Linux C/C++并发编程实战C/C++ 6种内存时序memory_order详解
Posted 奇妙之二进制
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Linux C/C++并发编程实战C/C++ 6种内存时序memory_order详解相关的知识，希望对你有一定的参考价值。
文章目录
接口预览
#include <stdatomic.h>
_Bool atomic_compare_exchange_strong( volatile A* obj,
C* expected, C desired ); (1) (since C11)
_Bool atomic_compare_exchange_weak( volatile A *obj,
C* expected, C desired ); (2) (since C11)
_Bool atomic_compare_exchange_strong_explicit( volatile A* obj,
C* expected, C desired,
memory_order succ,
memory_order fail );(3) (since C11)
_Bool atomic_compare_exchange_weak_explicit( volatile A *obj,
C* expected, C desired,
memory_order succ,
memory_order fail ); (4) (since C11)
上面几个函数的作用：原子的比较obj所指对象和expected所指对象的内容，如果它们相等，则将desired的值赋值给obj。否则，将obj所指向的内容load到expected所指向的空间。
上述过程用代码表示如下：
if (memcmp(obj, expected, sizeof *obj) == 0)
memcpy(obj, &desired, sizeof *obj);
return true;
else
memcpy(expected, obj, sizeof *obj);
return false;
obj指向我们想修改的对象，
desired是我们想赋值给obj的值，
但是在修改之前，我们要先测试下obj是否是我们期待的值*expected，这是以防我们修改obj之前，obj的值已经被其他人偷偷改了（多线程场景）。
如果改之前是我们期待的，说明还没有人对他进行修改，那就可以成功的修改*obj为desired，并返回true；
否则，将*obj赋值给*expected，并返回false。
这个操作被记为CAS。
strong和weak的区别
如果仅仅是这样子，这些接口还是不安全的，因为只保证了原子性，没有保证可见性，所以引入了一个内存时序/模型的概念（第四个参数）。
关于重排/乱序
说到内存模型，首先需要明确一个普遍存在，但却未必人人都注意到的事实：程序通常并不是总按着照源码中的顺序一一执行，此谓之乱序，乱序产生的原因可能有好几种：
- 编译器出于优化的目的，在编译阶段将源码的顺序进行交换。
- 程序执行期间，指令流水被 cpu 乱序执行。
- inherent cache 的分层及刷新策略使得有时候某些写读操作的从效果上看，顺序被重排。
以上乱序现象虽然来源不同，但从源码的角度，对上层应用程序来说，他们的效果其实相同：写出来的代码与最后被执行的代码是不一致的。这个事实可能会 让人很惊讶：有这样严重的问题，还怎么写得出正确的代码？这担忧是多余的了，乱序的现象虽然普遍存在，但它们都有很重要的一个共同点：在单线程执行的情况 下，乱序执行与不乱序执行，最后都会得出相同的结果 (both end up with the same observable result), 这是乱序被允许出现所需要遵循的首要原则，也是为什么乱序虽然一直存在但却多数程序员大部分时间都感觉不到的根本原因。
乱序的出现说到底是编译器，CPU 等为了让你程序跑得更快而作出无限努力的结果，程序员们应该为它们的良苦用心抹一把泪。
从乱序的种类来看，乱序主要可以分为如下4种：
-
写写乱序(store store), 前面的写操作被放到了后面的操作之后，比如：
a = 3; b = 4; 被乱序为： b = 4; a = 3; -
写读乱序(store load)，前面的写操作被放到了后面的读操作之后，比如：
a = 3; load(b); 被乱序为： load(b); a = 3; -
读读乱序(load load)， 前面的读操作被放到了后一个读操作之后，比如：
load(a); load(b); 被乱序为： load(b); load(a); -
读写乱序(load store), 前面的读操作被放到了后一个写操作之后，比如：
load(a); b = 4; 被乱序为： b = 4; load(a);
程序的乱序在单线程的世界里多数时候并没有引起太多引人注意的问题，但在多线程的世界里，这些乱序就制造了特别的麻烦，究其原因，最主要的有2个：
- 并发导致修改和访问共享变量的存在不确定性，使得一些不期望的中间状态（往往是错误状态）被暴露了出去，因此像 mutex，各种 lock 之类的东西在写多线程时被频繁地使用。
- 变量被修改后，该修改未必能被另一个线程及时观察到，因此需要“同步”。
解决同步问题就需要确定内存模型，也就是需要确定线程间应该怎么通过共享内存来进行交互。
- 编译器的重排序
编译器在不改变单线程程序语义的前提下，可以重新安排语句的执行顺序。下面的例子来自并发编程网的文章
int A = 1+1;
int B = 2;
int C = A;
“编译期重排序的典型就是通过调整指令顺序，在不改变程序语义的前提下，尽可能减少寄存器的读取、存储次数，充分复用寄存器的存储值。假设第一条指令计算一个值赋给变量A并存放在寄存器中，第二条指令与A无关但需要占用寄存器（假设它将占用A所在的那个寄存器），第三条指令使用A的值且与第二条指令无关。那么如果按照顺序一致性模型，A在第一条指令执行过后被放入寄存器，在第二条指令执行时A不再存在，第三条指令执行时A重新被读入寄存器，而这个过程中，A的值没有发生变化。通常编译器都会交换第二和第三条指令的位置，这样第一条指令结束时A存在于寄存器中，接下来可以直接从寄存器中读取A的值，降低了重复读取的开销。”
int A = 0;
int B = 0;
void fun()
A = B + 1; // L5
B = 1; // L6
int main()
fun();
return 0;
- 指令间的直接依赖关系
编译器和处理器可能会对操作做重排序，但是要遵守数据依赖关系，编译器和处理器不会改变存在数据依赖关系的两个操作的执行顺序。如果两个操作访问同一个变量，且这两个操作中有一个为写操作，此时这两个操作之间就存在数据依赖性。数据依赖分下列三种类型：
名称 代码示例说明
写后读 a = 1;b = a;写一个变量之后，再读这个位置。
写后写 a = 1;a = 2;写一个变量之后，再写这个变量。
读后写 a = b;b = 1;读一个变量之后，再写这个变量。
上面三种情况，只要重排序两个操作的执行顺序，程序的执行结果将会被改变。像这种有直接依赖关系的操作，是不会进行重排序的。特别注意：这里说的依赖关系仅仅是在单个线程内。
- 指令间的隐式依赖关系
编译器和CPU都必须保证程序上下文的因果关系不发生改变。因此，在绝大多数情况下，我们写程序都不会去考虑乱序所带来的影响。但是有些程序逻辑，单纯从上下文是看不出它们的因果关系的。比如：
*addr = 5;
val = *data;
从表面上看，addr和data是没有什么联系的，完全可以放心的去乱序执行。但是如果这是在某某设备驱动程序中，这两个变量却可能对应到设备的地址端口和数据端口。并且，这个设备规定了，当你需要读写设备上的某个寄存器时，先将寄存器编号设置到地址端口，然后就可以通过对数据端口的读写而操作到对应的寄存器。那么这么一来，对前面那两条指令的乱序执行就可能造成错误。对于这样的逻辑，我们姑且将其称作隐式因果关系；而指令与指令之间直接的输入输出依赖，也姑且称作显式因果关系。CPU的乱序执行或者编译器的重排序是以保持显式因果关系不变为前提的，但是它们都无法识别隐式因果关系。再举个例子：
obj->data = xxx;
obj->ready = 1;
当设置了data之后，记下标志，然后在另一个线程中可能执行：
if (obj->ready)
do_something(obj->data);
虽然这个代码看上去有些别扭，但是似乎没错。不过，考虑到乱序，如果标志被置位先于data被设置，那么结果很可能就杯具了。因为从字面上看，前面的那两条指令其实并不存在显式的因果关系，乱序是有可能发生的。总的来说，如果程序具有显式的因果关系的话，乱序一定会尊重这些关系；否则，乱序就可能打破程序原有的逻辑。这时候，就需要使用屏障来抑制乱序，以维持程序所期望的逻辑。
类似的例子还有
a = 0;
//写线程
public void writer()
a = 1;
flag = true;
//读线程
public void reader()
if (flag)
int i = a * a;
虽然从读线程和写线程的逻辑关系上来说，写线程不能进行重排序，否则读线程会产生错误的结果；但是实际上写线程还是可能发生重排序的，因为从写线程自身的角度来看，a和flag的写操作完全没有依赖关系，可以随意重排序。编译器没有那么智能，它只能将依赖分析限制在单线程中，不能跨线程进行依赖分析。
这种场景下，假设writer发生了重排，那么reader得到的i结果是0。
4.CPU的重排序（指令乱序执行）
现在的CPU一般采用流水线来执行指令。一个指令的执行被分成：取指、译码、访存、执行、写回、等若干个阶段。然后，多条指令可以同时存在于流水线中，同时被执行。指令流水线并不是串行的，并不会因为一个耗时很长的指令在“执行”阶段呆很长时间，而导致后续的指令都卡在“执行”之前的阶段上。相反，流水线是并行的，多个指令可以同时处于同一个阶段，只要CPU内部相应的处理部件未被占满即可。比如说CPU有一个加法器和一个除法器，那么一条加法指令和一条除法指令就可能同时处于“执行”阶段, 而两条加法指令在“执行”阶段就只能串行工作。
然而，这样一来，乱序可能就产生了。比如一条加法指令原本出现在一条除法指令的后面，但是由于除法的执行时间很长，在它执行完之前，加法可能先执行完了。再比如两条访存指令，可能由于第二条指令命中了cache而导致它先于第一条指令完成。一般情况下，指令乱序并不是CPU在执行指令之前刻意去调整顺序。CPU总是顺序的去内存里面取指令，然后将其顺序的放入指令流水线。但是指令执行时的各种条件，指令与指令之间的相互影响，可能导致顺序放入流水线的指令，最终乱序执行完成。这就是所谓的“顺序流入，乱序流出”。
指令流水线除了在资源不足的情况下会卡住之外（如前所述的一个加法器应付两条加法指令的情况），指令之间的相关性也是导致流水线阻塞的重要原因。CPU的乱序执行并不是任意的乱序，而是以保证程序上下文因果关系为前提的。有了这个前提，CPU执行的正确性才有保证。比如：
a++;
b=f(a);
c--;
由于b=f(a)这条指令依赖于前一条指令a++的执行结果，所以b=f(a)将在“执行”阶段之前被阻塞，直到a++的执行结果被生成出来；而c–跟前面没有依赖，它可能在b=f(a)之前就能执行完。（注意，这里的f(a)并不代表一个以a为参数的函数调用，而是代表以a为操作数的指令。C语言的函数调用是需要若干条指令才能实现的，情况要更复杂些）。
像这样有依赖关系的指令如果挨得很近，后一条指令必定会因为等待前一条执行的结果，而在流水线中阻塞很久，占用流水线的资源。而编译器的乱序，作为编译优化的一种手段，则试图通过指令重排将这样的两条指令拉开距离, 以至于后一条指令进入CPU的时候，前一条指令结果已经得到了，那么也就不再需要阻塞等待了。比如将指令重排为：
a++;
c--;
b=f(a);
由于重排序和乱序执行的存在，如果在并发编程中，没有做好共享数据的同步，很容易出现各种看似诡异的问题。
限制重排的几种语义
#include <atomic>
typedef enum memory_order
memory_order_relaxed,
memory_order_consume,
memory_order_acquire, (since C++11)
memory_order_release, (until C++20)
memory_order_acq_rel,
memory_order_seq_cst
memory_order;
这6种语义描述了原子操作上下附近的内存操作顺序的重排约束。
relaxed 语义
首先是 relaxed 语义，这表示一种最宽松的内存操作约定，该约定对其他读写操作没有同步和顺序约束的限制，只保证操作本身的原子性，以这种方式修改内存时，不需要保证该修改会不会及时被其它线程看到，也不对乱序做任何要求，因此当对公共变量以 relaxed 方式进行读写时，编译器，cpu 等是被允许按照任意它们认为合适的方式来加以优化处理的。
x=y=0
// Thread 1:
r1 = y.load(std::memory_order_relaxed); // A
x.store(r1, std::memory_order_relaxed); // B
// Thread 2:
r2 = x.load(std::memory_order_relaxed); // C
y.store(42, std::memory_order_relaxed); // D
如果没法发生重排，就不可能出现r1=r2=42（D->A->B->C的结果），结果是会出现的。
releaxed语义的应用场景就是计数器自增。例如std::shared_ptr的引用计数的自增。但引用计数的自减需要使用acquire-release 。
示例：
#include <vector>
#include <iostream>
#include <thread>
#include <atomic>
std::atomic<int> cnt = 0;
void f()
for (int n = 0; n < 1000; ++n)
cnt.fetch_add(1, std::memory_order_relaxed);
int main()
std::vector<std::thread> v;
for (int n = 0; n < 10; ++n)
v.emplace_back(f);
for (auto& t : v)
t.join();
std::cout << "Final counter value is " << cnt << '\\n';
Output:
Final counter value is 10000
不使用原子变量的情况：
#include <vector>
#include <iostream>
#include <thread>
#include <atomic>
int cnt = 0;
void f()
for (int n = 0; n < 1000; ++n)
cnt++;
int main()
std::vector<std::thread> v;
for (int n = 0; n < 10; ++n)
v.emplace_back(f);
for (auto& t : v)
t.join();
std::cout << "Final counter value is " << cnt << '\\n';
Output:
Final counter value is 9046
acquire语义
保证其他的读写操作不会重排到load操作之前，其他线程的release写操作在当前线程可见。
这样设计的含义是结合经验的，我们往往读取某个数，然后进行运算或者赋值，所以不期待load之后的读写操作重排到load之前。至于load之前的读写操作我们并不关心。
release语义
当前线程的store之前的任何读写操作不会重排到store之后。所有的写操作，对其他使用acquire操作的线程可见。
data.push_back(42);
flag.store(1, std::memory_order_release);
例如上述代码的语义是，push了数据之后，标识flag为1，如果push操作被重排到store之后，那么就会出现flag为1，但data里面却没数据的情况，这是我们不期待的。
release保证了两个事情，指令重排和可见性。
release-acquire 语义
如果你曾经去看过别的介绍内存模型相关的文章，你一定会发现 release 总是和 acquire 放到一起来讲，这并不是偶然。事实上，release 和 acquire 是相辅相承的，它们必须配合起来使用，这俩是一个 “package deal”， 分开使用则完全没有意义。具体到其中， release 用于进行写操作，acquire 则用于进行读操作，它们结合起来表示这样一个约定：
如果一个线程A对一块内存 m 以 release 的方式进行修改，那么在线程 A 中，所有在该 release 操作之前进行的内存操作，都在另一个线程 B 对内存 m 以 acquire 的方式进行读取之后，变得可见。
cppref的解释：
If an atomic store in thread A is tagged memory_order_release and an atomic load in thread B from the same variable is tagged memory_order_acquire, all memory writes (non-atomic and relaxed atomic) that happened-before the atomic store from the point of view of thread A, become visible side-effects in thread B. That is, once the atomic load is completed, thread B is guaranteed to see everything thread A wrote to memory. This promise only holds if B actually returns the value that A stored, or a value from later in the release sequence.
互斥锁std::mutex和自旋锁便是release-acquire同步机制的例子。
cppref的解释：
Mutual exclusion locks, such as std::mutex or atomic spinlock, are an example of release-acquire synchronization: when the lock is released by thread A and acquired by thread B, everything that took place in the critical section (before the release) in the context of thread A has to be visible to thread B (after the acquire) which is executing the same critical section.
举个粟子，假设线程 A 执行如下指令：
a.store(3);
b.store(4);
m.store(5, release);
线程 B 执行如下:
e.load();
f.load();
m.load(acquire);
g.load();
h.load();
如上，假设线程 A 先执行，线程 B 后执行, 因为线程 A 中对 m 以 release 的方式进行修改， 而线程 B 中以 acquire 的方式对 m 进行读取，所以当线程 B 执行完 m.load(acquire) 之后， 线程 B 必须已经能看到 a == 3, b == 4. 以上死板的描述事实上还传达了额外的不那么明显的信息：
- release 和 acquire 是相对两个线程来说的，它约定的是两个线程间的相对行为：如果其中一个线程 A 以 release 的方式修改公共变量 m， 另一个线程 B 以 acquire 的方式时读取该 m 时，要有什么样的后果，但它并不保证，此时如果还有另一个线程 C 以非 acquire 的方式来读取 m 时，会有什么后果。
- 一定程度阻止了乱序的发生，因为要求 release 操作之前的所有操作都在另一个线程 acquire 之后可见，那么：
- release 操作之前的所有内存操作不允许被乱序到 release 之后。
- acquire 操作之后的所有内存操作不允许被乱序到 acquire 之前。
而在对它们的使用上，有几点是特别需要注意和强调的：
- release 和 acquire 必须配合使用，分开单独使用是没有意义。
- release 只对写操作(store) 有效，对读 (load) 是没有意义的。
- acquire 则只对读操作有效，对写操作是没有意义的。
综合例子：
#include <thread>
#include <atomic>
#include <cassert>
#include <string>
std::atomic<std::string*> ptr;
int data;
void producer()
std::string* p = new std::string("Hello");
data = 42;
ptr.store(p, std::memory_order_release);
void consumer()
std::string* p2;
while (!(p2 = ptr.load(std::memory_order_acquire)))
;
assert(*p2 == "Hello"); // never fires
assert(data == 42); // never fires
int main()
std::thread t1(producer);
std::thread t2(consumer);
t1.join(); t2.join();
#include <thread>
#include <atomic>
#include <cassert>
#include <vector>
std::vector<int> data;
std::atomic<int> flag = 0;
void thread_1()
data.push_back(42);
flag.store(1, std::memory_order_release);
void thread_2()
int expected=1;
// memory_order_relaxed is okay because this is an RMW,
// and RMWs (with any ordering) following a release form a release sequence
while (!flag.compare_exchange_strong(expected, 2, std::memory_order_relaxed))
expected = 1;
void thread_3()
while (flag.load(std::memory_order_acquire) < 2)
;
// if we read the value 2 from the atomic flag, we see 42 in the vector
assert(data.at(0) == 42); // will never fire
int main()
std::thread a(thread_1);
std::thread b(thread_2);
std::thread c(thread_3);
a.join(); b.join(); c.join();
Release-Consume语义
从上面对Acquire-Release模型的分析可以知道，虽然可以使用这个模型做到两个线程之间某些操作的synchronizes-with关系，然而这个粒度有些过于大了。
在很多时候，线程间只想针对有依赖关系的操作进行同步，除此之外线程中的其他操作顺序如何无所谓。比如下面的代码中：
b = *a;
c = *b;
其中第二行代码的执行结果依赖于第一行代码的执行结果，此时称这两行代码之间的关系为“carry-a-dependency ”。
consume语义要比acquire弱一些，和load操作有依赖的数据，
#include <thread>
#include <atomic>
#include <cassert>
#include <string>
std::atomic<std::string*> ptr;
int data;
void producer()
std::string* p = new std::string("Hello");
data = 42;
ptr.store(p, std::memory_order_release);
void consumer()
std::string* p2;
while (!(p2 = ptr.load(std::memory_order_consume)))
;
assert(*p2 == "Hello"); // never fires: *p2 carries dependency from ptr
assert(data == 42); // may or may not fire: data does not carry dependency from ptr
int main()
std::thread t1(producer);
std::thread t2(consumer);
t1.join(); t2.join();
现代的处理器通常都支持一些 read-modify-write 之类的指令，对这种指令，有时我们可能既想对该操作 执行 release 又要对该操作执行 acquire，因此 c++11 中还定义了 memory_order_acq_rel，该类型的操作就是 release 与 acquire 的结合，除前面提到的作用外，还起到了 memory barrier 的功能。
sequential consistency
sequential consistency 相当于 release + acquire 之外，还加上了一个对该操作加上全局顺序的要求，这是什么意思呢？
简单来说就是，对所有以 memory_order_seq_cst 方式进行的内存操作，不管它们是不是分散在不同的 cpu 中同时进行，这些操作所产生的效果最终都要求有一个全局的顺序，而且这个顺序在各个相关的线程看起来是一致的。
举个粟子,假设 a, b 的初始值都是0：
线程 A 执行：
a.store(3, seq_cst);
线程 B 执行：
b.store(4, seq_cst);
如上对 a 与 b 的修改虽然分别放在两个线程里同时进行，但是这多个动作毕竟是非原子的，因此这些操作地进行在全局上必须要有一个先后顺序：
- 先修改a, 后修改 b，或
- 先修改b, 把整个a。
而且这个顺序是固定的，必须在其它任意线程看起来都是一样，因此 a == 0 && b == 4 与 a == 3 && b == 0 不允许同时成立。
#include <atomic>
#include <thread>
#include <assert.h>
std::atomic<bool> x,y;
std::atomic<int> z;
void write_x_then_y()
x.store(true,std::memory_order_relaxed); //1
y.store(true,std::memory_order_release); //2 //保证 第1句的x写入在y的写入之前
void read_y_then_x()
while(!y.load(std::memory_order_acquire)); //3 //保证 y的读取在 第4句x的读取之前。所以 如果y看到修改后的值，肯定 第4句看到的也是x的新的值了。
if(x.load(std::memory_order_relaxed)) //4
++z;
int main()
x=false;
y=false;
z=0;
std::thread a(write_x_then_y);
std::thread b(read_y_then_x);
a.join();
b.join();
assert(z.load() != 0); //5 //这个肯定通过的了。
参考：
https://preshing.com/20120612/an-introduction-to-lock-free-programming/
以上是关于Linux C/C++并发编程实战C/C++ 6种内存时序memory_order详解的主要内容，如果未能解决你的问题，请参考以下文章
9月11日 | C/C++视频教程29GB/C基础+C++编程+Linux核心+Win32/MFC编程+项目