同步

发表于 更新于 Waline： 阅读次数： 本文字数： 11k 阅读时长 ≈ 41 分钟

并发的挑战

并发程序的核心挑战在于其执行顺序的不确定性。即使是很简单的代码片段，在多线程环境下，由于线程调度和指令交错执行，可能产生多种不同的最终状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 共享变量
int x = 0, y = 0;

// 线程 T1
void T1() {
    x = 1;
    int t = y;
    // ... 使用 t ...
}

// 线程 T2
void T2() {
    y = 1;
    int t = x;
    // ... 使用 t ...
}

这种不确定性使得我们难以枚举和预测所有可能的状态，尤其是当线程数量增多时。

为何需要同步

并发程序中，并非所有可能的状态都是我们期望的。

  • 不期望的状态：例如，多个线程同时进入临界区，破坏数据一致性。
  • 期望的状态：例如，主线程调用 pthread_join 等待子线程结束后才能继续执行。

为了避免不期望的状态，并确保程序能达到期望的状态，我们需要对线程的执行进行控制。这种控制通常要求线程：

  1. 不能太快：在某个条件满足之前必须等待。例如，线程 B 必须等待线程 A 完成某项操作后才能继续。
  2. 不能太慢：一旦某个条件满足，等待的线程应该能够继续执行（避免死锁或饥饿）。

同步

同步（Synchronization）是指控制并发进程（或线程）的执行，使得它们在变化过程中保持一定的相对关系，以协调它们对共享资源的访问或保证特定的执行顺序。

互斥也是同步的一种特殊形式，它保证了在任意时刻只有一个线程能访问临界区。但同步的概念更广泛，还包括了执行顺序的协调。

简单的互斥锁（如 lock()/unlock()）虽然能保证临界区内的原子性，但无法直接控制线程间的执行顺序。

经典同步问题：生产者-消费者

生产者-消费者问题（Producer-Consumer Problem）是并发领域一个非常经典且具有代表性的同步问题，由 Edsger Dijkstra 首次提出。

  • 场景：一个或多个生产者线程和一个或多个消费者线程共享一个固定大小的有界缓冲区。
  • 生产者任务：生成数据项，如果缓冲区未满，则将数据放入缓冲区；如果缓冲区已满，则必须等待。
  • 消费者任务：从缓冲区取走数据项，如果缓冲区非空，则取出数据进行消费；如果缓冲区为空，则必须等待。

graph LR
    P(生产者) -- 生产数据 --> B{有界缓冲区};
    B -- 消费数据 --> C(消费者);

    subgraph 约束条件
        direction LR
        P -- 缓冲区满？ --> W1[等待];
        C -- 缓冲区空？ --> W2[等待];
    end

    style P fill:#f9f,stroke:#333,stroke-width:2px
    style C fill:#9cf,stroke:#333,stroke-width:2px
    style B fill:#ccf,stroke:#333,stroke-width:2px
    style W1 fill:#f99,stroke:#333,stroke-width:1px
    style W2 fill:#f99,stroke:#333,stroke-width:1px

简化模型：打印括号

为了更清晰地理解同步机制，我们将问题简化为打印括号：

  • 生产者：打印左括号 (，相当于向缓冲区放入一个元素。
  • 消费者：打印右括号 )，相当于从缓冲区取出一个元素。
  • 缓冲区：用括号的嵌套深度（depth）来表示，容量为 n。
  • 合法序列：任何时刻，括号深度 0 <= depth <= n。
1
2
void produce() { printf("("); }
void consume() { printf(")"); }

