Lecture 17: Transactional Memory

// stm.hpp —— 教学用软件事务内存 (Software Transactional Memory, STM)
// 设计要点:
//   1) 全局版本时钟 (global version clock) 给每个已提交事务分配提交序号 => 可串行化
//   2) 细粒度"锁位"表达写集所有权: 版本字最低位 = 1 表示被某事务独占
//   3) 数据版本策略可选: 惰性(写缓冲, LAZY) / 急切(undo log, EAGER)
//   4) 冲突检测: 读时检查版本(悲观/急切的"见锁即退") + 提交时验证读集(乐观)
//   5) 前进保证: 有限次退避重试后进入"排空式全局串行"回退, 消除活锁
// 用法/编译: 本文件是头文件, 与使用方一起编译, 例如
//   g++ -O3 -std=c++17 -pthread transfer.cpp -o transfer            (惰性版本)
//   g++ -O3 -std=c++17 -pthread -DSTM_EAGER=1 transfer.cpp -o transfer_eager
#pragma once
#include <atomic>
#include <cstdint>
#include <vector>
#include <utility>
#include <functional>
#include <mutex>
#include <thread>
#include <chrono>

#ifndef STM_EAGER
#define STM_EAGER 0          // 0 = 惰性版本(写缓冲), 1 = 急切版本(undo log)
#endif

namespace stm {

struct Word {                                   // 一个"事务化内存字"
    std::atomic<uint64_t> ver{0};               // 偶数=可访问; 奇数=被某事务独占
    long val{0};                                // 数据本体
};
static_assert(sizeof(Word) == 16, "Word should be 16B");

inline std::atomic<uint64_t> g_clock{0};       // 全局版本时钟, 每次提交 +2
inline std::atomic<bool>     g_serial{false};  // 全局串行模式(回退)标志
inline std::mutex            g_serial_lock;    // 回退互斥

// 统计(可选)
inline std::atomic<uint64_t> g_n_abort{0}, g_n_fallback{0}, g_n_commit{0};

struct Tx {
    uint64_t snapshot = 0;
    bool app_abort = false;                              // 应用层显式回滚
    std::vector<std::pair<Word*,uint64_t>> read_set;     // (地址, 当时版本)
    std::vector<Word*> owned;                            // 已持有的写集
#if STM_EAGER
    std::vector<std::pair<Word*,long>> undo;             // 旧值(undo log)
#else
    std::vector<std::pair<Word*,long>> wbuf;             // 新值(写缓冲)
#endif

    void begin() {
        app_abort = false;
        read_set.clear(); owned.clear();
        if (read_set.capacity() == 0) { read_set.reserve(32); owned.reserve(8); }  // 避免事务内扩容
#if STM_EAGER
        undo.clear(); if (undo.capacity() == 0) undo.reserve(8);
#else
        wbuf.clear(); if (wbuf.capacity() == 0) wbuf.reserve(8);
#endif
        snapshot = g_clock.load(std::memory_order_acquire);
    }

    bool read(Word* w, long& out) {
#if STM_EAGER
        for (Word* p : owned) if (p == w) { out = w->val; return true; }        // 值已在内存
#else
        for (auto& e : wbuf) if (e.first == w) { out = e.second; return true; } // 取自写缓冲
#endif
        if (read_committed(w, out)) return true;
        // 争用管理器(contention manager): 面对"被独占"的字, 决定等还是退
        if (!owned.empty()) return false;          // 已持写集: 立刻退, 避免环形等待/死锁
        if (!wait_unlocked(w)) return false;       // 未持写集: 有界等待持有者提交或回滚
        return read_committed(w, out);             // 重新读一次
    }

    bool read_committed(Word* w, long& out) {      // 读一个未被独占的字
        uint64_t v1 = w->ver.load(std::memory_order_acquire);
        if (v1 & 1ULL) return false;               // 被某事务独占
        long v = w->val;
        uint64_t v2 = w->ver.load(std::memory_order_acquire);
        if (v1 != v2) return false;                // 读到撕裂
        if (v1 > snapshot) return false;           // 提交于本事务开始之后 => 不可串行化
        read_set.emplace_back(w, v1);
        out = v;
        return true;
    }

