1.12 并发编程

1.11 的 fork 服务器每来一个连接就复制一个进程：地址空间隔离，写起来简单，但连接数上千时，页表复制和上下文切换会把 CPU 吃光。要支撑更多连接，必须让多个执行流共享同一份地址空间，或者让一个执行流同时管理多条连接。前者是线程，后者是 I/O 多路复用。

并发（concurrency）与并行（parallelism）不是一回事。并发是「多个任务在时间上重叠」，单核分时也能并发；并行是「多个任务在同一时刻真正执行」，需要多核。并发解决的是「等待时别闲着」，并行解决的是「算得更快」。两者的收益模型不同：前者看吞吐与延迟，后者受 Amdahl 定律约束（1.1.9）。

这一章的主线是共享状态。只要两个执行流能读写同一块内存，就必须回答：谁先谁后？读到的值是否完整？失败时怎么回滚？信号量、互斥锁、条件变量、线程池，都是围绕这三个问题的工程答案。1.12.7 把「共享内存 + 锁」换成「单线程 + 事件」，这是另一种同样重要的并发模型。

前七节给的是「会用」：三种模型、pthread 基本 API、锁与条件变量、信号量、线程池、死锁、epoll。后九节补「用对」：线程的完整生命周期与错误码（1.12.8）、C11 原子与内存序（1.12.9）、锁在内核里怎么实现（1.12.10）、读写锁/seqlock/RCU 的取舍（1.12.11）、死锁的复现与证据链（1.12.12）、线程池的容量与关闭语义（1.12.13）、C11/C++/Rust 三语言对照（1.12.14）、TSan 与 perf 实验（1.12.15）、十八条陷阱（1.12.16）。

1.12.1 三种并发模型

CSAPP 用三种服务器并发展示这个谱系：每连接一个进程（12.1）、I/O 多路复用（12.2）、每连接一个线程（12.3）。三者的接口都是「accept 一个连接，处理请求」，但资源模型完全不同。

进程模型的关键成本是 fork：即使有 COW，页表复制和调度器里的任务增多仍然存在。线程模型把「地址空间」这一层省掉，创建线程默认只需分配栈（glibc 默认 8 MiB 虚拟栈，按需缺页），切换不换页表，代价小得多。事件模型进一步把「执行流」也省掉，一个线程用 epoll 管理大量 fd，只有在事件就绪时才处理。

没有一种模型全面占优。计算密集型任务用线程/多进程吃多核；大量长连接、低计算量的网关用事件循环；需要强隔离的插件用多进程。真实系统常常混用：主线程 epoll 收连接，把请求交给线程池处理。

1.12.2 线程：共享地址空间带来的自由与麻烦

pthread_create 创建一个线程，新线程从指定的函数开始执行；pthread_join 等待它结束并回收资源。线程之间共享：

• 全局变量、静态变量、堆；
• 打开的文件描述符、当前工作目录、umask；
• 信号处理表（但每个线程有独立的信号掩码和 pending 集合）。

每个线程私有：

• 栈、寄存器上下文；
• 线程 ID、errno（线程局部）；
• 线程局部存储（thread-local storage, TLS）；
• 信号掩码、pthread_cleanup 栈。

共享是性能的来源，也是 bug 的来源。下面的程序让 4 个线程各把同一个计数器加 100 万次：

/* race.c：数据竞争演示；cc ... -pthread */
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <pthread.h>
#include <stdio.h>

#define NTHREADS 4
#define NLOOP    1000000L

static long counter;

static void *worker(void *arg)
{
    (void)arg;
    for (long i = 0; i < NLOOP; i++)
        counter++;                     /* 读-改-写，非原子 */
    return NULL;
}

int main(void)
{
    pthread_t t[NTHREADS];
    for (int i = 0; i < NTHREADS; i++) {
        if (pthread_create(&t[i], NULL, worker, NULL) != 0) {
            perror("pthread_create");
            return 1;
        }
    }
    for (int i = 0; i < NTHREADS; i++)
        pthread_join(t[i], NULL);

    printf("counter=%ld expected=%ld\n", counter, NTHREADS * NLOOP);
    return 0;
}

结果通常小于 4000000，且每次运行不同。counter++ 在机器层是「load → add → store」三步，两个线程可能同时读到旧值，各自加一后写回，更新丢失一次。更糟的是，C11 把这种无同步的并发读写定义为数据竞争（data race），行为未定义——编译器可以假设它不发生，从而做出让你意外的优化。

修法有三层：

volatile 不是同步手段。它只告诉编译器「每次都从内存读」，不保证原子性，也不建立线程间的 happens-before 关系。用 volatile 修数据竞争，是把 bug 藏得更深。

1.12.3 互斥锁与条件变量

互斥锁（mutex）保证同一时刻只有一个线程进入临界区：

pthread_mutex_lock(&mtx);
/* 临界区：读写共享状态 */
pthread_mutex_unlock(&mtx);

锁的粒度是设计问题：太大，所有线程排队，并发退化为串行；太小，保护不完整，仍会竞争。一个实用规则是「锁保护的是不变量，不是代码行」：先写下「哪些字段必须一起更新」，再决定锁的范围。

条件变量（condition variable）解决「等某个条件成立」的问题。它必须和互斥锁配合：

pthread_mutex_lock(&mtx);
while (!condition)                 /* 必须用 while，不能用 if */
    pthread_cond_wait(&cond, &mtx);
/* 条件成立，且持有锁 */
pthread_mutex_unlock(&mtx);

pthread_cond_wait 做三件事：释放 mutex、睡眠、被唤醒后重新获取 mutex。它不是「睡到条件为真」，而是「睡到有人通知」；通知可能丢失、可能虚假唤醒（spurious wakeup），所以必须用 while 重新检查条件。

另一个线程在改变条件后调用 pthread_cond_signal（唤醒一个）或 pthread_cond_broadcast（唤醒全部）。改变条件必须在持锁时进行，否则会出现「检查完条件、尚未睡眠，通知先到」的丢失唤醒。

条件变量适合「等待某个状态变化」，信号量适合「计数资源」。两者都能实现生产者-消费者，但语义不同：条件变量没有记忆，通知时没有等待者就丢失；信号量的计数会保留，sem_post 在没人等待时也会加一。

1.12.4 信号量与生产者-消费者

信号量是一个非负整数，支持两个原子操作：

• sem_wait（P）：若值 > 0 则减一；否则阻塞，直到有人 sem_post。
• sem_post（V）：值加一，唤醒一个等待者。

用三个信号量实现有界缓冲区：

• empty = N：空槽数量，生产者等待它；
• full = 0：已填槽数量，消费者等待它；
• mutex = 1：保护 in/out 与缓冲区。

/* pc.c：信号量版有界缓冲区；cc ... -pthread */
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <errno.h>
#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>

#define N         8
#define PRODUCE   100

static int buf[N];
static int in, out;
static sem_t empty_sem, full_sem;
static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;

static int wait_sem(sem_t *s)
{
    while (sem_wait(s) == -1) {
        if (errno == EINTR)
            continue;
        return -1;
    }
    return 0;
}

