并发编程：乱序执行的那些事儿五分钟给你整明白

什么是乱序执行

乱序执行 [1] ，简单说就是程序里面的代码的执行顺序，有可能会被编译器、CPU 根据某种策略调整顺序（俗称，“打乱”）——虽然从单线程的角度看，乱序执行不影响执行结果。

为什么需要乱序执行

主要原因是 CPU 内部采用 流水线技术 [2] 。抽象且简化地看，一个 CPU 指令的执行过程可以分成 4 个阶段：取指、译码、执行、写回。

这 4 个阶段分别由 4 个独立物理执行单元来完成。这种情况下，如果指令之间没有依赖关系，后一条指令并不需要等到前一条指令完全执行完成再开始执行。而是前一条指令完成取指之后，后一条指令便可以开始执行取指操作。

比较理想的情况如下图所示：指令之间无依赖，可以使流水线的并行度最大化。

在 按序执行 的情况下，一旦遇到指令依赖的情况，流水线就会停滞。比如：

指令 1: Load R3 <- R1(0) # 从内存中加载数据到 R3 寄存器指令 2: Add R3 <- R3, R1 # 加法，依赖指令 1 的执行结果指令 3: Sub R1 <- R6, R7 # 减法指令 4: Add R4 <- R6, R8 # 加法

上面的伪代码中，指令 2 依赖指令 1 的执行结果。该指令 1 执行完成之前，指令 2 无法执行，这会让流水线的执行效率大大降低。

观察到，指令 3 和指令 4 对其它指令没有依赖，可以考虑将这两条指令”乱序“到指令 2 之前。

这样，流水线执行单元就可以尽可能处于工作状态。

总的来说，通过乱序执行，合理调整指令的执行顺序，可以提高流水线的运行效率，让指令的执行能够尽可能地并行起来。

Compiler Fence

在多线程的环境下，乱序执行的存在，很容易打破一些预期，造成一些意想不到的问题。

乱序执行有两种情况：

  1. 在编译期，编译器进行指令重排。
  2. 在运行期，CPU 进行指令乱序执行。

我们先来看一个编译器指令重排的例子：

#include <atomic>// 按序递增发号器std::atomic<int> timestamp_oracle{0};// 当前处理的号码int now_serving_ts{0};int shared_value;int compute();void memory_reorder() {    // 原子地获取一个号码    int ts = timestamp_oracle.fetch_add(1);    // 加锁：判断当前是否轮到这个号码，否则就循环等    while (now_serving_ts != ts);    // 临界区：开始处理请求    shared_value = compute();        // 编译器 memory barrier    asm volatile('' : : : 'memory');    // 解锁：下一个要处理的号码    now_serving_ts = ts + 1;}

简单解释一下这段代码：

  1. 这个程序通过维护一个“发号器 timestamp_oracle”，来实现按顺序处理每个线程的请求。
  2. 每个线程先从“发号器”取一个号，然后不停判断当前是否轮到自己执行，类似自旋锁的逻辑。
  3. 每个线程执行完，将“号码”切换到下一个。

在 O1 的编译优化选项下，编译出来的汇编指令没有被重排（通过左右两边的代码行背景色就可以看出来）。

在 O2 的编译优化选项下，出现了指令被重排了，并且这里的指令重排打破了程序的预期，先切换了 now_serving_ts，再更新 shared_value，导致 shared_value 可能被并发修改。

为了阻止编译器重排这两句代码的指令，需要在它们之间插入一个 compiler fence。

asm volatile('': : :'memory');

这个是 GCC 扩展的 compiler fence 的写法。这条指令告诉编译器（ GCC 官方文档 [3] ）：

  1. 防止这条 fence 指令上方的内存访问操作被移到下方，同时防止下方的内存访问操作移到上面，也就是防止了乱序。
  2. 让编译器把所有缓存在寄存器中的内存变量 flush 到内存中，然后重新从内存中读取这些值。

对于第 2 点，有时候我们只需要刷新部分变量。刷新所有寄存器并不一定完全符合我们的预期，并且会引入不必要的开销。GCC 支持指定变量的 compiler fence。

write(x)asm volatile('': '=m'(y) : 'm'(x):)read(y)

