从零写OS内核 | 同步原语——spinlock与信号量，操作系统是怎么协调并发访问的

两个进程同时往同一个文件里写数据——进程 A 写前 100 字节，进程 B 写后 100 字节。如果不做任何协调，会发生什么？

进程 A 刚读到一个”当前写入位置”，还没来得及写，进程 B 也读到了同一个位置——两个进程都以为自己应该写在这里。结果是数据交错、文件损坏。这就是竞态条件（Race Condition）——并发访问共享资源时，由于执行顺序的不确定性，导致结果错误。

同步原语（Synchronization Primitives）就是用来解决这个问题的：让并发访问变成串行的、有序的。今天，我们来搞清楚 spinlock 和信号量是怎么实现的。

1. 竞态条件：问题的本质

Bash
竞态条件示例：

进程 A 和进程 B 共享变量 `counter = 0`，都想执行 counter++

正确的执行顺序（A 和 B 各执行一次）：
  counter = 0
  A: read counter → 0
  A: counter++   → 1
  A: write counter → 1
  B: read counter → 1
  B: counter++   → 2
  B: write counter → 2
  最终 counter = 2 ✓

但如果执行顺序错乱：
  A: read counter → 0
  B: read counter → 0   ← A 还没写，B 就读了
  A: counter++   → 1
  A: write counter → 1
  B: counter++   → 1   ← B 基于旧值 0 加的
  B: write counter → 1
  最终 counter = 1 ✗  ← 错误！两个进程各加了一次，结果只加了 1

问题出在哪里？read-modify-write 不是原子操作。要解决这个问题，必须保证同一时间只有一个进程能执行这段代码——这就是互斥（Mutual Exclusion）。

2. spinlock：忙等的互斥锁

spinlock 是最简单的互斥原语——如果锁被占用，CPU 就一直自旋（while loop），直到拿到锁。

Bash
spinlock 原理：

struct spinlock {
    int locked;  // 0=未占用，1=占用
}

void lock(struct spinlock *lock) {
    // 如果锁被占用，自旋等待
    while (atomic_test_and_set(&lock->locked, 1) == 1) {
        // 空循环，不断检查锁是否释放
        // 期间 CPU 不能做别的事（busy wait）
    }
    // 拿到了！locked = 1，退出循环
}

void unlock(struct spinlock *lock) {
    // 释放锁
    atomic_clear(&lock->locked, 0);
}

2.1 atomic_test_and_set：硬件保证的原子性

为什么 while (x == 1) 这样的检查不能保证原子性？因为检查和设置之间可能被其他 CPU 打断。

atomic_test_and_set 是一种 CPU 指令级别的原子操作：

Asm
# x86 lock 指令前缀（让下一条指令原子化）
lock xchg %eax, (lock_addr)  # 原子交换，返回旧值

# 在 C 语言里对应：
bool test_and_set(int *ptr) {
    bool old = *ptr;
    *ptr = 1;
    return old;
}

lock 前缀让 CPU 在执行 xchg 时锁定总线（或者用 cache coherency 机制），保证这个操作不被其他 CPU 打断——这就是原子性的来源。

2.2 spinlock 的问题：多核下的缓存行抖动

Bash
多核 CPU 上的 spinlock 性能问题：

CPU0 和 CPU1 共享同一个 lock 变量的缓存行：

CPU0: lock = 0 → 设为 1（拿到锁）→ 缓存行变为 M（Modified，独占）
CPU1: 读取 lock → 发现缓存行是 M → CPU1 需要等待 CPU0 释放
       → 触发 cache coherency protocol（MSI/MESI）的总线事务
CPU0: 释放 lock = 0 → 缓存行变为 I（Invalid）
CPU1: 收到 invalidate → 缓存行变为 S（Shared）→ 可以再次读取

问题：当很多 CPU 同时竞争同一个 spinlock 时，
     所有 CPU 都在同一个缓存行上自旋，
     缓存行的 invalidate/upgrade 流量会成为瓶颈
     → 这叫 "cache line ping-ponging"

解决方案：用每核独立的 flag + CPU pause 指令：

