自定义自旋锁实现¶

以锁的抽象方式用于保护执行时间极短（远短于线程上下文切换耗时）的共享资源操作，如订单簿实时更新、行情数据快照抓取、交易信号快速计算等场景。

原子操作、自旋锁、互斥锁实现对比

特性	std::atomic	Spinlock	std::mutex
核心机制	单条CPU硬件原子指令执行，或基于LDREX/STREX指令的循环逻辑，确保操作不可分割	在用户空间循环中执行原子“读-改-写”指令，通过忙等待获取锁，不主动让出CPU	结合用户空间原子操作（抢锁）与内核空间线程调度（阻塞/唤醒），平衡抢锁效率与CPU资源利用
底层依赖	依赖CPU指令集： - x86-64：LOCK前缀指令（如LOCK XADD、LOCK CMPXCHG） - ARM：LDREX/STREX指令、LSE（Large System Extensions）指令集	依赖CPU原子指令，通常基于std::atomic_flag或std::atomic，通过test_and_set或compare_exchange_strong指令在循环中实现	依赖操作系统同步原语： - Linux：futex系统调用（FUTEX_WAIT/FUTEX_WAKE） - Windows：WaitOnAddress/WakeByAddress，或早期内核对象（如Mutex）
竞争时行为	操作本身不可中断；若为CAS操作，失败后立即返回false，是否重试由上层代码逻辑决定	线程持续在循环中执行CAS等原子指令，占用CPU资源忙等待，直到成功获取锁	1. 首次尝试：用户态执行CAS操作抢锁； 2. 抢锁失败：调用系统调用（如futex），线程进入内核态并加入等待队列，释放CPU进入睡眠状态
开销	极低：仅消耗几个CPU时钟周期，无内核陷入（用户态直接执行）	低：无系统调用和线程上下文切换开销；但竞争时会持续占用CPU周期，增加CPU缓存一致性流量	无竞争时：极低，接近单次原子操作开销；有竞争时：高，涉及2次系统调用+2次上下文切换（睡眠、唤醒各1次）
适用场景	实现无锁数据结构、性能计数器、线程安全标志位等	内核编程、驱动开发，或用户态下“临界区执行时间远小于线程上下文切换时间”的场景	保护绝大多数用户态临界区，尤其适合临界区内包含IO操作、等待其他事件，或计算时间不可控的场景
C++体现	std::atomic（支持基础类型及可平凡复制类型）、std::atomic_flag（无锁布尔标志）	标准库无直接实现，需通过std::atomic_flag手动构建循环抢锁逻辑	std::mutex（基础互斥锁）、std::recursive_mutex（可重入互斥锁）、std::shared_mutex（读写分离互斥锁）

1. 实现一：测试并设置自旋锁 (Test-and-Set Spinlock)

#include <atomic>
#include <cassert>
#include <thread>
#include <vector>

class spin_lock {
 public:
  spin_lock() : locked(0) {}

  void lock() {
    int zero = 0;
    while (!locked.compare_exchange_weak(zero, 1, std::memory_order_acq_rel)) {
      zero = 0;
    }
  }

  void unlock() { locked.store(0, std::memory_order_release); }

 private:
  std::atomic<int> locked;
};

constexpr auto concurrent = 16;
constexpr auto work_per_thread = 100000;

spin_lock lock;
int counter = 0;

auto add() {
  for (int i = 0; i < work_per_thread; ++i) {
    lock.lock();
    ++counter;
    lock.unlock();
  }
}

int main() {
  std::vector<std::thread> thrds;
  for (int i = 0; i < concurrent; ++i) {
    thrds.emplace_back(add);
  }
  for (auto& thrd : thrds) {
    thrd.join();
  }
  assert(counter == concurrent * work_per_thread);
}

使用一个单一的原子变量 locked 作为锁的状态标志（0表示未锁定，1表示已锁定）。线程通过一个原子性的“比较并交换”（Compare-and-Swap, CAS）操作来尝试获取锁。

void lock() {
  int zero = 0;
  // 尝试将 locked 从 0 交换为 1
  // 如果交换成功，!true -> false，循环结束，获得锁
  // 如果交换失败（说明锁已被其他线程持有，locked 不为 0），!false ->
  // true，继续循环
  while (!locked.compare_exchange_weak(zero, 1, std::memory_order_acq_rel)) {
    // 如果失败，compare_exchange_weak 会将 locked 的当前值加载到 zero 中
    // 所以需要将 zero 重置为 0，以便下次循环时继续尝试与 0 比较
    zero = 0;
  }
}

