23.3 互斥

互斥量 (mutex) 保证同一时刻至多只有一个线程持有它。把对共享对象的访问放在加锁与解锁之间（这段代码称为临界区），就能保证这些访问不会同时进行。

下文示例基于 <threads.h>；若目标工具链未提供该头文件，请把同一思路迁移到平台线程接口。

1. 生命周期

互斥量的典型流程是：mtx_init 初始化、mtx_lock 加锁、mtx_unlock 解锁、mtx_destroy 销毁。任何一条执行路径只要拿到锁，就必须保证最终释放。

#include <stdio.h>
#include <threads.h>

static mtx_t lock;
static int counter = 0;

int worker(void *arg) {
    (void)arg;
    for (int i = 0; i < 100000; ++i) {
        mtx_lock(&lock);
        counter++;
        mtx_unlock(&lock);
    }
    return 0;
}

int main(void) {
    thrd_t t1, t2;

    if (mtx_init(&lock, mtx_plain) != thrd_success) {
        return 1;
    }

    thrd_create(&t1, worker, NULL);
    thrd_create(&t2, worker, NULL);
    thrd_join(t1, NULL);
    thrd_join(t2, NULL);

    printf("counter = %d\n", counter);
    mtx_destroy(&lock);
    return 0;
}

输出为：

bash

counter = 200000

2. 易错点

忘记解锁，会使其他线程在 mtx_lock 处永远等待。多个线程以不同的顺序获取多把锁时，也可能互相等待对方释放而形成死锁：例如线程甲持有锁 A 并等待锁 B，同时线程乙持有锁 B 并等待锁 A。程序偶尔卡死时，应先检查锁的获取顺序，以及各个提前返回的分支是否都解了锁。

3. 缩短临界区

临界区应只包含必须受保护的访问，耗时的计算和输入输出放在锁外，以减少其他线程的等待时间。

4. 锁类型选择

mtx_init 的类型参数决定互斥量的类型：mtx_plain 是普通互斥量；mtx_timed 支持超时，可以使用 mtx_timedlock；再与 mtx_recursive 按位或，就得到递归互斥量，同一线程可以多次加锁，并需要相同次数的解锁。没有特殊需要时使用 mtx_plain，它的控制流最容易检查。

5. 锁的获取顺序

模块中有多把锁时，应规定一个固定的获取顺序，并写进文档。所有代码都按这个顺序加锁，就不会出现上面那种互相等待的死锁。

6. 失败路径上的解锁

临界区中有多个提前返回的分支时，很容易在某个分支上漏掉解锁。可以让所有分支跳转到同一个出口，在出口处统一解锁：

int update_shared(void) {
    int rc = -1;
    if (mtx_lock(&lock) != thrd_success) {
        return -1;
    }

    if (!ready()) {
        goto out_unlock;
    }

    rc = apply_change();

out_unlock:
    mtx_unlock(&lock);
    return rc;
}

7. 互斥与原子

互斥量适合保护由多个对象组成、需要整体保持一致的状态，例如一个链表的头指针和元素个数。原子类型（见 23.2）适合只涉及单个对象的计数器或标志。若一个状态包含多个成员，且成员之间有必须保持的关系，应使用互斥量；分别把每个成员声明为原子类型，并不能保证它们之间的关系。

8. 习题

8.1 巩固

#12314

⚡2⏳2

下面的线程加锁之后直接返回，没有解锁：

#include <stdio.h>
#include <threads.h>

static mtx_t lock;

static int worker(void *arg) {
    (void)arg;
    mtx_lock(&lock);
    puts("worker: locked, returning without unlock");
    return 0;
}

int main(void) {
    thrd_t t;

    mtx_init(&lock, mtx_plain);
    thrd_create(&t, worker, NULL);
    thrd_join(t, NULL);
    int rc = mtx_trylock(&lock);
    printf("trylock = %d (success=%d, busy=%d)\n", rc, rc == thrd_success,
           rc == thrd_busy);
    return 0;
}

mtx_trylock 尝试加锁但不阻塞，成功返回 thrd_success，锁已被占用返回 thrd_busy。

要求：

1. 先预测 trylock 的结果，再编译运行。线程已经退出，它持有的锁会随线程结束自动释放吗？
2. 在 printf 之后把 return 0; 前面加一句 mtx_lock(&lock);，用 timeout 5 ./a.out 运行，记录退出码。程序停在哪里？此时锁在谁手里？
3. 把 mtx_destroy(&lock) 加在 thrd_join 之后、mtx_trylock 之前。程序行为是否改变？按本节第 1 小节，销毁锁的正确时机是什么？
4. 如果线程退出时锁确实没有释放，标准为什么不允许实现替线程自动解锁？从「临界区里的共享对象可能只改到一半」这个角度说明。

