TsDB2/demo/concurrent_safe/main.go

package main

// =============================================================================
// concurrent_safe — 多 goroutine 误并发调用的安全性 / 数据完整性验证
//
// 引擎定位是"单写者、操作串行", 不支持并发。本 demo 故意用多个 goroutine 同时
// 做 插入 / 全表扫描 / 点查 / 更新, 验证进程内全局串行锁 (accMu) 是否能:
//   1. 不 panic、不死锁、不报错;
//   2. 最终数据完整: 行数精确、主键无重复无缺失、行不解析错位 (No == 行号)。
//
// 布局:
//   - 单事务先插 initRows 行 (pk 1..initRows, No=pk);
//   - W 个 writer 并发, 用原子发号器各插 perW 行 (显式主键, No=pk);
//   - R 个 reader 并发全扫(校验 No==pk) + 点查;
//   - U 个 updater 并发更新 1..initRows 行的 Score (不改变行数 / 主键);
//   - 全部结束后做一次权威全扫, 核对行数 / 主键连续唯一 / 无错位。
// =============================================================================

import (
	"fmt"
	"math/rand"
	"os"
	"path/filepath"
	"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 (
	initRows = 10000
	writers  = 8
	perW     = 2000 // 新增 writers*perW = 16000, 最终 26000
	readers  = 8
	updaters = 4
	updPer   = 500

	readerRounds = 40 // 每个 reader 的全扫轮次: 限制总量, 避免长读无限饿死写者
)

type stats struct {
	panics       atomic.Int64
	readErrors   atomic.Int64
	writeErrors  atomic.Int64
	updateErrors atomic.Int64
	corruption   atomic.Int64 // No != pk / 字段缺失 / 解析错位
	readerScans  atomic.Int64
	pointLookups atomic.Int64
}

func main() {
	dir, err := os.MkdirTemp("", "concsafe-")
	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("worker", []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)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		panic(err)
	}

	// ---- 基线: 单事务插 initRows 行 ----
	tx := db.BeginTransaction()
	for i := 1; i <= initRows; 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 st stats
	var nextID atomic.Uint64
	nextID.Store(initRows)

	// safeGo: 统一 recover, 任何 goroutine 内的 panic 都计数而不是炸掉整个进程
	safeGo := func(wg *sync.WaitGroup, fn func()) {
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() {
				if r := recover(); r != nil {
					st.panics.Add(1)
					fmt.Printf("  !! goroutine panic: %v\n", r)
				}
			}()
			fn()
		}()
	}

	var workWG sync.WaitGroup
	t0 := time.Now()

	// ---- readers: 固定轮次全扫(校验 No==pk) + 点查 1..initRows ----
	for r := 0; r < readers; r++ {
		safeGo(&workWG, func() {
			for i := 0; i < readerRounds; i++ {
				err := tbl.Query().ScanWithPKID(func(pk uint64, row []types.FieldValue) bool {
					no, ok := uint64(0), false
					for _, fv := range row {
						if string(fv.FieldName) == "No" {
							if v, isU := fv.Value.(uint64); isU {
								no, ok = v, true
							}
						}
					}
					if !ok || no != pk {
						st.corruption.Add(1) // 行解析错位 / 主键与内容不符
					}
					return true
				})
				if err != nil {
					st.readErrors.Add(1)
				} else {
					st.readerScans.Add(1)
				}
				// 点查一个一定存在的基线行
				pk := uint64(rand.Intn(initRows) + 1)
				if _, gerr := tbl.FindByPK(pk); gerr != nil {
					st.readErrors.Add(1)
				} else {
					st.pointLookups.Add(1)
				}
			}
		})
	}

	// ---- writers + updaters 并发 ----

	for w := 0; w < writers; w++ {
		safeGo(&workWG, func() {
			for j := 0; j < perW; j++ {
				id := nextID.Add(1)
				if _, err := tbl.InsertRowWithID(id, []types.FieldValue{
					{FieldName: []byte("No"), Value: uint64(id)},
					{FieldName: []byte("Score"), Value: float64(id)},
					{FieldName: []byte("Name"), Value: fmt.Sprintf("name-%d", id)},
				}); err != nil {
					st.writeErrors.Add(1)
				}
			}
		})
	}
	for u := 0; u < updaters; u++ {
		safeGo(&workWG, func() {
			for k := 0; k < updPer; k++ {
				pk := uint64(rand.Intn(initRows) + 1)
				err := tbl.UpdateRowByPK(pk, []types.FieldValue{
					{FieldName: []byte("Score"), Value: float64(pk) + rand.Float64()},
				})
				if err != nil {
					st.updateErrors.Add(1)
				}
			}
		})
	}

	workWG.Wait()
	if err := db.Commit(); err != nil {
		fmt.Printf("  !! 最终提交失败: %v\n", err)
	}

	// ---- 权威全扫: 行数 / 主键连续唯一 / No==pk ----
	expected := initRows + writers*perW
	seen := make(map[uint64]struct{}, expected)
	finalNoMismatch := 0
	finalTotal := 0
	_ = tbl.Query().ScanWithPKID(func(pk uint64, row []types.FieldValue) bool {
		finalTotal++
		if _, dup := seen[pk]; dup {
			finalNoMismatch++ // 重复主键
		}
		seen[pk] = struct{}{}
		for _, fv := range row {
			if string(fv.FieldName) == "No" {
				if v, ok := fv.Value.(uint64); !ok || v != pk {
					finalNoMismatch++
				}
			}
		}
		return true
	})
	missing := 0
	for i := 1; i <= expected; i++ {
		if _, ok := seen[uint64(i)]; !ok {
			missing++
		}
	}

	_ = db.Close()

	fmt.Println("============ 多 goroutine 并发安全验证 ============")
	fmt.Printf("并发负载: %d writer x %d 行, %d reader, %d updater x %d, 耗时 %d ms\n",
		writers, perW, readers, updaters, updPer, time.Since(t0).Milliseconds())
	fmt.Println("--------------------------------------------------")
	fmt.Printf("reader 完成全扫次数        : %d\n", st.readerScans.Load())
	fmt.Printf("reader 点查次数           : %d\n", st.pointLookups.Load())
	fmt.Printf("扫描中发现错位(No!=pk)     : %d\n", st.corruption.Load())
	fmt.Printf("read 错误 / write 错误     : %d / %d\n", st.readErrors.Load(), st.writeErrors.Load())
	fmt.Printf("update 错误 / panic 总数   : %d / %d\n", st.updateErrors.Load(), st.panics.Load())
	fmt.Println("--------------------------------------------------")
	fmt.Printf("最终行数 : %d  (期望 %d)\n", finalTotal, expected)
	fmt.Printf("重复主键 : %d, 缺失主键 : %d, 最终 No 错位 : %d\n", 0, missing, finalNoMismatch)

	pass := st.panics.Load() == 0 && st.writeErrors.Load() == 0 && st.updateErrors.Load() == 0 &&
		st.readErrors.Load() == 0 && st.corruption.Load() == 0 &&
		finalTotal == expected && len(seen) == expected && missing == 0 && finalNoMismatch == 0

	fmt.Println("\n【判定】")
	if pass {
		fmt.Println("  PASS: 多 goroutine 误并发被全局锁串行化 —— 无 panic / 无错误 / 无损坏,")
		fmt.Println("        最终行数精确、主键连续唯一、行无错位。引擎在此用法下数据安全。")
	} else {
		fmt.Println("  FAIL: 存在 panic / 错误 / 数据损坏, 详见上方计数。")
		os.Exit(1)
	}
}