static void *producer(void *arg)
{
    (void)arg;
    for (int i = 0; i < PRODUCE; i++) {
        if (wait_sem(&empty_sem) == -1) {
            perror("sem_wait empty");
            return NULL;
        }
        pthread_mutex_lock(&mtx);
        buf[in] = i;
        in = (in + 1) % N;
        pthread_mutex_unlock(&mtx);
        if (sem_post(&full_sem) == -1)
            perror("sem_post full");
    }
    return NULL;
}

static void *consumer(void *arg)
{
    (void)arg;
    long sum = 0;
    for (int i = 0; i < PRODUCE; i++) {
        if (wait_sem(&full_sem) == -1) {
            perror("sem_wait full");
            return NULL;
        }
        pthread_mutex_lock(&mtx);
        sum += buf[out];
        out = (out + 1) % N;
        pthread_mutex_unlock(&mtx);
        if (sem_post(&empty_sem) == -1)
            perror("sem_post empty");
    }
    printf("consumer sum=%ld (expected %d)\n", sum, PRODUCE * (PRODUCE - 1) / 2);
    return NULL;
}

int main(void)
{
    if (sem_init(&empty_sem, 0, N) == -1 || sem_init(&full_sem, 0, 0) == -1) {
        perror("sem_init");
        return 1;
    }

    pthread_t p, c;
    if (pthread_create(&p, NULL, producer, NULL) != 0 ||
        pthread_create(&c, NULL, consumer, NULL) != 0) {
        perror("pthread_create");
        return 1;
    }
    pthread_join(p, NULL);
    pthread_join(c, NULL);

    sem_destroy(&empty_sem);
    sem_destroy(&full_sem);
    pthread_mutex_destroy(&mtx);
    return 0;
}

顺序不能颠倒：

• 生产者先 sem_wait(empty)，再拿 mutex；如果先拿 mutex 再等 empty，缓冲区满时生产者握着锁睡眠，消费者永远拿不到锁，死锁。
• 消费者先 sem_wait(full)，再拿 mutex；同理。
• in/out 只在持有 mutex 时修改；缓冲区槽位的写入/读取也在锁内，避免读到半写状态。

这个模型可以推广到连接池、工作队列、限流器：empty 表示「还有多少容量」，full 表示「有多少待处理任务」。信号量的初值就是资源数量，sem_wait 就是申请资源。

1.12.5 线程池与有界队列

每次请求都 pthread_create 的服务器，在请求率高时会把时间花在创建/销毁线程上。线程池预先创建固定数量的工作线程，任务提交到队列，空闲线程取出执行。

                 submit(task)
                      │
                      ▼
              ┌───────────────┐
              │  有界任务队列  │  capacity = 8
              │ [t][t][t][ ]  │
              └───────┬───────┘
        ┌─────────────┼─────────────┐
        ▼             ▼             ▼
   worker 1      worker 2      worker 3  ... worker N
        │             │             │
        └──── 从队列取任务，执行 fn(arg) ────┘
​
  队列满时 submit 阻塞（背压），而不是无限堆积内存
  shutdown 时广播，worker 取完剩余任务后退出

读图抓住三个同步点：queue 由 mutex 保护；not_empty 让 worker 在队列空时睡眠；not_full 让 submit 在队列满时等待。shutdown 是一个受锁保护的布尔量，广播后所有 worker 醒来，取完剩余任务退出。

/* pool.c：固定线程池 + 有界队列 + 优雅关闭；cc ... -pthread */
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <errno.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#define NWORKERS 4
#define QCAP     8

typedef struct task {
    void (*fn)(void *);
    void *arg;
    struct task *next;
} task_t;

typedef struct {
    pthread_mutex_t mtx;
    pthread_cond_t  not_empty;
    pthread_cond_t  not_full;
    task_t *head, *tail;
    int     count;
    int     shutdown;
} pool_t;

static void *worker(void *arg)
{
    pool_t *p = arg;
    for (;;) {
        pthread_mutex_lock(&p->mtx);
        while (p->count == 0 && !p->shutdown)
            pthread_cond_wait(&p->not_empty, &p->mtx);
        if (p->count == 0 && p->shutdown) {
            pthread_mutex_unlock(&p->mtx);
            break;
        }
        task_t *t = p->head;
        p->head = t->next;
        if (p->head == NULL)
            p->tail = NULL;
        p->count--;
        pthread_cond_signal(&p->not_full);
        pthread_mutex_unlock(&p->mtx);

        t->fn(t->arg);
        free(t);
    }
    return NULL;
}

static int pool_init(pool_t *p)
{
    p->head = p->tail = NULL;
    p->count = 0;
    p->shutdown = 0;
    if (pthread_mutex_init(&p->mtx, NULL) != 0)
        return -1;
    if (pthread_cond_init(&p->not_empty, NULL) != 0)
        return -1;
    if (pthread_cond_init(&p->not_full, NULL) != 0)
        return -1;
    return 0;
}

static int pool_submit(pool_t *p, void (*fn)(void *), void *arg)
{
    task_t *t = malloc(sizeof(*t));
    if (t == NULL)
        return -1;
    t->fn = fn;
    t->arg = arg;
    t->next = NULL;

    pthread_mutex_lock(&p->mtx);
    while (p->count == QCAP && !p->shutdown)
        pthread_cond_wait(&p->not_full, &p->mtx);
    if (p->shutdown) {
        pthread_mutex_unlock(&p->mtx);
        free(t);
        return -1;
    }
    if (p->tail == NULL)
        p->head = t;
    else
        p->tail->next = t;
    p->tail = t;
    p->count++;
    pthread_cond_signal(&p->not_empty);
    pthread_mutex_unlock(&p->mtx);
    return 0;
}

static void pool_shutdown(pool_t *p, pthread_t *workers, int n)
{
    pthread_mutex_lock(&p->mtx);
    p->shutdown = 1;
    pthread_cond_broadcast(&p->not_empty);
    pthread_cond_broadcast(&p->not_full);
    pthread_mutex_unlock(&p->mtx);

    for (int i = 0; i < n; i++)
        pthread_join(workers[i], NULL);

    pthread_mutex_destroy(&p->mtx);
    pthread_cond_destroy(&p->not_empty);
    pthread_cond_destroy(&p->not_full);
}

static void job(void *arg)
{
    long id = (long)(intptr_t)arg;
    usleep(10000);                     /* 模拟 10 ms 工作 */
    printf("job %ld done\n", id);
}

int main(void)
{
    pool_t p;
    if (pool_init(&p) != 0) {
        fprintf(stderr, "pool_init failed\n");
        return 1;
    }

    pthread_t workers[NWORKERS];
    for (int i = 0; i < NWORKERS; i++) {
        if (pthread_create(&workers[i], NULL, worker, &p) != 0) {
            perror("pthread_create");
            return 1;
        }
    }

    for (long i = 0; i < 20; i++) {
        if (pool_submit(&p, job, (void *)(intptr_t)i) != 0)
            fprintf(stderr, "submit %ld rejected\n", i);
    }

    pool_shutdown(&p, workers, NWORKERS);
    return 0;
}

几个设计点：

