TsDB2/DB/core/access_lock_test.go

package core

import (
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// TestAccessLock_ReadReadParallel 验证多个读者真正并行 (不互相阻塞)。
// 原理: 10 个 goroutine 同时 RLock, 每个 sleep 200ms, 总耗时应 ~200ms 而非 ~2000ms。
func TestAccessLock_ReadReadParallel(t *testing.T) {
	var l accessLock
	const readers = 10
	const hold = 200 * time.Millisecond

	start := time.Now()
	var wg sync.WaitGroup
	wg.Add(readers)
	for i := 0; i < readers; i++ {
		go func() {
			defer wg.Done()
			l.RLock()
			time.Sleep(hold)
			l.RUnlock()
		}()
	}
	wg.Wait()
	elapsed := time.Since(start)

	if elapsed >= hold*2 { // 给 2x 余量, 超过说明有串行化
		t.Errorf("读读应该并行, 但耗时 %v (期望 < %v)", elapsed, hold*2)
	}
	t.Logf("10 个读者并行耗时: %v", elapsed)
}

// TestAccessLock_WriteBlocksRead 验证写锁阻塞读。
func TestAccessLock_WriteBlocksRead(t *testing.T) {
	var l accessLock
	const hold = 200 * time.Millisecond
	const waitTolerance = hold / 2

	// goroutine A: 拿写锁, sleep 200ms
	doneA := make(chan struct{})
	go func() {
		l.Lock()
		time.Sleep(hold)
		l.Unlock()
		close(doneA)
	}()

	// 等 goroutine A 真正拿到锁
	time.Sleep(20 * time.Millisecond)

	// goroutine B: 尝试 RLock, 应该被阻塞, 直到 A 释放
	startB := time.Now()
	l.RLock()
	elapsedB := time.Since(startB)
	l.RUnlock()

	if elapsedB < hold-waitTolerance {
		t.Errorf("写锁应该阻塞读, 但 B 只等了 %v (期望 >= %v)", elapsedB, hold-waitTolerance)
	}
	t.Logf("读等写释放耗时: %v", elapsedB)

	<-doneA
}

// TestAccessLock_ReentrantWrite 验证同 goroutine 写锁重入不卡死。
func TestAccessLock_ReentrantWrite(t *testing.T) {
	var l accessLock
	const depth = 10

	// 同 goroutine 连续 Lock 10 次, 再 Unlock 10 次
	for i := 0; i < depth; i++ {
		l.Lock()
	}
	for i := 0; i < depth; i++ {
		l.Unlock()
	}
	t.Logf("写锁 %d 层重入正常", depth)
}

// TestAccessLock_ReentrantRLockUnderWrite 验证同 goroutine 持有写锁时再 RLock 不卡死。
// 这是 TsDB2 特有的关键路径: BeginTransaction (Lock) → InsertRow → scanWithPKID (RLock)
func TestAccessLock_ReentrantRLockUnderWrite(t *testing.T) {
	var l accessLock

	l.Lock()
	l.RLock() // 同 goroutine, 已持写锁, 应该跳过
	l.RUnlock()
	l.Unlock()
	t.Logf("同 goroutine 写锁下 RLock 正常")
}

// TestAccessLock_WriteWaitsForReaders 验证写锁等待现有读者离开。
func TestAccessLock_WriteWaitsForReaders(t *testing.T) {
	var l accessLock
	const hold = 200 * time.Millisecond
	const tolerance = hold / 2

	// goroutine A: RLock, sleep 200ms
	doneA := make(chan struct{})
	go func() {
		l.RLock()
		time.Sleep(hold)
		l.RUnlock()
		close(doneA)
	}()

	// 等 A 拿到 RLock
	time.Sleep(20 * time.Millisecond)

	// goroutine B: 尝试 Lock, 应该等 A 的 RUnlock
	startB := time.Now()
	l.Lock()
	elapsedB := time.Since(startB)
	l.Unlock()

	if elapsedB < hold-tolerance {
		t.Errorf("写锁应该等待读者, 但 B 只等了 %v (期望 >= %v)", elapsedB, hold-tolerance)
	}
	t.Logf("写锁等待读者耗时: %v", elapsedB)

	<-doneA
}

// TestAccessLock_DeadlockStress 长时间压力测试, 确保没有死锁。
// 10 个读者 goroutine + 2 个写者 goroutine, 混合跑 3 秒。
func TestAccessLock_DeadlockStress(t *testing.T) {
	var l accessLock
	var reads, writes int64

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

	// 读者: RLock → 少量工作 → RUnlock
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-stop:
					return
				default:
					l.RLock()
					atomic.AddInt64(&reads, 1)
					l.RUnlock()
				}
			}
		}()
	}

	// 写者: Lock → 少量工作 → Unlock
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-stop:
					return
				default:
					l.Lock()
					atomic.AddInt64(&writes, 1)
					l.Unlock()
				}
			}
		}()
	}

	// 跑 3 秒后停止
	time.Sleep(3 * time.Second)
	close(stop)

	done := make(chan struct{})
	go func() { wg.Wait(); close(done) }()

	select {
	case <-done:
		r := atomic.LoadInt64(&reads)
		w := atomic.LoadInt64(&writes)
		t.Logf("3 秒压力测试通过, reads=%d writes=%d", r, w)
	case <-time.After(2 * time.Second):
		t.Fatal("压力测试超时 —— 疑似死锁!")
	}
}