并发操作的同步

前面学习了如何保护线程间的共享数据。然而，有时候我们不仅需要保护共享数据，还需要令独立线程上的行为同步。例如，某线程只有先等另一线程的任务完成，才可以执行自己的任务。一般而言，线程常常需要等待特定事件的发生，或等待某个条件成立。只要设置一个“任务完成”的标志，或者利用共享数据存储一个类似的标志，通过定期查验该标志就可以满足需求，但这远非理想方法

上述线程间的同步操作很常见，C++标准库专门为之提供了处理工具：条件变量（conditional variable）和future

条件变量（conditional variable）

设想你坐夜行列车外出。如果要保证在正确的站点下车，一种方法是彻夜不眠，留心列车停靠的站点，那样就不会错过。可是，到达时你可能会精神疲倦。或者，你可以查看时刻表，按预定到达时间提前设定闹钟，随后安心入睡。这种方法还算管用，一般来说，你不会误站。但若列车晚点，你反而会太早醒来；也可能不巧，闹钟的电池刚好耗尽，结果你睡过头而错过下车站点。最理想的方法是，安排人员或设备，无论列车在什么时刻抵达目的站点，都可以将你唤起，那么你大可“高枕无忧”

同样的，如果线程甲要等待线程乙完成任务，可以采取集中不同的方式

方式一：在共享数据内部维护一标志（受互斥保护），线程乙完成任务后，就设置标志成立

该方式存在双重浪费：线程甲须不断查验标志，浪费原本有用的处理时间；另外，一旦互斥被锁住，则其他任何线程无法再加锁。这两点都是线程甲的弊病：如果它正在运行，就会限制线程乙可用的算力；还有，线程甲每次查验标志，都要锁住互斥以施加保护，那么，若线程乙恰好同时完成任务，也意欲设置标志成立，则无法对互斥加锁。这就像是你整晚熬夜，不停地与列车司机攀谈，于是他不得不放慢车速，因为你老使他走神，结果列车晚点。类似地，线程甲白白耗费了计算资源，它们本来可用于系统中的其它线程，最终导致毫无必要的等待时间

方式二：让线程甲调用 std:this_thread:sleep for( )函数，在各次查验之间短期休眠

bool flag;
mutex m;
void wait_for_flag()
{
	unique_lock<mutex> lk(m);
	while(!flag)
	{
		lk.unlock();
		this_thread::sleep_for(chrono::milliseconds(100));
		lk.lock();
	}
}

上面的代码在每轮循环中，先将互斥解锁，随之休眠，再重新加锁，从而其它线程有机会获取锁，得以设置标志成立

这确有改进，因为线程休眠，所以处理时间不再被浪费。然而，休眠期的长短却难以预知。休眠期太短，线程仍会频繁查验，虚耗处理时间；休眠期太长，则令线程过度休眠。如果线程乙完成了任务，线程甲却没有被及时唤醒，就会导致延迟。过度休眠很少直接影响普通程序的运作。但是，对于高速视频游戏，过度休眠可能会造成丢帧；对于实时应用，可能会使某些时间片计算超时

方式三：使用C++标准库的工具等待事件发生

以上述甲、乙两线程的二级流水线模式为例，若数据要先进行前期处理，才可以开始正式操作，那么线程甲则需等待线程乙完成并且触发事件，其中最基本的方式是条件变量。按照“条件变量”的概念，若条件变量与某一事件或某一条件关联，一个或多个线程就能以其为依托，等待条件成立。当某线程判定条件成立时，就通过该条件变量，知会所有等待的线程，唤醒它们继续处理

condition_variable

头文件<condition_variable>

condition_variable
condition_variable_any

相同点：两者都能与std::mutex一起使用

不同点：前者仅限于与 std::mutex 一起工作，而后者可以和任何满足最低标准的互斥量一起工作，从而加上了_any的后缀，condition_variable_any会产生额外的开销

一般只推荐使用condition_variable，除非对灵活性有硬性要求，才会考虑condition_variable_any

条件变量的构造函数：

std::condition_variable::condition_variable
constructor:
    condition_variable();   //默认构造函数无参
    condition_variable(const condition_variable&) = delete;   //删除拷贝构造函数

示例代码：

#include <iostream>                // std::cout
#include <thread>                // std::thread
#include <mutex>                // std::mutex, std::unique_lock
#include <condition_variable>    // std::condition_variable

std::mutex mtx; // 全局互斥锁.
std::condition_variable cv; // 全局条件变量.
bool ready = false; // 全局标志位.

void do_print_id(int id)
{
    std::unique_lock <std::mutex> lck(mtx);
    while (!ready) // 如果标志位不为 true, 则等待...
        cv.wait(lck); // 当前线程被阻塞, 当全局标志位变为 true 之后,
    // 线程被唤醒, 继续往下执行打印线程编号id.
    std::cout << "thread " << id << '\n';
}

