《探索C++多线程》：condition_variable源码（一）-526互联

https://blog.csdn.net/hujingshuang/article/details/70596630

现在接着学习关于多线程编程的特征，在这一节，将会了解到多线程中的condition_variable（条件变量）的相关知识。

在头文件<condition_variable>中有两种条件变量的类声明与定义：condition_variable、condition_variable_any，在这一节中先来了解condition_variable。

condiction_variable

条件变量是一种多线程的同步机制，它能够阻塞线程，直到某一条件满足。条件变量要与互斥量联合使用，以避免出现竞争的情况，当调用condition_variable的一个等待函数时，它使用一个unique_lock对象来锁定线程。

源码之前，了无秘密。下面就将来看一下condition_variable源码，如下：

class condition_variable {
public:
typedef _Cnd_t native_handle_type;
condition_variable() { // 构造函数，初始化条件变量，所有的条件变量必须初始化后才能使用。
_Cnd_initX(&_Cnd);
}
~condition_variable() _NOEXCEPT { // 析构函数
_Cnd_destroy(&_Cnd);
}
condition_variable(const condition_variable&) = delete;
condition_variable& operator=(const condition_variable&) = delete;
void notify_one() _NOEXCEPT { // 唤醒一个在等待线程
_Cnd_signalX(&_Cnd);
}
void notify_all() _NOEXCEPT { // 唤醒所有在等待的线程
_Cnd_broadcastX(&_Cnd);
}
void wait(unique_lock<mutex>& _Lck) { // 等待
_Cnd_waitX(&_Cnd, &_Lck.mutex()->_Mtx);
}
template<class _Predicate>
void wait(unique_lock<mutex>& _Lck, _Predicate _Pred) { // 等待，带有描述式
while (!_Pred())
wait(_Lck);
}
template<class _Rep, class _Period>
_Cv_status wait_for(unique_lock<mutex>& _Lck, const chrono::duration<_Rep, _Period>& _Rel_time) {
stdext::threads::xtime _Tgt = _To_xtime(_Rel_time);
return (wait_until(_Lck, &_Tgt));
}
template<class _Rep, class _Period, class _Predicate>
bool wait_for(unique_lock<mutex>& _Lck, const chrono::duration<_Rep, _Period>& _Rel_time, _Predicate _Pred) {
stdext::threads::xtime _Tgt = _To_xtime(_Rel_time);
return (wait_until(_Lck, &_Tgt, _Pred));
}
template<class _Clock, class _Duration>
_Cv_status wait_until( unique_lock<mutex>& _Lck, const chrono::time_point<_Clock, _Duration>& _Abs_time) {
typename chrono::time_point<_Clock, _Duration>::duration
_Rel_time = _Abs_time - _Clock::now();
return (wait_for(_Lck, _Rel_time));
}
template<class _Clock, class _Duration, class _Predicate>
bool wait_until(unique_lock<mutex>& _Lck, const chrono::time_point<_Clock, _Duration>& _Abs_time, _Predicate _Pred) {
typename chrono::time_point<_Clock, _Duration>::duration
_Rel_time = _Abs_time - _Clock::now();
return (wait_for(_Lck, _Rel_time, _Pred));
}
_Cv_status wait_until( unique_lock<mutex>& _Lck, const xtime *_Abs_time) {
if (!_Mtx_current_owns(&_Lck.mutex()->_Mtx))
_Throw_Cpp_error(_OPERATION_NOT_PERMITTED);
int _Res = _Cnd_timedwaitX(&_Cnd, &_Lck.mutex()->_Mtx, _Abs_time);
return (_Res == _Thrd_timedout ? cv_status::timeout : cv_status::no_timeout);
}
template<class _Predicate>
bool wait_until(unique_lock<mutex>& _Lck, const xtime *_Abs_time, _Predicate _Pred) {
bool _Res = true;
while (_Res && !_Pred())
_Res = wait_until(_Lck, _Abs_time)
!= cv_status::timeout;
return (_Pred());
}
native_handle_type native_handle() { // 返回条件变量的句柄
return (_Cnd);
}
void _Register(unique_lock<mutex>& _Lck, int *_Ready) {
_Cnd_register_at_thread_exit(&_Cnd, &_Lck.release()->_Mtx, _Ready);
}
void _Unregister(mutex& _Mtx) {
_Cnd_unregister_at_thread_exit(&_Mtx._Mtx);
}
private:
_Cnd_t _Cnd;
};

