23.4 条件等待与通知

条件变量 (condition variable) 让一个线程阻塞等待，直到另一个线程通知它某个共享状态可能已经改变。条件变量总是与互斥量配合使用：共享状态由互斥量保护，条件变量只负责阻塞和唤醒。

下文示例同样基于 <threads.h>。

1. 典型模式

等待方必须在循环中检查条件，而不是只用一次 if。一方面，cnd_wait 可能在没有收到通知的情况下返回（虚假唤醒，spurious wakeup）；另一方面，即使收到了通知，在等待方重新获得互斥量之前，其他线程也可能已经再次改变了状态。

#include <stdio.h>
#include <threads.h>

static mtx_t lock;
static cnd_t cond;
static int ready = 0;

int producer(void *arg) {
    (void)arg;

    mtx_lock(&lock);
    ready = 1;
    cnd_signal(&cond);
    mtx_unlock(&lock);
    return 0;
}

int consumer(void *arg) {
    (void)arg;

    mtx_lock(&lock);
    while (!ready) {
        cnd_wait(&cond, &lock);
    }
    puts("consumer: ready");
    mtx_unlock(&lock);
    return 0;
}

int main(void) {
    thrd_t t1, t2;

    mtx_init(&lock, mtx_plain);
    cnd_init(&cond);
    thrd_create(&t1, consumer, NULL);
    thrd_create(&t2, producer, NULL);
    thrd_join(t1, NULL);
    thrd_join(t2, NULL);
    cnd_destroy(&cond);
    mtx_destroy(&lock);
    return 0;
}

输出为：

bash

consumer: ready

2. `signal` 与 `broadcast`

cnd_signal 唤醒一个等待线程，cnd_broadcast 唤醒全部等待线程。若条件满足后可能有多个线程都应继续执行，使用 broadcast 更合适。

3. `wait` 的锁语义

调用 cnd_wait 前，调用线程必须已经锁定互斥量。cnd_wait 原子地解锁互斥量并开始阻塞；线程被唤醒后，cnd_wait 先重新锁定互斥量，然后返回。由于“解锁”与“开始等待”是一个不可分割的操作，其他线程不可能恰好在两者之间修改状态并发出通知，因此通知不会被错过。返回时调用线程再次持有互斥量，可以直接检查受保护的状态，再决定继续等待还是退出循环。

4. 状态优先于通知

条件变量本身不保存任何状态，通知只表示“状态可能已改变”。判断能否继续执行的依据始终是受互斥量保护的状态对象。通知方应先在持有互斥量时修改状态，再发出通知；等待方被唤醒后应先检查状态，再决定继续等待还是退出循环。

5. 习题

5.1 巩固

#12321

⚡2⏳3

下面的程序先让生产者运行结束，再创建消费者。消费者不看条件，直接调用 cnd_wait：

#include <stdio.h>
#include <threads.h>

static mtx_t lock;
static cnd_t cond;
static int ready;

static int producer(void *arg) {
    (void)arg;

    mtx_lock(&lock);
    ready = 1;
    cnd_signal(&cond);
    mtx_unlock(&lock);
    return 0;
}

static int consumer(void *arg) {
    (void)arg;

    mtx_lock(&lock);
    cnd_wait(&cond, &lock);
    printf("consumer: ready = %d\n", ready);
    mtx_unlock(&lock);
    return 0;
}

int main(void) {
    thrd_t p, c;

    mtx_init(&lock, mtx_plain);
    cnd_init(&cond);
    thrd_create(&p, producer, NULL);
    thrd_join(p, NULL);
    thrd_create(&c, consumer, NULL);
    thrd_join(c, NULL);
    puts("main: done");
    return 0;
}

要求：