验收：第 1 步 trylock = 1，success=0、busy=1；第 2 步退出码 124，主线程阻塞在 mtx_lock；第 3 步行为不变，但销毁一把仍被持有的锁本身就是错误用法，销毁应当在所有加锁路径都结束之后。

#12315

⚡3⏳3

下面的函数在三条提前返回的分支上都漏了解锁，第二次调用就会卡住：

#include <stdio.h>
#include <threads.h>

static mtx_t lock;
static int count;

static int update(int kind) {
    if (mtx_lock(&lock) != thrd_success) {
        return -1;
    }
    if (kind < 0) {
        return -1;
    }
    if (kind == 0) {
        return 0;
    }
    ++count;
    mtx_unlock(&lock);
    return 0;
}

int main(void) {
    mtx_init(&lock, mtx_plain);
    printf("update(0) = %d\n", update(0));
    printf("update(-1) = %d\n", update(-1));
    printf("update(1) = %d\n", update(1));
    printf("count = %d\n", count);
    return 0;
}

要求：

1. 先只跑第一次调用（把后两次注释掉），打印后立即 fflush(stdout)，再调用一次 update(1)。用 timeout 5 运行，记录退出码。说明卡死的位置。
2. 用本节第 6 小节的单一出口写法重写 update：所有分支跳转到 out_unlock，在出口处统一解锁。验收：三次调用依次返回 0、-1、0，count = 1，程序正常退出。
3. 把 goto 改成在每个分支前手动 mtx_unlock 再 return。两种写法各有几处解锁？如果再增加一条返回 -2 的分支，各需要改几处？
4. 让 update 在临界区里调用另一个也加这把锁的函数，记录程序行为。这与第 2 步的单一出口写法能否同时成立？说明为什么在「统一出口」与「可重入调用」之间必须选一个。

验收：第 1 步退出码 124；第 2 步输出 update(0) = 0、update(-1) = -1、update(1) = 0、count = 1；第 4 步卡死（timeout 返回 124），因为同一线程对 mtx_plain 锁二次加锁会永远等待。

8.2 编程

#12316

⚡3⏳4

下面的注册表在持锁状态下遍历，回调里又想注册一项：

#include <stdio.h>
#include <threads.h>

static mtx_t lock;
static int items[8];
static int count;

static void registry_add(int v) {
    mtx_lock(&lock);
    if (count < 8) {
        items[count++] = v;
    }
    mtx_unlock(&lock);
}

static void on_item(int v) {
    printf("item %d\n", v);
    if (v == 1) {
        registry_add(99);
        puts("callback added 99");
    }
}

static void registry_foreach(void (*cb)(int)) {
    mtx_lock(&lock);
    int n = count;
    for (int i = 0; i < n; ++i) {
        cb(items[i]);
    }
    mtx_unlock(&lock);
}

int main(void) {
    mtx_init(&lock, mtx_plain);
    registry_add(1);
    registry_add(2);
    puts("foreach begins");
    fflush(stdout);
    registry_foreach(on_item);
    puts("foreach ends");
    printf("count = %d\n", count);
    return 0;
}

要求：

1. 用 timeout 5 运行，记录退出码和已经打印的行。说明死锁是怎样形成的。
2. 修复一：把锁换成递归类型 mtx_plain | mtx_recursive，其余不动。验收：正常退出，输出 item 1、callback added 99、item 2、count = 3。
3. 修复二：在持锁时把 items 复制到本地数组，解锁之后再逐个调用回调。验收与本机输出一致：

foreach begins
item 1
callback added 99
item 2
foreach ends
count = 3

4. 两种修复的语义差别在哪？回调在遍历过程中新增的项，第二次修复里当前这一轮会不会被访问到？递归锁版本里会不会？如果回调可能来自库的使用者、行为无法预知，应选哪一种？
5. 递归锁有一个陷阱：加锁几次就必须解锁几次。在修复一的 main 末尾加一次 mtx_trylock 检查返回值，说明在 registry_foreach 正常返回后锁的计数是几，以及为什么 mtx_destroy 必须在计数归零之后。

