1 Commits
Author SHA1 Message Date
chiyi 81d36f827e fix(stream): OnClose 可在返回前注册,消除 Inject 循环等待死锁
旧实现 OnClose 先等待 ready,但 Inject 只有业务方法返回后才发生:业务同步 OnClose → 等 Inject;框架等业务返回 → 永久死锁,流首帧永不输出。改为仅持锁注册回调,不等待 channel 注入;已关闭场景锁外立即回调,避免重入自锁。新增两条回归测试。
2026-09-16 13:54:00 +08:00
2 changed files with 54 additions and 3 deletions
+8 -3
View File
@@ -67,14 +67,19 @@ func (s *Stream) Close() {
close(s.ch) close(s.ch)
} }
// OnClose 注册关闭回调;流已关闭时立即执行 // OnClose 注册关闭回调;可在业务方法 return *Stream 之前同步调用。
// 不得等待 Inject:Inject 只有业务方法返回后才发生,等待会形成循环依赖死锁。
// 流已关闭时在锁外立即执行回调,避免回调重入 Stream 时自锁。
func (s *Stream) OnClose(cb func()) { func (s *Stream) OnClose(cb func()) {
<-s.getReady() if cb == nil {
return
}
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock()
if s.closed { if s.closed {
s.mu.Unlock()
cb() cb()
return return
} }
s.onClose = cb s.onClose = cb
s.mu.Unlock()
} }
+46
View File
@@ -0,0 +1,46 @@
package fun
import (
"sync/atomic"
"testing"
"time"
)
// OnClose 必须允许业务方法在 return *Stream 之前同步注册;旧实现等待 Inject,
// 而 Inject 只有 invoke 在业务方法返回后才调用,形成精确死锁:
//
// business OnClose -> wait ready -> invoke wait business return -> never Inject.
func TestStreamOnCloseBeforeInjectDoesNotBlock(t *testing.T) {
st := &Stream{}
registered := make(chan struct{})
var called atomic.Int32
go func() {
st.OnClose(func() { called.Add(1) })
close(registered)
}()
select {
case <-registered:
// expected: registration is independent from channel injection
case <-time.After(time.Second):
t.Fatal("OnClose blocked before Inject (business method would deadlock)")
}
ch := make(chan any)
done := make(chan struct{})
st.Inject(ch, done)
st.Close()
if got := called.Load(); got != 1 {
t.Fatalf("callback called %d times, want 1", got)
}
}
func TestStreamOnCloseAfterAlreadyClosedRunsImmediately(t *testing.T) {
st := &Stream{}
st.Inject(make(chan any), make(chan struct{}))
st.Close()
var called atomic.Int32
st.OnClose(func() { called.Add(1) })
if got := called.Load(); got != 1 {
t.Fatalf("callback called %d times, want 1", got)
}
}