先写出预测，再编译运行。用 timeout 5 ./a.out 启动，记录退出码，124 表示程序没有退出。ready 在等待之前已经是 1，消费者为什么仍然不返回？
把创建生产者的 thrd_create 和 thrd_join 移到创建消费者之后，并在两者之间插入 thrd_sleep 睡 100 ms，让消费者先进入等待。其余不动，连跑 5 次，记录退出码，说明与第 1 步的差别。
把消费者的等待改成 while (!ready) { cnd_wait(&cond, &lock); }，两种启动顺序各跑 5 次。验收：退出码都是 0，输出 consumer: ready = 1。
再把循环换成 if (!ready) { cnd_wait(&cond, &lock); }。本程序中它也能通过。另写一个线程，用 thrd_sleep 安排在消费者已经进入等待、生产者尚未把 ready 置 1 的时刻调用 cnd_broadcast，但不修改 ready。用 if 时消费者会打印什么？换回 while 后，这个线程还会造成错误吗？

验收：第 1 步进程不退出（124），consumer 一行也不打印；第 2 步 5 次都正常结束；第 3 步两种顺序都正常；第 4 步用 if 时输出 consumer: ready = 0，用 while 时消费者重新睡下，最后等到 ready = 1 才打印。

5.2 编程

#12322

⚡3⏳3

一个“起跑门”：4 个工作线程各自等待主线程发出开始信号，然后打印一行。主线程把 go 置 1 后唤醒等待者。下面只给出共享对象和线程函数，main 自己补全：

static mtx_t lock;
static cnd_t cond;
static int go;

static int worker(void *arg) {
    int id = *(int *)arg;

    mtx_lock(&lock);
    while (!go) {
        cnd_wait(&cond, &lock);
    }
    printf("worker %d: go\n", id);
    mtx_unlock(&lock);
    return 0;
}

要求：

主线程用一次 cnd_signal 发出通知，用 timeout 5 ./a.out 连跑 5 次，记录退出码和每次打印的行数。
换成 cnd_broadcast，再跑 5 次。验收：每次打印 4 行，退出码 0。
把 main 改成创建两个等待者，go = 1 之后只调用一次 cnd_signal。连跑 5 次，记录退出码；再把这一次调用换成 cnd_broadcast。解释为什么这里一次 signal 会让程序停住，而 broadcast 不会。
共享状态换成“任务计数”：int pending 表示尚未被取走的任务数，每个线程最多取一个任务，取到就退出，没有任务时继续等待。另加 int closed，主线程在发完任务后置 1 并广播，让没取到任务的线程也能退出。主线程放入两个任务后连续调用两次 cnd_signal。运行并记录打印了几行、pending 最后是多少。说明这种“一个通知对应一个任务”的场合为什么用 signal；把这两次调用换成 broadcast，结果会有什么变化。

验收：第 1 步记录到的行数不超过 1 行且退出码为 124。被 timeout 杀掉的进程可能来不及刷新标准输出的缓冲区，一行也看不到是正常的，判断依据是退出码。第 2 步每次 4 行、退出码 0。第 3 步 signal 版 124，broadcast 版正常。第 4 步两次 signal 之后恰好打印两行取任务的输出，pending 为 0，进程正常退出；换成 broadcast 后取到的任务数和 pending 都相同，被唤醒的线程多出两个，它们检查条件后发现没有任务可取，随即退出。

#12323

⚡4⏳4

实现一个带计数的事件对象，多次通知可以累积：

typedef struct {
    mtx_t lock;
    cnd_t cond;
    unsigned long count;
} event_t;

void event_init(event_t *e);
void event_post(event_t *e);
int event_wait(event_t *e, unsigned long target, int timeout_ms);

event_wait 在 count >= target 时立即返回 1，超时返回 0。要求：