非法序列示例（假设缓冲区容量为 3）

  • (()))：非法，消费时 depth 变为 -1（缓冲区为空无法消费）。
  • (((())))：非法，生产第四个 ( 时 depth 变为 4（超出缓冲区容量）。

同步目标

我们需要确保：

  • 生产者在 depth < n 时才能打印 (。
  • 消费者在 depth > 0 时才能打印 )。

这本质上是要求线程在特定条件满足时才能执行，即实现 wait_until 语义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 伪代码
void produce() {
    wait_until(depth < n) {
        printf("(");
        // depth++; // 更新状态
    }
}

void consume() {
    wait_until(depth > 0) {
        printf(")");
        // depth--; // 更新状态
    }
}

问题在于如何高效且正确地实现 wait_until。

尝试与陷阱

  1. 无锁尝试（错误）

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    // 错误：竞态条件
    int n; // 容量
    int depth = 0; // 当前深度

    void produce() {
        while(1) {
            retry:
            // 检查和修改 depth 不是原子操作
            int ready = (depth < n);
            if (!ready) goto retry; // 自旋等待
            printf("(");
            depth++; // 可能与其他线程冲突
        }
    }
    // consume 类似
    • 问题：对共享变量 depth 的读写存在竞态条件，多个线程可能同时读到旧值并进行修改，导致 depth 值不正确，违背安全性。

  2. 加锁尝试 1（错误）

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    // 错误：条件检查与操作分离
    mutex_t lk;
    int n, depth = 0;

    void produce() {
        while(1) {
            retry:
            mutex_lock(&lk);
            int ready = (depth < n);
            mutex_unlock(&lk); // 释放锁

            // 在 unlock 和 if 之间，depth 可能已被其他线程修改！
            if (!ready) goto retry; // 条件可能已失效

            mutex_lock(&lk);
            printf("(");
            depth++;
            mutex_unlock(&lk);
        }
    }
    // consume 类似
    • 问题：在检查条件（depth < n）和根据条件执行操作（printf, depth++）之间释放了锁。这期间，其他线程可能修改 depth，使得之前检查的条件失效。

  3. 加锁尝试 2（正确但低效）

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    // 正确但低效：忙等待
    mutex_t lk;
    int n, depth = 0;

    void produce() {
        while(1) {
            retry:
            mutex_lock(&lk);
            int ready = (depth < n);
            if (!ready) {
                mutex_unlock(&lk);
                goto retry; // 持有锁时条件不满足，释放锁并重试 (忙等待)
            }
            // 条件满足，持有锁执行操作
            printf("(");
            depth++;
            mutex_unlock(&lk);
        }
    }
    // consume 类似
    • 问题：虽然保证了正确性（检查和操作都在锁的保护下），但在条件不满足时，线程会不断地 lock -> check -> unlock -> retry，形成忙等待或自旋，浪费 CPU 资源。

  4. 阻塞尝试（潜在问题：唤醒丢失）
    为了避免忙等待，我们希望线程在条件不满足时阻塞，让出 CPU，并在条件可能满足时被唤醒。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    // 潜在问题：唤醒丢失
    mutex_t lk;
    int n, depth = 0;
    queue_t producer_waiting_list, consumer_waiting_list;

    void produce() {
        while(1) {
            retry:
            mutex_lock(&lk);
            int ready = (depth < n);
            if (!ready) {
                mutex_unlock(&lk);
                wait(&producer_waiting_list); // 阻塞生产者
                goto retry;
            }
            printf("(");
            depth++;
            // 缓冲区从空变为非空，可能需要唤醒消费者
            wakeup(&consumer_waiting_list);
            mutex_unlock(&lk);
        }
    }

    void consume() {
        while(1) {
            retry:
            mutex_lock(&lk);
            int ready = (depth > 0);
            if (!ready) {
                mutex_unlock(&lk);
                wait(&consumer_waiting_list); // 阻塞消费者
                goto retry;
            }
            printf(")");
            depth--;
            // 缓冲区从满变为未满，可能需要唤醒生产者
            wakeup(&producer_waiting_list);
            mutex_unlock(&lk);
        }
    }
    • 旧有的问题：唤醒丢失。考虑生产者 P 和消费者 C：
      1. C 检查 depth > 0，发现为 false。
      2. 在 C 调用 wait 之前，发生上下文切换。
      3. P 执行，增加 depth，调用 wakeup(&consumer_waiting_list)。但此时 C 还没睡，唤醒信号丢失。
      4. C 恢复执行，调用 wait 进入睡眠，可能永远不会被唤醒。
    sequenceDiagram
    participant C as 消费者
    participant OS as 操作系统
    participant P as 生产者

    C->>C: mutex_lock()
    C->>C: check depth (== 0) -> ready = false
    Note right of C: 在 wait() 前发生上下文切换
    P->>P: mutex_lock()
    P->>P: 生产，depth++
    P->>OS: wakeup(consumer_list)
    Note left of OS: 无消费者在等待，唤醒丢失
    P->>P: mutex_unlock()
    Note left of C: 上下文切换回消费者
    C->>OS: wait(consumer_list)
    Note right of C: 沉睡，可能永远不会被唤醒
    OS-->>C: Blocked

    为了解决唤醒丢失，需要一种机制能原子地完成「释放锁并进入睡眠」的操作。这引出了条件变量。

条件变量（Condition Variable）

条件变量（Condition Variable, CV）是一种同步原语，它允许线程在某个条件不满足时原子地阻塞，并等待该条件被其他线程改变后发出信号而被唤醒。条件变量总是与一个互斥锁关联使用，以保护共享状态（即「条件」）。

条件变量

条件变量本身不存储条件的状态（它没有记忆），它只提供阻塞和唤醒的机制。条件的判断需要程序员显式地在代码中进行。

核心操作

  1. cond_wait(cond_t *cv, mutex_t *lk)
    • 原子操作：
      1. 释放传入的互斥锁 lk。
      2. 将当前线程加入条件变量 cv 的等待队列，并使其阻塞。
    • 唤醒后：当线程被唤醒时，cond_wait 会在返回前重新获取互斥锁 lk。
    • 前提：调用 cond_wait 时，必须已经持有互斥锁 lk。
  2. cond_signal(cond_t *cv)
    • 唤醒至少一个（通常是一个）正在 cond_wait(cv, ...) 上等待的线程。
    • 如果没有线程在等待 cv，则此操作无效（信号不会被「保存」）。
    • 建议：调用 cond_signal 时通常也应持有与 cv 关联的互斥锁 lk，以确保状态修改和信号发送之间的一致性。
  3. cond_broadcast(cond_t *cv)
    • 唤醒所有正在 cond_wait(cv, ...) 上等待的线程。
    • 同样，建议在持有锁时调用。

POSIX API

POSIX 线程库 (pthread) 提供了标准的条件变量 API：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#include <pthread.h>

// 声明和初始化
pthread_cond_t cv = PTHREAD_COND_INITIALIZER;
// 或者动态初始化
int pthread_cond_init(pthread_cond_t *restrict cv,
                      const pthread_condattr_t *restrict attr);
int pthread_cond_destroy(pthread_cond_t *cv);

// 等待
int pthread_cond_wait(pthread_cond_t *restrict cv,
                      pthread_mutex_t *restrict mutex);

// 唤醒单个
int pthread_cond_signal(pthread_cond_t *restrict cv);

// 唤醒所有
int pthread_cond_broadcast(pthread_cond_t *restrict cv);

基于 Futex 的简易实现

条件变量可以通过 Linux 的 futex 系统调用实现，以避免唤醒丢失。futex 允许在用户空间进行快速检查，仅在需要阻塞或唤醒时才陷入内核。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
// 简化的概念性实现（可能存在整数溢出等问题）
typedef struct {
    unsigned value; // 用于 futex 的内部状态
    // 可能还需要一个等待队列等内部结构
} cond_t;

// 与 CV 关联的互斥锁
mutex_t lk;

void cond_wait(cond_t *cv, mutex_t *lk) {
    // 1. 读取当前 futex 值（原子操作）
    //    这个值用于检测在释放锁和调用 futex_wait 之间是否有 signal 发生
    int current_value = atomic_load(&cv->value);

    // 2. 释放互斥锁
    mutex_unlock(lk);

    // 3. 调用 futex_wait
    //    原子地检查 cv->value 是否仍等于 current_value。
    //    如果是，则阻塞；
    //    如果不是（说明期间有 signal 修改了 value），则立即返回。
    futex_wait(&cv->value, current_value);

    // 4. 被唤醒后，重新获取互斥锁
    mutex_lock(lk);
}

void cond_signal(cond_t *cv) {
    // 1. 原子地增加 futex 值（通知潜在的 waiter 状态已变）
    atomic_fetch_add(&cv->value, 1);

    // 2. 调用 futex_wake 唤醒一个等待者
    futex_wake(&cv->value);
}

关键点在于，futex_wait 的原子检查机制确保了即使在 mutex_unlock 和 futex_wait 调用之间发生了 cond_signal（它会改变 cv->value），futex_wait 也会检测到这种变化并立即返回，从而避免了唤醒丢失。

Signal 语义：Hansen/Mesa vs. Hoare

cond_signal 的行为细节，特别是唤醒线程和锁的交接方式，存在两种主要语义：

  1. Hansen/Mesa 语义（常用，如 POSIX, Java）

    • cond_signal 只是将一个等待线程从等待队列移动到就绪队列，并不保证该线程立即运行，也不直接传递锁。
    • 发出 signal 的线程继续持有锁并执行，直到它释放锁。
    • 被唤醒的线程需要重新竞争获取互斥锁，当它最终获得锁并从 cond_wait 返回时，不能保证之前等待的条件仍然满足（因为其他线程可能在此期间运行并改变了状态）。
    sequenceDiagram
    participant T_Signal as 发信号线程（T1）
    participant T_Wait as 等待线程（T2）
    participant Mutex as 互斥锁
    participant CV as 条件变量

    T_Wait->>Mutex: 加锁
    T_Wait->>CV: cond_wait(cv, mutex)
    Mutex-->>T_Wait: 解锁（由 wait 隐式完成）
    CV-->>T_Wait: 阻塞
    T_Signal->>Mutex: 加锁
    T_Signal->>T_Signal: 修改共享状态
    T_Signal->>CV: cond_signal(cv)
    CV-->>T_Wait: 将 T2 从等待队列移动到就绪队列
    T_Signal->>T_Signal: 保持锁并继续执行
    T_Signal->>Mutex: 解锁
    Note over T_Wait: T2 竞争互斥锁
    T_Wait->>Mutex: 加锁（当被调度且成功时）
    Mutex-->>T_Wait: 获得锁
    CV-->>T_Wait: cond_wait 返回
    T_Wait->>T_Wait: 重新检查条件！

  2. Hoare 语义（理论上简洁，实现复杂）

    • cond_signal 立即将锁和 CPU 控制权从发信号的线程（T1）转移给被唤醒的线程（T2）。
    • T1 阻塞，直到 T2 执行完毕并释放锁（或者 T2 再次调用 cond_wait）时，锁和控制权才交还给 T1。
    • 优点：当 T2 从 cond_wait 返回时，可以保证其等待的条件仍然满足（因为没有其他线程能在这期间运行）。
    • 缺点：实现复杂，需要调度器进行特殊处理，可能导致更多的上下文切换。
    sequenceDiagram
    participant T_Signal as 发信号线程（T1）
    participant T_Wait as 等待线程（T2）
    participant Mutex as 互斥锁
    participant CV as 条件变量

    T_Wait->>Mutex: 加锁
    T_Wait->>CV: cond_wait(cv, mutex)
    Mutex-->>T_Wait: 解锁（由 wait 隐式完成）
    CV-->>T_Wait: 阻塞
    T_Signal->>Mutex: 加锁
    T_Signal->>T_Signal: 改变共享状态
    T_Signal->>CV: cond_signal(cv)
    Note over T_Signal, T_Wait: 原子地转移锁和 CPU 控制权
    Mutex-->>T_Wait: （从 T1）获取锁
    CV-->>T_Wait: cond_wait 返回
    T_Signal-->>T_Signal: T1 阻塞
    T_Wait->>T_Wait: 执行临界区（条件成立）
    T_Wait->>Mutex: 解锁
    Note over T_Signal, T_Wait: 将锁和控制权还给 T1
    Mutex-->>T_Signal: （从 T2）获取锁
    T_Signal-->>T_Signal: T1 恢复

现实世界的选择

由于实现简单且对调度器的要求较低，Hansen/Mesa 语义是现代操作系统和编程语言（如 POSIX, Java）的普遍选择。这意味着使用这些系统的条件变量时，必须遵循特定的编程模式。

Hansen/Mesa 语义下的正确用法：while 循环

由于 Mesa 语义下，线程被唤醒时不保证条件仍然满足，因此必须在 cond_wait 返回后重新检查条件。这通常通过将 cond_wait 放在 while 循环中实现：

1
2
3
4
5
6
7
// 正确模式（Mesa 语义）
mutex_lock(&lk);
while (!condition_is_met) { // 使用 while 而不是 if
    cond_wait(&cv, &lk);
}
// 条件满足，执行操作...
mutex_unlock(&lk);

原因：

  1. 虚假唤醒：某些实现可能在没有 signal 的情况下唤醒线程。
  2. 信号时机：在你被唤醒和重新获得锁之间，其他线程可能已经运行并再次改变了条件。
  3. broadcast：如果使用 cond_broadcast，多个线程被唤醒，但只有一个（或少数几个）线程的条件可能真正满足。

经验法则

在 Mesa 语义下，cond_wait 必须始终在 while 循环中调用。

生产者-消费者问题的条件变量解决方案

现在我们可以使用条件变量和 while 循环模式来正确解决生产者-消费者问题。

方案 1：使用两个条件变量

为生产者（等待缓冲区非满）和消费者（等待缓冲区非空）分别使用不同的条件变量，可以更精确地唤醒目标线程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
mutex_t lk = MUTEX_INIT();
cond_t cv_p = COND_INIT(); // 生产者的 CV（如果满则等待）
cond_t cv_c = COND_INIT(); // 消费者的 CV（如果空则等待）
int n; // 容量
int depth = 0; // 当前深度

void produce() {
    while(1) {
        mutex_lock(&lk);
        while (!(depth < n)) { // while 循环检查条件
            cond_wait(&cv_p, &lk); // 等待缓冲区非满
        }
        // 条件满足（depth < n）
        printf("(");
        depth++;
        // 唤醒可能在等待的消费者
        cond_signal(&cv_c);
        mutex_unlock(&lk);
    }
}

void consume() {
    while(1) {
        mutex_lock(&lk);
        while (!(depth > 0)) { // while 循环检查条件
            cond_wait(&cv_c, &lk); // 等待缓冲区非空
        }
        // 条件满足（depth > 0）
        printf(")");
        depth--;
        // 唤醒可能在等待的生产者
        cond_signal(&cv_p);
        mutex_unlock(&lk);
    }
}
  • 优点：signal 的目标明确，只唤醒需要被唤醒的角色（生产者唤醒消费者，消费者唤醒生产者）。
  • 缺点：需要管理两个条件变量。

方案 2：使用单个条件变量和 broadcast

使用一个条件变量，并在状态改变时唤醒所有等待者。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
mutex_t lk = MUTEX_INIT();
cond_t cv = COND_INIT(); // 单个 CV
int n, depth = 0;

void produce() {
    while(1) {
        mutex_lock(&lk);
        while (!(depth < n)) {
            cond_wait(&cv, &lk);
        }
        printf("(");
        depth++;
        // 唤醒所有等待者（生产者和消费者）
        cond_broadcast(&cv);
        mutex_unlock(&lk);
    }
}

void consume() {
    while(1) {
        mutex_lock(&lk);
        while (!(depth > 0)) {
            cond_wait(&cv, &lk);
        }
        printf(")");
        depth--;
        // 唤醒所有等待者
        cond_broadcast(&cv);
        mutex_unlock(&lk);
    }
}
  • 优点：实现相对简单，只需一个 CV。
  • 缺点：可能导致不必要的唤醒（例如，生产者唤醒了其他生产者，但缓冲区仍然是满的）。由于 while 循环的存在，这些被错误唤醒的线程会再次检查条件并继续等待，因此正确性不受影响，但可能带来性能开销（所谓的 thundering herd 问题[1]）。

单 CV + signal 的陷阱

如果只用一个 CV 并且使用 cond_signal，可能会出现问题。考虑缓冲区大小为 1，两个消费者 C1, C2 和一个生产者 P1：

  1. C1, C2 发现缓冲区为空，调用 cond_wait(&cv, &lk) 阻塞。
  2. P1 生产一个 item，调用 cond_signal(&cv)。假设 C1 被唤醒。
  3. 在 C1 重新获取锁之前，P1 再次运行（Mesa 语义允许），发现缓冲区已满 (depth=1)，调用 cond_wait(&cv, &lk) 阻塞。
  4. C1 获取锁，消费 item，depth 变为 0。C1 调用 cond_signal(&cv)。
  5. 问题：此时等待队列中有 P1 (需要 depth < 1) 和 C2 (需要 depth > 0)。如果 signal 唤醒了 C2，C2 会发现 depth == 0 并再次 wait。而 P1 则永远等待下去，导致死锁。

使用两个 CV 或使用 broadcast 可以避免此问题。

条件变量使用法则

  1. 必须有关联的互斥锁：保护共享状态（条件）。
  2. 必须有共享状态：用于判断条件是否满足。
  3. wait/signal/broadcast 时必须持有锁。（wait 内部会原子释放和重获锁）
  4. wait 必须在 while 循环内（Mesa 语义）。
  5. 优先为不同条件使用不同 CV，除非 broadcast 是明确需要或更简单的选择。

应用：协调计算依赖

条件变量的 wait/signal 机制非常适合用于实现计算依赖关系（happens-before）。

  • 计算图：可以将一个复杂的计算任务表示为一个有向无环图（DAG），其中节点 vv 代表计算任务（事件），边 (u,v)∈E(u, v) \in E 表示任务 vv 依赖于任务 uu 的结果（uu happens-before vv）。
  • 并行化：没有依赖关系（图中没有路径相连）的任务可以并行执行。

通用并行化方案

  1. 为每个计算节点 vv 分配一个线程（或将其视为一个待调度任务）。
  2. 等待条件：节点 vv 开始计算前，必须 wait 直到所有前驱节点 uu（即存在边 (u,v)(u, v)）都已完成。
  3. 完成信号：节点 uu 完成计算后，需要 signal（或 broadcast）通知所有后继节点 vv（即存在边 (u,v)(u, v)），告知它们依赖的条件可能已满足。

示例：并行化动态规划

动态规划算法通常涉及计算子问题的最优解，子问题之间存在依赖关系，形成一个 DAG。

编辑距离计算将字符串 A 转换为字符串 B 所需的最少编辑操作（插入、删除、替换）次数。有递推公式：dist(i, j) 依赖于 dist(i-1, j), dist(i, j-1), dist(i-1, j-1)。

  • 单线程：通常按行、按列或按对角线（拓扑序）计算。
  • 多线程：观察到同一条反斜线（i + j = const）上的所有 dist(i, j) 可以并行计算，因为它们只依赖于上一条反斜线 i + j - 1 和 i + j - 2 上的结果。
    • 可以为每条反斜线（或每个节点）创建一个任务/线程。
    • 使用条件变量（或信号量）来同步：计算 dist(i, j) 的线程需要等待计算 dist(i-1, j), dist(i, j-1), dist(i-1, j-1) 的线程完成。完成后，发出信号。

通用并发设计框架：调度器-工作者

一种常见的并发模式是将任务调度和执行分离：

  • 调度器线程（生产者）：维护计算图（或任务列表）和依赖关系。根据已完成的任务，找出当前可以运行的任务（满足依赖条件），将它们放入一个就绪队列，并 signal 工作者线程。
  • 工作者线程（消费者）：从就绪队列中取出任务并执行。执行前可能需要 wait 等待调度器的信号（如果队列为空）。执行完成后，通知调度器任务已完成（可能通过 signal 或更新共享状态）。这类似于线程池（Thread Pool）模型。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
// 概念代码
Queue ready_queue;
Mutex queue_lock;
CondVar queue_not_empty;
// ... 任务依赖、完成状态等共享数据 ...
Mutex state_lock;
CondVar state_changed; // 用于通知调度器

void T_worker() {
    while (1) {
        mutex_lock(&queue_lock);
        while (ready_queue.empty()) {
            cond_wait(&queue_not_empty, &queue_lock);
        }
        Task task = ready_queue.pop();
        mutex_unlock(&queue_lock);

        task.run(); // 执行计算

        // 通知调度器任务完成
        mutex_lock(&state_lock);
        update_completion_status(task);
        cond_signal(&state_changed); // 或 broadcast
        mutex_unlock(&state_lock);
    }
}

void T_scheduler() {
    while (!all_tasks_done()) {
        mutex_lock(&state_lock);
        // 等待工作者完成任务，触发状态更新
        // 可以使用 cond_wait(&state_changed, &state_lock) 或定期检查
        List<Task> ready_tasks = find_ready_tasks(); // 根据依赖和完成状态查找
        mutex_unlock(&state_lock);

        mutex_lock(&queue_lock);
        for (Task task : ready_tasks) {
            ready_queue.push(task);
            cond_signal(&queue_not_empty); // 通知工作者
        }
        mutex_unlock(&queue_lock);

        // 可能需要等待 state_changed 信号
        mutex_lock(&state_lock);
        while(find_ready_tasks().empty() && !all_tasks_done()) {
             cond_wait(&state_changed, &state_lock);
        }
        mutex_unlock(&state_lock);
    }
}

复杂示例：打印 fish

  • 任务：有三种线程，分别无限打印 <, >, -。需要同步它们，使得屏幕交替打印 <><_ 和 ><>_ 的组合序列。
  • 分析：这是一个状态同步问题。打印哪个字符取决于当前处于哪个状态。可以用有限状态机（FSM）来描述合法的前缀。还没学所以略。

信号量（Semaphore）

信号量（Semaphore）是 Dijkstra 在 1965 年提出的另一种强大的同步原语。与条件变量不同，信号量包含一个内部状态：一个非负整数值。

信号量

信号量是一个整数变量，只能通过两个原子操作来访问和修改：

  • P(sem)（来自「荷兰语」[2] prolaag，尝试减少），也称为 wait, down, acquire
  • V(sem)（来自荷兰语 verhoog，增加），也称为 signal, up, post, release

核心操作

  1. P(sem_t *sem) 或 sem_wait(sem_t *sem)
    • 原子地：
      1. 检查信号量 sem 的值。
      2. 如果 sem > 0，则将其减 1，操作完成，线程继续执行。
      3. 如果 sem <= 0，则线程阻塞，直到其他线程调用 V(sem) 使其值变为正数。
  2. V(sem_t *sem) 或 sem_post(sem_t *sem)
    • 原子地：
      1. 将信号量 sem 的值加 1。
      2. 如果此时有线程因 P(sem) 而阻塞在该信号量上，则唤醒其中一个线程。

POSIX API

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#include <semaphore.h>

sem_t sem;

// 初始化信号量
// value: 初始值
// pshared: 0 表示线程间共享，非 0 表示进程间共享
int sem_init(sem_t *sem, int pshared, unsigned int value);

// 销毁信号量
int sem_destroy(sem_t *sem);

// P 操作（等待/减少）
int sem_wait(sem_t *sem);
int sem_trywait(sem_t *sem); // 非阻塞版本
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout); // 带超时的版本