compare_exchange_weak：这是一个核心的原子操作。它会比较 locked 的值和 zero 的值。
如果相等（即 locked 是 0），它就原子性地将 locked 的值更新为 1，并返回 true。线程成功获取锁，循环结束。
如果不相等（即 locked 是 1），它就更新失败，将 locked 的当前值（也就是 1）加载到 zero 中，并返回 false。
while 循环：只要 compare_exchange_weak 返回 false（表示没抢到锁），循环就会继续。因为失败后 zero 的值会被改成 locked 的当前值（1），所以必须在下一次循环前把它重置为 0，这样才能继续尝试“将0换成1”。
memory_order_acq_rel：这是一个内存序。acq_rel = acquire + release。
Acquire (获取): 确保在成功获取锁之后的所有内存读写操作，都不会被重排到获取锁之前。这保证了线程能看到前一个持有锁的线程在临界区内做的所有修改。
Release (释放): 确保在尝试获取锁之前的所有内存操作都已完成。

void unlock() { locked.store(0, std::memory_order_release); }

store(0): 直接将 locked 的值设为 0，表示锁已释放。
memory_order_release：确保在释放锁之前的所有内存读写操作（比如 ++counter）都已完成，并且对其他线程可见。它与 lock() 中的 acquire 配对，构成了同步关系。
优点: 实现非常简单。
缺点:
不公平: 线程获得锁的顺序是随机的，可能导致某些线程“饿死”（starvation）。
高缓存争用 (High Cache Contention): 这是最大的问题。所有等待的线程都在同一个原子变量 locked 上自旋。在一个多核系统上，当一个线程释放锁（修改locked）时，会导致所有其他正在自旋的核心的缓存行（Cache Line）失效。这会引发大量的缓存一致性流量（Cache Coherence Traffic），占满内存总线，严重影响性能，尤其是在核心数很多的情况下。

2. 实现二：MCS 锁 (Mellor-Crummey and Scott Lock)

#include <atomic>
#include <cassert>
#include <thread>
#include <vector>

class spin_lock {
 public:
  struct lock_node {
    std::atomic<lock_node*> next;
    std::atomic<int> locked;
  };

  spin_lock() : tail(nullptr) {}

  void lock(lock_node* node) {
    node->next.store(nullptr, std::memory_order_relaxed);
    node->locked.store(0, std::memory_order_relaxed);

    auto prev = tail.exchange(node, std::memory_order_relaxed);
    if (not prev) {
      return;
    }

    prev->next.store(node, std::memory_order_relaxed);
    while (node->locked.load(std::memory_order_acquire) == 0);
  }

  void unlock(lock_node* node) {
    auto next = node->next.load(std::memory_order_relaxed);
    if (not next) {
      auto desired = node;
      if (tail.compare_exchange_strong(desired, nullptr,
                                       std::memory_order_release,
                                       std::memory_order_relaxed)) {
        return;
      }

      while ((next = node->next.load(std::memory_order_relaxed)) == nullptr);
    }

    next->locked.store(1, std::memory_order_release);
  }

 private:
  std::atomic<lock_node*> tail;
};

constexpr auto concurrent = 16;
constexpr auto work_per_thread = 100000;

spin_lock lock;
int counter = 0;

auto add() {
  for (int i = 0; i < work_per_thread; ++i) {
    spin_lock::lock_node node;
    lock.lock(&node);
    ++counter;
    lock.unlock(&node);
  }
}

int main() {
  std::vector<std::thread> thrds;
  for (int i = 0; i < concurrent; ++i) {
    thrds.emplace_back(add);
  }
  for (auto& thrd : thrds) {
    thrd.join();
  }
  assert(counter == concurrent * work_per_thread);
}

MCS 锁组织了一个等待线程的队列（逻辑上的链表）。每个希望获取锁的线程都将自己的一个 lock_node 节点加入到队列的尾部。然后，每个线程只在自己的 lock_node 节点上自旋，等待前一个线程释放锁后通知它。

void lock(lock_node* node) {
  // 1. 初始化自己的节点
  node->next.store(nullptr, std::memory_order_relaxed);
  node->locked.store(0, std::memory_order_relaxed);

  // 2. 将自己设为新的队尾，并获取旧的队尾
  auto prev = tail.exchange(node, std::memory_order_relaxed);

  // 3. 如果旧队尾是 nullptr，说明队列为空，直接获得锁
  if (not prev) {
    return;
  }

  // 4. 将自己连接到旧队尾的后面
  prev->next.store(node, std::memory_order_relaxed);

  // 5. 在自己的 locked 标志上自旋，等待前一个线程的通知
  while (node->locked.load(std::memory_order_acquire) == 0);
}