void go()
{
    std::unique_lock <std::mutex> lck(mtx);
    ready = true; // 设置全局标志位为 true.
    cv.notify_all(); // 唤醒所有线程.
}

int main()
{
    std::thread threads[10];
    // spawn 10 threads:
    for (int i = 0; i < 10; ++i)
        threads[i] = std::thread(do_print_id, i);

    std::cout << "10 threads ready to race...\n";
    go(); // go!

  for (auto & th:threads)
        th.join();

    return 0;
}

打印输出：

10 threads ready to race...
thread 6
thread 7
thread 8
thread 9
thread 5
thread 4
thread 3
thread 2
thread 1
thread 0

wait( )与notify_one( )

wait( )函数

void wait( std::unique_lock<std::mutex>& lock );

//Predicate是lambda表达式
template< class Predicate >
void wait( std::unique_lock<std::mutex>& lock, Predicate pred );
//以上二者都被notify_one())或notify_all()唤醒，但是
//第二种方式是唤醒后也要满足Predicate的条件
//如果不满足条件，继续解锁互斥量，然后让线程处于阻塞或等待状态
//第二种等价于
while (!pred())
{
    wait(lock);
}

std::condition_variable 提供了两种 wait() 函数。当前线程调用 wait() 后将被阻塞(此时当前线程应该获得了锁（mutex），不妨设获得锁 lck)，直到另外某个线程调用 notify_* 唤醒了当前线程

在线程被阻塞时，该函数会自动调用 lck.unlock() 释放锁，使得其他被阻塞在锁竞争上的线程得以继续执行
另外，一旦当前线程获得通知(notified，通常是另外某个线程调用 notify_* 唤醒了当前线程)，wait() 函数也是自动调用 lck.lock( )，使得 lck 的状态和 wait 函数被调用时相同

notify_one 唤醒等待的一个线程，注意只唤醒一个
notify_all 唤醒所有等待的线程，使用该函数时应避免出现惊群效应

wait( )函数在第二种情况下（即设置了 Predicate），只有当 pred 条件为 false 时调用 wait( ) 才会阻塞当前线程，并且在收到其他线程的通知后只有当 pred 为 true 时才会被解除阻塞

示例代码：

//  condition_variable simple example
//  Created by lei on 2022/05/15

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
using namespace std;

vector<int> data_vec;
int vec_size = 10;
bool prepared = false;
bool processed = false;

mutex m_mutex;
condition_variable m_cond_var;

void prepareData()
{
    data_vec.reserve(vec_size);
    for (int i = 0; i < vec_size; ++i)
    {
        data_vec.emplace_back(i + 1);
    }
}

void processData()
{
    for (int i = 0; i < vec_size; ++i)
    {
        data_vec[i] *= 2;
    }
}

void showData()
{
    for (int i = 0; i < vec_size; ++i)
    {
        cout << data_vec[i] << endl;
    }
}

void work()
{
    unique_lock<mutex> lk(m_mutex);
    m_cond_var.wait(lk, []
                    { return prepared; });
    cout << "Work thread is processing data..." << endl;
    processData();
    this_thread::sleep_for(chrono::seconds(2));
    processed = true;
    lk.unlock();
    m_cond_var.notify_one();
}

// use producer-consumer pattern
int main()
{
    thread worker(work);

    {
        lock_guard<mutex> lk(m_mutex);
        cout << "Preparing data..." << endl;
        prepareData();
        cout << "Before process:" << endl;
        showData();
        prepared = true;
    }
    m_cond_var.notify_one();

    {
        unique_lock<mutex> lk(m_mutex);
        m_cond_var.wait(lk, []
                        { return processed; });
        cout << "After process:" << endl;
        showData();
    }

    if (worker.joinable())
    {
        worker.join();
    }

    return 0;
}

打印输出：

Preparing data...
Before process:
1
2
3
4
5
6
7
8
9
10
Work thread is processing data...
After process:
2
4
6
8
10
12
14
16
18
20

spurious wakeups(虚假唤醒)

需要注意的一点是, wait有时会在没有任何线程调用notify的情况下返回，这种情况就是有名的spurious wakeup

这种情况会出现在第二种wait函数中，即：

template< class Predicate >
void wait( std::unique_lock<std::mutex>& lock, Predicate pred );

spurious wake产生的原因就是当前置判断条件pred为true时，即使没有线程notify，只要锁已经被释放，wait函数就会立刻返回，继续执行下面的代码

如将上面的示例程序稍加修改：

// use producer-consumer pattern
int main()
{
    thread worker(work);

    {
        lock_guard<mutex> lk(m_mutex);
        cout << "Preparing data..." << endl;
        prepareData();
        cout << "Before process:" << endl;
        showData();
        prepared = true;
        this_thread::sleep_for(chrono::seconds(5));
    }
    // m_cond_var.notify_one();