M4: 同步互斥与死锁

// 共享变量
int turn = 0;  // 谁的回合

// 进程 0
while (true) {
    while (turn != 0);  // 忙等待
    // 临界区
    turn = 1;
    // 剩余区
}

// 进程 1
while (true) {
    while (turn != 1);  // 忙等待
    // 临界区
    turn = 0;
    // 剩余区
}

bool flag[2] = {false, false};

// 进程 i
while (true) {
    while (flag[j]);     // 等待对方出来
    flag[i] = true;       // 表示想进入
    // 临界区
    flag[i] = false;
    // 剩余区
}

#include <stdatomic.h>

atomic_bool flag[2] = {false, false};
atomic_int turn;

void enter_critical_section(int i) {
    int j = 1 - i;
    atomic_store_explicit(&flag[i], true, memory_order_relaxed);
    atomic_store_explicit(&turn, j, memory_order_release); // 确保 flag 写入在 turn 之前
    
    // 全局屏障, 强制同步内存视图
    atomic_thread_fence(memory_order_seq_cst); 
    
    while (atomic_load_explicit(&flag[j], memory_order_acquire) && 
           atomic_load_explicit(&turn, memory_order_acquire) == j) {
        // busy wait
    }
}

void leave_critical_section(int i) {
    atomic_store_explicit(&flag[i], false, memory_order_release);
}

假设两进程都在临界区:
- flag[0] = flag[1] = true
- turn 只能是 0 或 1 (不能同时为两个值)
- 若 turn = 0, 则进程 1 在 while 循环等待
- 若 turn = 1, 则进程 0 在 while 循环等待
矛盾! 故最多一个进程在临界区.

// 编译器或 CPU 可能重排这两条指令
flag[i] = true;   // (1)
turn = j;         // (2)

// 如果 (2) 在 (1) 之前执行, 互斥可能被破坏

void enter_critical_section(int i) {
    int j = 1 - i;
    flag[i] = true;
    __sync_synchronize();  // 内存屏障
    turn = j;
    __sync_synchronize();
    while (flag[j] && turn == j);
}

// 生产者-消费者 (错误版本, 有 lost wakeup 问题)
#define N 100
int count = 0;

void producer() {
    while (true) {
        item = produce();
        if (count == N)
            sleep();       // 缓冲区满, 睡眠等待
        insert(item);
        count++;
        if (count == 1)
            wakeup(consumer);  // 唤醒消费者
    }
}

void consumer() {
    while (true) {
        if (count == 0)
            sleep();       // 缓冲区空, 睡眠等待
        item = remove();
        count--;
        if (count == N - 1)
            wakeup(producer);  // 唤醒生产者
        consume(item);
    }
}

1. count = 0
2. 消费者检查 count == 0, 准备睡眠
3. 调度器切换到生产者
4. 生产者生产一项, count = 1
5. 生产者调用 wakeup(consumer), 但消费者还没睡
6. wakeup 信号丢失!
7. 消费者睡眠
8. 生产者继续生产直到满, 然后睡眠
9. 两者都在睡眠 → 死锁

// CAS 语义
bool CAS(int *addr, int expected, int new_val) {
    if (*addr == expected) {
        *addr = new_val;
        return true;
    }
    return false;
}

// x86 内联汇编
static inline bool cas(volatile int *addr, int expected, int new_val) {
    char result;
    __asm__ __volatile__ (
        "lock cmpxchgl %3, %1\n\t"
        "sete %0"
        : "=q"(result), "+m"(*addr), "+a"(expected)
        : "r"(new_val)
        : "memory", "cc"
    );
    return result;
}

// Linux 内核中的屏障
smp_mb();    // 全屏障
smp_rmb();   // 读屏障
smp_wmb();   // 写屏障

CPU 0 执行 LOCK CMPXCHG:
1. 获取缓存行独占权 (MESI Exclusive/Modified)
2. 锁定总线或使用缓存锁定协议
3. 执行 CAS 操作
4. 释放锁定

