C++ 原子操作的底层指令：`compare_exchange` 的硬件支持

好的，各位观众老爷们，今天咱们来聊聊C++原子操作里一个非常关键，但又经常让人挠头的家伙：compare_exchange。这玩意儿，说白了，就是原子操作里的“乾坤大挪移”，能让你在并发编程的世界里，以一种非常优雅（有时候也很痛苦）的方式，保证数据的安全性和一致性。

一、啥是compare_exchange？—— 故事的开始

想象一下，你和你的小伙伴同时想给一个银行账户增加存款。如果你们各自先读取账户余额，然后各自加上自己的存款，最后再写回账户，那就会出现问题，也就是经典的“竞态条件”。

compare_exchange就是来解决这个问题的。它的工作原理是：

比较（Compare）： 首先，它会比较当前内存地址的值是不是等于你期望的旧值（expected）。
交换（Exchange）： 如果相等，它就会把内存地址的值替换成你期望的新值（desired）。
原子性（Atomicity）： 整个比较和交换的过程是原子性的，也就是说，在执行这个操作的过程中，不会被其他线程打断。

是不是有点像武侠小说里的乾坤大挪移？先看看对方是不是你要找的那个“目标”，如果是，就把“目标”替换成新的“目标”。

二、compare_exchange 的两种形态：weak 和 strong

C++标准库提供了两种compare_exchange函数：compare_exchange_weak 和 compare_exchange_strong。它们之间的区别在于：

compare_exchange_strong： 保证原子性的比较和交换一定成功，除非当前内存地址的值和期望的旧值不相等。也就是说，如果比较成功，那么交换一定成功。
compare_exchange_weak： 允许出现伪失败（spurious failure）。即使当前内存地址的值和期望的旧值相等，交换也可能失败。听起来有点坑爹，对吧？但是，compare_exchange_weak在某些情况下性能更好，因为它可以利用CPU的一些优化机制。

那么，什么时候用weak，什么时候用strong呢？

一般来说，如果你的代码需要绝对的可靠性，就用strong。如果你的代码可以容忍一些伪失败，并且希望获得更好的性能，就可以用weak。

一个通用的模式是用一个循环来重试compare_exchange_weak，直到成功为止：

#include <iostream>
#include <atomic>

int main() {
  std::atomic<int> value{0};
  int expected = 0;
  int desired = 1;

  while (!value.compare_exchange_weak(expected, desired)) {
    // compare_exchange_weak 可能会伪失败，所以我们需要更新 expected 的值
    // 重要：如果compare_exchange_weak失败，expected的值会被更新为atomic变量的当前值
    std::cout << "Weak exchange failed, retrying. Expected: " << expected << ", Current value: " << value.load() << std::endl;
  }

  std::cout << "Value successfully changed to: " << value.load() << std::endl;

  return 0;
}

三、compare_exchange的硬件支持：指令集大揭秘

好了，故事讲完了，现在我们来聊点硬核的。compare_exchange的底层实现，离不开CPU提供的原子指令。不同的CPU架构，提供的原子指令也可能不同。

下面我们以x86架构为例，看看compare_exchange背后到底发生了什么。

在x86架构下，compare_exchange通常使用cmpxchg指令来实现。cmpxchg指令会将一个寄存器的值与内存中的一个值进行比较。如果相等，就把另一个寄存器的值写入内存；如果不相等，就把内存中的值加载到第一个寄存器中。整个过程是原子性的。

指令	功能描述
`cmpxchg8b`	比较并交换8字节（64位），用于原子地更新双字
`cmpxchg16b`	比较并交换16字节（128位），用于原子地更新四字 (需要CPU支持，例如某些较新的Intel和AMD处理器)
`lock` 前缀	用于保证指令的原子性，防止其他CPU核心或线程同时访问同一块内存

举个例子：

假设我们要用compare_exchange来更新一个std::atomic<int>变量的值。

加载期望的旧值到寄存器EAX。
加载内存地址的值到寄存器EDX。
执行cmpxchg指令：
- 如果EAX == EDX，就把期望的新值加载到内存地址。
- 如果EAX != EDX，就把内存地址的值加载到EAX。
根据比较结果，设置CPU的零标志位（ZF）。
C++的compare_exchange函数会根据ZF的值来判断操作是否成功。

简化的汇编代码（x86）：

; 假设 atomic_variable 的地址在 RDI， expected 的地址在 RSI， desired 的值在 RDX

