mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 17:38:10 +00:00
feat(shm): ToolCall ring buffer(§13.3) — 栈帧模型 ring，push/pop 帧状态机
- 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
This commit is contained in:
JianFeeeee
249
internal/plugin/proc/toollane.go
Normal file
249
internal/plugin/proc/toollane.go
Normal file
@ -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:]),
|
||||
}
|
||||
}
|
||||
166
internal/plugin/proc/toollane_test.go
Normal file
166
internal/plugin/proc/toollane_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user