关键问题与本章定位

前两章我们知道：解决各种并发问题既需要锁（lock），也需要条件变量（condition variable）。Dijkstra 和同事们很早就意识到了这一点，于是发明了信号量（semaphore）——一个想"包打天下"的同步原语：同一个东西，初值设成 1 就是锁，设成 0 就能表达"等某件事发生"。

本章先给出信号量的精确定义，然后用它重做一遍熟悉的问题（锁、join、生产者/消费者），再看几个经典问题（读写锁、哲学家就餐、线程节流），最后用锁 + 条件变量亲手实现一个信号量。本章中"拿着一把锁去等另一样东西"引发的死锁，将在下一章《常见并发问题》里被系统地分析。

核心概念速览

信号量的定义

信号量是一个带整数值的对象。因为初值决定了它的行为，所以用它之前必须先初始化（Figure 31.1）：

#include <semaphore.h>
sem_t s;
sem_init(&s, 0, 1);   // 第 3 个参数：初值 1；第 2 个参数 0：在同一进程的线程之间共享

初始化之后，只能用两个函数操作它（Figure 31.2）。先不管它们内部怎么实现（显然需要小心处理并发），先假设这两个操作都是原子的：

int sem_wait(sem_t *s) {
    decrement the value of semaphore s by one     // 先减 1
    wait if value of semaphore s is negative      // 减完是负数就睡
}

int sem_post(sem_t *s) {
    increment the value of semaphore s by one     // 先加 1
    if there are one or more threads waiting, wake one   // 有人在等就唤醒一个
}

关于这两个接口，原书强调三点：

1. sem_wait() 要么立即返回（调用时值 ≥ 1），要么让调用者挂起，等待之后的某次 post。多个线程可以同时卡在 sem_wait() 里排队。
2. sem_post() 从不等待任何条件。它只是把值加 1，如果有线程在等就唤醒一个。
3. 值为负时，其绝对值等于等待线程数。使用者通常看不到这个值，但记住这个不变量有助于理解信号量如何运作。

二值信号量：把信号量当锁用

第一个用法我们已经很熟悉了：用 sem_wait()/sem_post() 包住临界区（Figure 31.3）。关键问题是：初值 X 应该是多少？先想一想再往下看。

sem_t m;
sem_init(&m, 0, X);   // X 应该是多少？

sem_wait(&m);
// 临界区
sem_post(&m);

答案是 1。第一个线程调用 sem_wait() 把值从 1 减到 0，0 不是负数，于是直接返回、进入临界区；如果这时第二个线程也来 sem_wait()，它把值减到 -1，只能睡眠。第一个线程离开时 sem_post() 把值加回 0 并唤醒第二个线程；第二个线程用完后再 post，值回到 1。

原书 Figure 31.5 给出了两个线程争锁的完整轨迹：T0 进入临界区后被中断，T1 调用 sem_wait() 把值减到 -1 并睡眠；T0 回来后 post，值变回 0 并唤醒 T1；T1 从 sem_wait() 返回、执行临界区、post，值回到 1。在下方"信号量执行器"里选择 Figure 31.5 预设，可以逐行复现这张表。注意一个细节：T1 被唤醒后从 sem_wait() 返回时不会再减一次——它的那次减 1 在睡下去之前就做完了。

如果只把信号量当二值锁用，其实可以实现得比通用信号量更简单。

信号量用于排序：实现 join

信号量也可以用来给事件排序：一个线程等某件事发生，另一个线程让这件事发生后发信号——这和上一章用条件变量做的事一样。经典例子是父线程等待子线程结束（Figure 31.6）：

sem_t s;

void *child(void *arg) {
    printf("child\n");
    sem_post(&s);        // 发信号：子线程完成了
    return NULL;
}

int main(int argc, char *argv[]) {
    sem_init(&s, 0, X);  // X 应该是多少？
    printf("parent: begin\n");
    pthread_t c;
    Pthread_create(&c, NULL, child, NULL);
    sem_wait(&s);        // 在这里等子线程
    printf("parent: end\n");
    return 0;
}

