关键问题与本章位置

前几章我们一直在"造工具"：锁（lock）、条件变量（condition variable）、信号量（semaphore）。这一章换个角度——真实世界里的程序员拿这些工具写代码时，到底犯了哪些错？

原书借助 Lu 等人对 MySQL、Apache、Mozilla、OpenOffice 四个成熟开源项目的研究，把并发缺陷分成两大类：非死锁缺陷（non-deadlock bug），其中绝大多数是"原子性违反"和"顺序违反"；以及经典的死锁（deadlock）。对死锁，本章系统地讲四个必要条件，以及预防（prevention）、避免（avoidance）、检测与恢复（detect and recover）三类对策。下一章（基于事件的并发）则干脆换一种不靠多线程的并发方式，从根上绕开其中一部分问题。

核心概念速览

真实世界里有哪些并发缺陷？

Lu 等人（ASPLOS '08）逐个分析了四个著名开源项目中已经被发现并修复的并发缺陷，把开发者的修复记录变成了定量统计。原书 Figure 32.1 的数据如下：

两个结论值得记住：第一，大多数（74/105）并发缺陷不是死锁，所以只盯着死锁是不够的；第二，缺陷数与代码规模、并发复杂度相关——OpenOffice 只有 8 个，Mozilla 将近 60 个。

非死锁缺陷：原子性违反与顺序违反

原子性违反（Atomicity Violation）——MySQL 的例子

先别看解释，自己找找下面这段 MySQL 代码（原书 Figure 32.2）的问题：

Thread 1::
if (thd->proc_info) {            // ① 检查：非 NULL 吗？
    fputs(thd->proc_info, ...);  // ② 使用：打印它
}

Thread 2::
thd->proc_info = NULL;           // ③ 置空

线程 1 先检查 proc_info 非 NULL，再去用它。如果恰好在 ① 和 ② 之间发生中断、线程 2 执行了 ③，线程 1 恢复后就会把 NULL 交给 fputs，程序崩溃。

Lu 等人的正式定义是："多次内存访问之间所期望的可串行性被违反了（即一段代码本意是原子的，但执行时原子性没有被强制保证）。"这里的"原子性假设"就是：对 proc_info 的非 NULL 检查与 fputs() 中对它的使用是一个整体。

修复通常很直接：给共享变量的访问加锁（原书 Figure 32.3）。注意所有访问这个结构的代码都要拿同一把锁，只给线程 1 加锁是没用的：

pthread_mutex_t proc_info_lock = PTHREAD_MUTEX_INITIALIZER;

Thread 1::
pthread_mutex_lock(&proc_info_lock);
if (thd->proc_info) {
    fputs(thd->proc_info, ...);
}
pthread_mutex_unlock(&proc_info_lock);

Thread 2::
pthread_mutex_lock(&proc_info_lock);
thd->proc_info = NULL;
pthread_mutex_unlock(&proc_info_lock);

顺序违反（Order Violation）——Mozilla 的例子

再看原书 Figure 32.4：

Thread 1::
void init() {
    mThread = PR_CreateThread(mMain, ...);   // 创建线程 2，返回后才赋值给 mThread
}

Thread 2::
void mMain(...) {
    mState = mThread->State;                 // 假设 mThread 已经被赋值
}

线程 2 的代码默认 mThread 已经初始化好了。但新线程一被创建就可能立即运行——那时 PR_CreateThread 还没返回，mThread 仍是 NULL，线程 2 解引用 NULL 就崩溃（如果初值不是 NULL，更糟：会访问任意内存）。

正式定义："两组内存访问之间所期望的顺序被颠倒了（即 A 应该总在 B 之前执行，但执行时这个顺序没有被强制保证）。"修复方法就是强制顺序——条件变量正是为此而生（原书 Figure 32.5）：

pthread_mutex_t mtLock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t  mtCond = PTHREAD_COND_INITIALIZER;
int mtInit = 0;                          // 状态变量：线程是否已初始化

