9.2 条件变量、信号量与 Barrier

地心探测台上的线程已经学会用 mutex 保护同一块仪表，但“轮到谁动手”仍没有答案。数据采集线程要等队列非空，四名分析线程要等所有人完成当前阶段，稀缺设备又只能同时借给两个 worker。只会互斥，结果往往是大家抱着锁轮流询问条件，CPU 白白发热。

9.1 线程、数据竞争与互斥保护 shared invariant。这一课处理三种协调关系：等待 predicate、限制并发资源数量，以及让一组 thread 在阶段边界汇合。故事里的“通知铃”不会保存业务事实；正式模型中的状态始终放在受保护的 predicate 里。

condition variable 不保存“发生过一次通知”

consumer 需要等 queue 非空。若它反复解锁、检查、sleep，会浪费 CPU；若先检查再单独进入 sleep，producer 可能恰好在两步之间发出通知，consumer 从此睡过头。

condition variable 把“释放 mutex 并进入等待”做成一个原子协议：

text

lock mutex
while predicate is false:
    cond_wait(condition, mutex)
consume / update state
unlock mutex

pthread_cond_wait 成功返回时已经重新持有 mutex。调用前也必须持锁，否则 predicate check、wait 和 producer update 不在同一个同步协议里。

通知本身不排队成业务事件。signal 表示“某个 waiter 应重新检查状态”，broadcast 表示“所有 waiter 都应重新检查”；若当时没有 waiter，通知可以没有任何后续记忆。真正的事实保存在受 mutex 保护的 predicate 中。

为什么一定用 `while`

wait 返回后，predicate 仍可能为假：

POSIX 允许 spurious wakeup；
多个 consumer 被唤醒后，另一个 thread 可能先取得 mutex 并取走数据；
broadcast 本来就会唤醒不一定都能继续的 waiter； -程序未来扩展出更多修改 predicate 的路径。

所以 while 不是单纯防御某个罕见 OS bug，而是 Mesa-style condition variable 的语义。thread 被唤醒只获得重新竞争 mutex 的资格，不获得“条件仍然成立”的预约券。

一个单槽 channel 的完整协议

下面的 producer 写入 1–5，consumer 求和。occupied 与 closed 都由同一 mutex 保护；condition variable 只负责 sleep/wake。

#define _POSIX_C_SOURCE 200809L
#include <stdbool.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

struct Channel {
    pthread_mutex_t mutex;
    pthread_cond_t changed;
    bool occupied;
    bool closed;
    int value;
};

struct ConsumerArgument {
    struct Channel *channel;
    int sum;
};

static void check_pthread(const char *operation, int error) {
    if (error != 0) {
        fprintf(stderr, "%s: %s\n", operation, strerror(error));
        abort();
    }
}

static void *produce(void *raw_channel) {
    struct Channel *channel = raw_channel;
    for (int value = 1; value <= 5; value++) {
        check_pthread("lock", pthread_mutex_lock(&channel->mutex));
        while (channel->occupied) {
            check_pthread("wait",
                          pthread_cond_wait(&channel->changed,
                                            &channel->mutex));
        }
        channel->value = value;
        channel->occupied = true;
        check_pthread("broadcast", pthread_cond_broadcast(&channel->changed));
        check_pthread("unlock", pthread_mutex_unlock(&channel->mutex));
    }

    check_pthread("lock", pthread_mutex_lock(&channel->mutex));
    while (channel->occupied) {
        check_pthread("wait",
                      pthread_cond_wait(&channel->changed, &channel->mutex));
    }
    channel->closed = true;
    check_pthread("broadcast", pthread_cond_broadcast(&channel->changed));
    check_pthread("unlock", pthread_mutex_unlock(&channel->mutex));
    return NULL;
}

static void *consume(void *raw_argument) {
    struct ConsumerArgument *argument = raw_argument;
    struct Channel *channel = argument->channel;

    check_pthread("lock", pthread_mutex_lock(&channel->mutex));
    for (;;) {
        while (!channel->occupied && !channel->closed) {
            check_pthread("wait",
                          pthread_cond_wait(&channel->changed,
                                            &channel->mutex));
        }
        if (!channel->occupied && channel->closed) {
            break;
        }

        argument->sum += channel->value;
        channel->occupied = false;
        check_pthread("broadcast", pthread_cond_broadcast(&channel->changed));
    }
    check_pthread("unlock", pthread_mutex_unlock(&channel->mutex));
    return NULL;
}

