条件变量中的唤醒丢失问题分析

本文是在其他作者博文的基础上进行了部分补充。原文：https://zhuanlan.zhihu.com/p/55123862

0. 前言

条件变量（condition variable）和互斥锁（mutex）为什么总是搭配着使用 ？互斥锁的作用是什么？非加互斥锁不可吗 ？

1. 错误写法

初始变量：

bool ready = false;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

Thread A 使用了互斥锁对操作条件变量的代码区域（临界区，包括2-4共3行代码）：

1: pthread_mutex_lock(&mutex);
2: while (false == ready) {
3:     pthread_cond_wait(&cond, &mutex);
4: }
5: pthread_mutex_unlock(&mutex);

Thread B 没有使用互斥锁对操作条件变量的代码区域（临界区，下面两行都属于临界区）进行保护。

1: ready = true;
2: pthread_cond_signal(&cond);

2. 怎么就错了

证明一个东西是错的，简单，给出一种错误的 case 就好，如下的执行序列 Thread A 就会丢失 Thread B 的条件变量的唤醒永久 wait，如下表格，在  false = ready  的时候，Thread A 进入 while 循环，但是还没有执行 wait 的时候，（因为没有thread B中没有使用互斥锁锁定临界区，造成了thread B可以随意执行代码；如有有互斥锁锁定了thread B的临界区，则因为thread A和B使用了同一个互斥锁mutex，则thread A在执行1-3的过程中，thread B的1-2无法执行，只能等待在那里，等到thread A中的代码进入到第3行，一旦进入第3行代码，则thread A会原子地释放占用的互斥锁mutex，然后thread B就可以占用mutex了，就可以执行1-2的代码了）thread B 执行了 ready = true 和 signal 唤醒，那么就出现条件变量唤醒 signal 先于 wait，那么相当于 Thread A 还没有被加入唤醒队列（thread A进入pthread_cond_wait()之后，才会加入到待唤醒队列中），这个时候，你已经 signal 唤醒了，那么这次唤醒自然就丢失了（后续如果还调用了唤醒，则能唤醒thread A了，如果不再调用pthread_cond_signal()则thread A就不会再被唤醒，只能永远休眠），执行序列的第 5 行，也就是 Thread 的第 3 行pthread_cond_wait(&cond, &mutex) 就会一直等在那里了。

3. 错在哪

核心出错的原因就是一个点，当判断 ready 为 false 进入 while 循环（Thread A 第 2 行），调用 pthread_cond_wait （Thread A 第 3 行）之前，Thread B 修改了 ready 为 true，并且 signal 了条件变量，导致 signal 先于 wait，出现丢失。这里根本的原因就是没有保证 ready == false 判断成立和 Thread A 调用 wait 进入唤醒队列的原子性，也就是我一旦判定 ready 为 false，那么必须进入等待队列，且在这期间不允许有人修改 ready 为 true，并执行 signal 操作。所以正确使用条件变量有两个约束：

• (1) 保证 ready == false 判断成立和 Thread A 调用 wait 进入唤醒队列的原子性
• (2) 设置 ready = true 在 signal 唤醒之前，也就是 Thread B 的 1、2 两行的顺序绝对不能反过来

上面错误的写法仅仅满足了约束 （2），但是却确忽略了（1）。一般来说约束（2）大部分人都能够意识到，因为错误 case 比较容易想到，这里为了不保持连贯性，错误 case 放在附录 1，后面再看。

上面错误的写法虽然给 Thread A 加锁了，但是这个锁加得有点糊里糊涂，没弄明白为什么要给 pthread_cond_wait(&cond, &mutex) 传递一个互斥量，以为是 pthread_cond_t 内部存在竞态条件，其实并不是，pthread_cond_wait(&cond, &mutex)调用之前会加锁，然后在内部将 thread 加入唤醒队列，然后才释放锁，其实就是为了保证约束（1），显然仅仅 Thread A 加锁，而 Thread B 设置 ready = true 没有加锁，并不能保证约束（1）的原子性。所以为了满足约束（1），需要给 Thread B 设置 ready = true 也加锁，正确的写法如下：

Thread A