// V 操作（发布/增加）
int sem_post(sem_t *sem);

// 获取当前值（主要用于调试）
int sem_getvalue(sem_t *sem, int *sval);

基于互斥锁和条件变量的实现

信号量的行为可以用互斥锁和条件变量来模拟：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
typedef struct {
    int value;             // 信号量的值
    pthread_cond_t cond;   // 用于阻塞等待的条件变量
    pthread_mutex_t lock;  // 保护 value 和 cond 操作的互斥锁
} sem_t;

void sem_init(sem_t *s, int pshared, int value) {
    // pshared 参数在此实现中忽略
    s->value = value;
    pthread_cond_init(&s->cond, NULL);
    pthread_mutex_init(&s->lock, NULL);
}

void sem_wait(sem_t *s) { // P 操作
    pthread_mutex_lock(&s->lock);
    while (s->value <= 0) { // 值不大于 0 则等待
        pthread_cond_wait(&s->cond, &s->lock);
    }
    s->value--; // 值大于 0，减 1
    pthread_mutex_unlock(&s->lock);
}

void sem_post(sem_t *s) { // V 操作
    pthread_mutex_lock(&s->lock);
    s->value++; // 值加 1
    // 值增加后，可能使等待条件（value > 0）满足，唤醒一个等待者
    pthread_cond_signal(&s->cond);
    pthread_mutex_unlock(&s->lock);
}

