并发竞态测试：数据竞争检测、确定性复现、TSan/Loom 与调度扰动

同一段并发代码，每次执行可能走不同的交错：
  · CPU 调度：线程何时被抢占
  · 内存可见性：写何时对其他核可见（缓存一致性）
  · 编译器/CPU 重排：指令顺序被优化
  · 系统负载：高负载时交错更密集，更易暴露

结论：并发缺陷的"复现"本质上是"控制随机性"。
  单核机器几乎测不出，多核 + 高负载才容易触发。

# 编译时启用，运行时自动检测
go test -race ./...
go run -race main.go

# 输出示例：
# ==================
# WARNING: DATA RACE
# Write at 0x00c0000b4018 by goroutine 7:
#   main.(*Counter).Inc()
#       counter.go:12 +0x38
# Previous read at 0x00c0000b4018 by goroutine 8:
#   main.(*Counter).Get()
#       counter.go:18 +0x44
# ==================

func TestCounterConcurrent(t *testing.T) {
    c := &Counter{}
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(2)
        go func() { defer wg.Done(); c.Inc() }()
        go func() { defer wg.Done(); _ = c.Get() }()
    }
    wg.Wait()
}
// go test -race 会捕获 Inc 与 Get 之间的无同步竞争

// 修复：用原子操作或互斥锁
type Counter struct {
    mu sync.Mutex
    n  int64
}
func (c *Counter) Inc() { c.mu.Lock(); c.n++; c.mu.Unlock() }
func (c *Counter) Get() int64 { c.mu.Lock(); defer c.mu.Unlock(); return c.n }

# 编译时插桩
clang++ -fsanitize=thread -g -O1 counter.cpp -o counter
./counter

# CMake 集成
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread -g -O1")

# Rust：TSan 需要 nightly 与 -Zsanitizer
# .cargo/config.toml
[target.x86_64-unknown-linux-gnu]
rustflags = ["-Zsanitizer=thread"]

RUSTFLAGS="-Zsanitizer=thread" cargo +nightly test --target x86_64-unknown-linux-gnu

# Java：用 jcstress 做并发正确性测试（OpenJDK 官方工具）
mvn test -Dtest=CounterStressTest

@JCStressTest
@Outcome(id = "100", expect = Expect.ACCEPTABLE, desc = "正确累加")
@Outcome(expect = Expect.FORBIDDEN, desc = "丢失更新")
@State
public class CounterStress {
    private int n;
    @Actor public void actor1() { for (int i = 0; i < 50; i++) n++; }
    @Actor public void actor2() { for (int i = 0; i < 50; i++) n++; }
}

复现三要素：
  1. 固定种子：随机 sleep/调度都从同一种子派生
  2. 记录交错：把线程切换点记录下来，可回放
  3. 注入延迟：在关键点插入 sleep 放大竞争窗口

// 在被测代码里留出可注入的 hook
var randDelayHook func()

func transfer(from, to *Account, amount int) {
    if randDelayHook != nil { randDelayHook() }   // 注入点：读取与写入之间
    if from.balance >= amount {
        from.balance -= amount
        to.balance += amount
    }
}

func TestTransferRace(t *testing.T) {
    rand := mrand.New(mrand.NewSource(42))   // 固定种子，可复现
    randDelayHook = func() {
        if rand.Intn(2) == 0 { time.Sleep(time.Duration(rand.Intn(100)) * time.Microsecond) }
    }
    defer func() { randDelayHook = nil }()

    acc := &Account{balance: 100}
    var wg sync.WaitGroup
    for i := 0; i < 2; i++ {
        wg.Add(1)
        go func() { defer wg.Done(); transfer(acc, &Account{}, 80) }()
    }
    wg.Wait()
    // 若发生丢失更新/双花，最终余额会异常
    if acc.balance < 0 {
        t.Fatalf("竞态触发：余额为负 %d", acc.balance)
    }
}

#[cfg(loom)]
use loom::sync::atomic::{AtomicUsize, Ordering};
#[cfg(not(loom))]
use std::sync::atomic::{AtomicUsize, Ordering};

#[test]
fn concurrent_increment() {
    loom::model(|| {
        let counter = Arc::new(AtomicUsize::new(0));
        let c1 = counter.clone();
        let c2 = counter.clone();

        let t1 = loom::thread::spawn(move || { c1.fetch_add(1, Ordering::SeqCst); });
        let t2 = loom::thread::spawn(move || { c2.fetch_add(1, Ordering::SeqCst); });
        t1.join().unwrap();
        t2.join().unwrap();

        assert_eq!(counter.load(Ordering::SeqCst), 2);
    });
}

# Cargo.toml
[dev-dependencies]
loom = "0.7"
[target.'cfg(loom)'.dev-dependencies]
loom = "0.7"

