C++ 自旋锁（Spinlock）的性能调优：CPU 缓存与退避策略

好的，各位观众，欢迎来到今天的“C++ 自旋锁性能调优：CPU 缓存与退避策略”讲座！

今天咱们不讲那些枯燥的理论，直接上干货，用大白话聊聊自旋锁这玩意儿，以及怎么让它跑得飞起。

一、啥是自旋锁？别告诉我你没听过！

想象一下，你去银行取钱，只有一个柜台，如果前面有人在办理，你是不是只能站在那儿“自旋”等待？这就是自旋锁的本质。

在多线程编程中，自旋锁是一种锁机制，当一个线程试图获取一个已经被其他线程持有的锁时，它不会立即进入睡眠状态，而是不断地循环检查锁是否释放，直到获取到锁为止。这种循环检查的过程就叫做“自旋”。

自旋锁的优点是避免了线程上下文切换的开销（因为线程一直处于运行状态），但缺点是如果锁被长时间占用，会浪费大量的 CPU 资源。

二、自旋锁的简单实现：写个简陋的玩具

先来一个最最最简单的自旋锁实现，让你感受一下：

#include <atomic>
#include <thread>
#include <iostream>

class SimpleSpinLock {
private:
  std::atomic_flag locked = ATOMIC_FLAG_INIT; // 原子标志，用于表示锁的状态

public:
  void lock() {
    while (locked.test_and_set(std::memory_order_acquire)) {
      // 自旋等待，直到锁被释放
    }
  }

  void unlock() {
    locked.clear(std::memory_order_release);
  }
};

// 使用示例
int counter = 0;
SimpleSpinLock spinLock;

void incrementCounter() {
  for (int i = 0; i < 100000; ++i) {
    spinLock.lock();
    counter++;
    spinLock.unlock();
  }
}

int main() {
  std::thread t1(incrementCounter);
  std::thread t2(incrementCounter);

  t1.join();
  t2.join();

  std::cout << "Counter value: " << counter << std::endl; // 预期结果：200000
  return 0;
}

这段代码里，std::atomic_flag 是关键，它保证了锁操作的原子性。 test_and_set() 尝试设置标志，如果设置成功（之前是未设置状态），就表示获取到了锁，否则就继续自旋。 clear() 释放锁。

这个玩具虽然能用，但是性能嘛… 只能说“能用”和“好用”之间隔着十万八千里。

三、CPU 缓存：自旋锁性能的幕后推手

自旋锁的性能，很大程度上受到 CPU 缓存的影响。要理解这一点，先要简单了解一下 CPU 缓存的结构。

CPU 缓存分为 L1、L2、L3 缓存，L1 缓存速度最快，但容量最小，L3 缓存速度最慢，但容量最大。CPU 访问数据的顺序是：L1 -> L2 -> L3 -> 内存。

当多个 CPU 核心同时访问同一个内存地址时，会涉及到缓存一致性协议（例如 MESI 协议）。简单的说，如果一个核心修改了某个数据，其他核心的缓存中对应的缓存行就会失效，需要重新从内存或者其他核心的缓存中加载数据。

3.1 缓存行伪共享（False Sharing）：性能杀手

缓存行伪共享是指，多个线程访问的数据位于同一个缓存行中，即使它们访问的是不同的变量。当其中一个线程修改了数据时，会导致其他线程的缓存行失效，从而引发不必要的缓存同步，降低性能。

在自旋锁的场景下，如果锁变量和其他变量位于同一个缓存行中，当一个线程持有锁时，其他线程自旋等待，不断地读取锁变量，这会导致缓存行频繁失效，造成严重的性能问题。

举个栗子：

struct Data {
  std::atomic_flag lock = ATOMIC_FLAG_INIT;
  int value;
};

Data data;

void worker(int id) {
  for (int i = 0; i < 1000000; ++i) {
    data.lock.lock();
    data.value++;
    data.lock.unlock();
  }
  std::cout << "Thread " << id << " finished." << std::endl;
}