我们希望输出总是 parent: begin → child → parent: end。答案是 X = 0。要考虑两种情况：

情况 1：子线程已创建但还没运行，父线程先调用 sem_wait()。我们希望父线程等待，唯一的办法是值不大于 0——所以初值是 0。父线程把值减到 -1 后睡眠；子线程运行后 sem_post() 把值加到 0 并唤醒父线程。情况 2：子线程在父线程调用 sem_wait() 之前就跑完了，post 把值从 0 加到 1；之后父线程 sem_wait() 把 1 减到 0，不需要等待就返回。两种情况都达到了目的。

生产者/消费者（有界缓冲区）问题

上一章用条件变量解决过这个问题；现在只用信号量来做。先写好 put 和 get（Figure 31.9），注意这里 put/get 本身不带任何同步：

int buffer[MAX];
int fill = 0;
int use  = 0;

void put(int value) {
    buffer[fill] = value;       // Line F1
    fill = (fill + 1) % MAX;    // Line F2
}

int get() {
    int tmp = buffer[use];      // Line G1
    use = (use + 1) % MAX;      // Line G2
    return tmp;
}

第一次尝试：empty 和 full

引入两个信号量：empty 表示空槽数，full 表示满槽数（Figure 31.10）。生产者先等一个空槽再放数据，然后宣布多了一个满槽；消费者对称。

sem_t empty;
sem_t full;

void *producer(void *arg) {
    for (int i = 0; i < loops; i++) {
        sem_wait(&empty);       // Line P1
        put(i);                 // Line P2
        sem_post(&full);        // Line P3
    }
}

void *consumer(void *arg) {
    int tmp = 0;
    while (tmp != -1) {
        sem_wait(&full);        // Line C1
        tmp = get();            // Line C2
        sem_post(&empty);       // Line C3
        printf("%d\n", tmp);
    }
}

int main(int argc, char *argv[]) {
    // ...
    sem_init(&empty, 0, MAX);   // MAX 个空槽
    sem_init(&full, 0, 0);      // 0 个满槽
    // ...
}

先看 MAX = 1、一个生产者一个消费者。假设消费者先运行：它在 C1 调用 sem_wait(&full)，full 初值 0，被减到 -1，消费者睡眠。然后生产者运行：P1 把 empty 从 1 减到 0，通过；P2 放入数据；P3 把 full 从 -1 加到 0 并唤醒消费者。之后如果生产者继续跑，它在 P1 把 empty 减到 -1 而睡眠；如果换消费者跑，它从 C1 返回、取走数据。两种情况都正确。多个生产者、多个消费者（MAX = 1）也仍然正确。

MAX > 1 时的竞争：数据被覆盖

现在让 MAX = 10，并且有多个生产者。问题出在 put() 和 get() 上：设生产者 Pa 先运行，在 F1 往 buffer[0] 写数据，还没来得及执行 F2 把 fill 加到 1 就被中断；生产者 Pb 运行，它在 F1 也往 buffer[0] 写——Pa 的数据被覆盖、永远丢失了。原因是忘了互斥：填缓冲区和移动 fill 下标是一个临界区。

加锁，但放错了位置：死锁

用一个二值信号量 mutex 把整个 put/get 部分包起来（Figure 31.11，NEW LINE 标出的是新加的两行）：

void *producer(void *arg) {
    for (int i = 0; i < loops; i++) {
        sem_wait(&mutex);       // Line P0 (NEW LINE)
        sem_wait(&empty);       // Line P1
        put(i);                 // Line P2
        sem_post(&full);        // Line P3
        sem_post(&mutex);       // Line P4 (NEW LINE)
    }
}

void *consumer(void *arg) {
    for (int i = 0; i < loops; i++) {
        sem_wait(&mutex);       // Line C0 (NEW LINE)
        sem_wait(&full);        // Line C1
        int tmp = get();        // Line C2
        sem_post(&empty);       // Line C3
        sem_post(&mutex);       // Line C4 (NEW LINE)
        printf("%d\n", tmp);
    }
}

