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

C++自旋锁的后退机制简介

java 2025/8/19 4:02:49

看一下TAS算法实现的自旋锁。

volatile int cnt = 0;class spin_lock {
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;
};

如果一个线程在调用lock()方法时，迟迟获取不到锁，说明该锁肯定存在高争用：即多个线程试图同时获取同一个锁。显然，此时线程获得锁的机会非常小，试图获得一个存在高争用的锁是一种应该回避的情形，因为这种自旋重试会导致总线流量的增加。如果遇到了这种情况，若让线程“后退”一下，给正在自旋中的线程以结束的机会，将会更加有效。因此，一个线程每当发现锁可能空闲但却无法获得它时，可以在重试之前主动后退，以确保发生冲突的并发线程不进入剧烈竞争状态，即在同一时刻所有线程都试图获得锁，该线程应该随机的后退一段时间。

所谓后退机制（backoff），也就是如果TAS申请不到锁，那就先主动后退一段时间，然后再回来竞争。下面看一下后退的机制。

本地自旋

下面是使用TTAS算法实现的自旋锁。它实际上也是一种后退机制，即如果TAS没有成功，那就后退到本地自旋，本地自旋时所检查的锁状态数据是在CPU本地的cache中，此时读取时不会和其它线程竞争内存总线。只有在本地自旋中检测到有可能会获得锁时，再次使用TAS算法进行竞争，这样能够降低大家都在TAS竞争时对锁释放操作的影响。这种后退机制完全是由用户层的软件代码来完成的，不涉及到其它因素。

下面是实现代码：

