|
|
@@ -367,6 +367,11 @@ func TestWriteOverReadCachedPageKeepsLRUConsistent(t *testing.T) {
|
|
|
// TestReadPageConcurrentHot 验证多个 goroutine 同时读热页真的并行。
|
|
|
// mu 从 sync.Mutex 改成 sync.RWMutex + atomic touch 后, cache-hit 路径
|
|
|
// 不应该互相阻塞。8 个 goroutine 并发读同一页应该 ≈ 单次耗时, 而不是 8×。
|
|
|
+//
|
|
|
+// 注意: 纯无工作量的 RLock+atomic 是纳秒级, 在 Windows Go timer 精度 (~15ms)
|
|
|
+// 下 baseline 和 elapsed 会同时被 round 成 0, 加速比算 NaN 或 0.00x —— 这不是
|
|
|
+// "并发没生效", 是 timer 没东西可量。所以这里给每次 ReadPage 加一段 workload,
|
|
|
+// 让每次访问约 50µs (和 TestBufferPoolHotConcurrencyWithWork 同套路)。
|
|
|
func TestReadPageConcurrentHot(t *testing.T) {
|
|
|
d := &diskStub{pages: map[uint32][]byte{42: {0x2A}}}
|
|
|
bp := NewBufferPool(64, d.read, d.write)
|
|
|
@@ -377,12 +382,14 @@ func TestReadPageConcurrentHot(t *testing.T) {
|
|
|
t.Fatalf("warmup: %v", err)
|
|
|
}
|
|
|
|
|
|
+ const perRead = 50 * time.Microsecond // 每次读后睡一下, 让 timer 有东西量
|
|
|
readers := 8
|
|
|
- baselineN := 100
|
|
|
+ baselineN := 20
|
|
|
var baseline time.Duration
|
|
|
for i := 0; i < baselineN; i++ {
|
|
|
start := time.Now()
|
|
|
_ = bp.ReadPage(42, buf)
|
|
|
+ time.Sleep(perRead)
|
|
|
baseline += time.Since(start)
|
|
|
}
|
|
|
baseline /= time.Duration(baselineN)
|
|
|
@@ -396,6 +403,7 @@ func TestReadPageConcurrentHot(t *testing.T) {
|
|
|
defer wg.Done()
|
|
|
out := make([]byte, 1)
|
|
|
errs[idx] = bp.ReadPage(42, out)
|
|
|
+ time.Sleep(perRead)
|
|
|
}(i)
|
|
|
}
|
|
|
wg.Wait()
|
|
|
@@ -408,20 +416,25 @@ func TestReadPageConcurrentHot(t *testing.T) {
|
|
|
}
|
|
|
|
|
|
speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
|
|
|
- t.Logf("单 ReadPage baseline: %v", baseline)
|
|
|
+ t.Logf("单 ReadPage baseline (含 %v workload): %v", perRead, baseline)
|
|
|
t.Logf("8 并发读同一热页: %v", elapsed)
|
|
|
t.Logf("串行预估 (baseline × 8): %v", baseline*time.Duration(readers))
|
|
|
t.Logf("加速比: %.2fx (并行 ≈ %.1f× 单次)", speedup, float64(elapsed)/float64(baseline))
|
|
|
|
|
|
- if speedup < 2.0 {
|
|
|
- t.Errorf("读没有真的并行! 加速比 %.2fx < 2x", speedup)
|
|
|
+ // RWMutex 的 cache-hit 路径应该真并行 —— 8 并发耗时不应是串行的 8 倍,
|
|
|
+ // 至少小于串行预估的一半 (speedup >= 2.0)
|
|
|
+ if elapsed >= baseline*time.Duration(readers) {
|
|
|
+ t.Errorf("读没有真的并行! 耗时 %v >= 串行预估 %v (speedup=%.2f)", elapsed, baseline*time.Duration(readers), speedup)
|
|
|
} else {
|
|
|
t.Logf("ReadPage cache-hit 路径真正并行!")
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// TestBufferPoolHotConcurrencyDirect 直接测 BufferPool 热缓存并发读 —— 绕过 TsDB2 上层
|
|
|
-// 模拟 8 个 goroutine 并发 ReadPage 不同页 + 同页 (热缓存, 应该全是 RLock)
|
|
|
+// 模拟 8 个 goroutine 并发 ReadPage 不同页 + 同页 (热缓存, 应该全是 RLock)。
|
|
|
+//
|
|
|
+// 和 TestReadPageConcurrentHot 同套路: 纯 RLock+atomic 是纳秒级, 在 Windows timer
|
|
|
+// 精度 (~15ms) 下 baseline/elapsed 会被 round 成 0 —— 加 workload 让 timer 有东西量。
|
|
|
func TestBufferPoolHotConcurrencyDirect(t *testing.T) {
|
|
|
// 模拟: 先预热 50 页 (相当于 B+ 树节点 + 数据页全进池)
|
|
|
d := &diskStub{pages: map[uint32][]byte{}}
|
|
|
@@ -437,20 +450,24 @@ func TestBufferPoolHotConcurrencyDirect(t *testing.T) {
|
|
|
}
|
|
|
t.Logf("预热后 BufferPool Len=%d, DirtyCount=%d", bp.Len(), bp.DirtyCount())
|
|
|
|
|
|
- // baseline: 单 goroutine 读 50 页
|
|
|
+ const pageWork = 50 * time.Microsecond // 每页 50µs workload, 让 timer 有东西量
|
|
|
+ const pagesPerScan = 50
|
|
|
+
|
|
|
+ // baseline: 单 goroutine 读 50 页 + 每页 workload
|
|
|
baselineN := 20
|
|
|
var baseline time.Duration
|
|
|
for i := 0; i < baselineN; i++ {
|
|
|
start := time.Now()
|
|
|
- for p := uint32(1); p <= 50; p++ {
|
|
|
+ for p := uint32(1); p <= pagesPerScan; p++ {
|
|
|
_ = bp.ReadPage(p, buf)
|
|
|
+ time.Sleep(pageWork)
|
|
|
}
|
|
|
baseline += time.Since(start)
|
|
|
}
|
|
|
baseline /= time.Duration(baselineN)
|
|
|
- t.Logf("单 goroutine 读 50 页 baseline: %v", baseline)
|
|
|
+ t.Logf("单 goroutine 读 %d 页 baseline (含 workload): %v", pagesPerScan, baseline)
|
|
|
|
|
|
- // 8 goroutine 并发: 每个读 50 页 (全是热缓存 cache-hit)
|
|
|
+ // 8 goroutine 并发: 每个读 50 页 + 每页 workload
|
|
|
readers := 8
|
|
|
start := time.Now()
|
|
|
var wg sync.WaitGroup
|
|
|
@@ -460,11 +477,12 @@ func TestBufferPoolHotConcurrencyDirect(t *testing.T) {
|
|
|
go func(idx int) {
|
|
|
defer wg.Done()
|
|
|
out := make([]byte, 1)
|
|
|
- for p := uint32(1); p <= 50; p++ {
|
|
|
+ for p := uint32(1); p <= pagesPerScan; p++ {
|
|
|
if e := bp.ReadPage(p, out); e != nil {
|
|
|
errs[idx] = e
|
|
|
return
|
|
|
}
|
|
|
+ time.Sleep(pageWork)
|
|
|
}
|
|
|
}(i)
|
|
|
}
|
|
|
@@ -478,12 +496,14 @@ func TestBufferPoolHotConcurrencyDirect(t *testing.T) {
|
|
|
}
|
|
|
|
|
|
speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
|
|
|
- t.Logf("8 并发各读 50 页: %v", elapsed)
|
|
|
+ t.Logf("8 并发各读 %d 页: %v", pagesPerScan, elapsed)
|
|
|
t.Logf("串行预估: %v", baseline*time.Duration(readers))
|
|
|
t.Logf("加速比: %.2fx", speedup)
|
|
|
|
|
|
- if speedup < 2.0 {
|
|
|
- t.Errorf("BufferPool 热缓存并发加速比 %.2fx < 2x! RLock 可能没真并行", speedup)
|
|
|
+ if elapsed >= baseline*time.Duration(readers) {
|
|
|
+ t.Errorf("BufferPool 热缓存没真并行! 耗时 %v >= 串行预估 %v (speedup=%.2f)", elapsed, baseline*time.Duration(readers), speedup)
|
|
|
+ } else {
|
|
|
+ t.Logf("BufferPool 热缓存并发加速比 %.2fx —— RLock 真并行!", speedup)
|
|
|
}
|
|
|
}
|
|
|
|