Thread 1::
void init() {
    mThread = PR_CreateThread(mMain, ...);
    pthread_mutex_lock(&mtLock);         // 通知：线程已创建好
    mtInit = 1;
    pthread_cond_signal(&mtCond);
    pthread_mutex_unlock(&mtLock);
}

Thread 2::
void mMain(...) {
    pthread_mutex_lock(&mtLock);         // 等待初始化完成
    while (mtInit == 0)
        pthread_cond_wait(&mtCond, &mtLock);
    pthread_mutex_unlock(&mtLock);
    mState = mThread->State;             // 现在一定安全
}

如果线程 2 先跑，它会在 wait 上睡着，等线程 1 设置 mtInit=1 并发信号；如果线程 2 后跑，它看到 mtInit 已是 1，直接继续。两种顺序都正确。（原书提到也可以直接用 mThread 本身当状态变量，这里为了清晰单独用了 mtInit。）

死锁：为什么会发生

除了上面两类，另一类经典问题出现在锁协议复杂的系统里：死锁。比如线程 1 持有锁 L1、正在等 L2，而持有 L2 的线程 2 正在等 L1 被释放。下面这段代码（原书 Figure 32.6）就可能死锁：

Thread 1:                    Thread 2:
pthread_mutex_lock(L1);      pthread_mutex_lock(L2);
pthread_mutex_lock(L2);      pthread_mutex_lock(L1);

注意是"可能"：如果线程 1 一口气拿到 L1 和 L2，就不会出事。只有当线程 1 拿到 L1 后发生上下文切换、线程 2 又拿到了 L2，二者才会各自握着一把锁去等另一把——然后永远等下去。依赖图里出现了环，就是死锁的标志：

既然这么明显，为什么还会发生？

上面的例子只要两个线程按相同顺序拿锁就能避免。可死锁在真实系统里仍然屡见不鲜，原书给了两个原因：

① 大型代码库中组件之间有复杂的依赖。以操作系统为例：虚拟内存系统换入页面时要访问文件系统读磁盘块；文件系统读块时又需要一页内存来存放数据，于是回头找虚拟内存系统。两个子系统各自持有自己的锁再调用对方，就天然存在循环依赖，锁策略必须精心设计。

② 封装（encapsulation）与锁不太合拍。我们被教导要隐藏实现细节，可有些看似无害的接口简直是在"邀请"你死锁。比如 Java 的 Vector.AddAll()：

Vector v1, v2;
v1.AddAll(v2);    // 内部：先锁 v1，再锁 v2

为了线程安全，这个方法内部要同时锁住 v1 和参数 v2，以某种固定顺序（比如先 v1 后 v2）。如果另一个线程几乎同时调用 v2.AddAll(v1)，它就会先锁 v2 再锁 v1——和 Figure 32.6 一模一样的死锁，而调用者完全看不出来。（下面的模拟器里有这个预设。）

死锁的四个必要条件

Coffman 等人（1971）指出，死锁发生需要同时满足四个条件。只要有任何一个不成立，死锁就不可能发生。因此每一种"预防"技术，都是想办法让其中一个条件不成立：

预防（Prevention）：逐个打破四个条件

循环等待：给锁定一个顺序

最实用、也最常用的预防技术：写加锁代码时保证永远不会产生循环等待。最直接的办法是给锁的获取规定一个全序（total ordering）：系统里只有 L1、L2 两把锁时，规定总是先拿 L1 再拿 L2，就不会形成环。

更复杂的系统里锁很多，全序难以做到（也未必需要），偏序（partial ordering）就是一种有用的折中。原书举了 Linux 内存映射代码（mm/filemap.c）的真实例子：文件开头的注释列出了十组加锁顺序，简单的如"i_mutex 在 i_mmap_rwsem 之前"，复杂的如"i_mmap_rwsem 在 private_lock 之前，在 swap_lock 之前，在 i_pages lock 之前"。

