2.12 线程

2.11 里的并发还只是「信号打断主流程」。从这一章开始，多个执行流真正同时跑在同一片地址空间里：它们共享堆、全局变量、fd 表，各自拥有栈、寄存器和 errno。共享让通信变得极快——一个指针就能传结构体；也让错误变得极难复现——两个线程谁先写那个指针，取决于调度器。

线程不是「更轻的进程」。进程是资源隔离的单位，线程是调度的单位。Linux 内核不区分两者：fork() 与 pthread_create() 最终都走 clone()，区别只在 flags——是否共享地址空间、fd 表、信号处理器、文件系统信息。理解这一点，就能解释本章所有「为什么」：为什么一个线程 close(fd) 会影响同进程其他线程，为什么一个线程 exit() 会杀掉整个进程，为什么 fork() 在多线程程序里危险。

这一章先把线程的生命周期讲对（创建、连接、分离、取消、清理），再讲每线程状态（栈、TLS、errno、信号掩码），最后落到一个可运行的线程池。线程之间的数据竞争与同步原语留给 2.13，但线程池已经需要互斥锁和条件变量——你会先看到它们怎么用，下一章再讲透为什么。

2.12.1 线程模型：clone 出来的执行流

 高地址
 ┌──────────────────────────────┐
 │ 主线程栈（默认 8 MB，向下增长） │
 ├──────────────────────────────┤
 │ 共享库 / mmap 区域             │
 ├──────────────────────────────┤
 │ 堆（brk / mmap）               │   ← 所有线程共享，malloc 需要加锁
 ├──────────────────────────────┤
 │ .bss / .data（全局变量）        │   ← 所有线程共享，竞态的发源地
 ├──────────────────────────────┤
 │ .text（只读代码）               │
 ├──────────────────────────────┤
 │ 线程 1 栈（默认 8 MB）          │   ← 每线程独立，溢出踩到 guard page
 ├──────────────────────────────┤
 │ 线程 2 栈（默认 8 MB）          │
 ├──────────────────────────────┤
 │ TLS + 线程控制块（TCB）         │   ← errno、__thread 变量
 └──────────────────────────────┘
 低地址

读图：线程栈不是从堆里 malloc 出来的，而是 mmap 出来的独立区域（glibc 用 mmap 分配，带一个 guard page）。所以「栈大小」是创建线程时就要决定的属性；线程跑起来之后不能扩容。图中堆只有一份，两个线程同时 malloc 时，glibc 内部会用 arena 锁保护——这也是为什么 fork() 在多线程程序里可能死锁：子进程可能继承了「已被另一个线程锁住」的 malloc 锁。

内核视角的线程就是一个 task_struct，有自己的 pid（内核叫 TID）、tgid（线程组 ID，等于进程 ID）。gettid() 返回 TID，getpid() 返回 TGID；ps -eLf 或 ls /proc/PID/task 能看到同一进程的多个 TID。glibc 2.30 起可直接调用 gettid()，更早版本用 syscall(SYS_gettid)。

2.12.2 创建、连接与分离

int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                   void *(*start_routine)(void *), void *arg);
int pthread_join(pthread_t thread, void **retval);
int pthread_detach(pthread_t thread);
void pthread_exit(void *retval) __attribute__((noreturn));

Pthreads 的错误约定与系统调用相反：成功返回 0，失败返回错误码，不设置 errno。所以不能写 if (pthread_create(...) == -1) perror(...)；正确写法是拿返回值去 strerror()。

/* thread_basic.c — 创建、传参、连接、取返回值 */
#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

struct job {
    int id;
    long result;
};

static void *worker(void *arg)
{
    struct job *j = arg;
    long sum = 0;

    for (int i = 1; i <= j->id; i++)
        sum += i;
    j->result = sum;

    printf("worker %d: pthread_t=%lu sum=%ld\n",
           j->id, (unsigned long)pthread_self(), sum);
    return j;                       /* 返回值由 pthread_join 的 retval 收到 */
}

int main(void)
{
    enum { N = 3 };
    struct job jobs[N];
    pthread_t tids[N];

    for (int i = 0; i < N; i++) {
        jobs[i].id = i + 1;
        jobs[i].result = 0;

        int rc = pthread_create(&tids[i], NULL, worker, &jobs[i]);
        if (rc != 0) {
            fprintf(stderr, "pthread_create: %s\n", strerror(rc));
            return EXIT_FAILURE;
        }
    }

    for (int i = 0; i < N; i++) {
        void *ret = NULL;
        int rc = pthread_join(tids[i], &ret);
        if (rc != 0) {
            fprintf(stderr, "pthread_join: %s\n", strerror(rc));
            return EXIT_FAILURE;
        }
        struct job *j = ret;
        printf("joined job %d: result=%ld\n", j->id, j->result);
    }
    return EXIT_SUCCESS;
}

编译与运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread thread_basic.c -o thread_basic
./thread_basic
# 输出顺序不固定：三个 worker 谁先打印由调度器决定

pthread_join 做三件事：等待线程终止、回收它的栈与 TCB、取出返回值。不 join 的 joinable 线程会泄漏资源，类似僵尸进程，但泄漏的是用户态栈（默认 8 MB）而不是 task_struct。

线程的终止方式有四种：

pthread_detach 适合「发射后不管」的后台任务，例如日志刷盘、心跳。但 detached 线程崩溃时没人收尸，也没有返回值可查；生产代码里更常见的是「joinable + 统一 join」，或者用线程池管理生命周期。

pthread_attr_t：属性对象里究竟有什么

attr 是不透明结构体（glibc 里 pthread_attr_t 是 56 字节的联合体），你只能通过 API 读写。但每个字段都对应 NPTL 或内核的一个真实行为，设错不会报错，只会在运行时表现成「参数没生效」。这张表回答「哪些属性值得设、默认值是什么」：

读表要点：默认值是「继承创建者的调度与掩码 + 8 MB 栈 + 一页 guard」。所谓「pthread_attr_setschedparam 不起作用」，九成是因为没有先调用：

pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);

不改这一项，pthread_create 会用创建者的调度参数覆盖你在 attr 里写的一切，而且返回 0，不报错。这是 Pthreads 里最隐蔽的一类「成功但无效」。

线程 ID：三个不同的编号

同一个线程有三个「ID」，混用是排查时最常见的迷惑源：

三者的关系值得逐条说清：

• pthread_t 不能用 == 比较，标准只给 pthread_equal(t1, t2)；也不能假设它是整数。glibc 的实现让它恰好是一个指针值，于是 printf("%lu", (unsigned long)t) 在 x86-64 上「能用」——这是巧合，不是契约。musl 同样是指针，某些商业 UNIX 是整数或结构体下标。
• 内核完全不知道 pthread_t。要把两者对上，只能在线程入口里调 gettid() 并自己记录，或者读 /proc/PID/task（目录名就是 TID）。
• getpid() 在所有线程里返回同一个值（TGID）。kill(getpid(), sig) 是进程级信号，tgkill(tgid, tid, sig) 才精确到线程；pthread_kill 是后者的包装。

/* 把 pthread_t 与 TID 对上：入口处记录，退出前打印 */
static __thread pid_t my_tid;           /* 每线程一份，不需要加锁 */

static void *worker(void *arg)
{
    my_tid = gettid();                  /* glibc < 2.30 用 syscall(SYS_gettid) */
    printf("pthread_t=%p tid=%d\n", (void *)pthread_self(), (int)my_tid);
    return NULL;
}

一个进程里 pthread_t 与 TID 都不是连续分配的：TID 由内核的 PID 分配器给，回收后会被复用；pthread_t 指向的 TCB 在 join 之后被 munmap，旧值再拿来用就是悬空句柄。不要缓存 pthread_t 跨线程生命周期使用，除非你同时持有「线程还活着」的同步保证。

2.12.3 线程栈与属性

glibc 在 x86-64 上默认给每个线程 8 MB 栈（受 RLIMIT_STACK 影响），外加一个 4 KB 的 guard page。1000 个线程就是 8 GB 虚拟地址空间——虚拟内存不心疼，但每个栈的物理页在用到时才分配，所以真正的问题是「线程数 × 栈峰值」的 RSS，而不是虚拟大小。