    static bool wait_unlocked(Word* w) {           // 有界等待: 持有者很快会提交/回滚
        for (int r = 0; r < 8192; ++r) {
            if (!(w->ver.load(std::memory_order_acquire) & 1ULL)) return true;
            if ((r & 255) == 255) std::this_thread::yield();
        }
        return false;
    }

    bool write(Word* w, long x) {
        for (Word* p : owned) if (p == w) {
#if STM_EAGER
            w->val = x;
#else
            for (auto& e : wbuf) if (e.first == w) { e.second = x; break; }
#endif
            return true;
        }
        uint64_t v = w->ver.load(std::memory_order_acquire);
        if (v & 1ULL) return false;
        if (v > snapshot) return false;
        uint64_t expect = v;
        if (!w->ver.compare_exchange_strong(expect, v + 1, std::memory_order_acq_rel)) return false;
        owned.push_back(w);                   // 抢到所有权
#if STM_EAGER
        undo.emplace_back(w, w->val);
        w->val = x;
#else
        wbuf.emplace_back(w, x);
#endif
        return true;
    }

    bool commit() {
        for (auto& r : read_set) {            // 1) 验证读集(乐观检测在此发生)
            bool mine = false;
            for (Word* p : owned) if (p == r.first) { mine = true; break; }
            if (!mine && r.first->ver.load(std::memory_order_acquire) != r.second) { abort(); return false; }
        }
        uint64_t c = g_clock.fetch_add(2, std::memory_order_acq_rel) + 2;   // 2) 申请提交版本
#if STM_EAGER
        for (Word* w : owned) w->ver.store(c, std::memory_order_release);
#else
        for (auto& e : wbuf) {                // 3) 冲刷写缓冲: 先写值, 再 release 发布版本
            e.first->val = e.second;
            e.first->ver.store(c, std::memory_order_release);
        }
#endif
        g_n_commit.fetch_add(1, std::memory_order_relaxed);
        read_set.clear(); owned.clear();
#if STM_EAGER
        undo.clear();
#else
        wbuf.clear();
#endif
        return true;
    }

    void abort() {                            // 回滚 + 释放所有锁
#if STM_EAGER
        for (auto& e : undo) e.first->val = e.second;    // 急切版本: 恢复旧值
#endif
        for (Word* w : owned) {
            uint64_t cur = w->ver.load(std::memory_order_relaxed);
            w->ver.store(cur & ~1ULL, std::memory_order_release);
        }
        read_set.clear(); owned.clear();
#if STM_EAGER
        undo.clear();
#else
        wbuf.clear();
#endif
    }
};

inline void backoff(int attempt) {
    int r = 1 << (attempt < 6 ? attempt : 6);
    for (volatile int i = 0; i < r * 64; ++i) { }
    if (attempt >= 4) std::this_thread::yield();
}

// 正常路径: 乐观重试; 超过 tries 次则进入排空式串行回退(保证前进)
template <class F>
bool run_tx(F body, uint64_t* n_abort = nullptr, uint64_t* n_serial = nullptr, int tries = 16) {
    for (int a = 0; a < tries; ++a) {
        if (g_serial.load(std::memory_order_acquire)) { std::this_thread::yield(); backoff(a); continue; }
        Tx tx; tx.begin();
        bool body_ok = body(tx);
        bool done = body_ok && tx.commit();
        bool app = tx.app_abort;
        if (!done) tx.abort();                 // 所有失败路径都必须回滚(释放锁)
        if (done) return true;
        if (app) return false;                 // 应用层拒绝(如"余额不足"): 干净回滚, 不重试
        g_n_abort.fetch_add(1, std::memory_order_relaxed);
        if (n_abort) ++*n_abort;
        backoff(a);
    }
    // 回退: 全局串行执行一次(带时限的停滞重试), 保证前进
    std::lock_guard<std::mutex> g(g_serial_lock);
    g_serial.store(true, std::memory_order_release);
    g_n_fallback.fetch_add(1, std::memory_order_relaxed);
    if (n_serial) ++*n_serial;
    auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(1000);
    for (;;) {
        Tx tx; tx.begin();
        bool body_ok = body(tx);
        bool done = body_ok && tx.commit();
        bool app = tx.app_abort;
        if (!done) tx.abort();
        if (done || app || std::chrono::steady_clock::now() > deadline) {
            g_serial.store(false, std::memory_order_release);
            return done;
        }
        std::this_thread::yield();
    }
}

} // namespace stm