int main() {
  std::thread t1(worker, 1);
  std::thread t2(worker, 2);

  t1.join();
  t2.join();

  std::cout << "Final value: " << data.value << std::endl;
  return 0;
}

在这个例子中，lock 和 value 位于同一个结构体中，很可能在同一个缓存行中。当一个线程持有锁，修改 value 时，会导致另一个线程的缓存行失效，即使它只是在自旋等待锁的释放。

3.2 如何避免缓存行伪共享？

填充（Padding）： 在锁变量和其他变量之间添加足够的填充，使它们位于不同的缓存行中。缓存行的大小通常是 64 字节，所以可以使用 alignas(64) 来保证变量的对齐。
```
struct Data {
  std::atomic_flag lock = ATOMIC_FLAG_INIT;
  char padding[64 - sizeof(std::atomic_flag)]; // 填充
  int value;
};
```
使用线程局部变量（Thread-Local Storage）： 如果可能，尽量使用线程局部变量，避免多个线程访问同一个内存地址。

四、退避策略（Backoff Strategies）：让自旋更优雅

纯粹的自旋等待会消耗大量的 CPU 资源。为了缓解这个问题，可以引入退避策略，让线程在自旋等待一段时间后，主动让出 CPU 资源。

4.1 指数退避（Exponential Backoff）：

指数退避是一种常用的退避策略。线程在每次自旋失败后，等待的时间呈指数增长。这样可以减少线程之间的竞争，降低 CPU 占用率。

#include <atomic>
#include <thread>
#include <chrono>
#include <iostream>
#include <random>

class ExponentialBackoffSpinLock {
private:
  std::atomic_flag locked = ATOMIC_FLAG_INIT;
  std::mt19937 rng; // 随机数生成器

public:
  ExponentialBackoffSpinLock() : rng(std::random_device{}()) {}

  void lock() {
    int delay = 1;
    while (locked.test_and_set(std::memory_order_acquire)) {
      std::this_thread::sleep_for(std::chrono::microseconds(delay));
      delay = std::min(1024, delay * 2); // 限制最大延迟
    }
  }

  void unlock() {
    locked.clear(std::memory_order_release);
  }
};

// 使用示例
int counter = 0;
ExponentialBackoffSpinLock spinLock;

void incrementCounter() {
  for (int i = 0; i < 100000; ++i) {
    spinLock.lock();
    counter++;
    spinLock.unlock();
  }
}

int main() {
  std::thread t1(incrementCounter);
  std::thread t2(incrementCounter);

  t1.join();
  t2.join();

  std::cout << "Counter value: " << counter << std::endl; // 预期结果：200000
  return 0;
}

在这个例子中，每次自旋失败后，线程会等待的时间翻倍，直到达到最大延迟（1024 微秒）。

4.2 随机退避（Random Backoff）：

随机退避是在指数退避的基础上，引入随机性。线程在每次自旋失败后，等待一个随机的时间。这样可以避免多个线程同时进入睡眠状态，然后又同时醒来，再次发生竞争。

#include <atomic>
#include <thread>
#include <chrono>
#include <iostream>
#include <random>

class RandomBackoffSpinLock {
private:
  std::atomic_flag locked = ATOMIC_FLAG_INIT;
  std::mt19937 rng; // 随机数生成器
  std::uniform_int_distribution<> distribution;

public:
  RandomBackoffSpinLock() : rng(std::random_device{}()), distribution(1, 1024) {}

  void lock() {
    while (locked.test_and_set(std::memory_order_acquire)) {
      int delay = distribution(rng); // 生成一个 1 到 1024 的随机数
      std::this_thread::sleep_for(std::chrono::microseconds(delay));
    }
  }

  void unlock() {
    locked.clear(std::memory_order_release);
  }
};

// 使用示例
int counter = 0;
RandomBackoffSpinLock spinLock;

void incrementCounter() {
  for (int i = 0; i < 100000; ++i) {
    spinLock.lock();
    counter++;
    spinLock.unlock();
  }
}

int main() {
  std::thread t1(incrementCounter);
  std::thread t2(incrementCounter);

  t1.join();
  t2.join();

  std::cout << "Counter value: " << counter << std::endl; // 预期结果：200000
  return 0;
}