C
// 改进的 MCS spinlock（每个 CPU 有一个本地节点）
struct mcs_lock_node {
    struct mcs_lock_node *next;  // 指向下一个等待者
    int waiting;                 // 本地标志
};

void lock_mcs(struct mcs_lock *lock, struct mcs_lock_node *local) {
    local->next = NULL;
    local->waiting = 1;

    // 原子地把本地节点放到锁的尾部，返回前一个持有者
    struct mcs_lock_node *prev = atomic_swap(&lock->tail, local);
    if (prev != NULL) {
        // 锁被占用，把本地节点接到前一个持有者后面
        prev->next = local;
        // 自旋等待，直到 waiting 变为 0
        while (local->waiting == 1) {
            cpu_relax();  // PAUSE 指令，减少缓存冲突
        }
    }
    // 拿到了！
}

void unlock_mcs(struct mcs_lock *lock, struct mcs_lock_node *local) {
    if (local->next == NULL) {
        // 自己是最后一个，检查是否还有人排在自己后面
        if (atomic_compare_exchange(&lock->tail, local, NULL) == SUCCESS) {
            // 没人排在自己后面，锁已释放
            return;
        }
        // 有人在排队，等它设置 next
        while (local->next == NULL) {
            cpu_relax();
        }
    }
    // 通知下一个等待者
    local->next->waiting = 0;
}

3. 信号量：允许有限并发访问的锁

spinlock 的问题是：如果锁被占用，CPU 只能空转。如果持锁的时间很长（比如等 I/O），这完全浪费 CPU。

信号量（Semaphore） 解决这个问题——不让 CPU 空等，而是让进程进入睡眠（阻塞），等锁可用时再唤醒。

Bash
信号量 vs spinlock：

spinlock：自旋，进程一直在 CPU 上跑（不能去睡觉）
  → 适合持锁时间很短的情况（< 1ms）

信号量：睡眠，进程让出 CPU（去等待队列睡觉）
  → 适合持锁时间较长的情况（I/O、等待外设等）

信号量 = 一个整数值（计数器）+ 一个等待队列

3.1 信号量实现

C
struct semaphore {
    int count;                    // 计数器（资源数量）
    struct list_head waiters;     // 等待队列（FIFO）
    spinlock_t lock;               // 保护信号量本身
};

void sem_init(struct semaphore *sem, int count) {
    sem->count = count;
    list_init(&sem->waiters);
    spinlock_init(&sem->lock);
}

void sem_down(struct semaphore *sem) {
    // P 操作（proberen，荷兰语"测试"）
    spinlock_lock(&sem->lock);
    if (sem->count > 0) {
        // 有资源，占用一个
        sem->count--;
        spinlock_unlock(&sem->lock);
    } else {
        // 没有资源，当前进程进入睡眠
        struct process *current = get_current_process();
        list_add_tail(&current->wait_link, &sem->waiters);
        set_process_state(current, STATE_BLOCKED);
        spinlock_unlock(&sem->lock);
        // 主动让出 CPU（调度到其他进程）
        schedule();
    }
}

void sem_up(struct semaphore *sem) {
    // V 操作（verhogen，荷兰语"增加"）
    spinlock_lock(&sem->lock);
    if (list_empty(&sem->waiters)) {
        // 没有进程在等，增加计数器
        sem->count++;
    } else {
        // 有进程在等，唤醒一个
        struct process *waiter = list_entry(sem->waiters.next,
                                           struct process, wait_link);
        list_del(&waiter->wait_link);
        set_process_state(waiter, STATE_RUNNABLE);
        // 被唤醒的进程会在下次调度时拿到信号量
    }
    spinlock_unlock(&sem->lock);
}

3.2 二值信号量（互斥锁）

当信号量初始值 = 1 时，它退化为一个互斥锁（Mutex）：

C
// 用信号量实现互斥锁
struct semaphore mutex;
sem_init(&mutex, 1);  // count=1，同时只有一个进程能进入

// 访问临界区
sem_down(&mutex);      // 获取锁（count--），如果 count=0 则睡眠
// 临界区代码
sem_up(&mutex);        // 释放锁（count++），唤醒一个等待者

