C11线程管理：条件变量

Posted 2020-10-16 渔舟唱晚

1、简介

　　C11提供另外一种用于等待的同步机制，它可以阻塞一个或者多个线程，直到收到另外一个线程发出的通知或者超时，才会唤醒当前阻塞的线程。条件变量要和互斥量配合起来使用。

　　condition_variable，配合std::unique_lock<std::mutex>进行wait操作。
　　condition_variable_any，和任意带有lock、unlock语意 的mutex搭配使用，比较灵活，但是效率比condition_variable低。

　　条件变量的使用过程如下：

　　a.拥有条件变量的线程获取互斥量。
　　b.循环检查某个条件，如果条件不满足，则阻塞线程直到满足；如果条件满足，则向下执行。
　　c.某个线程满足条件并执行完成之后，调用notify_one或者notify_all来唤醒一个或者多个线程。

2、实践

　　可以用条件变量来实现一个同步队列，同步队列作为一个线程安全的数据共享区，经常用于线程之间的读取，比如半同步半异步线程池的同步队列。

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

template<typename T>
class SyncQueue
{
public:
    SyncQueue(int maxSize) :m_maxSize(maxSize){}

    void Put(const T & t)
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        while (IsFull())
        {
            std::cout << "缓冲区满了，需要等待..." << std::endl;
            m_notFull.wait(m_mutex);
        }

        m_queue.push_back(t);
        m_notEmpty.notify_one();
    }

    void Take(const T & t)
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        while (IsEmpty())
        {
            std::cout << "缓冲区空了，需要等待..." << std::endl;
            m_notEmpty.wait(m_mutex);
        }
    
        t = m_queue.front();
        m_queue.pop_front(t);
        m_notFull.notify_one();
    }

    bool Empty()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.empty();
    }

    bool Full()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.size() == m_maxSize;
    }

    size_t Size()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.size();
    }

private:
    bool IsFull() const
    {
        return m_queue.size() == m_maxSize;
    }

    bool IsEmpty() const
    {
        return m_queue.empty();
    }

private:
    std::list<T> m_queue;  　　　　　　　　　　　　 //缓冲区
    std::mutex m_mutex;  　　　　　　　　　　　　   //互斥量
    std::condition_variable_any m_notEmpty;     //不为空的条件变量
    std::condition_variable_any m_notFull;      //没有满的条件变量
    int m_maxSize;            　　　　　　　　　　 //同步队列最大容量
};

　　这个队列中，没有满的情况下可以插入数据，如果满了，则会调用m_notFull阻塞线程等待，等待消费线程取出数据之后发出一个未满的通知，然后前面阻塞的线程会被唤醒继续往下执行；如果队列为空，不能取出数据，调用m_notEmpty来阻塞当前线程，等待插入数据的线程插入数据发出不为空的通知，唤醒被阻塞的线程，往下执行读出数据。

　　条件变量的wait方法还有个重载方法，可以接受一个条件。

std::lock_guard<std::mutex> locker(m_mutex);
while (IsFull())
{
    std::cout << "缓冲区满了，需要等待..." << std::endl;
    m_notFull.wait(m_mutex);
}

　　可以写为这样：

std::lock_guard<std::mutex> locker(m_mutex);
m_notFull.wait(locker, [this]{ return !IsFull();});

　　两种写法都一样，后者代码更加简洁，条件变量先检查判断式是否满足条件，如果满足，重新获取mutex，结束wait，继续往下执行；如果不满足条件，则释放mutex，将线程置为waiting状态，继续等待。

　　需要注意的是，wait函数会释放掉mutex，而lock_guard还拥有mutex，他只在出了作用域之后才会释放掉mutex，所以这时并不会释放，但是执行wait会提前释放，而在wait提前释放掉锁之后，会处于等待状态，在notify_one/all唤醒之后，会先获取mutex，相当于之前的mutex又获取到了，所以在出作用域的时候，lock_guard释放锁不会产生问题。

　　在这种情况下，如果用unique_lock语意更准确，因为unique_lock不像lock_guard一样只能在析构的时候才能释放锁，它可以随时释放锁，在wait的时候让uniq_lock释放锁，语意更加准确。

　　上述例子中，可以用unique_lock来替换掉lock_guard，condition_variable来替换掉condition_variable_any，会使代码更加清晰，效率也更高。

3、超时等待

　　除了wait还可以使用超时等待函数std::condition_variable::wait_for和std::condition_variable::wait_until。

　　与 std::condition_variable::wait() 类似，不过 wait_for 可以指定一个时间段，在当前线程收到通知或者指定的时间 rel_time 超时之前，该线程都会处于阻塞状态。而一旦超时或者收到了其他线程的通知，wait_for 返回，剩下的处理步骤和 wait() 类似。

　　与 std::condition_variable::wait_for 类似，但是 wait_until 可以指定一个时间点，在当前线程收到通知或者指定的时间点 abs_time 超时之前，该线程都会处于阻塞状态。而一旦超时或者收到了其他线程的通知，wait_until 返回，剩下的处理步骤和 wait_for() 类似。

// condition_variable::wait_for example
#include <iostream>           // std::cout
#include <thread>             // std::thread
#include <chrono>             // std::chrono::seconds
#include <mutex>              // std::mutex, std::unique_lock
#include <condition_variable> // std::condition_variable, std::cv_status

std::condition_variable cv;

int value;

void read_value() {
  std::cin >> value;
  cv.notify_one();
}

int main ()
{
  std::cout << "Please, enter an integer (I‘ll be printing dots): \n";
  std::thread th(read_value);

  std::mutex mtx;
  std::unique_lock<std::mutex> lck(mtx);
  while (cv.wait_for(lck,std::chrono::seconds(1))==std::cv_status::timeout) {
    std::cout << ‘.‘ << std::endl;
  }
  std::cout << "You entered: " << value << ‘\n‘;

  //等待th线程执行完
  th.join();

  return 0;
}

　　如果用wait_until，只需要将条件改为时间点即可：

while (cv.wait_for(lck,std::chrono::seconds(1))==std::cv_status::timeout)
while (cv.wait_until(lck, std::chrono::system_clock::now() +std::chrono::seconds(1)) == std::cv_status::timeout)