线程在自己的栈上创建一个 lock_node。
使用 tail.exchange() 原子地将自己设置为新的队尾，并取回之前队尾的指针 prev。
如果 prev 是 nullptr，意味着锁是空闲的，当前线程直接获取锁并返回。
如果 prev 不是 nullptr，意味着有线程在它前面。它就把自己的 node 地址存入前一个节点的 next 指针中，形成链表。
最后，它在自己的 node->locked 标志上自旋。这至关重要，因为每个线程都在不同的内存地址上自旋，避免了缓存争用。

void unlock(lock_node* node) {
  // 1. 检查是否有后继者
  auto next = node->next.load(std::memory_order_relaxed);

  // 2. 如果没有后继者 (next == nullptr)
  if (not next) {
    // 尝试将 tail 从自己(node)更新为 nullptr，表示队列空了
    // 这是一个 CAS 操作，因为可能在我检查 next 和更新 tail
    // 之间，有新线程加入了队列
    auto desired = node;
    if (tail.compare_exchange_strong(desired, nullptr, ...)) {
      return;  // 成功，队列变空，解锁完成
    }
    // 如果 CAS 失败，说明有新线程刚刚把自己加入了队尾
    // 此时必须等待，直到它的 next 指针被新线程设置好
    while ((next = node->next.load(std::memory_order_relaxed)) == nullptr);
  }

  // 3. 通知后继者可以获取锁了
  next->locked.store(1, std::memory_order_release);
}

解锁时，线程检查自己的 node->next 是否有值。
如果没有后继者，它尝试将全局的 tail 指针从 node 改为 nullptr。这里有一个竞态条件：可能在检查 next 之后，但在更新 tail 之前，另一个线程加入了队列。因此需要用 compare_exchange_strong 来安全地更新 tail。如果 CAS 失败，就必须等待 next 指针被新来的线程更新。
如果有后继者 next，就将其 locked 标志设置为 1，从而唤醒正在 next 节点上自旋的下一个线程。release 语义保证了临界区的修改对下一个线程可见。
优点:
公平: 遵循先进先出（FIFO）的顺序。
高性能/高可伸缩性: 每个线程在不同的内存位置上自旋，极大地减少了缓存争用，因此在多核环境下性能非常出色。
缺点:
实现比其他两种复杂。
lock 和 unlock 接口需要传递一个 lock_node 对象，使用起来稍显不便.

3. 实现三：票据锁 (Ticket Lock)

#include <atomic>
#include <cassert>
#include <thread>
#include <vector>

class spin_lock {
 public:
  spin_lock() : head(0), tail(0) {}

  void lock() {
    auto ticket = tail.fetch_add(1, std::memory_order_relaxed);
    while (head.load(std::memory_order_acquire) != ticket);
  }

  void unlock() { head.fetch_add(1, std::memory_order_release); }

 private:
  std::atomic<int> head;
  std::atomic<int> tail;
};

constexpr auto concurrent = 16;
constexpr auto work_per_thread = 100000;

spin_lock lock;
int counter = 0;

auto add() {
  for (int i = 0; i < work_per_thread; ++i) {
    lock.lock();
    ++counter;
    lock.unlock();
  }
}

int main() {
  std::vector<std::thread> thrds;
  for (int i = 0; i < concurrent; ++i) {
    thrds.emplace_back(add);
  }
  for (auto& thrd : thrds) {
    thrd.join();
  }
  assert(counter == concurrent * work_per_thread);
}

模拟银行或熟食店的排队叫号系统。

tail: "取号机"，每个想获取锁的线程都从这里取一个唯一的票号。
head: "叫号屏"，表示当前正在服务的号码。

void lock() {
  // 1. 取一个票号
  auto ticket = tail.fetch_add(1, std::memory_order_relaxed);

  // 2. 等待叫号，即 head 等于自己的票号
  while (head.load(std::memory_order_acquire) != ticket);
}

tail.fetch_add(1): 原子地将 tail 加一，并返回加一之前的值。这保证了每个线程都获得一个唯一的、递增的 ticket 号。
while (head.load(...) != ticket): 线程不断地检查 head 的值，直到轮到自己的 ticket 号，循环结束，获得锁。acquire 语义保证了能看到前一个线程的修改。