看起来对，但会死锁（deadlock）。设一个生产者一个消费者，消费者先运行：它在 C0 拿到 mutex，然后在 C1 对 full 调用 sem_wait()——没有数据，于是睡眠。关键是：它睡着的时候仍然持有 mutex。接着生产者运行，它本来可以放入数据并唤醒消费者，可它第一件事是在 P0 等 mutex——mutex 已被占用，生产者也睡着了。

正确的解法：缩小锁的范围

只需把 mutex 的获取和释放挪到只包住临界区（put/get），full/empty 的等待和通知留在外面（Figure 31.12）：

void *producer(void *arg) {
    for (int i = 0; i < loops; i++) {
        sem_wait(&empty);       // Line P1
        sem_wait(&mutex);       // Line P1.5 (lock)
        put(i);                 // Line P2
        sem_post(&mutex);       // Line P2.5 (unlock)
        sem_post(&full);        // Line P3
    }
}

void *consumer(void *arg) {
    for (int i = 0; i < loops; i++) {
        sem_wait(&full);        // Line C1
        sem_wait(&mutex);       // Line C1.5 (lock)
        int tmp = get();        // Line C2
        sem_post(&mutex);       // Line C2.5 (unlock)
        sem_post(&empty);       // Line C3
        printf("%d\n", tmp);
    }
}

现在任何线程都不会拿着 mutex 去睡：它只在确定有空槽/有数据之后才进锁，锁内的 put/get 很短、不会阻塞。这就是常用的有界缓冲区模式——原书说"现在弄懂它，以后会用上"，面试和期末考试尤其爱考。（原书脚注：为了模块化，把 mutex 的获取/释放放进 put() 和 get() 里面可能更自然。）

读写锁

另一个经典问题来自这样的需求：不同的数据结构访问需要不同的锁。比如一个并发链表，插入会修改链表（需要传统的临界区），而查找只是读——只要保证没有插入在进行，多个查找完全可以同时进行。支持这种操作的锁叫读写锁（reader-writer lock）（Figure 31.13）：

typedef struct _rwlock_t {
    sem_t lock;       // 二值信号量（普通锁），保护 readers
    sem_t writelock;  // 允许 1 个写者 或 多个读者
    int   readers;    // 临界区里有几个读者
} rwlock_t;

void rwlock_init(rwlock_t *rw) {
    rw->readers = 0;
    sem_init(&rw->lock, 0, 1);
    sem_init(&rw->writelock, 0, 1);
}

void rwlock_acquire_readlock(rwlock_t *rw) {
    sem_wait(&rw->lock);
    rw->readers++;
    if (rw->readers == 1)          // 第一个读者负责拿 writelock
        sem_wait(&rw->writelock);
    sem_post(&rw->lock);
}

void rwlock_release_readlock(rwlock_t *rw) {
    sem_wait(&rw->lock);
    rw->readers--;
    if (rw->readers == 0)          // 最后一个读者负责放掉它
        sem_post(&rw->writelock);
    sem_post(&rw->lock);
}

void rwlock_acquire_writelock(rwlock_t *rw) { sem_wait(&rw->writelock); }
void rwlock_release_writelock(rwlock_t *rw) { sem_post(&rw->writelock); }

写者很简单：直接用 writelock 保证只有一个写者进入。读者更有意思：先拿 lock，把 readers 加 1；如果自己是第一个读者，就顺带替整个"读者群"拿走 writelock，然后释放 lock。之后的读者只需 readers++ 就能进去；而写者必须等到所有读者都离开——最后一个离开的读者 post writelock。

这个实现能用，但不公平：读者很容易饿死（starve）写者。更复杂的方案可以解决它——提示：一旦有写者在等，就不再让新读者进入。另外要注意，读写锁通常开销更大（尤其是复杂实现），和简单快速的锁相比，未必真能提速。

哲学家就餐问题

Dijkstra 提出并解决的最著名的并发问题之一是哲学家就餐问题（dining philosophers）。它有趣、也有点智力挑战，实用价值不高——但名气太大，面试可能会问到。

