C++内存模型与并发支持

本文是CppCon23演讲：C++ Memory Model：from C++11 to C++ 23的笔记，掺杂个人见解以及扩展

内存模型

操作系统的四个特性：虚拟，并发，持久
抽象中很重要的一部分就是内存虚拟。从编程的角度来看，编程就是将指令和数据从内存中拿出来处理一下然后再放回去。在最早期事情就这么简单，但是随着计算机不断发展，内存相关的机制也越来越复杂，内存机制主要由两部分组成：数据在内存中的布局，并发性的支持。

随着人们对编程的不断抽象，编程方式增加了诸多高级语言和高级特性，如C++中的全局变量，局部变量，类，虚函数表等等的出现，使得C++数据的布局变得复杂。
因为对运行速度的要求，增加了流水线技术，cache技术，分支预测技术，多发射，乱序执行技术，以及多线程，多进程出现后的同步/异步需求，导致操作系统以及编程语言内存模型对并发性的支持越来越复杂。

其中后者被称为内存模型，内存模型以load和store的方式定义了内存行为，并定义了内存访问行为的正确性。
内存模型主要的作用是：

指导编译器/硬件的实现，能够以什么样的规则来实现内存操作的优化。
指导程序员编程，写出的代码，有哪些行为是隐式保证的，哪些是需要显式保证的。

本文主要讨论对并发部分的支持,这部分在C++11以后才有较好的支持，C++11新增了多线程支持的内存模型，atomic操作的相关支持以及例如memory order等相关基础支持。

就目前的C++而言，大多数情况下，程序的运行并不是完全按照所写代码执行的，原因在于CPU以及编译器都会进行很多优化。
编译器会进行循环展开，函数内联，代码移动，指令调度等一系列优化。根据开启的优化等级不同，进行的优化也不同，优化等级越高，编译器优化越激进。

如果使用debug模式，那编译的结果是严格按照代码进行运行的，方便进行调试

编译器和CPU优化都遵循As-if-rule，但编译器如何优化不是重点，那些事情由编译器来做，C++给程序员提供了成套的工具，以支持程序员的代码不被编译器的优化影响。
内存问题在单线程时一般不会有太多问题，无非就是内存泄漏，内存泄漏有相应的工具检测。在多线程时问题较为复杂，主要体现在：多线程同步问题，线程安全问题。

SC内存模型

最早提出的内存模型即SC内存模型于1979年提出，Sequential Consistency，顺序一致性。即：内存的读写顺序，严格按照代码的顺序来。所有线程/核心是公用一个内存的，所以对于内存的读写，应该有一个全局顺序，SC要求这个顺序和单个线程的代码顺序是完全一致的，但是对于不同线程间的顺序，没有限制。即，在一个线程中先a再b，那么这两行代码对应的读写行为在全局的内存读写中，一定是先a再b，至于a和b之间是否有其他线程的读写行为，SC没有保证。

SC模型的要求是比较严格，会限制编译器和CPU的优化。例如在编程中先读a，读b，写a，写b，在没有数据依赖的情况下，完全可以优化为读a写a，读b写b。但是SC内存模型里，就不允许这样优化。

SC的规则，图片来自知乎博主Li Shi
RMW指read-modify-write操作
SC是最基本的内存模型，通过放宽SC可以得到其他的更为宽松的内存模型。例如：放宽S-L，可以得到TSO；放宽S-S可以得到PSO；放宽L-S和L-L可以得到很多模型例如weak模型，released模型等。

TSO内存模型

Total Store Order内存模型，X86中采用该内存模型。因为现在的处理器内核都是采用write buffer来保存提交的store指令，但是如果存在下面这种情况
在这里插入图片描述
C1将S1存入write buffer，C2将S2存入write buffer
C1执行L2，C2执行L2，但是这时write还没执行，r1和r2都得到0
接下来两个核心的write buffer将NEW值写入内存。

