diff --git a/internal/plugin/proc/host.go b/internal/plugin/proc/host.go index 38f3613..8c94103 100644 --- a/internal/plugin/proc/host.go +++ b/internal/plugin/proc/host.go @@ -9,48 +9,37 @@ import ( pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk" ) -// Host 持有**被全部子进程插件共享的一块 StageContext 段**，是共享内存数据面的 +// Host 持有**被全部子进程插件共享的统一内存区域**，是共享数据面的 // 所有权中心（§3.3/§3.4）。 // -// ❗ 为什么必须共享一块段（这是一个容易走错的关键点）： -// 若每个插件各持一块段，则「内核 ctx → 段 → 插件改 → 回读 ctx」在多插件下退化成 +// ❗ 为什么必须共享一块区域： +// 若每个插件各持一块段，则「内核 ctx → 段 → 插件改 → 回读 ctx」退化成 // 副本模型——两个插件各写各的段、各自回读，最后回读者覆盖前者， // lost update 原样复现（§8.4 实测 35.8~36.8%）。 -// 实验 8 的做法是 5 个 worker 进程 mmap **同一个 memfd**，本实现与之一致。 // -// 生命周期：Host 由 registry 创建一次，随内核存活；每个插件 spawn 时经 -// ExtraFiles 拿到同一 memfd（fd 3），mmap 后即看到同一份物理页。 +// 统一区域（§13.1）：单 memfd 包含 SuperBlock + StageContext + EvtRing。 +// 子进程经 fd 3 mmap 同一 memfd → 同一份物理页，区域头 SuperBlock +// 告知各 segment 的偏移与大小。 // -// 另外持有事件环段（§3.6）：独立于 StageContext 的事件通知通道， -// 子进程从 eventfd 感知新事件并从 mmap 读 slot。 -// fd 分配：fd 3 = StageContext，fd 4 = 事件环，fd 5 = eventfd。 +// fd 分配：fd 3 = 统一区域，fd 4 = eventfd。 type Host struct { memfd *os.File - data []byte - seg *Segment - shmSize int + data []byte // 统一区域完整 mmap + unified *unifiedRegion // SuperBlock 解析结果 + seg *Segment // StageContext segment（位于 unified ctxData） + shmSize int // 统一区域总大小 - // 事件环段（独立于 StageContext） - evtfd *os.File // Unix：eventfd/pipe 读端（fd 5）。Windows 为 nil，用 evtNotifyFd 。 - evtNotifyFd int // 通知句柄的平台无关标识（Unix 是真 fd，Windows 是伪 fd） - evtRing *EvtRing // 内核侧事件环句柄 - evtRingFd *os.File // Unix：事件环段 memfd（fd 4）。Windows 为 nil（命名段）。 - evtData []byte // 事件环段 mmap 数据 + // 事件通知（独立于共享段） + evtfd *os.File // Unix：eventfd/pipe 读端（fd 4）。Windows 为 nil。 + evtNotifyFd int // 通知句柄的平台无关标识 + evtRing *EvtRing // 内核侧事件环句柄（位于 unified evtData） - // evtSubscriber 由 internal/plugin 注入，coreHandler 用它接子进程的 events.subscribe 请求。 - // proc 包不依赖 internal/plugin（循环依赖），故用接口类型存储。 evtSubscriber EvtRingSubscriber - - locks *lockRegistry - stageMu sync.Mutex - coordMu sync.Mutex - coord *stageCoordinator - - // sup 是内核侧唯一的子进程台账，与共享段同生命周期。 - // - // 放在 Host 而不是 registry 的理由：能拿到 Host 的地方就能拿到台账， - // 而 Host 本就是「全部子进程插件共享的那一份内核侧状态」。 - sup *Supervisor + locks *lockRegistry + stageMu sync.Mutex + coordMu sync.Mutex + coord *stageCoordinator + sup *Supervisor } // NewHost 创建共享段（平台层 allocShm + 布局初始化）。 @@ -63,22 +52,34 @@ type Host struct { // 三者共同点：全部插件看到同一份物理页，段内一律用相对偏移而非指针 // （实验 2 已验证各进程 mmap 到不同虚拟地址时偏移解引用仍正确）。 func NewHost() (*Host, error) { - memfd, data, err := allocShm(shmDefaultSize) + // 统一区域大小：SuperBlock + StageContext segment + EvtRing segment + unifiedSize := superBlockSize + shmDefaultSize + evtTotalSize + memfd, data, err := allocShm(unifiedSize) if err != nil { return nil, err } - seg, err := NewSegment(data) + + // 初始化 SuperBlock + 两个 segment + ur, err := initUnifiedRegion(data, shmDefaultSize, evtTotalSize) if err != nil { freeShm(memfd, data) return nil, err } - // 创建事件环段（独立于 StageContext） - evtRingFd, evtData, efd, err := allocEvtRing() + // 创建 StageContext segment（位于 SuperBlock 之后） + seg, err := NewSegment(ur.ctxData()) if err != nil { freeShm(memfd, data) - return nil, fmt.Errorf("事件环: %w", err) + return nil, err } + + // 初始化 EvtRing 子区域头部（magic/version/cap） + evtData := ur.evtData() + putU32(evtData[evtOffMagic:], evtRingMagic) + putU32(evtData[evtOffVersion:], evtRingVersion) + putU32(evtData[evtOffCap:], evtRingCap) + + // 创建 EvtRing segment（位于 StageContext 之后） evtRing, err := NewEvtRing(evtData) if err != nil { freeShm(memfd, data) @@ -86,17 +87,23 @@ func NewHost() (*Host, error) { } evtRing.Init() + // eventfd 独立于共享段，仍为单独 fd + efd, err := evtfdCreate() + if err != nil { + freeShm(memfd, data) + return nil, fmt.Errorf("创建 eventfd: %w", err) + } + return &Host{ sup: NewSupervisor(), memfd: memfd, data: data, + unified: ur, seg: seg, - shmSize: shmDefaultSize, + shmSize: unifiedSize, evtfd: evtfdReadFile(efd), evtNotifyFd: efd, evtRing: evtRing, - evtRingFd: evtRingFd, - evtData: evtData, locks: &lockRegistry{}, }, nil } @@ -123,14 +130,7 @@ func (h *Host) Close() error { 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 + h.data, h.memfd, h.unified = nil, nil, nil } if h.evtfd != nil { h.evtfd.Close() @@ -321,7 +321,7 @@ func (h *Host) EvtNotifyFd() int { return h.evtNotifyFd } func (h *Host) SetEvtSubscriber(sub EvtRingSubscriber) { h.evtSubscriber = sub } // EvtData 返回事件环段 mmap 数据（子进程消费者用）。 -func (h *Host) EvtData() []byte { return h.evtData } +func (h *Host) EvtData() []byte { return h.unified.evtData() } // EvtfdReadFile 返回 eventfd 的 *os.File（供子进程读取消费）。 func (h *Host) EvtfdReadFile() *os.File { return h.evtfd } diff --git a/internal/plugin/proc/shmpass_unix.go b/internal/plugin/proc/shmpass_unix.go index daec8b2..1e094b7 100644 --- a/internal/plugin/proc/shmpass_unix.go +++ b/internal/plugin/proc/shmpass_unix.go @@ -11,11 +11,10 @@ func (h *Host) procEnvForShm() []string { return nil } // // 顺序即 fd 编号（cmd.ExtraFiles[0] → 子进程 fd 3）： // -// fd 3 = StageContext 段 -// fd 4 = 事件环段 -// fd 5 = 事件通知（eventfd / pipe 读端） +// fd 3 = 统一共享内存区域（SuperBlock + StageContext + EvtRing） +// fd 4 = 事件通知（eventfd / pipe 读端） // // 插件侧模板 z_proc_shm_unix.go 的常量与此严格对应。 func (h *Host) procExtraFilesForShm() []*os.File { - return []*os.File{h.memfd, h.evtRingFd, h.evtfd} + return []*os.File{h.memfd, h.evtfd} } diff --git a/internal/plugin/proc/streaming_test.go b/internal/plugin/proc/streaming_test.go index 0418f94..91f4391 100644 --- a/internal/plugin/proc/streaming_test.go +++ b/internal/plugin/proc/streaming_test.go @@ -26,18 +26,26 @@ func TestStreaming_SlowConsumerDoesNotBlockPublish(t *testing.T) { if err != nil { t.Fatalf("NewHost: %v", err) } - defer host.Close() ring := host.EvtRing() var consumed atomic.Int64 consumer := NewEvtConsumer(host.EvtData(), host.EvtfdReadFile(), 0, func(evt *pubsdk.Event) error { - time.Sleep(20 * time.Microsecond) // 刻意的慢订阅者 + time.Sleep(20 * time.Microsecond) consumed.Add(1) return nil }) - go consumer.Run() + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + consumer.Run() + }() + // defer LIFO：Stop → 关 evtfd 打断 Read → 等协程退出 → Close host + defer host.Close() + defer wg.Wait() + defer host.EvtfdReadFile().Close() defer consumer.Stop() const tokens = 5000 @@ -71,7 +79,9 @@ func TestStreaming_PublishLatencyFlatAcrossSubscribers(t *testing.T) { if err != nil { t.Fatalf("NewHost: %v", err) } + // defer LIFO：先停消费者再 Close host defer host.Close() + defer host.EvtfdReadFile().Close() ring := host.EvtRing() payload := []byte(`{"type":"content_delta","payload":{"text":"t"}}`) @@ -93,13 +103,14 @@ func TestStreaming_PublishLatencyFlatAcrossSubscribers(t *testing.T) { cc.Run() }(c) } + // defer LIFO：Stop → 等协程退出 + defer wg.Wait() defer func() { for _, c := range active { c.Stop() } }() - // 让消费者先就位 time.Sleep(10 * time.Millisecond) start := time.Now() diff --git a/internal/plugin/proc/testdata/appendplugin.go b/internal/plugin/proc/testdata/appendplugin.go index ba37df4..0d13a67 100644 --- a/internal/plugin/proc/testdata/appendplugin.go +++ b/internal/plugin/proc/testdata/appendplugin.go @@ -181,7 +181,10 @@ func main() { send(response{ID: req.ID, Error: fmt.Sprintf("mmap: %v", err)}) continue } - shm = m + // 统一区域：前 64B 是 SuperBlock，StageContext 段在其后 + ctxOff := binary.LittleEndian.Uint32(m[20:]) + ctxSize := binary.LittleEndian.Uint32(m[24:]) + shm = m[ctxOff : ctxOff+ctxSize] } send(response{ID: req.ID, Result: map[string]interface{}{ "protocol": 1, "sdk_version": "test", diff --git a/internal/plugin/proc/testdata/readonlyplugin.go b/internal/plugin/proc/testdata/readonlyplugin.go index 74a3fe8..24c64f5 100644 --- a/internal/plugin/proc/testdata/readonlyplugin.go +++ b/internal/plugin/proc/testdata/readonlyplugin.go @@ -133,7 +133,10 @@ func main() { send(response{ID: req.ID, Error: fmt.Sprintf("mmap: %v", err)}) continue } - shm = m + // 统一区域：前 64B 是 SuperBlock，StageContext 段在其后 + ctxOff := binary.LittleEndian.Uint32(m[20:]) + ctxSize := binary.LittleEndian.Uint32(m[24:]) + shm = m[ctxOff : ctxOff+ctxSize] } send(response{ID: req.ID, Result: map[string]interface{}{ "protocol": 1, "sdk_version": "test", "plugin_name": "readonly", "pid": os.Getpid(), diff --git a/internal/plugin/proc/testdata/stageplugin.go b/internal/plugin/proc/testdata/stageplugin.go index 3e76861..b496049 100644 --- a/internal/plugin/proc/testdata/stageplugin.go +++ b/internal/plugin/proc/testdata/stageplugin.go @@ -223,14 +223,17 @@ func main() { json.Unmarshal(req.Params, &hp) shmSize = hp.ShmSize if shmSize > 0 { - // fd 3 = 内核传入的共享段 memfd + // fd 3 = 内核传入的统一共享内存区域 m, err := syscall.Mmap(3, 0, shmSize, syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED) if err != nil { send(response{ID: req.ID, Error: fmt.Sprintf("mmap 共享段失败: %v", err)}) continue } - shm = m + // 统一区域：前 64B 是 SuperBlock，StageContext 段在其后 + ctxOff := binary.LittleEndian.Uint32(m[20:]) + ctxSize := binary.LittleEndian.Uint32(m[24:]) + shm = m[ctxOff : ctxOff+ctxSize] } send(response{ID: req.ID, Result: map[string]interface{}{ "protocol": 1, "sdk_version": "test", "plugin_name": "stage", "pid": os.Getpid(), diff --git a/internal/plugin/proc/unified.go b/internal/plugin/proc/unified.go new file mode 100644 index 0000000..02611b5 --- /dev/null +++ b/internal/plugin/proc/unified.go @@ -0,0 +1,212 @@ +package proc + +// 统一共享内存区域布局（§13.1）。 +// +// 之前的两块独立 memfd（StageContext 256KB + EvtRing ~320KB，fd 3/4） +// 合并为单一 memfd（fd 3），eventfd 独占 fd 4。 +// +// 区域内按固定偏移定位各 segment，所有偏移相对段起始，跨进程 mmap 到 +// 不同虚拟地址仍能正确解引用（§3.3 实验 2 已验证）。 +// +// ┌──────────────────────────────────────────────────────┐ +// │ SuperBlock 64B │ +// │ magic / version / generation / capacity / reserved │ +// │ ctxOff / ctxSize / evtOff / evtSize │ +// ├──────────────────────────────────────────────────────┤ +// │ StageContext segment （内部布局不变） │ +// ├──────────────────────────────────────────────────────┤ +// │ EvtRing segment （内部布局不变） │ +// ├──────────────────────────────────────────────────────┤ +// │ ToolCall lane [§13.3] │ +// ├──────────────────────────────────────────────────────┤ +// │ InputCh lane [§13.5] │ +// ├──────────────────────────────────────────────────────┤ +// │ OutputCh lane [§13.6] │ +// ├──────────────────────────────────────────────────────┤ +// │ Dynamic Arena │ +// └──────────────────────────────────────────────────────┘ + +import ( + "fmt" + "sync/atomic" + "unsafe" +) + +const ( + // unifiedMagic 标识统一共享内存区域。 + unifiedMagic uint32 = 0x554D5352 // "UMSR" — Unified Memory Shared Region + unifiedVersion uint32 = 1 + + superBlockSize = 64 // SuperBlock 占前 64 字节 +) + +// SuperBlock 布局：64 字节，所有偏移相对区域起始。 +// +// [0,4) magic +// [4,8) version +// [8,16) generation（atomic uint64，resize 时 bump） +// [16,20) capacity（区域总字节数） +// [20,24) ctxOff（StageContext segment 偏移） +// [24,28) ctxSize（StageContext segment 字节数） +// [28,32) evtOff（EvtRing segment 偏移） +// [32,36) evtSize（EvtRing segment 字节数） +// [36,48) reserved（未来 lane 偏移/大小） +// [48,56) reserved +// [56,64) reserved（对齐到 8 字节） +const ( + sbOffMagic = 0 + sbOffVersion = 4 + sbOffGeneration = 8 + sbOffCapacity = 16 + sbOffCtxOff = 20 + sbOffCtxSize = 24 + sbOffEvtOff = 28 + sbOffEvtSize = 32 +) + +// SharedRef 是跨进程共享内存描述符，替代内联 JSON 数据。 +// +// 所有数据交换（工具调用参数/结果、Cleaner、输入/输出通道消息） +// 都通过 SharedRef 传递：RPC 只传 16 字节描述符，实际数据在共享内存中。 +// +// Generation 防 ABA：扩容 remap 后旧描述符自动失效。 +type SharedRef struct { + Offset uint32 // 相对区域起始的偏移 + Length uint32 // 数据字节数 + Generation uint32 // 扩容后 bump + Flags uint32 // 保留，位 0 = 二进制，位 1 = JSON +} + +const sharedRefSize = 16 + +func (r SharedRef) IsZero() bool { + return r.Offset == 0 && r.Length == 0 +} + +// Slice 从共享内存中按 SharedRef 切片。data 必须是完整的 mmap 区域。 +func (r SharedRef) Slice(data []byte) []byte { + if r.IsZero() || int(r.Offset)+int(r.Length) > len(data) { + return nil + } + return data[r.Offset : r.Offset+r.Length] +} + +// unifiedRegion 统一共享内存区域的内核侧视图。 +type unifiedRegion struct { + data []byte + + ctxOff uint32 + ctxSize uint32 + evtOff uint32 + evtSize uint32 +} + +// initUnifiedRegion 在 mmap 区域上初始化 SuperBlock + 两个 segment。 +func initUnifiedRegion(data []byte, ctxTotal, evtTotal int) (*unifiedRegion, error) { + cap := uint32(len(data)) + total := superBlockSize + ctxTotal + evtTotal + if int(cap) < total { + return nil, fmt.Errorf("unified: 区域过小（%d 字节，至少需要 %d）", cap, total) + } + + ctxOff := uint32(superBlockSize) + evtOff := ctxOff + uint32(ctxTotal) + + putU32(data[sbOffMagic:], unifiedMagic) + putU32(data[sbOffVersion:], unifiedVersion) + putU64(data[sbOffGeneration:], 0) + putU32(data[sbOffCapacity:], cap) + putU32(data[sbOffCtxOff:], ctxOff) + putU32(data[sbOffCtxSize:], uint32(ctxTotal)) + putU32(data[sbOffEvtOff:], evtOff) + putU32(data[sbOffEvtSize:], uint32(evtTotal)) + + return &unifiedRegion{ + data: data, + ctxOff: ctxOff, + ctxSize: uint32(ctxTotal), + evtOff: evtOff, + evtSize: uint32(evtTotal), + }, nil +} + +// attachUnifiedRegion 从已有 mmap 区域解析 SuperBlock（插件侧调用）。 +func attachUnifiedRegion(data []byte) (*unifiedRegion, error) { + if len(data) < superBlockSize { + return nil, fmt.Errorf("unified: 区域过小（%d 字节）", len(data)) + } + magic := getU32(data[sbOffMagic:]) + if magic != unifiedMagic { + return nil, fmt.Errorf("unified: 魔数不匹配（0x%x，期望 0x%x）", magic, unifiedMagic) + } + ver := getU32(data[sbOffVersion:]) + if ver != unifiedVersion { + return nil, fmt.Errorf("unified: 版本不匹配（%d，期望 %d）", ver, unifiedVersion) + } + ctxOff := getU32(data[sbOffCtxOff:]) + ctxSize := getU32(data[sbOffCtxSize:]) + evtOff := getU32(data[sbOffEvtOff:]) + evtSize := getU32(data[sbOffEvtSize:]) + + if int(evtOff)+int(evtSize) > len(data) { + return nil, fmt.Errorf("unified: EvtRing 越界（off=%d size=%d total=%d）", evtOff, evtSize, len(data)) + } + + return &unifiedRegion{ + data: data, + ctxOff: ctxOff, + ctxSize: ctxSize, + evtOff: evtOff, + evtSize: evtSize, + }, nil +} + +func (r *unifiedRegion) generation() uint64 { + return getU64(r.data[sbOffGeneration:]) +} + +func (r *unifiedRegion) bumpGeneration() uint64 { + for { + old := getU64(r.data[sbOffGeneration:]) + new := old + 1 + p := (*uint64)(unsafe.Pointer(&r.data[sbOffGeneration])) + if atomic.CompareAndSwapUint64(p, old, new) { + return new + } + } +} + +func (r *unifiedRegion) ctxData() []byte { + return r.data[r.ctxOff : r.ctxOff+r.ctxSize] +} + +func (r *unifiedRegion) evtData() []byte { + return r.data[r.evtOff : r.evtOff+r.evtSize] +} + +func putU32(b []byte, v uint32) { + b[0] = byte(v) + b[1] = byte(v >> 8) + b[2] = byte(v >> 16) + b[3] = byte(v >> 24) +} + +func getU32(b []byte) uint32 { + return uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24 +} + +func putU64(b []byte, v uint64) { + b[0] = byte(v) + b[1] = byte(v >> 8) + b[2] = byte(v >> 16) + b[3] = byte(v >> 24) + b[4] = byte(v >> 32) + b[5] = byte(v >> 40) + b[6] = byte(v >> 48) + b[7] = byte(v >> 56) +} + +func getU64(b []byte) uint64 { + return uint64(b[0]) | uint64(b[1])<<8 | uint64(b[2])<<16 | uint64(b[3])<<24 | + uint64(b[4])<<32 | uint64(b[5])<<40 | uint64(b[6])<<48 | uint64(b[7])<<56 +}