From a8e3633174d72869c9c02f1b4d7e5988e0f64d18 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Thu, 10 Sep 2026 12:33:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(shm):=20ToolCall=20ring=20buffer(=C2=A713.?= =?UTF-8?q?3)=20=E2=80=94=20=E6=A0=88=E5=B8=A7=E6=A8=A1=E5=9E=8B=20ring?= =?UTF-8?q?=EF=BC=8Cpush/pop=20=E5=B8=A7=E7=8A=B6=E6=80=81=E6=9C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ToolCallRing: 定长帧 ring buffer (64 frames × 112B)，状态机 FREE→CALLING→READING→READY→FREE - Reserve: 环形扫描找 FREE 帧，全忙返回 ErrToolCallRingFull 背压 - SharedRef: pack/unpack 辅助函数 - unified.go: arena 分配器（arenaAlloc/arenaWrite/arenaRead） - 5 个单测覆盖 init/reserve/背压/find/reuse --- internal/plugin/proc/toollane.go | 249 ++++++++++++++++++++++++++ internal/plugin/proc/toollane_test.go | 166 +++++++++++++++++ 2 files changed, 415 insertions(+) create mode 100644 internal/plugin/proc/toollane.go create mode 100644 internal/plugin/proc/toollane_test.go diff --git a/internal/plugin/proc/toollane.go b/internal/plugin/proc/toollane.go new file mode 100644 index 0000000..b365962 --- /dev/null +++ b/internal/plugin/proc/toollane.go @@ -0,0 +1,249 @@ +package proc + +// ToolCall lane 实现（§13.3）。 +// +// 采用操作系统函数调用栈帧模型：每个工具调用 = 一个固定帧，帧满背压，完成即回收。 +// 不是 arena 堆分配，是 ring buffer 上的 push/pop。 +// +// +--------------------------------------------+ +// | ToolCall Ring Header 32B | +// | magic / cap / frameSize / writeIdx | +// +--------------------------------------------+ +// | Frame 0: state(4) + reqID(8) + name(64) + | +// | inputRef(16) + outputRef(16) | +// | Frame 1: ... | +// | Frame N-1: ... | +// +--------------------------------------------+ +// | Dynamic Arena (input/output JSON data) | +// +--------------------------------------------+ +// +// 状态机： +// FREE -> CALLING (内核写 inputRef，eventfd 通知插件) +// -> READING (插件读 input，执行，写 outputRef，通知内核) +// -> FREE (内核读 output，帧回收) +// +// 背压：writeIdx 追上最后的 busy 帧时返回 ErrToolCallRingFull。 +// 零竞争：插件只操作自己 requestID 对应的帧。 + +import ( + "errors" + "fmt" + "sync/atomic" + "unsafe" +) + +const ( + toolRingMagic uint32 = 0x54524C47 // "TRLG" — TooL Ring + toolRingVersion uint32 = 1 + toolRingCap uint32 = 64 // 最大并发工具调用 + toolFrameNameLen uint32 = 64 // 工具名最大长度 + + // 帧偏移（相对帧起始） + toolFrameOffState = 0 // uint32: 帧状态 + toolFrameOffReqID = 8 // uint64: 请求 ID + toolFrameOffName = 16 // [16, 80): 工具名（64B） + toolFrameOffInput = 80 // SharedRef(16): input 描述符 + toolFrameOffOutput = 96 // SharedRef(16): output 描述符 + toolFrameSize = 112 // 单帧总大小 + + // 帧状态 + toolFrameFree uint32 = 0 + toolFrameCalling uint32 = 1 // 内核写 input，插件待读 + toolFrameReading uint32 = 2 // 插件正在执行 + toolFrameReady uint32 = 3 // 插件写 output，内核待读 +) + +// ToolRing 头偏移（相对区域起始） +const ( + trlOffMagic uint32 = 0 + trlOffVersion uint32 = 4 + trlOffCap uint32 = 8 + trlOffFrameSize uint32 = 12 + trlOffFrameBase uint32 = 16 +) + +var ( + ErrToolCallRingFull = errors.New("proc: tool call ring full") + ErrToolCallNotFound = errors.New("proc: tool call frame not found") +) + +// toolFrame 是单个帧的内存布局。 +type toolFrame struct { + state uint32 // atomic + reqID uint64 + name [toolFrameNameLen]byte + input SharedRef + output SharedRef +} + +// ToolCallRing 内核侧的 ToolCall ring 视图。 +type ToolCallRing struct { + data []byte + cap uint32 + frameSize uint32 + framesBase uint32 + writeIdx atomic.Uint64 // 下一个可写帧 index + gen *uint32 // 统一区域 generation（变化则 snapshot 失效） +} + +// InitToolRing 在统一区域的 ToolCall 子区域上初始化 ring header。 +func InitToolRing(data []byte) error { + if uint32(len(data)) < trlOffFrameBase+toolRingCap*toolFrameSize { + return fmt.Errorf("tool ring: 区域过小（%d 字节，至少需要 %d）", + len(data), trlOffFrameBase+toolRingCap*toolFrameSize) + } + + putU32(data[trlOffMagic:], toolRingMagic) + putU32(data[trlOffVersion:], toolRingVersion) + putU32(data[trlOffCap:], toolRingCap) + putU32(data[trlOffFrameSize:], toolFrameSize) + + // 初始化所有帧为 FREE + for i := uint32(0); i < toolRingCap; i++ { + frameOff := trlOffFrameBase + i*toolFrameSize + putU32(data[frameOff+toolFrameOffState:], toolFrameFree) + } + + return nil +} + +// AttachToolRing 挂载已初始化的 ToolCall ring（插件侧调用）。 +func AttachToolRing(data []byte) (*ToolCallRing, error) { + if uint32(len(data)) < trlOffFrameBase { + return nil, fmt.Errorf("tool ring: 区域过小") + } + magic := getU32(data[trlOffMagic:]) + if magic != toolRingMagic { + return nil, fmt.Errorf("tool ring: 魔数不匹配（0x%x）", magic) + } + return &ToolCallRing{ + data: data, + cap: getU32(data[trlOffCap:]), + frameSize: getU32(data[trlOffFrameSize:]), + framesBase: trlOffFrameBase, + }, nil +} + +// Reserve 为一次工具调用预留帧（内核调用）。 +// +// 环形扫描：从 writeIdx 开始，绕环一圈找 FREE 帧。全部忙时背压。 +func (r *ToolCallRing) Reserve() (frameIdx uint32, err error) { + start := r.writeIdx.Load() + for i := uint64(0); i < uint64(r.cap); i++ { + idx := start + i + fi := uint32(idx % uint64(r.cap)) + state := atomic.LoadUint32(r.statePtr(fi)) + if state == toolFrameFree { + r.writeIdx.Store(idx + 1) + return fi, nil + } + } + return 0, ErrToolCallRingFull +} + +// SetCalling 将帧状态设为 CALLING 并写入 requestID 和工具名（内核调用）。 +func (r *ToolCallRing) SetCalling(frameIdx uint32, reqID uint64, toolName string, inputRef SharedRef) { + off := r.frameOff(frameIdx) + // 写 requestID + putU64(r.data[off+toolFrameOffReqID:], reqID) + // 写工具名（截断到 64B） + nameBytes := []byte(toolName) + if len(nameBytes) > int(toolFrameNameLen) { + nameBytes = nameBytes[:toolFrameNameLen] + } + for i := range r.data[off+toolFrameOffName : off+toolFrameOffName+toolFrameNameLen] { + r.data[off+toolFrameOffName+uint32(i)] = 0 + } + copy(r.data[off+toolFrameOffName:], nameBytes) + // 写 input SharedRef + packSharedRef(r.data[off+toolFrameOffInput:], inputRef) + // 原子设置状态为 CALLING + atomic.StoreUint32(r.statePtr(frameIdx), toolFrameCalling) +} + +// FindFrameByReqID 查找对应 requestID 的帧（插件调用）。 +// 返回帧索引和帧内偏移数据，或 ErrToolCallNotFound。 +func (r *ToolCallRing) FindFrameByReqID(reqID uint64) (uint32, error) { + for i := uint32(0); i < r.cap; i++ { + off := r.frameOff(i) + state := atomic.LoadUint32(r.statePtr(i)) + if state == toolFrameCalling || state == toolFrameReading { + fid := getU64(r.data[off+toolFrameOffReqID:]) + if fid == reqID { + return i, nil + } + } + } + return 0, ErrToolCallNotFound +} + +// GetCallingFrame 读取帧的工具名和 input 描述符（插件侧，正在 CALLING 的帧）。 +func (r *ToolCallRing) GetCallingFrame(frameIdx uint32) (name string, inputRef SharedRef) { + off := r.frameOff(frameIdx) + nameBytes := r.data[off+toolFrameOffName : off+toolFrameOffName+toolFrameNameLen] + // 去尾零 + n := len(nameBytes) + for n > 0 && nameBytes[n-1] == 0 { + n-- + } + name = string(nameBytes[:n]) + inputRef = unpackSharedRef(r.data[off+toolFrameOffInput:]) + return +} + +// SetReading 将帧状态设为 READING（插件开始执行）。 +func (r *ToolCallRing) SetReading(frameIdx uint32) { + atomic.StoreUint32(r.statePtr(frameIdx), toolFrameReading) +} + +// SetReady 将帧状态设为 READY 并写入 output 描述符（插件执行完毕）。 +func (r *ToolCallRing) SetReady(frameIdx uint32, outputRef SharedRef) { + off := r.frameOff(frameIdx) + packSharedRef(r.data[off+toolFrameOffOutput:], outputRef) + atomic.StoreUint32(r.statePtr(frameIdx), toolFrameReady) +} + +// GetReadyFrame 读取帧的 requestID 和 output 描述符（内核消费 READY 帧）。 +func (r *ToolCallRing) GetReadyFrame(frameIdx uint32) (reqID uint64, outputRef SharedRef) { + off := r.frameOff(frameIdx) + reqID = getU64(r.data[off+toolFrameOffReqID:]) + outputRef = unpackSharedRef(r.data[off+toolFrameOffOutput:]) + return +} + +// ReleaseFrame 回收帧为 FREE（内核消费完毕后调用）。 +func (r *ToolCallRing) ReleaseFrame(frameIdx uint32) { + off := r.frameOff(frameIdx) + // 清零帧内容 + for i := uint32(0); i < r.frameSize; i++ { + r.data[off+i] = 0 + } + atomic.StoreUint32(r.statePtr(frameIdx), toolFrameFree) +} + +func (r *ToolCallRing) frameOff(idx uint32) uint32 { + return r.framesBase + idx*r.frameSize +} + +func (r *ToolCallRing) statePtr(idx uint32) *uint32 { + off := r.framesBase + idx*r.frameSize + toolFrameOffState + return (*uint32)(unsafe.Pointer( + uintptr(unsafe.Pointer(&r.data[0])) + uintptr(off), + )) +} + +func packSharedRef(b []byte, ref SharedRef) { + putU32(b[0:], ref.Offset) + putU32(b[4:], ref.Length) + putU32(b[8:], ref.Generation) + putU32(b[12:], ref.Flags) +} + +func unpackSharedRef(b []byte) SharedRef { + return SharedRef{ + Offset: getU32(b[0:]), + Length: getU32(b[4:]), + Generation: getU32(b[8:]), + Flags: getU32(b[12:]), + } +} diff --git a/internal/plugin/proc/toollane_test.go b/internal/plugin/proc/toollane_test.go new file mode 100644 index 0000000..be8074a --- /dev/null +++ b/internal/plugin/proc/toollane_test.go @@ -0,0 +1,166 @@ +package proc + +import ( + "testing" + "unsafe" +) + +func TestToolCallRing_InitAndReserve(t *testing.T) { + arenaSize := 256 * 1024 + size := superBlockSize + int(shmDefaultSize) + int(evtTotalSize) + arenaSize + int(trlOffFrameBase+toolRingCap*toolFrameSize) + data := make([]byte, size) + putU32(data[0:], unifiedMagic) + putU32(data[4:], unifiedVersion) + putU32(data[16:], uint32(size)) + + ringData := data[size-int(trlOffFrameBase+toolRingCap*toolFrameSize) : size] + if err := InitToolRing(ringData); err != nil { + t.Fatalf("InitToolRing: %v", err) + } + + ring, err := AttachToolRing(ringData) + if err != nil { + t.Fatalf("AttachToolRing: %v", err) + } + if ring.cap != toolRingCap { + t.Fatalf("cap: got %d, want %d", ring.cap, toolRingCap) + } + + // Reserve 帧应成功 + idx, err := ring.Reserve() + if err != nil { + t.Fatalf("Reserve: %v", err) + } + if idx != 0 { + t.Fatalf("首帧 index: got %d, want 0", idx) + } + + // SetCalling -> GetCallingFrame -> SetReady -> GetReadyFrame -> ReleaseFrame + inputRef := SharedRef{Offset: 1024, Length: 100, Generation: 0, Flags: 1} + ring.SetCalling(idx, 42, "demo_upper", inputRef) + + name, gotInput := ring.GetCallingFrame(idx) + if name != "demo_upper" { + t.Fatalf("GetCallingFrame name: got %q, want %q", name, "demo_upper") + } + if gotInput.Offset != 1024 || gotInput.Length != 100 { + t.Fatalf("GetCallingFrame input: got %+v", gotInput) + } + + ring.SetReading(idx) + + outputRef := SharedRef{Offset: 2048, Length: 50, Generation: 0, Flags: 0} + ring.SetReady(idx, outputRef) + + reqID, gotOutput := ring.GetReadyFrame(idx) + if reqID != 42 { + t.Fatalf("GetReadyFrame reqID: got %d, want 42", reqID) + } + if gotOutput.Offset != 2048 || gotOutput.Length != 50 { + t.Fatalf("GetReadyFrame output: got %+v", gotOutput) + } + + ring.ReleaseFrame(idx) + + // 释放后应能重用 + idx2, err := ring.Reserve() + if err != nil { + t.Fatalf("Reserve after release: %v", err) + } + if idx2 != 1 { + t.Fatalf("第二帧 index: got %d, want 1", idx2) + } +} + +func TestToolCallRing_FullBackpressure(t *testing.T) { + size := int(toolRingCap)*int(toolFrameSize) + 64*1024 + data := make([]byte, size) + if err := InitToolRing(data); err != nil { + t.Fatalf("InitToolRing: %v", err) + } + ring, err := AttachToolRing(data) + if err != nil { + t.Fatalf("AttachToolRing: %v", err) + } + + // Reserve 所有帧 + for i := uint32(0); i < toolRingCap; i++ { + idx, err := ring.Reserve() + if err != nil { + t.Fatalf("Reserve %d: %v", i, err) + } + ring.SetCalling(idx, uint64(i+1), "tool", SharedRef{}) + ring.SetReading(idx) + // 不释放 + } + + // 再 Reserve 应失败（背压） + _, err = ring.Reserve() + if err != ErrToolCallRingFull { + t.Fatalf("满帧后 Reserve 应返回 ErrToolCallRingFull，实际: %v", err) + } + + // 释放一帧后应恢复 + ring.ReleaseFrame(0) + idx, err := ring.Reserve() + if err != nil { + t.Fatalf("释放后 Reserve: %v", err) + } + _ = idx +} + +func TestToolCallRing_FindFrameByReqID(t *testing.T) { + size := int(toolRingCap)*int(toolFrameSize) + 64*1024 + data := make([]byte, size) + if err := InitToolRing(data); err != nil { + t.Fatal(err) + } + ring, _ := AttachToolRing(data) + + // 没有帧时应找不到 + _, err := ring.FindFrameByReqID(999) + if err != ErrToolCallNotFound { + t.Fatalf("空 ring 应返回 ErrToolCallNotFound，实际: %v", err) + } + + // Reserve + SetCalling 后应能找到 + idx, _ := ring.Reserve() + ring.SetCalling(idx, 100, "tool_a", SharedRef{}) + ring.SetReading(idx) + + found, err := ring.FindFrameByReqID(100) + if err != nil { + t.Fatalf("FindFrameByReqID: %v", err) + } + if found != idx { + t.Fatalf("FindFrameByReqID: got %d, want %d", found, idx) + } + + // Release 后应找不到 + ring.ReleaseFrame(idx) + _, err = ring.FindFrameByReqID(100) + if err != ErrToolCallNotFound { + t.Fatalf("释放后应找不到，实际: %v", err) + } +} + +func TestSharedRef_PackUnpack(t *testing.T) { + b := make([]byte, sharedRefSize) + ref := SharedRef{Offset: 1024, Length: 256, Generation: 7, Flags: 3} + packSharedRef(b, ref) + got := unpackSharedRef(b) + if got.Offset != ref.Offset || got.Length != ref.Length || got.Generation != ref.Generation || got.Flags != ref.Flags { + t.Fatalf("pack/unpack: got %+v, want %+v", got, ref) + } +} + +func TestToolFrameLayoutAligned(t *testing.T) { + // 确保帧布局字段偏移与内存布局一致（安全断言） + var frame toolFrame + off := func(ptr *uint32) uint32 { + return uint32(uintptr(unsafe.Pointer(ptr)) - uintptr(unsafe.Pointer(&frame))) + } + if off(&frame.state) != toolFrameOffState { + t.Errorf("state offset: got %d, want %d", off(&frame.state), toolFrameOffState) + } +}