这种顺序下，结局是r1和r0都是0，不符合SC模型。扔掉write buffer性能就下去了，所以提出了新的TSO内存模型。

TSO只要求三种顺序：
若L(a) <p L(b)，则L(a) <m L(b)
若L(a) <p S(b)，则L(a) <m S(b)
若S(a) <p S(b)，则S(a) <m S(b)
使用FIFO write buffer,相较于SC模型，放宽了store-load的约束，TSO比SC更宽松。

B指bypass，即允许load时直接从write buffer中直接读取。
TSO中引入了FENCE内存屏障行为，在内核中执行FENCE操作，则FENCE前的内存操作一定不会被重排到FENCE后面的内存操作之后。

SC-DRF内存模型

(Sequential consistency for data race free)内存模型是编程语言层面定义的内存模型，C++和Java都使用SC-DRF内存模型，内存模型就如同程序员与编译器/CPU之间的契约，需要彼此遵守承诺。C++的内存模型默认为SC-DRF，此外还支持更宽松的非SC-DRF的模型。
程序员既想要简单的SC推理，又想要宽松模型的高性能，所以定义了SC-DRF模型，即在SC-DRF中，程序员需要通过编写正确的同步指令来确保程序在SC下是DRF的，DRF(Data race free，无数据竞争)指当两个线程访问同一个内存位置，且至少有一个访问是store，且中间没有同步操作。同步指令的背后，就是内核内存模型中的操作，例如TSO中的FENCE。

CV(const & volatile) type qualifiers[CV限定符]

这两个限定符表明了对象的特性，是“不变”还是“易变”

联想到，OpenGL中的数据分为一致变量，易变变量等，根据变量类型不同，会有不同的存储方式

const表明该对象“不变”，

但是其实并不是真的不能变
如果想要直接修改，会导致编译错误，编译器这关都过不去
如果想要间接性修改，通过指向const的指针等方法，会导致未定义行为。
例如下面这段代码：

#include <iostream>using namespace std;int main()
{const int i = 10;int *p = (int*)&i;*p = 20;cout << i << endl; //10cout << *p << endl; //20cout << &i << endl; cout << p << endl; //输出的两个地址是一样的，同一地址，两个值却不一样，为什么会这样？想一想return 0;
}

在函数中，const表示该函数对输入参数/对象本身不发生改变，C++提供了mutable关键字，如果一个类的成员变量是mutable的，那么它可以被const函数改变。mutable一般常用在类中有与逻辑无关的辅助成员变量时，可以将其设定为mutable，以用于在不影响const的情况下给与一定的自由度。
例如：

#include <iostream>
#include <mutex>
using namespace std;class ThreadSafeCounter{
private:mutable mutex m;int i;
public:ThreadSafeCounter(): i(0) {}int get() const {lock_guard<mutex> lock(m);return i;}void increment() {lock_guard<mutex> lock(m);i++;}
};void CounterInfo(const ThreadSafeCounter& counter) {cout << "Counter value: " << counter.get() << endl;
}int main()
{ThreadSafeCounter counter;cout << counter.get() << endl;counter.increment();CounterInfo(counter);return 0;
}

volatile表明该对象有可能发生意外改变。使用volatile可以确保该类型不会被优化移除掉，每次访问时都是对内存里对象本身进行访问而不是使用之前的缓存。除此之外，别无他用，不能够替代多线程的同步机制。

现代系统都有专门的读写策略，写策略包括写回策略和写直达策略：写回策略不会在程序修改完数据后立刻将数据放到主存中，而是线放到cache中并标记为脏，只有当再次需要该cache的时候or其他合适的时间，才会判断这个cache中的数据是否为脏，是则写回，否则直接覆盖。写直达则是同时更新缓存和主存，保持主存中的数据永远是最新的。

volatile主要应用场景：

嵌入式与硬件打交道时，在程序运行时可能会隐式修改其他寄存器的值，使用volatile可以确保当用到这些寄存器时都是从实际位置访问的，而不是缓存的。