栈溢出不会返回错误，只会 SIGSEGV：

栈溢出的典型症状是「递归没有终止条件」或「大数组放在栈上」：每层消耗几 KB，最终踩到 guard page，进程收到 SIGSEGV。dmesg 里会看到 segfault at ... error 6 in a.out，gdb 的 backtrace 可能有几万层。排查手段：

• ulimit -s 看主线程栈上限；
• pthread_attr_getstacksize 或 pthread_getattr_np 看线程实际栈大小；
• gdb -p 后 thread apply all bt 看哪个线程爆了；
• cc -fstack-usage 编译，看每个函数的帧大小。

属性对象让创建线程前就能定下栈大小、分离状态、调度策略：

/* 创建 256 KB 栈、detached 的线程（节选） */
pthread_attr_t attr;
int rc = pthread_attr_init(&attr);
if (rc != 0) { /* 记录 rc 并返回 */ }

rc = pthread_attr_setstacksize(&attr, 256 * 1024);
if (rc != 0) { /* EINVAL：小于 PTHREAD_STACK_MIN */ }

rc = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
if (rc != 0) { /* ... */ }

pthread_t t;
rc = pthread_create(&t, &attr, worker, arg);
pthread_attr_destroy(&attr);        /* 属性对象用完即销毁，与线程生命周期无关 */

pthread_attr_destroy 不会影响已经创建的线程——pthread_create 会把属性复制进线程。栈大小也不要压得太小：glibc 的 PTHREAD_STACK_MIN 在 x86-64 上是 16 KB，但任何一次 printf、DNS 解析、信号处理器都可能需要更多。经验值是「正常业务 ≥ 128 KB，深度递归/大局部数组 ≥ 1 MB」。

自定义栈：setstack 之后，guard page 归你管

pthread_attr_setstack 把一块现成内存交给线程当栈。它适合两类场景：把线程栈放在大页（mmap(MAP_HUGETLB)）上减少 TLB miss；或者在实时/嵌入式环境里用静态数组，避免运行期 mmap 带来的不确定性。

代价是责任转移：glibc 不再为这块内存建立 guard page，也不再替你 munmap。

/* 1 MB 自定义栈 + 手工 guard page（片段，完整程序见 2.12.13） */
size_t size = 1 << 20;
void *area = mmap(NULL, size, PROT_READ | PROT_WRITE,
                  MAP_PRIVATE | MAP_ANONYMOUS | MAP_STACK, -1, 0);
if (area == MAP_FAILED) { /* 处理 */ }

/* 栈向低地址增长，所以 guard 放在 area 的低端 */
if (mprotect(area, 4096, PROT_NONE) == -1) { /* 处理 */ }

pthread_attr_t attr;
pthread_attr_init(&attr);
if (pthread_attr_setstack(&attr, area, size) != 0) { /* EINVAL：太小或地址未对齐 */ }

四个容易踩的点：

1. setstack 与 setstacksize 互斥。先 setstack 再 setstacksize 时，后者会把「用户提供的地址」清掉，退回 mmap 模式，而 setstacksize 的返回值仍是 0。
2. 对齐。栈顶由 glibc 计算为 (char *)area + size 再向下对齐；area 自身至少要是页对齐，否则第一次 mmap/mprotect 就失败。ABI 要求 16 字节对齐，页对齐自然满足。
3. guard 要自己留。setstack 之后 pthread_attr_getguardsize 仍返回默认值，但这个值不会被执行——glibc 只在它自己 mmap 时才应用 guard。自定义栈的溢出会直接写坏相邻内存，表现成「数据莫名被改」，而不是 SIGSEGV，更难查。
4. 回收。pthread_join 不会 munmap 用户提供的栈。线程 join 之后由调用者 munmap(area, size)；提前释放则是 use-after-free。

栈溢出：从 SIGSEGV 到一行 dmesg

栈溢出不会返回错误码，只会让进程收到 SIGSEGV。要判断「是不是栈溢出」，看三样东西：

1. dmesg 里的 segfault at 7ffd... error 6：error 6 = 用户态写 + 页不存在；地址通常紧贴某个线程栈的 guard page。
2. /proc/PID/maps 里该线程栈区间下方紧挨着一个 ---p（PROT_NONE）页。
3. gdb 的 bt：栈帧几千层，或最深帧地址贴近栈底。

-fstack-usage 会为每个翻译单元生成 .su 文件，逐函数给出 static（编译期可知的帧大小）与 dynamic（含 VLA/alloca）。把最大的几个函数挑出来，就是栈预算的主要开销：

cc -std=c17 -D_GNU_SOURCE -O2 -fstack-usage -c worker.c
sort -t$'\t' -k2 -n -r worker.su | head -5
# worker.c:31:13:process_request	1024	static
# worker.c:88:13:parse_frame	512	static

栈溢出的调试有一个反直觉之处：core dump 往往是空的或无法回溯。gdb 需要栈才能展开，而栈已经被踩坏。可行的办法是 ulimit -c unlimited 后在崩溃点看 info registers rsp，再和 /proc/PID/maps（崩溃前抓一份）对照。生产环境更实用的是给线程栈底预置一个魔数，线程定期自检：

/* 栈底放哨兵，溢出前就能发现（片段） */
#define CANARY 0xdeadbeefUL

static void check_stack_canary(const void *stack_low)
{
    if (*(const unsigned long *)stack_low != CANARY)
        fprintf(stderr, "stack overflow detected\n");
}

哨兵必须放在栈底之上一点点（guard page 之后的第一个可写位置），并且在线程运行期间周期性调用。它抓的是「还没踩到 guard」的早期溢出，比等 SIGSEGV 早得多。

2.12.4 取消与清理：让线程可被安全打断

pthread_cancel 不是「立刻杀死线程」。默认的取消类型是 deferred（延迟）：目标线程运行到下一个取消点时才响应。取消点是一组会阻塞或被 POSIX 指定为取消点的函数，包括 read、write、nanosleep、pthread_cond_wait、sem_wait、accept、connect、select、poll、wait 等。