这和 spinlock 的区别：spinlock 如果拿不到会自旋，mutex 如果拿不到会睡眠。睡眠避免了无意义的 CPU 消耗。

4. 互斥锁 vs spinlock：什么时候用哪个

Bash
选择原则：

持锁时间 < 1ms（临界区很短）：
  → 用 spinlock（睡眠+唤醒的开销 > 持锁时间）

持锁时间 > 1ms（临界区有 I/O）：
  → 用 mutex（让 CPU 做别的事）

中断处理函数里：
  → 用 spinlock（中断处理函数不能睡眠）
  → 但要尽可能快进快出，不要在中断里长时间持锁

多核环境：
  → spinlock 可能因缓存行竞争而性能差
  → 考虑 MCS lock 或 per-CPU 数据结构

5. 死锁：同步原语的最大陷阱

死锁（Deadlock）是最常见的同步错误——两个或多个进程互相等待对方持有的锁，谁都无法推进。

Bash
死锁示例（两把锁）：

进程 A：
  lock(lock1);       // 拿到 lock1
  lock(lock2);       // 等待 lock2

进程 B：
  lock(lock2);       // 拿到 lock2
  lock(lock1);       // 等待 lock1

结果：A 等 B 的 lock2，B 等 A 的 lock1，谁也拿不到

Bash
避免死锁的策略：

1. 锁的顺序：所有进程必须按相同顺序获取锁
   A: lock1 → lock2
   B: lock1 → lock2    ← 同一顺序，不会死锁

2. try-lock（不要阻塞式拿锁）
   if (try_lock(lock1) == FAIL) {
       unlock(lock2);  // 回退已拿的锁
       retry();
   }

3. 锁的层级检测（运行时）
   检测 lock graph 是否有环
   → Linux 的 lockdep 可以检测这个

4. 避免循环等待
   → 如果能确保所有进程都先拿 lock1 再拿 lock2，
     就不会有循环等待

6. Linux 实践：观察锁竞争

Bash
# 用 perf 观察锁争用
perf lock record -a sleep 10
perf lock report

# 显示所有锁的：
# 名称、持有时间、争用次数、等待时间

# 观察 spinlock 争用
perf stat -e 'lock:*' ./my_program

# 查看当前进程持有的锁
cat /proc/$$/status | grep -i lock

Bash
# 用 strace 观察 futex 系统调用（用户态锁的实现）
# futex(ADDR, FUTEX_WAIT, ...) = 等锁
# futex(ADDR, FUTEX_WAKE, ...) = 唤醒
strace -e trace=futex ./my_program

# 观察 futex 争用情况
cat /proc/sys/kernel/futex_timeout  # futex 等待超时

Bash
# 用 /proc/lock_stat 看全局锁统计（需要 root）
echo 1 > /proc/sys/kernel/lock_stat
# 运行程序
cat /proc/lock_stat
# 显示：
# 锁名 | 持有时间 | 争用次数 | 平均等待时间

7. 从零实现：wandos 的同步原语

⚠️ wandos 当前状态：kernel/sync/ 目录下有 spinlock 和 semaphore 实现。以下分析其实现。

7.1 spinlock 实现（kernel/sync/spinlock.cpp）

Cpp
// wandos spinlock 实现
// 源码：kernel/sync/spinlock.cpp

struct spinlock {
    volatile int value;   // 0=未占用，1=占用
};

// 原子 test-and-set（使用 x86 lock 前缀）
static inline bool atomic_test_and_set(volatile int *addr, int new_val) {
    // x86 的原子交换：返回旧值，设置新值
    // "lock" 前缀确保总线锁定，多核安全
    int old;
    __asm__ volatile(
        "lock xchg %0, %1"
        : "=r"(old), "+m"(*addr)
        : "0"(new_val)
        : "memory"
    );
    return old;
}

void spin_lock(spinlock_t *lock) {
    // 自旋等待，直到成功获取锁
    while (atomic_test_and_set(&lock->value, 1) == 1) {
        // 空循环，CPU 忙等
        // 可加 PAUSE 指令减少缓存竞争
        __asm__ volatile("pause" ::: "memory");
    }
}

