condition_variable
类是用来阻塞线程或者是多线程的同步原语,直到另一个线程修改完共享变量(the condition),并通知
condition_variable
。
尝试修改变量的线程必须:
- 获得std::mutex,典型的通过std::lock_guard()
- 加锁后对变量进行修改
- 在std::condition_variable上执行 notify_one or notify_all
即使共享变量是原子性的,为了正确的把修改传递给等待线程,共享变量在互斥量下也必须被修改。
任何在std::condition_variable下尝试等待的线程必须:
- 获得std::unique_lock<std::mutex>,该mutex通常用来保护共享变量
- 执行 wait, wait_for, or wait_until。该等待操作原子性的释放互斥量并阻塞线程执行。
- 当条件变量被通知,或者是超时,或者是虚假唤醒发生,该线程被唤醒,互斥量被原子性的获得。线程应该检查一下条件,如果是虚假唤醒的话应该继续等待;
std::condition_variable只能和std::unique_lock<std::mutex>一起使用,在某些平台上能够使资源最大限度的利用。
一些函数:
让我们来看一下wait的两种函数形式:
- notify_one 通知一个等待线程
- notify_all 通知所有等待线程
- wait 阻塞当前线程直到条件变量被唤醒
- wait_for 阻塞当前线程直到条件变量被唤醒或者是过了特定的时间间隔
- wait_until 阻塞当前线程直到条件变量被唤醒,或者是达到了特定的时间点
我们首先看一下wait()函数的两种形式:
void wait( std::unique_lock<std::mutex>& lock );
template< class Predicate >
void wait( std::unique_lock<std::mutex>& lock, Predicate pred );
wait()函数的主要功能,阻塞当前线程,将线程挂起,直到条件变量被通知或者是虚假唤醒出现。
1)调用wait()函数之后,原子性地释放锁(解锁),阻塞正在执行的线程,并把线程加入等待的线程列表(线程挂起),当调用 notify_all() or notify_one()的时候线程变为不阻塞状态(通过虚假唤醒也可),当线程不阻塞的时候,无论是什么原因,lock都被重新获得(加锁),wait()函数退出,如果函数是通过异常退出的,lock也会被重新获得。
2) 等价于:
while (!pred()) {
wait(lock);
}
这个重载函数可以通过设置pred值(true或者false)来忽略虚假唤醒,在while之前必须加锁,防止有别的线程修改pred,调用wait函数时执行解锁操作,使其它线程可以加锁,本线程挂起,收到通知后wait函数退出,再次加锁,线程开始执行。
lock -
std::unique_lock<std::mutex>的对象,必须被当前线程锁住;
pred - 如果要继续等待的话,返回false。
在详细讲解之前,我们看一段代码,了解一下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;
}
接下来基于上述代码,我们分析一下多线程代码是如何运行的。
首先,定义了几个全局变量,互斥量(mutex),条件变量(cv),标志量(ready)。整个代码的作用就是,当是个线程都准备好了之后,再并发执行,也就是说使用条件变量起到同步的作用。
在代码中结合使用了mutex与condition_variable,请大家仔细阅读一下代码,然后再阅读一下下面的解析。
1. 在print_id中,线程先将互斥量上锁(使用的unique_lock<mutex>),然后判断ready,若ready为false,说明条件不满足,那么调用条件变量的wait方法将线程挂起,同时解锁;
2. 当所有线程都处于等待状态的时候,说明所有线程都已就绪,此时在主线程中将reday设为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>
std::mutex m;
std::condition_variable cv;
std::string data;
bool ready = false;
bool processed = false;
void worker_thread()
{
// Wait until main() sends data
std::unique_lock<std::mutex> lk(m);
cv.wait(lk, [] {return ready; });
// after the wait, we own the lock.
std::cout << "Worker thread is processing data\n";
data += " after processing";
// Send data back to main()
processed = true;
std::cout << "Worker thread signals data processing completed\n";
// Manual unlocking is done before notifying, to avoid waking up
// the waiting thread only to block again (see notify_one for details)
lk.unlock();
cv.notify_one();
}
int main()
{
std::thread worker(worker_thread);
data = "Example data";
// send data to the worker thread
{
std::lock_guard<std::mutex> lk(m);
ready = true;
std::cout << "main() signals data ready for processing\n";
}
cv.notify_one();
// wait for the worker
{
std::unique_lock<std::mutex> lk(m);
cv.wait(lk, [] {return processed; });
}
std::cout << "Back in main(), data = " << data << '\n';
worker.join();
}
////output
main() signals data ready for processing
Worker thread is processing data
Worker thread signals data processing completed
Back in main(), data = Example data after processing
可以仔细瞧瞧上述代码,拢共两个线程:主线程、线程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()唤醒;
以上就是本篇博文所剖析的知识点,如果有表达不清或者有误的地方,请各位大佬多多指教~
参考博文:
https://blog.csdn.net/hujingshuang/article/details/70596630 《探索C++多线程》各种源码分析。