/* cancel_cleanup.c — 延迟取消 + 清理处理器释放锁 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static int counter;

static void unlock_cleanup(void *arg)
{
    pthread_mutex_t *m = arg;
    int rc = pthread_mutex_unlock(m);

    if (rc != 0)
        fprintf(stderr, "cleanup unlock: %s\n", strerror(rc));
    printf("cleanup: mutex unlocked at counter=%d\n", counter);
}

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

    int rc = pthread_mutex_lock(&lock);
    if (rc != 0) {
        fprintf(stderr, "lock: %s\n", strerror(rc));
        return NULL;
    }

    pthread_cleanup_push(unlock_cleanup, &lock);
    for (;;) {
        counter++;
        struct timespec ts = { .tv_nsec = 100 * 1000000L };
        /* nanosleep 是取消点；线程在这睡的时候收到 cancel */
        if (nanosleep(&ts, NULL) == -1 && errno == EINTR)
            continue;
    }
    pthread_cleanup_pop(1);         /* 不可达；但 push/pop 必须成对 */
    return NULL;
}

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

    struct timespec ts = { .tv_nsec = 350 * 1000000L };
    nanosleep(&ts, NULL);

    rc = pthread_cancel(t);
    if (rc != 0) {
        fprintf(stderr, "cancel: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }

    void *ret = NULL;
    rc = pthread_join(t, &ret);
    if (rc != 0) {
        fprintf(stderr, "join: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }
    printf("joined: ret=%s counter=%d\n",
           ret == PTHREAD_CANCELED ? "PTHREAD_CANCELED" : "other", counter);
    return EXIT_SUCCESS;
}

pthread_cleanup_push/pop 是两个宏，它们展开成「开一个块」和「关一个块」。因此二者必须出现在同一个词法作用域里，且中间不能有 return/goto 跳出——否则编译失败，或者清理链被破坏。这是 Pthreads 设计上最常被吐槽的地方；C++ 里通常用 RAII 替代。

工程上更推荐「协作式退出」：用一个 volatile sig_atomic_t 或 C11 原子变量做停止标志，线程在循环里检查它，配合 poll/epoll_wait 的超时或 eventfd 唤醒。取消机制适合「线程卡在不可中断的库调用里」的兜底，不适合日常控制流。

取消点清单：哪些函数会响应取消

POSIX 把取消点定义成一组「允许阻塞」的函数，并允许实现在其他函数里也插入取消点。glibc 的实际行为可以归成五类：

读表要点：pthread_mutex_lock 不在清单里。一个线程卡在等锁上，pthread_cancel 不会让它醒来；它必须等到拿到锁、走到下一个取消点才会响应。这也意味着「用一个高优先级线程不断 cancel 卡死的 worker」并不能解决死锁——死锁的线程往往正卡在 mutex_lock 里。

反过来，一个纯计算循环（没有函数调用、没有 I/O）永远不响应延迟取消。要让它可取消，必须显式插入：

for (long i = 0; i < n; i++) {
    do_work(i);
    if ((i & 0x3ff) == 0)
        pthread_testcancel();        /* 每 1024 次检查一次，开销可接受 */
}

pthread_testcancel() 在取消未挂起时是空操作，本质上只读一个线程局部标志，成本接近一次内存读。

sem_wait 被取消时不会消耗信号量计数（glibc 在 futex 唤醒后重新检查取消标志，取消路径上会补偿），但 sem_timedwait 的 ETIMEDOUT 与取消是两条完全不同的返回路径：前者 -1/ETIMEDOUT，后者线程直接终止、join 得到 PTHREAD_CANCELED。清理逻辑要能区分。

异步取消：为什么它几乎总是错的

PTHREAD_CANCEL_ASYNCHRONOUS 允许取消在任意机器指令处生效。后果是线程可能在最坏的位置消失：

• 正持有 malloc 的 arena 锁 → 其他线程永久阻塞；
• 正 write 到文件中间 → 留下半条记录；
• 刚 mmap 完、还没来得及记录指针 → 映射泄漏且无法找回；
• 正处在 pthread_cleanup_push 与 pop 之间 → 清理链状态不一致，甚至清理函数在自己都没准备好的状态下被调用。

POSIX 规定：异步取消的线程只能调用 async-signal-safe 函数，且不能持有需要释放的资源。实践中只有两类线程可以设成异步：纯计算、不分配内存、不持锁的「可抛弃」线程；以及进程马上要 _exit 时的收尾线程。

更常见的需求其实是「短暂屏蔽取消」，而不是异步取消：

/* 让 malloc 与链表插入要么都完成，要么都不发生 */
int old_state;
if (pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_state) != 0) { /* ... */ }

struct node *n = malloc(sizeof *n);
if (n != NULL) {
    n->next = head;
    head = n;
}

if (pthread_setcancelstate(old_state, NULL) != 0) { /* ... */ }
/* 这里才是安全的取消点 */
pthread_testcancel();

这段代码的模式值得记住：取消状态是「临界区的开关」，不是「线程的属性」。只在必须原子的几行里关掉，出来立刻恢复，并在恢复后主动 testcancel，避免取消被无限期推迟。

2.12.5 每线程数据：errno、TLS 与 pthread_once

errno 是编译器层面的 TLS：__thread int 或 C11 的 _Thread_local。你自己也可以定义：

static __thread int tls_counter;        /* GNU 扩展；每个线程一份 */
static _Thread_local char tls_buf[256]; /* C11 写法，语义相同 */

TLS 的生命周期与线程相同：线程创建时分配，线程退出时释放。它适合「每线程缓存」「每线程随机数种子」「每线程错误缓冲」。但 __thread 不能用于动态加载的库里的某些场景（dlopen 的 TLS 模型），也不适合需要析构函数的资源。

需要析构时用 pthread_key：

static pthread_key_t key;
static pthread_once_t once = PTHREAD_ONCE_INIT;

static void free_buf(void *p)
{
    free(p);                            /* 线程退出时自动调用 */
}

static void make_key(void)
{
    int rc = pthread_key_create(&key, free_buf);
    if (rc != 0) {
        fprintf(stderr, "pthread_key_create: %s\n", strerror(rc));
        abort();
    }
}

/* 任意线程、任意次数调用；初始化函数只执行一次 */
pthread_once(&once, make_key);

char *buf = pthread_getspecific(key);
if (buf == NULL) {
    buf = malloc(4096);
    if (buf == NULL) { /* ... */ }
    pthread_setspecific(key, buf);
}

pthread_once 保证初始化函数在整个进程内只执行一次，即使多个线程同时第一次调用。它比自己写「检查标志 + 加锁 + 再检查」可靠，因为 POSIX 保证等待者能看到完整的初始化结果。pthread_once_t 必须静态初始化为 PTHREAD_ONCE_INIT，不能 memset 或赋值。TLS 析构函数按未指定的顺序调用，最多重复 PTHREAD_DESTRUCTOR_ITERATIONS（Linux 上 4）轮；析构函数里不要再依赖其他线程的 TLS。

TLS 的四种模型：为什么 dlopen 会报 “cannot allocate memory in static TLS block”

同一个 __thread 变量，编译器可以生成四种寻址方式，选择取决于「变量在哪个模块、模块何时加载」：

initial-exec 最快，但它要求「模块在进程启动时就已经在内存里」，这样链接器才能把偏移量算进静态 TLS 块。dlopen 的库如果声明了 initial-exec 的 TLS，而静态 TLS 块剩余空间不够，加载就会失败：

dlopen: cannot allocate memory in static TLS block

glibc 为 dlopen 预留了一小块「静态 TLS 余量」（x86-64 上约 1.6 KB）。超过它，或者多个库叠加超过它，就会看到上面的错误。真实案例：某些 GPU/加速库把大结构体声明成 __thread，被 dlopen 时直接失败。

三种修法，按代价从低到高：

1. 启动时 LD_PRELOAD 该库，让它变成 initial-exec 的正常参与者（部署侧改动）；
2. 让库作者把 TLS 变量改成 -ftls-model=global-dynamic 重新编译（根治）；
3. 用 pthread_key 替代 __thread（改代码，访问变慢）。

pthread_key 与 TLS：什么时候必须用前者

pthread_key 的析构语义有三个细节必须知道：

1. 线程退出时，对每个非 NULL 的 key，先把它置为 NULL，再调用析构函数。所以析构函数里再 pthread_getspecific 拿到的是 NULL，不会重复释放。
2. 如果析构函数里又 pthread_setspecific 设置了新值，线程退出流程会再来一轮，最多 PTHREAD_DESTRUCTOR_ITERATIONS 轮（glibc 上是 4）。写一个「析构里重新申请、再注册」的循环，会白跑四轮然后静默放弃。
3. 析构函数的调用顺序未指定，且只在线程正常退出或被取消时执行；进程 exit() 时其他线程直接被终止，它们的析构不保证执行。

/* 每线程错误缓冲：key + 一次性创建 */
static pthread_key_t errbuf_key;
static pthread_once_t errbuf_once = PTHREAD_ONCE_INIT;

static void free_errbuf(void *p) { free(p); }

static void make_key(void)
{
    int rc = pthread_key_create(&errbuf_key, free_errbuf);
    if (rc != 0) { fprintf(stderr, "key_create: %s\n", strerror(rc)); abort(); }
}

char *get_errbuf(void)
{
    pthread_once(&errbuf_once, make_key);

    char *buf = pthread_getspecific(errbuf_key);
    if (buf == NULL) {
        buf = malloc(256);
        if (buf == NULL) return NULL;
        if (pthread_setspecific(errbuf_key, buf) != 0) { free(buf); return NULL; }
    }
    return buf;
}

这段代码里 pthread_once 不是可选项：两个线程同时第一次调用 pthread_getspecific，没有它就会创建两个 key，其中一个永久泄漏（PTHREAD_KEYS_MAX 是进程级资源）。

2.12.6 线程、信号与 fork

信号送到哪个线程

进程级信号（kill(pid, sig)、终端产生的 SIGINT）会送到「任意一个没有阻塞该信号的线程」。这意味着：

• 你不能假设主线程会收到 SIGINT；
• 如果所有线程都阻塞了它，信号会一直 pending，直到某个线程解除阻塞或调用 sigwait；
• pthread_kill(tid, sig) 可以指定线程，但 tid 必须是同进程的线程。

工程上的标准做法是：所有线程都阻塞业务信号，专门起一个线程 sigwait。

/* 信号线程骨架（节选） */
sigset_t set;
sigemptyset(&set);
sigaddset(&set, SIGINT);
sigaddset(&set, SIGTERM);

/* 主线程在创建任何线程之前阻塞；新线程继承这个掩码 */
if (pthread_sigmask(SIG_BLOCK, &set, NULL) != 0) { /* ... */ }

/* 信号线程 */
for (;;) {
    int sig = 0;
    int rc = sigwait(&set, &sig);
    if (rc != 0) { errno = rc; perror("sigwait"); break; }
    if (sig == SIGINT || sig == SIGTERM)
        break;                          /* 通知事件循环退出 */
}

pthread_sigmask 与 sigprocmask 的区别：POSIX 规定多线程程序中 sigprocmask 的行为未指定，必须用 pthread_sigmask。两者参数相同，但后者只改当前线程的掩码，且返回错误码而非设置 errno。

多线程程序里的 fork

fork() 只复制调用线程，其他线程在子进程里不存在。但子进程继承的是「其他线程可能正持有锁」的地址空间快照：

• 另一个线程 malloc 到一半，malloc 的 arena 锁是锁着的；子进程再 malloc 会死锁；
• 另一个线程持有 stdio 锁；子进程 printf 会死锁；
• 另一个线程正在写日志文件；子进程看到的是半条记录。

POSIX 只保证子进程可以安全调用 async-signal-safe 函数，直到 execve。所以多线程程序里 fork 之后的正确姿势是「立刻 exec」，中间只做 dup2、close、_exit 这类事。如果必须在 fork 后运行 C 代码，用 pthread_atfork 注册 prepare/parent/child 三个钩子，在 prepare 里获取所有相关锁，在 parent/child 里释放。

线程与进程控制：哪些操作会影响整个进程

多线程程序里，很多「进程级」操作比看起来更危险。把作用域列清，能避免「为什么另一个线程也退出了」这类问题：

两个最容易出事的组合：

• exit() 在库代码里被调用。库函数出错时调 exit()，会把宿主进程的所有线程一起杀掉。库代码只应返回错误码。
• exec 前没有清理。其他线程可能正持有文件锁、正写到一半；exec 不会等它们。多线程程序要 exec 时，先停掉所有工作线程，再在唯一存活的线程里 exec。

2.12.7 线程池：固定 worker + 有界队列

下面这个线程池是本章的落地代码：4 个固定 worker、容量 16 的任务队列、优雅关闭（关闭时把已入队的任务跑完）。它已经用到了 2.13 的互斥锁与条件变量，这里先看结构，下一章再解释每个 while 的必要性。

/* thread_pool.c — 固定 4 worker + 有界队列 + 优雅关闭 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#define POOL_SIZE 4
#define QUEUE_CAP 16

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

struct pool {
    pthread_mutex_t lock;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
    struct task *head;
    struct task *tail;
    size_t count;
    int shutdown;
    pthread_t workers[POOL_SIZE];
};

static void *worker(void *arg)
{
    struct pool *p = arg;

    for (;;) {
        int rc = pthread_mutex_lock(&p->lock);
        if (rc != 0) {
            fprintf(stderr, "worker lock: %s\n", strerror(rc));
            return NULL;
        }

        while (p->count == 0 && !p->shutdown) {
            rc = pthread_cond_wait(&p->not_empty, &p->lock);
            if (rc != 0) {
                fprintf(stderr, "cond_wait: %s\n", strerror(rc));
                pthread_mutex_unlock(&p->lock);
                return NULL;
            }
        }

        if (p->shutdown && p->count == 0) {
            pthread_mutex_unlock(&p->lock);
            break;                          /* 队列已空，退出 */
        }

        struct task *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->lock);

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

static int pool_init(struct pool *p)
{
    memset(p, 0, sizeof *p);

    int rc = pthread_mutex_init(&p->lock, NULL);
    if (rc != 0) return rc;

    rc = pthread_cond_init(&p->not_empty, NULL);
    if (rc != 0) { pthread_mutex_destroy(&p->lock); return rc; }

    rc = pthread_cond_init(&p->not_full, NULL);
    if (rc != 0) {
        pthread_cond_destroy(&p->not_empty);
        pthread_mutex_destroy(&p->lock);
        return rc;
    }

    for (int i = 0; i < POOL_SIZE; i++) {
        rc = pthread_create(&p->workers[i], NULL, worker, p);
        if (rc != 0) {
            /* 已创建的 worker 需要退出：置 shutdown 并唤醒 */
            pthread_mutex_lock(&p->lock);
            p->shutdown = 1;
            pthread_cond_broadcast(&p->not_empty);
            pthread_mutex_unlock(&p->lock);
            for (int j = 0; j < i; j++)
                pthread_join(p->workers[j], NULL);
            pthread_cond_destroy(&p->not_full);
            pthread_cond_destroy(&p->not_empty);
            pthread_mutex_destroy(&p->lock);
            return rc;
        }
    }
    return 0;
}

static int pool_submit(struct pool *p, void (*fn)(void *), void *arg)
{
    struct task *t = malloc(sizeof *t);
    if (t == NULL) return ENOMEM;

    t->fn = fn;
    t->arg = arg;
    t->next = NULL;

    int rc = pthread_mutex_lock(&p->lock);
    if (rc != 0) { free(t); return rc; }

    while (p->count == QUEUE_CAP && !p->shutdown) {
        rc = pthread_cond_wait(&p->not_full, &p->lock);
        if (rc != 0) {
            pthread_mutex_unlock(&p->lock);
            free(t);
            return rc;
        }
    }
    if (p->shutdown) {
        pthread_mutex_unlock(&p->lock);
        free(t);
        return ECANCELED;
    }

    if (p->tail != NULL)
        p->tail->next = t;
    else
        p->head = t;
    p->tail = t;
    p->count++;

    pthread_cond_signal(&p->not_empty);
    pthread_mutex_unlock(&p->lock);
    return 0;
}

static void pool_shutdown(struct pool *p)
{
    int rc = pthread_mutex_lock(&p->lock);
    if (rc != 0) return;

    p->shutdown = 1;
    pthread_cond_broadcast(&p->not_empty);
    pthread_cond_broadcast(&p->not_full);
    pthread_mutex_unlock(&p->lock);

    for (int i = 0; i < POOL_SIZE; i++) {
        rc = pthread_join(p->workers[i], NULL);
        if (rc != 0)
            fprintf(stderr, "join worker %d: %s\n", i, strerror(rc));
    }

    for (struct task *t = p->head, *next; t != NULL; t = next) {
        next = t->next;
        free(t);                            /* 理论上为空；保险起见清理 */
    }
    pthread_cond_destroy(&p->not_full);
    pthread_cond_destroy(&p->not_empty);
    pthread_mutex_destroy(&p->lock);
}

static void work(void *arg)
{
    long id = (long)arg;
    struct timespec ts = { .tv_nsec = 10 * 1000000L };

    nanosleep(&ts, NULL);                   /* 模拟 10ms 工作 */
    printf("task %2ld done by %lu\n", id, (unsigned long)pthread_self());
}

int main(void)
{
    struct pool p;
    int rc = pool_init(&p);
    if (rc != 0) {
        fprintf(stderr, "pool_init: %s\n", strerror(rc));
        return EXIT_FAILURE;
    }

    for (long i = 0; i < 24; i++) {
        rc = pool_submit(&p, work, (void *)i);
        if (rc != 0) {
            fprintf(stderr, "submit %ld: %s\n", i, strerror(rc));
            break;
        }
    }

    pool_shutdown(&p);
    return EXIT_SUCCESS;
}

关键设计点：

1. while (p->count == 0 && !p->shutdown)：条件变量必须用 while 而不是 if。醒来不代表条件成立——可能是虚假唤醒，也可能是别的 worker 先抢走了任务。
2. 提交者在队列满时也 while 等待：not_full 由 worker 在取走任务后 signal。这里用 signal 而不是 broadcast，因为一次只腾出一个空位。
3. shutdown 与队列为空同时判断：关闭时先把已入队任务跑完（shutdown && count == 0 才退出），这是「优雅关闭」；若要「立刻取消」，在 worker 里检查 shutdown 后直接 break，并丢弃队列。
4. pool_shutdown 先广播再 join：worker 可能卡在 cond_wait，不广播就永远等不到 shutdown 变化。
5. pthread_cond_signal 在锁内调用：POSIX 允许在锁内或锁外，但在锁内可以避免「唤醒后立刻又睡」的窗口，代码也更容易推理。

编译与运行：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Werror -O2 -g -pthread thread_pool.c -o thread_pool
./thread_pool | sort -k2 -n | head
# 24 个任务由 4 个 worker 分担；输出顺序不稳定，但每个任务只出现一次

2.12.8 观测与排查

线程程序出问题时，先回答三个问题：有几个线程、每个在干什么、谁持有锁。

# 1. 看线程数与名字
ps -eLf | grep myprog
ls /proc/$PID/task
cat /proc/$PID/task/*/comm

# 2. 看每个线程的内核栈（需要 root）
cat /proc/$PID/task/*/stack

# 3. 看谁在等锁（futex）
strace -f -e trace=futex -p $PID

# 4. gdb 看用户态栈
gdb -p $PID -ex 'thread apply all bt' -ex quit

给线程起名能极大降低排查成本：

/* Linux 专有：第二个参数是线程名，最长 15 字节 */
pthread_setname_np(pthread_self(), "worker-io");

名字会出现在 /proc/PID/task/TID/comm、ps -L、gdb 的 info threads 里。没有名字时，所有线程都叫进程名，只能靠 TID 猜。

栈溢出的排查前面讲过；数据竞争用 TSan：

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -g -O1 -fsanitize=thread -pthread racy.c -o racy
./racy
# 输出 "WARNING: ThreadSanitizer: data race"，并给出两个线程的栈

TSan 需要编译器和运行时支持，会把程序变慢 5–15 倍，不能用于生产，但能在测试里抓住大部分竞争。它抓不到「逻辑上依赖顺序」的 bug（例如 A 线程必须先写配置、B 线程才能读），那需要设计层面的同步。

2.12.9 Pthreads API 与错误码总表

这一节回答「手边这个 pthread_* 到底返回什么、该检查哪个值」。Pthreads 的错误约定与系统调用相反，把这张表放在手边，比每次翻 man 快。

/* 生命周期 */
int pthread_create(pthread_t *t, const pthread_attr_t *attr,
                   void *(*fn)(void *), void *arg);
int pthread_join(pthread_t t, void **retval);
int pthread_detach(pthread_t t);
void pthread_exit(void *retval) __attribute__((noreturn));
pthread_t pthread_self(void);
int pthread_equal(pthread_t a, pthread_t b);

/* 属性 */
int pthread_attr_init(pthread_attr_t *a);
int pthread_attr_destroy(pthread_attr_t *a);
int pthread_attr_setstacksize(pthread_attr_t *a, size_t n);
int pthread_attr_setstack(pthread_attr_t *a, void *addr, size_t n);
int pthread_attr_setguardsize(pthread_attr_t *a, size_t n);
int pthread_attr_setdetachstate(pthread_attr_t *a, int state);
int pthread_attr_setschedpolicy(pthread_attr_t *a, int policy);
int pthread_attr_setschedparam(pthread_attr_t *a, const struct sched_param *p);
int pthread_attr_setinheritsched(pthread_attr_t *a, int inherit);

/* 取消 */
int pthread_cancel(pthread_t t);
int pthread_setcancelstate(int state, int *old);
int pthread_setcanceltype(int type, int *old);
void pthread_testcancel(void);

/* 一次性初始化与 TLS */
int pthread_once(pthread_once_t *once, void (*init)(void));
int pthread_key_create(pthread_key_t *k, void (*dtor)(void *));
int pthread_key_delete(pthread_key_t k);
int pthread_setspecific(pthread_key_t k, const void *v);
void *pthread_getspecific(pthread_key_t k);

/* 信号 */
int pthread_kill(pthread_t t, int sig);
int pthread_sigmask(int how, const sigset_t *set, sigset_t *old);

这张表想说明一件事：Pthreads 函数不设置 errno。正确写法是把返回值直接送进 strerror：

int rc = pthread_create(&t, NULL, worker, arg);
if (rc != 0) {
    /* 不要用 perror：errno 可能还是上一次系统调用的值 */
    fprintf(stderr, "pthread_create: %s\n", strerror(rc));
    return -1;
}

2.12.10 NPTL 与 futex：一次 pthread_create 的内核路径

这一节回答「线程创建、退出、join 在内核里到底发生了什么」。看懂这条路径，就能解释三件事：为什么线程栈是 mmap 出来的、为什么 join 不需要轮询、为什么 robust mutex 能在 owner 死掉后恢复。

pthread_create
  └─ __pthread_create_2_1                     glibc nptl/pthread_create.c
       ├─ allocate_stack()                     mmap(栈 + TLS + TCB)，guard 页设 PROT_NONE
       ├─ clone3() / clone()                   glibc ≥ 2.34 优先 clone3，ENOSYS 回落 clone
       │    └─ kernel_clone()                  kernel/fork.c
       │         └─ copy_process()
       │              ├─ dup_task_struct()     task_struct + 内核栈（THREAD_SIZE）
       │              ├─ copy_mm()             CLONE_VM → mmget()，不复制页表
       │              ├─ copy_files() / copy_fs() / copy_sighand() / copy_signal()
       │              ├─ alloc_pid()           新 TID；tgid 指向线程组 leader
       │              └─ wake_up_new_task()    放入运行队列
       └─ 返回 pthread_t（TCB 地址）

pthread_exit / 从入口函数 return
  └─ start_thread → __do_cancel / exit
       └─ do_exit()                            kernel/exit.c
            ├─ exit_mm() → mm_release()        clear_child_tid 清零 + futex_wake
            ├─ exit_robust_list()              标记 FUTEX_OWNER_DIED，唤醒等待者
            └─ exit_notify()                   线程不产生僵尸，只有 leader 通知父进程

clone 的 flags 决定了「共享什么」，这张位表就是线程与进程的分界线：

NPTL 的线程就是 CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID 的组合。少了 CLONE_THREAD 就退化成进程（fork），少了 CLONE_VM 就退化成「共享 fd 但不共享内存」的怪东西。

join 为什么不用轮询

pthread_join 的核心是一行 futex 等待，等的是 TCB 里的 tid 字段：

/* pthread_join 的等待逻辑，简化示意（非逐字引用） */
while ((tid = atomic_load(&pd->tid)) != 0)
    futex_wait(&pd->tid, tid);       /* 值变了才返回 */

线程退出时，内核在 mm_release() 里做两件事：

/* kernel/fork.c, mm_release()，有删节 */
if (tsk->clear_child_tid) {
    if (atomic_read(&mm->mm_users) > 1) {
        u32 __user *tidptr = tsk->clear_child_tid;
        tsk->clear_child_tid = NULL;
        if (!put_user(0, tidptr))
            futex_wake(tidptr, 1, 1);
    }
}

put_user(0, tidptr) 把 TCB 的 tid 清零，futex_wake 唤醒正在 pthread_join 的线程。所以 join 是「睡到值变」，没有轮询、没有定时器，也不会因为信号而错过唤醒。

robust mutex 的内核支持

每个线程在创建时会通过 set_robust_list(2) 向内核登记一个链表头。线程退出时 exit_robust_list() 遍历这个链表，把每个 robust mutex 的 owner 字段标成 FUTEX_OWNER_DIED，然后唤醒一个等待者。等待者从 pthread_mutex_lock 得到 EOWNERDEAD——这就是 2.15 里跨进程崩溃恢复的基础。

2.12.11 程序一：无同步计数器的竞态与三种修法

这段程序要验证一件事：同一份工作量，换同步方式，结果与耗时都会变。它把「无同步」「互斥锁」「C11 原子」放在一个可执行文件里，用参数切换，方便你用同一个 perf/TSan 流程对比。

/* race_counter.c — 无同步 / 互斥锁 / C11 原子，三种计数器对比 */
#define _GNU_SOURCE
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#define THREADS    4
#define PER_THREAD 500000L

/* volatile 只阻止编译器把它缓存进寄存器，不提供原子性，也不建立屏障。
   这里用它是为了让竞态在 -O2 下仍然可见，而不是为了“修复”竞态。 */
static volatile long counter_plain;

static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static long counter_locked;

static atomic_long counter_atomic;

struct job {
    int mode;               /* 0=无同步 1=互斥锁 2=原子 */
    long iters;
};

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

static void *worker(void *arg)
{
    const struct job *j = arg;

    for (long i = 0; i < j->iters; i++) {
        switch (j->mode) {
        case 0:
            counter_plain++;                    /* load / add / store，三步 */
            break;
        case 1: {
            int rc = pthread_mutex_lock(&lock);
            if (rc != 0)
                die_pthread("mutex_lock", rc);
            counter_locked++;
            rc = pthread_mutex_unlock(&lock);
            if (rc != 0)
                die_pthread("mutex_unlock", rc);
            break;
        }
        default:
            atomic_fetch_add_explicit(&counter_atomic, 1, memory_order_relaxed);
            break;
        }
    }
    return NULL;
}

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;
}

/* 跑一轮，返回观察到的计数；elapsed 写入毫秒 */
static long run_round(int mode, long *elapsed)
{
    pthread_t tids[THREADS];
    struct job j = { .mode = mode, .iters = PER_THREAD };
    long start = now_ms();

    counter_plain = 0;
    counter_locked = 0;
    atomic_store_explicit(&counter_atomic, 0, memory_order_relaxed);

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

    *elapsed = now_ms() - start;

    switch (mode) {
    case 0:  return counter_plain;
    case 1:  return counter_locked;
    default: return atomic_load_explicit(&counter_atomic, memory_order_relaxed);
    }
}

int main(int argc, char **argv)
{
    int mode = 0;

    if (argc == 2) {
        if (strcmp(argv[1], "plain") == 0)       mode = 0;
        else if (strcmp(argv[1], "mutex") == 0)  mode = 1;
        else if (strcmp(argv[1], "atomic") == 0) mode = 2;
        else {
            fprintf(stderr, "usage: %s plain|mutex|atomic\n", argv[0]);
            return EXIT_FAILURE;
        }
    }

    const long expect = THREADS * PER_THREAD;
    long elapsed = 0;
    long got = run_round(mode, &elapsed);

    const char *name = mode == 0 ? "plain" : mode == 1 ? "mutex" : "atomic";
    printf("%-6s got=%ld expect=%ld lost=%ld elapsed=%ldms\n",
           name, got, expect, expect - got, elapsed);

    /* 无同步模式丢更新是预期行为；另外两种丢一个都说明程序有 bug */
    if (mode != 0 && got != expect) {
        fprintf(stderr, "BUG: synchronized counter lost updates\n");
        return EXIT_FAILURE;
    }
    return EXIT_SUCCESS;
}

编译与三次运行：

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

./race_counter plain
# plain  got=1387421 expect=2000000 lost=612579 elapsed=12ms
./race_counter mutex
# mutex  got=2000000 expect=2000000 lost=0      elapsed=158ms
./race_counter atomic
# atomic got=2000000 expect=2000000 lost=0      elapsed=14ms

plain 的 lost 每次运行都不同，甚至偶尔为 0——竞态不是稳定复现的 bug。mutex 正确但慢一个数量级：每次 ++ 两次原子操作，还可能进 futex 睡眠。atomic 在低竞争下接近 plain，因为它只是一条带 lock 前缀的 inc。

修法不止一种。如果这个计数器只是统计量，还有一个更快的方案：每线程一个局部计数，最后汇总。

/* 分片计数：没有共享写，就没有竞争 */
static _Thread_local long shard;

static void *worker_shard(void *arg)
{
    const struct job *j = arg;

    shard = 0;
    for (long i = 0; i < j->iters; i++)
        shard++;
    return (void *)(intptr_t)shard;     /* 用返回值带回，避免再引入共享 */
}

join 的返回值汇总即可。代价是「读到的是过期总数」——监控场景可接受，需要精确值时不行。

2.12.12 程序二：ABBA 死锁、trylock 与锁序

这段程序要验证三件事：死锁可以稳定复现、可以用超时检测、可以用两种方式修复。它用同一个二进制、三个模式，方便你反复对照。

/* abba_deadlock.c — 复现 ABBA 死锁，并给出两种修复 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define MODE_DEADLOCK 0
#define MODE_ORDER    1
#define MODE_TRYLOCK  2

static pthread_mutex_t lock_a = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t lock_b = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t done_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  done_cond = PTHREAD_COND_INITIALIZER;
static int done_count;
static int mode = MODE_DEADLOCK;

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

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

    while (nanosleep(&ts, &ts) == -1) {
        if (errno != EINTR) {
            perror("nanosleep");
            exit(EXIT_FAILURE);
        }
    }
}

static void mark_done(void)
{
    int rc = pthread_mutex_lock(&done_lock);
    if (rc != 0)
        die_pthread("done lock", rc);
    done_count++;
    rc = pthread_cond_signal(&done_cond);
    if (rc != 0)
        die_pthread("done signal", rc);
    rc = pthread_mutex_unlock(&done_lock);
    if (rc != 0)
        die_pthread("done unlock", rc);
}

/* 拿两把锁：根据模式选择“制造环”“按地址排序”“trylock 退避” */
static int lock_pair(pthread_mutex_t *first, pthread_mutex_t *second)
{
    if (mode == MODE_TRYLOCK) {
        for (int attempt = 0; attempt < 100000; attempt++) {
            int rc = pthread_mutex_lock(first);
            if (rc != 0)
                return rc;

            rc = pthread_mutex_trylock(second);
            if (rc == 0)
                return 0;
            if (rc != EBUSY) {
                pthread_mutex_unlock(first);
                return rc;
            }

            rc = pthread_mutex_unlock(first);
            if (rc != 0)
                return rc;

            /* 随机退避，避免两个线程同步重试形成活锁 */
            struct timespec ts = { .tv_nsec = (attempt % 7 + 1) * 100000L };
            nanosleep(&ts, NULL);
        }
        return EBUSY;
    }

    int rc = pthread_mutex_lock(first);
    if (rc != 0)
        return rc;
    sleep_us(200);                      /* 放大“持有一把、再要另一把”的窗口 */
    rc = pthread_mutex_lock(second);
    if (rc != 0) {
        pthread_mutex_unlock(first);
        return rc;
    }
    return 0;
}

struct worker_arg {
    pthread_mutex_t *first;
    pthread_mutex_t *second;
    const char *name;
};

static void *worker(void *arg)
{
    struct worker_arg *w = arg;

    int rc = lock_pair(w->first, w->second);
    if (rc != 0) {
        fprintf(stderr, "%s: lock_pair: %s\n", w->name, strerror(rc));
        return NULL;
    }

    printf("%s: got both locks\n", w->name);

    rc = pthread_mutex_unlock(w->second);
    if (rc != 0)
        die_pthread("unlock second", rc);
    rc = pthread_mutex_unlock(w->first);
    if (rc != 0)
        die_pthread("unlock first", rc);

    mark_done();
    return NULL;
}