int main(void) {
    struct Channel channel = {
        .mutex = PTHREAD_MUTEX_INITIALIZER,
        .changed = PTHREAD_COND_INITIALIZER,
        .occupied = false,
        .closed = false,
        .value = 0,
    };
    struct ConsumerArgument consumer_argument = {
        .channel = &channel,
        .sum = 0,
    };
    pthread_t producer;
    pthread_t consumer;

    check_pthread("create consumer",
                  pthread_create(&consumer, NULL, consume,
                                 &consumer_argument));
    check_pthread("create producer",
                  pthread_create(&producer, NULL, produce, &channel));
    check_pthread("join producer", pthread_join(producer, NULL));
    check_pthread("join consumer", pthread_join(consumer, NULL));

    printf("sum=%d\n", consumer_argument.sum);
    check_pthread("destroy condition", pthread_cond_destroy(&channel.changed));
    check_pthread("destroy mutex", pthread_mutex_destroy(&channel.mutex));
    return consumer_argument.sum == 15 ? 0 : 1;
}

这里用 broadcast 简化单槽协议；只有一个对应 waiter 时，signal 也够用。扩展到 bounded queue 后，通常分别使用 not_empty 与 not_full，减少无关唤醒。优化前先保证每次 state transition 都能唤醒所有可能变为可运行的角色。

timeout 要绑定合适的 clock

pthread_cond_timedwait 使用 condition attribute 选择的 clock。若采用 wall clock，管理员校时可能让 deadline 跳动；支持时可把 condition variable 配成 CLOCK_MONOTONIC。

timeout 返回后仍要在 mutex 内检查 predicate。deadline 到达与 producer 修改状态可能同时发生，API 返回 timeout 不代表状态必然仍为假。

相对 timeout 还要防止 spurious wakeup 每次重新计算完整等待时间，导致总时长无限延后。先计算一次 absolute deadline，再循环等待。

semaphore 保存的是 permit 数量

counting semaphore 有一个非负 permit count：

sem_wait 取得一个 permit；没有时阻塞；
sem_post 归还一个 permit，并可能唤醒 waiter。

它适合限制同时使用有限资源的 thread 数，例如只有 8 个数据库 connection。与 mutex 不同，semaphore 通常没有“必须由同一 owner 释放”的语义；错误地多 post 会凭空增加容量。

#define _POSIX_C_SOURCE 200809L
#include <errno.h>
#include <semaphore.h>

static int acquire_permit(sem_t *semaphore) {
    while (sem_wait(semaphore) != 0) {
        if (errno != EINTR) {
            return -1;
        }
    }
    return 0;
}

static int use_limited_resource(sem_t *semaphore) {
    if (acquire_permit(semaphore) != 0) {
        return -1;
    }

    int result = perform_operation();
    int saved_errno = errno;
    if (sem_post(semaphore) != 0 && result == 0) {
        result = -1;
        saved_errno = errno;
    }
    errno = saved_errno;
    return result;
}

perform_operation 代表业务函数。代码强调 permit 在所有退出路径归还；若 thread cancellation 可能发生，还需要 cleanup handler。

POSIX unnamed semaphore 并非每个 Unix 平台都支持 process-shared 用法，named semaphore 的生命周期也不同。跨平台库应查目标实现，不要把 Linux 行为当成 POSIX 全集。

binary semaphore 虽然 count 只有 0/1，却仍不自动拥有 mutex 的 ownership、priority inheritance 或 robust-recovery 语义。保护临界区优先用 mutex，表达资源数量才用 semaphore。

barrier 让整支探测队在阶段边界会合

四个 worker 可以各自处理一片数据，但下一轮汇总必须等所有人完成当前阶段。Barrier 保存的是这一代 participant 的到达数；它不表示某件稀缺设备有几个 permit。

barrier 的 predicate 是“本 generation 已到达 N 个 participant”。最后一个到达者推进 generation 并唤醒其他 thread。若 barrier 要重复使用，只有 count 没有 generation 会发生 ABA 式混淆：上一轮的迟到 wakeup 可能误读下一轮 count。

#define _POSIX_C_SOURCE 200809L
#include <assert.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

enum { THREADS = 4, ROUNDS = 20 };

struct Barrier {
    pthread_mutex_t mutex;
    pthread_cond_t changed;
    unsigned int participants;
    unsigned int arrived;
    unsigned int generation;
};

struct WorkerArgument {
    struct Barrier *barrier;
};

static _Atomic int arrivals[ROUNDS];

static void check(int error, const char *operation) {
    if (error != 0) {
        fprintf(stderr, "%s: %s\n", operation, strerror(error));
        abort();
    }
}