void sem_destroy(sem_t *s) {
    pthread_cond_destroy(&s->cond);
    pthread_mutex_destroy(&s->lock);
}

注意：sem_post 只需要 cond_signal 而不是 broadcast，因为 P 操作每次只消耗一个单位的「资源」，唤醒一个线程就足够了。while 循环在 sem_wait 中仍然是必要的，以处理虚假唤醒和保证条件的正确检查。

标准信号量

上面的「核心操作」说在 sem <= 0 时会阻塞线程，直到 > 0 时再减，这样理论上不会出现 sem < 0 的情况。实际上标准信号量是「先减再检查」：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
typedef struct {
    int value;
    pthread_cond_t cond;
    pthread_mutex_t lock;
} sem_t;

void sem_wait(sem_t *s) { // P 操作
    pthread_mutex_lock(&s->lock);
    s->value--; // 先减 1
    if (s->value < 0) { // 如果减后 < 0，说明资源不足，阻塞
        pthread_cond_wait(&s->cond, &s->lock);
    }
    pthread_mutex_unlock(&s->lock);
}

void sem_post(sem_t *s) { // V 操作
    pthread_mutex_lock(&s->lock);
    s->value++; // 先加 1
    if (s->value <= 0) { // 如果加后仍 <= 0，说明有线程在等待
        pthread_cond_signal(&s->cond);
    }
    pthread_mutex_unlock(&s->lock);
}

信号量的用途与「整数」理解

信号量的整数值可以被赋予不同的含义，从而实现不同的同步模式：

  1. 互斥

    • 将信号量初始化为 1。
    • P(sem)：获取锁（资源数为 1，只有一个线程能成功 P 操作）。
    • V(sem)：释放锁（将资源数恢复为 1）。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    sem_t mutex;
    sem_init(&mutex, 0, 1); // 初始化为 1

    void critical_section_example() {
        sem_wait(&mutex);   // 进入临界区（P）
        // ... 临界区代码 ...
        sem_post(&mutex);   // 退出临界区（V）
        // ... 剩余区代码 ...
    }

  2. 顺序控制（Ordering/Happens-Before）

    • 将信号量初始化为 0。
    • 线程 A 在完成某个操作后调用 V(sem)。
    • 线程 B 在需要等待 A 完成时调用 P(sem)。由于初始值为 0，B 会阻塞，直到 A 调用 V 使值变为 1。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    sem_t order_sem;
    sem_init(&order_sem, 0, 0); // 初始化为 0

    // 线程 T2（子线程）
    void* child(void *arg) {
        do_something_first();
        sem_post(&order_sem); // V 操作：通知 T1 我已完成
        return NULL;
    }

    // 线程 T1（主线程）
    int main() {
        pthread_t c;
        create_thread(&c, child);
        // ... 其他操作 ...
        sem_wait(&order_sem); // P 操作：等待 T2 完成
        do_something_after_child();
        return 0;
    }

  3. 通用资源计数（Counting Semaphore）

    • 将信号量初始化为可用资源的数量 N。
    • P(sem)：请求/获取一个资源。如果 value > 0，则资源可用，value 减 1；否则阻塞等待。
    • V(sem)：释放/归还一个资源。value 加 1，并可能唤醒一个等待资源的线程。
    • 这正是生产者-消费者问题中缓冲区空位和数据项数量的抽象。

利用信号量解决生产者-消费者问题

信号量提供了一种非常自然和简洁的方式来解决生产者-消费者问题。

  • empty 信号量：表示缓冲区中空闲槽位的数量，初始值为 MAX（缓冲区容量）。生产者 P 此信号量。
  • full 信号量：表示缓冲区中已填充数据的数量，初始值为 0。消费者 P 此信号量。
  • mutex 信号量（或互斥锁）：用于保护对缓冲区的互斥访问，初始值为 1。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#define MAX 10 // 缓冲区容量
sem_t empty;
sem_t full;
sem_t mutex; // 或者使用 pthread_mutex_t

// 缓冲区数据结构
int buffer[MAX];
int fill_ptr = 0;
int use_ptr = 0;

void put(int value) { /* 放入 buffer */ }
int get() { /* 从 buffer 取出 */ }

void init() {
    sem_init(&empty, 0, MAX); // MAX 个空槽位
    sem_init(&full, 0, 0);    // 0 个数据项
    sem_init(&mutex, 0, 1);   // 互斥访问锁
}

void *producer(void *arg) {
    int i = 0;
    while (1) {
        // 生产数据 item ...
        int item = i++;

        sem_wait(&empty); // 等待并获取一个空槽位（P(empty)）
        sem_wait(&mutex); // 获取缓冲区访问锁（P(mutex)）
        put(item);        // 访问临界区：放入数据
        sem_post(&mutex); // 释放缓冲区访问锁（V(mutex)）
        sem_post(&full);  // 通知增加了一个数据项（V(full)）
    }
}

void *consumer(void *arg) {
    while (1) {
        sem_wait(&full);  // 等待并获取一个数据项 (P(full))
        sem_wait(&mutex); // 获取缓冲区访问锁 (P(mutex))
        int item = get(); // 访问临界区：取出数据
        sem_post(&mutex); // 释放缓冲区访问锁 (V(mutex))
        sem_post(&empty); // 通知增加了一个空槽位 (V(empty))

        // 消费数据 item ...
        printf("Consumed: %d\n", item);
    }
}

死锁陷阱：错误的加锁顺序

如果在消费者（或生产者）中颠倒 P(full) 和 P(mutex) 的顺序：

1
2
3
4
// 错误顺序 - 可能导致死锁
sem_wait(&mutex); // 先获取互斥锁
sem_wait(&full);  // 再等待数据项
// ...

死锁场景：

  1. 消费者 C 运行，获取 mutex 锁。
  2. C 尝试 sem_wait(&full)，但缓冲区为空（full 值为 0），C 阻塞，但仍然持有 mutex 锁。
  3. 生产者 P 运行，尝试 sem_wait(&mutex) 以访问缓冲区放入数据，但 mutex 已被 C 持有，P 阻塞。
  4. C 等待 P 生产数据（V(full)），P 等待 C 释放 mutex。死锁发生。

正确做法：总是先获取表示资源可用性（empty 或 full）的信号量，再获取用于互斥访问（mutex）的信号量/锁。

P/V 操作后的代码保护

在上面基于互斥锁和条件变量的信号量实现中，sem_wait 和 sem_post 内部使用了锁，但在函数返回时锁已经释放。因此，紧跟在 sem_wait 或 sem_post 调用之后的代码并不受该信号量内部锁的保护。如果这些代码需要访问共享资源（如生产者-消费者例子中的 buffer），则必须使用额外的互斥锁（如 mutex 信号量）来保护。

信号量的局限性

虽然信号量很强大，但在某些复杂场景下可能不如条件变量灵活：

  • 复杂条件：信号量只能表示整数状态。对于涉及多个变量或非整数条件的判断（如 fish 例子中的状态转移），使用条件变量配合显式状态检查通常更清晰。
  • 唤醒目标：V 操作通常只唤醒一个等待者，且无法精确指定唤醒哪种类型的等待者（如果多种线程等待同一个信号量）。条件变量可以通过使用不同的 CV 来区分等待条件和唤醒目标。例如，在单 CV + signal 的生产者-消费者陷阱中，信号量也可能遇到类似问题（消费者 V 操作唤醒了另一个消费者而不是生产者）。

互斥锁、条件变量与信号量的比较

