linux 内核开发涉及的同步方式 分析

作者

在Linux内核开发中，由于内核运行在多处理器环境且需要处理中断、进程切换等并发场景，因此需要各种同步机制来保护共享资源。

首先，我们用一个表格来快速概览这些同步机制的特点：

同步机制核心思想是否可睡眠主要使用场景
原子操作 (Atomic Ops)不可中断的单条指令操作否对整型变量进行计数、设置标志位等简单操作。
自旋锁 (Spinlock)忙等待（Spinning），等待者持续循环检查锁否临界区非常短，且在中断上下文中必须使用。
互斥体 (Mutex)睡眠等待，等待者放弃 CPU 进入睡眠是临界区较长，或者临界区内可能发生阻塞（如 I/O）。
信号量 (Semaphore)睡眠等待，允许多个持有者（计数信号量）是控制对一组资源的访问，或用于任务间的同步。
读写锁 (RWLocks)允许多个读者或一个写者否（自旋锁版本）/是（信号量版本）读多写少，读操作远多于写操作的场景。
RCU (Read-Copy-Update)读者无锁，写者复制、修改、延迟释放否（读者侧）读操作极其频繁，写操作相对较少，且对读性能要求极高。
顺序锁 (Seqlocks)读者通过序列号判断数据有效性，写者优先否读多写少，数据简短，且要求写者延迟极低。
完成量 (Completions)一个任务等待另一个任务完成某项工作的信号是任务间简单的“完成”事件通知。

1. 原子操作 (Atomic Operations)

原子操作是所有同步机制中最基础的一种。它能确保对一个整型变量的操作是“原子的”，即在执行过程中不会被其他任务或中断打断。

  • 核心思想：利用特殊的 CPU 指令（如 lock; add）实现不可分割的读-改-写操作。这是最轻量级的同步方式。
  • 涉及接口：
  • 定义与初始化:
    • atomic_t v; 或 atomic64_t v;：定义原子变量。
    • atomic_set(atomic_t *v, int i);：设置原子变量 v 的值为 i。
    • ATOMIC_INIT(i)：在定义时静态初始化原子变量。例如：atomic_t my_counter = ATOMIC_INIT(0);
  • 操作接口:
    • atomic_read(const atomic_t *v);：读取原子变量的值。
    • atomic_add(int i, atomic_t *v);：给 v 增加 i。
    • atomic_sub(int i, atomic_t *v);：从 v 减去 i。
    • atomic_inc(atomic_t *v);：v 自增 1。
    • atomic_dec(atomic_t *v);：v 自减 1。
  • 测试并操作:
    • int atomic_inc_and_test(atomic_t *v);：v 自增 1，如果结果为 0，则返回 true。
    • int atomic_dec_and_test(atomic_t *v);：v 自减 1，如果结果为 0，则返回 true。常用于引用计数。
    • int atomic_sub_and_test(int i, atomic_t *v);：从 v 减去 i，如果结果为 0，则返回 true。
    • atomic_cmpxchg(atomic_t *v, int old, int new);：比较并交换（Compare-and-Swap, CAS）。如果 v 的值等于 old，则将其设置为 new，并返回 old；否则，直接返回 v 的当前值。
  • 使用方式：
  // 1. 定义和初始化一个引用计数器
  atomic_t ref_count;
  atomic_set(&ref_count, 1);

  // 2. 增加引用计数
  void get_reference(void) {
      atomic_inc(&ref_count);
  }

  // 3. 减少引用计数，并在计数为0时释放资源
  void put_reference(void) {
      if (atomic_dec_and_test(&ref_count)) {
          // 引用计数变为0，可以安全释放资源了
          free_the_resource();
      }
  }
  • 使用场景：
  • 引用计数：最经典的应用，用于跟踪对象被引用的次数。
  • 实现简单标志位：例如，一个标志位表示设备是否已打开。
  • 统计计数：例如，网络驱动中收发包的计数器。
  • 注意：原子操作只能保护单个整型变量，不能保护复杂的数据结构或多个变量。

2. 自旋锁 (Spinlock)

