本章要解决什么

第 26 章看到：我们想让一段指令原子地执行，但中断（或多个 CPU 上同时运行的线程）让这做不到。本章正面解决这个问题：程序员在临界区前后加上锁（lock），让临界区看起来就像一条原子指令。第 27 章只讲了 pthread_mutex_lock 怎么用；这一章讲它怎么造：从关中断、只用 load/store 的失败尝试，到 test-and-set、compare-and-swap、LL/SC、fetch-and-add 这些硬件原语，再到借助操作系统的 yield、park/unpark 和 Linux futex，避免白白自旋。

本章还有第二个关键问题（28.12 节）：如何避免自旋？怎样开发一种不会在 CPU 上无谓浪费时间自旋的锁？（CRUX: How to avoid spinning. How can we develop a lock that doesn't needlessly waste time spinning on the CPU?）

核心概念速览

锁的基本思想

典型的临界区是更新共享变量：

lock_t mutex;            // some globally-allocated lock 'mutex'
...
lock(&mutex);
balance = balance + 1;   // 临界区
unlock(&mutex);

语义很简单：调用 lock() 尝试获取锁；如果没有别的线程持有，就获得锁、进入临界区。如果别的线程已持有，lock() 不会返回，直到持有者调用 unlock()。释放时如果有线程在 lock() 里等待，其中一个会（最终）注意到或被告知锁的状态变化，获得锁并进入临界区。

锁把一点调度控制权还给了程序员：线程由程序员创建、由 OS 按它的方式调度；而用锁包住一段代码，程序员就能保证这段代码里至多只有一个线程活动。

POSIX 把锁叫 mutex（mutual exclusion，互斥）。注意 pthread_mutex_lock(&lock) 传入一个锁变量：不同的数据可以用不同的锁保护。一把大锁保护所有临界区叫粗粒度（coarse-grained）加锁；不同数据结构用不同的锁叫细粒度（fine-grained）加锁，允许更多线程同时处在加锁代码中。

怎样评价一个锁

最早的方案：关中断

void lock()   { DisableInterrupts(); }
void unlock() { EnableInterrupts(); }

在单处理器上，进入临界区前关中断，临界区就不会被打断，于是像原子的一样执行。优点只有一个：简单。缺点很多：

1. 要信任应用程序：必须允许任意线程执行特权操作（开关中断）。贪婪的程序一开始就 lock() 独占 CPU；恶意或出错的程序 lock() 后死循环，OS 再也拿不回控制权，只能重启。
2. 多处理器上无效：别的 CPU 上的线程照样能进入临界区，关本 CPU 的中断毫无用处。
3. 长时间关中断会丢中断：比如错过磁盘读完成的中断，OS 就不知道该唤醒等待的进程。

所以关中断只在有限场合使用：操作系统自己访问内部数据结构时，有时用屏蔽中断保证原子性——OS 总是信任自己，不存在信任问题。

失败的尝试：只用 load/store

试着只用一个普通变量 flag 表示锁是否被持有（图 28.1）：

typedef struct __lock_t { int flag; } lock_t;

void init(lock_t *mutex) {
    // 0 -> lock is available, 1 -> held
    mutex->flag = 0;
}

void lock(lock_t *mutex) {
    while (mutex->flag == 1)   // TEST the flag
        ;                      // spin-wait (do nothing)
    mutex->flag = 1;           // now SET it!
}

void unlock(lock_t *mutex) {
    mutex->flag = 0;
}

有两个问题。正确性：检查（TEST）和设置（SET）是两步。如果 T1 读到 flag=0、正要设置时被中断，T2 也读到 0 并设置为 1 进入临界区；T1 恢复后也把 flag 设为 1——两个线程都进了临界区（图 28.2）。性能：等待者不停地检查 flag，称为自旋等待（spin-waiting），在单处理器上尤其浪费——它等的那个线程在它自旋时根本没法运行。

用 test-and-set 构建能用的自旋锁

