并发修复

DB/core/endtoend_concurrent_read_test.go

@@ -0,0 +1,154 @@
+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:])
+}

DB/driver/driver.go

@@ -42,7 +42,7 @@ var registry = struct {
 type shared struct {
 	db   *core.DB
 	dsn  string
-	sem  chan struct{} // 容量1: 全局串行 (引擎单写者); 事务持有至 Commit/Rollback
+	sem  chan struct{} // 容量16: 允许多个读者并行; Begin/Commit 的写锁由 core.accMu 排他保证
 	refs int
 }
 
@@ -121,7 +121,7 @@ func sharedFor(dsn string) (*shared, error) {
 		if err != nil {
 			return nil, err
 		}
-		return &shared{db: db, dsn: dsn, refs: 1, sem: make(chan struct{}, 1)}, nil
+		return &shared{db: db, dsn: dsn, refs: 1, sem: make(chan struct{}, 16)}, nil
 	}
 	registry.Lock()
 	defer registry.Unlock()
@@ -133,7 +133,7 @@ func sharedFor(dsn string) (*shared, error) {
 	if err != nil {
 		return nil, err
 	}
-	s := &shared{db: db, dsn: dsn, refs: 1, sem: make(chan struct{}, 1)}
+	s := &shared{db: db, dsn: dsn, refs: 1, sem: make(chan struct{}, 16)}
 	registry.m[dsn] = s
 	return s, nil
 }
@@ -655,5 +655,5 @@ type result struct {
 	affected int64
 }
 
-func (r *result) LastInsertId() (int64, error)  { return r.lastID, nil }
+func (r *result) LastInsertId() (int64, error) { return r.lastID, nil }
 func (r *result) RowsAffected() (int64, error) { return r.affected, nil }