int main(int argc, char **argv)
{
    if (argc != 2) {
        fprintf(stderr, "usage: %s deadlock|order|trylock\n", argv[0]);
        return EXIT_FAILURE;
    }
    if (strcmp(argv[1], "deadlock") == 0)     mode = MODE_DEADLOCK;
    else if (strcmp(argv[1], "order") == 0)   mode = MODE_ORDER;
    else if (strcmp(argv[1], "trylock") == 0) mode = MODE_TRYLOCK;
    else {
        fprintf(stderr, "unknown mode: %s\n", argv[1]);
        return EXIT_FAILURE;
    }

    /* 锁序模式：按对象地址排序，所有线程得到同一个顺序 */
    pthread_mutex_t *lo = &lock_a;
    pthread_mutex_t *hi = &lock_b;
    if (mode == MODE_ORDER && (uintptr_t)lo > (uintptr_t)hi) {
        pthread_mutex_t *tmp = lo;
        lo = hi;
        hi = tmp;
    }

    struct worker_arg w1, w2;
    if (mode == MODE_DEADLOCK) {
        w1 = (struct worker_arg){ &lock_a, &lock_b, "t1(a->b)" };
        w2 = (struct worker_arg){ &lock_b, &lock_a, "t2(b->a)" };
    } else {
        w1 = (struct worker_arg){ lo, hi, "t1(lo->hi)" };
        w2 = (struct worker_arg){ lo, hi, "t2(lo->hi)" };
    }

    pthread_t t1, t2;
    int rc = pthread_create(&t1, NULL, worker, &w1);
    if (rc != 0)
        die_pthread("create t1", rc);
    rc = pthread_create(&t2, NULL, worker, &w2);
    if (rc != 0)
        die_pthread("create t2", rc);

    /* 最多等 2 秒：超时说明两个线程都没有返回，即死锁 */
    struct timespec deadline;
    if (clock_gettime(CLOCK_REALTIME, &deadline) == -1) {
        perror("clock_gettime");
        return EXIT_FAILURE;
    }
    deadline.tv_sec += 2;

    rc = pthread_mutex_lock(&done_lock);
    if (rc != 0)
        die_pthread("done lock", rc);

    while (done_count < 2) {
        rc = pthread_cond_timedwait(&done_cond, &done_lock, &deadline);
        if (rc == ETIMEDOUT) {
            pthread_mutex_unlock(&done_lock);
            fprintf(stderr, "watchdog: %d/2 workers finished -> deadlock\n",
                    done_count);
            /* 死锁无法在进程内安全拆解：直接退出，交给外部监控 */
            _exit(2);
        }
        if (rc != 0)
            die_pthread("done timedwait", rc);
    }
    rc = pthread_mutex_unlock(&done_lock);
    if (rc != 0)
        die_pthread("done unlock", rc);

    rc = pthread_join(t1, NULL);
    if (rc != 0)
        die_pthread("join t1", rc);
    rc = pthread_join(t2, NULL);
    if (rc != 0)
        die_pthread("join t2", rc);

    printf("both workers done\n");
    return EXIT_SUCCESS;
}

运行三个模式：

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

./abba_deadlock deadlock; echo "exit=$?"
# t1(a->b): got both locks        ← 有时只打印这一行就卡住，取决于交错
# watchdog: 1/2 workers finished -> deadlock
# exit=2

./abba_deadlock order; echo "exit=$?"
# t1(lo->hi): got both locks
# t2(lo->hi): got both locks
# both workers done
# exit=0

./abba_deadlock trylock; echo "exit=$?"
# 两种顺序都可能出现，但一定以 both workers done 结束

三个模式对应三种工程手段，各有代价：

注意 trylock 模式里的随机退避：如果两个线程都用固定 100 µs 重试，它们会一直同步碰撞，形成活锁——CPU 100%，进度为零。退避必须是随机的，或者按线程 ID 错开。

2.12.13 程序三：自定义栈、TLS 与线程 ID 的观测

这段程序要验证四件事：自定义栈的边界从哪来、guard page 触发时进程怎么死、TLS 是否真的每线程一份、pthread_t 与 TID 如何对应。它有两个模式：info 打印观测值，overflow 故意踩 guard page。

/* thread_stack_tls.c — 自定义栈 + TLS + gettid；overflow 模式演示栈溢出 */
#define _GNU_SOURCE
#include <dirent.h>
#include <errno.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>

#define STACK_SIZE (256 * 1024)
#define GUARD_SIZE 4096
#define CANARY     0xdeadbeefUL

static _Thread_local unsigned long tls_counter;   /* 每线程一份 */
static const char *g_mode = "info";
static void *g_stack;                             /* 自定义栈，fork 后子进程也能用 */

static void die(const char *msg)
{
    perror(msg);
    exit(EXIT_FAILURE);
}

/* 数 /proc/self/task 下的线程目录；非 Linux 平台返回 -1 */
static int count_tasks(void)
{
    DIR *d = opendir("/proc/self/task");
    if (d == NULL)
        return -1;

    int n = 0;
    struct dirent *e;
    while ((e = readdir(d)) != NULL) {
        if (e->d_name[0] != '.')
            n++;
    }
    closedir(d);
    return n;
}

/* 每层 2 KB，返回值让编译器无法把它优化成尾递归 */
static unsigned long burn_stack(unsigned long depth)
{
    volatile unsigned char pad[2048];

    pad[0] = (unsigned char)depth;
    pad[sizeof pad - 1] = (unsigned char)(depth >> 8);
    if (depth > 100000000UL)
        return depth;
    return burn_stack(depth + 1) + pad[0];
}

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

    int rc = pthread_setname_np(pthread_self(), "stack-tls");
    if (rc != 0)
        fprintf(stderr, "setname: %s\n", strerror(rc));

    printf("worker: pthread_t=%p tid=%d\n",
           (void *)pthread_self(), (int)gettid());

    tls_counter = 42;
    printf("worker: &tls_counter=%p value=%lu\n",
           (void *)&tls_counter, tls_counter);

    /* 从 NPTL 反查这块栈的真实范围与 guard 大小 */
    pthread_attr_t attr;
    rc = pthread_getattr_np(pthread_self(), &attr);
    if (rc != 0) {
        fprintf(stderr, "getattr_np: %s\n", strerror(rc));
        return NULL;
    }

    void *base = NULL;
    size_t size = 0, guard = 0;
    if (pthread_attr_getstack(&attr, &base, &size) != 0)
        fprintf(stderr, "getstack failed\n");
    if (pthread_attr_getguardsize(&attr, &guard) != 0)
        fprintf(stderr, "getguardsize failed\n");
    pthread_attr_destroy(&attr);

    printf("worker: stack=[%p,%p) size=%zu guard=%zu\n",
           base, (char *)base + size, size, guard);
    printf("worker: tasks=%d\n", count_tasks());

    if (strcmp(g_mode, "overflow") == 0) {
        printf("worker: burning stack until guard page...\n");
        fflush(stdout);
        unsigned long d = burn_stack(0);
        printf("worker: returned depth=%lu (no overflow)\n", d);
    }
    return NULL;
}