特性 条件变量（Condition Variable） 信号量（Semaphore）
内部状态 无（无记忆） 非负整数（有记忆）
核心操作 wait, signal, broadcast P(wait/down), V(post/up)
关联锁 必须与互斥锁配合使用 可独立使用，或与互斥锁配合
条件检查 程序员必须在 while 循环中显式检查共享状态 P 操作隐式检查 value > 0
唤醒 signal 唤醒一个，broadcast 唤醒所有 V 操作唤醒一个
主要用途 等待任意复杂条件满足、协调线程执行 互斥、顺序控制、资源计数
灵活性 更高，适用于复杂条件 相对较低，主要用于计数和简单同步
易用性 模式固定（while 循环），但概念稍复杂 概念直接（资源获取/释放），但易因顺序错误死锁

选择哪种同步原语取决于具体问题的需求。对于简单的互斥、顺序控制或资源计数，信号量通常更简洁。对于需要等待复杂条件或需要更精细唤醒控制的场景，条件变量通常是更好的选择。

互斥锁、条件变量与信号量回顾

在讨论更高级的同步机制之前，先简要回顾一下上面讲到的基础原语：

  • 互斥锁：提供最基本的互斥访问，确保临界区代码的原子性。
  • 条件变量：允许线程在特定条件不满足时阻塞等待，并在条件满足时被唤醒。它本身不存储条件状态，必须与互斥锁配合使用，并在 while 循环中检查条件（Mesa 语义）。
  • 信号量：包含一个内部计数值，通过 P (wait/down) 和 V (post/up) 原子操作进行管理。可用于互斥（初始值为 1）、顺序控制（初始值为 0）和资源计数（初始值为 N）。

工具间的关系与选择

  • 信号量可以实现互斥锁的功能（使用二元信号量）。
  • 信号量也能方便地实现某些线程执行顺序的控制。
  • 条件变量与互斥锁结合，可以实现信号量的功能（如上一章笔记所示）。
  • 然而，信号量不适合表达任意复杂的同步条件，此时条件变量通常更灵活。正如 Butler Lampson 提醒的，要小心过度泛化[3]。
  • 条件变量配合互斥锁，可以构建几乎任何同步逻辑。

接下来，我们将探讨更复杂的同步问题和更高级的同步技术。

读者-写者问题（Readers-Writers Problem）

这是一个经典的并发控制问题，旨在优化对共享数据的访问，特别是当读操作远多于写操作时。

问题定义与挑战

  • 场景：多个线程需要访问共享数据。部分线程只需要读取数据（读者，Reader），部分线程需要修改数据（写者，Writer）。
  • 核心矛盾：
    • 读取操作本身不会修改数据，因此多个读者同时读取是安全的，允许并发进行以提高性能。
    • 写入操作会修改数据，必须保证其原子性和排他性：
      • 任意时刻最多只能有一个写者在修改数据（写-写互斥）。
      • 当有写者在修改数据时，不能有任何读者或其他写者在访问（写-读/写-写互斥）。
graph LR
    subgraph "读者-写者并发控制"
        DB[(共享数据区)]

        subgraph "读操作"
            R1[读者 1]
            R2[读者 2]
            R3[读者 3]
        end

        subgraph "写操作"
            W1[写者 1]
            W2[写者 2]
        end

        R1 -->|可以<br>同时读取| DB
        R2 -->|可以<br>同时读取| DB
        R3 -->|可以<br>同时读取| DB

        W1 <== 互斥<br>写入 ==> W2

        classDef reader fill:#b5e2fa,stroke:#0077b6,color:black
        classDef writer fill:#ffafcc,stroke:#9d4edd,color:black
        classDef database fill:#caffbf,stroke:#38b000,color:black

        class R1,R2,R3 reader
        class W1,W2 writer
        class DB database
    end

    写操作 --> DB

    subgraph "访问规则"
        style 访问规则 fill:none,stroke:#666,stroke-width:1px

        Rule1["✓ 读-读并发: 多个读者可同时访问"]
        Rule2["✗ 写-写互斥: 写者之间互相排斥"]
        Rule3["✗ 读-写互斥: 读者与写者互相排斥"]

        style Rule1 fill:#d8f3dc,stroke:#2d6a4f,color:black
        style Rule2 fill:#ffddd2,stroke:#e71d36,color:black
        style Rule3 fill:#ffddd2,stroke:#e71d36,color:black
    end

目标

设计一种同步机制，允许多个读者并发访问，但写者必须独占访问权限。简单的全局互斥锁虽然能保证安全，但限制了读者的并发性，性能较低。

读写锁（Readers-Writers Lock）

为了解决读者-写者问题，引入了读写锁（Readers-Writers Lock, RW Lock）。典型的 API 接口如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 假设的读写锁结构体
typedef struct rwlock_t rwlock_t;

// 初始化
void rwlock_init(rwlock_t *rw);

// 读者获取锁
void acquire_readlock(rwlock_t *rw);
// 读者释放锁
void release_readlock(rwlock_t *rw);

// 写者获取锁
void acquire_writelock(rwlock_t *rw);
// 写者释放锁
void release_writelock(rwlock_t *rw);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 使用模式
rwlock_t rw_lock;
rwlock_init(&rw_lock);

// 读者线程
void reader() {
    acquire_readlock(rw_lock);
    // 读取共享数据…
    release_readlock(rw_lock);
    // ...
}

// 写者线程
void writer() {
    acquire_writelock(rw_lock);
    // 修改共享数据…
    release_writelock(rw_lock);
    // ...
}

实现方案与策略

读写锁的实现需要管理内部状态（如当前读者数量、是否有写者等），并处理读者和写者之间的竞争。不同的实现策略会导致不同的公平性表现。

方案 1：读者优先（Reader Preference）

这种策略下，只要当前没有写者持有锁，新的读者就可以立即获得读锁，即使有写者正在等待。

核心逻辑：

  • 使用一个计数器 readers 记录当前持有读锁的读者数量。
  • 使用一个互斥锁 lock 保护 readers 计数器的访问。
  • 使用一个信号量 rwlk（初始值为 1）控制对临界区的访问（写者直接获取，第一个读者获取，最后一个读者释放）。

进入条件：

  • 读者：如果临界区为空（readers == 0）或已有其他读者（readers > 0），且没有写者持有锁，则可进入。
  • 写者：只有当临界区为空（没有读者也没有写者）时才可进入。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
// 读者优先实现（基于信号量）
typedef struct _rwlock_t {
    sem_t lock;    // 保护 readers 计数的锁
    sem_t rwlk;    // 控制临界区访问（写者或第一个读者获取）
    int readers;   // 当前读者数量
} rwlock_t;

void rwlock_init(rwlock_t *rw) {
    rw->readers = 0;
    sem_init(&rw->lock, 0, 1);
    sem_init(&rw->rwlk, 0, 1); // 初始时临界区可用
}

void acquire_readlock(rwlock_t *rw) {
    P(&rw->lock); // 获取计数锁
    rw->readers++;
    if (rw->readers == 1) { // 第一个读者
        P(&rw->rwlk); // 尝试获取临界区访问权（阻止写者）
    }
    V(&rw->lock); // 释放计数锁
}

void release_readlock(rwlock_t *rw) {
    P(&rw->lock); // 获取计数锁
    rw->readers--;
    if (rw->readers == 0) { // 最后一个读者
        V(&rw->rwlk); // 释放临界区访问权（允许写者进入）
    }
    V(&rw->lock); // 释放计数锁
}

void acquire_writelock(rwlock_t *rw) {
    P(&rw->rwlk); // 获取临界区访问权（阻止其他读者和写者）
}

void release_writelock(rwlock_t *rw) {
    V(&rw->rwlk); // 释放临界区访问权
}

写者饿死

如果读者持续不断地到来，readers 计数将一直大于 0，导致 rwlk 信号量一直被读者持有（从第一个读者获取后直到最后一个读者释放前）。这会使得等待的写者永远无法获得 rwlk，发生「饿死」。

方案 2：写者优先（Writer Preference）

这种策略下，一旦有写者在等待，后续到达的读者将被阻塞，直到所有等待的写者完成操作。

核心逻辑：

  • 增加 writers 计数，记录等待和正在写的写者总数。
  • 增加一个信号量 tryRead，当有写者希望进入时，写者会持有 tryRead，阻止新读者尝试获取读锁。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
// 写者优先实现（概念性，基于课件代码）
typedef struct _rwlock_t {
    sem_t rlock;     // 保护 readers 计数的锁
    sem_t wlock;     // 保护 writers 计数的锁
    sem_t rwlk;      // 控制临界区访问
    sem_t tryRead;   // 读者尝试进入前检查，写者等待时持有
    int readers;     // 当前读者数
    int writers;     // 等待或正在写的写者数
} rwlock_t;

void rwlock_init(rwlock_t *rw) {
    rw->readers = 0;
    rw->writers = 0;
    sem_init(&rw->rlock, 0, 1);
    sem_init(&rw->wlock, 0, 1);
    sem_init(&rw->rwlk, 0, 1);
    sem_init(&rw->tryRead, 0, 1); // 初始允许读者尝试
}