void lock() {while (true) {if (!flag.test_and_set(memory_order_acquire)) {break;}// 如果TAS失败，进行退让，后退到本地自旋while (flag.test(memory_order_relaxed));}
}

不过，当test()测试为false时，指令流要向后跳转到test_and_set()去申请锁，CPU的指令流水线可能需要重新加载指令，执行单元需要等待加载、译码等准备工作，会造成执行性能损失。在高争用的场景下，线程是大概率获取不到锁的，既然test_and_set()大概率要失败，从而进入后续的test()自旋，与其这样，不如为了指令执行时有更好的局部性，改用下面的方式：

void lock() {while (true) {while (flag.test(memory_order_relaxed));if (!flag.test_and_set(memory_order_acquire)) {break;}}
}

即把test()放到test_and_set()前面执行，这样当test()测试结果为false时，指令流不会发生跳转，而是直接进入test_and_set()去获取锁，省去了指令跳转造成的性能损失，让检查到flag的值为false到把flag设置为true，时间间隙更小一点。

不过TTAS的后退机制并不高效，lock()方法使用了两个步骤：它不断地用test()读锁状态，当锁可能空闲时，则去调用test_and_set()来获取锁。两步之间存在间隙，一个线程离开第一步时，有可能其它某个线程抢先获得了锁，意味着该锁极有可能存在高争用，那个线程只能回到第一步重新本地自旋。虽然所有线程在第一步test()成功后，全部都去竞争test_and_set()，也是高争用，但是因为大部分时间都在本地自旋，不会竞争内存总线，当线程释放锁时不需要竞争内存总线而能够立即释放锁（因为当锁被一个线程占有时，说明其它线程都在本地自旋中，访问的是CPU本地cahce，没有访问系统内存）。

pause指令

pause指令是另一种形式的后退机制，这种后退机制需要控制CPU，暂时让CPU暂停一下，由CPU自己决定什么时候从暂停中恢复。不过实现这种后退机制，要依赖于CPU是否提供了该功能的指令，比如x86处理器就支持这个功能，指令是：pause指令（有时候也汇编为：rep nop）。

lock()代码如下：

void lock() {while (flag.test_and_set(std::memory_order_acquire)) {asm("pause"); }
}

在提供超线程（hyperthread）的CPU上，一个自旋中的线程抢占了CPU核心的执行资源，有可能让原本共同使用这个CPU核心（即逻辑核）的另一个线程可能无法得到执行。此外，自旋导致CPU发热和电量高功耗是彻头彻尾的浪费，它纯粹是在做无用功，一直在制造热量之外，没有别的用处。

而pause指令是让一个CPU逻辑核暂停，把执行资源全部让给处于同一个物理核中的其它超线程使用，既然除了自旋忙等，什么也干不了，不如把执行资源让出来。这非常划算，如果有超线程应该考虑使用pause指令。

上面的两种后退机制都没有脱离开自旋锁的特征，即获取不到锁时线程就处于等待中，线程不会被阻塞。这些后退机制都是在用户层上进行的，如果自旋锁在线程获取不到锁时不想被阻塞，可以使用它们优化算法。

下面介绍的后退机制，就要求线程在无法得到锁时被调度或者阻塞了，此时CPU
会被调度去执行别的线程了。既然涉及到线程被调度或者被阻塞的情况，要实现这些后退机制，显然在用户层是无法做到的，就得需要操作系统内核参与了。

下面我们看一下操作系统参与进来的后退机制，首先是yield。

yield-让步方式

即线程让步，意思是当前线程主动让出CPU资源，让其它就绪中的同优先级线程先执行，在合适的时候再调度回来。C++11线程库有std::this_thread::yield()方法，当一个线程调用它时，如果有别的相同优先级的线程（优先级高的不用让就把它抢占了）处于就绪状态时，就把当前线程调度出去，如果没有，则线程继续执行，yield()不会产生任何效果，除了进入和离开一次操作系统内核之外。

lock()代码如下：

void lock() {while (flag.test_and_set(std::memory_order_acquire)) {std::this_thread::yield();}
}

在这里，当TAS算法失败后，就调用线程的yield()接口，进入操作系统内核，让内核进行调度。当调度回来之后，继续进行TAS算法，直到成功获得锁。显然，这样获得自旋锁的延迟就更大了，例如，如果释放锁和TAS算法是同时进行的，此时TAS再继续循环检查一次，可能就能获得锁了，但是因为线程直接调用yield()被调度出去了，只能等到再次被调度回来时才有可能去申请锁。

TAS一旦失败就让出CPU，简单粗暴了一些，yield操作有可能导致线程切换，那就失去了自旋锁的应用初衷，可以在自旋循环一定次数之后，再调用yield，毕竟线程切换要借助到系统调用和内核调度，是一个非常大的开销。

如果yield没有把CPU让出去，说明此时没有同优先级的线程在就绪队列中，那么线程就独占CPU，处于持续自旋中。它的坏处是CPU耗电量大，此时可以结合前面的pause指令共同完成后退策略；还有一个坏处就是尽管系统中还有线程处于就绪状态中，只不过是优先级较低，也无法被分配这个让出来的CPU，造成资源浪费，此时可以考虑使用下面的睡眠模式。

sleep-睡眠方式

强制自旋的线程睡眠一段时间，把CPU调度给别的线程使用，包括同优先级或低于它的优先级的线程。但是如果在睡眠时，别的线程已经释放锁了，自己睡眠时间没有到，也只能一直等到睡眠时间到达之后才有机会去申请锁，会延迟获得锁。C++11线程库有std::this_thread::sleep_for()方法，当一个线程调用它时，可以按照指定的参数休眠一段时间。

lock()代码如下：

void lock() {while (flag.test_and_set(std::memory_order_acquire)) {std::this_thread::sleep_for(std::chrono::milliseconds(5));}
}

阻塞-唤醒方式

也称为通知-等待方式。yield让步方式可能后退不成功，例如线程调用yield()进入OS之后，如果操作系统内核发现没有同优先级的就绪线程，线程会被频繁地在用户层和内核层切换；而sleep睡眠方式，因为sleep()是强制休眠，如果睡眠时间设置太长，可能别的线程已经释放锁了，被强制休眠的线程也无法及时去获取锁，如果睡眠时间设置太短获取不到锁时，同样，线程也会被频繁的调度。比较合理的后退方式是阻塞-唤醒方式：当获取不到锁时进入休眠，当有机会获得锁时，就被唤醒去申请锁，在一定程度上既能保证获得锁的及时性又能保证没有得到锁时CPU不会被自旋操作占用过长。

C++20原子类提供了wait()和notify_one()接口。wait()执行原子等待操作，如果等待的值和原子变量相等，则阻塞线程直到被notify_one()通知。notify_one()执行原子通知操作，如果有线程在原子变量上的wait()中被阻塞，则解除至少一个此类线程的阻塞。这两个接口是属于阻塞等待-唤醒的机制，这种形式的状态变化检测通常比纯自旋锁更有效，可以使用这两个接口来实现自旋锁的后退机制，即在自旋等不到锁释放时进入阻塞等待状态。

因此，需要lock()和unlock()配合起来实现，lock()得不到锁时调用wait()进入阻塞等待状态，当锁释放了，unlock()调用notify_one()负责唤醒线程去竞争锁，功能类似于条件变量通知和等待操作。如下所示：

class spin_lock {// 自旋变量同时也是等待变量 atomic_flag flag{false};public:void lock() {while (flag.test_and_set(memory_order_acquire)) {// 如果锁变量为true就阻塞等待flag.wait(true, memory_order_relaxed);}  }void unlock() {flag.clear(memory_order_release);flag.notify_one(); // 锁变量为false，并发送通知}
};

此时自旋锁状态变量flag同时也是等待变量，如果TAS操作没有成功，可以调用它的wait()方法进入阻塞等待状态。当释放锁时，先把锁变量flag的值设置为false，同时调用它的notify_one()方法，这样，如果某个线程处在flag.wait(true)等待中，可以被这个notify_once()唤醒。唤醒后为了防止是假唤醒，使用TAS算法把它由false设置为true，如果设置成功，则获得锁，否则说明是假唤醒或者被别的线程抢先获得锁，调用它的wait()继续阻塞等待。

实际上，该机制虽然是自旋锁的后退机制方案，你会发现此时自旋锁就具有了和阻塞锁一样的特性：如果得不到锁就被阻塞，那么自旋锁的表现行为就是阻塞锁了，如同C++11线程库中的std::mutex。

后退机制总结

介绍了5种后退机制，前两种是在用户层进行的，线程不会阻塞，第一种是纯指令的软件实现的本地自旋，防止内存总线出现流量风暴，“后退”让出的是内存总线资源，而第二种需要CPU的特殊状态，以节省CPU耗电量，“后退”让出的是CPU逻辑核的执行资源。剩下的三种都需要操作系统介入了，因为涉及到线程的调度和阻塞，“后退”让出的都是CPU资源。让线程申请不到锁就进入阻塞状态，一是为了防止自旋锁在高争用时线程过度占有CPU资源，不做无意义的无用功，二是即使自旋锁在低争用时，如果占有锁的线程恰好被调度出去了，能让正在申请锁的线程暂时放弃CPU。

可以把几种后退机制放在一起考虑，通常要使用本地自旋，如果CPU支持pause指令一般也会使用，这两者可以组合成一个不涉及到线程阻塞的自旋锁。如果自旋锁保护的临界区非常小，执行时间非常短，很少发生高争用的情况，可以使用此方式的后退机制。

如果需要线程让出CPU资源，可以在这个基础再加上yield、sleep或者wait-notify中的一个，组成一个综合的后退机制。如果保护的临界区执行时间长，或者是高争用的情况，应该考虑使用这些方式的后退机制，以降低长时间自旋时过多的占有CPU和过高的消耗电量。

考虑到自旋锁保护的临界区一般很小，执行时间很短，如果TAS失败之后，说明有别的线程已获得锁并正处于临界区中，正常情况下应该很快就会离开临界区并释放锁，可能接下来几个自旋循环就能释放锁。因此等待中的线程再自旋一定次数内大概率是能等到锁释放的，一般会让线程再继续进行一定循环次数的自旋。如果循环次数用完了还没有得到锁，说明占用锁的那个线程大概率是被抢占了，此时可能会长时间的等待下去，继续自旋下去意义就不大了，不如使用yield、sleep或者wait-notify中的某个后退机制，让线程进入阻塞状态，把CPU调度给别的线程使用。

通过这种机制，让TAS返回失败时不再无条件的进入阻塞状态以释放CPU资源，而是让TAS返回失败和进入阻塞状态之间有了一小段时间的缓冲。通过这种方式，可以平衡申请锁时的延迟过长（线程上下文切换）和CPU空耗过大（长时间自旋）的情况。

常见平台的自旋锁实现

下面是一些平台所实现的自旋锁加锁函数的代码，也采用了上面介绍的后退实现方式的一种或者几种，只是选择有所不同，其中nginx还对pause的次数进行定制，执行一定次数之后还会调用yield，让出CPU的资源，大家不妨分析一下：

1、DPDK，使用了TTAS算法，同时也使用了pause指令后退机制

typedef struct {/**< lock status 0 = unlocked, 1 = locked */volatile int locked;
} rte_spinlock_t;static inline void
rte_spinlock_lock(rte_spinlock_t *sl) {while (__sync_lock_test_and_set(&sl->locked, 1))while(sl->locked)rte_pause();
}

2、NGINX（去掉了一些不相关代码），也是TTAS算法+pause
指令后退机制，在几次pause之后，如果仍然获取到不到锁，就调用yield让出CPU。

void ngx_spinlock(ngx_atomic_t *lock, ngx_atomic_int_t value, ngx_uint_t spin)
{ngx_uint_t  i, n;for ( ;; ) {if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {return;}if (ngx_ncpu > 1) {for (n = 1; n < spin; n <<= 1) {for (i = 0; i < n; i++) {ngx_cpu_pause();}if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {return;}}}ngx_sched_yield();}
}

受这些实现算法的启发，我们也可以在x86平台上，采用这样的实现方案：如果获取不到锁，先本地自旋，在自旋一定次数之后获还取不到锁就pause，pause一定次数之后还仍然获取不到锁，则最后再阻塞。

class spin_lock {atomic_flag flag{false};public:void lock() {while (flag.test_and_set(memory_order_acquire)) {// pause16次之后进入阻塞等待for (int i=0; i<16; i++) {int cnt = 0;// 自旋64次之后调用pausewhile (flag.test(memory_order_relaxed) && cnt++ < 64);if (flag.test_and_set(memory_order_acquire))asm("pause");elsereturn;}// 进入阻塞等待时做最后的尝试if (flag.test_and_set(memory_order_acquire))flag.wait(true);elsereturn;}}void unlock() {flag.clear(memory_order_release);flag.notify_one();}
};