// Bank transfer(银行转账)微基准: 用同一套 STM 跑"组合式"事务, 并检验不变量
// 编译(惰性版本): g++ -O3 -std=c++17 -pthread transfer.cpp -o transfer
// 编译(急切版本): g++ -O3 -std=c++17 -pthread -DSTM_EAGER=1 transfer.cpp -o transfer_eager
#include "stm.hpp"
#include <cstdio>
#include <cstdlib>
#include <thread>
#include <vector>
#include <atomic>

struct Account { stm::Word balance; };
static Account acct[4096];

// ---- 事务化的原子操作(可组合的构建块) ----
static bool deposit(stm::Tx& tx, Account& a, long amt) {
    long b; if (!tx.read(&a.balance, b)) return false;      // 冲突 => 重试
    return tx.write(&a.balance, b + amt);
}
static bool withdraw(stm::Tx& tx, Account& a, long amt, long* ryow_bad) {
    long b; if (!tx.read(&a.balance, b)) return false;
    if (b < amt) { tx.app_abort = true; return false; }     // "余额不足"=应用层拒绝, 干净回滚
    if (!tx.write(&a.balance, b - amt)) return false;
    long again;                                             // 检验"读己之所写"(写缓冲/内存一致性)
    if (!tx.read(&a.balance, again)) return false;
    if (again != b - amt && ryow_bad) ++*ryow_bad;
    return true;
}
// transfer 组合 withdraw/deposit: 最外层事务定义原子性边界
// fail_deposit=true 模拟"入账阶段抛异常": 已完成的取款必须被整体回滚(failure atomicity)
static bool transfer(stm::Tx& tx, Account& f, Account& t, long amt, bool fail_deposit, long* ryow_bad) {
    if (!withdraw(tx, f, amt, ryow_bad)) return false;
    if (fail_deposit) { tx.app_abort = true; return false; }
    return deposit(tx, t, amt);
}

int main(int argc, char** argv) {
    const int T = (argc > 1) ? atoi(argv[1]) : 4;
    const int ITERS = (argc > 2) ? atoi(argv[2]) : 200000;
    const int N = (argc > 3) ? atoi(argv[3]) : 1024;
    const long INIT = 1000000;
    for (int i = 0; i < N; ++i) acct[i].balance.val = INIT;
    long expect = (long)N * INIT;

    std::atomic<uint64_t> aborts{0}, serials{0}, app_fail{0}, lost{0}, ryow_bad{0};
    auto work = [&](int tid) {
        uint32_t seed = 12345u + tid * 977u;
        uint64_t ab = 0, se = 0; long rb = 0;
        for (int i = 0; i < ITERS; ++i) {
            seed = seed * 1664525u + 1013904223u;
            int a = (int)((seed >> 8) % (unsigned)N);
            int b = (int)((seed >> 16) % (unsigned)N);
            if (a == b) b = (b + 1) % N;
            if (seed & 0x8000u) { int t = a; a = b; b = t; }        // 双向, 避免单向抽干账户
            long amt = 1 + (long)((seed >> 24) % 100);
            bool must_fail = ((i & 15) == 15);                       // 每 16 次一次"入账失败"
            bool ok = stm::run_tx([&](stm::Tx& tx) {
                return transfer(tx, acct[a], acct[b], amt, must_fail, &rb);
            }, &ab, &se);
            if (must_fail ? ok : !ok) lost.fetch_add(1, std::memory_order_relaxed);
            if (must_fail) app_fail.fetch_add(1, std::memory_order_relaxed);
        }
        aborts.fetch_add(ab, std::memory_order_relaxed);
        serials.fetch_add(se, std::memory_order_relaxed);
        ryow_bad.fetch_add((uint64_t)rb, std::memory_order_relaxed);
    };

    auto t0 = std::chrono::steady_clock::now();
    std::vector<std::thread> th;
    for (int i = 0; i < T; ++i) th.emplace_back(work, i);
    for (auto& t : th) t.join();
    double sec = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();

    long total = 0; for (int i = 0; i < N; ++i) total += acct[i].balance.val;
    printf("%s T=%2d N=%5d : %7.1f ns/committed-tx  %6.2f M tx/s  abort=%5.1f%%  fallback=%llu  "
           "lost=%llu  ryow_bad=%llu  sum=%s\n",
#if STM_EAGER
           "EAGER",
#else
           "LAZY ",
#endif
           T, N, sec * 1e9 / ((double)T * ITERS) * T / T * 1.0,
           (double)T * ITERS / sec / 1e6,
           100.0 * aborts.load() / (aborts.load() + (double)T * ITERS),
           (unsigned long long)serials.load(),
           (unsigned long long)lost.load(), (unsigned long long)ryow_bad.load(),
           total == expect ? "OK" : "*** CORRUPTED ***");
    (void)app_fail;
    return (total == expect && lost.load() == 0 && ryow_bad.load() == 0) ? 0 : 1;
}

