dailz
408138b3c8
Put/Delete checked writeStopped then called queue.Submit (which sends on a channel). Close set writeStopped then closed the channel. With no synchronization between the check and Submit, a concurrent Close could close the channel during the window, causing "send on closed channel" panic in Put/Delete. The CommitQueue.Close comment said "Callers must stop submitting before Close" but Put/Delete didn't enforce this. Changes: - wal/writer.go: add submitMu sync.RWMutex. Put/Delete construct entry outside the lock, then hold RLock during writeStopped check + Submit; release before <-req.Result. Close holds write lock during writeStopped.Store + queue.Close, guaranteeing no Submit is in progress when the channel is closed. - wal/writer_test.go: add TestConcurrentPutCloseNoPanic and TestConcurrentDeleteCloseNoPanic. 100 goroutines + 1 closer, recover panics. Run with -count=50 for regression detection. Verified: all existing tests pass. New concurrent tests pass with -count=50 (0 panics). go test -race ./... green. Audit context: docs/audit-3.2.md C7.
340 lines
10 KiB
Markdown
340 lines
10 KiB
Markdown
# C7 修复方案：Put/Delete vs Close 并发竞态
|
||
|
||
## TL;DR
|
||
|
||
> **目标**：消除 Put/Delete 检查 `writeStopped` 和 `Submit` 之间的时间窗，防止 Close 并发关闭 channel 时 send-on-closed panic。
|
||
>
|
||
> **交付**：
|
||
> - `WalWriter` 加 `submitMu sync.RWMutex`
|
||
> - Put/Delete 持读锁保护 check+Submit 临界区
|
||
> - Close 持写锁保护 writeStopped+queue.Close
|
||
> - 2 个并发测试（Put+Close、Delete+Close）
|
||
> - 单次 commit
|
||
>
|
||
> **预估工时**：1.5-2 小时
|
||
> **风险**：低。改动局限在 `WalWriter` 一个文件，3 个函数
|
||
|
||
---
|
||
|
||
## Context
|
||
|
||
### Bug 摘要
|
||
|
||
`wal/writer.go:313-326` Put（Delete 同理 328-341）：
|
||
|
||
```go
|
||
func (ww *WalWriter) Put(key, value []byte) error {
|
||
if ww.writeStopped.Load() { // ① 检查
|
||
return errkit.ErrWriteStopped
|
||
}
|
||
req := ww.queue.Submit(...) // ② Submit（内部 `cq.ch <- req`）
|
||
result := <-req.Result
|
||
return result.Err
|
||
}
|
||
```
|
||
|
||
`wal/writer.go:369-382` Close：
|
||
|
||
```go
|
||
func (ww *WalWriter) Close() error {
|
||
ww.closeOnce.Do(func() {
|
||
ww.writeStopped.Store(true) // A 设标志
|
||
ww.queue.Close() // B 关 channel
|
||
close(ww.done)
|
||
ww.wg.Wait()
|
||
...
|
||
})
|
||
}
|
||
```
|
||
|
||
`wal/commit_queue.go:30-37` Submit：
|
||
|
||
```go
|
||
func (cq *CommitQueue) Submit(entries []*WalEntry) *CommitRequest {
|
||
req := &CommitRequest{...}
|
||
cq.ch <- req // ← send on possibly-closed channel
|
||
return req
|
||
}
|
||
```
|
||
|
||
`CommitQueue.Close` 的注释**自己写着** "Callers must stop submitting before Close" —— 但 Put/Delete 并没有保证这一点。
|
||
|
||
### 竞态场景
|
||
|
||
```
|
||
Goroutine 1 (Put) Goroutine 2 (Close)
|
||
──────────────── ────────────────────
|
||
writeStopped.Load() = false
|
||
writeStopped.Store(true)
|
||
queue.Close() → close(cq.ch)
|
||
queue.Submit():
|
||
cq.ch <- req ← PANIC
|
||
"send on closed channel"
|
||
```
|
||
|
||
时间窗：Put 的 ①→② 之间，Close 的 A→B 发生。
|
||
|
||
### 为什么没被抓到
|
||
|
||
现有测试串行调用 Put → Close，不模拟并发。要抓 C7 需要：
|
||
- 多 goroutine 并发 Put + 1 goroutine Close
|
||
- `defer recover` 捕获 panic
|
||
- 高并发 + 多次运行提高触发概率
|
||
|
||
属于"低概率高破坏"型 —— 平时不发作，发作就 panic 整个进程。
|
||
|
||
---
|
||
|
||
## 执行计划
|
||
|
||
### Phase A：代码改动（30 分钟）
|
||
|
||
#### A.1 加 `submitMu sync.RWMutex` 到 WalWriter
|
||
|
||
```go
|
||
type WalWriter struct {
|
||
...
|
||
writeStopped atomic.Bool
|
||
submitMu sync.RWMutex // C7 fix: 保护 Submit vs Close 临界区
|
||
done chan struct{}
|
||
...
|
||
}
|
||
```
|
||
|
||
#### A.2 修改 Put
|
||
|
||
```go
|
||
// 改后：
|
||
func (ww *WalWriter) Put(key, value []byte) error {
|
||
// Construct entry outside the lock to minimize critical section
|
||
// (Oracle nice-to-have: reduces contention under high concurrency).
|
||
entry := &WalEntry{
|
||
OpType: OpPut,
|
||
ValueKind: VKInline,
|
||
Key: cloneBytes(key),
|
||
Value: cloneBytes(value),
|
||
}
|
||
|
||
// C7 fix: RLock 保护 check + Submit 临界区。Close 持写锁时，
|
||
// 所有 RLock 释放后才能 queue.Close，保证 Submit 不会 send on closed channel。
|
||
ww.submitMu.RLock()
|
||
if ww.writeStopped.Load() {
|
||
ww.submitMu.RUnlock()
|
||
return errkit.ErrWriteStopped
|
||
}
|
||
req := ww.queue.Submit([]*WalEntry{entry})
|
||
ww.submitMu.RUnlock()
|
||
|
||
result := <-req.Result
|
||
return result.Err
|
||
}
|
||
```
|
||
|
||
**关键**：RLock 在 Submit 之后、`<-req.Result` 之前释放。这样：
|
||
- Submit 在锁保护内（防止 Close 并发关 channel）
|
||
- `<-req.Result` 在锁外（不阻塞其他 Put，不阻塞 Close 拿写锁）
|
||
|
||
#### A.3 修改 Delete
|
||
|
||
同 Put 模式。
|
||
|
||
#### A.4 修改 Close
|
||
|
||
```go
|
||
// 改后：
|
||
func (ww *WalWriter) Close() error {
|
||
ww.closeOnce.Do(func() {
|
||
// C7 fix: 写锁等待所有 Put/Delete 的 RLock 释放。
|
||
// 持锁期间关 channel 是安全的：没有 Submit 在进行中。
|
||
ww.submitMu.Lock()
|
||
ww.writeStopped.Store(true)
|
||
ww.queue.Close()
|
||
ww.submitMu.Unlock()
|
||
|
||
close(ww.done)
|
||
ww.wg.Wait()
|
||
if err := ww.segManager.Sync(); err != nil {
|
||
ww.closeErr = err
|
||
return
|
||
}
|
||
ww.closeErr = ww.segManager.Close()
|
||
})
|
||
return ww.closeErr
|
||
}
|
||
```
|
||
|
||
**关键顺序**：
|
||
1. 写锁（等所有 in-flight Submit 完成）
|
||
2. `writeStopped = true`（新 Put 看到后返回错误）
|
||
3. `queue.Close`（此刻无 Submit 在进行，安全）
|
||
4. 写锁释放
|
||
5. `close(done)` + `wg.Wait`（runLoop 退出）
|
||
|
||
#### 死锁分析
|
||
|
||
| 场景 | 分析 |
|
||
|------|------|
|
||
| Put 持 RLock + Submit 阻塞（channel 满）| runLoop 持续消费（不持任何锁），Submit 最终完成，Put 释放 RLock |
|
||
| Close 等写锁 | 等 RLock 释放；RLock 释放依赖 Submit 完成；Submit 完成依赖 runLoop 消费；runLoop 不受锁影响 → 最终完成 |
|
||
| processBatch 失败后 writeStopped=true | 不影响 runLoop 继续消费 channel，不持任何锁，无死锁 |
|
||
|
||
**无死锁**。Close 可能因 Submit 阻塞而延迟（等 I/O），但最终完成。这是 graceful shutdown 的正确语义 —— **保留 backpressure，不创建锁环**。
|
||
|
||
### Phase B：测试（45 分钟）
|
||
|
||
#### B.1 TestConcurrentPutCloseNoPanic
|
||
|
||
新增到 `wal/writer_test.go`：
|
||
|
||
```go
|
||
// Regression guard for C7: concurrent Put + Close must not panic.
|
||
// Buggy code has a race window between writeStopped check and Submit's
|
||
// send on channel; Close closing the channel during that window causes
|
||
// "send on closed channel" panic.
|
||
func TestConcurrentPutCloseNoPanic(t *testing.T) {
|
||
dir := t.TempDir()
|
||
cfg := config.Defaults()
|
||
ww, err := NewWalWriter(&cfg, dir, 0, 0)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
|
||
const goroutines = 100
|
||
var wg sync.WaitGroup
|
||
var panicCount atomic.Int32
|
||
start := make(chan struct{})
|
||
|
||
for i := 0; i < goroutines; i++ {
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
panicCount.Add(1)
|
||
}
|
||
}()
|
||
<-start
|
||
_ = ww.Put([]byte("k"), []byte("v"))
|
||
}()
|
||
}
|
||
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
<-start
|
||
_ = ww.Close()
|
||
}()
|
||
|
||
close(start)
|
||
wg.Wait()
|
||
|
||
if panicCount.Load() > 0 {
|
||
t.Fatalf("concurrent Put + Close caused %d panic(s) — C7 regression",
|
||
panicCount.Load())
|
||
}
|
||
}
|
||
```
|
||
|
||
#### B.2 TestConcurrentDeleteCloseNoPanic
|
||
|
||
同 B.1 但调 `ww.Delete([]byte("k"))`。
|
||
|
||
#### B.3 测试稳定性说明
|
||
|
||
> **Oracle 修订（bg_96db47f5）**：测试定位是"fix 后**保证不 panic**"，不是"fix 前**必 panic**"。buggy 代码的 race 窗口非常窄（Put check → Submit 之间几条指令），默认 queue capacity=10000，100 goroutines 通常不足以触发 channel 满阻塞。`-count=50` 能提高触发概率但不保证每次都抓到。
|
||
>
|
||
> 建议用 `go test -race ./wal -run TestConcurrent -count=10` 作为可选补充。确定性测试需要在 check 和 Submit 之间注入 test seam（过度侵入性，Phase 1 不值得）。
|
||
|
||
### Phase C：验证（15 分钟）
|
||
|
||
```bash
|
||
# 1. 编译
|
||
go build ./...
|
||
|
||
# 2. 并发测试（单次）
|
||
go test ./wal -run 'TestConcurrent' -count=1 -v
|
||
|
||
# 3. 并发测试（多次，抓 flaky race）
|
||
go test ./wal -run 'TestConcurrent' -count=50
|
||
|
||
# 4. wal 包全量
|
||
go test ./wal/... -count=1
|
||
|
||
# 5. 全仓
|
||
go test ./... -count=1
|
||
|
||
# 6. race
|
||
go test -race ./... -count=1
|
||
|
||
# 7. vet
|
||
go vet ./...
|
||
```
|
||
|
||
### Phase D：Commit message draft
|
||
|
||
```
|
||
fix: prevent send-on-closed-channel panic in concurrent Put/Delete + Close (C7)
|
||
|
||
Put/Delete checked writeStopped then called queue.Submit (which sends on
|
||
a channel). Close set writeStopped then closed the channel. With no
|
||
synchronization between the check and Submit, a concurrent Close could
|
||
close the channel during the window, causing "send on closed channel"
|
||
panic in Put/Delete.
|
||
|
||
The CommitQueue.Close comment said "Callers must stop submitting before
|
||
Close" but Put/Delete didn't enforce this.
|
||
|
||
Changes:
|
||
- wal/writer.go: add submitMu sync.RWMutex. Put/Delete hold RLock during
|
||
writeStopped check + Submit; release before <-req.Result. Close holds
|
||
write lock during writeStopped.Store + queue.Close, guaranteeing no
|
||
Submit is in progress when the channel is closed.
|
||
- wal/writer_test.go: add TestConcurrentPutCloseNoPanic and
|
||
TestConcurrentDeleteCloseNoPanic. 100 goroutines + 1 closer, recover
|
||
panics. On buggy code, this test triggers "send on closed channel"
|
||
panic (may require -count=N for reliable reproduction due to narrow
|
||
race window). On fixed code, never panics.
|
||
|
||
Verified: all existing tests pass. New concurrent tests pass with
|
||
-count=50. go test -race ./... green.
|
||
|
||
Audit context: docs/audit-3.2.md C7.
|
||
```
|
||
|
||
---
|
||
|
||
## 验收清单
|
||
|
||
- [ ] Phase A.1：`submitMu sync.RWMutex` 字段存在
|
||
- [ ] Phase A.2：Put 在 RLock 内 check + Submit，RUnlock 后 `<-req.Result`
|
||
- [ ] Phase A.3：Delete 同 Put 模式
|
||
- [ ] Phase A.4：Close 在写锁内 writeStopped + queue.Close
|
||
- [ ] Phase B.1：`TestConcurrentPutCloseNoPanic` 存在
|
||
- [ ] Phase B.2：`TestConcurrentDeleteCloseNoPanic` 存在
|
||
- [ ] `go test ./wal/... -count=1` 全绿
|
||
- [ ] `go test ./wal -run TestConcurrent -count=50` 全绿
|
||
- [ ] `go test -race ./... -count=1` 全绿
|
||
- [ ] `go vet ./...` 无新增警告
|
||
- [ ] 单次 commit，message 引用 audit C7
|
||
|
||
---
|
||
|
||
## 不在本次范围内
|
||
|
||
| 项 | 为什么不放进来 |
|
||
|------|---------------|
|
||
| CommitQueue API 改动 | 不需要。RWMutex 在 WalWriter 层就够 |
|
||
| DB 层并发保护 | 不需要。WalWriter 的保护对 DB 层透明生效（db.Put 调 writer.Put，writer 的 submitMu 已经保护）|
|
||
| processBatch 失败后的 channel 状态 | 不影响。writeStopped 后 runLoop 继续消费，不关 channel |
|
||
|
||
---
|
||
|
||
## 修订记录
|
||
|
||
- **v1（原始）**：C7 修复方案初稿，送 Momus 审
|
||
- **v1.0（Momus 审核 bg_66fea405）**：[OKAY]，无 blocking
|
||
- **v1.1（Oracle 审核 bg_96db47f5）**：**approve**（无 blocking）。3 个 nice-to-have 已采纳：
|
||
- 缩小临界区：WalEntry 构造移到 RLock 之前
|
||
- 死锁分析补充"保留 backpressure，不创建锁环"
|
||
- 测试定位澄清：是"fix 后保证不 panic"，不是"fix 前必 panic"
|