void acquire_readlock(rwlock_t *rw) {
    P(&rw->tryRead); // 尝试获取读权限（若有写者等待则阻塞）
    P(&rw->rlock);
    rw->readers++;
    if (rw->readers == 1) {
        P(&rw->rwlk); // 第一个读者获取临界区锁
    }
    V(&rw->rlock);
    V(&rw->tryRead); // 允许其他读者尝试（如果此时没有写者等待）
}

void release_readlock(rwlock_t *rw) {
    P(&rw->rlock);
    rw->readers--;
    if (rw->readers == 0) {
        V(&rw->rwlk); // 最后一个读者释放临界区锁
    }
    V(&rw->rlock);
}

void acquire_writelock(rwlock_t *rw) {
    P(&rw->wlock);
    rw->writers++;
    if (rw->writers == 1) {
        P(&rw->tryRead); // 第一个写者阻止后续读者尝试进入
    }
    V(&rw->wlock);
    P(&rw->rwlk); // 获取临界区锁
}

void release_writelock(rwlock_t *rw) {
    V(&rw->rwlk); // 释放临界区锁
    P(&rw->wlock);
    rw->writers--;
    if (rw->writers == 0) {
        V(&rw->tryRead); // 最后一个写者离开，允许读者尝试进入
    }
    V(&rw->wlock);
}

读者饿死

如果写者持续不断地到来，writers 计数将一直大于 0，导致 tryRead 信号量一直被写者持有。这会使得等待的读者永远无法通过 P(&rw->tryRead)，发生「饿死」。

方案 3：公平性考量（Ticket Lock）

为了避免读者或写者饿死，可以引入排队机制，确保请求按到达顺序得到服务。Ticket Lock 是一种实现 FIFO（先进先出）排队的方式。

Ticket Lock 机制：

  • ticket：一个全局递增的票号分发器。
  • turn：当前允许进入临界区的票号。
  • 获取锁：原子地获取当前 ticket 值作为自己的 myturn，并将 ticket 加 1。然后自旋等待，直到 turn 等于 myturn。
  • 释放锁：将 turn 加 1，允许下一位等待者进入。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// Ticket Lock 实现
typedef struct lock_ticket {
    int ticket; // 当前已分发的最大票号
    int turn;   // 当前服务的票号
} ticket_t;

void lock_init(ticket_t* tlk) {
    tlk->ticket = 0;
    tlk->turn = 0;
}

void ticket_lock(ticket_t* tlk) {
    int myturn = __sync_fetch_and_add(&tlk->ticket, 1); // 原子增并返回旧值（GCC）
    while (tlk->turn != myturn) {
        // 自旋等待（或使用更节能的阻塞方式）
        ; // spin
    }
}

void ticket_unlock(ticket_t* tlk) {
    // 确保 turn 的更新对其他 CPU 可见（内存屏障隐含在原子操作中）
    __sync_fetch_and_add(&tlk->turn, 1);
}

基于 Ticket Lock 的公平读写锁：

  • 所有读者和写者先通过同一个 Ticket Lock (queue) 排队获取进入许可。
  • 获得许可后，再根据是读者还是写者，执行类似读者优先的逻辑（使用 rlock 保护 readers，使用 rwlk 控制实际临界区访问）。
  • 关键在于，Ticket Lock 保证了无论是读者还是写者，都按照到达顺序获得尝试进入临界区的机会，避免了某一方无限等待。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
// 公平读写锁（基于 Ticket Lock 和读者优先逻辑）
typedef struct _rwlock_t {
    sem_t rlock;     // 保护 readers 计数
    sem_t rwlk;      // 控制临界区访问
    ticket_t queue;  // FIFO 排队锁
    int readers;     // 当前读者数
} rwlock_t;

void rwlock_init(rwlock_t *rw) {
    rw->readers = 0;
    sem_init(&rw->rlock, 0, 1);
    sem_init(&rw->rwlk, 0, 1);
    lock_init(&rw->queue);
}

void acquire_readlock(rwlock_t *rw) {
    ticket_lock(&rw->queue); // 先排队获取进入许可
    P(&rw->rlock);
    rw->readers++;
    if (rw->readers == 1) {
        P(&rw->rwlk); // 第一个读者获取临界区锁
    }
    V(&rw->rlock);
    ticket_unlock(&rw->queue); // 释放排队锁（允许下一个人排队后操作）
}

void release_readlock(rwlock_t *rw) {
    P(&rw->rlock);
    rw->readers--;
    if (rw->readers == 0) {
        V(&rw->rwlk); // 最后一个读者释放临界区锁
    }
    V(&rw->rlock);
}

void acquire_writelock(rwlock_t *rw) {
    ticket_lock(&rw->queue); // 先排队获取进入许可
    P(&rw->rwlk); // 获取临界区锁（此时 queue 锁仍持有，保证顺序）
}

void release_writelock(rwlock_t *rw) {
    V(&rw->rwlk); // 释放临界区锁
    ticket_unlock(&rw->queue); // 确保写者完成操作后才释放排队锁
}

公平性与性能

公平锁通常能避免饿死，但可能牺牲一定的吞吐量。例如，即使临界区当前只有读者，一个等待的写者也会阻止后续读者进入，直到写者完成。

POSIX 读写锁

POSIX 标准提供了 pthread_rwlock_t 类型和相关函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#include <pthread.h>

pthread_rwlock_t rw_lock;

// 初始化（attr 可为 NULL）
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
                        const pthread_rwlockattr_t *restrict attr);
// 销毁
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

// 获取读锁（阻塞）
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
// 获取写锁（阻塞）
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);

// 尝试获取读锁（非阻塞）
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);
// 尝试获取写锁（非阻塞）
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);

// 释放读锁或写锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

Linux Pthread 实现

在 Linux 中，pthread_rwlock_t 的默认行为通常是读者优先。但是，可以通过 pthread_rwlockattr_setkind_np 函数设置属性，选择写者优先（PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP）或其他策略（具体可用策略依赖于库版本）。

无锁编程：Read-Copy-Update (RCU)

在追求极致性能的场景下，即使是读写锁带来的开销（尤其是写锁的排他性）也可能成为瓶颈。Read-Copy-Update(RCU) 是一种高级的、主要用于内核的同步技术，旨在实现极低开销的读取操作，甚至达到「无锁」读取。

动机——锁的开销与瓶颈：

  • 锁操作代价：获取和释放锁（即使是互斥锁或读写锁）涉及原子操作、缓存一致性协议交互、可能的上下文切换，其 CPU 周期消耗远高于普通指令。
  • 短临界区问题：当临界区非常短时，锁本身的开销可能超过临界区内有效工作的时间，成为性能瓶颈。线程越多，竞争越激烈，性能可能反而下降。
  • 读写锁的局限：虽然读写锁允许多读并发，但写操作仍然需要排他性，会阻塞所有读者和写者。

RCU 的思想源于对数据一致性要求的放宽：

  • 强一致性：任何更新操作完成后，所有后续读取都能立即看到新数据。锁机制通常提供强一致性。
  • 最终一致性（Eventual Consistency）：系统保证如果没有新的更新，最终所有访问都会读取到最后更新的值。但在更新期间，读取操作可能读到旧数据。

RCU 利用最终一致性，允许读者在写者更新数据期间继续访问数据的旧版本，从而避免阻塞。

RCU 读者视角

当写者更新数据时，正在进行的读者操作：

  • 要么读到更新前的旧数据。
  • 要么（如果读操作开始在写者发布更新之后）读到更新后的新数据。
  • 绝不会读到写操作过程中的中间状态（RCU 保证写的「发布」是原子的）。

RCU 的基本思路是：

  1. 读者：读取数据时不加锁。它们直接访问数据。
  2. 写者：
    • 复制（Copy）：不直接在原地修改数据，而是创建一个数据的副本。
    • 修改（Update）：在副本上进行修改。
    • 发布（Publish/Removal）：通过一次原子操作（通常是修改指针），将共享引用指向新的、修改后的副本。旧版本的数据暂时保留。
    • 等待宽限期（Grace Period）：等待系统中所有可能还在引用旧版本数据的读者完成它们的读操作。
    • 回收（Reclamation）：在宽限期结束后，安全地释放（回收）旧版本数据的内存。

RCU 特点

  • 读者零开销：读操作几乎没有同步开销（可能只有内存屏障或简单的计数器操作）。
  • 写者不阻塞读者：写操作（复制、修改、发布）可以与读者并发进行。
  • 写者有额外成本：写者需要复制数据、等待宽限期和管理旧版本回收，成本相对较高。
  • 适用于读多写少：非常适合读取操作远多于写入操作的场景。
  • 允许多版本共存：在更新期间，新旧版本数据会短暂共存。读者可能读到旧版本，也可能读到新版本，但保证读到的版本是一致的（不会读到写操作进行到一半的中间状态）。这是一种最终一致性的体现。

RCU 机制详解

RCU 主要由针对读者和写者的不同机制组成。

读者：近乎零开销

  • RCU 读者的临界区（read-side critical section）通常不需要获取任何锁，也不需要进行原子写操作。
  • 读者只需标记其临界区的开始和结束（例如，通过 rcu_read_lock() 和 rcu_read_unlock()）。这些标记本身开销极低，在某些实现中甚至可能只是禁用/启用抢占或中断。
  • 由于不获取锁，读者不会被写者阻塞，可以并发执行。