其他 CPU 对该缓存行的访问被阻塞 → 性能开销

typedef struct {
    volatile int locked;
} spinlock_t;

void spin_lock(spinlock_t *lock) {
    while (__sync_lock_test_and_set(&lock->locked, 1)) {
        // 忙等待 (自旋)
        while (lock->locked) {
            __builtin_ia32_pause();  // CPU hint: 减少功耗
        }
    }
}

void spin_unlock(spinlock_t *lock) {
    __sync_lock_release(&lock->locked);
}

typedef struct {
    volatile unsigned int next;   // 下一个票号
    volatile unsigned int owner;  // 当前服务的票号
} ticket_lock_t;

void ticket_lock(ticket_lock_t *lock) {
    unsigned int my_ticket = __sync_fetch_and_add(&lock->next, 1);
    while (lock->owner != my_ticket) {
        __builtin_ia32_pause();
    }
}

void ticket_unlock(ticket_lock_t *lock) {
    lock->owner++;
}

// include/linux/spinlock_types.h
typedef struct raw_spinlock {
    arch_spinlock_t raw_lock;
} raw_spinlock_t;

// 使用
spinlock_t lock;
spin_lock_init(&lock);

spin_lock(&lock);
// 临界区
spin_unlock(&lock);

### 3.4 Spinlock 公平性演进

#### 3.4.1 基础 Spinlock (不公平)

```c
// 测试并设置 (TAS)
void spin_lock(spinlock_t *lock) {
    while (test_and_set(&lock->locked))
        ;  // 自旋
}

struct ticket_lock {
    atomic_t next_ticket;  // 下一个可取号码
    atomic_t now_serving;  // 当前服务号码
};

void spin_lock(struct ticket_lock *lock) {
    int my_ticket = atomic_fetch_add(&lock->next_ticket, 1);
    while (atomic_load(&lock->now_serving) != my_ticket)
        cpu_relax();  // 自旋
}

**不同变体**:

| API | 作用 |
|-----|:---:|
| `spin_lock()` | 获取锁 |
| `spin_lock_irq()` | 获取锁并关闭本地中断 |
| `spin_lock_irqsave()` | 获取锁并保存/关闭中断状态 |
| `spin_lock_bh()` | 获取锁并关闭下半部 |

---

## 4. 信号量 (Semaphore)

### 4.1 信号量本质

**信号量 (Semaphore)** 是一个整型变量, 支持两个原子操作:

| 操作 | 荷兰语 | 英文 | 行为 |
|:---:|--------|:---:|------|
| P | Proberen (尝试) | wait / down | `S--`; 若 `S < 0` 则阻塞 |
| V | Verhogen (增加) | signal / up | `S++`; 若有等待者则唤醒 |

### 4.2 信号量类型

| 类型 | 初始值 | 用途 |
|:---:|--------|:---:|
| **二值信号量** | 1 | 互斥锁替代品 |
| **计数信号量** | N | 资源池管理 (如连接池) |
| **同步信号量** | 0 | 线程间同步/通知 |

### 4.3 信号量核心用途

```mermaid
graph TB
    subgraph SemUses ["信号量作为基石"]
        direction LR
        Mutex[1. 互斥<br/>sem=1<br/>保护临界区]
        Sync[2. 同步<br/>sem=0<br/>A先B后]
        Count[3. 计数<br/>sem=N<br/>资源池控制]
    end

typedef struct {
    int value;                    // 资源计数
    struct list_head wait_list;   // 等待队列
    spinlock_t lock;              // 保护信号量本身
} semaphore_t;

void P(semaphore_t *sem) {        // Wait / Down
    spin_lock(&sem->lock);
    sem->value--;
    if (sem->value < 0) {
        // 加入等待队列并睡眠
        add_to_wait_list(&sem->wait_list, current);
        spin_unlock(&sem->lock);
        schedule();               // 让出 CPU
    } else {
        spin_unlock(&sem->lock);
    }
}