static void barrier_wait(struct Barrier *barrier) {
    check(pthread_mutex_lock(&barrier->mutex), "lock");
    unsigned int generation = barrier->generation;

    barrier->arrived++;
    if (barrier->arrived == barrier->participants) {
        barrier->arrived = 0;
        barrier->generation++;
        check(pthread_cond_broadcast(&barrier->changed), "broadcast");
    } else {
        while (generation == barrier->generation) {
            check(pthread_cond_wait(&barrier->changed, &barrier->mutex),
                  "wait");
        }
    }
    check(pthread_mutex_unlock(&barrier->mutex), "unlock");
}

static void *run_rounds(void *raw_argument) {
    struct WorkerArgument *argument = raw_argument;
    for (int round = 0; round < ROUNDS; round++) {
        atomic_fetch_add_explicit(&arrivals[round], 1,
                                  memory_order_relaxed);
        barrier_wait(argument->barrier);
        assert(atomic_load_explicit(&arrivals[round],
                                    memory_order_relaxed) == THREADS);
    }
    return NULL;
}

int main(void) {
    struct Barrier barrier = {
        .mutex = PTHREAD_MUTEX_INITIALIZER,
        .changed = PTHREAD_COND_INITIALIZER,
        .participants = THREADS,
        .arrived = 0,
        .generation = 0,
    };
    struct WorkerArgument argument = {.barrier = &barrier};
    pthread_t threads[THREADS];

    for (int index = 0; index < THREADS; index++) {
        check(pthread_create(&threads[index], NULL, run_rounds, &argument),
              "pthread_create");
    }
    for (int index = 0; index < THREADS; index++) {
        check(pthread_join(threads[index], NULL), "pthread_join");
    }

    check(pthread_cond_destroy(&barrier.changed), "cond_destroy");
    check(pthread_mutex_destroy(&barrier.mutex), "mutex_destroy");
    puts("all barrier rounds completed");
    return 0;
}

barrier participant 数在运行中不变，且 destroy 前所有 thread 已离开。支持 dynamic participant、cancellation 或 broken barrier 时，状态机会复杂得多；优先使用平台的 pthread_barrier_t 或成熟库，而不是复制教学实现。

read-write lock 不保证读多就更快

RW lock 允许多个 reader 或一个 writer。它适合读临界区足够长、真正可并行且 write 较少的 workload；以下情况普通 mutex 可能更好：

临界区很短，RW bookkeeping 超过并行收益；
cache line 因 reader-count 更新仍高度争用；
write 频率不低；
policy 导致 reader 或 writer starvation；
read path 仍会修改 lazy cache/statistics，并不是真只读。

公平性与 writer preference 往往是 implementation policy，不应假设。升级 read lock 到 write lock 还可能 deadlock，除非 API 明确提供且 protocol 处理竞争。

第 11 章会讨论 spin/futex 与 lock implementation；第 16 章再解释 atomic memory order。这里只把 RW lock 当成可测量的策略，不承诺“一定快一个数量级”。

选择原语时先写 predicate

问题	首选表达
一组字段必须作为整体修改	mutex
等待 queue 非空/不满	mutex + condition variable
同时最多 N 个使用者	counting semaphore
N 个 participant 阶段汇合	barrier
长读临界区、少量写	benchmark 后考虑 RW lock
单个计数/flag	C atomic，并证明 memory order

多个 primitive 可以组合，但每增加一种等待关系，deadlock、cancellation 与 shutdown path 都更难。bounded queue 用 mutex+condition variable 往往比“三个 semaphore 模拟 mutex”更容易维护 invariant 与 close semantics。

动手验证通知协议

把单槽 channel 的 while 改成 if，加入两个 consumer，构造 predicate 被先行消费的时序。
将 broadcast 有选择地改为 signal，说明每个位置为何足够或会遗漏角色。
给 channel 增加 cancellation/timeout，保证 producer 不会在 close 后永久等待。
把 reusable barrier 删除 generation，寻找跨轮次错误。
用 semaphore 限制 3 个 concurrent operation，验证任何 error path 都归还 permit。
对同一 read-heavy map 比较 mutex 与 RW lock，报告 critical-section 长度和 starvation policy。

runnable thread 最终还要排队

同步原语决定 thread 何时 blocked 或 runnable，却不决定 runnable 之后何时获得 CPU。下一章进入CPU 调度，比较 response time、turnaround、fairness 与 deadline，而不是只背算法名字。

第1章逻辑

第2章集合论

第3章证明方法

第4章递归、归纳与递推

第5章关系与函数

第6章组合数学

第7章图论

第8章概率与统计

第9章数论

第10章线性代数

第11章微积分与优化

第12章信息论与数值计算