mov eax, [rsi]      ; 将 expected 的值加载到 EAX 寄存器
lock cmpxchg dword ptr [rdi], edx  ; 原子比较并交换 RDI 指向的内存地址的值
                                    ; 如果 [rdi] == eax, 则 [rdi] = edx, 否则 eax = [rdi]
je success            ; 如果 ZF=1 (EAX == [rdi]), 跳转到 success
mov [rsi], eax      ; 如果 ZF=0 (EAX != [rdi]), 将 EAX 的值写回 expected
success:
; ...

关于weak的实现：

compare_exchange_weak的底层实现，可能会利用CPU的一些优化机制，比如允许出现虚假失败。这可以减少CPU的缓存一致性开销，从而提高性能。

四、代码示例：compare_exchange 的各种用法

光说不练假把式，现在我们来看一些compare_exchange的实际应用。

1. 原子计数器：

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

std::atomic<int> counter{0};

void increment_counter() {
  for (int i = 0; i < 10000; ++i) {
    int expected = counter.load();
    int desired = expected + 1;
    while (!counter.compare_exchange_weak(expected, desired)) {
      // 如果失败，更新 expected 的值并重试
    }
  }
}

int main() {
  std::vector<std::thread> threads;
  for (int i = 0; i < 4; ++i) {
    threads.emplace_back(increment_counter);
  }

  for (auto& thread : threads) {
    thread.join();
  }

  std::cout << "Counter value: " << counter.load() << std::endl;
  return 0;
}

2. 自旋锁：

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

class SpinLock {
 public:
  SpinLock() : locked(false) {}

  void lock() {
    bool expected = false;
    while (!locked.compare_exchange_weak(expected, true)) {
      expected = false; // 关键：每次循环都要重置 expected 为 false
                         // 避免ABA问题， 虽然ABA问题通常更多出现在指针上，但在布尔类型上也需要注意。
                         // 考虑以下情况：线程A尝试加锁，读取locked为false，准备cas操作
                         // 此时线程B加锁成功，然后解锁，locked再次变为false
                         // 线程A继续执行cas操作，虽然locked的值又变回了false，但可能线程A已经不应该获得锁了。
      // 可以添加一些 backoff 策略，减少 CPU 占用
      std::this_thread::yield(); // 放弃时间片，让其他线程有机会执行
    }
  }

  void unlock() {
    locked.store(false); // 直接释放锁，不需要 compare_exchange
  }

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

SpinLock lock;

void critical_section(int thread_id) {
  lock.lock();
  std::cout << "Thread " << thread_id << " is in the critical section." << std::endl;
  std::this_thread::sleep_for(std::chrono::milliseconds(100));
  std::cout << "Thread " << thread_id << " is leaving the critical section." << std::endl;
  lock.unlock();
}

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

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

  return 0;
}

3. 原子标志：

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

std::atomic<bool> ready{false};

void worker_thread() {
  std::cout << "Worker thread is waiting for the signal..." << std::endl;
  while (!ready.load()) {
    std::this_thread::yield();
  }
  std::cout << "Worker thread received the signal and is now working." << std::endl;
}

int main() {
  std::thread worker(worker_thread);

  std::this_thread::sleep_for(std::chrono::seconds(2));

  // 使用 compare_exchange_strong 设置 ready 标志
  bool expected = false;
  bool desired = true;
  if (ready.compare_exchange_strong(expected, desired)) {
    std::cout << "Main thread sent the signal." << std::endl;
  } else {
    std::cout << "Failed to send the signal." << std::endl;
  }

  worker.join();
  return 0;
}

五、compare_exchange的注意事项和坑

ABA问题： compare_exchange可能会受到ABA问题的影响。也就是说，一个值从A变成了B，又变回了A，compare_exchange可能会误认为这个值没有发生变化。解决ABA问题的方法有很多，比如使用版本号、指针等。
性能问题： 频繁的compare_exchange操作可能会导致CPU的缓存一致性开销，从而影响性能。因此，要尽量避免不必要的compare_exchange操作。
内存模型： C++的内存模型非常复杂，compare_exchange操作的内存顺序（memory order）也会影响程序的行为。要根据实际情况选择合适的内存顺序。

六、总结

compare_exchange是一个非常强大的原子操作，它可以让你在并发编程的世界里，以一种非常优雅的方式，保证数据的安全性和一致性。但是，compare_exchange的使用也需要非常小心，要避免各种坑。

希望今天的讲解对大家有所帮助。记住，并发编程的世界充满了挑战，但只要我们掌握了正确的工具和技巧，就能战胜一切困难！

发表回复 取消回复

发表回复取消回复