当一个任务试图获取一个已经被占用的自旋锁时，它不会进入睡眠，而是会“自旋”（在一个循环中反复检查锁），直到锁被释放。

  • 核心思想：忙等待。为了避免死锁，在持有自旋锁的期间，当前 CPU 的抢占会被禁用。
  • 涉及接口:
  • 定义与初始化:
    • spinlock_t my_lock;：定义自旋锁。
    • spin_lock_init(&my_lock);：动态初始化。
    • DEFINE_SPINLOCK(my_lock);：静态定义并初始化。
  • 加锁/解锁:
    • spin_lock(spinlock_t *lock);：获取锁，如果锁已被占用则自旋。
    • spin_unlock(spinlock_t *lock);：释放锁。
  • 中断安全版本:
    • spin_lock_irqsave(spinlock_t *lock, unsigned long flags);：在加锁前禁用本地中断并保存当前中断状态。
    • spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags);：解锁并恢复之前的中断状态。这是中断处理程序中最常用的版本。
    • spin_lock_irq(spinlock_t *lock);: 禁用中断并加锁。
    • spin_unlock_irq(spinlock_t *lock);: 解锁并使能中断。
  • 其他:
    • spin_trylock(spinlock_t *lock);：尝试获取锁，如果无法立即获取则返回 false，不自旋。
    • spin_lock_bh(spinlock_t *lock);：在加锁前禁用软中断（Bottom Half）。
  • 使用方式:
  DEFINE_SPINLOCK(list_lock);
  struct list_head my_list;

  void add_to_list(struct my_data *data) {
      unsigned long flags;

      // 使用 irqsave 版本防止中断和本地 CPU 抢占
      spin_lock_irqsave(&list_lock, flags);

      // --- 临界区开始 ---
      list_add(&data->list, &my_list);
      // --- 临界区结束 ---

      spin_unlock_irqrestore(&list_lock, flags);
  }
  • 使用场景:
  • 中断上下文：中断处理程序中必须使用自旋锁，因为中断上下文不能睡眠。
  • 临界区极短：当锁的持有时间非常短（通常是几条指令）时，自旋的开销比上下文切换（睡眠）的开销要小。
  • 保护 SMP 系统中的共享数据：防止不同 CPU 同时访问同一数据。
  • 警告：持有自旋锁的代码绝对不能调用任何可能导致睡眠的函数（如 kmalloc, copy_from_user, I/O 操作等），否则会导致系统死锁。

3. 互斥体 (Mutex)

互斥体是一种“睡眠锁”。当一个任务试图获取一个已经被占用的互斥体时，它会被放入一个等待队列并进入睡眠状态，将 CPU 让给其他任务。

  • 核心思想：睡眠等待。开销比自旋锁大，但不会浪费 CPU 周期。
  • 涉及接口:
  • 定义与初始化:
    • struct mutex my_mutex;：定义互斥体。
    • mutex_init(&my_mutex);：动态初始化。
    • DEFINE_MUTEX(my_mutex);：静态定义并初始化。
  • 加锁/解锁:
    • mutex_lock(struct mutex *lock);：获取锁，如果被占用则睡眠。
    • mutex_unlock(struct mutex *lock);：释放锁。
  • 其他版本:
    • mutex_lock_interruptible(struct mutex *lock);：尝试获取锁，但如果进程在睡眠时收到信号，会中断等待并返回 -EINTR。
    • mutex_trylock(struct mutex *lock);：尝试获取锁，如果无法立即获取则返回 false，不睡眠。
  • 使用方式:
  DEFINE_MUTEX(my_device_mutex);

  ssize_t my_device_write(struct file *filp, const char __user *buf, size_t count, loff_t *f_pos) {
      // 获取锁，如果其他进程正在写入，则当前进程会睡眠
      if (mutex_lock_interruptible(&my_device_mutex)) {
          return -ERESTARTSYS;
      }

      // --- 临界区开始 ---
      // 这里可以执行可能阻塞的操作，例如 copy_from_user
      if (copy_from_user(..., buf, ...)) {
          // ...
      }
      // --- 临界区结束 ---

      mutex_unlock(&my_device_mutex);

      return count;
  }
  • 使用场景:
  • 临界区较长：当锁的持有时间可能很长时。
  • 临界区内有阻塞操作：当临界区代码需要调用 kmalloc、copy_from_user/copy_to_user 或进行文件/网络 I/O 时，必须使用互斥体。
  • 进程上下文：只能在进程上下文中使用，绝对不能在中断上下文（硬中断、软中断、tasklet）中使用，因为它们不能睡眠。