# 运行：用 loom 配置替换标准库
# RUSTFLAGS="--cfg loom" cargo test --release

// 把 Ordering::Relaxed 改成 SeqCst 之前，Loom 会报出失败的交错：
// assertion failed: left == right
//   left: 1, right: 2
// 并给出导致该结果的具体线程切换序列

func TestOrderViolation(t *testing.T) {
    aReached, bReached := make(chan struct{}), make(chan struct{})
    var shared int

    go func() {
        shared = 1
        close(aReached)     // 通知：我写完了
        <-bReached
        _ = shared
    }()
    go func() {
        <-aReached
        shared = 2          // 与上面的读交错
        close(bReached)
    }()
    // 用通道精确编排两个 goroutine 的交错顺序
}

# Go：-count 重复运行 + 高并发
go test -race -count=100 -cpu=1,2,4,8 -run TestConcurrent ./...

# JVM：多轮 + 不同 JIT 状态
for i in $(seq 1 50); do mvn -q test -Dtest=ConcurrentTest || break; done

# 用 stress-ng 给系统加压，改变调度时序
stress-ng --cpu $(nproc) --timeout 60s &
go test -race -count=200 ./internal/concurrent/...

import random, threading, time, functools

class ChaosScheduler:
    def __init__(self, seed=0):
        self.rand = random.Random(seed)
        self.enabled = True

    def maybe_pause(self):
        if self.enabled and self.rand.random() < 0.3:   # 30% 概率暂停
            time.sleep(self.rand.uniform(0, 0.005))

    def patch(self, cls, methods):
        for m in methods:
            orig = getattr(cls, m)
            @functools.wraps(orig)
            def wrapper(*a, _orig=orig, **kw):
                self.maybe_pause()
                return _orig(*a, **kw)
            setattr(cls, m, wrapper)

def test_account_transfer_race():
    chaos = ChaosScheduler(seed=12345)
    chaos.patch(Account, ["balance", "withdraw", "deposit"])
    # 固定 seed 后，失败可复现：改 seed 直到复现，再保留该 seed 作为回归用例
    for _ in range(1000):
        run_transfer_scenario()

# TLC 模型检查器穷举所有状态，验证 Invariant 在所有可达状态下成立
java -cp tla2tools.jar tlc2.TLC Transfer.tla

// 死锁的最小复现：两个锁的获取顺序相反
func TestDeadlock(t *testing.T) {
    var mu1, mu2 sync.Mutex
    var wg sync.WaitGroup
    wg.Add(2)
    go func() { defer wg.Done(); mu1.Lock(); time.Sleep(time.Millisecond); mu2.Lock(); mu2.Unlock(); mu1.Unlock() }()
    go func() { defer wg.Done(); mu2.Lock(); time.Sleep(time.Millisecond); mu1.Lock(); mu1.Unlock(); mu2.Unlock() }()

    done := make(chan struct{})
    go func() { wg.Wait(); close(done) }()
    select {
    case <-done:
    case <-time.After(2 * time.Second):
        t.Fatal("检测到死锁：2s 内未完成")
    }
}

# 运行时死锁检测：Go 在全部 goroutine 阻塞时会 panic
# fatal error: all goroutines are asleep - deadlock!

# JVM：用 jstack 抓线程转储，查找 "Found one Java-level deadlock"
jstack $(pgrep -f MyApp) | grep -A20 "deadlock"

def test_no_starvation_under_contention():
    import threading, time
    served = []
    lock = threading.Lock()

    def worker(wid):
        for _ in range(100):
            with lock:
                served.append(wid)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
    start = time.time()
    for t in threads: t.start()
    for t in threads: t.join(timeout=10)
    assert all(not t.is_alive() for t in threads), "存在线程饥饿"
    assert len(served) == 1000

jobs:
  race-test:
    runs-on: ubuntu-latest          # 核数影响交错，固定 runner 规格
    steps:
      - run: go test -race -count=50 -cpu=4 ./...
      - name: Repeat to catch flaky races
        run: |
          for i in $(seq 1 10); do
            go test -race -run TestConcurrent ./... || exit 1
          done

种子管理规则：
  · 所有随机 sleep/调度都从显式 seed 派生
  · 失败时打印 seed，把它固化为回归用例
  · 回归用例用固定 seed + 增大迭代次数，确保稳定触发

判断流程：
  1. 失败是否可复现？（换 seed 是否稳定触发）
     · 可复现 → 真实并发缺陷，修复生产代码
     · 不可复现 → 可能是测试自身问题（共享状态、顺序依赖）
  2. 测试是否隔离？（每个用例独立的状态、独立的进程）
     · 不隔离 → 修测试（用 t.Parallel 时尤其注意共享变量）
     · 隔离 → 继续排查生产代码