设定：5 位哲学家围坐一桌，每两人之间放一把叉子（共 5 把）。哲学家时而思考（不需要叉子），时而吃饭；吃饭需要左右两把叉子。每位哲学家 p（0~4）的循环是：

while (1) {
    think();
    get_forks(p);
    eat();
    put_forks(p);
}

挑战是写出 get_forks() 和 put_forks()，使得没有死锁、没有哲学家饿死、并发度尽量高（尽可能多的人同时吃）。按 Downey 的写法，先定义两个辅助函数和 5 个信号量（每把叉子一个，初值 1）：

int left(int p)  { return p; }
int right(int p) { return (p + 1) % 5; }   // p=4 时右叉是 fork 0
sem_t forks[5];

错误的解法

void get_forks(int p) {
    sem_wait(&forks[left(p)]);
    sem_wait(&forks[right(p)]);
}

void put_forks(int p) {
    sem_post(&forks[left(p)]);
    sem_post(&forks[right(p)]);
}

直觉很简单：先"锁住"左叉，再锁住右叉，吃完都放下。可惜简单在这里意味着错误：如果每位哲学家都在任何人拿右叉之前拿起了左叉——P0 拿 f0、P1 拿 f1、P2 拿 f2、P3 拿 f3、P4 拿 f4——所有叉子都被拿走，每个人都攥着一把叉子等另一把，永远等下去：死锁。

解法：打破依赖

最简单的办法是让至少一位哲学家换个顺序拿叉子——Dijkstra 本人就是这样解决的。让编号最大的 P4 先拿右叉再拿左叉（Figure 31.16），put_forks() 不变：

void get_forks(int p) {
    if (p == 4) {
        sem_wait(&forks[right(p)]);
        sem_wait(&forks[left(p)]);
    } else {
        sem_wait(&forks[left(p)]);
        sem_wait(&forks[right(p)]);
    }
}

因为最后一位哲学家先拿右叉，就不可能出现"每人拿一把、都在等下一把"的局面——等待的环被打破了。想一想为什么：P4 的右叉是 f0，它和 P0 抢的是同一把"第一把叉子"；谁没抢到，谁手里就是空的，不会占着叉子等人。下一章会把这个思想总结为"破坏循环等待（circular wait）"。

类似的"著名问题"还有吸烟者问题（cigarette smoker's problem）、睡眠理发师问题（sleeping barber problem）等，它们大多是练习并发思维的借口。Downey 的《The Little Book of Semaphores》收录了大量这类题目。

线程节流

信号量还有一个简单用途：程序员怎样防止"太多"线程同时做某件事、把系统拖垮？答案：定一个"太多"的阈值，然后用信号量限制同时执行那段代码的线程数。这叫节流（throttling），是一种准入控制（admission control）。

具体例子：你创建了几百个线程并行处理一个问题，但代码中某一段（称为内存密集区）每个线程都要申请一大块内存。如果所有线程同时进入这一段，内存申请总量超过物理内存，机器就开始抖动（thrashing）——不停地换页，整个计算慢如蜗牛。只要把信号量初值设为"允许同时进入的最大线程数"，并在这段代码前后放 sem_wait() 和 sem_post()，就能自然地限流：

sem_t gate;
sem_init(&gate, 0, K);     // 最多 K 个线程同时在内存密集区

void *worker(void *arg) {
    ...                    // 普通计算
    sem_wait(&gate);       // 进门拿一个名额
    memory_intensive_region();
    sem_post(&gate);       // 出门还回名额
    ...
}

如何实现信号量：Zemaphore

最后，用底层同步原语——锁和条件变量——构造我们自己的信号量，叫作……（敲鼓）……Zemaphore（Figure 31.17）：

typedef struct __Zem_t {
    int value;
    pthread_cond_t cond;
    pthread_mutex_t lock;
} Zem_t;

// 只能由一个线程调用
void Zem_init(Zem_t *s, int value) {
    s->value = value;
    Cond_init(&s->cond);
    Mutex_init(&s->lock);
}

