TsDB2/demo/rw_concurrency/main.go

package main

// =============================================================================
// rw_concurrency — 读写并发实测
//
// 目的: 用真实运行回答两个问题
//   1. 读和写是不是真并发? (读者能否在写事务进行中完成扫描, 而不是干等)
//   2. 并发读者读到的是什么? 是事务前的一致快照(固定 20000 行), 还是随写者
//      进度不断变化、未提交的"中间态"(行数在 20000~60000 之间跳), 甚至 panic?
//
// 布局:
//   - 先在一个事务里插 20000 行并提交 (基线);
//   - 写者开一个长事务, 再插 40000 行 (pk 20001..60000), 中途不提交;
//   - 4 个读者在写事务期间反复全表扫描 + 随机点查, 记录行数 / 错误 / panic。
// =============================================================================

import (
	"fmt"
	"os"
	"path/filepath"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

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

const (
	baseRows  = 20000 // 基线已提交行数
	writeRows = 40000 // 长事务里新插行数 (未提交)
	readers   = 4
)

func main() {
	dir, err := os.MkdirTemp("", "rwconc-")
	if err != nil {
		panic(err)
	}
	defer os.RemoveAll(dir)
	path := filepath.Join(dir, "t.db")

	db, err := core.Open(path)
	if err != nil {
		panic(err)
	}

	tbl, err := db.CreateTable("emp", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("No")},
		{FieldType: types.FieldType_Float64, FieldName: []byte("Score")},
		{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(24), FieldName: []byte("Name")},
	})
	if err != nil {
		panic(err)
	}
	// 主键 B+ 树(键=行号 pkID): 建了它 Query 全表扫描 / FindByPK 才能定位行。
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		panic(err)
	}

	// ---- 基线: 一个事务插 baseRows 行后提交 ----
	tx := db.BeginTransaction()
	for i := 1; i <= baseRows; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("No"), Value: uint64(i)},
			{FieldName: []byte("Score"), Value: float64(i)},
			{FieldName: []byte("Name"), Value: fmt.Sprintf("name-%d", i)},
		}); err != nil {
			panic(err)
		}
	}
	if err := tx.Commit(); err != nil {
		panic(err)
	}

	// ---- 统计量 ----
	var (
		scans      atomic.Int64
		overlap    atomic.Int64 // 与写事务窗口重叠的扫描次数
		snapshot   atomic.Int64 // 恰好读到基线 baseRows 行 (事务前快照)
		dirty      atomic.Int64 // 读到 baseRows < n < 总数 (未提交中间态)
		readErrors atomic.Int64
		panics     atomic.Int64
		pkErrors   atomic.Int64
	)
	var mu sync.Mutex
	minN, maxN := int(1<<30), 0

	// 写事务窗口 (unixnano, 0=尚未开始)
	var wStart, wEnd atomic.Int64

	stop := make(chan struct{})
	var readersWG sync.WaitGroup

	oneScan := func() (n int, panicked bool, err error) {
		defer func() {
			if r := recover(); r != nil {
				panicked = true
				err = fmt.Errorf("panic: %v", r)
			}
		}()
		err = tbl.Query().ScanWithPKID(func(_ uint64, _ []types.FieldValue) bool {
			n++
			return true
		})
		return
	}

	for r := 0; r < readers; r++ {
		readersWG.Add(1)
		go func(id int) {
			defer readersWG.Done()
			for {
				select {
				case <-stop:
					return
				default:
				}
				t0 := time.Now().UnixNano()
				n, panicked, err := oneScan()
				if panicked {
					panics.Add(1)
				}
				if err != nil {
					readErrors.Add(1)
				}
				ws, we := wStart.Load(), wEnd.Load()
				inWindow := ws != 0 && ws <= t0 && (we == 0 || t0 <= we)

				// 随机点查一个基线行, 验证结构是否可解析
				if _, gerr := tbl.FindByPK(uint64((id*7919 + int(scans.Load()))%baseRows + 1)); gerr != nil {
					pkErrors.Add(1)
				}

				if n > 0 || err == nil {
					scans.Add(1)
					if inWindow {
						overlap.Add(1)
					}
					switch {
					case n == baseRows:
						snapshot.Add(1)
					case n > baseRows && n < baseRows+writeRows:
						dirty.Add(1)
					}
					mu.Lock()
					if n < minN {
						minN = n
					}
					if n > maxN {
						maxN = n
					}
					mu.Unlock()
				}
			}
		}(r)
	}

	// ---- 写者: 长事务插入 writeRows 行 ----
	time.Sleep(300 * time.Millisecond) // 让读者先跑起来
	wStart.Store(time.Now().UnixNano())
	wtx := db.BeginTransaction()
	for i := baseRows + 1; i <= baseRows+writeRows; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("No"), Value: uint64(i)},
			{FieldName: []byte("Score"), Value: float64(i)},
			{FieldName: []byte("Name"), Value: fmt.Sprintf("name-%d", i)},
		}); err != nil {
			panic(err)
		}
		if (i-baseRows)%4000 == 0 {
			time.Sleep(5 * time.Millisecond) // 拉长中间态窗口, 便于读者撞上
		}
	}
	if err := wtx.Commit(); err != nil {
		panic(err)
	}
	wEnd.Store(time.Now().UnixNano())

	time.Sleep(300 * time.Millisecond) // 提交后再让读者跑一会
	close(stop)
	readersWG.Wait()

	finalN, _, _ := oneScan()
	_ = db.Close()

	fmt.Println("================ 读写并发实测结果 ================")
	fmt.Printf("CPU 核数 GOMAXPROCS = %d\n", runtime.GOMAXPROCS(0))
	fmt.Printf("写事务窗口约 %d ms (插入 %d 行, 未提交期间读者并发扫描)\n",
		(wEnd.Load()-wStart.Load())/1_000_000, writeRows)
	fmt.Println("------------------------------------------------")
	fmt.Printf("读者完成全表扫描总次数      : %d\n", scans.Load())
	fmt.Printf("  其中与写事务窗口【重叠】  : %d   <- 生命周期在墙上时钟重叠(实际在锁上排队, 见判定)\n", overlap.Load())
	fmt.Printf("  读到事务前快照(%d行)     : %d\n", baseRows, snapshot.Load())
	fmt.Printf("  读到未提交中间态(行数跳)  : %d   <- 行数在 %d~%d 之间, 无快照隔离\n",
		dirty.Load(), baseRows, baseRows+writeRows)
	fmt.Printf("扫描行数范围               : min=%d  max=%d\n", minN, maxN)
	fmt.Printf("扫描返回错误次数           : %d\n", readErrors.Load())
	fmt.Printf("扫描 panic 次数            : %d   (结构解析越界等硬损坏)\n", panics.Load())
	fmt.Printf("点查 FindByPK 错误次数     : %d\n", pkErrors.Load())
	fmt.Println("------------------------------------------------")
	fmt.Printf("写事务提交后最终行数       : %d (应为 %d)\n", finalN, baseRows+writeRows)

	// 判定
	fmt.Println("\n【判定】")
	if dirty.Load() == 0 && minN == baseRows && maxN == baseRows+writeRows {
		fmt.Println("  - 无脏读: 写事务未提交期间, 读者在全局串行锁(accMu)上排队, 读不到",
			fmt.Sprintf("%d~%d 的中间态;", baseRows, baseRows+writeRows))
		fmt.Printf("    扫描结果只有两种 —— 写事务拿锁前读到事务前状态(%d), 提交后读到完整结果(%d)。\n",
			baseRows, baseRows+writeRows)
	}
	if overlap.Load() > 0 {
		fmt.Printf("  - 注: %d 次扫描的生命周期与写窗口在墙上时钟重叠, 但它们在锁上排队, 并未读到中间态。\n", overlap.Load())
	}
	if dirty.Load() > 0 || panics.Load() > 0 || readErrors.Load() > 0 {
		fmt.Println("  - 仍存在脏读 / panic / 错误: 串行锁未生效, 属缺陷。")
		os.Exit(1)
	}
}