• 有界队列：容量 8，满了 pool_submit 阻塞，形成背压（backpressure）。无界队列在过载时会把内存吃光，最后 OOM，比拒绝请求更糟。
• while 而不是 if：pthread_cond_wait 可能虚假唤醒；shutdown 与任务到达也可能竞争。
• 任务所有权：pool_submit 分配 task_t，worker 执行完 free；submit 失败时由调用者释放。
• 优雅关闭：shutdown 置位后广播，worker 把队列里剩余任务执行完再退出；pool_shutdown 负责 join 和销毁同步对象。
• 不要在执行任务时持锁：worker 弹出任务后立刻解锁，fn 里可能阻塞或再次提交任务，持锁会死锁。

这个池的 job 用 usleep 模拟 I/O；真实服务里任务可能是解析请求、查数据库、写日志。若任务本身是 CPU 密集型，worker 数应接近核数；若是 I/O 密集型，可以多于核数，但要用压测确认。

1.12.6 读写者、线程安全与死锁

读写锁

当共享数据「读多写少」时，互斥锁让读操作也串行，浪费并行度。读写锁允许：

• 多个读者同时持有读锁；
• 写者独占，读者和写者互斥。

pthread_rwlock_rdlock(&rw);   /* 读 */
pthread_rwlock_wrlock(&rw);   /* 写 */
pthread_rwlock_unlock(&rw);

读写锁的代价是复杂度：写者可能饥饿（读者源源不断），读者也可能饥饿（写者优先）。glibc 的 pthread_rwlock 默认读者优先，写多场景可能让写者长期拿不到锁；需要写者优先时用 pthread_rwlockattr_setkind_np 设 PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP。读临界区很长时，读写锁的收益会被缓存行争用抵消，未必比互斥锁快。

线程安全与可重入

• 线程安全：多个线程同时调用不会出错。靠锁、原子操作、线程局部存储实现。
• 可重入：函数在执行途中被再次调用（同一线程或信号处理）仍然正确。可重入通常要求不持有全局状态、不使用静态缓冲区。

strtok 用静态指针保存解析位置，不可重入；strtok_r 把状态交给调用者。errno 在 glibc 里是线程局部变量，所以多线程各自看到自己的错误码。localtime 返回指向静态 struct tm 的指针，多线程要用 localtime_r。

死锁的四个条件

死锁需要同时满足：互斥、持有并等待、不可剥夺、循环等待。打破任意一个即可：

工程上最有效的是锁顺序：给所有锁编号，永远按升序获取。两个函数分别按 A→B 和 B→A 加锁，迟早死锁；统一成 A→B 就不会有环。

哲学家就餐是这组条件的最小模型：五位哲学家围桌，每人的工作都要同时拿起左右两把叉子。若五人同时拿起左叉、再等右叉，就各持一把、互相等待，形成循环等待。修法正好对应上表：给叉子编号、每人先拿编号小的（锁顺序，破坏循环等待）；一次同时申请两把，拿不到就都不拿（破坏持有并等待）；或第二把拿不到就放下第一把（破坏不可剥夺）。

工具

cc -std=c17 -g -O1 -fsanitize=thread -pthread race.c -o race_tsan
./race_tsan
valgrind --tool=helgrind ./race

ThreadSanitizer 在编译期插桩，能报告数据竞争和锁顺序反转，开销约 5–15 倍；helgrind 不需要重编译，开销更大但能跑未插桩的二进制。两者都只能发现「执行到的」竞争，压测覆盖不到的路径仍可能漏。

伪共享（false sharing）是另一类性能问题：两个线程分别写同一缓存行里的不同变量，缓存一致性协议仍会把整行在核间来回搬。用 alignas(64) 或填充把热点变量分开，能显著减少无效流量。

1.12.7 事件驱动与 I/O 多路复用

线程池处理的是「任务」，事件循环处理的是「fd 就绪」。epoll 把大量 fd 注册到内核，内核在有事件时把就绪项放进链表，epoll_wait 只返回就绪的 fd，不需要遍历全部连接。

读图抓住三个结构：红黑树保存注册的 fd，就绪链表保存已触发的事件，等待队列让 epoll_wait 睡眠。epoll_ctl(ADD) 只注册一次；事件发生时，内核回调把 epitem 挂到就绪链表；epoll_wait 取链表，复杂度与「就绪数」相关，而不是「注册数」。

两种触发模式：

• 水平触发（level-triggered, LT）：只要 fd 还有数据，epoll_wait 就持续返回。编程简单，漏读会在下一次提醒。
• 边缘触发（edge-triggered, ET）：只在状态变化时返回一次。必须配合非阻塞 fd，一次读到 EAGAIN，否则剩余数据可能永远不再通知。

一个单线程 reactor 的结构（伪代码，省略错误处理）：

epfd = epoll_create1(0)
epoll_ctl(ADD, listen_fd, EPOLLIN)
loop:
    n = epoll_wait(epfd, events, MAX, -1)
    for e in events[0..n):
        if e.fd == listen_fd:
            c = accept4(listen_fd, SOCK_NONBLOCK)
            epoll_ctl(ADD, c, EPOLLIN | EPOLLET)
        else:
            read until EAGAIN
            process buffer
            write response (或注册 EPOLLOUT 等待可写)

事件循环的优点是「一个线程管一万条连接」，没有锁、没有线程切换；缺点是任何阻塞调用（磁盘 I/O、getaddrinfo、大文件读写）都会卡住整个循环。解决办法是把阻塞操作丢给线程池，或者用 io_uring 做异步 I/O。

真实的高性能服务器常把两者结合：epoll 线程负责收发包与协议解析，业务逻辑提交给线程池；线程池的回调再把结果写回 event loop。复杂度在于跨线程传递 fd/缓冲区，必须明确所有权与生命周期。

1.12.8 线程的完整生命周期：创建、分离、回收与退出

这一节回答三个问题：线程的资源由谁、在什么时候释放？pthread_* 失败怎么判断？哪些 API 的语义与直觉相反？

先记住一条与系统调用完全不同的约定：pthread_* 不设置 errno，而是直接返回错误码（成功返回 0）。perror("pthread_create") 打印的是上一个系统调用残留的 errno，经常显示 Success 或一个无关的错误，把排查带偏。正确写法是：

int rc = pthread_create(&tid, NULL, worker, arg);
if (rc != 0) {
    fprintf(stderr, "pthread_create: %s\n", strerror(rc));   /* 用返回值，不用 errno */
    return -1;
}

strerror 自身不保证线程安全（见 1.12.16 第 9 条），库代码应改用 strerror_r；调试输出可以接受。

表 1.12.1 是本章用到的线程 API 全表。返回值列只写真实会出现的错误，不是「失败」两个字。

线程的生命周期只有两条合法路径：

                 pthread_create
                        │
          ┌─────────────┴─────────────┐
          ▼                           ▼
     joinable（默认）             pthread_detach
          │                           │
   线程运行结束，资源保留        线程运行结束，资源立即回收
   （栈、TLS、返回值）           （不能 join，无返回值可取）
          │                           │
   pthread_join 回收 ◄───────────────┘
   （返回值写进 retval）
          │
   线程 ID 可能被下一个线程复用

读图抓住两点：joinable 线程退出后不是「没了」，而是留下一个待回收的内核/库资源；pthread_t 只在 join 之前有意义，join 之后可能被复用，把它当稳定身份存进哈希表是典型 bug。glibc 的栈是「虚拟分配、按需缺页」，所以 1000 个线程约占 8 GiB 虚拟地址而不是 8 GiB 物理内存；但每个线程的 guard page（默认 1 页）和内核栈是实打实的，容器里还要看 pids.max。