4. 信号量 (Semaphore)

信号量是一个更广义的锁，它内部有一个计数器。当计数器大于 0 时，任务可以获取信号量（并将计数器减 1）。当计数器为 0 时，任务必须等待。

  • 核心思想：计数和睡眠等待。当计数为 1 时，其行为类似互斥体（称为二值信号量）；当计数大于 1 时，允许多个任务同时进入临界区（称为计数信号量）。
  • 涉及接口:
  • 定义与初始化:
    • struct semaphore my_sem;：定义信号量。
    • sema_init(&my_sem, int count);：动态初始化，count 是初始计数值。
  • 获取/释放:
    • void down(struct semaphore *sem);：获取信号量（P 操作），计数器减 1。如果计数器为 0，则睡眠。
    • void up(struct semaphore *sem);：释放信号量（V 操作），计数器加 1，并唤醒一个等待者（如果有）。
  • 其他版本:
    • int down_interruptible(struct semaphore *sem);：可被信号中断的 down 操作。
    • int down_trylock(struct semaphore *sem);：非阻塞的 down 操作。
  • 使用方式：
  // 假设我们有一个包含3个可用资源的资源池
  struct semaphore resource_pool_sem;
  sema_init(&resource_pool_sem, 3);

  void* get_resource(void) {
      // 请求一个资源，如果3个都已被占用，则睡眠
      if (down_interruptible(&resource_pool_sem)) {
          return NULL; // 被中断
      }

      // 成功获取，从池中取出一个资源
      return pool_get();
  }

  void release_resource(void* res) {
      // 将资源放回池中
      pool_put(res);

      // 增加可用资源计数，并唤醒一个等待者
      up(&resource_pool_sem);
  }
  • 使用场景:
  • 控制对有限资源的访问：如上例中的资源池。
  • 任务同步：一个任务可以通过 up 操作来通知另一个正在 down 操作上等待的任务某件事已经完成。
  • 注意：对于简单的互斥锁功能，内核社区现在更推荐使用互斥体（Mutex），因为它的实现更轻量，且有更严格的使用规则和调试支持（例如，死锁检测）。

5. 读写锁 (Read-Write Lock)

读写锁允许多个读者并发访问，但只允许一个写者独占访问。

  • 核心思想：“读共享，写独占”。
  • 涉及接口：
  • 定义与初始化:
    • rwlock_t my_rwlock;
    • rwlock_init(&my_rwlock);
    • DEFINE_RWLOCK(my_rwlock);
  • 加锁/解锁 (自旋锁版本):
    • void read_lock(rwlock_t *lock);
    • void read_unlock(rwlock_t *lock);
    • void write_lock(rwlock_t *lock);
    • void write_unlock(rwlock_t *lock);
    • 同样有 _irqsave, _irq, _bh 等变体。
  • 使用方式:
  DEFINE_RWLOCK(config_lock);
  struct my_config global_config;

  // 读者
  void read_config_value(void) {
      read_lock(&config_lock);
      // --- 临界区 ---
      // 读取 global_config 的值
      // ---
      read_unlock(&config_lock);
  }

  // 写者
  void update_config(struct my_config *new_config) {
      write_lock_irq(&config_lock); // 通常写操作需要更强的保护
      // --- 临界区 ---
      // 更新 global_config
      // ---
      write_unlock_irq(&config_lock);
  }
  • 使用场景:
  • 读多写少：当数据被读取的频率远高于被修改的频率时，使用读写锁可以显著提高并发性能。例如，一个路由表或系统配置信息，它被频繁查询，但很少被修改。

6. RCU (Read-Copy-Update)