void V(semaphore_t *sem) {        // Signal / Up
    spin_lock(&sem->lock);
    sem->value++;
    if (sem->value <= 0) {
        // 唤醒一个等待者
        struct task_struct *p = remove_from_wait_list(&sem->wait_list);
        wake_up_process(p);
    }
    spin_unlock(&sem->lock);
}

semaphore empty = N;    // 缓冲区空槽位
semaphore full = 0;     // 缓冲区已填充槽位
semaphore mutex = 1;    // 互斥访问缓冲区
// ... 代码实现 ...

#define N 100
semaphore_t mutex = {1};      // 互斥访问缓冲区
semaphore_t empty = {N};      // 空槽位
semaphore_t full = {0};       // 满槽位

void producer() {
    while (true) {
        item = produce();
        P(&empty);            // 等待空槽
        P(&mutex);            // 进入临界区
        insert(item);
        V(&mutex);            // 离开临界区
        V(&full);             // 增加满槽
    }
}

void consumer() {
    while (true) {
        P(&full);             // 等待满槽
        P(&mutex);            // 进入临界区
        item = remove();
        V(&mutex);            // 离开临界区
        V(&empty);            // 增加空槽
        consume(item);
    }
}

semaphore_t mutex = {1};       // 保护 read_count
semaphore_t wrt = {1};         // 写互斥
int read_count = 0;

void reader() {
    P(&mutex);
    read_count++;
    if (read_count == 1)       // 第一个读者
        P(&wrt);               // 阻止写者
    V(&mutex);
    
    // 读操作
    
    P(&mutex);
    read_count--;
    if (read_count == 0)       // 最后一个读者
        V(&wrt);               // 允许写者
    V(&mutex);
}

void writer() {
    P(&wrt);
    // 写操作
    V(&wrt);
}

typedef struct {
    int locked;                   // 0=未锁定, 1=已锁定
    int contended;                // 是否有竞争
} mutex_t;

void mutex_lock(mutex_t *m) {
    // 快速路径: 尝试 CAS 获取锁
    if (CAS(&m->locked, 0, 1))
        return;  // 成功
    
    // 慢速路径: 进入内核等待
    m->contended = 1;
    while (!CAS(&m->locked, 0, 1)) {
        futex_wait(&m->locked, 1);  // 调用内核
    }
}

void mutex_unlock(mutex_t *m) {
    m->locked = 0;
    if (m->contended) {
        m->contended = 0;
        futex_wake(&m->locked, 1);  // 唤醒一个等待者
    }
}

#include <linux/futex.h>
#include <sys/syscall.h>

// 等待
syscall(SYS_futex, uaddr, FUTEX_WAIT, val, timeout, NULL, 0);
// 如果 *uaddr == val, 则睡眠

// 唤醒
syscall(SYS_futex, uaddr, FUTEX_WAKE, n, NULL, NULL, 0);
// 唤醒最多 n 个等待者

// 必须使用 while, 因为条件可能在被唤醒后变化
while (!condition) {
    pthread_cond_wait(&cond, &mutex);
}

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int ready = 0;

// 等待者
pthread_mutex_lock(&mutex);
while (!ready) {                                // 循环检查条件
    pthread_cond_wait(&cond, &mutex);           // 原子释放锁并等待
}
// 条件满足, 继续执行
pthread_mutex_unlock(&mutex);

// 通知者
pthread_mutex_lock(&mutex);
ready = 1;
pthread_cond_signal(&cond);                     // 或 pthread_cond_broadcast
pthread_mutex_unlock(&mutex);

优先级: H > M > L

1. L 获取锁 lock
2. H 尝试获取 lock, 被阻塞
3. M 抢占 L (M 优先级高于 L)
4. H 等待 M 完成后 L 才能释放锁

结果: 高优先级 H 被中优先级 M 间接阻塞