CPU具有缓存一致性，能够确保多个核心看到的同一个数据是一样的。

多线程共享字段的标志位且常被修改。

volatile没有原子性，不能够用来替代进程间的同步机制。所以在C++20以后，很多volatile关键字的使用场景已经被deprecate。

这些限定符是可以进行转换的，由多限定转换到少限定【也就是去除限定】需要用到const_cast，添加限定则一般不需要显式调用，除非没有隐式添加的方法。

#include <iostream>
#include <thread>
using namespace std;class Type
{
public:int i;Type(): i(0) {}void set(int value) const {const_cast<int&>(this->i) = value;const_cast<Type*>(this)->i = value; // 两种方法都可以修改const的i}
};int main()
{volatile const int type = 1; //可以尝试不加volatile看会是什么效果volatile const int& type_ref = type;const_cast<int&>(type_ref) = 2;Type t;t.set(type);cout << t.i << endl;return 0;
}

线程间同步

C++11以前，使用操作系统提供的互斥机制来实现线程间同步，C++11以后·，提供了锁，原子操作等诸多相关支持。

锁Mutex

C++11中四种常见的锁，三种操作：lock，unlock，try_lock。lock是阻塞式的，try_lock不阻塞。：

mutex：基本互斥锁，
- 如果锁已经被当前线程lock，那么不论是try+lock还是lock两者都会产生死锁。
recursive_mutex：递归互斥锁，允许一个线程多次上锁，释放时需要多次释放
timed_mutex：限时等待互斥锁，可以设置超时时间，如果超时就返回false并接着执行，不等待了。
recursive_timed_mutex：限时递归等待互斥锁，可递归加锁，可限时等待，时上面两者特性的结合体。

C++17还新增了一种shared_mutex，是读写锁，即存在三种状态：读模式锁，写模式锁，不加锁；读模式锁时是共享的，其他想要读的也可以对其进行加锁，但是想要写的不行。这种锁又被称为共享-独占锁
相较于互斥锁，多了lock_shared ，try_lock_shared,unlock_sharedAPI，这些就是读模式的加锁解锁

为了避免程序员忘了unlock，还增加了一个lock_guard类型，可以自动在构造时加锁，析构时解锁。为了避免lock_guard的生命周期太长导致加锁时间太长，又增加了一个unique_lock类型，构造时加锁，可以手动unlock，也可以最后析构自行unlock。

感觉太离谱了，为了避免unlock，还新增个类型，为了多给一点自由度，又新增了一个类型，为什么不直接去掉lock_gurad直接使用unique_lock呢？答案是unique_lock中维护了一个锁的状态，性能不如lock_guard。好！不愧是C++
不过设计lock_guard也有RAII的原因，如果有一个线程在lock后跑崩了，lock_guard析构时会释放锁。

案例代码：

#include <iostream>
#include <mutex>
#include <thread>
#include <chrono>using namespace std;int main()
{int data = 0;int data2 = 0;int data3 = 0;mutex m;recursive_mutex rm;timed_mutex tm;thread t1,t2;t1 = thread([&data, &m] {for (int i = 0; i < 100; ++i) {lock_guard<mutex> lock(m);data += 1;}});t2 = thread([&data, &m]() {for(int i = 0; i < 100; ++i) {unique_lock<mutex> lock(m);data += 1;lock.unlock();this_thread::sleep_for(chrono::milliseconds(10));}});thread t3,t4;t3 = thread([&data2, &rm] {for (int i = 0; i < 100; ++i) {lock_guard<recursive_mutex> lock(rm);data2 += 1;}});t4 = thread([&data2, &rm] {for(int i = 0; i < 100; ++i) {if(rm.try_lock()){data2 += 1;rm.unlock();}else{cout << "Failed to lock" << endl;}}});thread t5;t5 = thread([&data3, &tm]{for(int i = 0; i < 100; ++i) {if(tm.try_lock_for(chrono::milliseconds(10))) {data3 += 1;tm.unlock();}}});t1.join();t2.join();t3.join();t4.join();t5.join();cout << "Data1: " << data << "\n" << endl;cout << "Data2: " << data2 << "\n" << endl;cout << "Data3: " << data3 << "\n" << endl;
}

Atomic操作

原子操作是用于解决数据竞争的一种方式。所谓原子操作，就是将对某块内存的”read，modify，write“三个操作不间断完成，以确保这期间不会有其他线程对该内存的读写导致数据竞争问题。C++11中引入了原子类型和原子操作。原子操作还可以配置同一线程内自己附近的非原子操作顺序，这部分在memory order中会谈到。
C++中有两种原子类型：

atomic：普通的原子类型，所有操作跟原始类型一样，要求原始类型必须是trivially copyable type
atomic_flag：无锁的原子bool类型。

C++中的atomic除了atomic_flag以外的其他原子类型可用互斥或其他锁定操作实现，而不一定用免锁的原子 CPU 指令，所以也提供了一个is_lock_free()API用于判断是否是无锁实现。
示例代码如下：

#include <iostream>
#include <thread>using namespace std;class Person{
public:Person(int age, int height): age(age), height(height){}// Person(const Person& p) = delete; //如果禁用拷贝构造函数，那么这个类就不是copy constructible和move constructible的// Person operator=(const Person& p) = delete; //如果禁用拷贝赋值函数，那么这个类就不是copy assignable和move assignable的
private:int age;int height;// char* temp;  // 如果加上这一行，Name就不是trivially copyable了,因为指针不是trivially copyable
};class Class
{
public:Class() = default;
private:int a[20]; //默认atomic不是lock-free的,因为a[20]太大了,如果改成a[2]就是lock-free的};int main()
{cout << is_trivially_copyable<Person>::value << endl;cout << is_move_constructible<Person>::value << endl;cout << is_copy_constructible<Person>::value << endl;cout << is_move_assignable<Person>::value << endl;cout << is_copy_assignable<Person>::value << endl;// 想要使用atomic，必须保证以上5个都是true// atomic只能用于trivially copyable类型，且atomic本身既不能复制，也不能移动atomic<int> a(10);cout << a.load() << endl; //load用于返回原子变量的值a.store(20);cout << a.load() << endl; //store用于设置原子变量的值cout << a.is_lock_free() << endl; //判断原子变量是否是lock-free的，如果是，返回true，否则返回falseatomic<Person> p(Person(10, 20));cout << p.is_lock_free() << endl;atomic<Class> c;cout << c.is_lock_free() << endl;return 0;
}

C++ atomic中有一个特殊情况需要单独领出来讲，那就是share_ptr智能指针。C++20提供了atomic<share_ptr>。share_ptr的计数器是线程安全的，但是使用share_ptr访问数据则不是线程安全的。atomic<share_ptr>是完全线程安全的。
示例：

#include <iostream>
#include <memory>
#include <atomic>using namespace std;class Test
{
public:Test() : a(0) {}int a;
};// 编译的时候记得指定C++20标准，否则编译不通过
int main()
{atomic<shared_ptr<Test>> data;atomic<Test> data2;data.is_lock_free() ? cout << "data is lock free" << endl : cout << "data is not lock free" << endl;data2.is_lock_free() ? cout << "data2 is lock free" << endl : cout << "data2 is not lock free" << endl;
}

memory order

对于多线程时，可以通过原子操作来避免数据竞争的情况。但是还有问题，考虑一种情况：线程A和B都要进行x++，且x是原子操作，如果A先到了这里，B在这里的时候会停下来等待，如果线程A中在x++附近同时还有y++操作，请问B能否看到A中的y++操作？答案是不确定，因为原子操作只能保证该操作是原子的不会被打断，没有同步效果，为了利用原子操作，让原子操作有同步效果，定义了memory order。主要是利用happens-before原则和Synchronized-with关系来是实现同步。

即：线程内的运行顺序会影响到多线程的结果，原子操作无法影响重排，所以提出memory order是为了限制同一线程内的运行顺序，进而避免多线程出现问题。
happen-before原则指：如果A指令在B指令之前执行，那么A指令的结果对B指令是可见的。在单线程时这是理所当然的，但是在多线程时这个原则很重要。
synchronizes-with relationship指：如果A在load原子对象，B在store同一个原子对象，那么A和B存在synchronizes-with relationship。

考虑mutex的情景，对于一个线程A代码如下：

y++;
mutex.lock();
x++;
mutex.unlock();

在线程A进入临界区时，可以保证同一时刻，只有A线程在运行，为了保证不出现数据竞争，即使重排，临界区内的代码也不会排在临界区以外，但是对于临界区以外的指令，倒是可以重排到临界区以内。所以，我们可以把lock和unlock看作一个屏障，lock以下的代码不能往上重排，unlock以上的代码不能往下重排。
C++内存模型借鉴这个，引入了两个等效概念：Acquire（类似lock）和Release（类似unlock），这两个都是单方向的屏障（One-way Barriers: acquire barrier, release barrier）。
在这里插入图片描述

6种memory order如下：

enum memory_order {memory_order_relaxed, //不管内存顺序memory_order_consume, // 类似于memory_order_acquire，也是用于load操作，但更为宽松// load操作所在的线程仅能看到对于依赖于该原子变量的写操作的结果。memory_order_acquire, // 对于使用该枚举值的load操作，不允许该load之后的操作重排到load之前。// 若同一原子变量的store操作(使用memory_order_release，在另一线程) Synchronizes-with 该load操作，// 则另一线程store操作之前的所有写操作的结果，对当前线程load操作之后的所有操作可见。memory_order_release, // 使用该枚举值的store操作，不允许store之前的操作重排到store之后。// Store(release)操作可以与load(Acquire/Consume)配合使用。memory_order_acq_rel, //用于RMW原子操作，RMW操作前后的语句都不允许跨越该操作而重排。memory_order_seq_cst // 顺序一致性模型，前后语句都不能跨越该操作进行重排，相当于一个双向屏障
};

所以虽然提供了六种memory order，但是一共是三种内存次序排列方式：

sequential consistent：顺序一致次序，要求最严，完全屏障，不允许前后重排
relaxed：宽松内存模型，压根儿不管内存排序，不推荐使用，主要是为了在一些保证顺序一致代价比较大的平台上使用，例如ArmV7
Acquire-Release次序：原子 load 操作是 acquire 操作, 原子 store 操作是release操作，原子read_modify_write操作(如fetch_add(),exchange())可以是 acquire, release 或两者皆是(memory_order_acq_rel)。同步是成对出现的，它出现在一个进行 release 操作和一个进行 acquire 操作的线程间。一个 release 操作 synchronizes-with 一个想要读取刚才被写的值的 acquire 操作。
- acquire-consume是acquire-release的一个特殊情况，引入了数据依赖关系，只限制有数据依赖关系的指令的重排，相较于acquire-release效率更高，如果明确只需要限制某个变量时可以用consume

如果原子操作不指定memory order，默认是memory_order_seq_cst

操作	有效的 Memory order 枚举值	备注
Load	memory_order_relaxed, memory_order_consume, memory_order_acquire, or memory_order_seq_cst	其它枚举值不合法， MS STL 的实现是将其当作 memory_order_seq_cst 处理
Store	memory_order_relaxed, memory_order_release, or memory_order_seq_cst	同上
read-modify-write	memory_order_relaxed, memory_order_consume, memory_order_acquire, memory_order_release, memory_order_acq_rel, or memory_order_seq_cst

案例代码：

#include <iostream>
#include <thread>
#include <atomic>using namespace std;int main()
{atomic<int> data(0);atomic<bool> ready(false);thread producer([&data, &ready](){data.store(10, memory_order_relaxed); // memory_order_relaxed表示只保证原子操作是原子的，不保证内存访问顺序，因此不会有任何同步开销ready.store(true, memory_order_release); // memory_order_release表示所有在该指令之前的读写操作，不会被重排序到该指令之后// 且该指令之前的所有写操作对所有其他访问同一个原子变量的线程可见});thread consumer([&data, &ready](){while(!ready.load(memory_order_acquire)); // memory_order_acquire表示所有在该指令之后的读写操作，不会被重排序到该指令之前// 且其他使用release语义访问同一个原子操作的线程之前的所有读写操作对该线程可见// acquire和release共同使用，保证了对同一个原子变量的读写操作的顺序一致性cout << data.load(memory_order_relaxed) << endl;});producer.join();consumer.join();return 0;
}

future & promise

C++11中增加的一种异步编程方式，promise表示该对象后续一定会被设置，在使用到该值的地方可以用p.get_future()来得到一个future，如果该值还没被设置，就会等待，直到被设置。在设置该值的地方，可以使用p.set_value来设置该值，使用p.set_exception来设置异常状态。
注意：

只能从promise共享状态获取一个future对象，不能把两个future关联到同一个promise
如果promise不设置值或者异常，promise 对象在析构时会自动地设置一个 future_error 异常(broken_promise)来设置其自身的就绪状态
promise 对象的set_value只能被调用一次，多次调用会抛出std::future_error异常（因为第一次调用后状态变更为ready）
std::future是通过std::promise::get_future获取到的，自己构造出来的无效
示例代码：

#include <iostream>
#include <thread>
#include <future>
using namespace std;int main()
{promise<int> p;thread t1([&p](){this_thread::sleep_for(chrono::seconds(10));p.set_value(10);});thread t2([&p](){future<int> f = p.get_future();if(f.wait_for(chrono::seconds(5)) == future_status::ready){cout << f.get() << endl;}else{cout << "timeout" << endl;}});t1.join();t2.join();// C++11中还新增加了async API，更加简洁// async()函数返回一个future对象，可以通过get()方法获取异步任务的返回值// 通过launch::async参数，可以指定异步任务在新线程中执行// 通过launch::deferred参数，可以指定异步任务在调用get()方法时执行// 通过launch::async | launch::deferred参数，可以指定异步任务在新线程中执行，也可以指定异步任务在调用get()方法时执行future<int> f = async(launch::async, [](){this_thread::sleep_for(chrono::seconds(5));return 10;});cout << f.get() << endl;// packaged_task是将函数包装成一个任务，可以异步执行packaged_task<int()> task([](){this_thread::sleep_for(chrono::seconds(10));return 10;});// task.get_future()返回一个future对象，用于获取任务的返回值future<int> f2 = task.get_future();cout << f2.get() << endl;return 0;
}

Summary

内存模型是程序员与编程语言开发者和计算机硬件开发者之间的约定，目的是为了解决互斥和同步两个问题，从硬件到语言再到程序员，一层一层向上层保证执行的正确性，保证越强，性能成本就越高。

互斥问题可以通过锁或原子操作来实现，原子操作在C++中既可以由CPU提供的原子指令来实现，也可以通过锁来实现。
同步势必会影响性能，最快的方法肯定是所有的同步问题都由程序员自己解决，下层只保证happen-before和synchronizes-with relationship，对应的是宽松内存模型。最慢的方法肯定是所有同步都由语言和硬件保证，完全按照程序的顺序执行，对应的是顺序一致内存模型。

所以这个talk前半部分在讲内存模型，后面基本都是在讲多线程的同步问题了，因为确保顺序一致性是语言开发人员和硬件开发人员的事情，只有如何手动同步是程序员的事情。

C++实在太复杂了，哪怕只是对并发的支持，都有非常非常多的内容，如下图
在这里插入图片描述
所以，用到啥学啥吧，实在是太多了。