用 cnd_timedwait 实现超时：
c
```
int cnd_timedwait(cnd_t *cond, mtx_t *mutex, const struct timespec *deadline);
```
1
它接收的是 TIME_UTC 的绝对时刻，不是等待的时长；到时返回 thrd_timedout，正常唤醒返回 thrd_success。写一个辅助函数把“从现在起 timeout_ms 毫秒”换算成 struct timespec，处理 tv_nsec 达到 10^9 时的进位。等待条件仍然放在 while 里。
先连续调用两次 event_post，再依次调用 event_wait(e, 1, 200)、event_wait(e, 2, 200)、event_wait(e, 3, 200)。前两次返回 1，第三次约 200 ms 后返回 0。记录第三次的耗时。
把 count 换成 int flag：event_post 只把 flag 置 1，等待者在返回前把它清 0。重复第 2 步的三次等待，但每次单独用一个 printf 打印返回值，输出变成 1、0、0。说明在“等待开始之前已经发生两次通知”的情况下，计数与标志为什么给出不同结果。若把三次调用写进同一个 printf 的实参列表，观察结果是否还稳定，并解释原因。
用另一个探针线程观察互斥量的归属。探针线程在等待者进入 cnd_timedwait 之后调用一次 mtx_trylock(&e->lock)，记录返回值；再让等待者被唤醒后先置一个 atomic_int 标志、睡 200 ms 才 mtx_unlock，探针等到这个标志后立刻再调用一次 mtx_trylock。mtx_trylock 尝试加锁但不阻塞，拿到锁返回 thrd_success，锁已被占用返回 thrd_busy。两次结果分别说明等待期间与被唤醒之后，互斥量在谁手里。
让 event_post 在发出通知之后、mtx_unlock 之前睡 200 ms，测量等待者从通知发出到 event_wait 返回的时间；再把通知移到 mtx_unlock 之后重测。用两次结果说明 cnd_wait 返回之前做了什么。

验收：第 2 步输出 1、1、0，第三次约 200 ms；第 3 步输出 1、0、0；第 4 步等待期间探针得到 thrd_success（互斥量已被 cnd_timedwait 释放），被唤醒后的 200 ms 内得到 thrd_busy（互斥量已由等待者重新持有）；第 5 步前一次约 200 ms，后一次接近 0。

5.3 综合与挑战

#12324

⚡5⏳4

实现一个可重复使用的屏障：4 个线程每轮都在屏障处对齐，全部到齐后才一起进入下一轮，连续使用 200 轮。

typedef struct {
    mtx_t lock;
    cnd_t cond;
    int arrived;
    int generation;
    int aborted;
} barrier_t;

void barrier_init(barrier_t *b, int n);
void barrier_wait(barrier_t *b);

要求：

实现 barrier_init 与 barrier_wait：n 个线程中最后一个到达的负责放行，其余线程等待；每一轮结束后状态要能接着用于下一轮。barrier_wait 正常放行返回 0，被中止返回 -1。
写验收程序：第 r 轮开始时，每个线程先执行 atomic_store(&stamp[id], r)，再调用 barrier_wait，然后读取全部 atomic_int stamp[4]。屏障保证此时每个 stamp[j] 只能是 r 或 r+1，其他值记为一次 mismatch。4 个线程各跑 200 轮，连续运行 20 次，输出应恒为 mismatch = 0。
变体一：去掉 generation，等待条件改成 while (arrived != 0)，最后一个线程把 arrived 清零后广播。运行会卡住。记录 timeout 的退出码，并说明卡在哪里：醒来较晚的线程还没来得及退出循环，下一轮已经有人把 arrived 加回了非零，它于是又睡回去，而这一轮再没有人清零。
变体二：保留 generation，把 cnd_broadcast 换成 cnd_signal。同样会卡住。说明屏障为什么必须唤醒全部等待者，而第 12322 题里“一个通知对应一个任务”的场合只需要唤醒一个。
给 barrier_wait 增加超时：等待者用 cnd_timedwait 最多等 300 ms，超时后把 aborted 置 1 并广播，所有等待者返回 -1；此后到达的线程看到 aborted 也立即返回 -1。用 4 个线程中只让 3 个调用 barrier_wait 来验证：3 个线程各打印 rc = -1，最后输出 arrived = 3, aborted = 1。

验收：第 2 步 20 次运行 mismatch = 0；第 3、4 步的变体都用 timeout 运行并得到退出码 124；第 5 步三个等待者都返回 -1，程序正常退出。

3.10 求值

4.3 选择语句

5.5 跳转语句

10.2 结构体

14.3 程序支持

15.1 声明

16.2 对象表示

18.2 标准库中的宏

23.4 条件等待与通知 ​

1. 典型模式 ​

2. signal 与 broadcast ​

3. wait 的锁语义 ​

4. 状态优先于通知 ​

5. 习题 ​

5.1 巩固 ​

5.2 编程 ​

5.3 综合与挑战 ​