void spin_unlock(spinlock_t *lock) {
    // 释放锁
    __asm__ volatile(
        "movl $0, %0"
        : "+m"(*lock)
        :
        : "memory"
    );
}

7.2 mutex 实现（基于信号量）

Cpp
// wandos mutex 实现（基于 semaphore）
// 源码：kernel/sync/mutex.cpp

struct mutex {
    struct semaphore sem;   // 用二值信号量实现
    struct process *owner; // 当前持有者（用于调试/递归检测）
};

void mutex_init(struct mutex *m) {
    sem_init(&m->sem, 1);  // count=1，二值
    m->owner = NULL;
}

void mutex_lock(struct mutex *m) {
    // 当前进程不能重复拿自己已经持有的锁（简单检测）
    if (m->owner == get_current_process()) {
        panic("recursive mutex lock!");
    }

    sem_down(&m->sem);     // P 操作
    m->owner = get_current_process();
}

void mutex_unlock(struct mutex *m) {
    m->owner = NULL;
    sem_up(&m->sem);       // V 操作，唤醒等待者
}

7.3 等待队列（用于阻塞）

Cpp
// wandos 等待队列（实现信号量的等待机制）
// 源码：kernel/sync/waitqueue.cpp

struct wait_queue {
    struct list_head tasks;  // 等待任务链表
    spinlock_t lock;
};

void wait_queue_init(struct wait_queue *wq) {
    list_init(&wq->tasks);
    spinlock_init(&wq->lock);
}

void wait_on(struct wait_queue *wq) {
    struct process *current = get_current_process();

    spinlock_lock(&wq->lock);
    // 加入等待队列
    list_add_tail(&current->wait_link, &wq->tasks);
    // 设置为阻塞状态
    set_process_state(current, TASK_BLOCKED);
    spinlock_unlock(&wq->lock);

    // 主动调度，让出 CPU
    schedule();
}

void wake_up(struct wait_queue *wq) {
    spinlock_lock(&wq->lock);

    if (!list_empty(&wq->tasks)) {
        // 取队列头的第一个进程唤醒
        struct process *waiter = list_entry(wq->tasks.next,
                                            struct process, wait_link);
        list_del(&waiter->wait_link);
        set_process_state(waiter, TASK_RUNNABLE);
    }

    spinlock_unlock(&wq->lock);
}

7.4 wandos 和 Linux 的主要差异

  1. 没有 RCU（Read-Copy-Update）：Linux 有 RCU 让读操作完全无锁，wandos 没有这层
  2. 没有 lockdep：Linux 的 lockdep 可以检测死锁和锁顺序问题，wandos 只有简单的递归检测
  3. 没有 per-CPU 变量：Linux 的 percpu 机制让每个 CPU 有独立数据，避免缓存行竞争，wandos 没有
  4. 没有 sleepable spinlock：Linux 可以在禁止抢占的情况下睡眠（sleepable RCU），wandos 不支持

8. 动手环节：实现一个生产者-消费者问题

今天的目标：用信号量实现一个环形 buffer 的生产者-消费者问题，验证同步原语的正确性。

任务 1：实现环形 buffer

C
#define BUFFER_SIZE 8
int buffer[BUFFER_SIZE];
int in = 0;   // 生产者写入位置
int out = 0;  // 消费者读取位置

任务 2：用信号量控制

C
struct semaphore mutex;    // 互斥锁（保护 buffer）
struct semaphore empty;   // 空槽数量（初始 = BUFFER_SIZE）
struct semaphore full;    // 已填充数量（初始 = 0）

任务 3：生产者/消费者

C
// 生产者：生产 100 个 item
void producer(void) {
    for (int i = 0; i < 100; i++) {
        sem_down(&empty);           // 等空槽
        sem_down(&mutex);           // 互斥
        buffer[in] = i;
        in = (in + 1) % BUFFER_SIZE;
        sem_up(&mutex);
        sem_up(&full);              // 通知有数据
    }
}