既然关中断在多处理器上无效、load/store 又不行，系统设计者开始发明加锁的硬件支持。最早的多处理器（如 1960 年代初的 Burroughs B5000）就有，今天所有系统都有，甚至单 CPU 系统也有。最简单的是 test-and-set（也叫原子交换 atomic exchange），用 C 描述它做的事：

int TestAndSet(int *old_ptr, int new) {
    int old = *old_ptr;   // fetch old value at old_ptr
    *old_ptr = new;       // store 'new' into old_ptr
    return old;           // return the old value
}   // 关键：这三步由硬件原子地完成

它返回旧值，同时把内存设为新值。在 SPARC 上叫 ldstub，在 x86 上就是带锁的 xchg。用它构建自旋锁（图 28.3）：

void lock(lock_t *lock) {
    while (TestAndSet(&lock->flag, 1) == 1)
        ;   // spin-wait (do nothing)
}
void unlock(lock_t *lock) {
    lock->flag = 0;
}

为什么正确？两种情况：① 锁空闲（flag=0）：TestAndSet 返回 0，线程跳出循环获得锁，同时flag 被原子地设为 1。② 锁被持有（flag=1）：TestAndSet 返回 1（又把 1 写了一遍，无害），线程继续自旋，直到持有者把 flag 清 0，下一次 TestAndSet 返回 0 并原子地设成 1。检查旧值和设置新值合成了一个原子操作，所以只有一个线程能拿到锁。

这种锁叫自旋锁（spin lock）。在单处理器上它需要抢占式调度（定时器中断会打断线程）；否则自旋的线程永远不会让出 CPU，持锁者永远没机会释放。

int flag[2];
int turn;
void init() {
    flag[0] = flag[1] = 0;   // 用 flag 表示"我想持有锁"
    turn = 0;                // 轮到谁？（线程 0 或 1）
}
void lock() {
    flag[self] = 1;          // self 是调用者的线程 ID
    turn = 1 - self;         // 谦让：让对方先
    while ((flag[1-self] == 1) && (turn == 1 - self))
        ;                    // 对方也想要，且轮到对方 → 自旋
}
void unlock() {
    flag[self] = 0;          // 撤销意愿
}

评价自旋锁

• 正确性：✔ 一次只允许一个线程进入临界区。
• 公平性：✘ 不保证。自旋的线程在竞争下可能永远拿不到锁——饿死（starvation）。
• 性能（单 CPU）：很痛苦。持锁线程在临界区里被抢占后，调度程序可能依次运行其他 N−1 个线程，每个都尝试获取锁、自旋整整一个时间片才放弃 CPU。
• 性能（多 CPU）：相当不错（线程数大致等于 CPU 数时）。A 在 CPU1 持锁，B 在 CPU2 自旋；临界区通常很短，锁很快就可用，B 没浪费多少周期。

compare-and-swap

另一个硬件原语：SPARC 上叫 compare-and-swap，x86 上叫 compare-and-exchange（cmpxchg）（图 28.4）：

int CompareAndSwap(int *ptr, int expected, int new) {
    int original = *ptr;
    if (original == expected)
        *ptr = new;
    return original;
}

检查 *ptr 是否等于 expected，是就更新为 new，否则什么也不做；无论如何都返回原值，调用者据此知道是否成功。把 TAS 锁的 lock() 换成下面这样即可：

void lock(lock_t *lock) {
    while (CompareAndSwap(&lock->flag, 0, 1) == 1)
        ;   // spin
}

CAS 比 TAS 更强大，后面讲无锁同步（lock-free synchronization）时会用到这种能力；但如果只是用它构建简单自旋锁，行为和 TAS 自旋锁完全相同。

load-linked / store-conditional

MIPS、Alpha、PowerPC、ARM 提供一对配合使用的指令（图 28.5）：

int LoadLinked(int *ptr) {
    return *ptr;
}
int StoreConditional(int *ptr, int value) {
    if (no update to *ptr since LL to this addr) {
        *ptr = value;
        return 1;   // success!
    } else {
        return 0;   // failed to update
    }
}