/* 在子进程里跑 overflow，父进程只负责解码 wait 状态 */
static int run_overflow(void)
{
    pid_t pid = fork();
    if (pid == -1)
        die("fork");

    if (pid == 0) {
        pthread_attr_t attr;
        pthread_t t;

        if (pthread_attr_init(&attr) != 0)
            _exit(10);
        if (pthread_attr_setstack(&attr, g_stack, STACK_SIZE) != 0)
            _exit(11);
        if (pthread_create(&t, &attr, worker, NULL) != 0)
            _exit(12);
        pthread_join(t, NULL);
        _exit(0);                       /* 正常情况不会到这 */
    }

    int status = 0;
    if (waitpid(pid, &status, 0) == -1)
        die("waitpid");

    if (WIFSIGNALED(status)) {
        printf("parent: child killed by signal %d (%s)\n",
               WTERMSIG(status), strsignal(WTERMSIG(status)));
        return 0;
    }
    printf("parent: child exited with %d\n", WEXITSTATUS(status));
    return 1;
}

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

    /* 一块 256 KB 的自定义栈，低端手工留一页 guard */
    g_stack = mmap(NULL, STACK_SIZE, PROT_READ | PROT_WRITE,
                   MAP_PRIVATE | MAP_ANONYMOUS | MAP_STACK, -1, 0);
    if (g_stack == MAP_FAILED)
        die("mmap stack");
    if (mprotect(g_stack, GUARD_SIZE, PROT_NONE) == -1)
        die("mprotect guard");

    /* 在 guard 之后的第一个可写字节放哨兵，溢出到它时能早一步发现 */
    *(unsigned long *)((char *)g_stack + GUARD_SIZE) = CANARY;

    tls_counter = 7;
    printf("main:   pthread_t=%p tid=%d &tls_counter=%p value=%lu\n",
           (void *)pthread_self(), (int)gettid(),
           (void *)&tls_counter, tls_counter);

    int ret;
    if (strcmp(g_mode, "overflow") == 0) {
        ret = run_overflow();
    } else {
        pthread_attr_t attr;
        pthread_t t;

        int rc = pthread_attr_init(&attr);
        if (rc != 0) { fprintf(stderr, "attr_init: %s\n", strerror(rc)); return 1; }
        rc = pthread_attr_setstack(&attr, g_stack, STACK_SIZE);
        if (rc != 0) { fprintf(stderr, "setstack: %s\n", strerror(rc)); return 1; }

        rc = pthread_create(&t, &attr, worker, NULL);
        if (rc != 0) { fprintf(stderr, "create: %s\n", strerror(rc)); return 1; }
        rc = pthread_join(t, NULL);
        if (rc != 0) { fprintf(stderr, "join: %s\n", strerror(rc)); return 1; }
        pthread_attr_destroy(&attr);

        if (*(unsigned long *)((char *)g_stack + GUARD_SIZE) != CANARY) {
            fprintf(stderr, "main: canary overwritten -> stack overflow\n");
            return 1;
        }
        printf("main: canary intact, threads=%d\n", count_tasks());
        ret = 0;
    }

    if (munmap(g_stack, STACK_SIZE) == -1)
        die("munmap");
    return ret;
}

编译与两个模式：

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

./thread_stack_tls info
# main:   pthread_t=0x16d0b7000 tid=51321 &tls_counter=0x16d0b6f80 value=7
# worker: pthread_t=0x16d0af000 tid=51322
# worker: &tls_counter=0x16d0aef80 value=42
# worker: stack=[0x16d06f000,0x16d0af000) size=262144 guard=4096
# worker: tasks=2
# main: canary intact, threads=1

./thread_stack_tls overflow
# worker: burning stack until guard page...
# parent: child killed by signal 11 (Segmentation fault)

两个线程的 &tls_counter 相差 4 KB，值分别是 7 和 42，互不干扰。stack=[0x16d06f000,0x16d0af000) 的 256 KB 正是 mmap 的长度；guard=4096 是属性对象里的默认值，真正生效的是我们 mprotect 的那一页。tasks=2 只在 Linux 上有意义，其他平台 /proc 不存在，返回 -1。

overflow 模式里，子进程的线程踩到 PROT_NONE 页，内核向整个进程发 SIGSEGV，父进程用 WIFSIGNALED/WTERMSIG 解码出 11。这解释了为什么「线程栈溢出会杀掉整个进程」——信号是进程级的。

2.12.14 实验：用 strace、perf 与 TSan 看线程

三个实验分别回答「线程创建/同步走了哪些系统调用」「时间花在等锁还是算」「有没有数据竞争」。

实验一：strace -f 看 clone 与 futex

strace -f -e trace=clone,clone3,futex,set_robust_list,exit_group \
       -o /tmp/trace.txt ./thread_basic
grep -E 'clone3|set_robust_list|FUTEX_WAIT' /tmp/trace.txt | head -8

典型输出（地址与 TID 每次不同）：

12345 clone3({flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID, exit_signal=SIGCHLD, stack=0x7f8c4a2f0000, stack_size=0x800000, tls=0x7f8c4a2ef640}, 88) = 12346
12346 set_robust_list(0x7f8c4a2ef660, 24) = 0
12345 futex(0x7f8c4a2ef640, FUTEX_WAIT_PRIVATE, 12346, NULL) = 0
12346 exit_group(0)                     = ?

读法：

• clone3 的 flags 就是 2.12.10 那张位表；stack_size=0x800000 是 8 MB 默认栈。
• set_robust_list 出现在新线程里：NPTL 在每个线程启动时向内核登记 robust 链表。
• FUTEX_WAIT_PRIVATE 的第三个参数 12346 是期望值——pthread_join 在等 TCB 里的 tid 等于 12346 时睡眠；线程退出把它清零并唤醒。
• exit_group 出现在 main 返回时，它终止整个线程组；线程自己退出用的是 exit（不是 exit_group）。

strace -f -c 给出系统调用耗时汇总，能看出线程程序里 futex 占了多少：

strace -f -c -e trace=futex ./thread_pool 2>&1 | tail -6
# % time     seconds  usecs/call     calls    errors syscall
#  100.00    0.004120          41       100           futex

实验二：perf 与 /proc 看上下文切换

perf stat -e context-switches,cpu-migrations,page-faults,instructions \
          ./thread_pool

perf sched record -- ./thread_pool
perf sched latency | head -12

没有 perf 权限时（容器、perf_event_paranoid 限制），用 /proc/PID/status 与 pidstat：

pidstat -w -t -p $(pgrep -n thread_pool) 1 3
# 每行给出该线程每秒的自愿/非自愿上下文切换次数
grep ctxt_switches /proc/$$/status
# voluntary_ctxt_switches:     12
# nonvoluntary_ctxt_switches:  3

判读标准：自愿切换高说明线程在等锁、等 I/O、等条件变量（正常）；非自愿切换高说明线程数超过 CPU 核数，或被更高优先级抢占（要减线程或调亲和性）。线程池里 worker 数远大于核数时，非自愿切换会显著上升，吞吐反而下降。

实验三：TSan 抓 race_counter 的竞争

cc -std=c17 -D_GNU_SOURCE -Wall -Wextra -g -O1 -fsanitize=thread \
   -pthread race_counter.c -o race_tsan
./race_tsan plain; echo "exit=$?"

==================
WARNING: ThreadSanitizer: data race (pid=23456)
  Write of size 8 at 0x... by thread T3:
    #0 worker /tmp/race_counter.c:47 (race_tsan+0x...)
  Previous write of size 8 at 0x... by thread T1:
    #0 worker /tmp/race_counter.c:47 (race_tsan+0x...)
  Location is global 'counter_plain' of size 8 at 0x... 
==================
ThreadSanitizer: reported 3 warnings
exit=66

读法：两条栈指向同一全局变量，至少一个是写，且 TSan 找不到任何 happens-before 边（没有锁、没有原子操作、没有 join 之外的同步）。exit=66 是 TSan 的默认退出码，CI 里可以用它让任务失败。

用同一个二进制跑 mutex 与 atomic 模式，TSan 应当零报告——这正好验证「锁和原子操作都能建立 happens-before」。TSan 报不报，不取决于代码看起来是否安全，而取决于它是否观测到同步边。

2.12.15 陷阱清单：线程的十二个坑

这一节把全章的失败模式压成一张表，按「现象 → 原因 → 修复」组织。遇到线上问题时按现象查，比从头读快。

表里第 5、6、7 条都指向同一个结论：取消是协作机制，不是 kill。需要「立刻停」的语义时，用 eventfd/pipe 唤醒 + 停止标志，把控制权交回线程自己。

第 9 条在多线程 + fork 的组合里最危险，因为它在测试环境常常不复现：单线程测试永远不触发，只有在生产的多核负载下才偶发。判断方法是在 fork 后立刻 strace，看子进程是否停在 futex 上——如果是，就是继承了别人的锁。

版本差异也要记一笔：gettid() 的 glibc wrapper 从 2.30 才有，更早要 syscall(SYS_gettid)；glibc 2.34+ 优先用 clone3，旧版 Docker 的默认 seccomp 会让它返回 ENOSYS 并回落到 clone，自定义 seccomp 策略必须同时放行两者。

2.12.16 本章出口检查

母本对应

下一站：2.13 线程同步与线程安全把线程池里出现的 mutex 与 cond 讲透：为什么条件变量必须配 while、什么是虚假唤醒与惊群、如何用 TSan 与 helgrind 把数据竞争钉死在测试阶段。