中间的内联汇编指令告诉编译器不要把 write(x) 和 read(y) 这两个操作乱序。

CPU Fence

先来看一个例子：

int x = 0;int y = 0;int r0, r1;// CPU1void f1(){ x = 1; asm volatile('': : :'memory'); r0 = y; }// CPU2void f2(){ y = 1; asm volatile('': : :'memory'); r1 = x;}

上面的例子中，由于 compiler fence 的存在，编译器不会对函数 f1 和函数 f2 内部的指令进行重排。

此时，如果 CPU 执行时也没有乱序，是不可能出现 r0 == 0 && r1 == 0 的情况的。不幸的是，由于 CPU 乱序执行的存在，这种情况是可能发生的。看下面这个例子：

#include <iostream>#include <thread>int x = 0;int y = 0;int r0 = 100;int r1 = 100;void f1() {    x = 1;    asm volatile('': : :'memory');    r0 = y;}void f2() {    y = 1;    asm volatile('': : :'memory');    r1 = x;}void init() {    x = 0;    y = 0;    r0 = 100;    r1 = 100;}bool check() {    return r0 == 0 && r1 == 0;}std::atomic<bool> wait1{true};std::atomic<bool> wait2{true};std::atomic<bool> stop{false};void loop1() {    while(!stop.load(std::memory_order_relaxed)) {        while (wait1.load(std::memory_order_relaxed));        asm volatile('' ::: 'memory');        f1();        asm volatile('' ::: 'memory');        wait1.store(true, std::memory_order_relaxed);    }}void loop2() {    while (!stop.load(std::memory_order_relaxed)) {        while (wait2.load(std::memory_order_relaxed));        asm volatile('' ::: 'memory');        f2();        asm volatile('' ::: 'memory');        wait2.store(true, std::memory_order_relaxed);    }}int main() {    std::thread thread1(loop1);    std::thread thread2(loop2);    long count = 0;    while(true) {        count++;        init();        asm volatile('' ::: 'memory');        wait1.store(false, std::memory_order_relaxed);        wait2.store(false, std::memory_order_relaxed);        while (!wait1.load(std::memory_order_relaxed));        while (!wait2.load(std::memory_order_relaxed));        asm volatile('' ::: 'memory');        if (check()) {            std::cout << 'test count ' << count << ': r0 == ' << r0 << ' && r1 == ' << r1 << std::endl;            break;        } else {            if (count % 10000 == 0) {                std::cout << 'test count ' << count << ': OK' << std::endl;            }        }    }    stop.store(true);    wait1.store(false);    wait2.store(false);    thread1.join();    thread2.join();    return 0;}

上面的程序可以很轻易就运行出 r0 == 0 && r1 == 0 的结果，比如：

test count 56: r0 == 0 && r1 == 0

为了防止 CPU 乱序执行，需要使用 CPU fence。我们可以将函数 f1 和 f2 中的 compiler fence 修改为 CPU fence：

void f1() {    x = 1;    asm volatile('mfence': : :'memory');    r0 = y;}void f2() {    y = 1;    asm volatile('mfence': : :'memory');    r1 = x;}

如此，便不会出现 r0 == 0 && r1 == 0 的情况了。

总结

指令乱序执行主要由两种因素导致：

  1. 编译期指令重排。
  2. 运行期 CPU 乱序执行。

无论是编译期的指令重排还是 CPU 的乱序执行，主要都是为了让 CPU 内部的指令流水线可以“充满”，提高指令执行的并行度。

上面举的插入 fence 的例子都是使用了 GCC 的扩展语法，实际上 C++ 标准库已经提供了类似的封装： std::atomic_thread_fence [4] ，跨平台且可读性更好。

一些无锁编程、追求极致性能的场景可能会需要手动在合适的地方插入合适 fence，这里涉及的细节太多，非常容易出错。原子变量操作根据不同的 memory order 会自动插入合适的 fence，建议优先考虑使用原子变量。

原文链接：
https://mp.weixin.qq.com/s?__biz=MzI0NjA1MTU5Ng==&mid=2247484193&idx=1&sn=
88968ef741f3d336276e23e577e21bf8&utm_source=tuicool&utm_medium=referral

赞 (0)

相关推荐