load-linked 就像普通的 load；关键在 store-conditional：只有自上次 LL 以来没有别的 store 写过这个地址，它才成功（写入并返回 1），否则失败（不写，返回 0）。用它构建锁（图 28.6）：

void lock(lock_t *lock) {
    while (1) {
        while (LoadLinked(&lock->flag) == 1)
            ;   // spin until it's zero
        if (StoreConditional(&lock->flag, 1) == 1)
            return;   // if set-to-1 was success: done
        // otherwise: try again
    }
}
void unlock(lock_t *lock) {
    lock->flag = 0;
}

失败是怎么发生的：T1 执行 LL 读到 0，正要 SC 时被中断；T2 也执行 LL 读到 0。现在两个线程都要 SC——只有一个能成功把 flag 改成 1；第二个 SC 会失败（因为另一个线程在它 LL 和 SC 之间写了 flag），只好重试。学生 David Capel 给出的更短写法（利用短路求值）：

void lock(lock_t *lock) {
    while (LoadLinked(&lock->flag) || !StoreConditional(&lock->flag, 1))
        ;   // spin
}

fetch-and-add 与 ticket 锁

fetch-and-add 原子地把一个值加 1 并返回旧值：

int FetchAndAdd(int *ptr) {
    int old = *ptr;
    *ptr = old + 1;
    return old;
}

Mellor-Crummey 和 Scott 用它构建了 ticket 锁（图 28.7）：

typedef struct __lock_t {
    int ticket;
    int turn;
} lock_t;

void lock_init(lock_t *lock) {
    lock->ticket = 0;
    lock->turn   = 0;
}
void lock(lock_t *lock) {
    int myturn = FetchAndAdd(&lock->ticket);   // 取号
    while (lock->turn != myturn)
        ;   // spin                             // 等叫号
}
void unlock(lock_t *lock) {
    lock->turn = lock->turn + 1;                // 叫下一个号
}

和之前方案的重要区别：它保证所有线程都能前进。一旦线程拿到号，它一定会在未来某个时刻被叫到（前面的人用完临界区之后）。而 TAS 自旋锁没有这种保证：一个线程可能一直自旋，而别的线程反复获取、释放锁。

自旋太多：怎么办？

硬件锁简单（几行代码）而且正确，但有时非常低效。想象单处理器上两个线程：线程 0 持锁时被中断，线程 1 尝试获取锁，发现被持有，于是开始自旋——自旋——继续自旋……直到定时器中断，线程 0 再次运行并释放锁。任何时候线程在这种情况下自旋，都浪费了整个时间片，去检查一个不会改变的值。N 个线程竞争时，可能浪费 N−1 个时间片。

这就是本章第二个关键问题：如何避免自旋？单靠硬件解决不了，还需要操作系统的支持。

简单方法：让出 CPU（yield）

void init() {
    flag = 0;
}
void lock() {
    while (TestAndSet(&flag, 1) == 1)
        yield();   // give up the CPU
}
void unlock() {
    flag = 0;
}

假设 OS 提供 yield() 系统调用：把调用者从运行态变成就绪态，让另一个线程运行——相当于线程主动把自己"取消调度"。两个线程一个 CPU 时效果很好：发现锁被持有就让出 CPU，持锁者得以运行并完成临界区。

但如果有 100 个线程反复竞争：持锁者被抢占后，其余 99 个都会执行"运行一下、发现锁被持有、yield"。比自旋（浪费 99 个时间片）好，但上下文切换的开销仍然可观。更糟的是，它没有解决饿死：一个线程可能陷入无休止的 yield 循环，而别的线程反复进出临界区。

使用队列：休眠代替自旋

前面方法的真正问题是把太多事留给运气：调度程序决定下一个运行谁，选错了，运行的线程要么自旋、要么立刻让出。我们需要明确控制锁释放后由谁获得。为此需要一点 OS 支持和一个等待队列。原书采用 Solaris 的两个调用：park() 让调用线程睡眠，unpark(threadID) 唤醒指定线程（图 28.9）：

typedef struct __lock_t {
    int flag;
    int guard;
    queue_t *q;
} lock_t;

