std::condition_variable详解

1. 条件变量概述

   多线程访问一个共享资源(或称临界区),不仅需要用互斥锁实现独享访问避免并发错误,在获得互斥锁进入临界区后,有时还需检查特定条件是否成立。

   当某个线程修改测试条件后,将通知其它正在等待条件的线程继续往下执行。

   条件变量需要和一个互斥锁绑定,这个互斥锁的作用为:a. 互斥地访问临界资源。 b. 保护测试条件。

  1)wait线程从条件不满足,等待到重新执行过程,以 pthread_cond_wait 为例。

     

    a. (wait前必须先加锁)调用线程将自己放入等待队列,mutex解锁。(调用线程己加入等待队列并解锁,此时,允许其他线程改变“测试条件”)

   b. 挂起,等待pthread_cond_signal或pthread_cond_broadcast去唤醒。(其他线程改变测试条件,当条件满足时会发出通知)

   c. 被唤醒,mutex加锁

     关于条件变量的几个问题:

   (1) 为什么在pthread_cond_wait之前需要加锁?

          mutex是用来保护“测试条件”的,调用者将mutex传递给pthread_cond_wait,该函数内部会自动将调用线程放到等待队列中,然后再解锁mutex,

          并等待“测试条件”成立。这种做法关闭了从我们检测“测试条件”的时刻到将线程放入到等待队列之间的这段“时间窗口”,使得“测试条件”

          在线程加入等待队列之前不会被其他线程修改,从而确保调用线程不会错过“测试条件”的改变。最后,当pthread_cond_wait返回前,mutex又被上锁了。

    (2) 为什么使用while语句来循环判断“测试条件”而不使用if语句?

        线程API存在一个事实(很多语言中都如此,不仅仅是C++),就是即使在没有通知条件变量的情况下线程也可能被唤醒,这样的唤醒称为虚假唤醒

          (spurious wakeups),但此时“测试条件”往往并没有被满足。因此正确的做法是,通过while循环确认等待的“测试条件”是否确己发生并将其作

          为唤醒后的首个动作来处理,一旦确认是“虚假唤醒”则继续wait等待。而如果使用if语句,则唤醒后无法进行这种确认从而可能导致错误。

     (3)pthread_cond_signal 和 pthread_mutex_unlock顺序问题

         a. pthread_cond_signal放于pthread_mutex_unlock之前

            在上面对wait线程的解析中,我们可以看到,wait线程被唤醒后是会对mutex重新加锁的,但此时锁可能还没有被notify线程释放(会发生这

            种现象就是因为系统对线程的调度),会造成等待线程从内核中唤醒然后又回到内核空间(因为cond_wait返回后会有原子加锁的行为),

            所以一来一回会有性能的问题。但在Linux中推荐使用这种模式。

         b. pthread_cond_signal放于pthread_mutex_unlock之后

            不会出现之前说的那个潜在的性能损耗,因为在signal之前就已经释放锁了。但如果unlock和signal之前,有个低优先级的线程正在mutex上

            等待的话,那么这个低优先级的线程就会抢占高优先级的线程(cond_wait的线程)。

  2)notify线程:在wait线程阻塞期间,notify线程获取互斥锁并进入临界区内访问共享资源,然后改变测试条件,当条件满足时通知在条件变

      量上等待的wait线程。wait线程确认条件成立后重新申请对该互斥锁加锁,否则继续等待。

   条件变量类部分定义如下:

class condition_variable 
{
public:
    using native_handle_type = _Cnd_t;

    condition_variable() { _Cnd_init_in_situ(_Mycnd()); }                     // 默认构造函数
    ~condition_variable() noexcept { _Cnd_destroy_in_situ(_Mycnd()); }        // 析构函数

    condition_variable(const condition_variable&) = delete;
    condition_variable& operator=(const condition_variable&) = delete;        // 不可复制和移动

    void notify_one() noexcept { _Check_C_return(_Cnd_signal(_Mycnd())); }    // 唤醒一个等待线程
    void notify_all() noexcept { _Check_C_return(_Cnd_broadcast(_Mycnd())); } // 唤醒所有的等待线程

    void wait(unique_lock<mutex>& _Lck) {                                      // 等待,直到被唤醒
        _Check_C_return(_Cnd_wait(_Mycnd(), _Lck.mutex()->_Mymtx()));
    }
    template <class _Predicate>
    void wait(unique_lock<mutex>& _Lck, _Predicate _Pred) {                   // 等待信号并测试条件
        while (!_Pred()) {                                                    // 判断测试条件,只有当Pred不成立时才阻塞
            wait(_Lck);
        }
    }
};

    使用条件变量的wait线程基本流程:

     

2. mutex+condition_variable实现信号量

   1)P和V操作信号量是一个整数 count,提供两个原子(atom,不可分割)操作:P 操作和 V 操作,或是说 wait 和 signal 操作。

      a. P操作 (wait操作):count 减1,如果 count < 0 那么挂起执行线程。

--count;          //表示申请一个资源
if (count < 0)    //表示没有空闲资源
{
    调用进程进入等待队列Queue;
    阻塞进程;
}

    b. V操作 (signal操作):count 加1,如果 count <= 0 那么唤醒一个执行线程。

++count;          //表示释放一个资源
if (count <= 0)   //表示有进程处于阻塞状态
{
    从等待队列Queue中取出一个进程P;
    进程P进入就绪队列;
}

    来一个进程取一把锁(count减1),如果发现锁的数量小于0,即没有锁了? 于是只能进行(wait),直到有其它进程释放出一把锁为止。

      进程的事情办完后,要出去了,还回一把锁(count加1),如果发现 count <=0,即有进程在等,于是把自己的锁给它,唤醒一个等待的线程。

   2)代码实现如下

class semaphore 
{
public:
    semaphore(int value = 1): count(value) {}
    void P() {
        std::unique_lock<std::mutex> lock(mutex);
        if (--count < 0) condition.wait(lock);
    }
    void V() {
        std::lock_guard<std::mutex> lock(mutex);
        if(++count <= 0) condition.notify_one();
    }

private:
    int count;
    std::mutex mutex;
    std::condition_variable condition;
};

  

原文地址:https://www.cnblogs.com/yanghh/p/12995084.html