condition_variable是一个类，它的定义有一点长，不过我们可以注意到，不外乎就是：私有成员变量、构造函数、析构函数、等待与唤醒方法。

首先，我们来看构造函数和析构函数：

condition_variable() { // 构造函数，初始化条件变量，所有的条件变量必须初始化后才能使用。
_Cnd_initX(&_Cnd);
}
~condition_variable() _NOEXCEPT { // 析构函数
_Cnd_destroy(&_Cnd);
}

构造函数只有一个，且不带任何参数；而且该类中禁止了拷贝构造函数和赋值函数，这样一来我们要构造一个对象就比较简单了。
其次，我们来分析condition_variable的等待、唤醒操作：

等待操作：wait()、wait_for()、wait_until()

唤醒操作：notify_one()、notify_all()

在详细讲解之前，先来看一段代码，然大家了解是如何使用condition_variable的，如下：

#include <iostream> // std::cout
#include <thread> // std::thread
#include <mutex> // std::mutex, std::unique_lock
#include <condition_variable> // std::condition_variable
using namespace std;
mutex mtx; // 互斥量
condition_variable cv; // 条件变量
bool ready = false; // 标志量
void print_id(int id) {
unique_lock<mutex> lck(mtx); // 上锁
while (!ready) {
cv.wait(lck); // 线程等待直到被唤醒（释放锁 + 等待，唤醒，在函数返回之前重新上锁）
}
cout << "thread " << id << '\n';
}
void go() {
unique_lock<mutex> lck(mtx); // 上锁
ready = true;
cv.notify_all(); // 唤醒所有正在等待（挂起）的线程（在这里面要释放锁，为了在wait函数返回之前能成功的重新上锁）
}
int main() {
thread threads[10];
for (int i = 0; i<10; ++i) {
threads[i] = thread(print_id, i);
}
cout << "10 threads ready to race...\n";
go();
for (auto& th : threads) {
th.join();
}
return 0;
}

接下来基于上述代码，我们来分析一下多线程代码时如何运行的。

首先，定义了几个全局变量：互斥量（mtx）、条件变量（cv）、标志量（ready）。整个代码的作用就是，当10个线程都准备好了之后，再并发执行（就好像赛马，把马牵出来，当所有马匹在赛道上就绪之后，再开始跑），也就是使用条件变量起到同步的作用。

在代码中结合使用了mutex与condition_variable，请大家仔细阅读一下代码，然后再仔细阅读下面的解析：

1、在print_id中，线程先将互斥量上锁（使用的unique_lock<mutex>），再判断ready，若ready为false，说明条件不满足，那么调用条件变量的wait()方法将线程挂起；

2、当所有的线程都在等待状态时，说明所有线程已就绪，此时在主线程中将ready设为true，并调用notify_all()将所有挂起线程唤醒；

3、所有线程被唤醒之后，并发执行打印自己的id；

4、使用thread::join()方法，等所有线程都执行完毕，主线程才接着执行，直到出现结束。

我们举例代码的运行过程就是这样，但是我们仔细想一想第1个过程：当有一个线程一来，就将互斥量先上了锁，然后发现条件不满足，就进入了挂起状态，此时代码中并没有解锁操作，那么该线程就一直持有锁（锁被独占了），这样的话其他线程根本就没有机会获取锁，那还谈什么后面的全部线程挂起、并发执行呢？

实际上，上述分析并没有错，确实是这样的道理。但是这种问题是怎么解决的呢，其实这就是条件变量对象中wait()、notify()方法要处理的了，过程如下：

1、线程A一来，就将互斥量上锁（持有了锁），ready为false，那么线程A将调用条件变量的wait()方法；