线程私有存储（thread-local storage, TLS）有三种写法：_Thread_local（C11，简单标量）、__thread（GNU 扩展，性能最好）、pthread_key_t（带析构函数，可在退出时释放堆内存，库代码首选）。

pthread_once 解决「初始化只做一次」：多线程同时第一次调用时，只有一个线程执行 init，其余阻塞到它完成。它比「检查一个 int inited 再赋值」正确，因为后者本身就是数据竞争。

join 与分离的差别可以现场验证：对一个已 pthread_detach 的线程调用 pthread_join，返回 EINVAL（22）；把 pthread_join 的错误用 perror 打印，会看到 Success——因为 errno 根本没被设置。pthread_once 则保证初始化函数在多个线程同时首次调用时也只执行一次，pthread_key_t 的析构函数在该线程退出时被调用。

取消（cancellation）要谨慎。默认是延迟取消：pthread_cancel 只置位，线程运行到取消点（read、write、accept、poll、pthread_cond_wait、sem_wait、nanosleep 等）才真正退出。取消会执行 pthread_cleanup_push 注册的清理函数，但不会替你解锁手工加的 mutex；在锁内被取消，锁永远不释放。工程上更常见的做法是「取消标志 + 条件变量」或直接让任务短小可丢弃，而不是异步取消。

1.12.9 原子操作与内存序：C11、C++ 与 Rust 的同一套规则

counter++ 的问题不是「加法不快」，而是「读—改—写」三步之间可以被插入。原子类型把这三步合成一个不可分割的操作，但原子性不等于顺序性：编译器与 CPU 仍可以重排原子操作与其他访存。C11 的内存模型用 memory_order 描述允许的重排，C++ 与 Rust 用的是同一套语义，只是拼写不同。

先看六个内存序。表 1.12.2 的「禁止的重排」一列是理解重点：它说的是「这个操作之前/之后的访存能不能越过它」。

三种语言的写法一一对应：

/* C11 */
atomic_int x = 0;
atomic_store_explicit(&x, 1, memory_order_release);
int v = atomic_load_explicit(&x, memory_order_acquire);

// C++20
std::atomic<int> x{0};
x.store(1, std::memory_order_release);
int v = x.load(std::memory_order_acquire);

#![allow(unused)]
fn main() {
// Rust 2021
use std::sync::atomic::{AtomicI32, Ordering};
let x = AtomicI32::new(0);
x.store(1, Ordering::Release);
let v = x.load(Ordering::Acquire);
}

release 存 + acquire 读，构成 synchronizes-with：读方看到 x == 1 时，写方在 store 之前的全部写入对读方可见。这就是「用标志发布数据」的全部秘密：

/* 发布者 */
data = 42;                                            /* 普通写 */
atomic_store_explicit(&ready, 1, memory_order_release);
/* 消费者 */
while (atomic_load_explicit(&ready, memory_order_acquire) == 0)
    ;                                                 /* 自旋等就绪 */
assert(data == 42);                                   /* 一定成立 */

把 release/acquire 换成 relaxed，assert 就可能失败——不是「概率低」，而是 C11 明确允许的行为。这也是 volatile 不能替代原子的原因：volatile 只禁止编译器缓存，不建立跨线程顺序，也不保证原子性。

seq_cst 是默认序，也是最贵的：x86-64 上 seq_cst 的 store 要 xchg 或 mfence，ARM64 上要 dmb ish，而 relaxed 的 RMW 只需一条 lock add/ldadd。所以「计数器只要最终值对」时用 relaxed；「用标志传递数据」时用 release/acquire；「多个标志之间要求全序」（Dekker、Peterson、双重检查锁）才用 seq_cst。

下面的程序把三种实现放在同一负载下测量，并验证原子类型是否真的无锁：

/* atomic_bench.c：relaxed / seq_cst / mutex 计数开销对照
 * cc -std=c17 -D_GNU_SOURCE -O2 -Wall -Wextra -Werror -pthread atomic_bench.c -o atomic_bench
 */
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <time.h>

#define NTHREADS 4
#define NLOOP    500000L

enum mode { M_RELAXED, M_SEQ_CST, M_MUTEX };

static atomic_long g_relaxed;
static atomic_long g_seqcst;
static long        g_mutex;
static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
static enum mode   g_mode;

static double now_ms(void)
{
    struct timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) != 0)
        return 0.0;
    return (double)ts.tv_sec * 1e3 + (double)ts.tv_nsec / 1e6;
}

static void *worker(void *arg)
{
    (void)arg;
    for (long i = 0; i < NLOOP; i++) {
        switch (g_mode) {
        case M_RELAXED:
            atomic_fetch_add_explicit(&g_relaxed, 1, memory_order_relaxed);
            break;
        case M_SEQ_CST:
            atomic_fetch_add_explicit(&g_seqcst, 1, memory_order_seq_cst);
            break;
        case M_MUTEX:
            if (pthread_mutex_lock(&mtx) != 0)
                return NULL;
            g_mutex++;
            pthread_mutex_unlock(&mtx);
            break;
        }
    }
    return NULL;
}

static int run(enum mode m, const char *name)
{
    pthread_t t[NTHREADS];
    g_mode = m;
    double t0 = now_ms();
    for (int i = 0; i < NTHREADS; i++) {
        int rc = pthread_create(&t[i], NULL, worker, NULL);
        if (rc != 0) {
            fprintf(stderr, "pthread_create: %d\n", rc);
            return -1;
        }
    }
    for (int i = 0; i < NTHREADS; i++)
        pthread_join(t[i], NULL);
    double ms = now_ms() - t0;
    long total = (m == M_MUTEX) ? g_mutex
               : (m == M_RELAXED) ? atomic_load(&g_relaxed)
                                  : atomic_load(&g_seqcst);
    printf("%-8s total=%ld  %.1f ms  %.2f ns/op\n",
           name, total, ms, ms * 1e6 / (double)total);
    return 0;
}

int main(void)
{
    printf("atomic_long lock-free: %d\n", (int)atomic_is_lock_free(&g_relaxed));
    if (run(M_RELAXED, "relaxed") != 0)
        return 1;
    if (run(M_SEQ_CST, "seq_cst") != 0)
        return 1;
    if (run(M_MUTEX, "mutex") != 0)
        return 1;
    return 0;
}

预期：三者最终值都是 NTHREADS * NLOOP = 2000000；relaxed 与 seq_cst 在 x86-64 上差距很小（RMW 都带 lock），在 ARM64 上 seq_cst 明显更慢；mutex 最慢，因为每次都可能有 futex 竞争与唤醒。注意这段测量本身受线程调度影响，只看数量级，不要当基准测试结论。

1.12.10 锁的内部：从自旋到 futex

上一节用锁，这一节看锁是怎么实现的。所有互斥锁都建立在一条硬件原语上：原子地「读旧值并写新值」。区别只在于「拿不到锁时怎么办」——自旋、让出、还是睡进内核。

表 1.12.3 是常见实现技术。注意 LL/SC 与 CAS 的差别：LL/SC 能避免 ABA，CAS 不能。