void lock_init(lock_t *m) {
    m->flag  = 0;
    m->guard = 0;
    queue_init(m->q);
}

void lock(lock_t *m) {
    while (TestAndSet(&m->guard, 1) == 1)
        ;                        // acquire guard lock by spinning
    if (m->flag == 0) {
        m->flag = 1;             // lock is acquired
        m->guard = 0;
    } else {
        queue_add(m->q, gettid());
        m->guard = 0;
        park();                  // 睡眠，直到被 unpark
    }
}

void unlock(lock_t *m) {
    while (TestAndSet(&m->guard, 1) == 1)
        ;                        // acquire guard lock by spinning
    if (queue_empty(m->q))
        m->flag = 0;             // let go of lock; no one wants it
    else
        unpark(queue_remove(m->q));   // hold lock (for next thread!)
    m->guard = 0;
}

逐点理解：

• guard 是一个"保护 flag 和队列"的自旋锁。所以这个方案并没有完全消除自旋：线程可能在获取/释放锁的过程中被中断，导致别人自旋等 guard。但自旋时间很有限——只是 lock/unlock 里的几条指令，而不是用户定义的临界区。
• 拿不到锁时，线程把自己加入队列、释放 guard、然后 park()。思考题：如果 guard 的释放放在 park() 之后会怎样？提示：很糟——线程带着 guard 睡着了，其他线程（包括要来 unlock 唤醒它的持锁者）都会在 guard 上永远自旋。
• 唤醒别的线程时 flag 不置 0。这不是错误而是必须：被唤醒的线程从 park() 返回时并不持有 guard，没法再去把 flag 设成 1。所以锁被直接从释放者传递给下一个获取者，flag 中间不变成 0。
• 用队列控制下一个获得锁的是谁，因此避免了饿死。

唤醒/等待竞争（wakeup/waiting race）与 setpark()

注意 park() 调用之前那一小段：线程已经决定要睡觉（因为锁被持有），但还没真正睡着。如果此时切换到持锁线程并且它释放了锁，它会 unpark 这个"还醒着"的线程——这次唤醒就落空了；随后该线程执行 park()，可能永远睡下去。Solaris 用第三个系统调用 setpark() 解决：线程先声明"我马上要 park 了"；若在真正 park 之前有人 unpark 了它，随后的 park 会立即返回而不是睡眠：

queue_add(m->q, gettid());
setpark();          // new code
m->guard = 0;
park();

另一种办法是把 guard 传进内核，由内核原子地释放锁并把当前线程移出运行队列。下面这个小模拟器对比有无 setpark() 时的同一交错：

下面的模拟器把三种等待策略放在同一个单 CPU 上比较：N 个线程循环执行"获取锁 → 临界区 → 释放 → 做其他工作"，时间片轮转调度，初始时 T1 持锁并即将在临界区中被抢占（原书讨论的最坏情形）。

不同的 OS，不同的支持：Linux futex

Linux 提供 futex，与 Solaris 的接口类似但把更多功能放进内核：每个 futex 关联一个物理内存地址和一个内核中的等待队列。futex_wait(address, expected)：如果 *address == expected 就让调用者睡眠，否则立即返回（这个"检查-睡眠"在内核中原子完成，天然避免了唤醒/等待竞争）。futex_wake(address) 唤醒队列中的一个线程。nptl 库（glibc）中 lowlevellock.h 的代码片段（图 28.10）：

void mutex_lock (int *mutex) {
    int v;
    // Bit 31 was clear, we got the mutex (fastpath)
    if (atomic_bit_test_set (mutex, 31) == 0)
        return;
    atomic_increment (mutex);
    while (1) {
        if (atomic_bit_test_set (mutex, 31) == 0) {
            atomic_decrement (mutex);
            return;
        }
        // Have to wait. First make sure futex value
        // we are monitoring is negative (locked).
        v = *mutex;
        if (v >= 0)
            continue;
        futex_wait (mutex, v);
    }
}