2、在wait()方法中，做的第一件事就是将互斥量解锁（释放持有权），并进入等待状态（在wait()中阻塞，线程A挂起）；

3、现在线程B来了，互斥量是没有上锁的，所以线程B能持有锁，同理，接下来线程B也会挂起；

4、当所有线程都挂起了（就绪），此时互斥量也没有被上锁，在主线程中将ready置为true，并调用notify_all()将所有挂起的线程都唤醒；

5、此时所有线程将从wait()方法中返回，比如线程C先返回，在return之前，wait()方法做的最后一件事就是自动将互斥量上锁（线程C重新持有锁，以配合unique_lock的析构函数）；

6、由于while循环，此时再判断到ready为true，那么线程C将执行打印id的语句，由于此时只有线程C持有锁，不存在线程竞争问题，执行完打印之后，线程C就结束了，此时由unique_lock的析构函数解锁，释放所有权。

7、由于在wait()方法return之前，会自动重新去持有锁，若此时锁由线程C持有，则其他线程将继续阻塞，直到线程C释放锁；若线程C执行完毕后释放了锁，那么其他线程将会争取锁的持有权，争取到锁的就会像之前的线程C一样；没有争取到的就继续阻塞；

8、以此类推，由于每个线程都join，那么当所有线程执行完毕后，主线程才会继续执行；

实际上，条件变量的wait()、wait_for()、wait_until()方法中所作的事是：解锁 + 等待、唤醒、加锁，这三个是有序发生的。综上所述，这就是condition_variable与mutex联合使用的大致过程，这也是条件变量同步机制的原理。可能写得有些啰嗦。

下面，我们再来看一个例子，代码如下：

#include <iostream>
#include <string>
#include <thread>
#include <mutex>
#include <condition_variable>
using namespace std;
mutex m;
condition_variable cv;
string data;
bool ready = false;
bool processed = false;
void worker_thread() {
unique_lock<mutex> lk(m); // 【1】
cv.wait(lk, []{return ready;}); // 等价于 while(!ready) { cv.wait(lk); }
cout << "Worker thread is processing data\n";
data += " after processing";
processed = true;
cout << "Worker thread signals data processing completed\n";
//lk.unlock(); // 手动解锁
cv.notify_one();
}
int main() {
thread worker(worker_thread);
data = "Example data";
{
lock_guard<mutex> lk(m); // 【2】
ready = true;
cout << "main() signals data ready for processing\n";
}
cv.notify_one(); // 唤醒线程worker
{
unique_lock<mutex> lk(m); // 【3】
cv.wait(lk, []{return processed; }); // 等价于 while(!processed) { cv.wait(lk); }
}
cout << "Back in main(), data = " << data << '\n';
worker.join();
getchar();
return 0;
}

可以仔细瞧瞧上述代码，拢共两个线程：主线程、线程worker。接着我们来分析一下：

1、主线程执行，并创建了线程worker，此时便有两个线程在运行了；

2、可以看到代码中标出的【1】、【2】位置处都对互斥量进行上锁，由于线程的执行是不确定的，所以并不清楚先执行【1】还是【2】，也就是说有可能主线程先持有锁，也有可能是worker先持有锁；

3、分为两种情况：

情况一：worker先持有了锁（先执行【1】），那么主线程将会被阻塞。此时由于ready为false，根据之前讲的，worker执行wait()方法时会解锁，并进入等待状态；解锁后，主线程就会争取持有锁，由于worker在等待状态，没有其他线程跟主线程竞争，主线程会立即获取锁的持有权（执行【2】），并将ready置为true，在【2】所在的代码块结束后，lock_guard的析构函数会释放锁，接下来主线程调用notify_one()方法唤醒一个正在等待的线程（这里只有worker线程处于等待状态）；当worker线程被唤醒后，也就意味着要离开wait()了，在离开之前要做的最后一件事是上锁，也就是使worker重新持有锁。那么此时问题来了，由于worker已经醒来，与主线程并发执行，那到底是worker先重新持有锁，还是主线程的【3】先持有锁呢？这是一个要考虑的问题；