自旋锁的最小正确实现只要十几行，但「先读后抢」的顺序决定了性能：

/* spinlock.c 片段：TTAS 自旋锁（放进你的锁模块） */
#include <stdatomic.h>
#include <stdbool.h>

typedef struct {
    atomic_bool locked;
} spinlock_t;

#define SPINLOCK_INIT { .locked = false }

static void spin_lock(spinlock_t *s)
{
    for (;;) {
        while (atomic_load_explicit(&s->locked, memory_order_relaxed))
            ;                                   /* 只读自旋：不抢缓存行所有权 */
        if (!atomic_exchange_explicit(&s->locked, true, memory_order_acquire))
            return;
    }
}

static void spin_unlock(spinlock_t *s)
{
    atomic_store_explicit(&s->locked, false, memory_order_release);
}

两个细节：自旋等待用 relaxed 读，因为「看到 false」本身不发布数据；真正获得锁的 exchange 用 acquire，解锁用 release，这样临界区内的读写不会溢出到锁外。x86 的 pause、ARM 的 yield 指令能降低自旋功耗与流水线惩罚，C 里没有标准写法，通常用内联汇编或 sched_yield（后者代价大得多）。

用户态的 pthread_mutex 不是纯自旋锁。glibc 的实现是一个三态状态字加 futex：

   状态字：0 = 未加锁   1 = 已加锁无等待者   2 = 已加锁可能有等待者
   lock:   CAS 0→1 成功 → 进临界区（无系统调用）
           CAS 失败 → CAS 1→2（或已是 2），futex_wait(2)
   unlock: 原值为 1 → 写 0，结束；原值为 2 → 写 0 并 futex_wake(1)

读图的关键是「快路径不进内核」：无竞争时加锁/解锁只是两次用户态原子操作；只有出现第二个等待者，才把状态改成 2 并调用 futex(FUTEX_WAIT) 睡眠。futex 的设计要点是「内核只做等待队列，状态字由用户态维护」——FUTEX_WAIT 会在睡眠前原子地比较状态字，避免「检查后发现要睡，但唤醒已经发生」的丢失唤醒。这也是为什么 futex 不能单独当锁用：没有用户态状态字，它无法判断该不该睡。

pthread_mutex 有五种类型，选错会得到难以理解的返回值：

ROBUST 解决的是「持锁进程被 SIGKILL」：下一个加锁者拿到 EOWNERDEAD，检查并修复共享数据后调用 pthread_mutex_consistent 把锁恢复为可用；不修复就 pthread_mutex_unlock，锁会永久不可用。共享内存 + 多进程的场景必须考虑这一点。

自旋还是睡眠？判据是「临界区长度 vs 一次上下文切换」：切换约 1–5 µs，临界区只有几十条指令时自旋更快；临界区里可能阻塞（I/O、malloc、条件等待）时绝不能自旋。pthread_spin_lock 只适合「确定不会睡眠、且持有时间极短」的内核式临界区，用户态误用会把 CPU 烧满而毫无进展。

1.12.11 读写锁、seqlock 与 RCU：读多写少的三种路线

互斥锁把读也串行化。当共享数据的访问是「99% 读、1% 写」时，让读者并行能直接换来吞吐。这一节比较三条路线：读写锁（读者共享）、seqlock（读者不加锁但可能重读）、RCU（读者零开销，写者等宽限期）。

先厘清读写锁的语义：

默认策略是读者优先（PTHREAD_RWLOCK_PREFER_READER_NP）：只要还有读者，新读者就能进，写者排在后面。读请求源源不断时写者饥饿——这是「读者优先」的定义性后果，不是实现 bug。要写者优先，必须显式设置 kind；但写者优先又会反过来饿死读者，且 PREFER_WRITER_NONRECURSIVE_NP 禁止递归读锁（递归读会死锁）。生产代码里更稳的做法是「读写锁 + 写者到达时短暂阻止新读者」或者干脆用 RCU。

下面的程序在同一读密集负载下比较互斥锁与读写锁。关键设计：读循环里的 sum 必须被使用，否则优化器会把整个循环删掉；用原子累加 sink 保留结果。

/* rwlock_bench.c：读密集负载下 mutex 与 rwlock 的吞吐对照
 * cc -std=c17 -D_GNU_SOURCE -O2 -Wall -Wextra -Werror -pthread rwlock_bench.c -o rwlock_bench
 */
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <time.h>

#define NREADERS 4
#define ITER     1000000L

struct args { int use_rw; };

static pthread_mutex_t  mtx = PTHREAD_MUTEX_INITIALIZER;
static pthread_rwlock_t rw  = PTHREAD_RWLOCK_INITIALIZER;
static long data[64];
static atomic_long sink;

static double now_ms(void)
{
    struct timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) != 0)
        return 0.0;
    return (double)ts.tv_sec * 1e3 + (double)ts.tv_nsec / 1e6;
}

static void *reader(void *arg)
{
    int use_rw = ((const struct args *)arg)->use_rw;
    long sum = 0;

    for (long i = 0; i < ITER; i++) {
        if (use_rw) {
            if (pthread_rwlock_rdlock(&rw) != 0)
                return NULL;
            sum += data[i & 63];
            pthread_rwlock_unlock(&rw);
        } else {
            if (pthread_mutex_lock(&mtx) != 0)
                return NULL;
            sum += data[i & 63];
            pthread_mutex_unlock(&mtx);
        }
    }
    atomic_fetch_add_explicit(&sink, sum, memory_order_relaxed);   /* 保住结果 */
    return NULL;
}

static double run(int use_rw)
{
    struct args a = { use_rw };
    pthread_t t[NREADERS];
    double t0 = now_ms();

    for (int i = 0; i < NREADERS; i++) {
        if (pthread_create(&t[i], NULL, reader, &a) != 0)
            return -1.0;
    }
    for (int i = 0; i < NREADERS; i++)
        pthread_join(t[i], NULL);
    return now_ms() - t0;
}

int main(void)
{
    for (int i = 0; i < 64; i++)
        data[i] = i;

    double tm = run(0);
    double tr = run(1);
    double ops = (double)NREADERS * (double)ITER;

    printf("mutex : %7.1f ms  %6.1f Mops/s\n", tm, ops / tm / 1e3);
    printf("rwlock: %7.1f ms  %6.1f Mops/s\n", tr, ops / tr / 1e3);
    printf("speedup: %.2fx（读临界区越短，收益越小）\n", tm / tr);
    return 0;
}

结果依 libc 与临界区长度而变：临界区只有一次数组读时，两者测的都是锁本身，读写锁不一定占优（在 macOS 的 libc 上实测慢 4 倍；glibc 上通常快，但加速比远小于 4）。把临界区换成「遍历 4 KiB 数据」再测，读者并行的收益才会显现。这正是「读写锁的收益取决于读临界区长度与写比例」的定量版本——先测，再选。

seqlock：读者不加锁，代价是重读

seqlock 把「读者并行」推到极致：读者完全不加锁，只在读前读后各取一次序号；序号是奇数（写者正在写）或前后不一致，就重读。写者之间仍然互斥。

