condition_variable

2020-12-23

为了实现同步化，被不同线程执行的task必须彼此等待。使用future可以停下来直到另外一个线程提供数据或直到另外一个线程结束，但是future从某个线程传递数据到另外一个线程时只能传递一次，并且future主要用来处理线程的返回值或者异常。

使用condition variable(条件变量),可用来同步化线程之间的数据流逻辑依赖关系。使用condition variable可以唤醒一个或多个其他等待中的线程。C++标准库在头文件中中声明了condition_variable。

用法：

1.在使用condition_variable的时候，必须同时包含头文件和，并声明一个mutex和一个condition_variable的变量。

2.等到条件满足的线程（或多线程之一）必须调用条件变量中的notify_one()或者notify_all()，通知那个线程，并唤醒条件满足的线程。notify_one()时唤醒其中一个满足条件的，而notify_all()则会唤醒所有等待的线程。

3.添加满足的线程必须先获取锁,并使用wait()等待锁是否满足，如果等到锁那么就会唤醒这个线程。

在等待函数wait()中可能会锁定或解锁mutex。使用condition variable可能会发生wait动作发生在notify之前，这样有可能线程所需要的条件并不满足，从而导致程序不可预期的行为。为了避免这种情况，需要在线程唤醒后去验证条件实际是否已经完成。示例代码如下：

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <future>
bool readyFlag;
std::mutex readyMutex;
std::condition_variable readyCondVar;
void thread1(){
    //do something for thread2
    std::cout << "thread 1" << std::endl;
    std::cin.get();
    //get lock and lock mutex
    {
        std::lock_guard<std::mutex> lg(readyMutex);
        readyFlag = true;
    }//release lock
    readyCondVar.notify_one(); // notify one
}
void thread2(){
    //wait for thread1
    {
        std::unique_lock<std::mutex> ul(readyMutex);
        readyCondVar.wait(ul,[]{return readyFlag;});
    }
    std::cout << " done " << std::endl;
}
int main(int argc,char *argv[])
{
    std::future<void> f1 = std::async(std::launch::async,thread1);
    std::future<void> f2 = std::async(std::launch::async,thread2);
    return 0;
}

数据供应者thread1()锁住readyMutex，更新数据readyFlag，解锁然后通知condition variable。等待者线程thread2(),对readyMutex加锁，然后使用wait()等待条件满足，wait()函数第一个参数为一个 lock ul，第二个参数是一个lambda匿名函数，用来二次检测条件是否满足，其中上面thread2()中的wait语句相当于下面的代码：

1
2
3

while(!readyFlag){
    readyCondVar.wait(ul);
}

下面使用condition_variable解决生成者和消费者问题。代码如下：

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <future>
#include <queue>
#include <chrono>
#include <thread>
std::queue<int> que;
std::mutex qMutex;
std::condition_variable queConVar;
void Provider(int val){
    for(int i=0;i<6;++i){
        std::lock_guard<std::mutex> lg(qMutex);
        que.push(val+i);
    }
    queConVar.notify_one();
    //sleep for val millseconds
    std::this_thread::sleep_for(std::chrono::microseconds(val));
}
void Consumer(int num){
    while(true){
        int val;
        {
            std::unique_lock<std::mutex> ul(qMutex);
            queConVar.wait(ul,[]{return !que.empty();});
            val = que.front();
            que.pop();
        }
        std::cout << "consumer : " << num <<  " : " << val << std::endl;
    }
}
int main(int argc,char **argv)
{
    //provider
    auto f1 = std::async(std::launch::async,Provider,100);
    auto f2 = std::async(std::launch::async,Provider,100);
    auto f3 = std::async(std::launch::async,Provider,100);
    //consumer
    auto f4 = std::async(std::launch::async,Consumer,1);
    auto f5 = std::async(std::launch::async,Consumer,2);
    return 0;
}

condition variable也提供了一个接口允许你等待某个最大的时间量：wait_for()用来等待一个时间段，wait_until()用来等待直到某个时间点。

在程序中如果无法建立condition variable，构造函数会抛出std::system_error异常，并有错误码resource_unavailable_try_again.对于条件变量的copy和赋值都是不允许的。condition variable的一些操作如下：

操作	含义
condition_variable cv	默认构造函数，建立一个condition_variable变量
cv.~condition_variable()	析构函数
cv.notify_one()	唤醒一个等待线程
cv.notify_all()	唤醒所有等待线程
cv.wait(ul)	使用uniqe_lock ul等待通知
cv.wait(ul,pred)	使用uniqe_lock ul等待通知,并且直到pred再一次唤醒后为true
cv.wait_for(ul,duration)	使用uniqe_lock ul等待通知，等待时间时长duration
cv.wait_for(ul,duration,pred)	使用uniqe_lock ul等待通知，等待时间时长duration或者并且直到pred再一次唤醒后为true
cv.wait_until(ul,timepoint)	使用uniqe_lock ul等待通知，等待到时间点timepoint
cv.wait_until(ul,timepoint,until)	使用uniqe_lock ul等待通知，等待时间到时间点timepoint或者并且直到pred再一次唤醒后为true
cv.native_handle()	返回一个因平台而异的类型
notify_all_at_thread_exit(cv,ul)	在调用所在的本线程唤醒所有使用unique lock ul来等待cv的线程

所有通知都会被自动同步化，所以并发调用notify_one()和notify_all()不会带来麻烦。