// 并发二叉搜索树插入: 粗粒度锁 / 手递手细粒度锁 / 事务, 三者对比
// 编译: g++ -O3 -std=c++17 -pthread bst.cpp -o bst
#include "stm.hpp"
#include <cstdio>
#include <cstdlib>
#include <thread>
#include <vector>
#include <mutex>
#include <chrono>
#include <functional>
#include <atomic>

// ---------------- 事务版本: 子指针是事务化内存字 ----------------
struct TNode {
    long key;                    // 发布后不可变: 可非事务化读
    stm::Word left, right;       // 子指针(以 long 存指针), 0 表示空
};
static stm::Word root_word;

static bool tx_insert(stm::Tx& tx, long k, TNode* nn) {
    stm::Word* cur = &root_word; long c;
    if (!tx.read(cur, c)) return false;
    for (;;) {
        if (c == 0) return tx.write(cur, (long)nn);        // 挂到空位上
        TNode* n = (TNode*)c;
        stm::Word* child = (k < n->key) ? &n->left : &n->right;
        long cc; if (!tx.read(child, cc)) return false;
        if (cc == 0) return tx.write(child, (long)nn);
        cur = child; c = cc;                               // 继续下降
    }
}
static void stm_insert(long k, uint64_t* ab, uint64_t* se) {
    TNode* nn = new TNode{k, {}, {}};                      // malloc 不可回滚 => 放在事务之外
    bool ok = stm::run_tx([&](stm::Tx& tx) { return tx_insert(tx, k, nn); }, ab, se);
    if (!ok) { fprintf(stderr, "insert %ld failed\n", k); exit(1); }
}

// ---------------- 手递手(hand-over-hand)细粒度锁版本 ----------------
struct LNode {
    long key;
    LNode *left = nullptr, *right = nullptr;
    std::mutex m;
};
static LNode* lock_root = nullptr;

static void fine_insert(long k) {
    LNode* cur = lock_root;
    cur->m.lock();                                          // 先锁父, 再锁子, 然后放父
    for (;;) {
        LNode** slot = (k < cur->key) ? &cur->left : &cur->right;
        if (*slot == nullptr) { *slot = new LNode{k}; cur->m.unlock(); return; }
        LNode* nxt = *slot;
        nxt->m.lock();
        cur->m.unlock();
        cur = nxt;
    }
}

// ---------------- 粗粒度锁版本 ----------------
static std::mutex g_big_lock;
static void coarse_insert(LNode*& root, long k) {
    std::lock_guard<std::mutex> g(g_big_lock);
    LNode** p = &root;
    while (*p) p = (k < (*p)->key) ? &(*p)->left : &(*p)->right;
    *p = new LNode{k};
}

