plugin: 事件环内核侧实现(§3.6 Part 5 核心)

事件环(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
This commit is contained in:
JianFeeeee
2026-09-02 16:48:27 +08:00
parent 77e6c70712
commit 91b22978fa
10 changed files with 690 additions and 28 deletions

View File

@ -36,9 +36,18 @@ type coreHandler struct {
invokeTool func(name string, args map[string]interface{}) (interface{}, error)
invokeStageFn func(ctx context.Context, stage string, seq uint64) error
invokeOutput func(channel string, args map[string]interface{}) (interface{}, error)
// evtRing 是事件环的订阅接口(实现由 internal/plugin 提供,避免循环依赖)。
evtRing EvtRingSubscriber
}
// EvtRingSubscriber 是事件环订阅接口,由 internal/plugin.EventRing 实现。
// proc 包不依赖 internal/plugin,通过接口解耦。
// EvtRingSubscribe 返回一个取消函数(与 Bus.Subscribe 约定一致)。
type EvtRingSubscriber interface {
EvtRingSubscribe(types []pubsdk.EventType) func()
}
// invokeStageWithCtx 反向调用插件执行 stage。
func (h *coreHandler) invokeStageWithCtx(ctx context.Context, stage string, seq uint64) error {
if h.invokeStageFn == nil {
return fmt.Errorf("插件 %s: stage 调用通道未就绪", h.name)
@ -447,12 +456,26 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{}
}
return nil, h.locks.release(h.name)
// ---- 事件订阅(原 case 23/24,今日空实现)----
case MethodEventsSubscribe, MethodEventsUnsubscribe:
// Part 5 通知面(事件环 + eventfd)落地后接线。
// 今日 C ABI 侧是空实现("给不了"而非"不给",§1.3);
// 明确返回未实现,比静默成功后收不到事件更容易排查。
return nil, fmt.Errorf("%s: 事件订阅待 Part 5 通知面落地(事件环 + eventfd)", method)
// ---- 事件订阅(原 case 23/24,子进程下首次真正可用,§3.6)----
case MethodEventsSubscribe:
var p struct {
Types []pubsdk.EventType `json:"types"`
}
if err := unmarshal(params, &p); err != nil {
return nil, err
}
if h.evtRing == nil {
return nil, fmt.Errorf("%s: 事件环未就绪", method)
}
// 订阅请求来自子进程——handler 直接注册到 Bus,
// 事件经 EventRing 写入环后由子进程消费。
h.evtRing.EvtRingSubscribe(p.Types)
return nil, nil
case MethodEventsUnsubscribe:
// 事件环的订阅没有持久化句柄(取消函数由 Subscribe 返回但子进程未保存)。
// 当前设计:子进程 Stop 时由内核统一清理其订阅。
return nil, nil
// ---- 多模态注入(C ABI 侧空实现)----
case MethodIOSetToolBlocks: