2.13 线程同步与线程安全

2.12 的线程池已经出现了 pthread_mutex_t 和 pthread_cond_t，但当时只讲了「怎么用」。这一章回答「为什么」：为什么共享变量必须加锁，为什么条件变量醒来后要重新检查条件，为什么两个锁的顺序能让整个进程卡死。

同步的本质不是「加锁」，而是建立 happens-before 关系。没有同步时，编译器和 CPU 都可以重排你的读写；有了正确的锁或原子操作，你才能保证「A 线程写的值，B 线程一定看得到，而且看得到完整的值」。C 语言的内存模型从 C11 起才正式定义这件事；在此之前，pthread 的锁被当作编译器不会跨越的屏障。这一章讲的每一条纪律，最终都能还原成「哪两个操作之间建立了 happens-before」。

这一章的顺序是「问题 → 原语 → 失效模式」：先用一个丢失更新的计数器说明竞态，再依次讲互斥锁、条件变量、信号量、读写锁与屏障；然后集中处理死锁、惊群、优先级反转；最后讲线程安全函数与 TSan/helgrind。读完你应该能判断：这段代码该用哪种原语，以及它会在什么条件下出错。

2.13.1 竞态与临界区：丢失的更新

/* racy_counter.c — 两个线程各加 1000 万次 */
#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define N 10000000L

static long counter;                    /* 全局变量：两个线程共享 */

static void *bump(void *arg)
{
    (void)arg;
    for (long i = 0; i < N; i++)
        counter++;                      /* 读-改-写，三步不是原子的 */
    return NULL;
}

int main(void)
{
    pthread_t a, b;
    int rc = pthread_create(&a, NULL, bump, NULL);
    if (rc != 0) {
        fprintf(stderr, "pthread_create: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }
    rc = pthread_create(&b, NULL, bump, NULL);
    if (rc != 0) {
        fprintf(stderr, "pthread_create: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }

    if (pthread_join(a, NULL) != 0 || pthread_join(b, NULL) != 0) {
        fprintf(stderr, "pthread_join failed\n");
        return EXIT_FAILURE;
    }

    printf("counter=%ld expected=%ld\n", counter, 2 * N);
    return EXIT_SUCCESS;
}

在 x86-64 上编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread racy_counter.c -o racy_counter
./racy_counter
# 典型输出：counter=10485763 expected=20000000

counter++ 在汇编层是三条指令：load、add、store。两个线程可能同时读到 100，各自加 1 写回 101，两次增量只生效一次。这就是丢失更新（lost update）。

 时间 ─────────────────────────────────────────────▶
 线程 A:   load counter(=100) ── add 1 ── store 101
 线程 B:        load counter(=100) ── add 1 ── store 101
                                   ▲
                          两次写都基于同一个旧值，一次增量被覆盖

读图：问题不在于「谁先写」，而在于「读和写之间被插入」。只要 load 与 store 之间没有别的线程修改，结果就对；一旦交错，就丢。临界区要保护的正是这段「读-改-写」区间。

修复方式一：互斥锁把「读-改-写」包成临界区。

static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static long counter;

static void *bump(void *arg)
{
    (void)arg;
    for (long i = 0; i < N; i++) {
        int rc = pthread_mutex_lock(&lock);
        if (rc != 0) { /* 记录并退出 */ }
        counter++;
        rc = pthread_mutex_unlock(&lock);
        if (rc != 0) { /* 记录并退出 */ }
    }
    return NULL;
}

修复方式二：用 C11 原子操作，让硬件保证「读-改-写」不可分割——把 counter 声明为 atomic_long，循环体改为 atomic_fetch_add_explicit(&counter, 1, memory_order_relaxed)。memory_order_relaxed 只保证这个变量自身的原子性，不建立与其他变量的顺序。计数器、统计量用它足够；如果这个值要用来「发布」另一块数据（例如 ready 标志保护 data），必须用 release/acquire 或互斥锁。原子操作不是「无锁就更快」的万能药：高竞争下，缓存行在核心之间来回弹跳，可能比一把锁还慢。

2.13.2 互斥锁：语义、类型与陷阱

int pthread_mutex_init(pthread_mutex_t *m, const pthread_mutexattr_t *attr);
int pthread_mutex_lock(pthread_mutex_t *m);
int pthread_mutex_trylock(pthread_mutex_t *m);          /* 忙时返回 EBUSY */
int pthread_mutex_timedlock(pthread_mutex_t *m, const struct timespec *abs);
int pthread_mutex_unlock(pthread_mutex_t *m);
int pthread_mutex_destroy(pthread_mutex_t *m);

所有函数成功返回 0，失败返回错误码。静态初始化用 PTHREAD_MUTEX_INITIALIZER，适合全局锁；动态初始化用 pthread_mutex_init，适合堆上或结构体里的锁。

四种互斥锁类型

设置类型要经过属性对象：

pthread_mutexattr_t attr;
pthread_mutex_t lock;

pthread_mutexattr_init(&attr);                       /* 每个调用都要检查返回值 */
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
pthread_mutex_init(&lock, &attr);
pthread_mutexattr_destroy(&attr);

递归锁不是「更安全」，而是「把设计问题藏起来」：一个函数在持锁时调用另一个也要加同一把锁的函数，说明临界区边界没划清。递归锁还要求解锁次数与加锁次数严格相等，漏一次就永久锁死。除非在维护无法改结构的旧代码，否则用 ERRORCHECK 找出重复加锁，再重构。

timedlock 用的是墙上时钟

pthread_mutex_timedlock 的 abs_timeout 默认基于 CLOCK_REALTIME：系统时间被调后，你可能多等；被调前，你可能立刻超时。glibc 2.30 起提供 pthread_mutex_clocklock，可以显式指定 CLOCK_MONOTONIC。条件变量的 pthread_cond_timedwait 也一样，可以用 pthread_condattr_setclock(CLOCK_MONOTONIC) 改掉默认时钟。

锁的粒度

锁保护的是数据，不是代码。一个常见的错误是「函数入口加锁、出口解锁」，把整个函数变成临界区，包括其中的 I/O、malloc、日志。正确的做法是：

1. 找出所有访问共享数据的语句；
2. 只把这些语句放进临界区；
3. 在临界区外做 I/O、内存分配、回调；
4. 保证每条加锁路径都有对应的解锁路径（包括错误分支）。

反例是在持锁状态下 snprintf + write 日志：write 可能阻塞，其他线程全被拖住。正例是锁内只把 shared 拷到一个局部变量，解锁后再格式化与写盘。锁保护的是数据访问，不是整个函数。

互斥锁属性：五组开关决定它的行为

pthread_mutexattr_t 比大多数人以为的复杂。glibc 里 pthread_mutex_t 是 40 字节，其中大部分字段由属性决定：

pthread_mutexattr_t a;
pthread_mutex_t m;

pthread_mutexattr_init(&a);
pthread_mutexattr_settype(&a, PTHREAD_MUTEX_ERRORCHECK);      /* 调试期 */
pthread_mutexattr_setpshared(&a, PTHREAD_PROCESS_PRIVATE);
pthread_mutexattr_setrobust(&a, PTHREAD_MUTEX_STALLED);
pthread_mutexattr_setprotocol(&a, PTHREAD_PRIO_NONE);
pthread_mutex_init(&m, &a);
pthread_mutexattr_destroy(&a);

两个属性需要展开：

PTHREAD_MUTEX_ROBUST：持有者线程/进程死亡时，内核的 exit_robust_list 把锁标成 FUTEX_OWNER_DIED，下一个 lock 返回 EOWNERDEAD。它只对「跨进程共享」和「可能崩溃的线程」有意义；单进程内线程崩溃通常整个进程都没了，robust 也救不回来。详见 2.15。

PTHREAD_PRIO_INHERIT：需要 SCHED_FIFO/SCHED_RR 才有意义，且要求内核支持 PI futex（CONFIG_FUTEX_PI，主流发行版默认开）。设置了它但线程仍是 SCHED_OTHER 时不会报错，只是不产生任何效果——这是「优先级反转没解决」的常见原因。

互斥锁的实际成本

pthread_mutex_lock 在无竞争时不进内核。glibc 的 fast path 是一条内联的原子比较交换：

无竞争：lock cmpxchg 成功 → 返回 0            （约 20 个周期）
有竞争：lock cmpxchg 失败 → __lll_lock_wait → futex(FUTEX_WAIT)  （微秒级 + 上下文切换）

TLPI 在同一台机器上的对照（第 30.1.3 节）：fcntl 文件锁加解锁 2000 万次约 44 秒，System V 信号量约 28 秒，而 futex 互斥量因为不进内核，通常要快一到两个数量级。差距不在「锁算法」，而在「有没有系统调用」。

这解释了两个常见现象：

• 「加锁很便宜」只在无竞争时成立。竞争一旦发生，成本从纳秒跳到微秒，因为线程要进内核睡眠、被唤醒、重新调度。
• 「改成无锁就一定快」也不对。高竞争下 CAS 失败重试同样烧 CPU，缓存行在核间来回弹跳（cache line ping-pong），吞吐可能比一把锁更低。

glibc 还提供一个非标准类型 PTHREAD_MUTEX_ADAPTIVE_NP：先自旋一小段时间（约几十到几百次），还拿不到才进 futex。它适合「临界区极短、竞争持续时间小于一次上下文切换」的场景，例如分配器的 arena 锁。但它不可移植，且自旋次数依赖 CPU 频率与负载，不要把它当成通用优化。

/* 差：持锁做 I/O 与格式化 */
pthread_mutex_lock(&log_lock);
snprintf(buf, sizeof buf, "%ld %s\n", now_ms(), msg);
write(log_fd, buf, strlen(buf));        /* 可能阻塞在磁盘/网络 */
pthread_mutex_unlock(&log_lock);
/* 好：锁内只做入队，写盘交给专门的线程 */
pthread_mutex_lock(&log_lock);
queue_push(&log_queue, msg);            /* 几条指令 */
pthread_mutex_unlock(&log_lock);

2.13.3 条件变量：为什么必须用 while

互斥锁解决「不能同时访问」，但不解决「条件不满足时该干什么」。消费者发现队列为空，可以解锁后忙等（浪费 CPU），也可以睡到「有数据」再醒——这就是条件变量。

        生产者                         消费者
          │                              │
   lock(mutex)                    lock(mutex)
   while (count == CAP)           while (count == 0)
       cond_wait(not_full)            cond_wait(not_empty)
   buf[tail] = item               item = buf[head]
   count++                        count--
   signal(not_empty)              signal(not_full)
   unlock(mutex)                  unlock(mutex)
          │                              │
          └──────── 同一把 mutex ────────┘
              cond 只是等待队列，数据仍由 mutex 保护

读图：条件变量本身不保存状态，状态在 count、head、tail 里，由互斥锁保护。wait 原子地「释放锁 + 睡眠」，避免了「检查到队列空 → 解锁 → 睡下」之间生产者插入并 signal 的丢失唤醒窗口。但醒来后锁已被重新获取，条件可能又变了，所以必须重新检查。

完整的有界队列

/* bounded_queue.c — 1 个生产者、3 个消费者、容量 8 的环形队列 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#define QUEUE_CAP 8
#define ITEMS     40
#define CONSUMERS 3

struct queue {
    int buf[QUEUE_CAP];
    size_t head;
    size_t tail;
    size_t count;
    int done;                           /* 生产者结束标志 */
    pthread_mutex_t lock;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
};
static int q_init(struct queue *q)
{
    memset(q, 0, sizeof *q);

    int rc = pthread_mutex_init(&q->lock, NULL);
    if (rc != 0) return rc;
    rc = pthread_cond_init(&q->not_empty, NULL);
    if (rc != 0) { pthread_mutex_destroy(&q->lock); return rc; }
    rc = pthread_cond_init(&q->not_full, NULL);
    if (rc != 0) {
        pthread_cond_destroy(&q->not_empty);
        pthread_mutex_destroy(&q->lock);
        return rc;
    }
    return 0;
}
static void q_destroy(struct queue *q)
{
    pthread_cond_destroy(&q->not_full);
    pthread_cond_destroy(&q->not_empty);
    pthread_mutex_destroy(&q->lock);
}
/* 返回 0 成功，-1 失败（errno 已设置） */
static int q_put(struct queue *q, int v)
{
    int rc = pthread_mutex_lock(&q->lock);
    if (rc != 0) { errno = rc; return -1; }

    while (q->count == QUEUE_CAP && !q->done) {
        rc = pthread_cond_wait(&q->not_full, &q->lock);
        if (rc != 0) {
            pthread_mutex_unlock(&q->lock);
            errno = rc;
            return -1;
        }
    }
    if (q->done) {                      /* 已被通知关闭 */
        pthread_mutex_unlock(&q->lock);
        errno = ECANCELED;
        return -1;
    }

    q->buf[q->tail] = v;
    q->tail = (q->tail + 1) % QUEUE_CAP;
    q->count++;

    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->lock);
    return 0;
}
/* 返回 1 取到数据，0 表示正常结束，-1 表示错误 */
static int q_get(struct queue *q, int *out)
{
    int rc = pthread_mutex_lock(&q->lock);
    if (rc != 0) { errno = rc; return -1; }

    while (q->count == 0 && !q->done) {
        rc = pthread_cond_wait(&q->not_empty, &q->lock);
        if (rc != 0) {
            pthread_mutex_unlock(&q->lock);
            errno = rc;
            return -1;
        }
    }
    if (q->count == 0 && q->done) {     /* 队列空且生产者已结束 */
        pthread_mutex_unlock(&q->lock);
        return 0;
    }

    *out = q->buf[q->head];
    q->head = (q->head + 1) % QUEUE_CAP;
    q->count--;

    pthread_cond_signal(&q->not_full);
    pthread_mutex_unlock(&q->lock);
    return 1;
}
static void q_finish(struct queue *q)
{
    int rc = pthread_mutex_lock(&q->lock);
    if (rc != 0) return;
    q->done = 1;
    pthread_cond_broadcast(&q->not_empty);  /* 唤醒全部消费者 */
    pthread_cond_broadcast(&q->not_full);
    pthread_mutex_unlock(&q->lock);
}
static void *producer(void *arg)
{
    struct queue *q = arg;
    for (int i = 1; i <= ITEMS; i++) {
        struct timespec ts = { .tv_nsec = 2 * 1000000L };
        nanosleep(&ts, NULL);               /* 模拟生产耗时 */
        if (q_put(q, i) == -1) {
            fprintf(stderr, "q_put: %s\n", strerror(errno));
            break;
        }
    }
    q_finish(q);
    return NULL;
}
static void *consumer(void *arg)
{
    struct queue *q = arg;
    long sum = 0;
    int n = 0;
    for (;;) {
        int v;
        int rc = q_get(q, &v);
        if (rc == -1) {
            fprintf(stderr, "q_get: %s\n", strerror(errno));
            break;
        }
        if (rc == 0)
            break;                          /* 正常结束 */
        sum += v;
        n++;
    }
    printf("consumer %lu: n=%d sum=%ld\n",
           (unsigned long)pthread_self(), n, sum);
    return NULL;
}
int main(void)
{
    struct queue q;
    int rc = q_init(&q);
    if (rc != 0) {
        fprintf(stderr, "q_init: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }

    pthread_t prod, cons[CONSUMERS];
    rc = pthread_create(&prod, NULL, producer, &q);
    if (rc != 0) {
        fprintf(stderr, "create producer: %s\n", strerror(rc));
        q_destroy(&q);
        return EXIT_FAILURE;
    }
    for (int i = 0; i < CONSUMERS; i++) {
        rc = pthread_create(&cons[i], NULL, consumer, &q);
        if (rc != 0) {
            fprintf(stderr, "create consumer %d: %s\n", i, strerror(rc));
            q_finish(&q);
            pthread_join(prod, NULL);
            for (int j = 0; j < i; j++)
                pthread_join(cons[j], NULL);
            q_destroy(&q);
            return EXIT_FAILURE;
        }
    }

    pthread_join(prod, NULL);
    for (int i = 0; i < CONSUMERS; i++)
        pthread_join(cons[i], NULL);

    q_destroy(&q);
    return EXIT_SUCCESS;
}

运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread bounded_queue.c -o bounded_queue
./bounded_queue
# 三个消费者的 n 之和为 40，sum 为 1+2+...+40=820（顺序不定）

为什么是 while 而不是 if

假设消费者写成：

if (q->count == 0)
    pthread_cond_wait(&q->not_empty, &q->lock);
/* 醒来后直接取数据 */

三种情况会让它出错：

1. 虚假唤醒：POSIX 允许 cond_wait 在没有 signal 的情况下返回。if 只检查一次，醒来后队列仍空，取到垃圾。
2. 唤醒被抢走：两个消费者同时被 broadcast 唤醒，A 先拿到锁取走唯一的数据，B 再拿到锁时 count 又是 0。if 不重新检查，B 越界读。
3. 条件被改回：生产者放入数据后立刻又取走（或另一个消费者取走），等待者醒来时条件已不成立。

while 的语义是「只要条件不成立就继续等」，每一次醒来都重新验证。这不是防御性编程，而是使用条件变量的强制契约。

条件变量属性与「虚假唤醒」的真实来源

int pthread_condattr_init(pthread_condattr_t *a);
int pthread_condattr_setpshared(pthread_condattr_t *a, int pshared);
int pthread_condattr_setclock(pthread_condattr_t *a, clockid_t clock);
int pthread_cond_init(pthread_cond_t *c, const pthread_condattr_t *a);

三个属性里只有 setclock 是日常会用到的：pthread_cond_timedwait 默认用 CLOCK_REALTIME，系统时间被 NTP 往回调时，超时可能被推迟；往前跳时，可能立刻返回 ETIMEDOUT。改成单调时钟：

pthread_condattr_t ca;
pthread_condattr_init(&ca);
pthread_condattr_setclock(&ca, CLOCK_MONOTONIC);
pthread_cond_init(&cv, &ca);
pthread_condattr_destroy(&ca);

/* 超时也必须用同一个时钟计算，否则语义错位 */
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
ts.tv_sec += 1;                                  /* 1 秒后 */
pthread_cond_timedwait(&cv, &m, &ts);

glibc 2.30 起还有 pthread_cond_clockwait，可以每次调用时指定时钟，不必改条件变量属性。

虚假唤醒（spurious wakeup）不是 bug，是契约。 它的来源有四个：

所以「醒来 = 条件成立」这个假设在任何实现上都不成立。while 不是防御式编程，是使用条件变量的强制契约。

signal 与 broadcast 的选择

两者不是「保守 vs 激进」，而是对应两种状态变化：

• 状态变化只允许一个等待者推进（队列里多了一个元素、一个连接就绪）→ signal；
• 状态变化允许多个等待者同时推进（关闭标志置位、缓冲区从空变满、配置热更新）→ broadcast。

用 broadcast 代替 signal 的代价是惊群：N 个线程被唤醒、抢同一把锁、只有一个推进，其余重新睡。每次多出 N−1 次上下文切换。用 signal 代替 broadcast 的代价更严重：丢失唤醒，剩下的等待者永远睡下去。

一个折中模式是「一个条件变量 + 状态代数」：

/* 等待「版本号变化」，而不是等待某个布尔量 */
unsigned long seen = 0;

pthread_mutex_lock(&m);
while (version == seen)
    pthread_cond_wait(&cv, &m);
seen = version;
pthread_mutex_unlock(&m);

生产者每次修改共享状态就 version++ 并 broadcast。这样即使多个消费者需要看到同一次变化，也不会因为「谁先醒」而漏掉。

2.13.4 信号量：计数与跨进程同步

信号量是一个非负计数器，sem_wait 减一（为 0 时阻塞），sem_post 加一并唤醒等待者。

#include <semaphore.h>

sem_t sem;
sem_init(&sem, 0, 1);            /* pshared=0：线程间；value=1：二值信号量 */
sem_wait(&sem);                  /* P 操作；可能被信号打断，返回 -1/EINTR */
sem_post(&sem);                  /* V 操作；async-signal-safe */
sem_destroy(&sem);

信号量最常见的两种用法：

• 二值信号量当事件标志：中断处理器/信号处理器里 sem_post，工作线程 sem_wait。这是把异步事件转成同步等待的经典手法。
• 计数信号量做资源池：初始化 N，每个线程取资源前 sem_wait，归还时 sem_post。不需要额外的条件变量和计数变量。

sem_getvalue 返回当前值，但这个值立即可能过时：返回值只适合监控，不能用来做「先检查再操作」的判断。Linux 的 futex 实现里，等待者的数量会以负数形式编码，sem_getvalue 可能返回 0 或负数；POSIX 只保证「返回值反映某个瞬间」。

命名信号量用 sem_open("/name", O_CREAT, 0600, 1) 创建，sem_close 关闭，sem_unlink 删除名字。名字在 /dev/shm/sem.name；跨进程同步时，pshared=1 的未命名信号量必须放在共享内存里（见 2.15）。

信号量的内部表示与超时时钟

glibc 的 sem_t 在 x86-64 上是 32 字节：一个 futex 字加一个计数。sem_wait 的 fast path 是一条原子减：

sem_wait:  原子地把 count 减 1；结果 >= 0 → 直接返回
           结果为负 → 把 count 加回，futex(FUTEX_WAIT) 睡眠
sem_post:  原子地把 count 加 1；若有等待者 → futex(FUTEX_WAKE, 1)

这带来三个可观测的后果：

1. pshared 决定 futex 用私有还是共享。pshared=0 时 glibc 用 FUTEX_WAIT_PRIVATE，只在本进程的 mm 内匹配；pshared=1 用 FUTEX_WAIT（共享），可以跨进程。把 pshared=0 的 sem_t 放进共享内存给另一个进程用，对方 sem_wait 直接返回，不会等——因为 futex 的等待队列根本不在一个命名空间里。
2. sem_getvalue 可能返回 0 或负数。POSIX 允许「有等待者时返回值无意义」；Linux 的实现里等待者信息编码在计数中，所以拿它做 if (sem_getvalue() > 0) sem_wait() 一定错。
3. sem_timedwait 默认用 CLOCK_REALTIME，且参数是绝对时间。glibc 2.30 起有 sem_clockwait，可以用 CLOCK_MONOTONIC。

/* 绝对超时 + 单调时钟，避免系统时间跳变 */
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
ts.tv_sec += 2;

int rc;
while ((rc = sem_clockwait(&sem, CLOCK_MONOTONIC, &ts)) == -1 && errno == EINTR)
    ;

if (rc == -1 && errno == ETIMEDOUT)
    fprintf(stderr, "acquire timeout\n");

sem_wait 是取消点，被取消时不消耗计数；sem_post 是 async-signal-safe，可以在信号处理器里调用。这一对不对称性决定了「信号处理器只 post，工作线程 wait」的固定模式。

读写锁的属性与饥饿

int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *a, int kind);
/* kind: PTHREAD_RWLOCK_PREFER_READER_NP（默认）
 *       PTHREAD_RWLOCK_PREFER_WRITER_NP
 *       PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP */

默认策略是读者优先：只要有读者持锁，新来的读者就能进。写者会一直等，直到「当前所有读者都退出」且「没有新读者进来」——在读请求持续不断时，写者可能永久饥饿。这是「读写锁比互斥锁快」这句话最常被忽略的前提。

PREFER_WRITER_NONRECURSIVE_NP 让写者排队优先，代价是「递归读锁」会死锁（因此名字里带 NONRECURSIVE）：线程持有读锁时再申请读锁，如果此时有写者在等，第二次读锁会排在写者后面，而写者又在等第一个读锁释放——环形成。

读写锁还有三个容易被忽略的事实：

1. 它比互斥锁慢。无竞争时 pthread_rwlock_rdlock 也要一次原子加，写锁要两次；有竞争时还有读者计数、写者等待队列的管理开销。
2. 不支持升级。持有读锁再申请写锁会死锁（或返回 EDEADLK，取决于实现）。正确做法是释放读锁、重新申请写锁、重新检查条件——因为在你放锁的窗口里数据可能已经变了。
3. tryrdlock/timedrdlock 才有非阻塞语义。pthread_rwlock_trywrlock 返回 EBUSY 是正常结果。

2.13.5 读写锁、屏障与自旋锁

读写锁

pthread_rwlock_t 允许「多读单写」：读锁可共享，写锁独占。

pthread_rwlock_t rw = PTHREAD_RWLOCK_INITIALIZER;

pthread_rwlock_rdlock(&rw);      /* 读：多个读者可同时持有 */
/* 只读共享数据 */
pthread_rwlock_unlock(&rw);

pthread_rwlock_wrlock(&rw);      /* 写：独占 */
/* 修改共享数据 */
pthread_rwlock_unlock(&rw);

读写锁不是「一定比互斥锁快」。它的开销更大，且默认策略可能让读者优先、写者饥饿；glibc 提供 PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP 属性来缓解。只有当临界区读多写少、读操作耗时明显（例如遍历大表）时才值得用；临界区只有几条指令时，互斥锁更快。

读写锁不支持升级：持有读锁时再申请写锁会死锁（或返回 EDEADLK，取决于实现）。需要「读后写」时，先释放读锁，再重新申请写锁，并重新检查条件。

屏障

pthread_barrier_t 让一组线程互相等待，全部到齐后一起继续：

pthread_barrier_t barrier;
pthread_barrier_init(&barrier, NULL, 4);   /* 4 个线程 */

/* 每个线程做完第一阶段 */
int rc = pthread_barrier_wait(&barrier);
if (rc != 0 && rc != PTHREAD_BARRIER_SERIAL_THREAD) { /* ... */ }
/* 所有线程都到达后，才一起进入第二阶段 */

pthread_barrier_wait 对其中一个线程返回 PTHREAD_BARRIER_SERIAL_THREAD（通常是 0 号到达者），其余返回 0。屏障可重复使用；pthread_barrier_destroy 前要确保没有线程在等。它适合分阶段的并行计算（每阶段结束后交换边界数据），不适合「等待某个事件」——那是条件变量的工作。

自旋锁

用户态自旋锁用原子变量实现，拿不到就循环重试：

#include <stdatomic.h>
static atomic_flag spin = ATOMIC_FLAG_INIT;

while (atomic_flag_test_and_set_explicit(&spin, memory_order_acquire))
    ;                                   /* 忙等，浪费 CPU */
/* 临界区 */
atomic_flag_clear_explicit(&spin, memory_order_release);

自旋只适合「临界区极短（几十条指令）且线程数不超过 CPU 核数」的场景。如果持锁线程被抢占，其他自旋线程会白烧完整个时间片。Linux 的 pthread_spin_lock 在拿不到锁时可能调用 sched_yield，并不是纯自旋；不要假设它「无系统调用」。生产代码里，除非你在写分配器、无锁队列或内核旁路，否则用互斥锁。

死锁的检测：用户态没有 lockdep

内核有 lockdep（CONFIG_PROVE_LOCKING），能在运行时给每把锁建立「谁在谁之后」的依赖图，一有环就打印报告。用户态没有等价物：pthread_mutex_t 只是一个 40 字节的结构体，内核不知道你的加锁顺序。所以用户态的死锁检测只能靠三样东西：

TSan 的报告长这样，注意它检测的是「潜在死锁」，即两条线程以相反顺序拿过同一对锁，即使这一次没有真的卡住：

==================
WARNING: ThreadSanitizer: lock-order-inversion (potential deadlock) (pid=34567)
  Cycle in lock order graph: M0 (0x...) => M1 (0x...) => M0

  Mutex M1 acquired here while holding mutex M0 in thread T1:
    #0 pthread_mutex_lock ...
    #1 worker_a abba.c:31
  Mutex M0 acquired here while holding mutex M1 in thread T2:
    #0 pthread_mutex_lock ...
    #1 worker_b abba.c:52
==================

这正是「静态锁序」为什么值得作为团队约定的原因：它把一类只能在生产环境偶发的问题，变成了代码评审时就能看出的问题。

还要区分三个容易混淆的概念：

• 死锁：每个线程都在等一个永远不会被释放的资源，系统没有任何进展。
• 活锁：线程都在运行（CPU 占用高），但因为不断重试、互相让路，业务没有进展。
• 饥饿：系统整体有进展，但某个线程长期拿不到资源（读者优先的读写锁、非公平的 signal 唤醒都可能造成）。

三者的排查入口不同：死锁看「谁持有锁」，活锁看「重试次数与退避策略」，饥饿看「调度与唤醒策略」。

2.13.6 死锁、惊群与优先级反转

死锁的四个必要条件

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

工程上最有效的是锁序。给每把锁编号，规定「只能按编号递增加锁」；两个线程就不会出现 A→B 与 B→A 的环。

线程 1: lock(&a); lock(&b);
线程 2: lock(&b); lock(&a);      /* 各持一把，互相等 → 死锁 */

用 trylock + 退避可以避免永久死锁，但会引入活锁：两个线程同时失败、同时重试、再次碰撞。加随机退避或固定优先级可以缓解，但不如锁序干净。

惊群

pthread_cond_broadcast 唤醒所有等待者，但通常只有一个能拿到任务，其余醒来、抢锁、发现条件不满足、再睡。这就是惊群（thundering herd）。连接时代的 accept 惊群、epoll 的 EPOLLEXCLUSIVE 都是同一类问题。

减少惊群的办法：

• 能用 signal 就用 signal（只有一个等待者能推进时）；
• 把「一个条件变量」拆成多个，让唤醒更精确；
• 事件循环里用 EPOLLEXCLUSIVE/SO_REUSEPORT（见 2.18）；
• 接受「醒来后重新检查」的成本，别为了省几次唤醒引入更复杂的协议。

优先级反转

高优先级线程 H 等一把被低优先级线程 L 持有的锁；中优先级线程 M 抢占 L 的 CPU，L 无法运行、无法释放锁，H 无限等待。这就是优先级反转。火星探路者号在 1997 年遇到过著名的优先级反转故障，最终靠优先级继承缓解。

修法：

• 优先级继承（priority inheritance）：持锁的 L 临时继承 H 的优先级，避免被 M 抢占。POSIX 用 pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT)。
• 优先级天花板：锁的优先级设为所有可能使用者的最高值。
• 实时系统里尽量避免「高优先级线程等低优先级线程」的设计。

普通分时调度（SCHED_OTHER）下优先级反转的影响不明显，因为所有线程按 nice 值共享 CPU；一旦用 SCHED_FIFO/SCHED_RR，就必须认真对待。

四种「安全」不是一回事

「这个函数线程安全吗」有四种不同强度的答案，混用会导致「在信号处理器里调用了一个线程安全函数然后死锁」：

四者的关系是包含而不是等价：可重入函数通常线程安全，但线程安全函数不一定可重入（malloc 就是反例）；async-signal-safe 是 POSIX 用列表钦定的一小撮，printf 不在其中。

实践上的判断顺序：

1. 这段代码会在信号处理器里跑吗？会 → 只查 async-signal-safe 列表，其他一律不用。
2. 会被多个线程同时调用吗？会 → 看它是否使用静态/全局可变状态；有就用 _r 版本或自己加锁。
3. 只在一个线程里用，但可能被同一线程的信号处理器打断吗？会 → 仍按可重入要求处理。

2.13.7 线程安全与可重入

常见的不可重入/非线程安全函数：

strerror_r 有两个不兼容的版本：GNU 版返回 char *（可能不写缓冲区），XSI 版返回 int。用 _GNU_SOURCE 时拿到的是 GNU 版；跨平台代码要么避免它，要么用条件编译。

线程安全函数也不一定可重入。malloc 是线程安全的（内部有锁），但不可重入：信号处理器里调用 malloc 可能死锁（中断正好发生在持锁期间）。POSIX 的 async-signal-safe 列表才是信号处理器里能用的函数清单。

TLS 是「让函数变为线程安全」的常用手段：把静态变量改成 __thread，每个线程一份。但 TLS 不能解决「跨线程共享」的问题，也不能让函数可重入（同一线程的信号处理器会看到同一份 TLS）。

2.13.8 用 TSan 与 helgrind 抓数据竞争

ThreadSanitizer

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -g -O1 -fsanitize=thread -pthread racy_counter.c -o racy_tsan
./racy_tsan

典型输出：

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

读法：两条栈指向同一内存位置，至少一个是写，且没有 happens-before 关系。TSan 在编译期插桩，运行时维护每个内存位置的「影子」访问历史（shadow memory），所以能给出精确的两个访问点。

TSan 的局限：

• 只能检测实际执行到的路径；没跑到的分支查不出。
• 需要所有代码（包括库）插桩；未插桩的库里的竞争可能漏报。
• 不能与 ASan 同时使用（-fsanitize=address,thread 会报错）。
• 运行慢 5–15 倍，内存占用高。

helgrind

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -g -O0 -pthread racy_counter.c -o racy_dbg
valgrind --tool=helgrind ./racy_dbg

典型输出：

==12345== Possible data race during write of size 8 at 0x...
==12345==    at 0x...: bump (racy_counter.c:14)
==12345==  This conflicts with a previous write of size 8
==12345==    at 0x...: bump (racy_counter.c:14)
==12345==  Lock at 0x... was held here ...

helgrind 不需要重新编译（但 -g 才有行号），能在二进制层面检测锁序问题：

valgrind --tool=helgrind --track-lockorders=yes ./program
# 输出 "Thread #1: lock order 'A -> B' violated by ..."

helgrind 比 TSan 慢得多（几十到上百倍），适合小测试用例；TSan 适合跑单元测试与集成测试。两者都只能证明「发现了竞争」，不能证明「没有竞争」。

2.13.9 同步原语 API 与错误码总表

这一节回答「这个同步函数成功返回什么、失败返回什么、错在哪」。Pthreads 与 POSIX 信号量的错误约定不同，混用会写出永远不成立的判断。

/* 互斥量 */
int pthread_mutex_init(pthread_mutex_t *m, const pthread_mutexattr_t *a);
int pthread_mutex_lock(pthread_mutex_t *m);
int pthread_mutex_trylock(pthread_mutex_t *m);
int pthread_mutex_timedlock(pthread_mutex_t *m, const struct timespec *abs);
int pthread_mutex_clocklock(pthread_mutex_t *m, clockid_t c, const struct timespec *abs);
int pthread_mutex_unlock(pthread_mutex_t *m);
int pthread_mutex_destroy(pthread_mutex_t *m);
int pthread_mutex_consistent(pthread_mutex_t *m);

/* 条件变量 */
int pthread_cond_init(pthread_cond_t *c, const pthread_condattr_t *a);
int pthread_cond_wait(pthread_cond_t *c, pthread_mutex_t *m);
int pthread_cond_timedwait(pthread_cond_t *c, pthread_mutex_t *m, const struct timespec *abs);
int pthread_cond_clockwait(pthread_cond_t *c, pthread_mutex_t *m, clockid_t k, const struct timespec *abs);
int pthread_cond_signal(pthread_cond_t *c);
int pthread_cond_broadcast(pthread_cond_t *c);
int pthread_cond_destroy(pthread_cond_t *c);

/* 读写锁、屏障、自旋锁 */
int pthread_rwlock_rdlock(pthread_rwlock_t *rw);
int pthread_rwlock_wrlock(pthread_rwlock_t *rw);
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rw);
int pthread_rwlock_trywrlock(pthread_rwlock_t *rw);
int pthread_rwlock_timedrdlock(pthread_rwlock_t *rw, const struct timespec *abs);
int pthread_barrier_init(pthread_barrier_t *b, const pthread_barrierattr_t *a, unsigned count);
int pthread_barrier_wait(pthread_barrier_t *b);
int pthread_spin_lock(pthread_spinlock_t *s);
int pthread_spin_trylock(pthread_spinlock_t *s);

/* POSIX 信号量：注意，这一组返回 -1 并设置 errno */
int sem_init(sem_t *s, int pshared, unsigned value);
int sem_wait(sem_t *s);
int sem_trywait(sem_t *s);
int sem_timedwait(sem_t *s, const struct timespec *abs);
int sem_clockwait(sem_t *s, clockid_t c, const struct timespec *abs);
int sem_post(sem_t *s);
int sem_getvalue(sem_t *s, int *val);

读表要点：Pthreads 返回错误码，POSIX 信号量返回 -1 并设置 errno。所以下面两种写法不能互换：

/* Pthreads */
int rc = pthread_mutex_lock(&m);
if (rc != 0) { fprintf(stderr, "%s\n", strerror(rc)); return -1; }

/* POSIX semaphore */
if (sem_wait(&s) == -1) {
    if (errno == EINTR) { /* 重试 */ }
    else { perror("sem_wait"); return -1; }
}

2.13.10 futex：互斥锁与条件变量的内核实现

这一节回答「明明加了锁，为什么 strace 里看不到系统调用；一旦有竞争又满屏 futex」。答案是：锁的快速路径完全在用户态，只有争用时才进内核，而进内核的入口就是 futex。

futex（fast userspace mutex）是「一个 32 位整数 + 一条等待队列」的组合。内核只在两种情况下介入：

• 用户态发现「值不是期望值」，调用 FUTEX_WAIT 睡眠；
• 用户态改了值，调用 FUTEX_WAKE 唤醒等待者。

pthread_mutex_lock
  ├─ 快速路径：lock cmpxchg 把 __lock 从 0 改成 1 → 成功，返回（无系统调用）
  └─ 慢速路径：__lll_lock_wait
        ├─ 把 __lock 置 2（表示有等待者）
        ├─ futex(&__lock, FUTEX_WAIT_PRIVATE, 2, NULL)   ← 睡眠
        └─ 被唤醒后重新尝试 cmpxchg

pthread_mutex_unlock
  ├─ 快速路径：store 0，若原值是 2 才进慢速路径
  └─ futex(&__lock, FUTEX_WAKE_PRIVATE, 1)

FUTEX_WAIT 的语义有一个关键点：它原子地检查 *uaddr == val，不等则立刻返回 EAGAIN，相等才睡眠。这避免了「检查后被抢占、信号丢失」的窗口——和条件变量的 wait 是同一个设计思想，只是层次更低。

FUTEX_PRIVATE_FLAG（0x80）可以和上面任意操作按位或，表示「只在本进程的 mm 内匹配」。glibc 对线程私有的锁用 _PRIVATE，对 pshared=1 的锁去掉这个标志——这就是 2.15 里跨进程互斥锁能工作的底层机制。

内核侧的数据结构是哈希桶 + 等待队列：

/* kernel/futex/core.c，简化示意 */
struct futex_hash_bucket {
    atomic_t waiters;
    spinlock_t lock;
    struct plist_head chain;        /* 所有等待同一个 uaddr 的 futex_q */
};

futex_wait 先用 futex_hash(uaddr) 找到桶，把当前任务挂到 chain 上，然后 schedule()。futex_wake 找到同一个桶，按 plist 顺序唤醒 N 个。哈希冲突只影响性能，不影响正确性，因为每个等待项都记录了完整的 uaddr。

条件变量怎么建在 futex 上

pthread_cond_wait 的核心是「释放互斥锁 + 睡在条件变量的 futex 字上」这两步的原子化。glibc 2.25 之后的实现用 64 位序号 __wseq 和两个等待组来避免惊群：

cond_wait:
  1. 取号：wseq = atomic_fetch_add(&cond->__wseq, 2)
  2. 释放 mutex
  3. futex_wait_bitset(&cond->__wseq, wseq, 组位掩码)   ← 睡在这里
  4. 被唤醒后重新 lock(mutex)
  5. 循环检查谓词（这就是 while 的来源）

cond_signal:
  1. 若无等待者，直接返回（信号丢失）
  2. 选组 1 的一个等待者，futex_wake_bitset(...)

cond_broadcast:
  1. 两组都唤醒（futex_wake_bitset 两次，或按组引用计数唤醒）

第 5 步是重点：醒来之后重新加锁、重新检查，这是库函数的职责边界；库只能保证「你被唤醒了」，不能保证「条件成立了」。

2.13.11 程序一：丢失唤醒、if 与 while 的代价

这段程序要验证三件事：丢失唤醒是确定性的 bug、if 的代价是读到空队列、while 才是正确契约。三个模式共用一个二进制，用 alarm 做超时看门狗。

/* cond_bug.c — 丢失唤醒 / if / while 三种条件变量写法 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <signal.h>      /* alarm/signal/SIGALRM */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

static pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  cv = PTHREAD_COND_INITIALIZER;
static pthread_barrier_t bar;

static int ready;                  /* lost 模式：生产者置位 */
static int items;                  /* if/while 模式：待消费数量 */
static int done;                   /* while 模式：生产者结束 */
static int bad;                    /* if 模式：读到空队列 */
static int consumed;               /* while 模式：消费总数 */
static const char *g_mode = "lost";

static void die_pthread(const char *what, int rc)
{
    fprintf(stderr, "%s: %s\n", what, strerror(rc));
    exit(EXIT_FAILURE);
}

static void sleep_ms(long ms)
{
    struct timespec ts = { .tv_sec = ms / 1000, .tv_nsec = (ms % 1000) * 1000000L };

    while (nanosleep(&ts, &ts) == -1 && errno == EINTR)
        ;
}

/* 只做 async-signal-safe 的事：write + _exit */
static void on_alarm(int sig)
{
    (void)sig;
    static const char msg[] = "timeout: consumer never woke (lost wakeup)\n";
    (void)!write(STDERR_FILENO, msg, sizeof msg - 1);
    _exit(3);
}

/* ---------- 模式 lost：无锁检查谓词，然后无条件 wait ---------- */
static void *consumer_lost(void *arg)
{
    (void)arg;

    if (!ready) {                       /* 无锁读：此刻是 0 */
        sleep_ms(100);                  /* 生产者在这段时间置位并 signal：无人等待，信号丢失 */
        int rc = pthread_mutex_lock(&m);
        if (rc != 0)
            die_pthread("lock", rc);
        rc = pthread_cond_wait(&cv, &m); /* 永远睡下去 */
        if (rc != 0)
            die_pthread("cond_wait", rc);
        rc = pthread_mutex_unlock(&m);
        if (rc != 0)
            die_pthread("unlock", rc);
    }
    printf("consumer_lost: woke, ready=%d\n", ready);
    return NULL;
}

static void *producer_lost(void *arg)
{
    (void)arg;

    sleep_ms(10);                       /* 让消费者先完成那次无锁检查 */
    int rc = pthread_mutex_lock(&m);
    if (rc != 0)
        die_pthread("lock", rc);
    ready = 1;
    rc = pthread_cond_signal(&cv);      /* 没有等待者 → 信号消失 */
    if (rc != 0)
        die_pthread("signal", rc);
    rc = pthread_mutex_unlock(&m);
    if (rc != 0)
        die_pthread("unlock", rc);
    return NULL;
}

/* ---------- 模式 if：醒来后不再检查谓词 ---------- */
static void *consumer_if(void *arg)
{
    (void)arg;

    (void)pthread_barrier_wait(&bar);

    int rc = pthread_mutex_lock(&m);
    if (rc != 0)
        die_pthread("lock", rc);

    if (items == 0)
        rc = pthread_cond_wait(&cv, &m); /* 只检查这一次 */
    if (rc != 0)
        die_pthread("cond_wait", rc);

    items--;                             /* 醒来后直接消费：可能是 0 → -1 */
    if (items < 0)
        bad = 1;

    rc = pthread_mutex_unlock(&m);
    if (rc != 0)
        die_pthread("unlock", rc);
    return NULL;
}

static void *producer_if(void *arg)
{
    (void)arg;

    (void)pthread_barrier_wait(&bar);
    sleep_ms(50);                        /* 等两个消费者都睡下 */

    int rc = pthread_mutex_lock(&m);
    if (rc != 0)
        die_pthread("lock", rc);
    items = 1;
    rc = pthread_cond_broadcast(&cv);
    if (rc != 0)
        die_pthread("broadcast", rc);
    rc = pthread_mutex_unlock(&m);
    if (rc != 0)
        die_pthread("unlock", rc);
    return NULL;
}

/* ---------- 模式 while：正确写法 ---------- */
static void *consumer_while(void *arg)
{
    (void)arg;

    (void)pthread_barrier_wait(&bar);

    for (;;) {
        int rc = pthread_mutex_lock(&m);
        if (rc != 0)
            die_pthread("lock", rc);

        while (items == 0 && !done)
            rc = pthread_cond_wait(&cv, &m);
        if (rc != 0)
            die_pthread("cond_wait", rc);

        if (items == 0) {                /* done 且队列空：正常结束 */
            rc = pthread_mutex_unlock(&m);
            if (rc != 0)
                die_pthread("unlock", rc);
            break;
        }

        items--;
        consumed++;
        rc = pthread_mutex_unlock(&m);
        if (rc != 0)
            die_pthread("unlock", rc);
    }
    return NULL;
}

static void *producer_while(void *arg)
{
    (void)arg;

    (void)pthread_barrier_wait(&bar);

    for (int round = 0; round < 4; round++) {
        int rc = pthread_mutex_lock(&m);
        if (rc != 0)
            die_pthread("lock", rc);
        items++;
        rc = pthread_cond_signal(&cv);
        if (rc != 0)
            die_pthread("signal", rc);
        rc = pthread_mutex_unlock(&m);
        if (rc != 0)
            die_pthread("unlock", rc);
        sleep_ms(5);
    }

    int rc = pthread_mutex_lock(&m);
    if (rc != 0)
        die_pthread("lock", rc);
    done = 1;
    rc = pthread_cond_broadcast(&cv);
    if (rc != 0)
        die_pthread("broadcast", rc);
    rc = pthread_mutex_unlock(&m);
    if (rc != 0)
        die_pthread("unlock", rc);
    return NULL;
}

int main(int argc, char **argv)
{
    if (argc == 2)
        g_mode = argv[1];

    int rc;
    if (strcmp(g_mode, "lost") == 0) {
        alarm(2);                        /* 看门狗：2 秒后强制报告 */
        signal(SIGALRM, on_alarm);

        pthread_t c, p;
        rc = pthread_create(&c, NULL, consumer_lost, NULL);
        if (rc != 0)
            die_pthread("create consumer", rc);
        rc = pthread_create(&p, NULL, producer_lost, NULL);
        if (rc != 0)
            die_pthread("create producer", rc);
        rc = pthread_join(p, NULL);
        if (rc != 0)
            die_pthread("join producer", rc);
        rc = pthread_join(c, NULL);      /* 卡在这里，直到 alarm 触发 */
        if (rc != 0)
            die_pthread("join consumer", rc);
        return 0;
    }

    if (strcmp(g_mode, "if") == 0) {
        rc = pthread_barrier_init(&bar, NULL, 3);
        if (rc != 0)
            die_pthread("barrier_init", rc);

        pthread_t c1, c2, p;
        if ((rc = pthread_create(&c1, NULL, consumer_if, NULL)) != 0)
            die_pthread("create c1", rc);
        if ((rc = pthread_create(&c2, NULL, consumer_if, NULL)) != 0)
            die_pthread("create c2", rc);
        if ((rc = pthread_create(&p, NULL, producer_if, NULL)) != 0)
            die_pthread("create producer", rc);

        pthread_join(c1, NULL);
        pthread_join(c2, NULL);
        pthread_join(p, NULL);
        pthread_barrier_destroy(&bar);

        printf("if mode: items=%d bad=%d\n", items, bad);
        return bad ? 4 : 0;
    }

    if (strcmp(g_mode, "while") == 0) {
        rc = pthread_barrier_init(&bar, NULL, 3);
        if (rc != 0)
            die_pthread("barrier_init", rc);

        pthread_t c1, c2, p;
        if ((rc = pthread_create(&c1, NULL, consumer_while, NULL)) != 0)
            die_pthread("create c1", rc);
        if ((rc = pthread_create(&c2, NULL, consumer_while, NULL)) != 0)
            die_pthread("create c2", rc);
        if ((rc = pthread_create(&p, NULL, producer_while, NULL)) != 0)
            die_pthread("create producer", rc);

        pthread_join(c1, NULL);
        pthread_join(c2, NULL);
        pthread_join(p, NULL);
        pthread_barrier_destroy(&bar);

        printf("while mode: consumed=%d expect=4\n", consumed);
        return consumed == 4 ? 0 : 5;
    }

    fprintf(stderr, "usage: %s lost|if|while\n", argv[0]);
    return EXIT_FAILURE;
}

编译与三种模式：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread cond_bug.c -o cond_bug

./cond_bug lost; echo "exit=$?"
# timeout: consumer never woke (lost wakeup)
# exit=3        ← 消费者睡死，信号在无人等待时被丢弃

./cond_bug if; echo "exit=$?"
# if mode: items=-1 bad=1
# exit=4        ← 两个消费者都被唤醒，第二个读到空队列

./cond_bug while; echo "exit=$?"
# while mode: consumed=4 expect=4
# exit=0

三个结果对应三种失败模式：

1. 丢失唤醒（lost）：谓词检查发生在锁外，signal 发生时没有等待者，信号被丢弃。修复不是「多 signal 几次」，而是「检查与等待必须在锁内，且等待前重新检查」——即 while 契约。
2. if 的代价（if）：broadcast 唤醒所有等待者，但只有一个能推进。if 不重新检查，第二个消费者把 items 减成 -1，生产中表现为「读到空队列/空指针」。
3. while：醒来后重新检查，队列空且 done 才退出。consumed=4 正好等于生产者放入的数量。

2.13.12 程序二：自旋、互斥锁与原子操作的竞争代价

这段程序要验证一件事：没有一种同步方式在所有竞争强度下都最快。它用同一份工作量跑三种实现，并把「临界区长度」做成了可调参数。

/* lock_cost.c — 自旋锁 / 互斥锁 / 原子操作，不同线程数下的耗时 */
#define _GNU_SOURCE
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define ITERS 200000L

enum { MODE_MUTEX = 0, MODE_SPIN = 1, MODE_ATOMIC = 2 };

static pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
static atomic_flag spin = ATOMIC_FLAG_INIT;
static atomic_long g_atomic;
static long g_shared;
static int g_work = 16;              /* 临界区里的“工作量” */
static int g_mode = MODE_MUTEX;

static void die_pthread(const char *what, int rc)
{
    fprintf(stderr, "%s: %s\n", what, strerror(rc));
    exit(EXIT_FAILURE);
}

static long now_ms(void)
{
    struct timespec ts;

    if (clock_gettime(CLOCK_MONOTONIC, &ts) == -1) {
        perror("clock_gettime");
        exit(EXIT_FAILURE);
    }
    return ts.tv_sec * 1000L + ts.tv_nsec / 1000000L;
}

static void *worker(void *arg)
{
    (void)arg;

    switch (g_mode) {
    case MODE_MUTEX:
        for (long i = 0; i < ITERS; i++) {
            int rc = pthread_mutex_lock(&g_mutex);
            if (rc != 0)
                die_pthread("mutex_lock", rc);
            for (int k = 0; k < g_work; k++)
                g_shared += k;
            rc = pthread_mutex_unlock(&g_mutex);
            if (rc != 0)
                die_pthread("mutex_unlock", rc);
        }
        break;

    case MODE_SPIN:
        for (long i = 0; i < ITERS; i++) {
            while (atomic_flag_test_and_set_explicit(&spin, memory_order_acquire))
                ;                                   /* 忙等，烧 CPU */
            for (int k = 0; k < g_work; k++)
                g_shared += k;
            atomic_flag_clear_explicit(&spin, memory_order_release);
        }
        break;

    default:
        for (long i = 0; i < ITERS; i++)
            atomic_fetch_add_explicit(&g_atomic, 1, memory_order_relaxed);
        break;
    }
    return NULL;
}

int main(int argc, char **argv)
{
    int nthreads = 4;

    if (argc >= 2) {
        if (strcmp(argv[1], "mutex") == 0)       g_mode = MODE_MUTEX;
        else if (strcmp(argv[1], "spin") == 0)   g_mode = MODE_SPIN;
        else if (strcmp(argv[1], "atomic") == 0) g_mode = MODE_ATOMIC;
        else {
            fprintf(stderr, "usage: %s mutex|spin|atomic [threads] [work]\n", argv[0]);
            return EXIT_FAILURE;
        }
    }
    if (argc >= 3)
        nthreads = atoi(argv[2]);
    if (argc >= 4)
        g_work = atoi(argv[3]);
    if (nthreads < 1 || nthreads > 64 || g_work < 1) {
        fprintf(stderr, "bad threads/work\n");
        return EXIT_FAILURE;
    }

    pthread_t tids[64];
    long start = now_ms();

    for (int i = 0; i < nthreads; i++) {
        int rc = pthread_create(&tids[i], NULL, worker, NULL);
        if (rc != 0)
            die_pthread("pthread_create", rc);
    }
    for (int i = 0; i < nthreads; i++) {
        int rc = pthread_join(tids[i], NULL);
        if (rc != 0)
            die_pthread("pthread_join", rc);
    }

    long elapsed = now_ms() - start;
    const char *name = g_mode == MODE_MUTEX ? "mutex"
                     : g_mode == MODE_SPIN ? "spin" : "atomic";

    printf("%-6s threads=%d work=%2d elapsed=%4ldms  checksum=%ld\n",
           name, nthreads, g_work, elapsed,
           g_mode == MODE_ATOMIC ? (long)atomic_load(&g_atomic) : g_shared);
    return EXIT_SUCCESS;
}

在一台 8 逻辑核的机器上（ITERS=20 万、临界区 16 次加法）的一次实测：

绝对值随机器变化很大，趋势是稳定的：自旋在超订时恶化一到两个数量级，互斥锁随竞争线性增长但斜率小得多。

读表要点：

1. 原子操作只在「一个计数器」场景成立。atomic 模式没有临界区，所以它不可比较；把它放进来是为了给出「同步成本的下限」。
2. 自旋在超订（线程数 > 核数）时急剧恶化。持锁线程被调度走，其他线程自旋等待，CPU 全花在 pause 上。这就是「自旋锁只适合临界区极短且不超订」的量化依据。
3. 互斥锁的成本主要来自竞争，不是加锁本身。无竞争时它和自旋几乎一样快（都是原子指令），有竞争时才进 futex 睡眠。

2.13.13 程序三：屏障与单生产者单消费者队列

这段程序要验证两件事：屏障如何让「同时开始」变得确定，release/acquire 如何在不加锁的情况下安全传递数据。它用屏障对齐生产者与消费者的起跑线，用环形缓冲区传递 200 万个整数并校验总和。

/* spsc_ring.c — 屏障 + release/acquire 的单生产者单消费者队列 */
#define _GNU_SOURCE
#include <pthread.h>
#include <sched.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define CAP 1024UL
#define N   2000000UL

/* head 与 tail 各占一条缓存行：避免伪共享（false sharing） */
struct ring {
    unsigned long buf[CAP];
    _Atomic unsigned long head __attribute__((aligned(64)));
    _Atomic unsigned long tail __attribute__((aligned(64)));
};

static struct ring g_ring;
static pthread_barrier_t g_bar;

static void die_pthread(const char *what, int rc)
{
    fprintf(stderr, "%s: %s\n", what, strerror(rc));
    exit(EXIT_FAILURE);
}

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

    int rc = pthread_barrier_wait(&g_bar);
    if (rc != 0 && rc != PTHREAD_BARRIER_SERIAL_THREAD)
        die_pthread("barrier_wait(producer)", rc);

    for (unsigned long i = 0; i < N; i++) {
        unsigned long tail = atomic_load_explicit(&g_ring.tail, memory_order_relaxed);

        /* 队列满：等消费者推进 head。acquire 保证看到 head 更新前的读取 */
        while (tail - atomic_load_explicit(&g_ring.head, memory_order_acquire) == CAP)
            sched_yield();

        g_ring.buf[tail % CAP] = i;

        /* release：保证上面的 buf 写入不会被重排到这条 store 之后 */
        atomic_store_explicit(&g_ring.tail, tail + 1, memory_order_release);
    }
    return NULL;
}

static void *consumer(void *arg)
{
    unsigned long long sum = 0;

    (void)arg;

    int rc = pthread_barrier_wait(&g_bar);
    if (rc != 0 && rc != PTHREAD_BARRIER_SERIAL_THREAD)
        die_pthread("barrier_wait(consumer)", rc);

    for (unsigned long i = 0; i < N; i++) {
        unsigned long head = atomic_load_explicit(&g_ring.head, memory_order_relaxed);

        while (atomic_load_explicit(&g_ring.tail, memory_order_acquire) == head)
            sched_yield();

        sum += g_ring.buf[head % CAP];

        /* release：保证「已经读完这个槽位」对生产者可见 */
        atomic_store_explicit(&g_ring.head, head + 1, memory_order_release);
    }

    return (void *)(uintptr_t)sum;
}

int main(void)
{
    int rc = pthread_barrier_init(&g_bar, NULL, 2);
    if (rc != 0)
        die_pthread("barrier_init", rc);

    pthread_t p, c;
    rc = pthread_create(&p, NULL, producer, NULL);
    if (rc != 0)
        die_pthread("create producer", rc);
    rc = pthread_create(&c, NULL, consumer, NULL);
    if (rc != 0)
        die_pthread("create consumer", rc);

    void *ret = NULL;
    rc = pthread_join(p, NULL);
    if (rc != 0)
        die_pthread("join producer", rc);
    rc = pthread_join(c, &ret);
    if (rc != 0)
        die_pthread("join consumer", rc);

    pthread_barrier_destroy(&g_bar);

    unsigned long long got = (unsigned long long)(uintptr_t)ret;
    unsigned long long expect = (unsigned long long)N * (N - 1) / 2;

    printf("sum=%llu expect=%llu %s\n", got, expect,
           got == expect ? "OK" : "MISMATCH");
    return got == expect ? 0 : 1;
}

编译运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread spsc_ring.c -o spsc_ring
./spsc_ring
# sum=1999999000000 expect=1999999000000 OK

四个设计点：

1. 屏障只用于「同时开始」。屏障的 count 在 init 时固定，两个线程各等一次；主线程不参与，所以 count 是 2。超过 count 的等待者会让屏障进入下一轮，这是「可重用」的副作用，也是误用的常见来源。
2. 索引用 relaxed 读、release 写、acquire 读对方索引。tail 的 release 与消费者的 acquire 配对，保证「先写 buf 再发布索引」；head 方向同理。
3. head/tail 各占一条缓存行。放在同一行时，生产者写 tail 会让消费者手里的 head 缓存行失效，每次读写都跨核同步——这就是伪共享，实测能差出数倍。
4. 没有锁。SPSC 是唯一能安全做到「无锁」的常见队列形态：只有一个生产者写 tail，一个消费者写 head，不存在写-写竞争。

2.13.14 实验：用 strace、perf 与 TSan 看同步

实验一：futex 调用次数就是竞争次数

strace -f -c -e trace=futex ./lock_cost mutex 1 16
# % time     seconds  usecs/call     calls    errors syscall
# 100.00    0.000000           0         0           futex     ← 无竞争，零系统调用

strace -f -c -e trace=futex ./lock_cost mutex 8 16
# % time     seconds  usecs/call     calls    errors syscall
# 100.00    1.284300          18     68000           futex     ← 有竞争，每次睡眠一次

读法：futex 调用次数 ≈ 线程「进入睡眠」的次数 + 唤醒次数。无竞争时 mutex 完全在用户态完成，这解释了为什么「加锁不影响性能」的说法在低竞争下成立。竞争激烈时，每次 FUTEX_WAIT 都伴随一次上下文切换，耗时从纳秒级跳到微秒级。

用 strace -T 看单次耗时，或用 -e trace=futex -o 落盘后再统计：

strace -f -T -e trace=futex ./lock_cost mutex 8 16 2>&1 | \
  awk -F'[<>]' '/futex/{print $2}' | sort -n | tail -3
# 0.000120  ← 最慢的几次 futex 等待，单位秒

实验二：perf 看阻塞在哪里

perf stat -e context-switches,cpu-migrations,page-faults,instructions \
          ./lock_cost mutex 8 16
perf sched record -- ./lock_cost mutex 8 16
perf sched latency | head -10

perf sched latency 会给出每个线程的「等待时间 / 运行时间」，以及最长的调度延迟。如果 worker 的等待时间远大于运行时间，说明时间花在等锁而不是计算上——这时候优化临界区长度比换锁类型更有效。

没有 perf 权限时：

grep ctxt_switches /proc/$$/status
# voluntary_ctxt_switches:     41
# nonvoluntary_ctxt_switches:  2

自愿切换高 = 主动睡眠（等锁、等 I/O、等条件变量），正常；非自愿切换高 = 被抢占，通常是线程数超过核数。

实验三：TSan/helgrind 的锁序报告

用 2.12.12 的 abba_deadlock.c 编译两个版本：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -g -O1 -fsanitize=thread \
   -pthread abba_deadlock.c -o abba_tsan
./abba_tsan order
# ==================
# WARNING: ThreadSanitizer: lock-order-inversion (potential deadlock)
#   Cycle in lock order graph: M0 => M1 => M0
# ==================

valgrind --tool=helgrind --track-lockorders=yes ./abba_deadlock_dbg order
# Thread #1: lock order '0x... -> 0x...' violated by thread #2
#   Earlier lock order: 0x... -> 0x...

注意：两个线程都按 lo->hi 顺序加锁时，TSan 不应报告；它报的是「同一个程序里出现过相反顺序」。所以它既能抓当前 bug，也能抓「某条分支走错顺序」的潜在 bug。

2.13.15 陷阱清单：同步的十五个坑

这十五条里，第 1–5 条属于「条件变量协议」，第 6–9 条属于「锁的生命周期与死锁」，第 10–11 条属于「读写锁策略」，第 12–15 条属于「信号量与跨进程」。按这个分组记，比逐条背更容易在排障时想起来。

2.13.16 本章出口检查

母本对应

下一站：2.14 管道、FIFO 与消息队列把同步从「线程共享内存」换到「进程间传递字节」：管道什么时候阻塞、什么时候 EOF，FIFO 的打开顺序为什么会死锁，POSIX 消息队列如何保留消息边界。