RCU 是一种非常高级且高效的同步机制，专为读多写少的场景优化。它的核心是让读者几乎无开销地访问数据，而写者通过“复制-更新”的方式工作。

  • 核心思想：
  • 读者：不加锁，仅通过 rcu_read_lock/unlock 声明一个 RCU 读侧临界区，这通常只是禁用/启用抢占，开销极小。
  • 写者：
    1. 复制：创建一个要修改数据的副本。
    2. 更新：在副本上进行修改。
    3. 发布：使用 rcu_assign_pointer 将指针原子地指向新副本。
    4. 等待：等待一个“宽限期”（Grace Period）结束。这个宽限期保证了所有在发布新版本之前进入 RCU 读临界区的读者都已退出。
    5. 释放：宽限期结束后，可以安全地释放旧数据的内存。
  • 涉及接口:
  • 读者侧:
    • rcu_read_lock();：标记 RCU 读临界区开始。
    • rcu_read_unlock();：标记 RCU 读临界区结束。
    • rcu_dereference(p);：在 RCU 读临界区内安全地读取受 RCU 保护的指针。
  • 写者侧:
    • rcu_assign_pointer(p, new_p);：发布新指针。
    • synchronize_rcu();：同步等待，阻塞直到一个宽限期结束。
    • call_rcu(struct rcu_head *head, rcu_callback_t func);：异步方式，在宽限期结束后，通过回调函数 func 来处理旧数据的释放。
  • 使用方式:
  struct foo {
      int a;
      struct rcu_head rcu;
  };
  struct foo __rcu *g_foo_ptr; // RCU保护的全局指针

  // 读者
  void reader_func(void) {
      struct foo *p;
      rcu_read_lock();
      p = rcu_dereference(g_foo_ptr);
      if (p) {
          // 使用 p->a
          printk("Value: %d\n", p->a);
      }
      rcu_read_unlock();
  }

  // 写者
  void writer_func(int new_val) {
      struct foo *new_p = kmalloc(sizeof(*new_p), GFP_KERNEL);
      struct foo *old_p;

      new_p->a = new_val;
      old_p = rcu_dereference_protected(g_foo_ptr, lockdep_is_held(...)); // 在写者侧获取旧指针

      rcu_assign_pointer(g_foo_ptr, new_p); // 发布新版本

      if (old_p) {
          // 在宽限期后释放旧数据
          call_rcu(&old_p->rcu, free_foo_callback);
      }
  }

  void free_foo_callback(struct rcu_head *head) {
      struct foo *p = container_of(head, struct foo, rcu);
      kfree(p);
  }
  • 使用场景:
  • 性能要求极高的读密集型数据：例如，内核的网络路由表、文件系统的 dentry 缓存。
  • 读者不能被阻塞：RCU 保证了读者的无锁、无阻塞访问。
  • 注意：RCU 的写者端逻辑比其他锁复杂得多，且只适用于通过指针访问的数据结构。

总结与选择

机制等待方式可否睡眠使用上下文核心场景
原子操作不等待否任何上下文简单的计数器、标志位
自旋锁忙等待否任何上下文 (主要用于中断)保护极短的、不能睡眠的临界区
互斥锁睡眠等待是仅进程上下文保护可能睡眠的、较长的临界区
信号量睡眠等待是仅进程上下文控制对有限个资源的并发访问
读写锁忙等待/睡眠否/是相应上下文读多写少的场景，提升读并发
RCU读不等待，写者等待是任何上下文 (读者)读操作远多于写，追求极致读性能

选择原则：

  1. 能用原子操作就不用锁。
  2. 在进程上下文，如果临界区可能睡眠，用互斥锁（Mutex）。
  3. 在中断上下文或持有自旋锁时，绝对不能用互斥锁。
  4. 如果临界区很短且不睡眠，用自旋锁（Spinlock）。
  5. 如果数据是“读多写少”，考虑用读写锁或 RCU。
  6. 如果读性能是首要瓶颈，且可以接受更新延迟，RCU是最佳选择。

本文版权归原作者zhaofujian所有，采用 CC BY-NC-ND 4.0 协议进行许可，转载请注明出处。

发表评论