写者：Removal 与 Reclamation

RCU 的写操作（更新）被分解为两个关键阶段：

  1. Removal（移除）阶段：

    • 写者创建一个数据的新副本，并在副本上进行修改。
    • 通过一次原子操作（如原子指针赋值，通常需要内存屏障保证顺序）将指向旧数据的指针更新为指向新副本。
    • 这个阶段完成后，新的读者将看到新数据。但旧数据不能立即删除，因为可能还有读者正在访问它。
    • 关键：Removal 阶段可以与读者并发进行。

  2. Reclamation（回收）阶段：

    • 写者需要等待一个宽限期结束。
    • 宽限期是指从写者完成 Removal 操作开始，直到系统中所有在 Removal 操作之前就已经进入 RCU 读临界区的读者全部退出临界区为止的时间段。
    • 宽限期结束后，写者可以安全地回收（例如，释放内存）旧版本的数据，因为此时已没有任何读者会访问它。
    • 关键：Reclamation 阶段需要与读者进行同步（等待宽限期）。

示例：链表更新

假设要更新链表中的节点 d 为 d'：

RCU 的一个特点是需要维护数据的多个版本（至少是旧版本和新版本），直到旧版本可以安全回收。

保证 Removal 原子性：订阅-发布与内存屏障

简单地在写者端加锁、读者端不加锁，并不能保证 Removal 的原子性。编译器优化和弱内存模型可能导致指令重排，使得读者看到处于中间状态的数据（例如，新节点的数据还没完全初始化好，指针就已经被更新了）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// 潜在问题代码（无内存屏障）
struct foo { int a, b, c; };
struct foo *gp = NULL;
mutex_t lk;

void foo_update() {
    struct foo *new_gp = kmalloc(...);
    mutex_lock(&lk);
    new_gp->a = 1; // 写新数据
    new_gp->b = 2;
    new_gp->c = 3;
    struct foo *old_gp = gp;
    gp = new_gp; // 发布新指针 <--- 问题点
    mutex_unlock(&lk);
    // 稍后 kfree(old_gp)
}

void foo_read() {
    struct foo *p = gp; // 读取指针 <--- 问题点
    if (p != NULL) {
        // 可能读到 p 指向 new_gp，但 a, b, c 尚未完全写入！
        do_something(p->a, p->b, p->c);
    }
}

RCU 使用了类似订阅-发布（Publish-Subscribe）的模式，并结合内存屏障来解决这个问题：

  • 发布（Write/Publish）：写者使用特殊原语（如 Linux 内核的 rcu_assign_pointer()）来更新指针。这个原语内部包含内存屏障，确保在更新指针之前，对新数据的所有修改都已完成并对其他 CPU 可见。
  • 订阅（Read/Subscribe）：读者使用特殊原语（如 rcu_dereference()）来读取指针。这个原语内部也包含内存屏障，确保读取指针的操作不会被重排到读取数据内容之后，并且能正确观察到写者发布的数据。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
// RCU API 使用示例（概念性）
foo *gp = NULL; // 受 RCU 保护的全局指针

// 写者线程（更新）
void foo_update() {
    foo *new_gp = kmalloc(...); // 分配新结构
    // 初始化 new_gp 的内容（a=1, b=2, c=3）…
    new_gp->a = 1; new_gp->b = 2; new_gp->c = 3;

    foo *old_gp = gp; // 保存旧指针，用于后续回收

    // 发布更新（原子，带内存屏障）
    rcu_assign_pointer(gp, new_gp);

    // 等待宽限期结束（同步点）
    synchronize_rcu();

    // 回收旧数据
    kfree(old_gp);
}

// 读者线程（读取）
void foo_read() {
    // 标记 RCU 读临界区开始
    rcu_read_lock();

    // 获取指针（保证读到有效指针，带内存屏障）
    foo *p = rcu_dereference(gp);

    if (p != NULL) {
        // 访问数据（保证在获取指针之后）
        do_something(p->a, p->b, p->c);
    }

    // 标记 RCU 读临界区结束
    rcu_read_unlock();
}

保证 Reclamation 安全性：宽限期（Grace Period）

写者如何知道何时可以安全回收旧数据？需要等待所有在 Removal 之前就已存在的读者完成它们的读操作。

  • 读者标记：rcu_read_lock() 和 rcu_read_unlock() 用于显式标记读者访问 RCU 保护数据的区间。
  • 写者等待：synchronize_rcu() 是一个阻塞函数，调用它的写者线程会一直等待，直到当前 RCU 宽限期结束。

宽限期图示（Grace Period Timeline）:

sequenceDiagram
    participant Writer as 写者
    participant RCU as RCU 子系统
    participant Reader1 as 读者 1（旧）
    participant Reader2 as 读者 2（旧）
    participant Reader3 as 读者 3（新）
    participant Reader4 as 读者 4（新）

    Note over Writer: T0: 发布更新（e.g., rcu_assign_pointer）

    Reader1->>Reader1: rcu_read_lock()
    Reader1->>Reader1: 读取旧数据（R1）
    Reader2->>Reader2: rcu_read_lock()
    Reader2->>Reader2: 读取旧数据（R2）

    Writer->>RCU: synchronize_rcu()/开始等待宽限期

    Reader1->>Reader1: rcu_read_unlock() (T_R1_end)
    Note over RCU: 检测到 CPU_Reader1 静止

    Reader3->>Reader3: rcu_read_lock()
    Reader3->>Reader3: 读取新数据（R3）

    Reader2->>Reader2: rcu_read_unlock() (T_R2_end)
    Note over RCU: 检测到 CPU_Reader2 静止

    Reader4->>Reader4: rcu_read_lock()
    Reader4->>Reader4: 读取新数据（R4）

    Note over RCU: 所有 CPU 均经历静止状态
    RCU-->>Writer: 宽限期结束（Grace Period Expires）
    Writer->>Writer: 安全回收旧数据

    Reader3->>Reader3: rcu_read_unlock() (T_R3_end)
    Reader4->>Reader4: rcu_read_unlock() (T_R4_end)

宽限期的实现机制

如何检测宽限期结束？关键在于检测所有 CPU 是否都经历了一个静止状态（quiescent state），表明它们已经处理完了在宽限期开始前就在运行的 RCU 读临界区。

经典内核实现：

  • 读者 rcu_read_lock() 可能只是禁用抢占（阻止当前线程被切换出去）。
  • 读者 rcu_read_unlock() 则重新启用抢占。
  • 静止状态可以是 CPU 发生上下文切换、进入 idle 状态、或执行用户态代码（因为内核 RCU 读临界区必须在内核态）。
  • synchronize_rcu() 会等待，直到检测到系统中的所有 CPU 都至少经历过一次静止状态。这保证了任何在 synchronize_rcu() 调用前开始的 RCU 读临界区都已经结束。

RCU 的具体实现非常复杂，涉及多种优化和变体（如 Tree RCU）。上述只是简化模型。关键在于理解「宽限期」的概念和「静止状态」检测的必要性。

异步 Reclamation 与用户态 RCU

如果写者不想在 synchronize_rcu() 处阻塞，可以使用 call_rcu(callback_func, data)。它注册一个回调函数，在宽限期结束后，系统会自动调用该函数来执行旧数据的回收操作。

RCU 的概念也可以应用于用户态程序，例如 liburcu 库。用户态实现需要不同的机制来标记读临界区和检测静止状态（例如，基于线程信号、内存屏障和 epoch 机制）。

RCU 适用场景

RCU 是一种强大的优化技术，但并非万能。

  • 优点：
    • 读取操作非常快（接近零开销），且不被写者阻塞。
    • 在高并发、读多写少（read-mostly）的场景下性能极佳。
    • 适用于需要高可扩展性的数据结构（如内核路由表、系统配置信息等）。
  • 缺点：
    • 写操作相对复杂，需要管理数据副本和处理回收逻辑。
    • 写者可能因等待宽限期而延迟。
    • 需要额外的内存来存储旧版本数据。
    • 只适用于基于指针访问的数据结构。
    • 编程模型比锁更复杂，需要仔细理解其语义。

何时考虑 RCU？

当你的应用场景符合「读操作极其频繁、写操作相对较少、读取性能是瓶颈、可以容忍读取到稍旧的数据（最终一致性）」这些特点时，RCU 是一个值得考虑的选项。

经典同步问题：哲学家就餐（Dining Philosophers Problem）

这是由 Edsger Dijkstra 提出的另一个经典的并发问题，用于说明死锁（Deadlock）和饿死（Starvation）的概念及避免策略。

问题描述

  • 场景：N 位哲学家围坐在一张圆桌旁，桌上放着 N 支筷子[4]，每两位哲学家之间放一支。
  • 行为：哲学家要么思考（不需要资源），要么吃饭（需要资源）。
  • 资源需求：每位哲学家需要同时拿起左右两边的两支筷子才能吃饭。
  • 目标：设计一个协议（算法），让哲学家们能够合理地交替思考和吃饭，满足以下条件：
    1. 互斥：一支筷子同一时刻只能被一位哲学家持有。
    2. 无死锁：系统不会进入所有（或部分）哲学家永远无法获得足够筷子吃饭的状态。
    3. 无饿死：任何一位想吃饭的哲学家最终都能吃到饭（不会无限期等待）。
    4. 高并发：允许多位哲学家同时吃饭（理想情况下最多 N/2 位）。