4.3 std::this_thread::yield()：主动让出 CPU 资源

std::this_thread::yield() 可以让线程主动让出 CPU 资源，让其他线程有机会运行。

#include <atomic>
#include <thread>
#include <iostream>

class YieldSpinLock {
private:
  std::atomic_flag locked = ATOMIC_FLAG_INIT;

public:
  void lock() {
    while (locked.test_and_set(std::memory_order_acquire)) {
      std::this_thread::yield(); // 主动让出 CPU 资源
    }
  }

  void unlock() {
    locked.clear(std::memory_order_release);
  }
};

// 使用示例
int counter = 0;
YieldSpinLock spinLock;

void incrementCounter() {
  for (int i = 0; i < 100000; ++i) {
    spinLock.lock();
    counter++;
    spinLock.unlock();
  }
}

int main() {
  std::thread t1(incrementCounter);
  std::thread t2(incrementCounter);

  t1.join();
  t2.join();

  std::cout << "Counter value: " << counter << std::endl; // 预期结果：200000
  return 0;
}

五、自旋锁 vs 互斥锁（Mutex）：选哪个？

自旋锁和互斥锁都是用于保护共享资源的锁机制，但它们的适用场景不同。

特性	自旋锁	互斥锁
线程等待方式	自旋等待，不断循环检查锁是否释放	阻塞等待，进入睡眠状态，等待操作系统唤醒
CPU 占用	如果锁被长时间占用，会浪费大量的 CPU 资源	占用较少 CPU 资源，因为线程处于睡眠状态
上下文切换	避免了线程上下文切换的开销	可能引起线程上下文切换，有一定开销
适用场景	锁占用时间短，竞争不激烈的场景	锁占用时间长，竞争激烈的场景
优先级反转	可能导致优先级反转（高优先级线程等待低优先级线程释放锁）	可以通过优先级继承等机制避免优先级反转

简单总结：

锁占用时间短，竞争不激烈： 自旋锁
锁占用时间长，竞争激烈： 互斥锁

六、CAS (Compare-and-Swap) 操作：自旋锁的基石

自旋锁的实现，通常依赖于原子操作，而 CAS 操作是原子操作中最常用的一种。

CAS 操作包含三个操作数：内存地址、期望值和新值。它会原子地比较内存地址的值和期望值，如果相等，则将内存地址的值更新为新值，否则不做任何操作。

std::atomic_flag 的 test_and_set() 和 clear() 方法，底层就是基于 CAS 操作实现的。

七、代码优化：让你的自旋锁更上一层楼

使用 std::atomic_flag： std::atomic_flag 是 C++ 标准库提供的原子标志，专门用于实现自旋锁，性能比自己实现的原子变量更好。
避免不必要的内存访问： 在自旋等待时，尽量减少对共享内存的访问，可以先读取一个本地变量，然后再判断锁是否释放。
使用合适的内存顺序（Memory Order）： std::memory_order_acquire 和 std::memory_order_release 是最常用的内存顺序，可以保证锁操作的正确性。

八、总结：自旋锁调优的葵花宝典

理解 CPU 缓存： 了解缓存行伪共享，并采取措施避免。
选择合适的退避策略： 根据实际情况选择指数退避、随机退避或 std::this_thread::yield()。
选择合适的锁机制： 根据锁占用时间和竞争程度，选择自旋锁或互斥锁。
代码优化： 使用 std::atomic_flag，避免不必要的内存访问，使用合适的内存顺序。

九、彩蛋：永远不要忘记测试！

理论再完美，也要经过实践的检验。在实际应用中，一定要进行充分的测试，才能找到最佳的自旋锁配置。

可以使用性能分析工具（例如 gprof、perf）来测量自旋锁的性能，并根据测试结果进行调整。

好了，今天的讲座就到这里。希望大家能够掌握自旋锁性能调优的技巧，写出高性能的多线程程序！感谢大家！

发表回复 取消回复

发表回复取消回复