情况二：主线程先持有了锁（先执行【2】），那么worker将会被阻塞。在主线程中会将ready置为true，并结束【2】所在的语句块，释放锁的持有权；由于线程worker一直阻塞的，一旦互斥量解锁了，线程worker就会争取锁的持有权，那到底是worker先持有锁，还是主线程的【3】先持有锁呢？这是一个要考虑的问题。

4、

在3的情况一中：如果线程worker在wait()中争取到了锁的持有权，也就是在wait()方法中先重新上锁了，那么主线程将与worker并发执行；

① 若在worker线程手动解锁之前，执行了主线程的【3】，那么主线程将被阻塞（因为worker正持有锁），worker将processed置为true，然后手动解锁；此时主线程与worker并发执行，而且主线程将立即持有锁（没有线程竞争了），由于此时processed为ture，主线程就不会进入等待状态，而是继续执行，直到worker执行完毕后，主线程才会结束；

② 若在worker线程手动解锁之后，执行了主线程的【3】，那么主线程不会进入等待状态（worker中processed已置为true了），而是继续执行，直到worker执行完毕后，主线程才会结束；

在3的情况一中：如果主线程争取到了锁的持有权，也就是在wait()方法中重新上锁之前，执行了主线程的【3】，那么worker将继续阻塞在wait()方法中；此时processed为false，主线程调用wait()方法，释放锁，并进入等待状态，此时worker将在wait()中重新上锁成功（无线程竞争了），worker继续执行，将processed置为true，并进行手动解锁，最后调用notify_one()唤醒一个等待的线程（有且只有主线程）；主线程被唤醒后，继续执行，直到worker执行完毕后，主线程才会结束。

在3的情况二中，如果主线程争取到了锁的持有权，也就是【3】在【1】之前执行了，那么worker将继续阻塞，由于此时processed为false，主线程将调用wait()方法，解锁并进入等待状态；互斥量解锁之后，worker将会立即持有锁（由于主线程处于等待状态，没有竞争的线程了）；此时，worker中判断ready，发现ready已经是true了，那么worker将继续执行不进入等待状态，执行中将processed置为true，并进行了手动解锁，最后调用notify_one()方法来唤醒一个等待状态的线程（有且只有主线程被唤醒）；主线程被唤醒之后，将从wait()中返回，在返回之前做的最后一件事是重新持有锁（此时直接就持有了，没有竞争的线程）；最后继续执行，直到worker执行完毕，主线程才会结束。

再来说一说为什么需要做手动解锁这一步：

其实在这个例子中，做不做手动解锁这一步，没有太大的影响，也不会影响多线程的运行结果，根据上述分析，我所能想到的影响就是，存在这么一种情况：如果不手动释放锁（worker中的lk.unlock()这条语句被注释掉了），那么主线程在等待状态时，如果worker调用了notify_one()将主线程唤醒，此时由于worker还持有锁，所以主线程的wait()方法在重新上锁时，依然会被阻塞，这种情况直到线程worker运行完毕，unique_lock的析构函数释放了锁，那么此时主线程才能重新上锁成功；也就是说，手动解锁避免了主线程被唤醒后还要被阻塞的情况。

另外，还有一点：为什么要使用unique_lock对象来对加锁、解锁呢？请参见我的下一篇文章：《探索C++多线程》：condition_variable源码（二）。

这一节花了大篇幅分析了两个例子，我觉得这样才能更好理解和运用condition_variable，可能存在很多口水话的地方，希望不要介意。

condition_variable的方法：

condition_variable::notify_one()：唤醒一个处于等待状态的线程；

condition_variable::notify_all()：唤醒所有处于等待状态的线程；

condition_variable::wait()：将线程置于等待状态，直到被notify_xxx()唤醒；

condition_variable::wait_for()：将线程置于等待状态，直到一段时间结束后自动醒来或被notify_xxx()唤醒；

condition_variable::wait_until()：将线程置于等待状态，直到指定的时间点到来自动唤醒或被notify_xxx()唤醒；

以上就是本篇博文所剖析的知识点，如果有表达不清或者有误的地方，请各位大佬多多指教~