TsDB2/DB/core/endtoend_concurrent_read_test.go

package core

import (
	"path/filepath"
	"sort"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
)

// TestEndToEnd_ConcurrentReads 端到端验证并发读是否真正生效。
//
// 方案: 开真实库, 建表, 插 5000 行数据, 然后同时 N 个 goroutine 做全表 ScanWithPKID。
// 每个 Scan 本身耗时 ~X ms。如果是串行锁, N 个 goroutine 总耗时 ≈ N×X ms;
// 如果 RLock 真正并行, 总耗时应该接近 ≈ X ms (或略多, 远小于 N×X)。
func TestEndToEnd_ConcurrentReads(t *testing.T) {
	tmpDir := t.TempDir()
	dbPath := filepath.Join(tmpDir, "concurrent_test.db")

	db, err := Open(dbPath)
	if err != nil {
		t.Fatalf("Open: %v", err)
	}
	defer db.Close()

	// 建表
	tbl, err := db.CreateTable("items", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Dynamicbyte, FieldName: []byte("name")},
		{FieldType: types.FieldType_Int64, FieldName: []byte("v")},
	})
	if err != nil {
		t.Fatalf("CreateTable: %v", err)
	}
	// 主键索引 (独立项目需要手动建, DockerManager 版本在 CreateTable 内自动建)
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("CreateIndex pk: %v", err)
	}

	// 批量插 5000 行
	txn := db.BeginTransaction()
	for i := 0; i < 5000; i++ {
		pk := tbl.AllocPKID()
		name := []byte("item_" + itoa(i))
		if _, err := tbl.InsertRowWithID(pk, []types.FieldValue{
			{FieldName: []byte("id"), Value: pk},
			{FieldName: []byte("name"), Value: name},
			{FieldName: []byte("v"), Value: int64(i)},
		}); err != nil {
			txn.Rollback()
			t.Fatalf("InsertRowWithID(%d): %v", i, err)
		}
	}
	if err := txn.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
	t.Logf("插入 5000 行完成")

	// 验证 accMu 的 readers 计数: 临时注入, RLock 进来 +1, RUnlock -1
	// 如果并发, readers 应该同时 > 1
	t.Logf("=== 并发读验证 (临时 log) ===")

	// 串行基线: 单个 goroutine Scan 5 次取平均
	baselineN := 3
	baseline := time.Duration(0)
	for i := 0; i < baselineN; i++ {
		start := time.Now()
		c, e := fullScanCount(tbl)
		if e != nil {
			t.Fatalf("Scan baseline %d: %v", i, e)
		}
		if c != 5000 {
			t.Fatalf("Scan baseline: got %d want 5000", c)
		}
		baseline += time.Since(start)
	}
	baseline /= time.Duration(baselineN)
	t.Logf("单次全表 Scan 基线: %v", baseline)

	// 并发 8 个 goroutine 同时 Scan —— 测 BufferPool 内部锁的竞争
	readers := 8

	// ====== 阶段 1: 预热 ======
	// 先 8 个 goroutine 各跑 2 次全表 Scan, 把 B+ 树节点 + 数据页全读进 BufferPool
	t.Logf("=== 预热: 8 goroutine × 2 次全表 Scan ===")
	for warm := 0; warm < 2; warm++ {
		var wg sync.WaitGroup
		wg.Add(readers)
		for i := 0; i < readers; i++ {
			go func() {
				defer wg.Done()
				fullScanCount(tbl)
			}()
		}
		wg.Wait()
	}
	t.Logf("预热完成, BufferPool 应该热了")

	// ====== 阶段 2: 热缓存并发读 ======
	t.Logf("=== 热缓存并发读 (预期大部分 cache-hit) ===")
	start := time.Now()
	var wg2 sync.WaitGroup
	errs := make([]error, readers)
	counts := make([]int, readers)
	startTimes := make([]time.Time, readers)
	endTimes := make([]time.Time, readers)
	wg2.Add(readers)
	var activeReaders int32
	var maxActive int32
	for i := 0; i < readers; i++ {
		go func(idx int) {
			defer wg2.Done()
			startTimes[idx] = time.Now()
			cur := atomic.AddInt32(&activeReaders, 1)
			for {
				m := atomic.LoadInt32(&maxActive)
				if cur <= m || atomic.CompareAndSwapInt32(&maxActive, m, cur) {
					break
				}
			}
			c, e := fullScanCount(tbl)
			counts[idx] = c
			errs[idx] = e
			atomic.AddInt32(&activeReaders, -1)
			endTimes[idx] = time.Now()
		}(i)
	}
	wg2.Wait()
	elapsed := time.Since(start)

	t.Logf("同时在跑的最大 goroutine 数: %d (如果 = 8 说明真并行)", atomic.LoadInt32(&maxActive))
	startDeltas := make([]time.Duration, readers)
	for i := 0; i < readers; i++ {
		startDeltas[i] = startTimes[i].Sub(start)
	}
	sort.Slice(startDeltas, func(i, j int) bool { return startDeltas[i] < startDeltas[j] })
	t.Logf("8 个 goroutine 开始时间分布 (最早→最晚):")
	t.Logf("  最早: %v", startDeltas[0])
	t.Logf("  最晚: %v", startDeltas[readers-1])
	t.Logf("  开始时间离散度: %v", startDeltas[readers-1]-startDeltas[0])

	for i, e := range errs {
		if e != nil {
			t.Fatalf("reader %d error: %v", i, e)
		}
	}
	for i, c := range counts {
		if c != 5000 {
			t.Fatalf("reader %d scan: got %d want 5000", i, c)
		}
	}

	speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
	t.Logf("单次全表 Scan 基线: %v", baseline)
	t.Logf("串行预估 (baseline × 8): %v", baseline*time.Duration(readers))
	t.Logf("8 读者耗时 (热缓存): %v", elapsed)
	t.Logf("加速比 (串行/实际): %.2fx", speedup)

	if speedup > 1.5 {
		t.Logf("✅ 热缓存并发读真的并行了! (加速比 > 1.5x)")
	} else {
		t.Logf("⚠️ 加速比 %.2fx, 可能瓶颈不在 BufferPool 锁 (磁盘 IO/Go 调度等)", speedup)
	}
}