// kernel/locking/rtmutex.c
// 实现优先级继承

// 当高优先级任务等待锁时
static int rt_mutex_adjust_prio_chain(...) {
    // 遍历等待链
    // 将锁持有者的优先级提升到等待者中最高的
    task->prio = min(task->prio, waiter->prio);
}

Available = [3,3,2]  // 当前可用

        Max       Allocation    Need
P0    [7,5,3]      [0,1,0]    [7,4,3]
P1    [3,2,2]      [2,0,0]    [1,2,2]  ← 可满足
P2    [9,0,2]      [3,0,2]    [6,0,0]
P3    [2,2,2]      [2,1,1]    [0,1,1]  ← 可满足
P4    [4,3,3]      [0,0,2]    [4,3,1]

安全序列: <P1, P3, P4, P2, P0>

int Available[m];           // 可用资源向量
int Max[n][m];              // 最大需求矩阵
int Allocation[n][m];       // 已分配矩阵
int Need[n][m];             // 尚需矩阵 (Need = Max - Allocation)

bool is_safe() {
    int Work[m];
    bool Finish[n];
    
    // 初始化
    memcpy(Work, Available, sizeof(Work));
    memset(Finish, false, sizeof(Finish));
    
    int count = 0;
    while (count < n) {
        bool found = false;
        for (int i = 0; i < n; i++) {
            if (!Finish[i] && need_le_work(Need[i], Work)) {
                // 模拟进程 i 完成
                for (int j = 0; j < m; j++)
                    Work[j] += Allocation[i][j];
                Finish[i] = true;
                found = true;
                count++;
            }
        }
        if (!found) break;  // 无法继续
    }
    
    return (count == n);  // 所有进程都能完成则安全
}

进程   Max     Allocation   Need    Available
      A B C     A B C       A B C    A B C
P0    7 5 3     0 1 0       7 4 3    3 3 2
P1    3 2 2     2 0 0       1 2 2
P2    9 0 2     3 0 2       6 0 0
P3    2 2 2     2 1 1       0 1 1
P4    4 3 3     0 0 2       4 3 1

安全序列: `<P1, P3, P4, P0, P2>` (存在至少一个)

// 读侧 (无锁, 极低开销)
rcu_read_lock();
p = rcu_dereference(global_ptr);
// 使用 p 读取数据
rcu_read_unlock();

// 写侧
new = kmalloc(sizeof(*new), GFP_KERNEL);
*new = /* 新数据 */;
old = global_ptr;
rcu_assign_pointer(global_ptr, new);  // 原子发布新版本
synchronize_rcu();                     // 等待所有读者完成
kfree(old);                            // 安全释放旧版本

写者发布新版本
     │
     ↓
─────┼──────────────────────────────────────────
     │     ← 正在进行的读操作 →      │
─────┼──────────────────────────────────────────
                                     │
                            Grace Period 结束
                                     │
                                     ↓
                              可以安全释放旧数据

seqlock_t lock;

// 写者 (独占)
write_seqlock(&lock);
// 修改数据
write_sequnlock(&lock);

// 读者 (可能重试)
unsigned int seq;
do {
    seq = read_seqbegin(&lock);
    // 读取数据
} while (read_seqretry(&lock, seq));

// 内核配置
CONFIG_LOCKDEP=y
CONFIG_PROVE_LOCKING=y   // 更严格的检测

// 发生警告时的输出示例:
// ======================================================
// WARNING: possible circular locking dependency detected
// ======================================================
// swapper/0/1 is trying to acquire lock:
// (&rq->lock){-.-.}
// 
// but task is already holding lock:
// (&p->pi_lock){-.-.}
// 
// which lock already depends on the new lock.

// 同一锁类的不同实例
spinlock_t lock1, lock2;  // 同类

// 如果需要区分, 使用 lockdep_set_class
lockdep_set_class(&lock1, &key1);
lockdep_set_class(&lock2, &key2);

