From 5bbfcc02fb8a4d7ef05a2d8f0da5ac4128114acd Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Wed, 2 Sep 2026 16:48:27 +0800 Subject: [PATCH] =?UTF-8?q?plugin:=20=E4=BA=8B=E4=BB=B6=E7=8E=AF=E5=86=85?= =?UTF-8?q?=E6=A0=B8=E4=BE=A7=E5=AE=9E=E7=8E=B0=EF=BC=88=C2=A73.6=20Part?= =?UTF-8?q?=205=20=E6=A0=B8=E5=BF=83=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 事件环(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< cap { + c.readSeq = writeSeq - cap + } + idx := c.readSeq % cap + slotOff := evtOffSlots + uint32(idx)*evtRingSlotLen + seq := binary.LittleEndian.Uint64(c.ringData[slotOff:]) + etype := binary.LittleEndian.Uint32(c.ringData[slotOff+8:]) + off := binary.LittleEndian.Uint32(c.ringData[slotOff+12:]) + slen := binary.LittleEndian.Uint32(c.ringData[slotOff+16:]) + if seq != c.readSeq { + // slot 已被新事件覆盖——逐个扫太慢(溢出场景 readSeq=0 要跳 100+ 步), + // 直接跳到 writeSeq 附近找下一个可读 slot。 + // 简化:溢出后直接跳到 writeSeq - cap(最旧的可读事件)。 + if writeSeq > cap { + c.readSeq = writeSeq - cap + } else { + c.readSeq = writeSeq + } + continue + } + // 位掩码过滤 + if c.typeMask != 0 && (1< 0 && slen > 0 && uint64(off)+uint64(slen) <= uint64(len(c.ringData)) { + payload := make([]byte, slen) + copy(payload, c.ringData[off:off+slen]) + var evt pubsdk.Event + if err := json.Unmarshal(payload, &evt); err == nil { + c.handler(&evt) + } + } + c.readSeq++ + } +} + +func (c *EvtConsumer) Stop() { + c.mu.Lock() + defer c.mu.Unlock() + if c.running { + close(c.stop) + } +} diff --git a/internal/plugin/proc/host.go b/internal/plugin/proc/host.go index 1f9e7db..7efa044 100644 --- a/internal/plugin/proc/host.go +++ b/internal/plugin/proc/host.go @@ -20,24 +20,28 @@ import ( // // 生命周期:Host 由 registry 创建一次,随内核存活;每个插件 spawn 时经 // ExtraFiles 拿到同一 memfd(fd 3),mmap 后即看到同一份物理页。 +// +// 另外持有事件环段(§3.6):独立于 StageContext 的事件通知通道, +// 子进程从 eventfd 感知新事件并从 mmap 读 slot。 +// fd 分配:fd 3 = StageContext,fd 4 = 事件环,fd 5 = eventfd。 type Host struct { memfd *os.File data []byte seg *Segment shmSize int - // locks 被全部插件的 coreHandler 共享——同阶段并发扇出的插件在此排队, - // 语义等价于内置插件共享 *StageContext 的 sync.RWMutex(§0.2 第 1 条)。 - locks *lockRegistry + // 事件环段(独立于 StageContext) + evtfd *os.File // eventfd fd(fd 5 的句柄,子进程读取消费) + evtRing *EvtRing // 内核侧事件环句柄 + evtRingFd *os.File // 事件环段 memfd(fd 4,子进程 mmap 读事件) + evtData []byte // 事件环段 mmap 数据 - // stageMu 串行化「整次 stage 执行」对共享段的独占。 - // - // 必要性:内核可能在不同路径并发触发 RunStage(如 emitResponse 的 - // before_output 与主循环的其他阶段)。段只有一份,两次 stage 交叠会互相污染。 - // 由首个进入的插件加锁、最后离开的插件解锁;RunStage 的 wg.Wait() 保证 - // 每个 handler 的 defer 必然执行,故 inflight 必然归零,不会死锁。 + // evtSubscriber 由 internal/plugin 注入,coreHandler 用它接子进程的 events.subscribe 请求。 + // proc 包不依赖 internal/plugin(循环依赖),故用接口类型存储。 + evtSubscriber EvtRingSubscriber + + locks *lockRegistry stageMu sync.Mutex - coordMu sync.Mutex coord *stageCoordinator } @@ -60,12 +64,29 @@ func NewHost() (*Host, error) { return nil, err } + // 创建事件环段(独立于 StageContext) + evtRingFd, evtData, efd, err := allocEvtRing() + if err != nil { + freeShm(memfd, data) + return nil, fmt.Errorf("事件环: %w", err) + } + evtRing, err := NewEvtRing(evtData) + if err != nil { + freeShm(memfd, data) + return nil, fmt.Errorf("事件环初始化: %w", err) + } + evtRing.Init() + return &Host{ - memfd: memfd, - data: data, - seg: seg, - shmSize: shmDefaultSize, - locks: &lockRegistry{}, + memfd: memfd, + data: data, + seg: seg, + shmSize: shmDefaultSize, + evtfd: evtfdReadFile(efd), + evtRing: evtRing, + evtRingFd: evtRingFd, + evtData: evtData, + locks: &lockRegistry{}, }, nil } @@ -76,11 +97,27 @@ func NewHost() (*Host, error) { // 全部插件共享一块,总开销恒定,不随插件数增长。 const shmDefaultSize = 256 * 1024 -// Close 释放共享段。 +// Close 释放共享段(StageContext + 事件环)。 func (h *Host) Close() error { - data, f := h.data, h.memfd - h.data, h.memfd = nil, nil - return freeShm(f, data) + var firstErr error + if h.data != nil { + if err := freeShm(h.memfd, h.data); err != nil && firstErr == nil { + firstErr = err + } + h.data, h.memfd = nil, nil + } + if h.evtData != nil { + if h.evtRingFd != nil { + h.evtRingFd.Close() + h.evtRingFd = nil + } + h.evtData = nil + } + if h.evtfd != nil { + h.evtfd.Close() + h.evtfd = nil + } + return firstErr } // beginStage 由插件 handler 进入时调用。 @@ -199,3 +236,18 @@ func (c *stageCoordinator) leave() (last bool, err error) { // ShmSize 返回共享段大小(供诊断/日志)。 func (h *Host) ShmSize() int { return h.shmSize } + +// EvtRing 返回内核侧事件环句柄。 +func (h *Host) EvtRing() *EvtRing { return h.evtRing } + +// Evtfd 返回 eventfd 的 *os.File(供 EventRing 写通知)。 +func (h *Host) Evtfd() *os.File { return h.evtfd } + +// SetEvtSubscriber 注入事件环订阅接口(由 Registry 在创建 Host 后设置)。 +func (h *Host) SetEvtSubscriber(sub EvtRingSubscriber) { h.evtSubscriber = sub } + +// EvtData 返回事件环段 mmap 数据(子进程消费者用)。 +func (h *Host) EvtData() []byte { return h.evtData } + +// EvtfdReadFile 返回 eventfd 的 *os.File(供子进程读取消费)。 +func (h *Host) EvtfdReadFile() *os.File { return h.evtfd } diff --git a/internal/plugin/proc/plugin.go b/internal/plugin/proc/plugin.go index 0f33b0e..c20a68f 100644 --- a/internal/plugin/proc/plugin.go +++ b/internal/plugin/proc/plugin.go @@ -73,6 +73,7 @@ func (p *Plugin) Start(core CoreSDK) error { name: p.name, host: p.host, locks: p.host.locks, + evtRing: p.host.evtSubscriber, } // 反向调用闭包:注册回调时捕获,运行期经 RPC 打到插件进程。 p.handler.invokeTool = p.invokeTool @@ -82,8 +83,8 @@ func (p *Plugin) Start(core CoreSDK) error { proc, err := Spawn(p.name, p.bin, Options{ Dir: p.dir, Env: p.env, - // 子进程 fd 3 = 共享段 memfd(全部插件同一个,故看到同一份物理页) - ExtraFiles: []*os.File{p.host.memfd}, + // 子进程 fd 布局:3=StageContext 段,4=事件环段,5=eventfd + ExtraFiles: []*os.File{p.host.memfd, p.host.evtRingFd, p.host.evtfd}, ShmSize: p.host.shmSize, Handler: p.handler.Handle, OnExit: p.handleExit,