// ---------------- 校验: 中序遍历必须严格递增 ----------------
static long verify_tx(TNode* n, long* prev, bool* sorted) {
    if (!n) return 0;
    long c = verify_tx((TNode*)n->left.val, prev, sorted);
    if (n->key <= *prev) *sorted = false;
    *prev = n->key;
    return c + 1 + verify_tx((TNode*)n->right.val, prev, sorted);
}
static long verify_lock(LNode* n, long* prev, bool* sorted) {
    if (!n) return 0;
    long c = verify_lock(n->left, prev, sorted);
    if (n->key <= *prev) *sorted = false;
    *prev = n->key;
    return c + 1 + verify_lock(n->right, prev, sorted);
}

// 以"先插区间中点"的递归顺序插入 => 树高 O(log n), 避免退化成链表
static void fill_bal(long lo, long hi, const std::function<void(long)>& ins) {
    if (lo > hi) return;
    long mid = lo + (hi - lo) / 2;
    ins(mid);
    fill_bal(lo, mid - 1, ins);
    fill_bal(mid + 1, hi, ins);
}

static LNode* g_coarse_root = nullptr;
static std::atomic<uint64_t> g_ab{0}, g_se{0};

static void body_fine(int tid, long per) {
    long base = (long)tid * per;
    fill_bal(base + 1, base + per - 1, [](long k) { fine_insert(k); });
}
static void body_coarse(int tid, long per) {
    long base = (long)tid * per;
    fill_bal(base + 1, base + per - 1, [](long k) { coarse_insert(g_coarse_root, k); });
}
static void body_stm(int tid, long per) {
    long base = (long)tid * per;
    uint64_t ab = 0, se = 0;
    fill_bal(base + 1, base + per - 1, [&](long k) { stm_insert(k, &ab, &se); });
    g_ab.fetch_add(ab, std::memory_order_relaxed);
    g_se.fetch_add(se, std::memory_order_relaxed);
}

static double timed(const char* name, int T, long M, std::function<void(int,long)> body) {
    auto t0 = std::chrono::steady_clock::now();
    std::vector<std::thread> th;
    for (int i = 0; i < T; ++i) th.emplace_back(body, i, M + 1);
    for (auto& x : th) x.join();
    double sec = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
    printf("%-22s T=%2d : %8.1f ms  %7.1f ns/insert  %7.2f M insert/s\n",
           name, T, sec * 1e3, sec * 1e9 / ((double)T * M), (double)T * M / sec / 1e6);
    return sec;
}

int main(int argc, char** argv) {
    const int T = (argc > 1) ? atoi(argv[1]) : 4;
    const int M = (argc > 2) ? atoi(argv[2]) : 20000;      // 每线程插入数
    const long PER = M + 1;                                // 每线程独占的键区间长度

    root_word.val = 0;
    lock_root = new LNode{0};
    g_coarse_root = new LNode{0};
    { TNode* nn0 = new TNode{0, {}, {}};                   // 三棵树用同一套键: 0 与 T-1 个分界键
      stm::run_tx([&](stm::Tx& tx) { return tx_insert(tx, 0, nn0); }, nullptr, nullptr); }
    for (int t = 1; t < T; ++t) {                          // 预建骨架: T 个互不相交的键区间
        long k = (long)t * PER;
        LNode** p = &g_coarse_root; while (*p) p = (k < (*p)->key) ? &(*p)->left : &(*p)->right;
        *p = new LNode{k};
        fine_insert(k);
        TNode* nn = new TNode{k, {}, {}};
        stm::run_tx([&](stm::Tx& tx) { return tx_insert(tx, k, nn); }, nullptr, nullptr);
    }

    timed("hand-over-hand locks", T, M, body_fine);
    timed("single global lock", T, M, body_coarse);
    timed("STM (transactions)", T, M, body_stm);
    printf("  STM aborts=%llu fallbacks=%llu\n",
           (unsigned long long)g_ab.load(), (unsigned long long)g_se.load());

    long expect = (long)T * PER, prev = -1; bool sorted = true;
    long n1 = verify_tx((TNode*)root_word.val, &prev, &sorted);
    printf("  STM tree : nodes=%ld (expect %ld) sorted=%s\n", n1, expect, sorted ? "YES" : "NO");
    bool ok1 = (n1 == expect && sorted);
    prev = -1; sorted = true;
    long n2 = verify_lock(g_coarse_root, &prev, &sorted);
    printf("  lock tree: nodes=%ld (expect %ld) sorted=%s\n", n2, expect, sorted ? "YES" : "NO");
    prev = -1; sorted = true;
    long n3 = verify_lock(lock_root, &prev, &sorted);
    printf("  hoh tree : nodes=%ld (expect %ld) sorted=%s\n", n3, expect, sorted ? "YES" : "NO");
    return (ok1 && n2 == expect && n3 == expect) ? 0 : 1;
}

