TsDB2/DB/core/endtoend_concurrent_read_test.go

package core

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

	"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)
	}

	// 批量插 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 行完成")

	// 串行基线: 单个 goroutine Scan 5 次取平均
	baselineN := 5
	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
	readers := 8
	start := time.Now()
	var wg sync.WaitGroup
	errs := make([]error, readers)
	counts := make([]int, readers)
	wg.Add(readers)
	for i := 0; i < readers; i++ {
		go func(idx int) {
			defer wg.Done()
			c, e := fullScanCount(tbl)
			counts[idx] = c
			errs[idx] = e
		}(i)
	}
	wg.Wait()
	elapsed := time.Since(start)

	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)
		}
	}

	// 判定: 真正并行时 elapsed 应 < baseline * readers * 40%
	serialEstimate := baseline * time.Duration(readers)
	parallelThreshold := serialEstimate * 40 / 100

	t.Logf("=== 并发读验证结果 ===")
	t.Logf("读者 goroutine 数: %d", readers)
	t.Logf("单次 Scan 基线:     %v", baseline)
	t.Logf("串行预估耗时:       %v (baseline × readers)", serialEstimate)
	t.Logf("实际并发耗时:       %v", elapsed)
	t.Logf("并行判定阈值:       %v (串行预估 × 40%%)", parallelThreshold)

	if elapsed > parallelThreshold {
		t.Errorf("❌ 读没有真正并行! 耗时 %v > 阈值 %v (串行预估 %v)",
			elapsed, parallelThreshold, serialEstimate)
	} else {
		speedup := float64(serialEstimate) / float64(elapsed)
		t.Logf("✅ 读真正并行! 加速比 %.1fx (串行/实际)", 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:])
}