C++：并发概述（一）

这是学习 A Tour of C++ (3rd Edition) Chapter 18 Concurrency 的笔记。感觉并发这一块的知识量相当大，本章节也只是做了入门介绍，真正要掌握完整的知识体系还得啃 C++ Concurrency in Action by Anthony Williams 那本书。

区分并发与并行：

并发指的是多个任务的执行过程可以交错
并行指的是多个任务实际同时占用不同执行核心

Tasks and Threads

task 是一段可以和其它计算同时进行的工作；thread 是操作系统层面对 task 的执行载体

void f();

struct F { // 函数对象
    void operator()();   // call operator
};

void user() {
    std::thread t1 {f};
    std::thread t2 {F{}};

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

join() 表示等待线程，保证线程运行完毕之后才退出 user()。但是很容易忘记 join() ，因此可以使用 std::jthread（代表 joining thread），它会自动处理 join()，简化了代码：

void user() {
    std::jthread t1 {f};
    std::jthread t2 {F{}};
}

使用了 std::jthread 之后，join() 的操作由析构函数来负责。此后 join 的顺序与发起 jthread 的顺序相反。

线程共享同一个地址空间，因此可以方便地用共享对象来通信，但是也可能发生数据竞争。在上述的例子中，t1 和 t2 这两个线程的执行顺序是不确定的。如果我们在 f() 和 F 内部编写使用 std::cout 来输出的代码，则两个线程会同时使用 std::cout 对象，可能导致输出无法预测。当然，这个问题可以使用 std::osyncstream 来输出予以解决。

传递参数

这一块讲如何给线程函数传参数。std::thread （或者 jthread）的构造函数当中的第一个参数是线程函数的 callable，后续的参数是要传入这个 callable 的参数。callable 可以是函数指针、函数对象、lambda。如果直接传入函数名，则这个函数名会退化成函数指针。

以下特别说明要往线程函数当中传递引用类型的方法，有两种：

使用 std::ref() 把引用传入线程函数

void f(std::vector<double>& v);
std::jthread t1 {f, std::ref(some_vec)};

这里 f 接收的参数是引用。在 std::jthread 的构造函数中必须使用 std::ref(some_vec) 来保持 some_vec 的引用类型。它会告诉 std::jthread 的构造函数在调用 f 的时候传原来的 some_vec 的引用。之所以不可以直接传 some_vec，是因为 std::thread 或者 std::jthread 的构造函数会把你传进去的参数先按值保存一份，随后再使用这个副本来调用 f。

使用函数对象作为 callable，在函数对象内定义引用类型的成员

struct F {
    F(vector<double>& v) : v_(v) {}
    vector<double>& v_;
    operator ()(); 
};
vector<double> vec2;
jthread t2 {F{vec2}};

F() 的

F::v_ 成员是引用类型的。在 std::jthread 构造函数中构造 F 的时候把 vec2 作为引用传入并初始化 v_

总结一下线程任务启动时可以接收的参数：

值：复制数据，隔离性好
const&：const 引用，避免复制，但是要求被引用的对象在任务结束前仍然有效
普通引用或指针：允许任务修改外部对象，有同步和生命周期风险
std::ref 和 std::cref：显式告诉线程构造器使用引用

任务参数设计本身是并发设计的重要部分。参数涉及的部分原则包括：

默认优先按值传递，减少共享状态和数据竞争
只读大对象时，可使用 const& 但是要保证对象生命周期足够长
使用普通引用或指针，任务会修改外部数据，此时必须明确同步规则
结果优先通过 future 返回

返回结果

这一块讲线程函数如何把结果返回创建它的线程

普通函数可以直接返回，但是 std::thread 不会接收线程函数的返回值。因此做法有 2 种：

通过指针保存返回值

void f(const vector<double>& v, double* res);

线程在启动的时候把指针作为参数传入 task 函数，然后线程内部把结果写到指针

double result1;
thread t1{f, cref(vec1), &result};

lambda 通过引用捕获外部变量，线程函数直接修改它：

定义线程的任务函数 f()，然后再用 lambda 封装它。然后在外部定义用来保存任务结果的变量 res2，lambda 用引用捕获这个变量，然后在 lambda 内部调用 f() 并把结果存入 res2。之所以用引用捕获是为了在 lambda 内修改 res2 这个变量。

double f() {...}
std::vector<double> vec2 {1, 2, 3, 4};
double result2;

thread t2{[&]() {
    result2 = f(vec2);
}};

之后必须先确保 t1 和 t2 都 join() 了，才能安全地使用 result1 和 result2

这种“通过引用/指针把结果写出去”的方式很常见，但不够优雅，后面会介绍更好的方式，比如 future/promise。

共享数据

如果多个线程访问同一个可变对象，并且至少一个线程会写，就必须避免数据竞争

std::mutex 是一种资源

锁本身会带来风险：

忘记释放锁
异常路径没有解锁
锁粒度过大导致并发度下降
多把锁顺序不一致导致死锁

因此建议不要直接使用裸 std::mutex（以及手动 lock()、unlock()），而是使用资源句柄来管理裸锁，比如 std::scoped_lock 和 std::unique_lock。

std::scoped_lock

#include <mutex>

std::mutex m;
int sh; // 共享的数据

void f() {
    std::scoped_lock lck {m};
    sh += 7;
}

这里的 std::scoped_lock lck {m} 完整写法是 std::scoped_lock<mutex> lck {m}。这里 std::scoped_lock 的构造函数接收 mutex 对象，然后调用 m.lock() 尝试获取这个锁。如果其它线程已经获取了这个锁，则此时当前线程阻塞了，直到其它线程释放这个锁。

一旦当前线程执行 f() 完毕，即出作用域了，那么 RAII 就会调用 scoped_lock 的析构函数。这个析构函数会调用 m.unlock() 来释放这个锁。由此，其它等待这个锁的线程就可以被唤醒。

使用 std::mutex 来保护数据的惯常做法是把要保护的数据设为类的私有成员，然后在公有成员中暴露操作数据的方法，并且在方法上使用 mutex 来保护数据操作。这里需要依赖程序员的意识（如果忘记给数据操作方法写 Mutex，就相当于没有保护），而且如果把数据作为公有成员，也相当于没有保护（可以绕开数据操作函数直接操作数据）。

某些操作需要获取多把锁才能推进。比如在交易系统当中，跨多个交易品种更新组合风险或同时从订单簿移除订单并更新风险状态。

std::scoped_lock 支持同时等待多个 std::mutex：

void f() {
    std::scoped_lock lck {mutex1, mutex2, mutex3};
}

在这里，std::scoped_lock 会等待所有 mutex 都齐全之后才全部调用 lock() 并且推进任务，因此不会持有部分 mutex 等待其它 mutex（导致死锁）。

最后对比一下 共享数据通信 与 函数调用

通过共享数据通信：

两个线程在通信时会暴露内部状态（比如 bool done、int result 之类的）
需要程序员手动设计 mutex、设计 done 何时变为 true 等等

通过函数调用 int result = compute();：

有 “调用 -> 执行 -> 返回结果” 的固定流程
因此，“工作有没有完成” 这个问题已经由 call / return 模型解决了

有这种观念：“共享数据通信不用复制，所以一定更快”。其实不一定，因为现代 CPU 对于函数调用中的数据复制有很多优化，可以做得很快。

std::unique_lock

在仅需要约束一把 mutex 的前提下，std::unique_lock 比 std::scoped_lock 更灵活但是更复杂。std::mutex 支持：

手动 lock() / unlock()
延迟加锁
尝试加锁 std::try_to_lock
和 std::condition_variable 配合

而 std::scoped_lock 适合仅需要简单 RAII 机制的情况，或者需要同时管理多把锁的情况。

std::shared_lock

std::shared_lock 是给 std::shared_mutex 使用的 “共享读锁”。它允许多个线程同时读，但是只要有线程在写，读线程就要等。单一写入者 + 若干读取者是很常见的模型，可以用 std::shared_lock 来约束读者、用 std::unique_lock 来约束写入者。这样的效果是读者可以和其它读者同时读；写入者必须独占锁（不可同时有读者，也不可同时有其它写入者）。

std::shared_mutex mx;
int data = 0;

void reader() {
    std::shared_lock lck {mx};
    int x = data;
}

void writer() {
    std::unique_lock lck {mx};   
    data = 1;
}

std::atomics

mutex 是比较重的机制，它可以保护任意数量的任务完成、避免数据竞争。而 std::atomic 针对更轻量的任务避免数据竞争。std::atomic<T> 让操作不被其它线程观察到中间状态。

std::mutex mut;
std::atomic<bool> init_x;  // 表示是否已经初始化。若已经初始化则为 true
X x;

if (!init_x) {
    std::scoped_lock lck {mut};
    if (!init_x) {
        // nontrivial initialization of x
        init_x = true;
    }
}

这里有双重检查：

第一次 if (!init_x)：已经初始化时，跳过加锁
加锁后第二次检查 if(!init_x)：防止在 x 初始化前，多个线程同时看到 “未初始化”，排队获得锁之后又重复初始化（虚假唤醒）

等待事件：condition_variable

有时一个线程需要等待外部条件/某个事件（比如另一个线程完成某个任务）

这一块涉及生产者消费者模型。典型的生产者消费者模型是：

生产者：

加锁
放入消息
通知消费者
解锁

消费者：

加锁
如果队列为空，则等待
被唤醒后重新检查条件
取出消息
解锁
处理消息

std::queue<Message> mqueue;
std::condition_variable mcond;
std::mutex mmutex;

void consumer() {
    while (true) {
        std::unique_lock lck {mmutex};
        mcond.wait(lck, [] { return !mqueue.empty(); });

        auto m = mqueue.front();
        mqueue.pop();
        lck.unlock();  // 这里手动 unlock 了，因为在作用域内还要处理消息

        processMessage(m);
    }
}

void producer() {
    while (true) {
        Message m;
        std::scoped_lock lck {mmutex}; // 此处使用 scoped_lock 是为了简洁（没必要使用 unique_lock）
        mqueue.push(m);
        mcond.notify_one();
    } // 出作用域之后会自动释放 scoped_lock
}

之所以在 consumer() 当中，在 std::unique_lock 之后还有 mcond.wait() 检查队列是因为：

应当获取锁之后再检查队列，避免刚刚检查完队列，队列状态就变化
consumer() 获取锁是因为想要处理队列中的信息，但是有可能多个 consumer 都想要处理同样的信息。因此检查队列可以防止虚假唤醒（即唤醒了，但发现信息已经被其它 consumer 处理掉了）