void unlock() {
  // 服务完成，叫下一个号
  head.fetch_add(1, std::memory_order_release);
}

head.fetch_add(1): 原子地将 head 加一，相当于“叫下一个号”。release 语义保证了临界区的修改对下一个获得锁的线程可见。
优点:
公平: 严格的先进先出（FIFO）顺序。
实现简单: 比 MCS 锁简单得多。
unlock 操作非常快，只是一个 fetch_add。
缺点:
仍然存在缓存争用: 和 Test-and-Set 锁一样，所有等待的线程都在同一个原子变量 head 上自旋。当 unlock 修改 head 时，同样会使所有等待核心的缓存行失效，引起总线流量。虽然比实现一略好（因为 unlock 操作本身很快），但在高争用下的伸缩性远不如 MCS 锁。

特性	实现一 (Test-and-Set)	实现二 (MCS Lock)	实现三 (Ticket Lock)
公平性	❌ 不公平	✅ 公平 (FIFO)	✅ 公平 (FIFO)
缓存争用	🔥 高	✅ 低 (最优)	⚠️ 中等
实现复杂度	✅ 低	❌ 高	✅ 中等
使用便利性	✅ 高	❌ 低 (需传 node)	✅ 高
适用场景	低线程争用环境	高线程争用、追求极致性能和伸缩性的环境	需要公平性且实现简单的通用场景

自旋等待与PAUSE：自旋（锁）的优化

自旋等待循环（Spin-wait Loop）是一个tight-loop（紧密循环）

while (ring_buffer[index].flag != EXPECTED_VALUE) {
  // Spin...
}
// Data is ready, process it...

while 循环的条件检查 (... != EXPECTED_VALUE) 是一个条件分支 (Conditional Branch) 指令。CPU的分支预测器 (Branch Predictor) 会试图预测这个分支的结果（是继续循环，还是退出循环）。

在自旋等待期间，这个循环条件 绝大多数 时候都是 true（数据还没来，继续循环）。
分支预测器会预测下一次检查的结果是true。
CPU会推测性地开始执行循环体内的指令。
当数据最终到达时，循环条件突然变为 false。
CPU必须丢弃所有基于错误预测而推测执行的指令和中间结果，清空它的指令流水线，然后重新从正确的分支路径开始取指和执行。

内存排序违规 (Memory Order Violation）的机制

CPU推测性地执行（错误地预测循环继续）时，它可能已经开始执行循环之后的代码。
在自旋等待中，一个核心（消费者）在读取一个内存地址，而另一个核心（生产者）在写入该地址。
消费者核心的推测执行路径（基于“循环继续”的错误预测）与生产者核心的写入操作（导致“循环退出”）在CPU内部的内存排序缓冲区（Memory Order Buffer）中发生冲突。
CPU的内存模型必须保证：在循环退出之后的指令，必须能“看到”导致循环退出的那次写入。
但由于推测执行，CPU可能已经基于一个旧的内存状态（数据还没来）开始执行后续的指令。当它最终发现分支预测错误时，它也同时检测到这种由推测执行引起的内存时序混乱。
这种冲突被CPU的内部逻辑检测为一次（潜在的）内存排序违规。为了纠正这个错误并严格遵守内存一致性模型，CPU会触发一次更深、代价更高的流水线清空。

PAUSE 指令 是一种用于优化自旋等待循环（spin-wait loop）的汇编指令，最早在 Intel Pentium 4 处理器中引入，但向后兼容所有 IA-32 处理器。在早期处理器中，PAUSE 指令的行为类似于 NOP（No Operation）指令，而在现代处理器中，它具有更复杂的功能。

优化自旋等待循环性能 在自旋等待循环中，处理器可能会因检测到内存顺序违规而导致性能下降。PAUSE 指令向处理器提供提示，表明当前代码是一个自旋等待循环，从而帮助处理器避免内存顺序违规。这种优化显著提升了自旋锁的性能。
降低功耗 在自旋等待循环中，处理器可能以极高的速度执行循环，导致大量电力消耗。插入 PAUSE 指令可以降低处理器的功耗，使其在等待资源时更加节能。
减少流水线刷新 在多核或超线程环境下，PAUSE 指令可以减少因推测性执行导致的流水线刷新，从而提高资源利用率，避免浪费其他线程的计算资源。
延迟操作 PAUSE 指令在某些处理器上实现为预定义的延迟操作（pre-defined delay），延迟时间有限，甚至可能为零。它不会改变处理器的架构状态，仅执行延迟操作。