// 源代码
x = 1;      // 写 x
r1 = y;     // 读 y

// CPU 可能执行为:
r1 = y;     // 读 y (先执行)
x = 1;      // 写 x (后执行)

// 生产者 (Producer)
store_value = new_data;
smp_store_release(&flag, 1);  // 所有之前的写操作对消费者可见

// 消费者 (Consumer)
if (smp_load_acquire(&flag)) {    // 如果看到 flag=1
    use(store_value);              // 保证能看到 new_data
}

// litmus 测试文件示例 (MP - Message Passing)
C MP+porelease+poacquire

{
  x = 0; y = 0;
}

P0(int *x, int *y) {
  WRITE_ONCE(*x, 1);
  smp_store_release(y, 1);
}

P1(int *x, int *y) {
  int r0;
  int r1;
  r0 = smp_load_acquire(y);
  r1 = READ_ONCE(*x);
}

exists (1:r0=1 /\ 1:r1=0)  // 预期: 不可能

#include <stdatomic.h>

atomic_int counter = ATOMIC_VAR_INIT(0);

atomic_fetch_add(&counter, 1);  // 原子加
atomic_load(&counter);          // 原子读
atomic_store(&counter, 0);      // 原子写

struct node {
    void *data;
    struct node *next;
};

struct lockfree_stack {
    _Atomic(struct node *) head;
};

void push(struct lockfree_stack *s, void *data) {
    struct node *new_node = malloc(sizeof(*new_node));
    new_node->data = data;
    
    struct node *old_head;
    do {
        old_head = atomic_load(&s->head);
        new_node->next = old_head;
    } while (!atomic_compare_exchange_weak(&s->head, &old_head, new_node));
}

void *pop(struct lockfree_stack *s) {
    struct node *old_head;
    do {
        old_head = atomic_load(&s->head);
        if (old_head == NULL) return NULL;
    } while (!atomic_compare_exchange_weak(&s->head, &old_head, old_head->next));
    
    void *data = old_head->data;
    free(old_head);
    return data;
}

semaphore fork[5];  // 初始值均为 1

void philosopher(int i) {
    while (true) {
        think();
        P(&fork[i]);           // 拿起左边的叉子
        P(&fork[(i+1) % 5]);   // 拿起右边的叉子
        eat();
        V(&fork[(i+1) % 5]);   // 放下右边的叉子
        V(&fork[i]);           // 放下左边的叉子
    }
}

semaphore fork[5];   // 初始值均为 1
semaphore room = 4;  // 最多 4 人同时就餐

void philosopher(int i) {
    while (true) {
        think();
        P(&room);              // 进入餐厅
        P(&fork[i]);
        P(&fork[(i+1) % 5]);
        eat();
        V(&fork[(i+1) % 5]);
        V(&fork[i]);
        V(&room);              // 离开餐厅
    }
}

void philosopher(int i) {
    while (true) {
        think();
        if (i % 2 == 0) {
            P(&fork[i]);           // 偶数哲学家先拿左边
            P(&fork[(i+1) % 5]);
        } else {
            P(&fork[(i+1) % 5]);   // 奇数哲学家先拿右边
            P(&fork[i]);
        }
        eat();
        V(&fork[(i+1) % 5]);
        V(&fork[i]);
    }
}

monitor dining_philosophers {
    enum {THINKING, HUNGRY, EATING} state[5];
    condition self[5];
    
    void pickup(int i) {
        state[i] = HUNGRY;
        test(i);
        if (state[i] != EATING)
            self[i].wait();
    }
    
    void putdown(int i) {
        state[i] = THINKING;
        test((i + 4) % 5);  // 测试左邻居
        test((i + 1) % 5);  // 测试右邻居
    }
    
    void test(int i) {
        if (state[(i+4) % 5] != EATING &&
            state[i] == HUNGRY &&
            state[(i+1) % 5] != EATING) {
            state[i] = EATING;
            self[i].signal();
        }
    }
}