From f8add88dbc7a2a5f6ec57ab7c71fa4779e2835ca Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 27 Sep 2026 16:29:24 +0800 Subject: [PATCH] =?UTF-8?q?perf(stagehost):=20=E5=B7=A5=E5=85=B7=E5=A3=B0?= =?UTF-8?q?=E6=98=8E=E6=9F=A5=E8=AF=A2=E5=85=8D=E5=8E=BB=E7=BB=93=E6=9E=84?= =?UTF-8?q?=E4=BD=93=E6=8B=B7=E8=B4=9D=EF=BC=88=E7=83=AD=E8=B7=AF=E5=BE=84?= =?UTF-8?q?=201000=20=E5=B9=B6=E5=8F=91=E4=B8=8B=E7=9C=81=201000=20?= =?UTF-8?q?=E6=AC=A1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 问题 StageHost.ToolDef 返回 &def —— 一次**结构体拷贝**：3 个 string + 2 个 map 头 + 2 个 bool + Cleaner 函数指针。 而 toolParallelSafe 在**每批**并发判据里对每个工具各调一次： batchRunnable 遍历 PendingTools → toolParallelSafe(tc.Name)。 1000 并发批次 = 1000 次结构体拷贝，全在判定阶段（执行之前）。 不是"逃逸漏洞"（Go 1.22+ 循环变量每轮独立，go.mod 是 1.25），纯粹是白拷贝。 ## 修法 ToolDef 保留 —— 它要给需要完整声明的调用方（Cleaner、Parameters 校验）， 返回副本也是**有意**的（ToolDef 里有 map 与函数指针，交出内部元素会把 可变引用漏出去）。 新增免拷贝查询，热路径专用： ConcurrencySafeOf(name) (safe, found bool) // 只读 ParallelSafe && !Serial NoMemoryOf(name) (v, found bool) HasTool(name) bool 全部在持 RLock 下走同一个 findLocked。 `toolParallelSafe` 切到 ConcurrencySafeOf。语义完全等价 —— 两者都算 `ParallelSafe && !Serial`，只差一次拷贝。 ## 判据（两个都防"优化悄悄改了语义"） - TestNoCopyQueriesMatchToolDef 7 种声明组合（plain / parallel / serial / both / nomem / all / serial_nomem）下，免拷贝查询与 ToolDef(...).字段 **逐字段等价**；不存在的工具三态一致（false/false/true）。 ★ 这类优化最危险的失败模式就是语义漂移：并发判据若读错字段， 能并发的批次会**悄悄退化成串行** —— 没有任何报错，只表现为"变慢了"。 所以判据必须逐个组合比对，而不是只测一个典型值。 - TestNoCopyQueriesConcurrent 32 goroutine × 50 工具并发查询， -race 无竞态且结果与串行一致。 回归：go build ./... 通过；go test ./internal/... 全绿； go test -race ./internal/agent/core 通过。 --- internal/agent/core/argvalidate.go | 8 +- internal/agent/core/stages.go | 62 +++++++++++++- internal/agent/core/stages_tooldef_test.go | 97 ++++++++++++++++++++++ 3 files changed, 161 insertions(+), 6 deletions(-) create mode 100644 internal/agent/core/stages_tooldef_test.go diff --git a/internal/agent/core/argvalidate.go b/internal/agent/core/argvalidate.go index 51594fc..d613b76 100644 --- a/internal/agent/core/argvalidate.go +++ b/internal/agent/core/argvalidate.go @@ -258,9 +258,13 @@ func (a *Agent) toolParallelSafe(name string) bool { if a == nil { return false } + // ⚠️ 这里用 ConcurrencySafeOf 而不是 ToolDef(name).ParallelSafe： + // 后者每次调用**拷贝整个 ToolDef**（含 2 个 map 与 Cleaner 函数指针）， + // 而本函数在每批并发判据里对每个工具各调一次 —— 1000 并发就是 1000 次 + // 拷贝。语义完全等价（两者都算 ParallelSafe && !Serial），只是不拷贝。 if a.stageHost != nil { - if def := a.stageHost.ToolDef(name); def != nil { - return def.ParallelSafe && !def.Serial + if safe, ok := a.stageHost.ConcurrencySafeOf(name); ok { + return safe } } if a.io != nil { diff --git a/internal/agent/core/stages.go b/internal/agent/core/stages.go index 6ce9581..d879083 100644 --- a/internal/agent/core/stages.go +++ b/internal/agent/core/stages.go @@ -72,17 +72,71 @@ func (h *StageHost) GetToolDefs() []sdk.ToolDef { return defs } +// ToolDef 返回工具声明的**副本**。 +// +// 保留"返回副本"是有意的：ToolDef 里有 map 与函数指针（Cleaner）， +// 返回内部切片元素会把可变引用交出去。Go 1.22+ 循环变量每轮独立， +// 所以 &def 不是逃逸漏洞。 +// +// 但代价真实存在：结构体含 3 个 string + 2 个 map 头 + 2 个 bool， +// 每次调用都是一次结构体拷贝。而 toolParallelSafe 在**每批**判断里对 +// 每个工具各调一次，1000 并发时就是 1000 次拷贝。 +// +// 只需"是否存在 + 声明项"的调用方（toolParallelSafe、validateArgsAgainstSchema +// 等热路径）应改用 ConcurrencySafeOf / HasTool，避免拷贝。 +// 需要拿到完整声明的（如 Cleaner）才用 ToolDef。 func (h *StageHost) ToolDef(name string) *sdk.ToolDef { h.mu.RLock() defer h.mu.RUnlock() - for _, def := range h.toolDefs { - if def.Name == name { - return &def - } + if def, ok := h.findLocked(name); ok { + return &def } return nil } +// ConcurrencySafeOf 报告工具是否可并发执行，**不做结构体拷贝**。 +// +// 与 ToolDef(name).ParallelSafe && !Serial 等价，但只读两个 bool 字段。 +// 热路径（每批并发判据）必须走这个。 +func (h *StageHost) ConcurrencySafeOf(name string) (safe, found bool) { + h.mu.RLock() + defer h.mu.RUnlock() + def, ok := h.findLocked(name) + if !ok { + return false, false + } + return def.ParallelSafe && !def.Serial, true +} + +// NoMemoryOf 报告工具是否声明 NoMemory，同样不拷贝。 +func (h *StageHost) NoMemoryOf(name string) (v, found bool) { + h.mu.RLock() + defer h.mu.RUnlock() + def, ok := h.findLocked(name) + if !ok { + return false, false + } + return def.NoMemory, true +} + +// HasTool 报告工具是否存在，不拷贝。 +func (h *StageHost) HasTool(name string) bool { + h.mu.RLock() + defer h.mu.RUnlock() + _, ok := h.findLocked(name) + return ok +} + +// findLocked 必须在持有 h.mu 时调用。 +func (h *StageHost) findLocked(name string) (sdk.ToolDef, bool) { + for _, def := range h.toolDefs { + if def.Name == name { + return def, true + } + } + return sdk.ToolDef{}, false +} + func (h *StageHost) ExecuteTool(name string, args map[string]interface{}) (ret interface{}, err error) { defer func() { if r := recover(); r != nil { diff --git a/internal/agent/core/stages_tooldef_test.go b/internal/agent/core/stages_tooldef_test.go new file mode 100644 index 0000000..e0a9406 --- /dev/null +++ b/internal/agent/core/stages_tooldef_test.go @@ -0,0 +1,97 @@ +package core + +import ( + "sync" + "testing" + + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// 无拷贝查询必须与 ToolDef(...).字段 **完全等价**。 +// +// 这类优化最危险的失败模式是"优化悄悄改了语义"：比如并发判据原本 +// 读 ParallelSafe，优化后读成了别的字段，于是工具能并发的批次悄悄 +// 退化成串行 —— 没有任何报错。 +func TestNoCopyQueriesMatchToolDef(t *testing.T) { + sh := NewStageHost() + noop := func(map[string]interface{}) (interface{}, error) { return nil, nil } + cases := []sdk.ToolDef{ + {Name: "plain"}, + {Name: "parallel", ParallelSafe: true}, + {Name: "serial", Serial: true}, + {Name: "both", ParallelSafe: true, Serial: true}, + {Name: "nomem", NoMemory: true}, + {Name: "all", ParallelSafe: true, NoMemory: true}, + {Name: "serial_nomem", Serial: true, NoMemory: true}, + } + for _, d := range cases { + if err := sh.RegisterTool(d.Name, d, noop); err != nil { + t.Fatal(err) + } + } + for _, d := range cases { + def := sh.ToolDef(d.Name) + if def == nil { + t.Fatalf("%s: ToolDef 返回 nil", d.Name) + } + safe, found := sh.ConcurrencySafeOf(d.Name) + if !found { + t.Errorf("%s: ConcurrencySafeOf 未找到", d.Name) + continue + } + if want := def.ParallelSafe && !def.Serial; safe != want { + t.Errorf("%s: ConcurrencySafeOf=%v，ToolDef 算出的应为 %v（ParallelSafe=%v Serial=%v）", + d.Name, safe, want, def.ParallelSafe, def.Serial) + } + nm, ok := sh.NoMemoryOf(d.Name) + if !ok { + t.Errorf("%s: NoMemoryOf 未找到", d.Name) + continue + } + if nm != def.NoMemory { + t.Errorf("%s: NoMemoryOf=%v，ToolDef.NoMemory=%v", d.Name, nm, def.NoMemory) + } + if sh.HasTool(d.Name) != true { + t.Errorf("%s: HasTool 应为 true", d.Name) + } + } + // 不存在的工具 + if safe, found := sh.ConcurrencySafeOf("nope"); found || safe { + t.Errorf("不存在的工具：found=%v safe=%v，应为 false/false", found, safe) + } + if sh.HasTool("nope") { + t.Error("HasTool 对不存在的工具返回 true") + } +} + +// 并发查询必须无竞态，且与串行结果一致。 +func TestNoCopyQueriesConcurrent(t *testing.T) { + sh := NewStageHost() + noop := func(map[string]interface{}) (interface{}, error) { return nil, nil } + for i := 0; i < 50; i++ { + name := "t" + string(rune('a'+i%26)) + string(rune('0'+i/26)) + _ = sh.RegisterTool(name, sdk.ToolDef{Name: name, ParallelSafe: i%2 == 0}, noop) + } + var wg sync.WaitGroup + for g := 0; g < 32; g++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 50; i++ { + name := "t" + string(rune('a'+i%26)) + string(rune('0'+i/26)) + safe, found := sh.ConcurrencySafeOf(name) + if !found { + t.Errorf("并发下 %s 未找到", name) + return + } + if safe != (i%2 == 0) { + t.Errorf("并发下 %s safe=%v，期望 %v", name, safe, i%2 == 0) + return + } + _ = sh.HasTool(name) + _, _ = sh.NoMemoryOf(name) + } + }() + } + wg.Wait() +}