mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 01:18:08 +00:00
5bbfcc02fb
事件环（EvtRing）是子进程首次获得事件订阅能力的基础设施。 此前 case 23/24 明确返回未实现，现在经事件环真正可用。 核心设计（§3.6，实验 4 已验证 post-and-forget 加速比 2218x）： - 事件环放**独立共享段**（不与 StageContext 混放）：stage compact 会清 arena， 事件要独立于 stage 生命周期。Host 持有两块 memfd：fd 3 = StageContext， fd 4 = 事件环段，fd 5 = eventfd。 - 无锁数据结构：内核 WritePush 追加写 slot，子进程 EvtConsumer 消费。 writeSeq 原子递增（Bus.Publish 并发调用），readSeq 每订阅者独立。 - eventfd 通知：Linux 用 unix.Eventfd（计数合并，1000 token 只唤醒几次）， macOS 用 os.Pipe（阻塞模式走 netpoller，只 park goroutine，实验 1 验证 200 等待者仅 +1 OS 线程）。两者行为一致：Read 阻塞直到有新事件。 - 溢出语义：落后超 cap 时跳到最新，丢弃计数记入 dropped（消费者知道丢了）。 不静默覆盖最旧（写端直接覆盖 slot，读端靠 seq 判断跳过）。 - 事件类型编码：pubsdk.EventType 字符串 ↔ uint32 位索引（编译时映射表）， typeMask 位掩码过滤（1<<idx）。 Host 改动： - NewHost 同时创建事件环段和 eventfd（惰创建，一次分配）。 - Host 持有 evtSubscriber 接口（EvtRingSubscriber），由 Registry 注入 EventRing 实现——proc 包不依赖 internal/plugin（避免循环依赖）。 corehandler 改动： - events.subscribe（原 case 23）：子进程传事件类型列表，coreHandler 通过 evtRing 接口调用 EvtRingSubscribe，注册到 Bus 上。 事件经 EventRing 写入环后由子进程 mmap 读取。 - events.unsubscribe（原 case 24）：当前由内核统一清理（子进程 Stop 时）。 Registry 改动： - ensureProcHost 在创建 Host 后同时创建 EventRing（Bus → EvtRing → eventfd）， 并通过 Host.SetEvtSubscriber 注入给 coreHandler。 测试 3 项： - BasicWriteAndConsume：Host 创建 → EventRing 写入 → 消费者读到 - OverflowStillDelivers：写入超过 cap 后消费者仍能读到最新事件 - TypeMaskFiltering：typeMask 只订阅 tool_call，agent_output 被过滤 验证：go build ./... 通过；go test -race ./internal/plugin/... 全绿； 既有事件环测试 3/3 通过；proc 包测试未受影响。 Ref: docs/zh/架构迁移评估.md §3.6、docs/zh/plugin-migration-plan.md Part 5
56 lines
1.6 KiB
Go
56 lines
1.6 KiB
Go
package plugin
|
||
|
||
import (
|
||
"encoding/json"
|
||
"sync"
|
||
|
||
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
|
||
"gitcode.com/JianFeeeee/HomeAgent/internal/plugin/proc"
|
||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||
)
|
||
|
||
// EventRing 是 Bus 与 proc.EvtRing 之间的适配层。
|
||
//
|
||
// 把内核的事件总线接到共享内存事件环：Bus.Publish → handler
|
||
// 把事件序列化写入 EvtRing slot → eventfd 通知子进程。
|
||
// 不改 Bus 自身结构（保护零 API 变动）。
|
||
type EventRing struct {
|
||
ring *proc.EvtRing
|
||
bus *events.Bus
|
||
efd int
|
||
mu sync.Mutex
|
||
}
|
||
|
||
func NewEventRing(ring *proc.EvtRing, efd int, bus *events.Bus) *EventRing {
|
||
return &EventRing{ring: ring, bus: bus, efd: efd}
|
||
}
|
||
|
||
// Subscribe 在 Bus 上注册一个把事件分发到事件环的 handler，返回取消函数。
|
||
//
|
||
// 不改 Bus 自身结构——handler 把事件序列化后写入环并 post eventfd，
|
||
// Bus 侧按 EventType 精确匹配分发（与现有逻辑完全一致）。
|
||
func (er *EventRing) Subscribe(eventType pubsdk.EventType) func() {
|
||
return er.bus.Subscribe(events.EventType(eventType), func(evt *events.Event) {
|
||
payload, err := json.Marshal(evt)
|
||
if err != nil {
|
||
return
|
||
}
|
||
er.ring.WritePush(pubsdk.EventType(evt.Type), payload)
|
||
proc.EvtfdNotify(er.efd)
|
||
})
|
||
}
|
||
|
||
// EvtRingSubscribe 实现 proc.EvtRingSubscriber 接口。
|
||
// 按事件类型列表订阅，返回统一取消函数。
|
||
func (er *EventRing) EvtRingSubscribe(types []pubsdk.EventType) func() {
|
||
unsubscribes := make([]func(), 0, len(types))
|
||
for _, t := range types {
|
||
unsubscribes = append(unsubscribes, er.Subscribe(t))
|
||
}
|
||
return func() {
|
||
for _, fn := range unsubscribes {
|
||
fn()
|
||
}
|
||
}
|
||
}
|