void Zem_wait(Zem_t *s) {
    Mutex_lock(&s->lock);
    while (s->value <= 0)             // 没有名额就等（while，不是 if！）
        Cond_wait(&s->cond, &s->lock);
    s->value--;                       // 有名额才减
    Mutex_unlock(&s->lock);
}

void Zem_post(Zem_t *s) {
    Mutex_lock(&s->lock);
    s->value++;
    Cond_signal(&s->cond);
    Mutex_unlock(&s->lock);
}

只用了一把锁、一个条件变量，再加一个记录信号量值的状态变量。和 Dijkstra 定义的"纯"信号量有一个细微差别：Zemaphore 不维护"值为负时等于等待者数"这个不变量——它的值永远不会小于 0（先判断、有名额才减）。这种做法更容易实现，也和当前 Linux 的实现一致。

反过来，用信号量构造条件变量要难得多。一些经验丰富的并发程序员在 Windows 环境下尝试过，结果冒出了许多不同的 bug（Birrell 的文章记录了这段经历）。你可以自己试试，想想为什么它比看上去难——提示：条件变量的 signal 在没人等时什么都不做，而信号量的 post 会被"记住"。

先预测再验证

动手实验

cd labs/31-semaphores && make
./fork_join; ./fork_join 1
./pc -p 2 -c 3 -m 10 -l 100000
./pc -b
./dining broken; ./dining fixed
./barrier 5
./throttle 20 3; ./throttle 20 0

$ ./fork_join
parent: begin (sem 初值 = 0)
child
parent: end
$ ./fork_join 1
parent: begin (sem 初值 = 1)
parent: end
child

$ ./pc -p 2 -c 3 -m 10 -l 100000
正确版：2 个生产者，3 个消费者，MAX=10，每个生产者 100000 个
   消费者 C0 取走 64225 个
   消费者 C1 取走 76677 个
   消费者 C2 取走 59098 个
生产 200000 个 (校验和 109999900000)，消费 200000 个 (校验和 109999900000) → 一致 ✓
$ ./pc -b
错误（mutex 在外层）版：1 个生产者，1 个消费者，MAX=1，每个生产者 10000 个
!! 2 秒没有进展：检测到死锁。各线程状态：
   生产者 P0: 等 mutex
   消费者 C0: 持有 mutex 并等 full
   已完成 put/get 次数 = 0

$ ./dining broken
错误版（全部先左后右）：5 位哲学家，每人吃 1000 次
!! 2 秒内没有人吃上饭：检测到死锁（已吃 6 顿）
   P0 拿着 f0，等待 f1
   P1 拿着 f1，等待 f2
   P2 拿着 f2，等待 f3
   P3 拿着 f3，等待 f4
   P4 拿着 f4，等待 f0
   等待链：P0→f1(P1持有)→f2(P2持有)→…→f0(P0持有)，形成环
$ ./dining fixed
修复版（P4 先右后左）：5 位哲学家，每人吃 1000 次
全部吃完，用时 3.21 秒；每人吃的次数： P0=1000 P1=1000 P2=1000 P3=1000 P4=1000

$ ./barrier 5
线程 0: P1
线程 2: P1
线程 1: P1
线程 4: P1
线程 3: P1
线程 3:     P2 (此时已完成 P1 的线程数 = 5)
线程 2:     P2 (此时已完成 P1 的线程数 = 5)
线程 0:     P2 (此时已完成 P1 的线程数 = 5)
线程 1:     P2 (此时已完成 P1 的线程数 = 5)
线程 4:     P2 (此时已完成 P1 的线程数 = 5)
✓ 所有 P2 都发生在全部 P1 之后

$ ./throttle 20 3; ./throttle 20 0
20 个线程，K=3：内存密集区里同时最多有 3 个线程（峰值约 24 MB）
20 个线程，K=0（不节流）：内存密集区里同时最多有 20 个线程（峰值约 160 MB）

自测题

常见误区与小结