死锁与饿死

这两个概念都关乎系统的活性，即系统能否持续做有用的工作。它们不违背安全性。

  • 饿死：一个线程在有限时间内无法获得所需资源或 CPU 时间，导致其无法向前推进。
  • 死锁：一种特殊的饿死状态。它发生在多个线程形成一个循环等待链，每个线程持有至少一个资源，并等待链中下一个线程所持有的资源。由于等待关系是循环的，没有任何线程能打破这个循环，导致所有涉及的线程都永久阻塞。

死锁必然导致饿死，但饿死不一定是死锁。例如，由于调度策略不公或运气不好，一个线程可能总是抢不到锁，即使没有形成死锁环路，也会饿死。

解决方案探索

我们将筷子视为资源，每支筷子可以用一个信号量（初始值为 1）或互斥锁来表示。

方案 1：朴素尝试（导致死锁）

每个哲学家都先尝试拿起左手边的筷子，再尝试拿起右手边的筷子。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#define N 5
sem_t forks[N]; // 每支筷子是一个信号量，初始值为 1

int left(int i) { return i; }
int right(int i) { return (i + 1) % N; }

void philosopher(int i) {
    while (1) {
        think();
        P(&forks[left(i)]);  // 拿起左筷
        P(&forks[right(i)]); // 拿起右筷
        eat();
        V(&forks[left(i)]);  // 放下左筷
        V(&forks[right(i)]); // 放下右筷
    }
}

死锁风险

如果所有哲学家同时拿起他们左手边的筷子，那么每位哲学家都持有了一支筷子，并等待右手边的筷子。而右手边的筷子正被邻居持有并等待更右边的。形成了一个循环等待，所有哲学家都无法获得第二支筷子，导致死锁。

方案 2：尝试获取与释放（导致饿死/活锁）

为了避免死锁，哲学家在拿不到第二支筷子时，主动放下第一支筷子，稍等片刻再重试。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
int tryP(sem_t *s) { // 尝试 P 操作（非阻塞）
    // sem_trywait 返回 0 表示成功，否则失败
    return sem_trywait(s);
}

void philosopher(int i) {
    while (1) {
        think();
        while (1) { // 循环尝试直到成功拿起两支筷子
            P(&forks[left(i)]); // 拿起左筷
            if (tryP(&forks[right(i)]) == 0) { // 尝试拿右筷
                // 成功拿到两支
                break; // 跳出尝试循环，去吃饭
            } else {
                // 拿右筷失败
                V(&forks[left(i)]); // 放下左筷
                sleep(sometime); // 等待一段时间再试
            }
        }
        eat();
        V(&forks[left(i)]);
        V(&forks[right(i)]);
    }
}

活锁/饿死风险

虽然避免了死锁（因为线程会释放资源），但可能出现活锁（Livelock）或饿死。

例如，运气不好的话，所有哲学家可能在同一时间尝试拿筷子、失败、放下、等待，然后又在同一时间再次尝试……陷入一种不断尝试但无法前进的状态。

即使不是严格的活锁，某个哲学家也可能因为运气差或 sleep 时间不当而持续失败，导致饿死。

方案 2 改进：随机退避

将 sleep(sometime) 改为 sleep(random_time)。引入随机性可以大大降低活锁的可能性。

1
2
3
// ... 同方案 2，但使用随机等待时间 ...
sleep(random_time()); // 等待随机时间
// ...

实用性

随机退避是一种在分布式系统和网络协议中常用的解决冲突和避免活锁的方法。在哲学家就餐问题中，它在实践中通常能工作得很好，但理论上仍然不能完全排除极小概率下的饿死情况。对于安全攸关系统，可能需要更确定性的解决方案。

方案 3：全局锁（并发度低）

在哲学家尝试拿起任何筷子之前，先获取一个全局的互斥锁（例如 Ticket Lock）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
ticket_t turn_lock; // 全局排队锁
lock_init(&turn_lock);

void philosopher(int i) {
    while (1) {
        think();
        ticket_lock(&turn_lock); // 获取全局锁
        P(&forks[left(i)]);
        P(&forks[right(i)]);
        eat();
        V(&forks[left(i)]);
        V(&forks[right(i)]);
        ticket_unlock(&turn_lock); // 释放全局锁
    }
}

并发度极低

这种方法能有效避免死锁和饿死（因为访问是串行的），但同一时间最多只有一位哲学家能吃饭，并发度非常低，没有利用问题的潜在并行性。

方案 3 改进：限制并发数（鸽巢原理）

允许一部分哲学家（例如 N-1 位）同时尝试获取筷子。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#define N 5
sem_t capacity; // 最多允许 N-1 位哲学家同时进入「尝试拿筷子」阶段
sem_init(&capacity, 0, N - 1); // 初始化容量为 N-1

void philosopher(int i) {
    while (1) {
        think();
        P(&capacity); // 获取一个「尝试名额」
        P(&forks[left(i)]);
        P(&forks[right(i)]);
        eat();
        V(&forks[left(i)]);
        V(&forks[right(i)]);
        V(&capacity); // 释放「尝试名额」
    }
}

鸽巢原理保证无死锁

最多只有 N-1 位哲学家在竞争 N 支筷子。根据鸽巢原理，至少有一位哲学家能够成功拿到左右两支筷子（因为即使 N-1 位哲学家每人拿了一支，也还剩下一支筷子可以被某人拿起作为第二支）。这位能吃到饭的哲学家最终会释放筷子，从而让其他人有机会吃饭。这种方法提高了并发度（允许多人同时吃饭），并保证了无死锁。

方案 4：资源分级（破坏循环等待）

破坏死锁的四个必要条件之一：循环等待。可以通过给资源（筷子）编号，并规定所有哲学家必须按「固定顺序」（例如，先拿编号较小的筷子，再拿编号较大的筷子）来获取资源。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
void philosopher(int i) {
    int fork1 = left(i);
    int fork2 = right(i);

    // 确保总是先 P 编号小的筷子
    if (fork1 > fork2) {
        int temp = fork1;
        fork1 = fork2;
        fork2 = temp;
    }

    while (1) {
        think();
        P(&forks[fork1]); // 先拿编号小的筷子
        P(&forks[fork2]); // 再拿编号大的筷子
        eat();
        V(&forks[fork1]); // 按任意顺序释放（或按获取逆序）
        V(&forks[fork2]);
    }
}

无死锁且并发性好

这种方法通过强制规定资源获取顺序，打破了潜在的循环等待链，从而保证无死锁。它不需要额外的锁或信号量，并且允许多达 N/2 位哲学家同时吃饭（如果他们的筷子不冲突），具有较好的并发性。这是解决此类问题的常用且优雅的策略。

问题启示

哲学家就餐问题是理解和解决并发系统中资源竞争、死锁和饿死问题的经典模型。它展示了：

  • 简单直观的并发策略可能隐藏着严重的活性问题。
  • 避免死锁的常用策略包括：资源预分配（一次性获取所有资源，可能降低并发）、资源有序分配（破坏循环等待）、引入协调者或限制并发数。
  • 需要在安全性（不发生错误）、活性（系统能持续运行）和性能（高并发、低延迟）之间做出权衡。

总结

本章我们深入探讨了更高级的同步问题和技术：

  • 读者-写者问题：引出了读写锁，并讨论了不同策略（读者优先、写者优先、公平策略）及其对饿死风险的影响。
  • Read-Copy-Update(RCU)：一种用于高性能、读密集场景的无锁读取技术。通过版本控制、原子发布和延迟回收（需等待宽限期）机制，实现极低开销的读者操作，但写者逻辑更复杂。
  • 哲学家就餐问题：一个经典的用于演示死锁和饿死问题的模型。探讨了多种解决方案，包括可能导致问题的朴素方法、基于退避的方法、限制并发数的方法（鸽巢原理）以及通过资源分级破坏循环等待的有效方法。

  1. 惊群问题，指在多进程/多线程环境中，当多个等待者（进程/线程）同时等待同一个资源或事件时，一旦该资源可用或事件发生，所有等待者都会被唤醒，但最终只有一个能成功获取资源，其他所有被唤醒的进程/线程都会重新进入等待状态，造成不必要的上下文切换和系统资源浪费。 ↩︎

  2. 实际上 prolaag 是由 probeer(尝试) 和 verlaag(减少) 合并而来。另外引用一个荷兰学生的话：

    My professor told me he only says Probeer to prevent students asking the same question every week on the meaning of Prolagg :D

    ↩︎

  3. "All problems in computer science can be solved by another level of indirection, except for the problem of too many layers of indirection." - 虽然原话是关于间接层，但其对泛化的警示同样适用。 ↩︎

  4. 西方哲学家用筷子……我看到的版本是刀叉。 ↩︎