当前位置：首页 > news >正文

C++11原子操作实现公平自旋锁

news 2025/8/2 21:12:08

让我们继续深入探讨自旋锁的实现。
首先回顾一下基于TAS和CAS算法实现的自旋锁，分析它们共同的不足之处：
TAS算法

class spin_lock_tas final {
public:void lock() {while (flag.test_and_set(memory_order_acquire));}void unlock() {flag.clear(memory_order_release);}private:atomic_flag flag = ATOMIC_FLAG_INIT;
};

CAS算法

class spin_lock_cas final {atomic<bool> sync_var{false};public:void lock() {bool expected;do {expected = false;} while (!sync_var.compare_exchange_strong(expected, true,memory_order_acquire, memory_order_relaxed));}void unlock() {sync_var.store(false, memory_order_release);}
};

在TAS算法的lock()操作中，自旋通过while (lock_var.test_and_set())实现。所有等待的CPU都会在lock_var上持续自旋。当unlock()将lock_var设为false时，所有竞争线程会同时对lock_var执行TAS操作，只有成功获取false值的线程才能获得锁。这种情况下，可能出现先自旋的线程未能获得锁，而后来的线程反而成功的情况。同样的公平性问题也存在于CAS算法中，这两种算法都无法确保先到先得的锁获取顺序。

是否存在一种基于“先到先得”策略的自旋锁实现算法？这能更好地保证公平性。

我们先不妨看一个生活中银行服务的场景：当客户到达银行后，取号机会按到达顺序发放号码，随后系统根据号码顺序叫号服务。这种机制避免了服务窗口前无序争抢的情况，确保先到的客户能优先获得服务。公平锁采用了类似原理：每个申请锁的线程会被分配一个有序编号，只有当前面的编号持有者完成操作后，后续编号的线程才能获得锁资源。这种有序分配方式确保了资源获取的公平性。

基本思路是：线程在申请锁时会先获取一个序号，随后持续等待，直到没有更早序号的线程尝试进入临界区。实现时需要借助一种特殊的原子操作原语：FAA（Fetch-And-Add）。

FAA 原语

<< atomic >>
function faa(p : pointer to int, inc : int) returns int {int value <- *location*p <- value + increturn value
}

FAA 原语的语义是：首先读取变量 p 的当前值 value，然后对 p 执行增量操作（增加 inc），最后返回 p 的原始值 value。这个读-修改-写操作必须保证原子性。在 C++11 中，整数类型的原子类通过fetch_add() 函数可实现该功能，在 x86 CPU 架构下，该操作对应 lock xadd 指令。

下面是公平锁的实现代码：

公平自旋锁实现

class spin_lock_fair {atomic<int> sync{0}; // 锁变量atomic<int> next{0}; // 取号机public:void lock() {int ticket = next.fetch_add(1, std::memory_order_relaxed); // 领凭证while (sync.load(std::memory_order_acquire) != ticket);}void unlock() {sync.fetch_add(1, std::memory_order_release);}
};

定义了两个原子整型变量：sync用于表示锁状态，释放锁时其值递增；next则用于实现"先来先服务"机制，严格单调递增。每次调用lock()时，线程会获取一个唯一的next值作为凭证，只有当sync等于该凭证值时，线程才能获得锁。

next 机制类似于一个排队系统，线程首先获取一个凭证（ticket），然后等待该凭证被分发。这种设计确保了：如果线程B在线程A之后获取ticket，那么B的ticket值必定大于A的，即ticket的分配是严格单调递增的。释放锁时，sync会执行加1操作，从而确保ticket值较小的线程总能优先获得锁，实现了公平的锁获取顺序。

加锁算法

当线程申请锁时，首先需要获取一个唯一的ticket作为排队凭证。这个操作很简单：通过原子递增next整型变量来获取当前凭证值，每个线程都能获得一个按顺序递增的独立编号，也就是根据凭证的数值大小就为这些线程排好了队。

获取凭证后，线程会持续将自身的ticket与sync变量的当前值进行比对：当两者相等时，表示已轮到自己获取锁；否则，线程将保持自旋状态直到条件满足。这种机制确保了线程按申请顺序公平地获取锁资源。

解锁算法

非常简单，就是通过原子操作将 sync 值递增 1，让等待sync+1值的线程获得自旋锁。虽然所有线程都在自旋等待，但它们各自监测的是不同的 sync 值。这种设计避免了多个线程同时竞争同一个值的情况，确保每个线程只在自己对应的凭证 ticket 值等于 sync 值时才会获得锁。由于 ticket 是按先到先得原则分配的，因此该算法实现了公平的自旋锁机制。

此外，next和sync都是32位的整型原子变量。当它们持续递增时可能发生溢出问题：一旦数值溢出回绕，就可能与之前线程的ticket值重复。这种情况下，当锁被释放时，可能有两个线程的ticket值与当前值匹配，导致两个线程同时被唤醒并进入临界区。

然而这种场景在现实中几乎不可能发生。假设两个线程的ticket值相同，这意味着系统需要同时运行2^32=4G个线程。由于线程获取自旋锁时严格按ticket顺序排队，不会出现next线程仍在等待而next+1线程却已获得锁的情况。只要存在等待ticket的线程，就表明后续所有线程（包括已溢出环绕的next）都处于等待状态。

实际应用中，4G个线程同时自旋是不可行的。按每个线程占用8M内存计算，4G线程将消耗32PB内存，应用程序在启动前就会因内存不足而崩溃。因此，在常规编程实践中根本不会出现这种情况。如果将next和sync定义为64位的atomic类型，发生溢出环绕错误的情况就更加不可能了。

优化本地自旋
while (sync.load(std::memory_order_acquire) != ticket) 是一种本地自旋操作。如前文所述，这种操作会导致CPU高功耗。通常我们会使用pause指令来暂停执行，或者采用回退策略以减少空转时的电量消耗。实现代码如下：

    const int SPIN_TIMES = 64;void lock() {int loops = 0;int ticket = next.fetch_add(1, std::memory_order_relaxed);while (sync.load(std::memory_order_acquire) != ticket) {if (loops++ == SPIN_TIMES) {loops = 0;asm("pause");}}}

缺点
本该算法虽然确保了锁的公平性，但存在明显的性能隐患。当某个排队等待锁的线程被调度出去时，即便前一个线程已释放锁，由于该线程无法及时获取锁（仍处于调度状态），后续所有持有更大ticket编号的线程都会被阻塞。这些线程必须等待被调出的线程重新获得CPU时间片并完成锁的获取与释放操作，从而导致整个系统的线程性能因这一次上下文切换而普遍下降。

参考：