// 硬件事务内存 (Intel TSX / RTM) 示例: 事务化转账 + 不变量读取者
// 编译: g++ -O3 -std=c++17 -pthread -mrtm tsx.cpp -o tsx
//   (必须加 -mrtm: RTM 内建函数在未开启该特性时无法内联, 会报 always_inline 错误)
// 说明: 需要 CPU 支持 RTM(Intel Haswell/Broadwell 及后续部分型号)。
//       无 RTM 时 _xbegin() 立即返回失败状态, 本程序自动走互斥锁回退路径。
#include <immintrin.h>
#include <cpuid.h>
#include <cstdio>
#include <cstdlib>
#include <thread>
#include <vector>
#include <atomic>
#include <mutex>
#include <chrono>

struct Account { long balance; };
static Account acct[8];
static std::mutex g_fallback_lock;         // 回退路径: 普通互斥锁保证前进

static std::atomic<uint64_t> n_htm_commit{0}, n_conflict{0}, n_capacity{0},
                             n_explicit{0}, n_other{0}, n_mutex{0}, n_tear{0};

static bool g_has_rtm = false;             // 必须先探测: 无 RTM 时执行 XBEGIN 会产生 #UD 非法指令异常

static bool cpu_has_rtm() {
    unsigned a, b, c, d;
    if (__get_cpuid_max(0, nullptr) < 7) return false;
    __cpuid_count(7, 0, a, b, c, d);
    return (b & (1u << 11)) != 0;          // CPUID.07H:EBX[11] = RTM
}

// 用硬件事务完成一次转账; 冲突/容量溢出时解码状态位并退避重试, 最终回退到锁
static void htm_transfer(int from, int to, long amt) {
    if (!g_has_rtm) {                                  // 无硬件事务支持: 直接走锁路径
        std::lock_guard<std::mutex> g(g_fallback_lock);
        long a = acct[from].balance, b = acct[to].balance;
        if (a >= amt) { acct[from].balance = a - amt; acct[to].balance = b + amt; }
        n_mutex.fetch_add(1, std::memory_order_relaxed);
        return;
    }
    for (int attempt = 0; ; ++attempt) {
        unsigned st = _xbegin();
        if (st == _XBEGIN_STARTED) {
            // ★ 事务体内是"普通"读写: 硬件用缓存记录读集/写集
            long a = acct[from].balance;
            long b = acct[to].balance;
            if (a < amt) { _xabort(1); }            // 显式回滚(应用层拒绝)
            acct[from].balance = a - amt;           // 写集: 只在提交时对其他核可见
            acct[to].balance   = b + amt;
            _xend();                                // 提交: 写集原子生效
            n_htm_commit.fetch_add(1, std::memory_order_relaxed);
            return;
        }
        if (st & _XABORT_CONFLICT)      n_conflict.fetch_add(1, std::memory_order_relaxed);
        else if (st & _XABORT_CAPACITY) n_capacity.fetch_add(1, std::memory_order_relaxed);
        else if (st & _XABORT_EXPLICIT) n_explicit.fetch_add(1, std::memory_order_relaxed);
        else                            n_other.fetch_add(1, std::memory_order_relaxed);
        if (attempt >= 8) {                         // 回退: 用锁串行完成
            std::lock_guard<std::mutex> g(g_fallback_lock);
            long a = acct[from].balance, b = acct[to].balance;
            if (a >= amt) { acct[from].balance = a - amt; acct[to].balance = b + amt; }
            n_mutex.fetch_add(1, std::memory_order_relaxed);
            return;
        }
        for (volatile int i = 0; i < (1 << attempt) * 64; ++i) { }   // 指数退避
    }
}