当然，全序和偏序都需要精心设计，而且顺序只是约定：一个粗心的程序员无视它，就可能导致死锁。它还要求对代码库、各函数如何被调用有深入理解，一个失误就会招来那个"D"开头的词。

if (m1 > m2) {   // 按地址从高到低加锁
    pthread_mutex_lock(m1);
    pthread_mutex_lock(m2);
} else {
    pthread_mutex_lock(m2);
    pthread_mutex_lock(m1);
}
// 假设 m1 != m2（不是同一把锁）

持有并等待：一次性拿齐所有锁

让线程原子地一次拿到所有需要的锁，就不存在"拿着一些等另一些"的情况：

pthread_mutex_lock(prevention);   // 开始获取
pthread_mutex_lock(L1);
pthread_mutex_lock(L2);
...
pthread_mutex_unlock(prevention); // 结束

先抢到全局的 prevention 锁，就保证了获取锁的过程中不会有"不合时宜"的线程切换插进来。另一个线程即使以不同顺序拿 L1、L2 也没关系，因为它也得先持有 prevention。前提是：任何线程拿任何锁之前都要先拿 prevention。

缺点很明显：封装又一次和我们作对——调用一个函数前必须准确知道它要拿哪些锁并提前拿好；而且所有锁都得提前一起拿，而不是真正需要时再拿，并发度会下降。

非抢占：trylock，拿不到就退回去

锁一般要等到 unlock 才释放，这正是"拿着一个、等着另一个"的根源。很多线程库提供 pthread_mutex_trylock()：锁空闲就拿到并返回成功，否则立即返回错误码，而不是睡眠等待。用它可以构造一个不怕加锁顺序、不会死锁的协议：

top:
    pthread_mutex_lock(L1);
    if (pthread_mutex_trylock(L2) != 0) {  // L2 拿不到？
        pthread_mutex_unlock(L1);          // 那就把 L1 也放掉
        goto top;                          // 从头再来
    }

另一个线程按 L2、L1 的顺序走同样的协议，程序依然不会死锁。但出现了新问题——活锁（livelock）：两个线程可能一直重复这个序列、一直失败，都在运行（所以不是死锁），却都没有进展。

解决活锁也有办法：比如在跳回开头前随机延迟一段时间，降低两个线程反复互相干扰的概率（下面模拟器的"随机退避"选项）。

这个方案还回避了 trylock 的一些难点。首先又是封装：如果某把锁藏在被调用的某个函数深处，"跳回开头"就很难实现；其次，如果拿到 L1 之后还申请了别的资源（例如分配了内存），失败时必须小心地把这些也都释放掉。不过在一些受限场景（比如前面的 Java vector 方法）里，这种方法可以工作得很好。还要注意，它并没有真正实现"抢占"（强行从别人手里夺走锁），而是让开发者体面地主动放弃自己的所有权——原书承认它在这一点上不够完美，但它很实用。

互斥：干脆不用锁（lock-free）

最后一种预防技术是彻底避免互斥。一般来说这很难，因为代码里确实有临界区。Herlihy 提出可以借助强大的硬件指令，设计完全不用锁的数据结构：无锁（lock-free）以及相关的无等待（wait-free）方法。回忆比较并交换（compare-and-swap）指令：

int CompareAndSwap(int *address, int expected, int new) {
    if (*address == expected) {   // 整个函数由硬件原子执行
        *address = new;
        return 1;  // 成功
    }
    return 0;      // 失败
}

void AtomicIncrement(int *value, int amount) {
    do {
        int old = *value;
    } while (CompareAndSwap(value, old, old + amount) == 0);
}

AtomicIncrement 不获取任何锁，只是不断尝试用 CAS 把值更新为新值：如果期间别人改过，CAS 失败就重试。没有锁，就不会死锁（但活锁仍可能存在，健壮的实现会比这复杂）。

稍复杂一点的例子是链表头插。朴素版本在多线程下有竞态：