验收：第 1 步退出码 124；标准输出被重定向时按块缓冲，item 1 可能来不及输出，能看到 foreach begins 即可判断卡死位置；第 2、3 步都输出两行 item 且 count = 3；第 5 步 mtx_trylock 成功，说明计数已归零。

#12317

⚡4⏳4

4 个工作线程各做 5 轮：先进行 10 ms 的「耗时计算」，再对共享计数器加一。测量整段程序的墙钟时间。

(1) [3.3] 先用下面两种写法各测一次：

static void slow_work(void) {
    thrd_sleep(&(struct timespec){ .tv_nsec = 10000000 }, NULL);
}

/* 写法一：整段放进临界区 */
mtx_lock(&lock);
slow_work();
++counter;
mtx_unlock(&lock);

/* 写法二：只保护计数 */
slow_work();
mtx_lock(&lock);
++counter;
mtx_unlock(&lock);

本机实测前者 201 ms、后者 50 ms。解释两个数字分别由什么决定，并说明为什么 4 个线程的 20 次 10 ms 计算在写法二里能并行成 50 ms。

(2) [4.3] 把初始化改成 mtx_init(&lock, mtx_timed)，把 mtx_lock 换成 mtx_timedlock，超时设为 150 ms，另写一个线程先持锁 1 秒后再释放。记录 mtx_timedlock 的返回值和实际等待时间。本机返回 thrd_timedout（数值为 4），等待 150 ms。说明超时参数用的是绝对时刻还是相对时长，以及收到 thrd_timedout 时调用者有没有拿到锁。

(3) [4.4] 用 mtx_trylock 写一个「拿不到就去做别的事」的版本：每轮先 mtx_trylock，失败就打印一次 skip 并继续下一轮，不阻塞。让每轮在尝试之前先睡 2 ms，运行 200 轮、4 个线程。本机 3 次得到 skipped = 16、3、8，counter 依次为 784、797、792。说明这种写法适合什么场景，以及它为什么不能保证每一轮都完成计数。

(4) [5.4] 临界区里调用 printf 会不会拖长临界区？把 4 个线程各输出 500 行（每行十几字节，输出重定向到文件并每行 fflush），分别在锁内与锁外各测一次。本机两次都在 4 到 6 ms 之间，看不出差别。说明为什么这个规模的输出测不出差异，并设计一个能测出差异的实验（提示：把单行输出加长，或让输出落到较慢的设备上，再用两个版本的耗时对比）。

验收：第 1 步两个数字分别为约 200 ms 与约 50 ms；第 2 步返回 thrd_timedout，等待约 150 ms，未取得锁；第 3 步能说明 trylock 失败时不持有锁、不需要解锁，skipped 大于 0；第 4 步能说清「输出本身很快、锁内外的差别被其他开销淹没」是测不出差异的原因。

8.3 综合与挑战

#12318

⚡5⏳5

两个对象 x、y 必须同步增长，读线程检查它们是否相等。

(1) [4.4] 4 个写线程各自执行 atomic_fetch_add(&gx, 1); atomic_fetch_add(&gy, 1);，一个读线程循环 200000 次，每次连续读 gx、gy 并统计不等的次数。运行并记录。本机五次都得到 18 万次左右（上限 20 万）。说明每一次加法都是原子的，为什么两次读取之间仍然能看到中间状态。

(2) [5.5] 把写线程和读线程都改成用同一把 mtx_t 保护 ++x; ++y; 与「读 x、读 y」。运行并记录不等的次数。验收：x = 800000、y = 800000、mismatches = 0。

(3) [5.5] 只把读线程的两次读取放进锁里，写线程保持两个独立的原子加法。运行并记录不等的次数。本机仍得到约 4 万次。说明为什么只锁读端没有用，以及「锁保护的是数据」还是「锁保护的是访问」在这个例子里怎样体现。

(4) [6.5*] 把两个对象换成结构体的两个成员，用一把锁保护整个结构体，读线程在锁内一次性复制整个结构体。说明这种「整体快照」的写法与两个原子变量相比，多出了哪些保证；如果读端每秒钟要读一百万次，可以怎样减少锁的竞争（提示：读写各一把锁、每线程各持有一份快照、用序号校验）。

验收：第 1 步不为 0 且每次不同；第 2 步恒为 0；第 3 步仍明显不为 0。

#12319

⚡6⏳5

下面的转账程序让两个线程以相反的顺序获取两把锁：

static mtx_t lock_a, lock_b;
static long acct[2];

