C++同步机制之条件变量（std::condition_variable）

2022-09-10 22:57:19

文章目录

一、使用机制
二、成员函数
- 2.1 唤醒函数
- 2.2 等待函数
三、为什么需要互斥锁
四、虚假唤醒
五、示例

对多线程编程有了解的对条件变量（condition_variable）这个概念一定不陌生，相较于另一种常见的同步机制互斥锁，条件变量是一种更高层次的同步手段，互斥锁用于保护一段临界区，也就是各个线程互斥地去访问同一块资源，这是一种很简单直接的同步手段，可以cover住很大一类的同步场景，比如多线程修改一个非线程安全的unordered_map。但是对于另外一类很常见的同步需求却很吃力：线程需要用到其他线程产出的资源，也就是需要其他线程产出资源后才能继续进行本线程的处理，在此之前只能等待。此类场景仅使用mutex也是可以的，比如可以通过结合sleep和死循环，但实现方式和效率都不尽如人意。而condition_variable则可以比较好地应对此类场景，它可以让wait线程在condition不满足的时候阻塞等待在一个condition_variable变量上，随后notify线程可以修改condition并唤醒wait在condition_variable上的线程进行后续处理。
Linux下面提供了一系列pthread_cond_*的函数用于条件变量操作，在c++11里引入了std::condition_variable和std::condition_variable_any，正式将cv作为了c++的语言标准，机制上和pthread版本很类似，std::condition_variable_any和std::condition_variable的区别在于支持更多的锁，但在一些平台上性能更差，这里以std::condition_variable为例聊一聊条件变量。

一、使用机制

1.1 涉及的主要概念

在condition_variable的使用中，通常需要牵涉到以下角色：

condition_variable类型的变量：等待线程和唤醒线程之间的同步操作都依赖该变量。
一个共享变量 (condition)：上述cv变量只是提供了一个根据condition同步的机制，既然是根据condition同步同步，那么必然需要一个condition。
一个互斥锁mutex：这个额外的mutex是为了保证能正常唤醒，即便共享变量是 atomic也是需要的，后面会具体分析。
wait线程：当condition不满足的时候需要等待，直到条件满足了才做自己的事情。
notify线程：做完自己的事情后修改condition变量并唤醒等待线程。

1.2 wait线程中的用法

获得mutex。
进行wait过程，总的来说就是循环判断检查conditon，如果已经满足了就直接往下走自己的逻辑，否则以mutex作为参数执行wait, wait_for, wait_until三者之一的wait相关函数进行阻塞等待，mutex在wait函数阻塞当前线程前会被释放，具体机制参考第三部分。之所以需要循环检查condition是因为会有虚假唤醒的问题。

1.3 notify线程中的用法

获得mutex。
在这个mutex的保护下修改共享的条件变量，之所以在修改的时候保证获得锁是为了保证正确地唤醒，下面会讲到这个mutex的必要性。
调用notify_one或者notify_all唤醒wait在condition_variable类型变量上的线程，唤醒的时候不要求继续持有上述互斥锁，因为唤醒操作并不需要这个mutex的保护。

二、成员函数

condition_variable的成员函数有五种，分别是用于唤醒的notify_one和notify_all，以及用于wait的wait、wait_for和 wait_until。

2.1 唤醒函数

唤醒函数有两个，按需使用即可。
（1）notify_one

//唤醒一个wait在cv上的线程
void notify_one() noexcept;

（2）notify_all

//唤醒所有wait在cv上的线程
void notify_all() noexcept;

2.2 等待函数

（1）wait
调用condition_variable上的该函数会让当前线程阻塞，直到另外的线程调用了notify函数或者发生了虚假唤醒。具体到实现上主要就是把线程挂到等待队列里并释放外面获得的mutex，等待被唤醒，被唤醒后会重新获取前面释放的mutex。

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

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

（2）wait_for
和（1）基本一致，区别在于到了一定的超时时长即便没有notify也会被唤醒。

std::cv_status wait_for( std::unique_lock<std::mutex>& lock,
                         const std::chrono::duration<Rep, Period>& rel_time);

template< class Rep, class Period, class Predicate >
bool wait_for( std::unique_lock<std::mutex>& lock,
               const std::chrono::duration<Rep, Period>& rel_time,
               Predicate pred);

（3）wait_until
和（1）基本一致，区别在于到一定的绝对时间点即便没有notify也会被唤醒。