void insert(int value) {
    node_t *n = malloc(sizeof(node_t));
    assert(n != NULL);
    n->value = value;
    n->next = head;     // 两个线程可能读到同一个 head……
    head = n;           // ……后写的覆盖先写的，丢一个节点
}

传统做法是在最后两行外面加锁（listlock）。原书留了个思考题：为什么锁拿得这么晚、不在函数一开头拿？——因为前面的 malloc 和初始化只涉及新节点这个私有对象（假设 malloc 本身线程安全），不必放进临界区。无锁版本则用 CAS：

void insert(int value) {
    node_t *n = malloc(sizeof(node_t));
    assert(n != NULL);
    n->value = value;
    do {
        n->next = head;
    } while (CompareAndSwap(&head, n->next, n) == 0);
}

当然，一个真正可用的链表不止插入，还要删除、查找；构建一个无锁的完整链表并不简单，有大量文献专门研究它。

模拟器：死锁检测器

下面你来当调度器：每一步选一个线程执行一条"指令"（加锁、临界区、解锁）。右侧实时画出资源分配图——实线"锁 → 线程"表示持有，虚线"线程 → 锁"表示等待。一旦图中出现环，模拟器会把环标红，并逐条说明四个必要条件如何同时成立。换一种策略，再试试能不能把它逼进死锁。

避免（Avoidance）：通过调度绕开死锁

另一条思路不是"预防"，而是避免：需要事先掌握全局信息——各个线程在执行过程中可能会拿哪些锁——然后据此调度，保证死锁不可能发生。

假设有 2 个 CPU、4 个线程。T1 会拿 L1 和 L2（顺序不定、时机不定），T2 也会拿 L1 和 L2，T3 只拿 L2，T4 不拿锁：

聪明的调度器能算出：只要 T1 和 T2 不同时运行，就不会死锁。T3 与 T1（或 T2）重叠没关系——T3 只拿一把锁，永远不会因为"拿着一把等另一把"而参与到环里。

第二个例子竞争更激烈：T1、T2、T3 都要拿 L1 和 L2：

这类方法的著名代表是 Dijkstra 的银行家算法（Banker's Algorithm）（1964 年；有趣的是，Dijkstra 当时把死锁叫做"致命拥抱"（deadly embrace），好在这个名字没流行开来）。遗憾的是，它们只在非常受限的环境里有用——比如嵌入式系统，事先完全知道要运行哪些任务、它们需要哪些锁；而且如例 2 所示，它们会限制并发。所以通过调度避免死锁并不是被广泛使用的通用方案。

检测与恢复（Detect and Recover）

最后一种通用策略：允许死锁偶尔发生，发生后再处理。如果操作系统一年才死机一次，重启一下接着干活就是了——当死锁足够罕见时，这种"非方案"其实相当务实。

先预测，再验证

动手实验：亲手制造（和消灭）一次死锁

cd labs/32-concurrency-bugs && make
./vector_deadlock naive
./vector_deadlock ordered 1000000
./vector_deadlock trylock 1000000
./vector_deadlock global 1000000
./atomicity bug
./atomicity fixed

[naive] 疑似死锁！已完成 3 / 200000 次 vector_add，1 秒内毫无进展
  线程 0: 持有 v0 的锁，等待 v1 的锁
  线程 1: 持有 v1 的锁，等待 v0 的锁
  等待图: T0 -> v1 -> T1 -> v0 -> T0  (出现环)
[naive] 完成 200000 次 vector_add，用时 0.014 秒        ← 另一次运行：侥幸没死锁
[ordered] 完成 2000000 次 vector_add，用时 0.157 秒
[trylock] 完成 2000000 次 vector_add，用时 0.250 秒，trylock 失败重试 3420323 次
[global] 完成 2000000 次 vector_add，用时 0.178 秒
[bug] 检查通过后成功使用 1757021 次；检查通过、使用时却已是 NULL（原子性违反）3103 次
[fixed] 检查通过后成功使用 683798 次；检查通过、使用时却已是 NULL（原子性违反）0 次

自测题

常见误区与本章小结