static int forward(void *arg) {
    (void)arg;
    for (int i = 0; i < 100000; ++i) {
        mtx_lock(&lock_a);
        mtx_lock(&lock_b);
        acct[0] -= 1;
        acct[1] += 1;
        mtx_unlock(&lock_b);
        mtx_unlock(&lock_a);
    }
    return 0;
}

static int backward(void *arg) {
    (void)arg;
    for (int i = 0; i < 100000; ++i) {
        mtx_lock(&lock_b);
        mtx_lock(&lock_a);
        acct[1] -= 1;
        acct[0] += 1;
        mtx_unlock(&lock_a);
        mtx_unlock(&lock_b);
    }
    return 0;
}

(1) [4.4] 用 timeout 8 连跑 3 次，记录退出码。本机 3 次都是 124，但这只是概率高，不是必然。写出一个必定复现的版本：让 forward 在拿到 lock_a 之后、拿 lock_b 之前睡 150 ms，backward 先拿 lock_b、睡 50 ms 再拿 lock_a。验收：3 次运行都是退出码 124。

(2) [5.5] 按本节第 5 小节规定统一的获取顺序（先小地址、先编号小的锁），把两个函数改成以相同顺序加锁。验收：acct[0] + acct[1] = 0，连跑 3 次都正常结束。说明为什么「规定顺序」比「小心一点」可靠。

(3) [6.5] 用 mtx_trylock 实现「拿不到第二把锁就退回第一把」的写法：

for (;;) {
    if (mtx_trylock(&lock_a) != thrd_success) {
        thrd_yield();
        continue;
    }
    if (mtx_trylock(&lock_b) == thrd_success) {
        break;
    }
    mtx_unlock(&lock_a);
    thrd_yield();
}

两个方向都改成这个写法，各做 100000 次，运行并记录 sum、重试次数和耗时。本机 sum = 0，重试 9 万次上下，耗时 48 到 50 ms。说明重试为什么不会活锁，以及这个写法与第 2 步的固定顺序相比各有什么代价。

(4) [7.5*] 把 mtx_trylock 换成 mtx_timedlock，超时 1 ms，其余结构相同。跑 2000 轮并计时。本机 2000 轮耗时 11 秒到 17 秒；同样 2000 轮的 trylock 版本只用 1 ms 左右。解释这个差距：超时版本在等锁期间做了什么，为什么它不适合作为「避免死锁」的默认手段。

验收：第 1 步 3 次都是 124；第 2 步恒为 0；第 3 步 sum = 0 且能正常结束；第 4 步耗时以秒计，明显比 trylock 版本慢。

#12320

⚡6⏳6

把一把全局锁换成一组按桶分配的锁，是「缩短临界区」之外另一种常见的做法。下面是一个 16 桶的哈希表，插入与查找都要加锁：

#define BUCKETS 16
#define THREADS 4
#define PER 100000

typedef struct entry {
    int key;
    struct entry *next;
} entry;

static entry *table[BUCKETS];
static mtx_t lock;   /* 改成一桶一把：static mtx_t lock[BUCKETS]; */

static unsigned hash(int key) { return (unsigned)key % BUCKETS; }

(1) [5.5] 先用一把全局锁实现插入，4 个线程各插入 100000 个键，测量耗时，跑 3 次。本机在 83 到 104 ms 之间。再把锁改成 lock[BUCKETS]，按 hash(key) 选择，重新测量 3 次，本机在 38 到 59 ms 之间。说明加速来自哪里：是锁变快了，还是等待次数变少了？

(2) [5.5] 把哈希函数改成 return 0;，其余不变（每把锁仍然只在对应的桶上使用）。重新测量并解释结果。本机 3 次在 87 到 96 ms 之间，与全局锁版本接近。这说明「按桶分锁」在什么条件下才有效？

(3) [6.5] 哈希表扩容时要把旧桶里的节点搬进新桶，这时候一个线程可能同时需要两个桶的锁。用两把锁实现搬移，说明必须遵守的获取顺序，并写一个最小程序验证：以相反顺序搬移的两个线程会在 timeout 下卡死（退出码 124），统一顺序后正常结束。

(4) [7.5*] 桶数从 16 改成 64、256，测量插入耗时的变化。桶数增加后，锁的数目、缓存占用和查找性能各有什么变化？为什么桶数不能无限增加？

验收：第 1 步按桶分锁的耗时约为全局锁的一半；第 2 步各线程全部争同一把锁，耗时回到与全局锁版本相近；第 3 步未规定顺序的版本退出码 124，统一顺序后正常。