fix: 写队列满时 Close 不再与 enqueue 死锁
This commit is contained in:
Nixevol
@@ -1728,3 +1728,13 @@ issue #3 未关闭，`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是
|
||||
- 原因：终态回执、超限后的后续 pending、撤回腾出窗口后都依赖再唤醒，否则空闲在线端收不到。
|
||||
- 备选方案：在同一次 `PushPending` 里循环扫完整 pending 表（否决，指令要求合并唤醒下一轮）。
|
||||
- 影响：仅 `internal/app/message/`；相关单测见 `review_r3_65_test.go`。
|
||||
|
||||
### 复审修复 R3-03
|
||||
|
||||
1. **写队列满时 Close 不再与 enqueue 死锁**
|
||||
- 日期：2026-09-30
|
||||
- 原条款：issue #67；停机需 Drain/Close 在超时内返回；已入队任务仍由写 goroutine 处理完。
|
||||
- 实际做法：Queue 增加 closing 信号。enqueue 在持 sendMu.RLock 时 select 同时等待 q.ch、closing 与 ctx.Done()，通道满时不再无限阻塞。Close 先 Swap(closed) 并 close(closing) 唤醒在途发送方释放读锁，再在写锁内 close(q.ch)，最后等 loop 退出。不向已关闭 channel 发送。
|
||||
- 原因：通道满时并发 Do 占着读锁堵在发送上，Close 的写锁拿不到，停机卡死；未提交的在途写也会丢。
|
||||
- 备选方案：入队改为非阻塞，满则立即 ErrBusy（否决：改变背压语义，正常高峰会误伤提交）。
|
||||
- 影响：仅 internal/store/queue.go 与测试；不改迁移编号。
|
||||
|
||||
+15
-2
@@ -38,6 +38,7 @@ type Queue struct {
|
||||
|
||||
ch chan writeJob
|
||||
done chan struct{}
|
||||
closing chan struct{} // Close 时关闭，唤醒持读锁阻塞在发送上的 enqueue
|
||||
closed atomic.Bool
|
||||
sendMu sync.RWMutex
|
||||
|
||||
@@ -52,10 +53,15 @@ type Queue struct {
|
||||
|
||||
// NewQueue 创建合并写入队列并启动写 goroutine。
|
||||
func NewQueue(db *sql.DB) *Queue {
|
||||
return newQueue(db, queueBuffSize)
|
||||
}
|
||||
|
||||
func newQueue(db *sql.DB, buffSize int) *Queue {
|
||||
q := &Queue{
|
||||
db: db,
|
||||
ch: make(chan writeJob, queueBuffSize),
|
||||
ch: make(chan writeJob, buffSize),
|
||||
done: make(chan struct{}),
|
||||
closing: make(chan struct{}),
|
||||
ready: true,
|
||||
}
|
||||
go q.loop()
|
||||
@@ -115,10 +121,15 @@ func (q *Queue) enqueue(job writeJob) error {
|
||||
return ErrQueueClosed
|
||||
}
|
||||
q.addPending(1)
|
||||
// 通道满时不得只堵在发送上持有读锁：Close 需要写锁关闭 q.ch。
|
||||
select {
|
||||
case q.ch <- job:
|
||||
q.sendMu.RUnlock()
|
||||
return nil
|
||||
case <-q.closing:
|
||||
q.addPending(-1)
|
||||
q.sendMu.RUnlock()
|
||||
return ErrQueueClosed
|
||||
case <-job.ctx.Done():
|
||||
q.addPending(-1)
|
||||
q.sendMu.RUnlock()
|
||||
@@ -376,11 +387,13 @@ func (q *Queue) Drain(ctx context.Context) error {
|
||||
}
|
||||
|
||||
// Close 关闭队列：不再接受新任务，并等待写 goroutine 处理完已入队任务后退出。
|
||||
// 在写锁内关闭数据通道，避免并发 Do 向已关闭 channel 发送而 panic。
|
||||
// 先关闭 closing 唤醒因通道满而阻塞的发送方并释放读锁，再在无发送者时关闭数据通道，
|
||||
// 避免向已关闭 channel 发送而 panic，也避免与持读锁的 enqueue 死锁。
|
||||
func (q *Queue) Close() error {
|
||||
if q.closed.Swap(true) {
|
||||
return nil
|
||||
}
|
||||
close(q.closing)
|
||||
q.sendMu.Lock()
|
||||
close(q.ch)
|
||||
q.sendMu.Unlock()
|
||||
|
||||
@@ -289,3 +289,137 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.upd
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// openSmallQueue 用小缓冲队列替换默认队列，便于测满通道时的 Close/Drain。
|
||||
func openSmallQueue(t *testing.T, buf int) *DB {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
db, err := Open(dir, "FULL")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Queue.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db.Queue = newQueue(db.Write, buf)
|
||||
return db
|
||||
}
|
||||
|
||||
func TestQueueCloseUnblocksFullChannel(t *testing.T) {
|
||||
t.Parallel()
|
||||
const buf = 4
|
||||
db := openSmallQueue(t, buf)
|
||||
defer func() { _ = db.Close() }()
|
||||
q := db.Queue
|
||||
|
||||
hold := make(chan struct{})
|
||||
blockerStarted := make(chan struct{})
|
||||
blockerErr := make(chan error, 1)
|
||||
go func() {
|
||||
blockerErr <- q.Do(context.Background(), func(tx *sql.Tx) error {
|
||||
close(blockerStarted)
|
||||
<-hold
|
||||
return nil
|
||||
})
|
||||
}()
|
||||
<-blockerStarted
|
||||
|
||||
ctx := context.Background()
|
||||
var fillWG sync.WaitGroup
|
||||
for i := 0; i < buf; i++ {
|
||||
fillWG.Add(1)
|
||||
go func() {
|
||||
defer fillWG.Done()
|
||||
_ = q.Do(ctx, func(tx *sql.Tx) error { return nil })
|
||||
}()
|
||||
}
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for q.Len() < buf+1 && time.Now().Before(deadline) {
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
}
|
||||
if q.Len() < buf+1 {
|
||||
t.Fatalf("channel not full: len=%d", q.Len())
|
||||
}
|
||||
|
||||
blockedErr := make(chan error, 1)
|
||||
go func() {
|
||||
blockedErr <- q.Do(ctx, func(tx *sql.Tx) error { return nil })
|
||||
}()
|
||||
// 等额外 Do 堵在 enqueue（pending 超过通道容量+正在执行的一条）。
|
||||
deadline = time.Now().Add(2 * time.Second)
|
||||
for q.Len() < buf+2 && time.Now().Before(deadline) {
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
}
|
||||
|
||||
closeDone := make(chan error, 1)
|
||||
go func() { closeDone <- q.Close() }()
|
||||
|
||||
select {
|
||||
case err := <-blockedErr:
|
||||
if !errors.Is(err, ErrQueueClosed) {
|
||||
t.Fatalf("blocked Do: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Close did not unblock full-channel enqueue")
|
||||
}
|
||||
|
||||
close(hold)
|
||||
select {
|
||||
case err := <-closeDone:
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Close hung after writer released")
|
||||
}
|
||||
<-blockerErr
|
||||
fillWG.Wait()
|
||||
}
|
||||
|
||||
func TestQueueDrainTimeoutWhileWriterBlocked(t *testing.T) {
|
||||
t.Parallel()
|
||||
const buf = 4
|
||||
db := openSmallQueue(t, buf)
|
||||
defer func() { _ = db.Close() }()
|
||||
q := db.Queue
|
||||
|
||||
hold := make(chan struct{})
|
||||
started := make(chan struct{})
|
||||
go func() {
|
||||
_ = q.Do(context.Background(), func(tx *sql.Tx) error {
|
||||
close(started)
|
||||
<-hold
|
||||
return nil
|
||||
})
|
||||
}()
|
||||
<-started
|
||||
|
||||
ctx := context.Background()
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < buf; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
_ = q.Do(ctx, func(tx *sql.Tx) error { return nil })
|
||||
}()
|
||||
}
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for q.Len() < buf+1 && time.Now().Before(deadline) {
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
}
|
||||
|
||||
drainCtx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
err := q.Drain(drainCtx)
|
||||
elapsed := time.Since(start)
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("Drain err=%v want deadline", err)
|
||||
}
|
||||
if elapsed > 500*time.Millisecond {
|
||||
t.Fatalf("Drain took %s, should return on timeout", elapsed)
|
||||
}
|
||||
|
||||
close(hold)
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user