void mutex_unlock (int *mutex) {
    // Adding 0x80000000 to counter results in 0 if and
    // only if there are not other interested threads
    if (atomic_add_zero (mutex, 0x80000000))
        return;
    // There are other threads waiting for this mutex,
    // wake one of them up.
    futex_wake (mutex);
}

这段真实代码有两点值得注意：① 用一个整数同时记录锁是否被持有（最高位）和等待者数量（其余位），所以锁被持有时这个整数是负数；② 为常见情况（无竞争）优化：只有一个线程加锁、解锁时，只做一次原子位测试并设置（加锁）和一次原子加法（解锁），完全不进内核。

两阶段锁（two-phase lock）

Linux 的做法有一种古老方法的味道，至少可追溯到 1960 年代初的 Dahm 锁，现在叫两阶段锁：它意识到自旋可能是有用的，尤其是锁即将被释放时。第一阶段先自旋一会儿，希望能拿到锁；如果没拿到，进入第二阶段：调用者睡眠，直到锁空闲时被唤醒。上面的 Linux 锁只自旋一次；推广一下，可以在循环里自旋固定时间再用 futex 睡眠。

两阶段锁是又一个混合方案：把两个好主意结合起来可能得到更好的主意。但效果强烈依赖于硬件环境、线程数和工作负载。做出一个对所有情况都好的通用锁，是个真正的挑战。

把所有实现放在一起比较

先预测，再验证

动手实验：用 GCC 原子内建造锁

cd labs/28-locks && make
make correct     # 正确性
make bench       # 1/2/4/8/16 线程
make oversub     # 线程数 = 2 × CPU 数
make single      # 所有线程钉在 CPU 0 上

none   threads=4   counter=1070265   expected=4000000   WRONG time=0.006s
flag   threads=4   counter=1057400   expected=4000000   WRONG time=0.011s
tas    threads=4   counter=4000000   expected=4000000   OK    time=0.140s
cas    threads=4   counter=4000000   expected=4000000   OK    time=0.148s
ticket threads=4   counter=4000000   expected=4000000   OK    time=0.308s
mutex  threads=4   counter=4000000   expected=4000000   OK    time=0.075s

$ make oversub     # 32 线程 × 20 万次
tas    threads=32  counter=6400000   expected=6400000   OK    time=1.517s
ttas   threads=32  counter=6400000   expected=6400000   OK    time=1.341s
ticket threads=32  TIMEOUT after 10.0s: only 18964 of 6400000 increments done (1896 per second)
yield  threads=32  counter=6400000   expected=6400000   OK    time=0.030s
mutex  threads=32  counter=6400000   expected=6400000   OK    time=0.143s
$ make single      # taskset -c 0，4 线程 × 500 万次
tas    threads=4   counter=20000000  expected=20000000  OK    time=0.136s
ticket threads=4   TIMEOUT after 5.0s: only 534588 of 20000000 increments done (106918 per second)
yield  threads=4   counter=20000000  expected=20000000  OK    time=0.093s
mutex  threads=4   counter=20000000  expected=20000000  OK    time=0.150s

自测题

常见误区

本章小结

• 锁让临界区像单条原子指令一样执行。评价锁看三点：正确性、公平性、性能（无竞争 / 单 CPU / 多 CPU）。
• 关中断只适合单处理器上的 OS 内部；只用 load/store 的 flag 不正确；Peterson/Dekker 只用 load/store 能工作，但在现代宽松内存模型下失效。
• 硬件原语——test-and-set、compare-and-swap、LL/SC、fetch-and-add——把"检查 + 修改"合成一个原子操作，可以构建正确的自旋锁；ticket 锁进一步保证了公平。
• 自旋在单 CPU 或线程多于 CPU 时浪费严重。OS 支持能改善：yield 让出 CPU；队列 + park/unpark（加 setpark 防唤醒竞争）让等待者睡眠并按序交接；Linux futex 加上两阶段锁是今天真实的做法：硬件原子指令 + OS 睡眠/唤醒原语。
• 下一章（第 29 章）把锁用到计数器、链表、队列、哈希表中，看看锁放在哪里、放几把如何影响正确性与性能；第 30 章的条件变量处理"等待某个条件成立"。