本檔收錄 6 題由易到難的練習。建議流程：先蓋住解答自己寫，卡住再看
「解題思路」，寫完對照「參考解答」。每題完整程式碼在 exercises/：
cc -std=c11 -Wall -Wextra -pthread -o /tmp/out exercises/ex1_thread_ids.c && /tmp/out本單元所有程式編譯必須加 -pthread。並行程式的輸出天生帶有
不確定性（交錯順序、競爭統計、計時數字每次不同）——各題的
「範例輸出」是某一次實際執行的結果，講解中會註明哪些部分會變動。
練習 1：八個執行緒報到（難度：基礎）
題目
建立 8 個執行緒，每個印出自己的編號與 pthread_self() 的值，
主執行緒 join 全部後收工。跑幾次觀察輸出順序的變化。
解題思路
- create/join 的最小完整骨架——之後每一題都是它的擴充：
main: create x8 ──> worker 們自由奔跑 ──> join x8- 參數「一人一格」的鐵律：8 個執行緒要知道自己的編號。
錯誤寫法
pthread_create(..., &i)把同一個 i 的位址 發給所有人——i 在迴圈中持續變動，執行緒醒來時看到的 是「當下的 i」，可能重複、可能是 8。正解：struct worker_arg args[8]，第 i 個執行緒拿&args[i]， 而且陣列活得比所有執行緒久（main 的堆疊、join 之後才離開 作用域）。 - pthread_self 是什麼：目前執行緒的識別碼，型別
pthread_t是不透明的（macOS 上實際是指標、Linux 上 是無號整數）——印出來只能當「身分證號碼」看， 比對要用pthread_equal()，不能==（可攜性）。 - 輸出交錯就是課程重點：printf 各自為政，順序由排程器 即興決定。這次 0-7 整齊排隊，下次可能亂序—— 「不確定性」是並行的本性，也是後面幾題要馴服的對象。
參考解答
/*
* 練習 1：八個執行緒報到 — create/join 與輸出交錯觀察
*
* 編譯：cc -std=c11 -Wall -Wextra -pthread -o /tmp/out ex1_thread_ids.c
* 執行：/tmp/out
*
* 重點：參數「一人一格」的鐵律、pthread_self 的身分證、
* 輸出交錯 = 排程器的即興演出（每次執行都不同）。
*/
#include <stdio.h>
#include <pthread.h>
#define NTHREADS 8
/* 每個執行緒的專屬參數：一人一格，絕不共用迴圈變數 */
struct worker_arg {
int no; /* 我們給的編號（0..7） */
};
static void *worker(void *arg)
{
struct worker_arg *w = arg; /* void* 轉回真正的型別 */
/*
* pthread_self() 回傳目前執行緒的 pthread_t。
* pthread_t 是不透明型別（macOS 上是指標、Linux 上是整數），
* 印出時 cast 成 void* 只作辨識用，不可比大小；
* 比對執行緒要用 pthread_equal()。
*/
printf(" worker %d reporting, pthread_self()=%p\n",
w->no, (void *)pthread_self());
return NULL;
}
int main(void)
{
pthread_t tid[NTHREADS];
struct worker_arg args[NTHREADS]; /* 參數陣列：與執行緒同壽命 */
printf("main: creating %d threads...\n", NTHREADS);
for (int i = 0; i < NTHREADS; i++) {
args[i].no = i; /* 先填好，再交出去 */
if (pthread_create(&tid[i], NULL, worker, &args[i]) != 0) {
perror("pthread_create");
return 1;
}
}
/* join：等每個執行緒結束並回收資源（不 join 會累積 zombie） */
for (int i = 0; i < NTHREADS; i++)
pthread_join(tid[i], NULL);
printf("main: all %d threads joined.\n", NTHREADS);
printf("\nnote: the report order above is the scheduler's choice --\n"
"run it again and it will likely differ. that nondeterminism\n"
"is exactly why shared data needs synchronization.\n");
return 0;
}程式碼講解
args[i].no = i先填好再 create——create 之後執行緒 隨時可能開跑，參數必須在交棒前就緒。pthread_create的回傳值要檢查：它不設 errno， 直接回傳錯誤碼（本題簡化用 perror，嚴謹寫法是strerror(rc)）。- join 迴圈與 create 迴圈分開——如果 create 完立刻 join， 就變成「一次只跑一條」的偽並行。
- 這次執行恰好按 0..7 順序輸出——是排程器的巧合不是保證； printf 對 stdout 的單次呼叫在 POSIX 下是執行緒安全的 （不會字元交錯），但行與行的順序沒有任何承諾。
- 易錯處：忘記 join 就讓 main return——整個行程結束， 沒跑完的執行緒被無聲砍掉。
範例輸出
main: creating 8 threads...
worker 0 reporting, pthread_self()=0x16d6eb000
worker 1 reporting, pthread_self()=0x16d777000
worker 2 reporting, pthread_self()=0x16d803000
worker 3 reporting, pthread_self()=0x16d88f000
worker 4 reporting, pthread_self()=0x16d91b000
worker 5 reporting, pthread_self()=0x16d9a7000
worker 6 reporting, pthread_self()=0x16da33000
worker 7 reporting, pthread_self()=0x16dabf000
main: all 8 threads joined.
note: the report order above is the scheduler's choice --
run it again and it will likely differ. that nondeterminism
is exactly why shared data needs synchronization.（報到順序與 pthread_self 的值每次執行都不同。）
練習 2：分段加總 — 回傳 malloc 的部分和（難度：基礎）
題目
把 1..1000000 的加總分給 4 個執行緒，每個執行緒把部分和
放進 malloc 的記憶體並以回傳值交回；main 用
pthread_join 收集、加總、釋放，與公式 N(N+1)/2 對帳。
解題思路
- 執行緒怎麼「回傳」東西：進入點簽名是
void *worker(void *)——回傳一個指標，join 的第二個參數 接住它。問題是：指標指向的東西必須在執行緒結束後 還活著。 - 三個候選方案：
-
return &local;——區域變數在執行緒結束時隨堆疊蒸發， join 拿到懸空指標。絕對錯誤； - 寫進參數結構的欄位——可行（很多程式這樣做）； - malloc 一塊回傳——本題指定的做法，通用性最好 （回傳的東西大小不定時尤其有用），代價是呼叫端 要負責 free。 - 分工：N/4 一段，最後一段收尾到 N——N 不整除 執行緒數時不能漏元素（本題剛好整除，但程式要寫對）。
- 這裡不需要鎖！ data 陣列是唯讀共享（大家只讀）， 每個執行緒寫自己的 malloc 塊（不共享）—— 「沒有共享的寫入就沒有 race」，最好的同步是不用同步。
- 對帳：等差級數公式給出期望值，MATCH/MISMATCH 一翻兩瞪眼——並行程式的測試一定要有確定性的驗證錨點。
參考解答
/*
* 練習 2：分段加總 — 執行緒回傳 malloc 的部分和
*
* 編譯：cc -std=c11 -Wall -Wextra -pthread -o /tmp/out ex2_partial_sums.c
* 執行：/tmp/out
*
* 重點：執行緒回傳值的正確做法——malloc 的指標經 pthread_join
* 交回 main（回傳區域變數位址是懸空指標！）；
* main 收集、加總、釋放。
*/
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#define N 1000000
#define NTHREADS 4
static int data[N]; /* 唯讀共享：執行緒只讀，不需要鎖 */
struct seg {
int lo, hi; /* 本執行緒負責的半開區間 [lo, hi) */
};
/*
* 部分和放進 malloc 的記憶體回傳。
* 錯誤示範（絕對不行）：
* long sum; ...; return ∑ // 執行緒結束堆疊即回收 -> 懸空
*/
static void *sum_worker(void *arg)
{
const struct seg *s = arg;
long *result = malloc(sizeof *result); /* 活得比執行緒久 */
if (result == NULL)
return NULL; /* main 要檢查 NULL */
long sum = 0;
for (int i = s->lo; i < s->hi; i++)
sum += data[i];
*result = sum;
printf(" worker [%7d, %7d) partial sum = %ld\n", s->lo, s->hi, sum);
return result; /* 交棒給 join */
}
int main(void)
{
/* 填測試資料：1..N，總和公式 N*(N+1)/2 可對帳 */
for (int i = 0; i < N; i++)
data[i] = i + 1;
long expect = (long)N * (N + 1) / 2;
pthread_t tid[NTHREADS];
struct seg segs[NTHREADS];
int chunk = N / NTHREADS;
printf("summing 1..%d with %d threads:\n", N, NTHREADS);
for (int i = 0; i < NTHREADS; i++) {
segs[i].lo = i * chunk;
/* 最後一段收尾到 N：N 不整除時不能漏元素 */
segs[i].hi = (i == NTHREADS - 1) ? N : (i + 1) * chunk;
if (pthread_create(&tid[i], NULL, sum_worker, &segs[i]) != 0) {
perror("pthread_create");
return 1;
}
}
/* 收集：join 的第二個參數接住 worker 的回傳指標 */
long total = 0;
for (int i = 0; i < NTHREADS; i++) {
void *ret;
pthread_join(tid[i], &ret);
if (ret == NULL) {
fprintf(stderr, "worker %d failed\n", i);
return 1;
}
total += *(long *)ret;
free(ret); /* 誰 malloc 誰負責安排釋放：
worker 配置、main 用完釋放 */
}
printf("total = %ld\n", total);
printf("expected = %ld (N*(N+1)/2)\n", expect);
printf("%s\n", total == expect ? "MATCH" : "MISMATCH?!");
return total == expect ? 0 : 1;
}程式碼講解
sum_worker開頭就 malloc——失敗時回傳 NULL， main 檢查後放棄；把「配置失敗」也走執行緒的回傳通道。- 部分和先累加在區域變數
sum、最後才寫進*result： 迴圈裡直接寫*result += ...也對，但區域變數讓編譯器 能把它放進暫存器（快很多）。 - main 的
free(ret)：配置在 worker、釋放在 main—— 「誰 malloc 誰負責安排釋放」，跨執行緒交接時 所有權要在註解裡寫清楚。 - 部分和輸出順序可能亂（worker 各自印），總和不變。
- 易錯處：join 的第二個參數型別是
void **—— 傳&ret（ret 是void *），再 cast 成long *解讀。
範例輸出
summing 1..1000000 with 4 threads:
worker [ 0, 250000) partial sum = 31250125000
worker [ 250000, 500000) partial sum = 93750125000
worker [ 500000, 750000) partial sum = 156250125000
worker [ 750000, 1000000) partial sum = 218750125000
total = 500000500000
expected = 500000500000 (N*(N+1)/2)
MATCH（worker 行的順序可能不同，total 恆為 500000500000。）
練習 3：trylock 搶鎖統計（難度：進階）
題目
4 個執行緒搶同一把 mutex：用 pthread_mutex_trylock，
拿到就在臨界區做一單位的工作，拿不到就記一次失敗並讓出 CPU。
每人成功 50000 次收工。統計每人的成功/失敗次數，
驗證共享計數器恰為 200000。
解題思路
- lock vs trylock 的本質差異：
lock拿不到就睡 （排隊等叫號）；trylock拿不到立刻回 EBUSY—— 執行緒保有控制權，可以先去做別的事。適用場景： 「有鎖就做、沒鎖就跳過」（如背景清理）、 避免死鎖的退避策略（見練習 5）。 - 迴圈結構：
while (成功次數 < 目標)，每輪 trylock： - 回 0：進了臨界區——shared_work++、自己的成功數 ++、 解鎖； - 回 EBUSY：失敗數 ++、sched_yield()讓出 CPU 再試 ——不 yield 就是純忙等，燒 CPU 又搶不贏持鎖的人； - 其他值：真正的錯誤（EINVAL 等），要報。 - 統計自己的數字不用鎖：
st->success/st->fail是 每人一格的私有資料——再次體會「劃清私有與共享」 是並行設計的第一步。共享的只有shared_work， 它只在臨界區內被碰。 - 結果解讀：總成功數恆為 200000（驗證鎖的正確性）； 失敗數每次執行都不同（競爭取決於時序）， 臨界區越長失敗率越高——本題臨界區只有一個 ++， 失敗率通常不到 1%。
參考解答
/*
* 練習 3：trylock 搶鎖統計 — 拿不到就數一筆，不排隊
*
* 編譯：cc -std=c11 -Wall -Wextra -pthread -o /tmp/out ex3_trylock_stats.c
* 執行：/tmp/out
*
* 重點：pthread_mutex_trylock 不阻塞——拿不到立刻回 EBUSY。
* 統計每個執行緒的成功/失敗次數，親眼看到鎖競爭
* （contention）的分布；成功/失敗比反映臨界區長度。
*/
#include <stdio.h>
#include <pthread.h>
#include <errno.h>
#include <sched.h> /* sched_yield */
#define NTHREADS 4
#define GOAL 50000 /* 每個執行緒要「成功」這麼多次才收工 */
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
static long shared_work = 0; /* 鎖保護的共享資料 */
struct stats {
int no;
long success; /* trylock 拿到的次數 */
long fail; /* EBUSY 的次數 */
};
static void *fighter(void *arg)
{
struct stats *st = arg;
while (st->success < GOAL) {
int rc = pthread_mutex_trylock(&lock);
if (rc == 0) {
/* 臨界區：刻意做一點點事，製造競爭窗口 */
shared_work++;
st->success++;
pthread_mutex_unlock(&lock);
} else if (rc == EBUSY) {
st->fail++; /* 沒拿到：記一筆，讓出 CPU 再試 */
sched_yield();
} else {
fprintf(stderr, "trylock error: %d\n", rc);
break;
}
}
return NULL;
}
int main(void)
{
pthread_t tid[NTHREADS];
struct stats st[NTHREADS];
printf("%d threads fighting for one mutex, %d successes each:\n\n",
NTHREADS, GOAL);
for (int i = 0; i < NTHREADS; i++) {
st[i] = (struct stats){ .no = i };
if (pthread_create(&tid[i], NULL, fighter, &st[i]) != 0) {
perror("pthread_create");
return 1;
}
}
for (int i = 0; i < NTHREADS; i++)
pthread_join(tid[i], NULL);
long total_success = 0, total_fail = 0;
printf(" thread success fail fail%%\n");
for (int i = 0; i < NTHREADS; i++) {
double pct = st[i].fail * 100.0 / (st[i].success + st[i].fail);
printf(" %6d %8ld %8ld %5.1f%%\n",
st[i].no, st[i].success, st[i].fail, pct);
total_success += st[i].success;
total_fail += st[i].fail;
}
printf("\nshared_work = %ld (expect %d) %s\n",
shared_work, NTHREADS * GOAL,
shared_work == (long)NTHREADS * GOAL ? "MATCH" : "MISMATCH?!");
printf("total: %ld successes, %ld busy misses\n",
total_success, total_fail);
printf("\nnotes:\n"
" - fail counts differ per run: contention depends on timing\n"
" - trylock never blocks: useful for 'do something else if\n"
" busy' patterns and for deadlock avoidance (see ex5)\n");
return shared_work == (long)NTHREADS * GOAL ? 0 : 1;
}程式碼講解
trylock的回傳值判斷用rc == EBUSY而不是rc != 0： EBUSY 是「正常的失敗」，其他錯誤碼是 bug—— 兩者的處理天差地遠，不可混為一談。sched_yield()（<sched.h>）：告訴排程器「我先讓一步」 ——比空轉禮貌，比 usleep 輕量。struct stats用複合字面值初始化(struct stats){ .no = i }——其餘欄位自動歸零。- 驗證錨點
shared_work == NTHREADS * GOAL： 若把臨界區的保護拿掉（直接 ++），這個等式會破—— 可以自己實驗看看 race 的威力。 - 易錯處：成功路徑忘記 unlock——下一輪 trylock 對 非遞迴 mutex 再上鎖是 UB（或 EBUSY 假訊號）， 所有人從此全 EBUSY。
範例輸出
4 threads fighting for one mutex, 50000 successes each:
thread success fail fail%
0 50000 0 0.0%
1 50000 149 0.3%
2 50000 30 0.1%
3 50000 50 0.1%
shared_work = 200000 (expect 200000) MATCH
total: 200000 successes, 229 busy misses
notes:
- fail counts differ per run: contention depends on timing
- trylock never blocks: useful for 'do something else if
busy' patterns and for deadlock avoidance (see ex5)（fail 欄位每次執行都不同；success 與 shared_work 恆定。）
練習 4：毒藥丸關閉協定（難度：進階）
題目
為生產者-消費者系統加上「毒藥丸」（poison pill）關閉協定： 2 個生產者各生產 10 個項目後收工；主執行緒等生產者結束後， 往佇列投放「消費者數量」顆毒藥丸（特殊值 -1）； 3 個消費者各自吃到藥丸後優雅退出。驗證總消費量與 緩衝區清空。
解題思路
- 問題：消費者的迴圈是
for(;;) { v = get(); ... }—— 誰來告訴它「不會再有資料了，下班吧」？ 直接pthread_cancel太暴力（可能砍在臨界區中間， 鎖永遠不釋放）；設全域旗標 + 廣播喚醒可行但條件複雜。 毒藥丸是最優雅的方案：關閉訊號走資料本身的通道。 - 協定設計三原則：
- 藥丸值必須是正常資料不可能出現的值（本題資料
非負，用 -1）；
- 一人一顆：每個消費者吃到藥丸就退出、
不再拿新工作——所以要投放恰好 N_CONSUMERS 顆；
- 投放時機：確定不再有真資料之後（main join 完
所有生產者），藥丸走一般的
put()排在所有真資料 後面——FIFO 保證消費者把真資料吃完才碰到藥丸。 - 有界緩衝區本體是教科書寫法：mutex + 兩個條件變數
（
not_full、not_empty），等待一律用while重驗 ——假喚醒與「醒來後條件又被搶走」都靠它擋。 - 驗收：三個消費者的處理量相加 = 20（總生產量）、
count == 0（藥丸也吃完了）——分布（11/9/0）隨排程 而異，總量不變。 - 邊界思考：藥丸數多於消費者 → 剩藥丸卡在佇列 （無害但不乾淨）；少於 → 有消費者永遠等不到下班通知 ——「恰好 N 顆」是協定的一部分。
參考解答
/*
* 練習 4：毒藥丸關閉協定 — 生產者-消費者的優雅退場
*
* 編譯：cc -std=c11 -Wall -Wextra -pthread -o /tmp/out ex4_poison_pill.c
* 執行：/tmp/out
*
* 重點：消費者在 while(1) 裡等資料，誰來叫它下班？
* 毒藥丸（poison pill）＝佇列裡的特殊值，
* 每個消費者吃到一顆就退出——生產者收工後，
* 主執行緒投放「消費者數量」顆藥丸。
*/
#include <stdio.h>
#include <pthread.h>
#define BUF_SIZE 8
#define N_PRODUCERS 2
#define N_CONSUMERS 3
#define ITEMS_EACH 10 /* 每個生產者生產的項目數 */
#define POISON (-1) /* 藥丸：正常資料永遠不會是負數 */
/* ---- 有界緩衝區（環形陣列 + mutex + 兩個條件變數） ---- */
static int buf[BUF_SIZE];
static int count = 0, head = 0, tail = 0;
static pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t not_full = PTHREAD_COND_INITIALIZER;
static pthread_cond_t not_empty = PTHREAD_COND_INITIALIZER;
static void put(int v)
{
pthread_mutex_lock(&m);
while (count == BUF_SIZE) /* 滿了：等消費者挪出空位 */
pthread_cond_wait(¬_full, &m); /* while 不是 if！ */
buf[head] = v;
head = (head + 1) % BUF_SIZE;
count++;
pthread_cond_signal(¬_empty); /* 叫醒一個等資料的人 */
pthread_mutex_unlock(&m);
}
static int get(void)
{
pthread_mutex_lock(&m);
while (count == 0) /* 空的：等生產者放資料 */
pthread_cond_wait(¬_empty, &m);
int v = buf[tail];
tail = (tail + 1) % BUF_SIZE;
count--;
pthread_cond_signal(¬_full);
pthread_mutex_unlock(&m);
return v;
}
/* ---- 生產者：做完自己的量就收工（不投藥，投藥是 main 的事） ---- */
static void *producer(void *arg)
{
int no = *(int *)arg;
for (int i = 0; i < ITEMS_EACH; i++) {
int item = no * 100 + i; /* 100,101.. / 200,201.. */
put(item);
}
printf(" producer %d: done (%d items)\n", no, ITEMS_EACH);
return NULL;
}
/* ---- 消費者：吃到藥丸就下班 ---- */
static void *consumer(void *arg)
{
int no = *(int *)arg;
long taken = 0;
for (;;) {
int v = get();
if (v == POISON) { /* 藥丸：不再拿新工作 */
printf(" consumer %d: got poison pill after %ld items, "
"exiting\n", no, taken);
return (void *)taken;
}
taken++; /* 正常資料：處理它 */
}
}
int main(void)
{
pthread_t prod[N_PRODUCERS], cons[N_CONSUMERS];
int pno[N_PRODUCERS], cno[N_CONSUMERS];
printf("%d producers x %d items, %d consumers, buffer %d:\n\n",
N_PRODUCERS, ITEMS_EACH, N_CONSUMERS, BUF_SIZE);
for (int i = 0; i < N_CONSUMERS; i++) {
cno[i] = i + 1;
pthread_create(&cons[i], NULL, consumer, &cno[i]);
}
for (int i = 0; i < N_PRODUCERS; i++) {
pno[i] = i + 1;
pthread_create(&prod[i], NULL, producer, &pno[i]);
}
/* 1. 等所有生產者收工——確定不會再有新資料 */
for (int i = 0; i < N_PRODUCERS; i++)
pthread_join(prod[i], NULL);
/* 2. 投放 N_CONSUMERS 顆藥丸：每個消費者恰好吃到一顆。
藥丸走一般的 put()——排在所有真資料後面，
保證消費者把真資料吃完才會碰到藥丸。 */
printf(" main: producers done, dropping %d poison pills\n",
N_CONSUMERS);
for (int i = 0; i < N_CONSUMERS; i++)
put(POISON);
/* 3. 等所有消費者下班，順便收集各自的處理量 */
long total = 0;
for (int i = 0; i < N_CONSUMERS; i++) {
void *ret;
pthread_join(cons[i], &ret);
total += (long)ret;
}
printf("\ntotal consumed = %ld (expect %d) %s\n",
total, N_PRODUCERS * ITEMS_EACH,
total == N_PRODUCERS * ITEMS_EACH ? "MATCH" : "MISMATCH?!");
printf("buffer left = %d (expect 0)\n", count);
return (total == N_PRODUCERS * ITEMS_EACH && count == 0) ? 0 : 1;
}程式碼講解
put/get的條件變數用法逐行都是重點：while (count == BUF_SIZE) pthread_cond_wait(¬_full, &m);——wait 原子地「放鎖＋睡」，醒來自動重新持鎖， 然後 while 再驗一次條件。- 生產者不投藥：它不知道還有幾個同事沒收工—— 「誰有全局視野誰投藥」，本題是 main（join 過所有生產者）。
- 消費者的回傳值用
(void *)taken把處理量帶回 main ——順手複習練習 2 的回傳通道（這次不用 malloc， 整數塞指標——僅限「值不是位址」的私下約定用法）。 - 輸出中 consumer 2 吃到 0 個項目就退場——排程器沒給它 機會搶到真資料，這種不均是正常現象； 總量 MATCH 才是正確性的判準。
- 易錯處：把藥丸投在 join 生產者之前——藥丸可能 排在真資料前面，消費者提早下班，真資料爛在佇列裡。
範例輸出
2 producers x 10 items, 3 consumers, buffer 8:
producer 1: done (10 items)
producer 2: done (10 items)
main: producers done, dropping 3 poison pills
consumer 2: got poison pill after 0 items, exiting
consumer 3: got poison pill after 9 items, exiting
consumer 1: got poison pill after 11 items, exiting
total consumed = 20 (expect 20) MATCH
buffer left = 0 (expect 0)（各消費者的分配量隨排程而異，總量恆為 20。）
練習 5：親手做出死鎖，再修好它（難度：挑戰）
題目
用兩把鎖、相反的拿鎖順序做出一個確定會發生的死鎖； 主執行緒當監工，偵測到卡死後印出四要件分析； 然後跑「固定全域鎖序」的修正版證明它能跑完。 程式整體必須正常結束（exit 0），不得真的掛死。
解題思路
- 死鎖的配方（要重現就要湊齊四要件）：
thread 1: lock(A) --sleep--> lock(B) 持 A 等 B
thread 2: lock(B) --sleep--> lock(A) 持 B 等 A ← 環！中間的 usleep(100ms) 是關鍵：確保兩人都拿到第一把鎖
之後才去搶第二把——不睡的話，快的那條可能一口氣拿完
兩把，死鎖變成機率事件。教學程式要的是必然重現。
2. 怎麼安全地展示「掛死」：死鎖的執行緒叫不回來
（join 會跟著卡死）。監工模式：worker 完成時對
atomic_int done_count +1；main 睡 1 秒後看計數——
還是 0 就宣告死鎖成立，不 join、直接進第二幕，
最後 exit(0) 讓作業系統收走一切。
死鎖的執行緒停在 pthread_mutex_lock 裡毫無副作用，
放著不管是安全的。
3. lldb 觀察法（題目要求的除錯技能）：把偵測拿掉讓
程式真的卡住，另開終端 lldb -p <pid>，
thread backtrace all——兩條執行緒都停在
__psynch_mutexwait（macOS 的 mutex 等待 syscall），
各自的呼叫堆疊顯示持有的鎖與等待的鎖，互相指向對方。
4. 修法：全域鎖序。四要件中最好拆的是「循環等待」——
規定所有執行緒按同一順序拿鎖（先 C 後 D），
等待圖就不可能出現環：等 D 的人必然已持 C，
而持 D 的人絕不會反過來等 C。
5. 其他修法（README 提到的）：trylock + 退避
（拿不到第二把就放掉第一把重來——犧牲「持有並等待」）、
一次拿齊所有鎖。全域鎖序是工程上最常用的。
參考解答
/*
* 練習 5：親手做出死鎖，再用「固定鎖序」修好它
*
* 編譯：cc -std=c11 -Wall -Wextra -pthread -o /tmp/out ex5_deadlock.c
* 執行：/tmp/out
*
* 重點：兩把鎖、相反順序 = 死鎖四要件湊齊。
* 本程式先「安全地」重現死鎖（主執行緒當監工，
* 發現卡死就宣告並放棄那兩條執行緒），
* 再跑固定鎖序的修正版證明它能跑完。
*
* 觀察工具：程式卡住時（把 DEMO_DEADLOCK 的偵測拿掉的話）
* 可用 `lldb -p <pid>` attach，`thread backtrace all` 會看到
* 兩條執行緒都停在 __psynch_mutexwait —— 互相等待的鐵證。
*/
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h> /* usleep */
#include <pthread.h>
#include <stdatomic.h>
static pthread_mutex_t lock_a = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t lock_b = PTHREAD_MUTEX_INITIALIZER;
static atomic_int done_count = 0; /* 監工用：完成的執行緒數 */
/* ---------- 壞版本：兩條執行緒以相反順序拿鎖 ---------- */
static void *bad_thread_1(void *arg)
{
(void)arg;
pthread_mutex_lock(&lock_a); /* 1. 拿 A */
printf(" bad thread 1: got A, now wants B...\n");
usleep(100 * 1000); /* 讓對方也拿到第一把鎖 */
pthread_mutex_lock(&lock_b); /* 2. 等 B —— 永遠等不到 */
/* 死鎖成立時永遠到不了這裡 */
pthread_mutex_unlock(&lock_b);
pthread_mutex_unlock(&lock_a);
atomic_fetch_add(&done_count, 1);
return NULL;
}
static void *bad_thread_2(void *arg)
{
(void)arg;
pthread_mutex_lock(&lock_b); /* 1. 拿 B（順序相反！） */
printf(" bad thread 2: got B, now wants A...\n");
usleep(100 * 1000);
pthread_mutex_lock(&lock_a); /* 2. 等 A —— 永遠等不到 */
pthread_mutex_unlock(&lock_a);
pthread_mutex_unlock(&lock_b);
atomic_fetch_add(&done_count, 1);
return NULL;
}
/* ---------- 好版本：全域鎖序「永遠先 A 後 B」 ---------- */
static pthread_mutex_t lock_c = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t lock_d = PTHREAD_MUTEX_INITIALIZER;
static void *good_thread(void *arg)
{
int no = *(int *)arg;
for (int i = 0; i < 3; i++) {
pthread_mutex_lock(&lock_c); /* 所有人都先 C */
pthread_mutex_lock(&lock_d); /* 再 D —— 循環等待不可能 */
printf(" good thread %d: holding C+D (round %d)\n", no, i + 1);
pthread_mutex_unlock(&lock_d); /* 釋放順序相反（好習慣） */
pthread_mutex_unlock(&lock_c);
usleep(10 * 1000);
}
return NULL;
}
int main(void)
{
/* ===== 第一幕：重現死鎖 ===== */
printf("=== act 1: opposite lock order (A->B vs B->A) ===\n");
pthread_t t1, t2;
pthread_create(&t1, NULL, bad_thread_1, NULL);
pthread_create(&t2, NULL, bad_thread_2, NULL);
/* 監工：等 1 秒，看有沒有人完成 */
usleep(1000 * 1000);
int finished = atomic_load(&done_count);
if (finished == 0) {
printf(" main: 1 second passed, 0 of 2 threads finished\n");
printf(" main: DEADLOCK confirmed --\n");
printf(" thread 1 holds A, waits for B\n");
printf(" thread 2 holds B, waits for A\n");
printf(" (all four conditions met: mutual exclusion,\n");
printf(" hold-and-wait, no preemption, circular wait)\n");
/* 死鎖的執行緒無法回收——不 join（joins 會跟著卡死）。
它們會在行程結束時被作業系統一併收走。 */
} else {
/* 理論上極不可能（兩邊都睡了 100ms 才拿第二把鎖） */
printf(" main: threads finished?! scheduler was very kind\n");
pthread_join(t1, NULL);
pthread_join(t2, NULL);
}
/* ===== 第二幕：固定鎖序的修正版 ===== */
printf("\n=== act 2: fixed global lock order (everyone C then D) ===\n");
pthread_t g1, g2;
int no1 = 1, no2 = 2;
pthread_create(&g1, NULL, good_thread, &no1);
pthread_create(&g2, NULL, good_thread, &no2);
pthread_join(g1, NULL);
pthread_join(g2, NULL);
printf(" main: both good threads joined -- no deadlock possible\n");
printf("\nlesson: a single rule ('always lock in the same global\n"
"order') removes the circular-wait condition entirely.\n");
/*
* exit 而非 return：main return 會做正常的行程收尾，
* 但第一幕那兩條死鎖執行緒還掛著——exit 直接結束行程，
* 作業系統回收一切。這裡兩者效果相同，用 exit 表明意圖。
*/
exit(0);
}程式碼講解
done_count用atomic_int：監工與 worker 之間的 「完成了沒」訊號本身也是共享資料——用原子變數避免 在死鎖示範裡再引入一個 race（多諷刺）。- 第一幕結束不 join t1/t2 並註明原因——join 死鎖執行緒 = main 加入死者行列。這是「明知資源沒回收仍然正確」 的少數合法場景：行程結束時 OS 全部收走。
- 第二幕的 good_thread 跑三輪、兩條共六次「holding C+D」 全部順利印出——同樣的兩把鎖、同樣的兩條執行緒， 只差「順序統一」。
- 解鎖順序 D→C 與上鎖相反：非必要（解鎖不會死鎖） 但是好習慣——巢狀結構清晰，也方便換成 RAII 風格的包裝。
exit(0)而非return 0：兩者此處效果相同， exit 明示「我知道還有執行緒掛著，就是要整個行程收攤」。- 易錯處：用
sleep等死鎖執行緒「自己好起來」—— 死鎖不會自癒，等多久都一樣；監工要的是「確認卡死」 而不是「等待恢復」。
範例輸出
=== act 1: opposite lock order (A->B vs B->A) ===
bad thread 1: got A, now wants B...
bad thread 2: got B, now wants A...
main: 1 second passed, 0 of 2 threads finished
main: DEADLOCK confirmed --
thread 1 holds A, waits for B
thread 2 holds B, waits for A
(all four conditions met: mutual exclusion,
hold-and-wait, no preemption, circular wait)
=== act 2: fixed global lock order (everyone C then D) ===
good thread 1: holding C+D (round 1)
good thread 2: holding C+D (round 1)
good thread 1: holding C+D (round 2)
good thread 2: holding C+D (round 2)
good thread 2: holding C+D (round 3)
good thread 1: holding C+D (round 3)
main: both good threads joined -- no deadlock possible
lesson: a single rule ('always lock in the same global
order') removes the circular-wait condition entirely.（第二幕的輪次交錯順序每次不同。）
練習 6：SPSC 無鎖環形緩衝區（難度：挑戰）
題目
用 stdatomic.h 實作單生產者單消費者（SPSC）的無鎖環形
緩衝區（acquire/release 記憶體序），傳輸 200 萬個項目；
再實作一把大鎖的 mutex 版本，比較吞吐量；
兩版都用 checksum 驗證資料無遺漏、無重複。
解題思路
- 為什麼 SPSC 可以不用鎖——關鍵觀察：
-
head（寫入位置）只有生產者寫； -tail（讀取位置）只有消費者寫； 每個變數都只有一個寫者！剩下的問題只是「對方什麼時候 看見我的寫入」——這是記憶體序（memory ordering） 的問題，不是互斥的問題。 - release/publish 模式——無鎖程式設計的第一課：
生產者 消費者
1. buf[h % N] = v ┐
2. store(head, h+1, │ release 保證 1 在 2 之前
release) ┘ 對「acquire 讀到 h+1 的人」可見
1. load(head, acquire) 看到 h+1
2. 讀 buf[t % N] ← 保證看到 v沒有 release/acquire 配對，編譯器與 CPU 都可以重排——
消費者可能看到「index 前進了、資料還是舊的」。
3. 滿/空的判斷：head/tail 用不繞回的累計計數
（unsigned long 自然溢位也安全），
滿 = head - tail >= N、空 = head == tail；
索引用 & (N-1)（N 是 2 的冪）。
計數不繞回讓滿/空判斷不需要犧牲一格。
4. mutex 版對照組：同一個 ring，每次 push/pop 都
lock→檢查→操作→unlock，滿/空時放鎖 yield 重試。
兩版跑同樣的 200 萬項目，計時比較。
5. honest benchmark：吞吐量數字每次執行都不同
（機器負載、排程），本機上無鎖版約快 1.3~2 倍；
重點不是倍數而是checksum 兩版都 OK——
正確性先於效能。
6. 界限要講清楚：這個設計只適用單生產者單消費者。
兩個生產者同時 push 就會互踩 head——多生產者/多消費者
需要 CAS 迴圈或回到鎖。
參考解答
/*
* 練習 6：SPSC 無鎖環形緩衝區 — stdatomic vs mutex 吞吐量
*
* 編譯：cc -std=c11 -Wall -Wextra -pthread -o /tmp/out ex6_spsc_ring.c
* 執行：/tmp/out
*
* 重點：單生產者單消費者（SPSC）佇列可以完全不用鎖——
* head 只有生產者寫、tail 只有消費者寫，
* 靠 acquire/release 記憶體序讓「資料寫入」先於
* 「index 發布」被對方看見。
*/
#include <stdio.h>
#include <pthread.h>
#include <stdatomic.h>
#include <sched.h>
#include <time.h>
#define RING_SIZE 1024 /* 2 的冪：% 換成 & (SIZE-1) */
#define N_ITEMS 2000000L
/* ---------------- 無鎖版：SPSC ring ---------------- */
static int lf_buf[RING_SIZE];
static atomic_ulong lf_head = 0; /* 只有生產者寫 */
static atomic_ulong lf_tail = 0; /* 只有消費者寫 */
static void lf_push(int v)
{
unsigned long h = atomic_load_explicit(&lf_head, memory_order_relaxed);
/* 滿了（head 追上 tail + SIZE）：等消費者前進 */
while (h - atomic_load_explicit(&lf_tail, memory_order_acquire)
>= RING_SIZE)
sched_yield();
lf_buf[h & (RING_SIZE - 1)] = v; /* 1. 先寫資料 */
/* 2. 再發布 index：release 保證上面的寫入先被看見 */
atomic_store_explicit(&lf_head, h + 1, memory_order_release);
}
static int lf_pop(void)
{
unsigned long t = atomic_load_explicit(&lf_tail, memory_order_relaxed);
/* 空的：等生產者發布新資料 */
while (atomic_load_explicit(&lf_head, memory_order_acquire) == t)
sched_yield();
int v = lf_buf[t & (RING_SIZE - 1)]; /* 1. 先讀資料 */
atomic_store_explicit(&lf_tail, t + 1, memory_order_release);
return v;
}
/* ---------------- mutex 版：同樣的 ring，一把鎖包全部 ---------------- */
static int mx_buf[RING_SIZE];
static unsigned long mx_head = 0, mx_tail = 0;
static pthread_mutex_t mx_lock = PTHREAD_MUTEX_INITIALIZER;
static void mx_push(int v)
{
for (;;) {
pthread_mutex_lock(&mx_lock);
if (mx_head - mx_tail < RING_SIZE) {
mx_buf[mx_head & (RING_SIZE - 1)] = v;
mx_head++;
pthread_mutex_unlock(&mx_lock);
return;
}
pthread_mutex_unlock(&mx_lock); /* 滿：放鎖重試 */
sched_yield();
}
}
static int mx_pop(void)
{
for (;;) {
pthread_mutex_lock(&mx_lock);
if (mx_head != mx_tail) {
int v = mx_buf[mx_tail & (RING_SIZE - 1)];
mx_tail++;
pthread_mutex_unlock(&mx_lock);
return v;
}
pthread_mutex_unlock(&mx_lock); /* 空：放鎖重試 */
sched_yield();
}
}
/* ---------------- 共用的測試骨架 ---------------- */
struct api {
void (*push)(int);
int (*pop)(void);
};
static void *producer(void *arg)
{
const struct api *a = arg;
for (long i = 0; i < N_ITEMS; i++)
a->push((int)(i & 0x7fffffff));
return NULL;
}
static void *consumer(void *arg)
{
const struct api *a = arg;
long checksum = 0;
for (long i = 0; i < N_ITEMS; i++)
checksum += a->pop();
return (void *)checksum;
}
static double run_bench(const char *name, const struct api *a,
long *checksum_out)
{
struct timespec t0, t1;
pthread_t p, c;
clock_gettime(CLOCK_MONOTONIC, &t0);
pthread_create(&p, NULL, producer, (void *)a);
pthread_create(&c, NULL, consumer, (void *)a);
pthread_join(p, NULL);
void *ret;
pthread_join(c, &ret);
clock_gettime(CLOCK_MONOTONIC, &t1);
*checksum_out = (long)ret;
double sec = (double)(t1.tv_sec - t0.tv_sec) +
(double)(t1.tv_nsec - t0.tv_nsec) / 1e9;
printf(" %-10s %ld items in %.3f s (%.1f M items/s)\n",
name, N_ITEMS, sec, N_ITEMS / sec / 1e6);
return sec;
}
int main(void)
{
/* 正確性對帳：0..N-1 的和 */
long expect = 0;
for (long i = 0; i < N_ITEMS; i++)
expect += (int)(i & 0x7fffffff);
printf("SPSC ring buffer, %ld items, ring size %d:\n\n",
N_ITEMS, RING_SIZE);
struct api lockfree = { lf_push, lf_pop };
struct api mutexed = { mx_push, mx_pop };
long sum_lf, sum_mx;
double t_lf = run_bench("lock-free", &lockfree, &sum_lf);
double t_mx = run_bench("mutex", &mutexed, &sum_mx);
printf("\nchecksums: lock-free %s, mutex %s\n",
sum_lf == expect ? "OK" : "BAD!",
sum_mx == expect ? "OK" : "BAD!");
printf("speedup: mutex/lock-free = %.1fx\n", t_mx / t_lf);
printf("\nwhy it works without locks:\n"
" - head is written ONLY by the producer,\n"
" tail ONLY by the consumer (single writer each)\n"
" - release on publish / acquire on observe creates the\n"
" happens-before edge: data write < index publish\n"
" - two threads max. for MPMC you need CAS loops or locks\n");
return (sum_lf == expect && sum_mx == expect) ? 0 : 1;
}程式碼講解
lf_push讀自己的head用 relaxed（自己寫的值自己 當然看得到）、讀對方的tail用 acquire（要看見對方 release 之前的效果）——每個原子操作的記憶體序都有 理由，不是越強越好（seq_cst 全上會白白慢）。- 等待用
sched_yield()自旋：SPSC 佇列通常搭配 「幾乎不會滿/空」的工作負載；會長時間等的場景 應改用條件變數（回到鎖的世界）。 struct api { push, pop }函式指標表：同一個 producer/consumer 骨架跑兩種實作——函式指標單元的 手法在測試框架裡的實戰。clock_gettime(CLOCK_MONOTONIC)：量牆鐘時間用 MONOTONIC（不受系統時間調整影響）。- checksum 對帳：
0..N-1的和，兩版必須相等且等於預期 ——任何 index 錯誤（少讀、重讀）都會讓和對不上。 - 易錯處一：資料寫入與 index 發布都用 relaxed—— 在 x86 上「碰巧」對（強記憶體模型），搬到 ARM（Apple Silicon！）就是真 bug——本檔在 ARM Mac 上實測， acquire/release 是必要的。
- 易錯處二：ring 大小不是 2 的冪還用
& (N-1)—— 索引直接錯亂；要嘛強制 2 的冪，要嘛用%。
範例輸出
SPSC ring buffer, 2000000 items, ring size 1024:
lock-free 2000000 items in 0.063 s (31.9 M items/s)
mutex 2000000 items in 0.080 s (25.0 M items/s)
checksums: lock-free OK, mutex OK
speedup: mutex/lock-free = 1.3x
why it works without locks:
- head is written ONLY by the producer,
tail ONLY by the consumer (single writer each)
- release on publish / acquire on observe creates the
happens-before edge: data write < index publish
- two threads max. for MPMC you need CAS loops or locks（計時與倍數每次執行不同；checksums 兩行恆為 OK。）
常見錯誤解法對照
| 錯誤寫法 | 症狀 | 正解 |
|---|---|---|
pthread_create(..., &i) 傳迴圈變數 |
編號重複/亂跳 | 參數陣列一人一格（練習 1） |
return &local; 回傳堆疊位址 |
join 拿到懸空指標 | malloc 或寫進參數結構（練習 2） |
trylock 把 EBUSY 當錯誤處理 |
正常競爭被誤報 | EBUSY 與其他錯誤碼分開（練習 3） |
cond_wait 用 if 不用 while |
假喚醒後帶著壞條件前進 | while 重驗（練習 4） |
| 藥丸在生產者收工前投放 | 消費者提早下班、資料爛在佇列 | join 生產者後再投（練習 4） |
| 兩把鎖各處順序不一 | 偶發性整組卡死 | 全域鎖序（練習 5） |
| 無鎖佇列全用 relaxed | x86 碰巧對、ARM 上資料撕裂 | release 發布 / acquire 觀察（練習 6） |