// fullScanCount 全表扫描返回行数 (验证 scanWithPKID 的 RLock 并行)
func fullScanCount(tbl *Table) (int, error) {
	q := tbl.Query()
	count := 0
	err := q.ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
		count++
		return true
	})
	return count, err
}

// itoa 无分配 int→string
func itoa(i int) string {
	if i == 0 {
		return "0"
	}
	neg := i < 0
	if neg {
		i = -i
	}
	var buf [20]byte
	pos := len(buf)
	for i > 0 {
		pos--
		buf[pos] = byte('0' + i%10)
		i /= 10
	}
	if neg {
		pos--
		buf[pos] = '-'
	}
	return string(buf[pos:])
}

// TestAccMu_RLockParallel_EndToEnd 端到端验证 accMu RLock 真能让多个 Scan 并行
// 直接用 db.lockRead() + 模拟 Scan 工作 (sleep), 看 8 个 goroutine 是不是真并行
func TestAccMu_RLockParallel_EndToEnd(t *testing.T) {
	tmpDir := t.TempDir()
	dbPath := filepath.Join(tmpDir, "test.db")

	db, err := Open(dbPath)
	if err != nil {
		t.Fatalf("Open: %v", err)
	}
	defer db.Close()

	const readers = 8
	const hold = 30 * time.Millisecond // 每个 RLock 持有 30ms (模拟 Scan 耗时)

	// baseline: 串行拿 RLock
	baselineN := 5
	var baseline time.Duration
	for i := 0; i < baselineN; i++ {
		start := time.Now()
		unlock := db.lockRead()
		time.Sleep(hold)
		unlock()
		baseline += time.Since(start)
	}
	baseline /= time.Duration(baselineN)
	t.Logf("串行 RLock+sleep baseline: %v", baseline)

	// 8 个 goroutine 并发拿 RLock
	start := time.Now()
	var wg sync.WaitGroup
	wg.Add(readers)
	for i := 0; i < readers; i++ {
		go func() {
			defer wg.Done()
			unlock := db.lockRead()
			time.Sleep(hold)
			unlock()
		}()
	}
	wg.Wait()
	elapsed := time.Since(start)

	speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
	t.Logf("8 并发 RLock+sleep: %v", elapsed)
	t.Logf("串行预估: %v", baseline*time.Duration(readers))
	t.Logf("加速比: %.2fx", speedup)

	if speedup < 4.0 {
		t.Errorf("accMu RLock 并发加速比 %.2fx < 4x! 说明有别的锁串住了 Scan", speedup)
	} else {
		t.Logf("accMu RLock 真并行! 加速比 %.2fx", speedup)
	}
}

// TestScanConcurrencyLevels 测不同并发数下的加速比
func TestScanConcurrencyLevels(t *testing.T) {
	tmpDir := t.TempDir()
	dbPath := filepath.Join(tmpDir, "levels.db")

	db, err := Open(dbPath)
	if err != nil {
		t.Fatalf("Open: %v", err)
	}
	defer db.Close()

	tbl, err := db.CreateTable("items", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Dynamicbyte, FieldName: []byte("name")},
		{FieldType: types.FieldType_Int64, FieldName: []byte("v")},
	})
	if err != nil {
		t.Fatalf("CreateTable: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("CreateIndex pk: %v", err)
	}

	txn := db.BeginTransaction()
	for i := 0; i < 5000; i++ {
		pk := tbl.AllocPKID()
		name := []byte("item_" + itoa(i))
		if _, err := tbl.InsertRowWithID(pk, []types.FieldValue{
			{FieldName: []byte("id"), Value: pk},
			{FieldName: []byte("name"), Value: name},
			{FieldName: []byte("v"), Value: int64(i)},
		}); err != nil {
			txn.Rollback()
			t.Fatalf("InsertRowWithID(%d): %v", i, err)
		}
	}
	if err := txn.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}

	// baseline
	baselineN := 3
	var baseline time.Duration
	for i := 0; i < baselineN; i++ {
		start := time.Now()
		fullScanCount(tbl)
		baseline += time.Since(start)
	}
	baseline /= time.Duration(baselineN)
	t.Logf("单次 Scan baseline: %v", baseline)

	// 预热
	for i := 0; i < 3; i++ {
		fullScanCount(tbl)
	}

	for _, readers := range []int{1, 2, 4, 6, 8} {
		start := time.Now()
		var wg sync.WaitGroup
		wg.Add(readers)
		for i := 0; i < readers; i++ {
			go func() {
				defer wg.Done()
				fullScanCount(tbl)
			}()
		}
		wg.Wait()
		elapsed := time.Since(start)
		speedup := float64(baseline*time.Duration(readers)) / float64(elapsed)
		perReader := float64(elapsed) / float64(time.Duration(readers))
		baselineNs := float64(baseline) / 1e6
		t.Logf("%d readers: total %v, per-reader %.1fms (baseline %.1fms), speedup %.2fx",
			readers, elapsed, perReader, baselineNs, speedup)
	}
}