mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-05 16:57:32 +00:00
fix(rtchild): pump 裸读写 data race —— 锁贯彻到底
## 缺陷（-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 全绿
This commit is contained in:
JianFeeeee
@ -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）：保持原同步行为，
|
||||
|
||||
135
internal/agent/core/rtchild_pump_lock_test.go
Normal file
135
internal/agent/core/rtchild_pump_lock_test.go
Normal file
@ -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")
|
||||
}
|
||||
Reference in New Issue
Block a user