1: pthread_mutex_lock(&mutex);
2: while (false == ready) {
3:     pthread_cond_wait(&cond, &mutex);
4: }
5: pthread_mutex_unlock(&mutex);

Thread B

1: pthread_mutex_lock(&mutex);
2: ready = true;
3: pthread_mutex_unlock(&mutex);
4: pthread_cond_signal(&cond);

这样就保证了原子性约束（1），那么无论线程如何运行，都只会有只有两种情况，情况 1：如果 Thread A 先拿到 mutex，那么此时 ready 为 false，Thread A 调用 pthread_cond_wait 进入等待队列，接着释放 mutex，然后 Thread B 才能修改 ready，并 signal；情况 2：如果 Thread A 没有拿到 mutex，Thread B 拿到 mutex，然后修改 ready 为 true，然后释放锁，这样 Thread A 在拿到 mutex，就不会再进 while 循环调 wait 了。

4. 这样也对

其实 Thread B 还可以这么写，也是正确的：

1: pthread_mutex_lock(&mutex);
2: ready = true;
3: pthread_cond_signal(&cond);
4: pthread_mutex_unlock(&mutex);

这样写也是正确的，但是性能会稍差写，考虑一种情况，Thread B 执行完第3行，但是第4行还未执行，那么 Thread A 将被唤醒，然后 Thread A 尝试去加锁，但是 Thread B 还没释放锁，所以 Thread A 会继续睡眠，然后 Thread B 再释放锁，会再次唤醒 Thread A，所以这种写法相比上面正确的写法可能会多一次线程上下文切换。实际上这里将 pthread_cond_signal(&cond) 加入到临界区中，保证了整个原子性，那么就不需要上面的约束（2）了，因此下面的写法也是正确的。

1: pthread_mutex_lock(&mutex);
2：pthread_cond_signal(&cond);
3: ready = true; 
4: pthread_mutex_unlock(&mutex);

 指出在具体实现中，上述情况可能会多的一次线程上下切换在 pthread 中已经被优化，所以不会存在这个问题。

5. 附录

Thread A

1: pthread_mutex_lock(&mutex);
2: while (false == ready) {
3:     pthread_cond_wait(&cond, &mutex);
4: }
5: pthread_mutex_unlock(&mutex);

Thread B

1: pthread_cond_signal(&cond);
2: ready = true;

这种错误写法实际上既没有满足约束（1），也没有满足约束（2），约束（1）上面讲过了，这里不在赘述，直接给违背约束（2）带来的 error case：

补充一点，pthread_cond_wait(&cond, &mutex);这句话很有意思，它本身包含了两个相反的操作：

• 在线程A开始执行这句话时（前提时前面已经加了互斥锁了，也就是已经持有互斥锁了），它准备进入休眠状态，准备释放CPU资源，但与此同时，它会在释放CPU的瞬间释放其占有的那个互斥锁，然后进入休眠状态，线程被放进待唤醒队列。其他线程此时可以抢占该互斥锁。
• 在线程A休眠时，也就是已经在待唤醒队列中等待时，某个线程释放了互斥锁，并唤醒了线程A，则线程A会准备离开pthread_cond_wait(&cond, &mutex);这句话，准备执行下一条语句，在离开的时刻，它会占有互斥锁，然后在下次释放之前，一直占有该互斥锁。

最后给出条件变量的一种实现，通过其实现，可以理解上面的内容：

struct cond{//条件变量是一个结构体，里面包含了一个等待在它上面的线程的链表
    struct thread * wait_list;
};

void cond_wait(struct cond * cond)
{
    list_append(cond->waite_list, thread_self());
    atomic_block_unlock(mutex);//原子挂起并释放锁
    lock(mutex);//重新获得互斥锁,当线程被其他线程唤醒时，从这一句开始执行
}

void cond_signal(struct cond *cond)
{
    if(!list_empty(cond->wait_list))
        wakeup(list_remove(cond->wait_list));
}

void cond_broadcast(struct cond * cond)
{
    while(!list_empty(cond->wait_list))
        wakeup(list_remove(cond->wait_list));
}

原文：https://zhuanlan.zhihu.com/p/55123862