并发与同步：用状态谓词组织线程

并发表示多个执行流的生命周期交叠，并行表示它们在同一时刻执行。单核上交错调度也有并发错误，多核则增加了同时访问共享状态的机会。正确程序需要明确谁拥有数据、什么操作必须不可分割，以及等待条件如何被改变。

一条 C 语句并不自动原子

counter++ 可概念化成读取、加一、写回。两个线程都读到 7，各自写回 8，就丢失一次更新。在 C 的线程内存模型中，对同一非原子对象存在无同步的冲突访问，还构成 data race，行为未定义；不能只把它当成偶尔得到较小数字。

volatile 不提供互斥，也不建立线程间同步。它在设备寄存器或某些信号标志中有用途，但“防优化”不是共享数据正确性的完整条件。可以用锁保护多字段不变量，或在适当场景使用原子对象并明确所需内存顺序。学习开始时，使用简单锁设计比试图从汇编猜测是否“碰巧原子”可靠。

单槽通道的状态

仓库 sync_demo.c 让生产者依次写入 1 到 10000，消费者按顺序取出并累加。通道只有一个位置，关键状态是 full 与 item。两者由同一互斥锁保护。

角色	允许操作的谓词	改变状态	通知
生产者	!full	写入 item，再设 full=true	readable
消费者	full	读取 item，再设 full=false	writable

条件变量不是储存“已经通知几次”的计数器。线程等待的是共享状态变成允许继续的样子；通知只是提示等待者有机会重新检查。因此先写出谓词，再选择等待与唤醒操作。

pthread_mutex_lock(&c->lock);
while (!c->full)
    pthread_cond_wait(&c->readable, &c->lock);
/* 返回时重新持有锁；此处消费 item。 */
c->full = false;
pthread_cond_signal(&c->writable);
pthread_mutex_unlock(&c->lock);

pthread_cond_wait 把释放互斥锁与进入等待正确衔接，返回前重新获得锁。必须用 while，因为可能出现虚假唤醒，也可能在有多个消费者时，另一个消费者先获得锁并取走数据。通知不等于条件依然成立。

所有权怎样帮助推理

本例中生产者只在持锁时写 item，消费者也在同一锁下读；full 同样如此。total 只有消费者修改，主线程在 join 后读取。pthread_join 保证线程已终止，主线程不会与消费者同时读取修改中的 total。锁与线程生命周期共同构成同步关系。

sequenceDiagram
  participant P as 生产者
  participant L as 单槽与互斥锁
  participant C as 消费者
  P->>L: 锁内检查空槽
  P->>L: 写 item，full=true
  P-->>C: signal readable
  P->>L: 解锁
  C->>L: 获锁并重查 full
  C->>L: 读取，full=false
  C-->>P: signal writable
  C->>L: 解锁

查看流程图文本

sequenceDiagram
  participant P as 生产者
  participant L as 单槽与互斥锁
  participant C as 消费者
  P->>L: 锁内检查空槽
  P->>L: 写 item，full=true
  P-->>C: signal readable
  P->>L: 解锁
  C->>L: 获锁并重查 full
  C->>L: 读取，full=false
  C-->>P: signal writable
  C->>L: 解锁

同一顺序可以有许多实际线程调度，但每次取值必须是对应的下一项。测试同时检查 item == i 和最终总和，避免只验证总和时遗漏“顺序错了但总数碰巧正确”。

make -C labs/csapp sync_demo
./labs/csapp/sync_demo

预期为 sync: items=10000 sum=50005000，因为 $10000\times 10001/2=50005000$。运行时间不作为加速比基准；这个程序故意频繁同步，主要验证协议正确性。

/* A one-slot channel: condition variables guard predicates, not event counts. */
#include <assert.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#define ITEMS 10000

typedef struct {
    pthread_mutex_t lock;
    pthread_cond_t readable, writable;
    bool full;
    int item;
    long total;
} Channel;

static void checked(int result) { if (result != 0) { fprintf(stderr, "pthread error %d\n", result); abort(); } }
static void *producer(void *argument) {
    Channel *c = argument;
    for (int i = 1; i <= ITEMS; ++i) {
        checked(pthread_mutex_lock(&c->lock));
        while (c->full) checked(pthread_cond_wait(&c->writable, &c->lock));
        c->item = i;
        c->full = true;
        checked(pthread_cond_signal(&c->readable));
        checked(pthread_mutex_unlock(&c->lock));
    }
    return NULL;
}
static void *consumer(void *argument) {
    Channel *c = argument;
    for (int i = 1; i <= ITEMS; ++i) {
        checked(pthread_mutex_lock(&c->lock));
        while (!c->full) checked(pthread_cond_wait(&c->readable, &c->lock));
        assert(c->item == i);
        c->total += c->item;
        c->full = false;
        checked(pthread_cond_signal(&c->writable));
        checked(pthread_mutex_unlock(&c->lock));
    }
    return NULL;
}
int main(void) {
    Channel channel = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER,
                       PTHREAD_COND_INITIALIZER, false, 0, 0};
    pthread_t p, c;
    checked(pthread_create(&p, NULL, producer, &channel));
    checked(pthread_create(&c, NULL, consumer, &channel));
    checked(pthread_join(p, NULL));
    checked(pthread_join(c, NULL));
    assert(channel.total == (long)ITEMS * (ITEMS + 1) / 2);
    printf("sync: items=%d sum=%ld\n", ITEMS, channel.total);
    checked(pthread_mutex_destroy(&channel.lock));
    checked(pthread_cond_destroy(&channel.readable));
    checked(pthread_cond_destroy(&channel.writable));
    return 0;
}

死锁与临界区大小

两把锁 A、B 被不同线程按相反顺序获得，可能形成等待环：线程 1 持有 A 等 B，线程 2 持有 B 等 A。为所有代码制定一致的加锁顺序是常用解决方式。若锁内进行长时间 I/O，虽然未必死锁，却可能让所有线程排队，吞吐量明显下降。

把临界区缩小也不能随意移动语句。例如先解锁再读 item，会让生产者覆盖它；先设 full=false 再保存 item 也可能产生类似问题。先确定受保护的不变量，再考虑把与共享状态无关的计算移出锁。

并行程序还可能遇到 false sharing：不同线程写不同变量，但变量落在同一缓存行，导致缓存一致性反复转移所有权。这不是 C 层面的数据竞争，却影响性能。正确性与性能需要不同证据：锁或原子协议证明无竞争，测量与硬件事件分析瓶颈。

扩展练习：容量 N 的队列

把单槽改成环形数组，增加 head、tail、count。生产者等待 count < N，写 tail 后前移并增加 count；消费者等待 count > 0，读 head 后前移并减少 count。维护 0 <= count <= N，所有状态变化在同一锁下完成。

完成后再增加关闭协议：closed=true 后生产者不能继续放入；消费者在 count==0 && closed 时结束，而不是无限等待。关闭时通常要 broadcast 唤醒所有等待者，让它们检查结束条件。若只有一个 poison pill，却有多个消费者，其他消费者可能永远等不到结束标记。

自测：把 while 改成 if，在单生产者单消费者测试中没失败，能否说明正确？

不能。条件变量允许虚假唤醒，且设计未来可能扩展到多个等待者。正确性应基于接口保证与不变量，而不是某次调度没有暴露问题。

自测：锁保护了两个字段，读取它们时只锁其中一个是否足够？

锁并不附着在变量上；它保护所有遵守同一协议的临界区。若需要读取两个字段的一致快照，必须在所有修改者使用的锁下同时读取，或另行设计等价同步协议。

课程参考：CS:APP3e 学生资源。下一章：经典实验手册。