writer:  seq++（变奇数）; 写数据; seq++（变偶数）
reader:  do { s1 = seq; if (s1 & 1) continue;
              read data; s2 = seq; } while (s1 != s2);

适用条件很苛刻：数据小、可重复读、读者不睡眠、能接受「读到一半被改写就重来」。它不保证读者看到某一时刻的一致快照，只保证「不会把撕裂的数据当成有效值」。Linux 内核的 seqcount_t 就是这个结构，write_seqcount_begin/end 配 read_seqbegin/read_seqretry；用户态自己实现时，序号必须是原子变量，数据访问要用 atomic_thread_fence 或 READ_ONCE/WRITE_ONCE 语义约束，否则编译器重排会让校验失效。

RCU：读者零开销，写者等一个宽限期

RCU（read-copy-update）把「更新」拆成三步：复制新版本、原子替换指针、等所有旧读者离开后释放旧版本。读者侧只有「标记进入/退出」，没有原子 RMW、没有缓存行争用，因此读侧几乎零开销；代价落在写者：synchronize_rcu() 要等一个宽限期（grace period），期间所有已进入读临界区的线程都必须退出。

内核里的 RCU 读侧是 rcu_read_lock() / rcu_read_unlock()，写侧是 rcu_assign_pointer() 发布、synchronize_rcu() 等宽限期、kfree_rcu() 延迟释放。用户态对应的库是 liburcu（urcu/rcu.h：rcu_read_lock、rcu_dereference、rcu_assign_pointer、synchronize_rcu），每个线程要先 rcu_register_thread()。

RCU 的三条纪律，违反任何一条都会变成 use-after-free：

1. 读临界区内不能睡眠（经典 RCU）。睡眠的读者会无限拉长宽限期；需要睡眠的场景用 SRCU 或 sleepable RCU 变体。
2. 发布必须 release，读取必须 acquire。rcu_assign_pointer 与 rcu_dereference 负责这对顺序；用普通指针赋值，编译器可能把「初始化新对象」挪到「发布指针」之后。
3. 旧对象不能在替换后立刻 free。必须等宽限期；call_rcu 把释放回调挂到宽限期结束后执行。自己用引用计数实现时，计数增减的时机同样要落在正确的临界区里。

信号量版读者-写者：readcnt、mutex 与 write_sem

CSAPP 12.5.4 用三个信号量实现「读者优先」：

sem_t mutex;      /* 保护 readcnt */
sem_t write_sem;  /* 写者互斥，也挡住新读者 */
int readcnt = 0;

void reader(void) {
    sem_wait(&mutex);
    if (readcnt++ == 0) sem_wait(&write_sem);  /* 第一个读者锁住写者 */
    sem_post(&mutex);
    /* critical section: read */
    sem_wait(&mutex);
    if (--readcnt == 0) sem_post(&write_sem);  /* 最后一个读者放行写者 */
    sem_post(&mutex);
}

void writer(void) {
    sem_wait(&write_sem);
    /* critical section: write */
    sem_post(&write_sem);
}

这个版本写者会饥饿：只要读者不断到来，readcnt 永远不归零，写者永远等不到 write_sem。要公平，加一个 sem_t queue 让新读者排在写者后面（读者-写者公平版），代价是吞吐下降。

1.12.12 死锁：从进度图到证据链

1.12.6 给了死锁的四个条件和「锁顺序」这一条工程手段。这一节把它做成可复现、可取证、可修复的完整流程。

先看 CSAPP 12.5.1 的进度图（progress graph）：横轴是线程 1 的指令进度，纵轴是线程 2 的进度，每条轨迹是一次可能的交错执行。对共享变量的「读—改—写」会形成一个不安全区（unsafe region），轨迹进入它就可能产生错误结果；两个线程各自持有一把锁再申请对方那把，会形成一个矩形死锁区（deadlock region），轨迹一旦进入就再也出不来。进度图的价值是把「调度运气」变成几何：安全轨迹不进入不安全区，死锁轨迹进入死锁区。

工程上更常用的是可执行证据。下面的程序用 1 ms 的睡眠放大竞争窗口，制造稳定的 AB/BA 死锁；加 -DFIX 后改用 trylock + 退避，破坏「持有并等待」。

/* deadlock.c：AB/BA 锁序死锁；-DFIX 用 trylock + 退避修复
 * cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -pthread deadlock.c -o deadlock
 * timeout 3 ./deadlock; echo "exit=$?"     # 124 = 超时，说明死锁
 * cc -std=c17 -D_GNU_SOURCE -DFIX -Wall -Wextra -Werror -pthread deadlock.c -o deadlock_fix
 */
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <time.h>

static pthread_mutex_t lock_a = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t lock_b = PTHREAD_MUTEX_INITIALIZER;
static long shared;

static void msleep(long ms)
{
    struct timespec ts = { ms / 1000, (ms % 1000) * 1000000L };
    nanosleep(&ts, NULL);
}

static int lock_pair(pthread_mutex_t *first, pthread_mutex_t *second, int id)
{
#ifdef FIX
    for (int attempt = 0; attempt < 1000; attempt++) {
        if (pthread_mutex_lock(first) != 0)
            return -1;
        if (pthread_mutex_trylock(second) == 0)
            return 0;                        /* 两把都拿到 */
        pthread_mutex_unlock(first);         /* 拿不到第二把：全放，退避重试 */
        msleep(1 + (attempt + id) % 3);
    }
    return -1;
#else
    (void)id;
    if (pthread_mutex_lock(first) != 0)
        return -1;
    msleep(1);                               /* 放大窗口：让对方拿到 second */
    if (pthread_mutex_lock(second) != 0)
        return -1;
    return 0;
#endif
}

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

    for (int i = 0; i < 100; i++) {
        int rc = (id == 0) ? lock_pair(&lock_a, &lock_b, id)
                           : lock_pair(&lock_b, &lock_a, id);
        if (rc != 0) {
            fprintf(stderr, "thread %d: 放弃加锁\n", id);
            return NULL;
        }
        shared++;
        if (id == 0) {                       /* 按实际加锁的逆序解锁 */
            pthread_mutex_unlock(&lock_b);
            pthread_mutex_unlock(&lock_a);
        } else {
            pthread_mutex_unlock(&lock_a);
            pthread_mutex_unlock(&lock_b);
        }
    }
    printf("thread %d done\n", id);
    return NULL;
}

int main(void)
{
    pthread_t t[2];

    for (long i = 0; i < 2; i++) {
        int rc = pthread_create(&t[i], NULL, worker, (void *)(intptr_t)i);
        if (rc != 0) {
            fprintf(stderr, "pthread_create: %d\n", rc);
            return 1;
        }
    }
    for (int i = 0; i < 2; i++)
        pthread_join(t[i], NULL);

    printf("shared=%ld (expected 200)\n", shared);
    return 0;
}

取证分三步，缺一不可：

1. 确认卡住而不是算得慢：timeout 3 ./deadlock; echo $? 返回 124。ps -o stat,wchan -L -p <pid> 看到两个线程都是 S，wchan 为 futex_wait_queue。
2. 看它们各自持有什么、等什么：gdb -p <pid> -batch -ex 'thread apply all bt'，两个栈都停在 __lll_lock_wait，上一帧分别是 lock_pair 的第 1、2 次加锁。pstack <pid> 等价。
3. 确认锁序：在 lock_pair 里打印 id 与 first/second 地址，或直接用 TSan/helgrind 的锁序检查（TSAN_OPTIONS="second_deadlock_stack=1"）报告 lock-order-inversion。helgrind 会打印 Thread #1: lock order "0x... before 0x..." violated。

修复手段按代价从低到高：

std::scoped_lock（C++17）与 std::lock 内部用「try-and-backoff」一次锁多把，能避免 AB/BA；C 里没有标准等价物，只能自己写 lock_pair。注意「按地址排序加锁」也是全局锁序的一种：地址稳定（对象不移动）时可用，但堆对象被 realloc/移动后顺序会变，不如显式编号可靠。

死锁只是「永久阻塞」的一种。还有三种相邻故障要区分：活锁（线程都在跑但互相让路，没人推进）、饥饿（某个线程永远排不上队）、优先级反转（低优先级持锁、高优先级等待、中优先级插队）。优先级反转的经典案例是 Mars Pathfinder 的重启；修法是优先级继承（PTHREAD_PRIO_INHERIT）或优先级天花板。Linux 的 PI futex 支持优先级继承，但要求实时调度策略与相应权限，普通分时任务用不上。

1.12.13 线程池的容量、背压与关闭语义

1.12.5 的 pool.c 给了正确的骨架：有界队列、while 等待、shutdown 广播、join。这一节补三个工程问题：池开多大、队列满了怎么办、关闭时在途任务怎么算。

池开多大。设任务到达率 λ（个/秒）、每个任务平均停留时间 W（秒），Little 定律给出稳态并发数 L = λW。例如网关每秒 2 万请求、每个请求在池内平均 5 ms，则需要 L = 20000 × 0.005 = 100 个并发任务；若每个任务 90% 时间在等下游 I/O，CPU 密集部分只占 10%，线程数可以按 N ≈ 核数 × (1 + W_io/W_cpu) 估：4 核、I/O 等待是计算的 9 倍时约 40 个线程。两个公式都只是起点，最终要靠压测找拐点。

三个容易踩的环境差异：

• sysconf(_SC_NPROCESSORS_ONLN) 在容器里返回宿主机核数，不是 CPU 配额；要读 cgroup v2 的 /sys/fs/cgroup/cpu.max（quota period，如 200000 100000 表示 2 核）。
• RLIMIT_NPROC 与 cgroup pids.max 限制的是线程总数（Linux 线程即任务）；池大小 × 进程数不能超过限额，否则 pthread_create 返回 EAGAIN。
• 每个线程默认 8 MiB 虚拟栈；若真的开几百个线程，用 pthread_attr_setstacksize 降到 256 KiB–1 MiB，但要留足深递归与 alloca 的余量。

队列满了怎么办。这是策略问题，不是实现问题：

pool.c 选的是「阻塞提交」，适合「上游能等」的服务；面向用户的入口应选「拒绝 + 快速返回 503」，否则延迟会堆到超时。无论哪种，都要把队列长度暴露成指标：pool.count 的采样、等待时间的分位数（P50/P99）、拒绝次数。只看 CPU 利用率会漏掉「队列在涨、线程在睡」的假健康状态。

关闭语义。pool_shutdown 必须回答三个问题：还收不收新任务（不收）、队列里的任务跑不跑（跑完，叫 graceful drain）、等在 submit 上的调用者怎么办（唤醒并返回错误）。pool.c 的实现是「置 shutdown → 广播两个条件变量 → join」，pool_submit 在等待循环里检查 shutdown 并返回 -1，这是正确的最小版本。两个易错点：

• 广播而不是 signal：关闭要唤醒所有 worker，pthread_cond_signal 只唤醒一个，其余永远睡。
• shutdown 与「队列非空」的判断顺序：worker 的循环条件必须是 while (count == 0 && !shutdown)，退出条件必须是 count == 0 && shutdown；写成 if (shutdown) break; 会丢掉队列里剩余任务。

更强的关闭语义要支持「超时强制」：pthread_cond_timedwait 等到截止时间，仍未排空就返回未执行任务数，由调用者决定丢弃还是持久化。生产环境的「优雅关闭」通常还有一个宽限期（如 30 s），超过就 pthread_cancel 或直接 _exit，避免进程被容器编排器 SIGKILL。

强扩展、弱扩展与并行加速比

CSAPP 12.6 用两个指标描述「加核有没有用」：

• 强扩展（strong scaling）：问题规模固定，增加核数，看时间下降。受 Amdahl 限制：串行部分占比 s，加速比上限 1/s。
• 弱扩展（weak scaling）：问题规模随核数同比增加，看「每核处理量不变时时间是否不变」。受 Gustafson 影响，并行部分可摊薄。

/* speedup.c — 固定规模，测 P 个线程的加速比 */
double t1 = time_sum(n, 1);
for (int p = 2; p <= 8; p *= 2) {
    double tp = time_sum(n, p);
    printf("p=%d speedup=%.2f efficiency=%.2f\n", p, t1/tp, (t1/tp)/p);
}

效率 E = S(p)/p。若 E 随 p 快速下降，先查：串行比例、锁竞争、伪共享、内存带宽、负载不均。不要只看「核多」就假设更快。

基于锁的并发数据结构

OSTEP 第 29 章讲三类结构；本书在 1.12.5/1.12.11 讲了线程池队列与 RCU，这里补最小实现：

懒初始化链表：插入时只锁前驱，查找时无锁读 next（需要 atomic_load/memory_order_acquire）；删除要锁前驱与目标，并处理「被删节点还被读者持有」的回收问题（hazard pointer/RCU）。

hand-over-hand locking：遍历时先锁下一个节点，再释放当前节点；并发度高，但每次跳转两次锁，开销大，实际不如「每桶一把锁」的哈希表。

并发哈希表：固定桶数组，每桶一把 pthread_mutex_t；hash(key)%N 选桶，桶内链表加锁。这是最实用的「基于锁」结构：简单、可扩展、无全局锁。

#define NBUCKET 64
struct node { int key; struct node *next; };
struct bucket { pthread_mutex_t lock; struct node *head; };
struct table { struct bucket b[NBUCKET]; };

int table_get(struct table *t, int key)
{
    int i = (unsigned)key % NBUCKET;
    pthread_mutex_lock(&t->b[i].lock);
    for (struct node *p = t->b[i].head; p; p = p->next)
        if (p->key == key) { pthread_mutex_unlock(&t->b[i].lock); return 1; }
    pthread_mutex_unlock(&t->b[i].lock);
    return 0;
}

1.12.14 三种语言的并发原语对照：C11、C++20 与 Rust

同一套内存模型，三种语言给了三种「防错」方式：C11 只给原语，什么都不查；C++ 用 RAII 管生命周期，锁序仍靠人；Rust 用类型系统在编译期拒绝数据竞争。

/* C11 <threads.h>：glibc ≥ 2.28 提供；macOS 的 libc 没有这个头 */
#include <stdio.h>
#include <threads.h>

static mtx_t mtx;
static long counter;

static int worker(void *arg)
{
    (void)arg;
    for (long i = 0; i < 1000000; i++) {
        if (mtx_lock(&mtx) != thrd_success) return 1;
        counter++;
        mtx_unlock(&mtx);
    }
    return 0;
}