template< class Clock, class Duration >
std::cv_status wait_until( std::unique_lock<std::mutex>& lock,
                const std::chrono::time_point<Clock, Duration>& timeout_time );
                template< class Clock, class Duration, class Pred >

template< class Clock, class Duration, class Pred >
bool wait_until( std::unique_lock<std::mutex>& lock,
                 const std::chrono::time_point<Clock, Duration>& timeout_time,
                 Pred pred);

每种wait函数都有两个版本，分别是基本的版本和带Predicate的版本，Predicate在需要继续wait的时候返回false。对于前者，由于有虚假唤醒的存在，所以在用的过程需要结合一个对于condition的循环判断来使用，而后者则是内部帮我们做了这件事情，以wait为例，

cv.wait(lk, []{return i == 1;});

等价于

while (i != 1) {
    cv.wait(lock);
}

三、为什么需要互斥锁

首先我们需要明确一点，这个mutex不是用来保护condition_variable变量的，如果是保护condition_variable变量也用不着这么大费周章从外面传进去，这个锁是用来保证包括condition判断在内的整个wait过程是原子的，前面我们讲到了wait在使用过程中是先判断condition是否满足，如果不满足则会调用wait将线程阻塞放到等待队列里，这两个操作整体上并不是原子的，如果没有mutex的保护会出现这么一种情况：wait线程判断完condition不满足但在阻塞之前notify线程修改了condition并执行了notify，随后才被阻塞，这样wait线程就错过了这次唤醒，这也就是所谓的Lost wakeup问题，而额外的mutex则是避免了这个问题，因为在wait线程判断condition之前和唤醒线程修改condition之前都需要获取mutex，在wait线程的判断condition和wait之间condition无法被修改。实际上，condition_variable变量内部也有一个mutex，用于保护等待列表的修改，假设我们定义的mutex为m_a，cv内部的mutex为m_b,那么所以整个同步过程其实是

wait线程：

获取m_a，判断condition，发现不满足。
调用wait：
（1）获取m_b，获取后notify线程的notify也会被阻塞。
（2）释放m_a。
（3）线程挂到等待列表里。释放m_b后进行等待。
（4）被唤醒后重新获取m_a进行后续操作。

notify线程：

获取m_a。
修改condition。
调用notify ：
（1）获取m_b 。
（2）通知等待列表内的线程唤醒
释放m_a。

四、虚假唤醒

使用条件变量不能忽视的一种情况就是虚假唤醒（spurious wakeups），由于虚假唤醒的存在所以我们需要循环判读condition，根据字面意思，我们也能猜个八九不离十，简单来说就是不符合预期或者说异常的唤醒，在条件变量的使用场景下，虚假唤醒通常可以分为两种：

阻塞在cv上的线程在没有其他线程调用 notify或者notify_all的情况下解除了阻塞。
阻塞在cv上的线程确实是由于notify或者notify_all被调用而解除了阻塞，然而被唤醒后发现依赖的condition仍然是未满足的状态。

对于1，一种原因就是在阻塞的过程中收到了某种系统信号导致被唤醒，这里不展开讨论。对于2，这里贴一个典型的例子，假设有一个单生产者多消费者的队列由条件变量控制同步，线程1和线程2是消费者，线程3是生产者

线程1拿完了队列里最后一个item正在处理，此时队列为空。
线程2想去队列里拿发现已经空了所以wait在cv上。
线程3拿到mutex后往队列里发布了一个新item，修改了condition并执行了notify_one。
由于收到了notify线程2准备要被调度，但是线程1此时恰好处理完了手头的任务进行了下一轮循环抢在线程2之前拿到了mutex并取走了线程3刚放进去的item，此时线程2被阻塞，随后线程1释放了mutex。
线程2终于拿到了mutex却发现队列又是空的，这就是一次虚假唤醒，对于这种情况线程2需要继续wait。

五、示例

下面贴两个示例，分别用到了基础的wait和带Predicate的wait。

// condition_variable example
#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 print_id (int id) {
  std::unique_lock<std::mutex> lck(mtx);
  while (!ready) cv.wait(lck);
  // ...
  std::cout << "thread " << id << '\n';
}

void go() {
  std::unique_lock<std::mutex> lck(mtx);
  ready = true;
  cv.notify_all();
}

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

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

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

  return 0;
}

#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();
}

参考
std::condition_variable - cppreference.com
condition_variable - C++ Reference - Cplusplus.com
Why does pthread_cond_wait have spurious wakeups?
Calling pthread_cond_signal without locking mutex

码农公寓