// 读取者: 在同一瞬间观察两个账户, 校验总额不变(检验隔离性)
static void reader(int rounds, const long* expected_total, std::atomic<bool>* stop) {
    while (!*stop) {
        if (!g_has_rtm) {
            long s;
            { std::lock_guard<std::mutex> g(g_fallback_lock);
              s = 0; for (int i = 0; i < 8; ++i) s += acct[i].balance; }
            if (s != *expected_total) n_tear.fetch_add(1, std::memory_order_relaxed);
            continue;
        }
        unsigned st = _xbegin();
        if (st == _XBEGIN_STARTED) {
            long s = 0; for (int i = 0; i < 8; ++i) s += acct[i].balance;   // 8 个字必须"同一瞬间"
            _xend();
            if (s != *expected_total) n_tear.fetch_add(1, std::memory_order_relaxed);
        } else {
            long s;
            { std::lock_guard<std::mutex> g(g_fallback_lock);
              s = 0; for (int i = 0; i < 8; ++i) s += acct[i].balance; }
            if (s != *expected_total) n_tear.fetch_add(1, std::memory_order_relaxed);
        }
        (void)rounds;
    }
}

int main(int argc, char** argv) {
    const int T = (argc > 1) ? atoi(argv[1]) : 4;
    const int ITERS = (argc > 2) ? atoi(argv[2]) : 200000;
    g_has_rtm = cpu_has_rtm();
    printf("CPU supports RTM: %s\n", g_has_rtm ? "YES" : "NO (将全部走互斥锁回退路径)");
    for (int i = 0; i < 8; ++i) acct[i].balance = 1000000;
    const long expected = 8L * 1000000;        // 8 个账户总额守恒

    std::atomic<bool> stop{false};
    auto t0 = std::chrono::steady_clock::now();
    std::thread rd(reader, 0, &expected, &stop);
    std::vector<std::thread> th;
    for (int i = 0; i < T; ++i) th.emplace_back([&, i] {
        uint32_t seed = 99u + i * 7919u;
        for (int k = 0; k < ITERS; ++k) {
            seed = seed * 1664525u + 1013904223u;
            int a = (int)((seed >> 8) % 8), b = (int)((seed >> 16) % 8);
            if (a == b) b = (b + 1) % 8;
            htm_transfer(a, b, 1 + (long)((seed >> 24) % 50));
        }
    });
    for (auto& x : th) x.join();
    stop.store(true); rd.join();
    double sec = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();

    long total = 0; for (int i = 0; i < 8; ++i) total += acct[i].balance;
    printf("  time            = %.3f s  (%.1f ns/transfer)\n", sec, sec * 1e9 / ((double)T * ITERS));
    printf("  htm commits     = %llu\n", (unsigned long long)n_htm_commit.load());
    printf("  aborts: conflict=%llu capacity=%llu explicit=%llu other=%llu\n",
           (unsigned long long)n_conflict.load(), (unsigned long long)n_capacity.load(),
           (unsigned long long)n_explicit.load(), (unsigned long long)n_other.load());
    printf("  mutex fallbacks = %llu\n", (unsigned long long)n_mutex.load());
    printf("  torn reads      = %llu  (必须为 0)\n", (unsigned long long)n_tear.load());
    printf("  INVARIANT total = %ld (expected %ld) => %s\n", total, 8000000L,
           total == 8000000L ? "OK" : "*** CORRUPTED ***");
    return (total == 8000000L && n_tear.load() == 0) ? 0 : 1;
}