From 0c959e3ff3f6505fdca1b6560ed3d60d4f3c771f Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Wed, 4 Nov 2026 16:53:13 +0800 Subject: [PATCH] =?UTF-8?q?fix(rtchild):=20pump=20=E8=A3=B8=E8=AF=BB?= =?UTF-8?q?=E5=86=99=20data=20race=20=E2=80=94=E2=80=94=20=E9=94=81?= =?UTF-8?q?=E8=B4=AF=E5=BD=BB=E5=88=B0=E5=BA=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 缺陷（-race 实测） rtChild 里 pump 是**唯一**"构造后被改写、又有并发读者"的字段， 但它的读写绕过了 rc.mu： Dispose（StopRTChild 的 goroutine） rc.pump = nil ← 写 rtEmit （rtLoop 的 goroutine） rc.pump ← 读 -race 栈： Read at … core.(*Agent).rtEmit rtchild.go:1275 Previous write … core.(*rtChild).Dispose rtchild.go:377 ## 为什么之前没被发现 TestRTChild_E2E_Sustained 覆盖同一个窗口，但它是**偶发命中**： 全量负载下 5 次中 1 次，单跑 20 次 0 次，基线 af17af5 全量 0 次。 这种形状在 CI 里等于没有 —— 红了是运气，绿了不代表安全。 ## 修法：锁贯彻到底（不是只补这一处） rtChild 全部字段逐个判定过： 字段 构造后写? 遵守锁规? pump 是(Dispose) ★ 本提交修：读写都持 rc.mu light/mainRO 是(Dispose) ✓ 无并发读者，只在 Dispose 内用 slot/input/output/agent 否 ✓ 只读，不需要锁 history 是(append) ✓ 三处访问全部持锁（原有） round/lastSeq/curCancel/critical/suppressedCancels ✓ 原本就持锁 即：pump 是唯一的漏网者，其余要么本已正确、要么根本没有并发写。 ## 两个顺序要求（不只是"加个锁"） ① **取走并置 nil 在锁内，Close 在锁外** pump.Close() 内部 wg.Wait() 等消费 goroutine 收尾；持锁等待可能与 另一个持锁路径交叉。而"置 nil"必须与读者的"读"互斥，否则读者会 拿到一个已被清掉的指针。 ② **rtEmit 锁内取、锁外 Push** Push 在队列满时会阻塞，持锁会卡死 Dispose 的置 nil。 ## 判据（2 条，2 次变异全部实测变红） TestRTPumpAccessIsLocked 一边不停 emit、一边 Dispose，用 -race 判定。 变异 A（rtEmit 裸读） ⇒ 连跑 3 次全部命中 DATA RACE 变异 B（Dispose 裸写） ⇒ 第 1 次命中 DATA RACE TestRTPumpAfterDisposeIsNil 钉住顺序语义：Dispose 之后 rc.pump 必须为 nil（否则 emit 会推 送到已关闭的 pump），且兜底路径不 panic。 ## 判据第一版是假的（记下来） 第一版读者写的是 `for i := 0; i < 500; i++` —— 500 次 emit 微秒级跑完， 而写者 sleep 1ms 才动手，两者**根本不重叠**。变异 B（裸写）连跑 6 次 都不红。改成"持续跑到写者宣布结束"之后才真正抓到。 这正是「判据必须能变红」的意义：不跑变异就永远不知道它是真判据还是摆设。 ## 判据的边界（诚实说明） 这条判据的"红"依赖 -race 检测器。不带 -race 时它只能发现 panic/死锁。 所以必须与 `go test -race` 一起跑 —— 提交纪律第 4 条已要求全量 -race， 这里只是把「哪个窗口最危险」钉成可复跑的形态。 ## 验证（提交纪律四项全绿） go build ./... 0 go vet ./... 0 gofmt -l internal/ cmd/ pkg/ 干净 go test ./... -count=1 -race 0 竞态，62 包全过 干净检出（worktree） build + TestRTPump 全绿 --- internal/agent/core/rtchild.go | 31 +++- internal/agent/core/rtchild_pump_lock_test.go | 135 ++++++++++++++++++ 2 files changed, 161 insertions(+), 5 deletions(-) create mode 100644 internal/agent/core/rtchild_pump_lock_test.go diff --git a/internal/agent/core/rtchild.go b/internal/agent/core/rtchild.go index 33830b96..70d5aeeb 100644 --- a/internal/agent/core/rtchild.go +++ b/internal/agent/core/rtchild.go @@ -372,9 +372,22 @@ func (rc *rtChild) Dispose() string { } // ★ 先排空输出队列再退出：Close 内部 wg.Wait() 会等消费 goroutine // 把剩余片发完。不排空会把"最后半句"吞掉 —— 用户听到话说到一半就断了。 - if rc.pump != nil { - rc.pump.Close() - rc.pump = nil + // + // ★ 取走 pump 必须在锁内、关闭必须在锁外（2026-11-04 修的 data race）： + // + // Dispose（StopRTChild 的 goroutine） 写 rc.pump = nil + // rtEmit （rtLoop 的 goroutine） 读 rc.pump + // + // 两者无同步 ⇒ -race 报 DATA RACE（实测：全量负载下 + // TestRTChild_E2E_Sustained 偶发命中）。加上锁之后，「取走并置 nil」 + // 与「读它」互斥；Close 放在锁外是因为它内部 wg.Wait() 等消费 + // goroutine 收尾，持锁等待可能与另一个持锁路径交叉。 + rc.mu.Lock() + pump := rc.pump + rc.pump = nil + rc.mu.Unlock() + if pump != nil { + pump.Close() } // 图记忆句柄必须关（SQLite 句柄，不关会随 create/stop 累积泄漏）。 // @@ -703,6 +716,9 @@ func (a *Agent) SpawnRTChild(opts RTChildOptions) (string, error) { // resolve 是闭包而不是缓存的设备指针：插件崩溃时 releasePluginChannels // 会注销设备，恢复后重新注册 —— 缓存指针会一直指向死设备， // 现象是"插件重启后 rt 再也说不出话"。 + // ★ 此处置 pump **不需要加锁**：rc 还没被任何 goroutine 看到 + //（rtLoop 在 SpawnRTChild 之后才启动）。之后的读/清一律走 rc.mu —— + // 见 Dispose 与 rtEmit 里的说明。 rc.pump = newOutputPump(opts.Output, func() agentIO.Device { if a.io == nil { return nil @@ -1272,8 +1288,13 @@ func (a *Agent) rtEmit(rc *rtChild, text string) { // 原实现直接 `dev.Execute(...)`，插件侧延迟（TTS/网络编码/字幕渲染） // **直接节流模型的 token 消费** —— 「不攒整段」的收益全被吃掉。 // pump.Push 立即返回，交付由 pump 的消费 goroutine 负责。 - if rc.pump != nil { - rc.pump.Push(text) + // ★ 锁内取、锁外推（见 Dispose 的说明）： + // Push 可能阻塞（队列满时），持锁会卡住 Dispose 的 pump 置 nil。 + rc.mu.Lock() + pump := rc.pump + rc.mu.Unlock() + if pump != nil { + pump.Push(text) return } // 兜底（pump 未建时，如某些判据直接构造 rtChild）：保持原同步行为， diff --git a/internal/agent/core/rtchild_pump_lock_test.go b/internal/agent/core/rtchild_pump_lock_test.go new file mode 100644 index 00000000..0eb06d6b --- /dev/null +++ b/internal/agent/core/rtchild_pump_lock_test.go @@ -0,0 +1,135 @@ +package core + +// 判据：rtChild.pump 的读写必须走 rc.mu。 +// +// ## 守的是什么（2026-11-04 实测 data race） +// +// pump 是 rtChild 里**唯一**一个"构造后被改写、又有并发读者"的字段： +// +// Dispose（StopRTChild 的 goroutine） 写 rc.pump = nil +// rtEmit （rtLoop 的 goroutine） 读 rc.pump +// +// 两者无同步 ⇒ -race 报 DATA RACE，栈是： +// +// Read at … core.(*Agent).rtEmit rtchild.go:1275 +// Previous write … core.(*rtChild).Dispose rtchild.go:377 +// +// ## 为什么必须有这条判据（e2e 不够） +// +// TestRTChild_E2E_Sustained 覆盖了同一个窗口，但它是**偶发命中**： +// 实测全量负载下 5 次里中 1 次，单跑 20 次 0 次。这种判据在 CI 里 +// 等于没有 —— 它红了是运气，绿了也不代表安全。 +// +// 所以这里直接构造**最高并发压力**：一边不停 rtEmit，一边 Dispose， +// 用 -race 判定。它不依赖时序巧合。 +// +// ## 判据的边界（诚实说明） +// +// 这条判据的"红"依赖 -race 检测器。不带 -race 时它只能发现 +// panic / 死锁，不能发现数据竞争。所以必须与 `go test -race` 一起跑 +// —— 本仓的提交纪律第 4 条已经要求全量 -race，这里只是把 +// 「哪个窗口最危险」钉成可复跑的形态。 + +import ( + "sync" + "testing" + "time" + + agentIO "github.com/JianFeeeee/HologramAgent/internal/agent/io" +) + +// TestRTPumpAccessIsLocked 用并发压力 + -race 钉住加锁。 +// +// 变异：把 rtEmit 里的 rc.mu.Lock/Unlock 去掉（直接读 rc.pump） +// 或把 Dispose 里的置 nil 移出锁 ⇒ 本判据在 -race 下红。 +func TestRTPumpAccessIsLocked(t *testing.T) { + dev := &mockOutputDevice{ + name: "rtpumpdev", + caps: agentIO.CapText, + toolFn: func(string, map[string]interface{}) (interface{}, error) { + return "ok", nil + }, + } + rc := &rtChild{ + Child: nil, // 本判据只碰 pump，不经过骨架 + output: "rtpumpdev", + } + // 给 rtEmit 用的 Agent（它读 a.io 拿设备）。 + a := New(AgentConfig{ + ID: "rtpumpagent", + IO: agentIO.NewIOManager(), + StageHost: NewStageHost(), + }) + if err := a.io.RegisterDevice(dev); err != nil { + t.Fatalf("注册设备: %v", err) + } + rc.agent = a + rc.mu.Lock() + rc.pump = newOutputPump("rtpumpdev", func() agentIO.Device { + return a.io.GetDevice("rtpumpdev") + }) + rc.mu.Unlock() + + // ★ 读者必须**跑满整个写者窗口**，不能跑固定次数就先结束。 + // + // 第一版是 `for i := 0; i < 500; i++`：500 次 emit 在微秒级就跑完， + // 而写者 sleep 1ms 才动手 ⇒ 两者根本不重叠。实测变异验证时， + // 把 Dispose 的置 nil 移出锁（裸写）**连跑 6 次都不红** —— 判据是假的。 + // + // 现在改成"持续跑到写者宣布结束"，保证读写真正交错。 + stop := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + a.rtEmit(rc, "x") + } + } + }() + + // 写者：等读者进入循环后 Dispose（pump 置 nil + Close）。 + time.Sleep(2 * time.Millisecond) + rc.Dispose() + // Dispose 之后再放一小段：此时读者仍可能读 rc.pump（应读到 nil 或被锁挡住）， + // 这一小段正是"裸写会与读者撞车"的窗口。 + time.Sleep(2 * time.Millisecond) + close(stop) + wg.Wait() +} + +// TestRTPumpAfterDisposeIsNil 钉住 Dispose 之后 emit 不再推送到已关的 pump。 +// +// 这是加锁的**语义**副作用验证：只加锁但先 Close 后置 nil， +// 读者仍可能拿到一个已关闭的 pump 去 Push。顺序必须"取走并置 nil， +// 再在锁外 Close"。 +func TestRTPumpAfterDisposeIsNil(t *testing.T) { + a := New(AgentConfig{ + ID: "rtpumpagent2", + IO: agentIO.NewIOManager(), + StageHost: NewStageHost(), + }) + dev := &mockOutputDevice{name: "d2", caps: agentIO.CapText} + if err := a.io.RegisterDevice(dev); err != nil { + t.Fatalf("注册设备: %v", err) + } + rc := &rtChild{agent: a, output: "d2"} + rc.mu.Lock() + rc.pump = newOutputPump("d2", func() agentIO.Device { return a.io.GetDevice("d2") }) + rc.mu.Unlock() + + rc.Dispose() + + rc.mu.Lock() + p := rc.pump + rc.mu.Unlock() + if p != nil { + t.Fatal("★ Dispose 之后 rc.pump 仍非 nil —— emit 会推送到已关闭的 pump") + } + // 兜底路径必须能工作（不 panic）。 + a.rtEmit(rc, "after-dispose") +}