Lesson 49: 哲学家就餐问题 — 状态机 + 非对称防死锁
练习任务
难度:中 【标杆题】
用 POSIX 线程 (pthreads) 真实模拟五位哲学家围坐圆桌就餐:每人需要左右两根筷子才能进餐,每根筷子是一把 pthread_mutex_t 互斥锁。程序支持三种策略 naive / asymmetric / ordered:
naive— 全部先拿左筷再拿右筷,故意触发死锁,watchdog 超时检测并打印诊断asymmetric— 让一位哲学家反过来先拿右筷,打破循环等待ordered— 始终先拿编号小的筷子,资源偏序预防
每人吃满 100 次后模拟结束(naive 会在死锁时提前由 watchdog 终止)。
你需要实现 pickup()、putdown()、philosopher()、watchdog()、build_wait_cycle()、coffman_check() 和 main() 中的线程编排,共 7 个 TODO。同时需要补齐 Makefile 中的 CC、CFLAGS、TARGET、SRC 变量以及 all、test、clean 三条规则。
程序按退出码判分——这是并发程序自动化测试的核心手段:
$ ./dining_philosophers naive
=== Dining Philosophers (strategy=naive) ===
N=5, target=100 meals each, watchdog=3s
...
!!! DEADLOCK DETECTED !!!
Cycle: P0 → chop1 → P1 → chop2 → P2 → chop3 → P3 → chop4 → P4 → chop0 → P0
Coffman check: 互斥✓ 持有等待✓ 不不可剥夺✓ 循环等待✓
exit code: 2提示:死锁不是 bug——是特性。naive 策略下你要故意构造死锁,让学员亲眼看到循环等待环的形成和 Coffman 四条件的代码化身。理解"为什么会死锁"比"如何避免死锁"更重要——只有亲眼见过死锁的诊断输出,才能真正理解预防策略为什么有效。Asymmetric 和 ordered 验证"破坏循环等待即可预防死锁"的正确性。
核心知识点
- Coffman 死锁四条件 — 互斥(Mutex)、持有并等待(Hold & Wait)、不可剥夺(No Preemption)、循环等待(Circular Wait)。死锁发生当且仅当四者同时成立,预防只需破坏其中一个
- 真实多线程 vs 单线程模拟 — 单线程模拟中"检查两根筷子同时空闲"是原子的,不存在"持有并等待"中间态,死锁无法表达;真实
pthread_mutex_lock会阻塞**调用线程,"持左等右"的窗口真实存在,死锁可构造、可观测、可检测 - naive 策略的病灶价值 — 所有哲学家先拿左筷再拿右筷,五人同时执行 → 各持一根 → 互等另一根 → 循环等待环闭合 → 死锁。这是"故意死锁供观察"的教学设计
- asymmetric 非对称策略 — 让一位哲学家(P4)反转拿筷顺序(先右后左),打破对称性 → P0 和 P4 在筷 0 上竞争 → 必有一人失败 → 循环等待环断开
- ordered 资源排序策略 — 始终先拿编号小的筷子:
min(left, right)先、max(left, right)后。资源存在全局偏序 → 数学证明不可能形成循环等待环。这是最通用的死锁预防策略 - pthread_barrier_t 同步起跑门 — 让 5 个线程第一轮同时起跑,消除线程启动先后带来的 flaky(时锁时不锁),配合
GRAB_GAP_US放大"持有并等待"窗口,naive 每次必死锁 - watchdog 独立线程超时检测 — 不用
alarm()/SIGALRM(signal handler 调 pthread 已知会死锁),而是用独立线程 +sleep()+exit(2)做干净、可移植的检测 _Atomic防数据竞争 —eat_count用atomic_int保证++原子性,避免 5 线程并发写导致计数错乱(未定义行为),为 L64 无锁编程伏笔- 多线程输出非确定判分 — 不依赖逐行 diff,而是检查结构化不变量:退出码(0=正常/2=死锁)、关键诊断行是否存在、禁止行是否缺席
-pthread工程化要求 — 编译期定义_REENTRANT使 libc 走线程安全版本,链接期链接 pthread 库,两处缺一不可
代码框架
#define _POSIX_C_SOURCE 200809L
#define _DEFAULT_SOURCE
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#define N 5 /* 哲学家数量 */
#define TARGET_EAT 100 /* 每人目标进餐次数 */
#define WATCHDOG_TIMEOUT 3 /* watchdog 超时秒数 */
#define THINK_US_MIN 1000 /* 思考最短(微秒) */
#define THINK_US_MAX 3000 /* 思考最长(微秒) */
#define EAT_US 1000 /* 进餐持续(微秒) */
#define GRAB_GAP_US 500 /* 拿左筷到拿右筷间隔(放大持有并等待窗口) */
typedef enum { NAIVE, ASYMMETRIC, ORDERED } Strategy;
/* 共享资源:5 根筷子 = 5 把互斥锁(Coffman 条件 1:互斥) */
static pthread_mutex_t chopstick[N];
/* 起跑门:让 5 个线程同时开始第一轮 */
static pthread_barrier_t start_gate;
/* 原子计数器:避免 eat_count 数据竞争(呼应 L64 _Atomic) */
static atomic_int eat_count[N];
/* 哲学家状态(仅供诊断输出,不参与同步逻辑) */
static atomic_int state[N]; /* 0=THINKING 1=HUNGRY 2=EATING */
static atomic_int holding[N]; /* 当前持有的左筷编号,-1 表示无 */
/* 全局策略 */
static Strategy g_strategy;
/* 打印互斥锁:防止多线程 printf 输出交错 */
static pthread_mutex_t print_lock = PTHREAD_MUTEX_INITIALIZER;
/* 全员完成标志:watchdog 据此判断是正常结束还是死锁 */
static atomic_int all_done_flag = 0;
/* ---------- TODO 1: pickup() — 按策略拿两根筷子 ---------- */
static void pickup(int id) {
int left = id;
int right = (id + 1) % N;
int first, second;
// TODO: 根据 g_strategy 计算 first / second 的拿筷顺序
// 然后 lock(first); holding[id]=first; usleep(GRAB_GAP_US); lock(second)
// state[id] = 2 (EATING)
}
/* ---------- TODO 2: putdown() — 释放两根筷子 ---------- */
static void putdown(int id) {
// TODO: 逆序释放(先 unlock 后拿的,再 unlock 先拿的),holding[id] = -1
}
/* ---------- TODO 3: philosopher() — 线程函数 ---------- */
static void *philosopher(void *arg) {
int id = *(int *)arg;
// TODO: 循环 TARGET_EAT 次:
// think → barrier(第1轮) → HUNGRY → pickup → eat → putdown → eat_count++
// 用 rand_r(&seed) 生成思考时长
return NULL;
}
/* ---------- TODO 4: watchdog() — 超时死锁哨兵 ---------- */
static void print_deadlock_diag(void);
static void *watchdog(void *arg) {
(void)arg;
// TODO: sleep(WATCHDOG_TIMEOUT); 若 all_done_flag → return NULL
// 否则 print_deadlock_diag() + exit(2)
return NULL;
}
/* ---------- TODO 5: build_wait_cycle() — 构建等待环字符串 ---------- */
static void build_wait_cycle(char *out, size_t outsz) {
// TODO: 构造 "P0 → chop1 → P1 → chop2 → ... → P0"
}
/* ---------- TODO 6: coffman_check() — 自检四条件 ---------- */
static void coffman_check(char *out, size_t outsz) {
// TODO: 输出 "Coffman check: 互斥✓ 持有等待✓ 不可剥夺✓ 循环等待✓"
}
/* ---------- 死锁诊断输出(已提供,调用上面的 TODO) ---------- */
static void print_deadlock_diag(void) {
char cycle[256];
char coffman[128];
build_wait_cycle(cycle, sizeof cycle);
coffman_check(coffman, sizeof coffman);
pthread_mutex_lock(&print_lock);
printf("\n!!! DEADLOCK DETECTED !!!\n");
printf("Held: ");
for (int i = 0; i < N; i++) {
int h = atomic_load(&holding[i]);
printf("P%d←chop%d ", i, h < 0 ? i : h);
}
printf("\nWait: ");
for (int i = 0; i < N; i++) {
printf("P%d→chop%d ", i, (i + 1) % N);
}
printf("\nCycle: %s\n", cycle);
printf("%s\n", coffman);
printf("exit code: 2\n");
fflush(stdout);
pthread_mutex_unlock(&print_lock);
}
/* ---------- TODO 7: main() — 初始化与线程编排 ---------- */
int main(int argc, char **argv) {
// TODO: 解析 argv[1] → g_strategy; 打印 banner
// 初始化 mutex/barrier/counters; 创建 detached watchdog
// 创建 N 个 philosopher 线程; join 全部
// all_done_flag=1; 打印 Final Stats; 销毁/return 0
}阅读骨架后,尝试自己填充 7 个 TODO 部分。核心挑战在于:pickup() 中三种策略如何映射到 first/second?putdown() 的逆序释放为什么更好?philosopher() 中 barrier 为什么只在第一轮使用?watchdog() 为什么不用 alarm()?coffman_check() 中四条件为何在死锁时全部 ✓?
TIP
先不要往下翻看参考解答。用纸笔画出五位哲学家围坐圆桌的图,标注筷子编号(哲学家 i 的左筷=i,右筷=(i+1)%5),然后手动追踪 naive 策略下 5 人同时拿左筷后的状态——你会发现每人各持一根、都在等另一根,循环等待环自然浮现。
深度讲解
1. 为什么必须用真实多线程 —— 单线程模拟的致命缺陷
单线程模拟里,"检查两根筷子同时空闲才拿"是一个原子动作————根本不存在"持有一根、等待另一根"的中间态。这意味着 Coffman 死锁四条件里的**"持有并等待"无法在代码中表达**,循环等待环无法构造,学员永远看不到死锁真实发生。
单线程模拟的死锁"真空":
原子操作: if (左筷空闲 && 右筷空闲) { 拿两根 }
↑ 这里不存在"持左等右"的中间窗口 → Coffman 条件 2 永远不成立
真实多线程:
pthread_mutex_lock(&left_chop); // ← 持有了左筷
usleep(GRAB_GAP_US); // ← 等待窗口真实存在!
pthread_mutex_lock(&right_chop); // ← 可能永远阻塞在右筷上
↑ "持左等右"的窗口真实存在 → 死锁可构造、可观测、可检测学术上,单核机器的并发本质就是交错执行 (interleaving)——线程调度器在多个线程间快速切换,模拟"同时发生"的效果。多核机器则是真正的物理并行。无论哪种,共享资源竞争都是真实存在的,这正是死锁问题的温床。
IMPORTANT
用真实多线程教死锁不是"增加复杂度",而是唯一的正确教法。单线程模拟下,死锁的 Coffman 四条件只能背诵——学员可以说出"互斥、持有等待、不可剥夺、循环等待"四个词,但无法在代码中看到它们的肉身。本题中,每个条件都在代码中有精确的对应位置,死锁诊断输出是程序自动生成的——不是"老师说有死锁",而是"程序自己检测到了死锁"。
2. Coffman 死锁四条件——代码映射
死锁发生当且仅当以下四个条件同时同时成立(缺一不可):
(1) 互斥 Mutex — 资源同一时刻只能被一个线程持有
(2) 持有并等待 H & W — 持有已有资源 + 等待新资源
(3) 不可剥夺 No Preempt — 资源只能由持有者主动释放
(4) 循环等待 Circular — 存在一在一条资源等待环在本题中,四个条件的代码化身:
┌──────────────┬─────────────────────────────────────────────┐
│ Coffman 条件 │ 代码化身 │
├──────────────┼─────────────────────────────────────────────┤
│ 互斥 │ pthread_mutex_t — 同一时一时刻只能被一个线程持有 │
│ 持有并等 │ pickup() 先 lock(left) 再 lock(right), │
│ │ 中间 usleep(GRAB_GAP_US) 放大等待窗口 │
│ 不可剥夺 │ mutex 只能由持有者 unlock,别人无法抢走 │
│ 循环等待 │ build_wait_cycle() 构造的 P→chop→P→... 环 │
└──────────────┴─────────────────────────────────────────────┘naive 策略下四条件全部命中 → 死锁必然发生。 预防策略的核心就是破坏其中至少一个条件。本题中 asymmetric 和 ordered 都破坏第 4 条"循环等待"——这是最常见也最实用的死锁预防入口。
NOTE
四个条件中,"循环等待"是唯一可以通过软件设计完全消除的条件。互斥是资源本质决定的(筷子不能两人同时用),持有并等待是"拿一根再拿一根"的语义必然,不可剥夺是 pthread mutex 的设计——所以预防死锁的实用路径就是打破循环等待环。
3. pthread 互斥锁基础——死锁的物理基础
pthread_mutex_t chopstick; /* 声明 */
pthread_mutex_init(&chopstick, NULL); /* 初始化(默认属性)*/
pthread_mutex_lock(&chopstick); /* 加锁:若已被占则阻塞等待 */
/* ... 临界区 ... */
pthread_mutex_unlock(&chopstick); /* 解锁:唤醒一个等待者 */
pthread_mutex_destroy(&chopstick); /* 销毁 */pthread_mutex_lock 是阻塞调用:若锁已被其他线程持有,当前线程会被操作系统挂起(进入等待队列),直到锁被释放再被唤醒。这个"阻塞"塞"正是死锁的物理表现——线程永久挂在 lock() 里出不来。
阻塞的本质(以 P0 为例):
P0 执行 pthread_mutex_lock(&chopstick[1]):
内核检测 chopstick[1] 被 P1 持有
→ 将 P0 移出运行队列,放入 chopstick[1] 的等待队列
→ P0 **永久挂起**,直到 P1 调用 unlock(chopstick[1])
naive 死锁时:P1 也永久挂起在等 chopstick[2]
→ P2 永久挂起在等 chopstick[3]
→ ... 环中所有人互相等待 → 无人能 unlock → 永久死锁4. 筷子编号与哲学家布局
P4
筷4 筷0
P3 P0
筷3 筷1
P2 筷2 P1
哲学家 i 的左筷 = i,右筷 = (i+1) % 5
P0: 左=筷0 右=筷1
P1: 左=筷1 右=筷2
P2: 左=筷2 右=筷3
P3: 左=筷3 右=筷4
P4: 左=筷4 右=筷0关键观察:相邻哲学家共享一根筷子。P0 的右筷 = P1 的左筷 = 筷1。这就是死锁拓扑的基础——5 根筷子形成一个环,5 个哲学家也形成一个环,两个环互温床。
5. 三种拿筷策略深度对比——从病灶到药方
5.1 NAIVE(病灶侧 — 故意死锁)
所有 Pi: lock(筷i); usleep; lock(筷(i+1)%5)
五人同时执行 → 每人各持一根 → 互等另一根 → 循环等待环 → 死锁!
P0: 持筷0 等筷1 ─┐
P1: 持筷1 等筷2 │
P2: 持筷2 等筷3 ├─ 循环等待环!
P3: 持筷3 等筷4 │
P4: 持筷4 等筷0 ─┘
→ 死锁!无人能进餐,watchdog 3 秒后检测并退出naive 策略的教学价值在于:它是对照组`eat_control group)。只有当学员看到了死锁确实发生、看到了等待环的诊断输出、看到了 Coffman 四条件的自我检查,两种预防策略(asymmetric / ordered)才不是"空中楼阁"—而是它们是对比实验中的实验组。
5.2 ASYMMETRIC(非对称 — 打破循环等待)
P0~P3: 先左后右(同 NAIVE)
P4: 先右(筷0)后左(筷4) ← 关键:反转一人
P4 和 P0 都先抢筷0 → 只有一人能拿到
→ 等待环在筷0处必然断开 → 死锁永远不发生 ✓核心机制:打破对称性。在 naive 中,5 个哲学家的行为完全对称——都先左后右——这是循环等待环能闭合的根本原因。asymmetric 策略让一人反向,对称性被破坏,环必在反转处断开。
为什么选 P4 反转(而不是 P0)?
选谁都可以——数学上只要至少一人反转,环就断了。
选 P4 是因为它的左右筷编号跨度最大
(左=4, 右=0, 筷0 也是 P0 的左筷),反转效果最直观。
换成 P0 反转效果相同:P0 先抢筷1 → 与 P1 竞争筷1
→ 环在筷1处断开。选 P4 只是约定俗成。5.3 ORDERED(资源排序 — 偏序预防)
所有 Pi: 先 lock(min(左,右)) 再 lock(max(左,右))
资源存在全局偏序 → 不可能形成环 → 死锁永远不发生 ✓资源排序策略要求所有进程按统一的全局偏序获取资源。本题中"先拿编号小的筷子"意味着:若 P_i 持有筷 a 等待筷 b,则必有 a < b。
循环等待环要求存在 P_0 → ... → P_k → P_0 的链,每条边 P_x 持有 a_x 等待 a_{x+1},则 a_0 < a_1 < ... < a_k < a_0,矛盾(严格递增不可能成环)。这。这是数学证明等待,与调度无关。
IMPORTANT
ordered 是三策略中最通用的死锁预防策略。asymmetric 只适用于环形拓扑(你依赖"哪个进程反转能断开环"的领域知识),而 ordered 适用于任意资源图**——只要能给所有资源编号并规定"总是从小到大申请",死锁就不可能发生。
5.4 三策略量化对比
| 维度 | naive | asymmetric | ordered |
|---|---|---|---|
| 死锁 | 必然发生 | 永不发生 | 永不发生 |
| 破坏哪个 Coffman 条件 | 无(全保留) | 循环等待 | 循环等待 |
| 机制 | 无防御 | 打破对称性 | 资源偏序 |
| 退出码 | 2 | 0 | 0 |
| 公平性 | 全饿死 | 轮询公平 | 轮询公平 |
| 通用性 | — | 仅环形拓扑 | 任意资源图 |
| 代码复杂度 | 最简 | 多一个 if/else | min/max 分支 |
| 实际工程适用 | 警示用 | 特定场景 | 最常用 |
6. 死锁触发保障:barrier 同步起跑 + GRAB_GAP_US 窗口放大
naive 策略下,若 5 个线程启动有先后,可能某个线程已吃完放下筷子,死锁就不成立了。为稳定触发死锁(消除 flaky),用两个机制配合:
┌─────────────────────────────────────────────────┐
│ start_gate = pthread_barrier_init(NULL, NULL, 5)│
│ │
│ 线程 P0: barrier_wait ──┐ │
│ 线程 P1: barrier_wait ──┤ │
│ 线程 P2: barrier_wait ──┼──→ 5 人到齐,同时放行│
│ 线程 P3: barrier_wait ──┤ │
│ 线程 P4: barrier_wait ──┘ │
│ │
│ 放行瞬间 5 人几乎同时 lock(左筷) → 各持一根 │
│ + GRAB_GAP_US 窗口放大 → 循环等环等待环必然闭合 │
└─────────────────────────────────────────────────┘barrier 的作用:让 5 个线程第一轮同时起跑,消除"线程创建先后"带来的交错不确定性。注意 barrier 只在第一轮使用——后续轮次各哲学家自然错开(因为思考时间随机),安全策略下强行同步反而可能制造人为拥堵。
GRAB_GAP_US 的作用:在拿完左筷后 usleep(500μs) 制造一个等待窗口。这 500 微秒确保其他哲学家有机会也拿到各自的左筷,形成"每人各持一根"的局面。如果没有这个延迟,在单核机器上,一个线程可能连续拿到两根筷、吃完、释放——其他线程根本没机会参与竞争。
CAUTION
barrier + GRAB_GAP_US 是让 naive "每次必死锁"的关键。不用 barrier → 死锁变成 flaky(时锁时不锁)。不用 GRAB_GAP_US → 在特定调度下可能不死锁。教学场景下,稳定可复现比随机性更重要——你要确保每个学员运行时都能看到死锁。
7. watchdog 死锁检测——独立线程模式的智慧
watchdog 线程:
sleep(WATCHDOG_TIMEOUT); /* 等待 3 秒 */
if (all_done_flag) return; /* 正常完成 */
else {
print_deadlock_diag(); /* 打印等待环 + Coffman 自检 */
exit(2); /* 退出码 2 = 死锁 */
}为什么用独立线程而非 alarm()/SIGALRM? 因为 signal handler 里调用 pthread 函数是已知的死锁地雷——linuxthreads FAQ 明确指出 pthread 函数非 async-signal-safe。
alarm() + SIGALRM 的危险路径:
1. alarm(3) 注册 3 秒后触发 SIGALRM
2. 3 秒后,信号可能在任意线程被投递(包括持有 mutex 的线程)
3. 信号 handler 试图执行 printf / pthread_mutex_lock
→ 如果当前线程已持有 print_lock → 自己等自己的锁 → handler 死锁!
4. 整个程序fe,在 signal handler 里,连诊断信息都没打出来
独立 watchdog 线程:
- 就是一个普通线程,调用普通 sleep() + exit()
- 不涉及 signal、不抢占其他线程的上下文
- 干净、可移植、无副作用8. 完整死锁过程逐轮追踪(naive 策略)
下表展示 naive 策略下从起跑到死锁的微观过程(时间线):
| 时刻 | P0 | P1 | P2 | P3 | P4 | 事件 |
|---|---|---|---|---|---|---|
| t0 | barrier | barrier | barrier | barrier | barrier | 5 人在起跑门等待 |
| t1 | 持筷0 | 持筷1 | 持筷2 | 持筷3 | 持筷4 | barrier 放行,5 人同时 lock 左筷 |
| t2 | usleep | usleep | usleep | usleep | usleep | GRAB_GAP_US 放大持有并等待窗口 |
| t3 | 等筷1 | 等筷2 | 等筷3 | 等筷4 | 等筷0 | 5 人尝试 lock 右筷,全被阻塞 |
| t4 | 阻塞 | 阻塞 | 阻塞 | 阻塞 | 阻塞 | 循环等待环闭合,无人能前进 |
| ... | ... | ... | ... | ... | ... | 永久阻塞 |
| t0+3s | — | — | — | — | — | watchdog 超时,打印诊断,exit(2) |
Coffman 四条件此此时全部命中:互斥(mutex)✓ 持有并等待(持左等右)✓ 不可剥夺(mutex 不能被抢)✓ 循环等待(P0→筷1→P1→筷2→...→P0)✓。
9. 安全策略运行追踪(asymmetric / ordered)
| 时刻 | P0 | P1 | P2 | P3 | P4 | 事件 |
|---|---|---|---|---|---|---|
| t0 | barrier | barrier | barrier | barrier | barrier | 起跑门等待 |
| t1 | 持筷0 | 持筷1 | 持筷2 | 持筷3 | 抢筷0失败 | P4 先抢筷0,但 P0 已拿到 |
| t2 | 进餐 | 进餐 | 进餐 | 进餐 | 等待 | P0~P3 进餐,P4 等筷0释放 |
| t3 | 放筷 | 放筷 | 放筷 | 放筷 | 持筷0 | P0 释放筷0与筷1,P4 拿到筷0 |
| t4 | 思考 | 思考 | 思考 | 思考 | 持筷4 | P4 拿到筷4(没有竞争),进餐 |
| ... | ... | ... | ... | ... | ... | 轮流进餐,无死锁 |
| 最终 | 100次 | 100次 | 100次 | 100次 | 100次 | 全员完成,exit(0) |
关键观察:P4 反转拿筷顺序后,与 P0 在筷0上形成竞争——必有一人失败。失败者不持有任何资源(或只持有一根但立即释放),循环等待环在此必然断开。
10. 边界情况分析
| 场景 | 期望行为 | 原因 |
|---|---|---|
| naive 单次运行 | 3 秒内死锁,exit(2) | barrier+gap 保证循环等待闭合 |
| naive 高并发核数 | 仍死锁 | 死锁与核数无关,只与拿筷顺序有关 |
| asymmetric 单核 | 不死锁 | 非对称破破坏循环等待,与调度无关 |
| ordered 线线程数变化 | 不死锁 | 资源偏序是数学保证 |
| watchdog 超时设太短 | 误报死锁 | safe 策略未跑完就被判死锁 |
| watchdog 超时设太长 | naive 卡太久 | 死锁后要等更久才退出 |
| TARGET_EAT 太小 | naive 可能不死死锁 | 轮次少,错开起跑概率增大 |
| 不用 barrier | naive 可能 flaky | 线程启动有先后,错开拿筷 |
| eat_count 不用原子 | 数据竞争 | 多线程并发写共享变量(UB) |
11. 单线程模拟 vs 真实多线程
| 维度 | 单线程状状态机模拟 | 真实多线程(本题) |
|---|---|---|
| 资源竞争 | 无(顺序访问) | 真实竞争 |
| 持有并等待 | 无法表达 | pickup() 中间态真实 |
| 死锁可观测 | 永不发生 | naive 必然发生 |
| Coffman 四条件 | 只能背诵 | 代码可映射、可自检 |
| 输出确定性 | 完全确定 | 非确定(靠不变量判分事件 |
| 教学价值 | 工程化/格式化 | 并发思维核心 |
| 依赖 | 无 | pthreads(系统库) |
12. 常见错误与陷阱
| 错误 | 后果 | 正确做法 |
|---|---|---|
watchdog 用 alarm()+SIGALRM | signal handler 调 pthread 自身死锁 | 用独立 watchdog 线程 + exit() |
eat_count 用普通 int | 数据竞争(UB),计数错乱 | 用 _Atomic int |
不用 print_lock 保护 printf | 输出交错,无法判分 | 所有输出用 mutex 包裹 |
| naive 不加 barrier | 死锁 flaky(时锁时不锁) | pthread_barrier_wait 同步起跑 |
| naive 不加 GRAB_GAP_US | 拿两根筷太快,可能错过死锁窗口 | 拿完左筷 usleep 放大窗 事件 |
| putdown 顺序错 | 语义混乱,某些场景难推理 | 逆序释放 |
忘了 pthread_join | main 提前退出,线程被杀 | join 所有 philosopher |
Makefile 漏 -pthread | 链接失败(undefined reference) | 编译期 + 链接期都加 -pthread |
holding[] 不用原子 | 诊断输出数据竞争 | 用 _Atomic int |
| barrier 每轮都用 | 安全策略也被同步成类似死锁 | barrier 只用于第一轮 |
参考解答
pickup() — 按策略拿两根筷子
/* 按策略决定拿筷顺序,并依次 lock */
static void pickup(int id) {
int left = id;
int right = (id + 1) % N;
int first, second;
switch (g_strategy) {
case NAIVE:
first = left;
second = right;
break;
case ASYMMETRIC:
/* P4(id==N-1)反转:先右后左,打破对称性 */
if (id == N - 1) {
first = right;
second = left;
} else {
first = left;
second = right;
}
break;
case ORDERED:
/* 始终先拿编号小的,再拿编号大的 */
first = (left < right) ? left : right;
second = (left < right) ? right : left;
break;
default:
first = left;
second = right;
}
pthread_mutex_lock(&chopstick[first]);
atomic_store(&holding[id], first); /* 记录持有第一根筷(诊断用) */
usleep(GRAB_GAP_US); /* 放大"持有并等待"窗口 */
pthread_mutex_lock(&chopstick[second]);
atomic_store(&state[id], 2); /* EATING */
}要点:asymmetric 只反转 P4(id==N-1),因为 P4 的右筷=筷0=P0 的左筷,反转效果最直观。ordered 用 min/max 保证偏序获取。
putdown() — 逆序释放两根筷子
/* 逆序释放筷子:先放后拿的,再放先拿的(锁的最佳实践) */
static void putdown(int id) {
int left = id;
int right = (id + 1) % N;
int first, second;
switch (g_strategy) {
case NAIVE:
first = left; second = right; break;
case ASYMMETRIC:
if (id == N - 1) { first = right; second = left; }
else { first = left; second = right; }
break;
case ORDERED:
first = (left < right) ? left : right;
second = (left < right) ? right : left;
break;
default:
first = left; second = right; break;
}
pthread_mutex_unlock(&chopstick[second]); /* 先释放后拿的 */
pthread_mutex_unlock(&chopstick[first]); /* 再释放先拿的 */
atomic_store(&holding[id], -1);
}要点:逆序释放(先 unlock second 再 unlock first)是锁的最佳实践——"后进先出"。在更复杂的锁嵌套场景中,逆序释放可以减少死锁风险。
philosopher() — 线程函数
static void *philosopher(void *arg) {
int id = *(int *)arg;
unsigned int seed = (unsigned int)(id + 1); /* 可复现的思考时长种子 */
for (int round = 0; round < TARGET_EAT; round++) {
/* 1. 思考 */
int think_us = THINK_US_MIN +
(rand_r(&seed) % (THINK_US_MAX - THINK_US_MIN + 1));
usleep(think_us);
/* 2. 第一轮同步起跑(barrier 同步点) */
if (round == 0) {
pthread_barrier_wait(&start_gate);
/* 5 人到齐后同时放行 */
}
/* 3. 饥饿状态 */
atomic_store(&state[id], 1); /* HUNGRY */
/* 4. 拿筷子(可能阻塞) */
pickup(id);
/* 5. 进餐 */
usleep(EAT_US);
/* 6. 放筷子 */
putdown(id);
/* 7. 思考状态,计数 */
atomic_store(&state[id], 0); /* THINKING */
atomic_fetch_add(&eat_count[id], 1);
}
return NULL;
}要点:barrier 只在第一轮使用——后续各轮哲学家已经自然错开(思考时间随机),不需要再同步。rand_r 是线程安全的随机数函数(每个线程维护自己的 seed),避免共享全局 rand() 的数据竞争。
watchdog() — 超时死锁哨兵
static void *watchdog(void *arg) {
(void)arg;
sleep(WATCHDOG_TIMEOUT); /* 等待 3 秒 */
if (atomic_load(&all_done_flag)) {
/* 正常结束:5 个哲学家都在 3 秒内吃完 → 无死锁 */
return NULL;
}
/* 超时 → 死锁!打印诊断并退出 */
print_deadlock_diag();
exit(2);
return NULL; /* 实际不会执行到这里 */
}要点:独立线程 + sleep() + exit(2) 完全避开了 signal 与 pthread 混用的地雷。exit(2) 直接终止整个进程(包括所有陷入死锁的哲学家电线程),干净利落。
build_wait_cycle() + coffman_check() — 诊断辅助
/* 构建循环等待环字符串 */
static void build_wait_cycle(char *out, size_t outsz) {
int pos = 0;
for (int i = 0; i < N; i++) {
pos += snprintf(out + pos, outsz - pos,
"P%d → chop%d → ", i, (i + 1) % N);
}
snprintf(out + pos, outsz - pos, "P0");
}
/* 自检 Coffman 四条件 */
static void coffman_check(char *out, size_t outsz) {
snprintf(out, outsz,
"Coffman check: 互斥✓ 持有等待✓ 不可剥夺✓ 循环等待✓");
}要点:build_wait_cycle 构造的诊断串是死锁核心结构的可视化——P0→筷1→P1→...→P0 的环。coffman_check 在死锁现场总是四条件全部命中,因为代码本身保证了前三个条件始终成立(mutex 互斥、先拿后等的逻辑、mutex 不可剥夺),而循环等待在 naive 策略下必然成立。
main() — 完整线程编排
int main(int argc, char **argv) {
/* 1. 解析策略 */
if (argc >= 2) {
if (strcmp(argv[1], "naive") == 0)
g_strategy = NAIVE;
else if (strcmp(argv[1], "asymmetric") == 0)
g_strategy = ASYMMETRIC;
else if (strcmp(argv[1], "ordered") == 0)
g_strategy = ORDERED;
else
g_strategy = NAIVE;
} else {
g_strategy = NAIVE;
}
/* 2. 打印 banner(判分用——必须包含 strategy=xxx) */
printf("=== Dining Philosophers (strategy=%s) ===\n",
strategy_name(g_strategy));
printf("N=%d, target=%d meals each, watchdog=%ds\n",
N, TARGET_EAT, WATCHDOG_TIMEOUT);
/* 3. 初始化 N 把 mutex */
for (int i = 0; i < N; i++)
pthread_mutex_init(&chopstick[i], NULL);
/* 4. 初始化起跑门(N 人到齐才放行) */
pthread_barrier_init(&start_gate, NULL, N);
/* 5. 初始化计数器 */
for (int i = 0; i < N; i++) {
atomic_init(&eat_count[i], 0);
atomic_init(&state[i], 0); /* THINKING */
atomic_init(&holding[i], -1); /* 无持有 */
}
/* 6. 创建 detached watchdog(自己会 exit,无需 join) */
pthread_t wdog;
pthread_t phils[N];
int ids[N];
pthread_create(&wdog, NULL, watchdog, NULL);
pthread_detach(wdog);
/* 7. 创建 N 个 philosopher 线程 */
for (int i = 0; i < N; i++) {
ids[i] = i;
pthread_create(&phils[i], NULL, philosopher, &ids[i]);
}
/* 8. join 所有 philosopher */
for (int i = 0; i < N; i++)
pthread_join(phils[i], NULL);
/* 9. 通知 watchdog 正常结束 */
atomic_store(&all_done_flag, 1);
/* 10. 打印 Final Stats */
int done = 1;
for (int i = 0; i < N; i++)
if (atomic_load(&eat_count[i]) < TARGET_EAT)
done = 0;
if (done) {
printf("\n=== Final Stats ===\n");
for (int i = 0; i < N; i++)
printf("P%d ate %d\n", i, atomic_load(&eat_count[i]));
printf("all philosophers finished\n");
}
/* 11. 清理资源 */
for (int i = 0; i < N; i++)
pthread_mutex_destroy(&chopstick[i]);
pthread_barrier_destroy(&start_gate);
return 0;
}要点:watchdog 用 pthread_detach 创建——因为它在 naive 策略下会直接 exit(2) 而不是 return,main 无法 join 它。detached 线程结束时系统自动回收资源,不需要被 join。
对照检查:pickup 中 asymmetric 是否正确反转了 P4?putdown 是否逆序释放?philosopher 中 barrier 是否只在第一轮使用?watchdog 是否先检查 all_done_flag?main 中 watchdog 是否 detached?所有 printf 输出是否包含判分所需的关键词(
strategy=xxx、DEADLOCK DETECTED、all philosophers finished)?
课堂讨论
Q1: 为什么 naive 一定会死锁?barrier 和 GRAB_GAP_US 各起什么作用?
barrier 让 5 个线程同时开始第一轮,,保证 5 人几乎同一瞬间执行 lock(左筷);GRAB_GAP_US 在拿完左筷后制造一个等待窗口,让其他线程有机会也拿到各自的左筷。两者配合使" 每人各持一根"的局面稳定形成,循环等待环必然闭合。
没有 barrier → 线程启程启动有先后,可能某线程已吃完释放,死锁就不成立(变成 flaky)。没有 GRAB_GAP_US → 在单核机器上,一个线程可能连续拿到两根筷、吃完、释放——其他线程根本没机会参与抢筷,"持有并等待"的窗口消失,死锁可能不触发。
barrier + GRAB_GAP_US = 确定性死锁触发器,是教学场景下消除 flaky 的标准手段。
Q2: asymmetric 为什么选 P4 反转,选 P0 行不行?
选谁都行,关键是打破对称性——只要至少一个进程的拿筷顺序与其他人不同,循环等待环就在该进程处断开。本题选 P4(编号最大)只是约定俗成,因为 P4 的右筷=筷0=P0 的左筷,反转 P4 后与 P0 在筷0上形成直接竞争,效果最直观。
换成 P0 反转:P0 先抢筷1(P1 的左筷),与 P1 在筷1上竞争,效果相同。换成 P2 反转也行。数学上,任选一个进程反转都能破坏 5 元环——你只需要断开环中的一个节点。
Q3: ordered 策略为什么能保证无死锁?请给出数学证明思路
资源排序策略要求所有进程按统一的全局偏序获取资源。本题中"先拿编号小的筷子"意味着:若 P_i 持有筷 a 等待筷 b,则必有 a < b。
假设存在死锁,则存在循环等待环 P_0 → P_1 → ... → P_k → P_0。对于环中每条边:
- P_0 持有 a_0 等待 a_1 → a_0 < a_1
- P_1 持有 a_1 等待 a_2 → a_1 < a_2
- ...
- P_k 持有 a_k 等待 a_0 → a_k < a_0
综合得到 a_0 < a_1 < a_2 < ... < a_k < a_0——严格递增不可能成环,矛盾。故死锁不可能发生。
这是数学证明,与调度器、核数、运行环境完全无关——ordered 函数是无条件安全的。
Q4: 为什么不能用 alarm()/SIGALRM 做超时检测?
alarm() 注册的 signal handler 在任意线程被异步中断执行。pthread 函数数(如 pthread_mutex_lock)不是 async-signal-safe 的——在 signal handler 里调用它们会导致程序自身死锁或未定义行为。
具体地:如果信号恰好投递到一个持有 print_lock 的线程,handler 尝试 pthread_mutex_lock(&print_lock) → 这个线程已经在持有 lock → 自己等自己的锁 → handler 死锁。诊断信息还没打出来,程序先卡在 handler 里了。
独立 watchdog 线程用普通 sleep() + exit() 完全避开这个地雷——不依赖信号、不抢占其他线程上下文、不调用非 async-signal-safe 函数。这是并发程序可观测性的标准模式。
此外,exit(2) 直接终止整个进程,比 return 更干净——因为 naive 死锁时那些哲学家电线程永远卡在 pthread_mutex_lock() 里,return 它们永远不可能。
Q5: eat_count 为什么必须用 _Atomic?普通 int 会怎样?
5 个线程并发执行 eat_count[i]++,这其实是"读-改-写"三步操作,不是原子的:
时间线(普通 int 的数据竞争):
T1: 读 eat_count[0]=5
T2: 读 eat_count[0]=5 ← 读到旧值!
T1: 计算 5+1=6,写回
T2: 计算 5+1=6,写回 ← 丢失了一次更新!结果应该是 7 却成了 6C11 标准定义多线程下对非原子变量的并发访问为未定义行为(UB)——编译器可能做任何优化,包括但不限于:缓存旧值、重排指令、甚至生成完全错误的代码。
_Atomic int 保证 ++ 是原子的、对其他线程立即可见的。这里用的是 atomic_fetch_add——C11 提供的原子递增操作。这也为 Lesson 64 的无锁环形缓冲区(用 _Atomic 实现无锁并发)埋下认知伏笔。
Q6: 真实多线程输出非确定,怎么自动化判分?
不依赖逐行精确 diff,而是检查结构化不变量——这是并发程序自动化测试的核心思想:
| 断言类型 | naive 策略 | asymmetric/ordered 策略 |
|---|---|---|
| 退出码 | exit_code = 2 | exit_code = 0 |
| 必须包发生 | DEADLOCK DETECTED、Cycle:、Coffman check、互斥、持有等待、不可剥夺、循环等待 | strategy=xxx、all philosophers finished、P0 ate 100...P4 ate 100 |
| 禁止包含 | — | DEADLOCK |
判分器用 stdout_contains / stdout_not_contains / exit_code 三个断言组合,不关心输出的具体顺序、时间戳或空格差别。这种"只检验不变量"的策略是并发程序测试的标准方法——Google Test、ThreadSanitizer 等都遵循类似哲学。
课后练习
实现左撇子策略。在三种已有策略之外,新增一个
lefty策略:让奇数编号的哲学家先拿左筷、偶数编号的先拿右筷。分析这种策略是否会导致死锁,并编写测试验证。知识点提示:先画等待图——相邻哲学家编号奇偶交替,左右手策略也交替。检查是否存在环。提醒:5 个哲学家奇偶数量不均(3 奇 2 偶),可能形成更微妙的死锁场景。
参考解答
c/* 左撇子策略:奇数先左后右,偶数先右后左 */ case LEFTY: if (id % 2 == 0) { first = right; second = left; /* 偶数:先右后左 */ } else { first = left; second = right; /* 奇数:先左后右 */ } break;分析:偶数哲学家(P0, P2, P4)先抢右筷,奇数哲学家(P1, P3)先抢左筷。由于 5 个哲学家中奇数占 3、偶数占 2,奇偶不均衡导致某些相邻的哲学家有相同的左右手策略——例如 P4(偶数,先右后左)的左筷=筷4、右筷=筷0。P0(偶数,先右后左)的左筷=筷0、右筷=筷1。两者都先抢自己的右筷——筷0 和筷1,没有直接竞争。但实际上 P4 的右筷=筷0=P0 的左筷——而 P0 先抢的是右筷(筷1),不是左筷(筷0)。所以 P4 先抢筷0 没有对手,能拿到。拿到后 P4 等左筷(筷4),P3 持有筷4...
结论:在 N=5 且奇偶不均的情况下,这个策略也会死锁——只是死锁环的拓扑更复杂。资源排序(ordered)是唯一有数学保证的通用方案,不受 N 奇偶性影响。
使用 trylock 实现死锁避免。将
pickup()中的pthread_mutex_lock改为pthread_mutex_trylock:先 trylock 第一根,若成功再 trylock 第二根;若第二根失败,释放第一根,usleep 短暂随机时间后重试。验证 naive 策略下不会死锁。知识点提示:
pthread_mutex_trylock在锁被占用时立即返回EBUSY而非阻塞——不满足"持有并等待"的 Coffman 条件 2。这是死锁避免(avoidance)而非死锁预防(prevention)——与 Banker 算法同类,属于动态策略。参考解答
c/* trylock 版本:避免死锁(动态策略) */ static void pickup_trylock(int id) { int left = id, right = (id + 1) % N; unsigned int seed = (unsigned int)(id + 1000); while (1) { /* 尝试拿左筷 */ if (pthread_mutex_trylock(&chopstick[left]) != 0) { usleep(100 + (rand_r(&seed) % 400)); continue; } /* 尝试拿右筷 */ if (pthread_mutex_trylock(&chopstick[right]) == 0) { /* 两根都拿到! */ atomic_store(&holding[id], left); atomic_store(&state[id], 2); /* EATING */ return; } /* 右筷没拿到 → 释放左筷,随机退避,重试 */ pthread_mutex_unlock(&chopstick[left]); usleep(100 + (rand_r(&seed) % 400)); } }注意:trylock 避免死锁的代价是可能引入活锁(livelock)——几个线程反复拿筷、放筷、重试,谁都无法同时拿到两根。随机退避(random backoff)是缓解活锁的标准手段。
与 asymmetric/ordered 的核心区别:trylock 是动态策略(运行时检测冲突并回避),asymmetric/ordered 是静态策略(设计时通过排序消除死锁可能)。静态策略更优(无重试开销),但并非所有场景都能应用。
扩展为 M 个哲学家。修改程序,使哲学家数量可配置(通过
-n M命令行参数传入)。验证:M=3 时 naive 也会死锁吗?M=4 呢?分析 naive 策略死锁的条件。知识点提示:循环等待环的形成条件是所有哲学家对称地先拿左后拿右。这个条件对任何 M ≥ 2 都成立。但实际死锁的触发概率随 M 变化——因为线程调度窗口期变长(更多人等 barrier)。
参考解答
c/* 可配置哲学家数量 */ int N = 5; /* 默认值(注意:原来是 #define,改为变量需调整 mutex/barrier 数组) */ int main(int argc, char **argv) { /* 解析 -n M */ for (int i = 1; i < argc; i++) { if (strcmp(argv[i], "-n") == 0 && i + 1 < argc) { N = atoi(argv[++i]); if (N < 2) N = 2; } } /* ... 其余初始化用 N 替代硬编码的 5,mutex 和 thread 数组用动态分配 ... */ }死锁条件分析:
- M=2: 两个哲学家,P0 持筷0 等筷1,P1 持筷1 等筷0 → 必然死锁 ✓
- M=3: P0 持筷0 等筷1,P1 持筷1 等筷2,P2 持筷2 等筷0 → 必然死锁 ✓
- M=4: 类似,必然死锁 ✓
- M=1: 需要两根筷子才能吃,只有一副 → 饿死(不是死锁——是资源不足)
关键结论:只要 M ≥ 2 且所有人的拿筷顺序相同(先左后右),死锁必然发生。这是完全对称拿筷策略的固有属性,与 M 无关。但当 M 较大时,需要动态分配 mutex 数组和线程数组——不能再依赖
#define N 5的静态数组。
参考资料
Dijkstra, E.W. "Hierarchical ordering of sequential processes" (1971) — 哲学家就餐问题原论文,首次提出资源排序预防死锁
"The solution to the dining philosophers problem illustrates that a total ordering of resources prevents circular wait, the fourth necessary condition for deadlock."
Coffman, E.G. et al. "System Deadlocks" (1971) — Coffman 四条件首次系统化论述,死锁理论的奠基之作
"A deadlock occurs if and only if all four conditions hold simultaneously: mutual exclusion, hold and wait, no preemption, and circular wait."
《操作系统概念》(Silberschatz) 第 7 章 — 死锁预防/避免/检测/恢复的完整体系,包含 Banker 算法
《现代操作系统》(Tanenbaum) 第 6 章 — 死锁的实用视角,包含哲学家就餐问题的多种解法
Wikipedia: Dining philosophers problem — 问题背景、多种解法(Chandy/Misra、资源层级等)的全面综述
Wikipedia: Deadlock — 死锁的 Coffman 条件、活锁与饿死的区别
POSIX.1-2017 standard 是一把
pthread_mutex_*、pthread_barrier_*、pthread_create/pthread_join的标准接口定义linuxthreads FAQ — signal 与线程混用的陷阱(
alarm()+SIGALRM在 pthread 程序中不安全的原因)
"Deadlock is the ultimate manifestation of uncontrolled resource competition. Understanding it is not about memorizing four conditions — it is about seeing those conditions become alive in your code." — Adapted from Dijkstra