C++11并发——多线程条件变量std::condition_variable（四）

std::condition_variable 是条件变量，

当 std::condition_variable 对象的某个 wait 函数被调用的时候，它使用 std::unique_lock(通过 std::mutex) 来锁住当前线程。

当前线程会一直被阻塞，直到另外一个线程在相同的 std::condition_variable 对象上调用了 notification 函数来唤醒当前线程。

std::condition_variable 对象通常使用 std::unique_lock<std::mutex> 来等待，如果需要使用另外的 lockable 类型，可以使用 std::condition_variable_any 类，本文后面会讲到 std::condition_variable_any 的用法。

#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 << '
';
}

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...
";
    go(); // go!

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

    return 0;
}

concurrency ) ./ConditionVariable-basic1 
threads ready to race...
thread 1
thread 0
thread 2
thread 3
thread 4
thread 5
thread 6
thread 7
thread 8
thread 9

好了，对条件变量有了一个基本的了解之后，我们来看看 std::condition_variable 的各个成员函数。

std::condition_variable 构造函数

default (1)	condition_variable();
copy [deleted] (2)	condition_variable (const condition_variable&) = delete;

std::condition_variable 的拷贝构造函数被禁用，只提供了默认构造函数。

std::condition_variable::wait() 介绍

unconditional (1)	void wait (unique_lock<mutex>& lck);
predicate (2)	template <class Predicate> void wait (unique_lock<mutex>& lck, Predicate pred);

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

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

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

while (!pred()) wait(lck);

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

std::mutex mtx;
std::condition_variable cv;

int cargo = 0;
bool shipment_available()
{
    return cargo != 0;
}

// 消费者线程.
void consume(int n)
{
    for (int i = 0; i < n; ++i) {
        std::unique_lock <std::mutex> lck(mtx);
        cv.wait(lck, shipment_available);
        std::cout << cargo << '
';
        cargo = 0;
    }
}

int main()
{
    std::thread consumer_thread(consume, 10); // 消费者线程.

    // 主线程为生产者线程, 生产 10 个物品.
    for (int i = 0; i < 10; ++i) {
        while (shipment_available())
            std::this_thread::yield();
/*
std::this_thread::yield：　当前线程放弃执行，操作系统调度另一线程继续执行。
即当前线程将未使用完的“CPU时间片”让给其他线程使用，
等其他线程使用完后再与其他线程一起竞争"CPU"。
std::this_thread::sleep_for：　表示当前线程休眠一段时间，
休眠期间不与其他线程竞争CPU，根据线程需求，等待若干时间。

*/
        std::unique_lock <std::mutex> lck(mtx);
        cargo = i + 1;
        cv.notify_one();
    }

    consumer_thread.join();

    return 0;
}

1. std::condition_variable

条件变量提供了两类操作：wait和notify。这两类操作构成了多线程同步的基础。

1.1 wait

wait是线程的等待动作，直到其它线程将其唤醒后，才会继续往下执行。下面通过伪代码来说明其用法：

std::mutex mutex; std::condition_variable cv; 
// 条件变量与临界区有关，用来获取和释放一个锁，因此通常会和mutex联用。
 std::unique_lock lock(mutex); 
// 此处会释放lock，然后在cv上等待，直到其它线程通过cv.notify_xxx来唤醒当前线程，
//cv被唤醒后会再次对lock进行上锁，然后wait函数才会返回。 
// wait返回后可以安全的使用mutex保护的临界区内的数据。此时mutex仍为上锁状态 cv.wait(lock)

除wait外, 条件变量还提供了wait_for和wait_until，这两个名称是不是看着有点儿眼熟，std::mutex也提供了_for和_until操作。在C++11多线程编程中，需要等待一段时间的操作，

一般情况下都会有xxx_for和xxx_until版本。前者用于等待指定时长，后者用于等待到指定的时间。

1.2 notify

了解了wait，notify就简单多了：唤醒wait在该条件变量上的线程。notify有两个版本：notify_one和notify_all。

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

其使用方式见下例：

std::mutex mutex;
 std::condition_variable cv; 
std::unique_lock lock(mutex); 
// 所有等待在cv变量上的线程都会被唤醒。但直到lock释放了mutex，被唤醒的线程才会从wait返回。 
cv.notify_all(lock)