int main(void)
{
    thrd_t t[4];
    if (mtx_init(&mtx, mtx_plain) != thrd_success) return 1;
    for (int i = 0; i < 4; i++)
        if (thrd_create(&t[i], worker, NULL) != thrd_success) return 1;
    for (int i = 0; i < 4; i++) thrd_join(t[i], NULL);
    printf("counter = %ld\n", counter);
    mtx_destroy(&mtx);
    return 0;
}

// C++20：scoped_lock 一次锁多把，内部 try-and-backoff 避免 AB/BA
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

int main()
{
    long counter = 0;
    std::mutex mtx;
    std::vector<std::thread> pool;
    for (int i = 0; i < 4; ++i)
        pool.emplace_back([&] {
            for (long n = 0; n < 1000000; ++n) {
                std::scoped_lock lock(mtx);
                ++counter;
            }
        });
    for (auto &t : pool) t.join();
    std::cout << "counter = " << counter << '\n';
}

// Rust 2021：thread::scope 允许借用栈上数据，无需 Arc
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

fn main() {
    let counter = AtomicU64::new(0);
    thread::scope(|s| {
        for _ in 0..4 {
            s.spawn(|| {
                for _ in 0..1_000_000 {
                    counter.fetch_add(1, Ordering::Relaxed);
                }
            });
        }
    });
    println!("counter = {}", counter.load(Ordering::Relaxed));
}

差异不在语法，在错误被谁发现：C 里忘记 mtx_unlock 就是死锁，编译器不管，且 <threads.h> 的可用性依赖 libc（glibc/musl 有，macOS 没有）；C++ 的 lock_guard 在析构时解锁，异常路径安全，但两把锁的顺序仍靠人；Rust 的 Mutex<T> 把数据关进锁里，不拿锁拿不到 T，Rc 不是 Send、RefCell 不是 Sync，编译期就拒绝——代价是全局状态写起来更绕，且一旦出现 unsafe，保证立即失效。

1.12.15 并发调试：TSan、helgrind 与 perf

数据竞争的「测试通过」不代表没有：它依赖调度，可能一万次才出现一次。工具把碰运气变成可复现。

实验一：用 TSan 抓住 race.c 的竞争

cc -std=c17 -D_GNU_SOURCE -g -O1 -fsanitize=thread -pthread race.c -o race_tsan
TSAN_OPTIONS="halt_on_error=1 second_deadlock_stack=1" ./race_tsan

预期输出（地址与线程号会变）：

WARNING: ThreadSanitizer: data race (pid=...)
  Write of size 8 at 0x... by thread T2:
    #0 worker /tmp/race.c:22 (race_tsan+0x...)
  Previous write of size 8 at 0x... by thread T1:
    #0 worker /tmp/race.c:22 (race_tsan+0x...)
  Location is global 'counter' ... (race_tsan+0x...)

读法：TSan 维护每个内存字的「访问历史 + happens-before」，报告一定给出两次访问及各自调用栈；Location is global 'counter' 直接指认变量。看到报告先查两次访问之间有没有 release/acquire 或锁，而不是怀疑工具。修法是加锁或 atomic_fetch_add，重编译重跑，报告消失。

三个限制：TSan 与 ASan 不能同时开；只检查执行到的路径；推荐 -O1 -g。Linux 上地址随机化影响复现时，用 setarch -R ./race_tsan 关 ASLR。

helgrind 不需重编译（Linux）：valgrind --tool=helgrind ./race，关键行是 Possible data race during write of size 8、Locks held: none、This conflicts with a previous write；锁序反转会打印 lock order ... violated。它慢 20–100 倍，但能跑未插桩的二进制。

实验二：伪共享与线程身份

cat > fs.c <<'EOF'
#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>
#include <time.h>

#define ITER 20000000L

struct shared {
#ifdef ALIGN
    _Alignas(64) long a;
    _Alignas(64) long b;
#else
    long a;
    long b;
#endif
};

static struct shared s;

static void *bump_a(void *p) { (void)p; for (long i = 0; i < ITER; i++) s.a++; return NULL; }
static void *bump_b(void *p) { (void)p; for (long i = 0; i < ITER; i++) s.b++; return NULL; }

int main(void)
{
    struct timespec t0, t1;
    pthread_t x, y;
    clock_gettime(CLOCK_MONOTONIC, &t0);
    if (pthread_create(&x, NULL, bump_a, NULL) != 0 ||
        pthread_create(&y, NULL, bump_b, NULL) != 0)
        return 1;
    pthread_join(x, NULL);
    pthread_join(y, NULL);
    clock_gettime(CLOCK_MONOTONIC, &t1);
    printf("%s: %.1f ms a=%ld b=%ld\n",
#ifdef ALIGN
           "alignas(64)",
#else
           "same line",
#endif
           (t1.tv_sec - t0.tv_sec) * 1e3 + (t1.tv_nsec - t0.tv_nsec) / 1e6, s.a, s.b);
    return 0;
}
EOF
cc -std=c17 -D_GNU_SOURCE -O2 -Wall -Wextra -Werror -pthread fs.c -o fs_same
cc -std=c17 -D_GNU_SOURCE -O2 -Wall -Wextra -Werror -pthread -DALIGN fs.c -o fs_align
./fs_same; ./fs_align

预期：same line 慢 2–5 倍。原因不是锁，而是缓存一致性：两个核各持同一缓存行的独占副本，每次写都让对方失效（cache line ping-pong）；_Alignas(64) 把 a、b 分到不同行，流量消失。x86-64 与多数 ARM64 的缓存行是 64 B。

用 perf 量化（Linux）：

perf stat -e cache-references,cache-misses,context-switches ./fs_same
perf c2c record -o /tmp/c2c.data ./fs_same && perf c2c report -i /tmp/c2c.data --stdio | head -40

perf c2c 的 HITM 列直接显示「某缓存行被多核争用」，same line 版本会看到高 HITM，alignas 版本消失。没有 perf c2c 时，退一步看 perf stat 的 cache-misses 与 context-switches 差值。

把线程与内核 TID 对上号，才知道谁在抢：

./fs_same & PID=$!
ps -L -o pid,tid,psr,stat,pcpu,comm -p $PID
gdb -p $PID -batch -ex 'info threads' -ex 'thread apply all bt' -ex 'detach'

ps -L 的 TID 就是 gettid() 的返回值，PSR 是最后运行的 CPU；/proc/$PID/task/<tid>/stat 的第 39 字段是 processor，第 41、42 字段是自愿/非自愿上下文切换次数（comm 含空格时字段会移位）。把「在哪个核、切了多少次」与「缓存未命中」放在一起，才能区分锁竞争与伪共享。

1.12.16 九条并发陷阱

1.12.17 本章出口检查

母本对应

CSAPP 第 12 章小节清单（已覆盖＝独立展开；合并＝并入相邻节；背景＝只给上下文）

OSTEP 第 25–34 章小节清单

下一站：把进程、虚拟内存、文件、网络、并发放在一张桌上，操作系统的三大主题——虚拟化、并发、持久化——各自解决什么问题？李治军的 8 个实践项目又怎样把这些机制从零实现一遍？进入 1.13 操作系统导论。