// 消费者：消费 100 个 item
void consumer(void) {
    for (int i = 0; i < 100; i++) {
        sem_down(&full);            // 等数据
        sem_down(&mutex);
        int val = buffer[out];
        out = (out + 1) % BUFFER_SIZE;
        sem_up(&mutex);
        sem_up(&empty);             // 通知有空槽
        // 处理 val
    }
}

验收标准

  • 生产者和消费者并发运行，不丢失数据，不重复读取
  • 用 mutex 确保 buffer 读写不会交错（验证输出完整性）
  • 可以用两个线程（生产者/消费者）运行，观察结果

9. 踩坑与注意事项

坑 1：持锁时调用 sleep

在持有 spinlock 时调用 schedule() 或任何可能让进程睡眠的操作——这是灾难性的。持有 spinlock 意味着”我正在保护某个数据结构”，如果进程在保护期间睡觉，另一个进程可能试图获取同一个 spinlock，然后自旋到天荒地老（死锁）。Spinlock 只能在禁止抢占的上下文中使用（中断处理函数、禁止调度的临界区）。

坑 2：信号量 count 不匹配初始化

初始化信号量时，count 的值必须正确反映资源的数量。如果 sem_init(&mutex, 2)（count=2 的信号量当互斥锁用），会导致两个进程同时进入临界区。Mutex 必须是 count=1，二值信号量。

坑 3：唤醒后继续持有锁

sem_up 唤醒等待者后，等待者被加入就绪队列，但不立即拿到信号量——它只是被唤醒，等调度器选中后才能继续。如果 sem_up 后立即把锁变量设为 0，而唤醒的进程还没被调度，其他进程可能先拿到锁。正确的实现：唤醒者通过 wake_up 把进程设为 RUNNABLE，锁的状态由唤醒者（sem_up）设置，唤醒的进程在 sem_down 里再检查。

坑 4：锁粒度太大

如果一个锁保护了太多数据（比如整个进程表），所有进程访问任何进程信息都要竞争这把锁——系统变成单线程，性能灾难。解决：细粒度锁（每条链表一把锁）、读写锁（读多写少场景）、per-CPU 数据结构（根本不需要锁）。

写在最后

同步原语是并发安全的基石：spinlock 用硬件原子指令保证互斥，适合短临界区；信号量用睡眠/唤醒机制让 CPU 不空等，适合长临界区。两者配合，构成了操作系统并发控制的核心工具。

理解同步原语，你才能真正理解为什么多线程程序要小心翼翼地保护共享数据，以及为什么 Linux 内核里满地都是 spinlock——因为内核代码永远是并发的，不保护就有问题。

下篇预告：从零写OS内核系列阶段性总结——我们覆盖了哪些、还缺什么。

相关阅读

  • Linux Kernel Source: kernel/locking/spinlock.c（spinlock 实现）
  • Linux Kernel Source: kernel/locking/semaphore.c（信号量实现）
  • Linux Kernel Source: kernel/locking/mutex.c（互斥锁）
  • wandos: kernel/sync/spinlock.cpp
  • wandos: kernel/sync/mutex.cpp
  • wandos: kernel/sync/waitqueue.cpp
  • 本文 Demo: https://github.com/golang12306/os-kernel-from-scratch （demos/synchronization/）
  • https://github.com/zhangfuwen/wandos — Linux 内核教程
  • 下一篇：《从零写OS内核 | 系列总结——我们覆盖了哪些，还缺什么》

动手环节

想深入理解本文内容？动手实践是最好的方式：

今天的目标：下载 wandos 代码仓库，实现生产者-消费者问题，理解 Linux vs wandos 的差异。

  1. 下载 wandos：

    Bash
    git clone https://github.com/zhangfuwen/wandos.git
cd wandos

  2. 找到对应模块：查看 kernel/sync/ 下的同步原语实现

  3. 实现作业：根据文中”动手环节”章节的要求，完成代码编写

  4. 提交作业：Fork 仓库，提交你的改动，在 GitHub 上开一个 Pull Request

仓库：https://github.com/golang12306/os-kernel-from-scratch

最后修改： 2024年2月4日

作者

评论

发表评论

您的邮箱地址不会被公开。