// conditionVariable.cpp

#include <iostream>
#include <condition_variable>
#include <mutex>
#include <thread>

std::mutex mutex_;
std::condition_variable condVar;

void doTheWork(){
  std::cout << "Processing shared data." << std::endl;
}

void waitingForWork(){
    std::cout << "Worker: Waiting for work." << std::endl;

    std::unique_lock<std::mutex> lck(mutex_);
    condVar.wait(lck);
    doTheWork();
    std::cout << "Work done." << std::endl;
}

void setDataReady(){
    std::cout << "Sender: Data is ready."  << std::endl;
    condVar.notify_one();
}

int main(){

  std::cout << std::endl;

  std::thread t1(waitingForWork);
  std::thread t2(setDataReady);

  t1.join();
  t2.join();

  std::cout << std::endl;

}

该程序有两个子线程： t1和t2。它们在第33行和第34行中获得可调用的有效负载（函数或函子） waitingForWork和setDataReady。

函数setDataReady通过使用条件变量condVar调用condVar.notify_one()进行通知。在持有锁的同时，线程T2正在等待其通知: condVar.wait(lck).

虚假的唤醒

细节决定成败。事实上，可能发生的是，接收方在发送方发出通知之前完成了任务。这怎么可能呢？接收方对虚假的唤醒很敏感。所以即使没有通知发生，接收方也有可能会醒来。

为了保护它，我不得不向等待方法添加一个判断。这就是我在下一个例子中所做的：

// conditionVariableFixed.cpp

#include <iostream>
#include <condition_variable>
#include <mutex>
#include <thread>

std::mutex mutex_;
std::condition_variable condVar;

bool dataReady;

void doTheWork(){
  std::cout << "Processing shared data." << std::endl;
}

void waitingForWork(){
    std::cout << "Worker: Waiting for work." << std::endl;

    std::unique_lock<std::mutex> lck(mutex_);
    condVar.wait(lck,[]{return dataReady;});
    doTheWork();
    std::cout << "Work done." << std::endl;
}

void setDataReady(){
    std::lock_guard<std::mutex> lck(mutex_);
    dataReady=true;
    std::cout << "Sender: Data is ready."  << std::endl;
    condVar.notify_one();
}

int main(){

  std::cout << std::endl;

  std::thread t1(waitingForWork);
  std::thread t2(setDataReady);

  t1.join();
  t2.join();

  std::cout << std::endl;

}

View Code

与第一个示例的关键区别是在第11行中使用了一个布尔变量dataReady 作为附加条件。 dataReady在第28行中被设置为true。

它在函数waitingForWork中被检查：

condVar.wait(lck,[]{return dataReady;})

这就是为什么wait方法有一个额外的重载，它接受一个判定。判定是个callable，它返回true或false。
在此示例中，callable是lambda函数。因此，条件变量检查两个条件：判定是否为真，通知是否发生。

关于dataReady
dataReady是个共享变量，将会被改变。所以我不得不用锁来保护它。
因为线程T1只设置和释放锁一次，所以std::lock_guard已经够用了。但是线程t2就不行了，wait方法将持续锁定和解锁互斥体。所以我需要更强大的锁：std::unique_lock。
但这还不是全部，条件变量有很多挑战，它们必须用锁来保护，并且易受虚假唤醒的影响。
大多数用例都很容易用tasks来解决，后续再说task问题。

唤醒不了

条件变量的异常行为还是有的。大约每10次执行一次conditionVariable.cpp就会发生一些奇怪的现象：

我不知道怎么回事，这种现象完全违背了我对条件变量的直觉。
在安东尼·威廉姆斯的支持下，我解开了谜团。
问题在于，如果发送方在接收方进入等待状态之前发送通知，则通知会丢失。C ++标准同时也将条件变量描述为同步机制，“condition_variable类是一个同步原语，可以用来同时阻塞一个线程或多个线程。。。”。
因此，通知消息已经丢失了，但是接收方还在等啊和等啊等啊等啊…
怎么解决这个问题呢？去除掉wait第二个参数的判定可以有效帮助唤醒。实际上，在判定设置为真的情况下，接收器也能够独立于发送者的通知进而继续其工作。