【C++】锁与原子操作（四）：自旋锁的完整实现与性能优化

前面已经介绍了原子操作、CAS 和 C 内存序。本文继续分析一个完整的自旋锁实现。自旋锁的核心思想是当锁已经被其他线程占用时当前线程不进入休眠而是在原地循环检查锁的状态直到锁被释放。一个基础自旋锁主要需要提供以下接口spinlock_init(); // 初始化自旋锁 spinlock_lock(); // 阻塞式获取锁 spinlock_trylock(); // 尝试获取锁 spinlock_unlock(); // 释放锁 spinlock_destroy(); // 销毁锁本文将结合这些接口完整分析自旋锁从创建到使用的过程。一、自旋锁是如何表示锁状态的自旋锁只需要保存一个简单的状态0锁当前空闲 1锁已经被某个线程占用因此可以使用一个原子整数表示锁#include atomic struct spinlock { // 0表示锁空闲1表示锁已经被占用 std::atomic_int lock; };初始化时将锁设置为0static inline void spinlock_init( struct spinlock* lock) { // 初始化为0表示当前没有线程持有锁 std::atomic_init(lock-lock, 0); }多个线程争抢锁时目标都是尝试完成下面的修改lock0 → 1由于这个修改必须具有原子性因此不能使用普通赋值// 错误检查和修改不是一个原子操作 if (lock-lock 0) { lock-lock 1; }假设两个线程同时执行上面的代码线程A读取lock0 线程B读取lock0 线程A将lock改成1 线程B也将lock改成1两个线程都会认为自己获得了锁最终同时进入临界区自旋锁便失去了作用。正确做法是使用原子交换lock-lock.exchange( 1, std::memory_order_acquire );exchange()会完成两件事将原子变量设置成新值1 返回修改前的旧值。例如std::atomicint value{0}; int old_value value.exchange(1);执行以后old_value 0 value 1因此可以根据旧值判断是否成功获得锁int old_value lock-lock.exchange(1); if (old_value 0) { // 原来是空闲状态本线程成功获得锁 } else { // 原来已经是1说明锁被其他线程占用 }二、自旋锁为什么使用双层循环最简单的自旋锁可以写成static inline void spinlock_lock( struct spinlock* lock) { while (lock-lock.exchange( 1, std::memory_order_acquire)) { // 获取失败后不断重新尝试 } }这段代码能够保证正确性但存在一个性能问题。只要锁没有被释放所有等待线程都会不断执行lock-lock.exchange(1);exchange()不只是读取锁状态还会尝试执行写操作。多个处理器核心不断写同一个原子变量会频繁争抢该变量所在的缓存行增加缓存一致性开销。因此可以将加锁过程分成两步第一步使用exchange真正尝试获取锁 第二步失败后先使用普通原子读取等待 直到观察到锁变为0再重新使用exchange。完整实现如下static inline void spinlock_lock( struct spinlock* lock) { for (;;) { /* * exchange将lock设置为1 * 同时返回修改前的值。 * * 返回0 * 原来的锁是空闲状态加锁成功。 * * 返回1 * 锁已经被其他线程占用加锁失败。 */ if (!lock-lock.exchange( 1, std::memory_order_acquire)) { return; } /* * 获取锁失败后不要立即重复执行exchange。 * * 这里只使用load读取锁状态 * 等待锁从1恢复成0。 */ while (lock-lock.load( std::memory_order_relaxed)) { // 当前线程继续自旋 } } }这种方式通常称为Test-and-Test-and-Set 先读状态再尝试修改整个执行过程为执行exchange尝试加锁 ↓ 返回0 ↓ 成功获得锁 执行exchange尝试加锁 ↓ 返回1 ↓ 使用load循环读取锁状态 ↓ 发现锁变成0 ↓ 重新执行exchange争抢锁为什么不能只通过load()判断锁为空然后直接进入临界区// 错误示例 while (lock-lock.load() ! 0) { } lock-lock.store(1); // 进入临界区因为从load()看到锁为空到执行store(1)之间其他线程也可能同时观察到锁为空。因此load只能用于等待 真正获取锁仍然必须使用exchange或CAS。三、_mm_pause()为什么能优化自旋等待在 x86-64 平台上自旋循环中通常会加入_mm_pause();需要包含头文件#include immintrin.h优化后的自旋锁如下static inline void spinlock_lock( struct spinlock* lock) { for (;;) { if (!lock-lock.exchange( 1, std::memory_order_acquire)) { return; } while (lock-lock.load( std::memory_order_relaxed)) { /* * 告诉处理器当前处于自旋等待状态 * 减少空循环带来的处理器资源消耗。 */ _mm_pause(); } } }_mm_pause()并不会让线程真正休眠也不会主动把 CPU 交给其他线程。它更像是向处理器发出提示当前代码正在等待一个共享状态发生变化不需要按照普通计算循环积极执行。在支持超线程的处理器上同一个物理核心可能同时运行两个逻辑线程。如果一个逻辑线程执行紧密的空循环while (lock-lock.load()) { }它可能占用较多执行资源影响同一物理核心上的另一个逻辑线程。加入_mm_pause()后可以适当降低这种影响。为了兼容不同平台可以使用宏#if defined(__x86_64__) #include immintrin.h // x86-64平台使用pause指令 #define atomic_pause_() _mm_pause() #else // 其他平台暂时不执行额外操作 #define atomic_pause_() ((void)0) #endif加锁函数可以统一写成static inline void spinlock_lock( struct spinlock* lock) { for (;;) { if (!lock-lock.exchange( 1, std::memory_order_acquire)) { return; } while (lock-lock.load( std::memory_order_relaxed)) { atomic_pause_(); } } }需要注意_mm_pause()只能降低自旋等待的部分成本不能改变自旋锁本身的特点。如果锁被持有很长时间等待线程仍然会不断占用 CPU。四、尝试加锁、解锁与兼容封装1.spinlock_trylock()普通的spinlock_lock()会一直等待直到成功获得锁。有些场景并不希望线程阻塞式自旋而是只尝试一次if (spinlock_trylock(lock)) { // 成功获得锁 } else { // 锁正被占用执行其他操作 }实现如下static inline int spinlock_trylock( struct spinlock* lock) { /* * 第一步先读取锁状态。 * * 如果当前已经是1就没有必要立即执行exchange。 */ if (lock-lock.load( std::memory_order_relaxed)) { return 0; } /* * 即使刚才读取到0也不能直接认为加锁成功。 * * 因为其他线程可能在load之后抢先获得锁 * 所以仍然需要通过exchange完成最终竞争。 */ return !lock-lock.exchange( 1, std::memory_order_acquire ); }也可以写成static inline int spinlock_trylock( struct spinlock* lock) { return !lock-lock.load( std::memory_order_relaxed) !lock-lock.exchange( 1, std::memory_order_acquire); }这里利用了逻辑与的短路特性。如果第一次读取发现锁已经是1后面的exchange()不会执行。假设两个线程同时调用trylock()并且都先读取到0线程Aload得到0 线程Bload得到0它们随后都会执行exchange(1)。但原子交换保证只有一个线程能够看到旧值为0线程Aexchange返回0加锁成功 线程Bexchange返回1加锁失败因此前面的load()只是性能优化真正保证正确性的仍然是后面的原子交换。2.spinlock_unlock()解锁时将锁从1设置回0static inline void spinlock_unlock( struct spinlock* lock) { lock-lock.store( 0, std::memory_order_release ); }为什么加锁使用acquire解锁使用release加锁 acquire 保证获得锁以后能够看到上一个持锁线程 在解锁前对共享数据完成的修改。 解锁 release 保证当前线程临界区中的修改 在锁变为0之前完成并发布。例如int shared_data 0; spinlock lock; void writer() { spinlock_lock(lock); // 临界区内修改共享数据 shared_data 100; spinlock_unlock(lock); } void reader() { spinlock_lock(lock); /* * 成功获得锁以后可以看到 * 上一个线程释放锁前写入的shared_data。 */ int value shared_data; spinlock_unlock(lock); }锁上的release和后续成功加锁的acquire建立同步关系从而保护临界区中的普通共享数据。3.spinlock_destroy()原子变量本身不需要释放操作系统资源因此销毁函数可以为空static inline void spinlock_destroy( struct spinlock* lock) { /* * 当前原子自旋锁没有动态内存 * 也没有需要销毁的系统锁对象。 */ (void)lock; }虽然函数没有真正执行销毁操作但保留统一接口有两个好处调用代码结构保持一致 以后切换成pthread互斥锁时不需要修改使用方式。4. 使用宏统一调用方式可以定义几个辅助宏#define SPIN_INIT(object) \ spinlock_init((object)-lock) #define SPIN_LOCK(object) \ spinlock_lock((object)-lock) #define SPIN_UNLOCK(object) \ spinlock_unlock((object)-lock) #define SPIN_DESTROY(object) \ spinlock_destroy((object)-lock)例如有一个任务队列struct task_queue { spinlock lock; int task_count; };可以这样使用task_queue queue; // 初始化queue中的自旋锁 SPIN_INIT(queue); // 加锁 SPIN_LOCK(queue); queue.task_count; // 解锁 SPIN_UNLOCK(queue); // 销毁 SPIN_DESTROY(queue);这样上层代码不需要反复编写spinlock_lock(queue.lock);五、完整代码、兼容方案与使用场景下面给出一个可以直接使用的自旋锁头文件#ifndef SIMPLE_SPINLOCK_H #define SIMPLE_SPINLOCK_H #include atomic #if defined(__x86_64__) || defined(_M_X64) #include immintrin.h /* * x86-64平台使用pause指令 * 优化紧密的自旋等待循环。 */ #define SPIN_PAUSE() _mm_pause() #else #define SPIN_PAUSE() ((void)0) #endif struct spinlock { // 0表示空闲1表示已被占用 std::atomic_int lock; }; /** * 初始化自旋锁 */ static inline void spinlock_init( struct spinlock* lock) { std::atomic_init(lock-lock, 0); } /** * 阻塞式获取自旋锁 */ static inline void spinlock_lock( struct spinlock* lock) { for (;;) { /* * 真正尝试将锁从0修改为1。 */ if (!lock-lock.exchange( 1, std::memory_order_acquire)) { return; } /* * 获取失败后先使用只读方式等待。 */ while (lock-lock.load( std::memory_order_relaxed)) { SPIN_PAUSE(); } } } /** * 尝试获取自旋锁 * * 成功返回1 * 失败立即返回0。 */ static inline int spinlock_trylock( struct spinlock* lock) { return !lock-lock.load( std::memory_order_relaxed) !lock-lock.exchange( 1, std::memory_order_acquire); } /** * 释放自旋锁 */ static inline void spinlock_unlock( struct spinlock* lock) { lock-lock.store( 0, std::memory_order_release ); } /** * 原子自旋锁不持有额外系统资源 */ static inline void spinlock_destroy( struct spinlock* lock) { (void)lock; } #endif下面使用自旋锁保护 4 个线程共同累加一个普通整数#include iostream #include thread #include spinlock.h // 全局自旋锁 spinlock count_lock; // 被多个线程共同修改的普通变量 int count 0; /** * 将count增加指定次数 */ void increase(int number) { for (int i 0; i number; i) { // 进入临界区前加锁 spinlock_lock(count_lock); count; // 离开临界区后解锁 spinlock_unlock(count_lock); } } int main() { // 使用前初始化自旋锁 spinlock_init(count_lock); std::thread t1(increase, 500); std::thread t2(increase, 500); std::thread t3(increase, 500); std::thread t4(increase, 500); t1.join(); t2.join(); t3.join(); t4.join(); std::cout count count \n; // 使用结束后调用统一销毁接口 spinlock_destroy(count_lock); return 0; }最终结果为count 2000除了标准原子操作还可以为不支持标准原子库的编译环境提供编译器内置函数版本typedef int atomic_flag_; #define ATOMIC_FLAG_INIT_ 0 #define atomic_flag_test_and_set_(ptr) \ __sync_lock_test_and_set(ptr, 1) #define atomic_flag_clear_(ptr) \ __sync_lock_release(ptr) struct spinlock { atomic_flag_ lock; }; static inline void spinlock_init( struct spinlock* lock) { lock-lock ATOMIC_FLAG_INIT_; } static inline void spinlock_lock( struct spinlock* lock) { while (atomic_flag_test_and_set_( lock-lock)) { } } static inline int spinlock_trylock( struct spinlock* lock) { return atomic_flag_test_and_set_( lock-lock) 0; } static inline void spinlock_unlock( struct spinlock* lock) { atomic_flag_clear_(lock-lock); }还可以通过编译宏将同一套接口切换为互斥锁#include pthread.h struct spinlock { pthread_mutex_t lock; }; static inline void spinlock_init( struct spinlock* lock) { pthread_mutex_init( lock-lock, nullptr ); } static inline void spinlock_lock( struct spinlock* lock) { pthread_mutex_lock(lock-lock); } static inline int spinlock_trylock( struct spinlock* lock) { return pthread_mutex_trylock( lock-lock) 0; } static inline void spinlock_unlock( struct spinlock* lock) { pthread_mutex_unlock(lock-lock); } static inline void spinlock_destroy( struct spinlock* lock) { pthread_mutex_destroy(lock-lock); }这样上层代码始终使用spinlock_lock(lock); spinlock_unlock(lock);底层既可以选择原子自旋锁也可以选择pthread_mutex。使用自旋锁时需要注意以下问题1. 自旋锁适合非常短的临界区 2. 临界区中不要进行网络请求、文件读写和休眠 3. 等待期间线程会继续占用CPU 4. 高竞争情况下互斥锁可能比自旋锁更合适 5. 自旋锁通常不保证线程获取锁的公平性 6. 同一个线程重复获取普通自旋锁可能造成死锁 7. 成功加锁后必须保证最终执行解锁 8. 对简单计数器直接使用atomic通常更加合适。这一篇的核心可以总结为自旋锁使用原子变量表示锁状态 exchange负责真正争抢锁 load负责在锁被占用时低成本等待 双层自旋可以减少频繁原子写造成的缓存竞争 _mm_pause可以优化处理器中的紧密等待循环 加锁使用acquire解锁使用release trylock只尝试一次失败后立即返回 统一接口可以在自旋锁和互斥锁之间切换。0voice · GitHub