Linux 内核同步机制深度实战:从 Spinlock 到 RCU 的完全并发控制工程指南

一、为什么内核同步是操作系统的"操作系统"

在 Linux 内核中,并发是常态而非例外。无论是多核 SMP 对称多处理、硬件中断抢占、软中断软调度、可睡眠工作队列,还是底层抢占式内核配置,内核代码始终在多种上下文中并行执行。同步机制正是保证数据一致性与系统稳定性的基石,也是驱动开发者最容易犯错、最难调试的领域。

本指南从硬件原子指令讲起,逐层解析 spinlock、rwlock、seqlock、mutex、rw_semaphore、completion、percpu 变量、RCU 与内存屏障的完整原理,每个机制都给出内核源码函数签名、真实驱动示例与最佳实践规则。

二、原子操作与内存屏障 —— 一切同步的根基

2.1 原子变量的本质

现代 CPU 通过硬件级指令保证操作的不可分割性。x86 用LOCK前缀总线锁,ARM 用LDXR/STXR独家监视器,RISC-V 用LR/SC加载保留/条件存储。内核通过atomic_t封装:

// include/linux/atomic.h
typedef struct {
    int counter;
} atomic_t;

// 读写直接在架构层映射
#define atomic_read(v)      READ_ONCE((v)->counter)
#define atomic_set(v, i)    WRITE_ONCE(((v)->counter), (i))

// 操作 API
void atomic_add(int i, atomic_t *v);        // 加
void atomic_sub(int i, atomic_t *v);        // 减
int atomic_inc_return(atomic_t *v);          // 加一并返回
bool atomic_dec_and_test(atomic_t *v);       // 减一并测试是否为0
int atomic_cmpxchg(atomic_t *v, int old, int new); // CAS 比较交换

真实驱动示例 —— 设备打开计数:

static atomic_t device_open_count = ATOMIC_INIT(0);

static int mydev_open(struct inode *inode, struct file *filp)
{
    if (atomic_inc_return(&device_open_count) > 1) {
        atomic_dec(&device_open_count);
        return -EBUSY;  // 只允许单个进程打开
    }
    return 0;
}

static int mydev_release(struct inode *inode, struct file *filp)
{
    atomic_dec(&device_open_count);
    return 0;
}

2.2 为什么需要内存屏障(memory barrier)

编译器和 CPU 会对指令做重排序(reordering)优化。无屏障时:

// CPU 视角                     // 另一CPU视角
a = 1;                          // if (ready == 1)
ready = 1;                      // assert(a == 1); // 可能失败!

内核屏障家族:

屏障保证
smp_mb()全屏障,读写都不穿越
smp_rmb()读屏障,读不穿越
smp_wmb()写屏障,写不穿越
smp_store_release()保证之前所有读写在store之后可见
smp_load_acquire()保证之后的读写在load之后执行

现代内核更推荐acquire/release语义而非裸屏障,因其有更清晰的语义且跨架构高效:

 // RCU链表更新示例
list_rcu = new_head;
mp_store_release(&list_head, new_head); // release写
// ---
p = mp_load_acquire(&list_head);        // acquire读,保证看到完整初始化

三、Spinlock —— 多核互斥的黄金标准

3.1 硬件原理与内核实现

Spinlock 的核心是忙等待(busy-wait):获取失败时循环探测(自旋),不会让出 CPU。这意味着 spinlock 不可在睡眠上下文中使用(持有 spinlock 时不能调用 copy_to_user、kmalloc(GFP_KERNEL) 等可能睡眠的函数)。

内核 spinlock 的内部数据结构依赖于体系架构:

// include/linux/spinlock_types.h — 通用结构
typedef struct raw_spinlock {
    arch_spinlock_t raw_lock;   // 架构层:x86用ticket spinlock,ARM64用qspinlock
} raw_spinlock_t;

typedef struct spinlock {
    struct raw_spinlock rlock;
#ifdef CONFIG_DEBUG_LOCK_ALLOC
    struct dep_map dep_map;
#endif
} spinlock_t;

Linux 4.2 以后 x86_64 采用了 qspinlock (queued spinlock):每个锁竞争者只在自己的 per-cpu MCS 节点上自旋(本地缓存行),而非在全局变量上争抢,大幅降低缓存一致性流量。当核数 > 256 时退化为 test-and-set。

3.2 API 分类与用法

// 1. 基本获取/释放
void spin_lock(spinlock_t *lock);       // 简单获取,关闭内核抢占
void spin_unlock(spinlock_t *lock);

// 2. 保存中断状态并关闭本地中断(中断上下文必用)
void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);
void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags);

// 3. 仅关中断(性能更好但要求你保证上下文)
void spin_lock_irq(spinlock_t *lock);
void spin_unlock_irq(spinlock_t *lock);

// 4. 仅关底部半处理(BH)
void spin_lock_bh(spinlock_t *lock);
void spin_unlock_bh(spinlock_t *lock);

// 5. 非阻塞尝试(中断上下文推荐)
int spin_trylock(spinlock_t *lock);    // 成功返回1,失败返回0

如何选择变体?

  • 如果被中断处理函数或软中断访问 → spin_lock_irqsave(同时保护中断和进程上下文)
  • 如果只被进程上下文(多个CPU)访问 → spin_lock
  • 下半部(BH)tasklet 共享 → spin_lock_bh
  • 需要非阻塞获取(中断处理中的fallback逻辑) → spin_trylock

3.3 完整字符设备驱动示例

#include 
#include 

struct my_device {
    spinlock_t          lock;
    struct circ_buf     ring;
    wait_queue_head_t   read_wq;
};

/* 中断处理——必须 irqsave 变体 */
static irqreturn_t mydev_irq_handler(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;
    unsigned long flags;
    u32 data = read_reg(dev->reg_base + DATA_REG);

    spin_lock_irqsave(&dev->lock, flags);
    circ_buf_write(&dev->ring, data);
    spin_unlock_irqrestore(&dev->lock, flags);

    wake_up_interruptible(&dev->read_wq);
    return IRQ_HANDLED;
}

/* 进程上下文读取 */
static ssize_t mydev_read(struct file *filp, char __user *buf,
                           size_t count, loff_t *pos)
{
    struct my_device *dev = filp->private_data;
    unsigned long flags;
    ssize_t ret = 0;
    u8 tmp[64];

    if (wait_event_interruptible(dev->read_wq,
           !circ_buf_empty(&dev->ring)))
        return -ERESTARTSYS;

    spin_lock_irqsave(&dev->lock, flags);
    ret = circ_buf_to_user(&dev->ring, tmp, min(count, sizeof(tmp)));
    spin_unlock_irqrestore(&dev->lock, flags);

    if (copy_to_user(buf, tmp, ret))
        return -EFAULT;
    return ret;
}

四、读写 Spinlock (rwlock) —— 多读者单写者的自旋版

rwlock 分为读者和写者多个读者可并发,写者需独占等待所有读者退出。适用于读多写少的场景(如设备列表、路由表查询)。但注意 rwlock 有读者饥饿风险:高频写者可能被读者持续阻塞。

rwlock_t my_rwlock = __RW_LOCK_UNLOCKED(my_rwlock);

read_lock(&my_rwlock);    // 获取读锁
read_unlock(&my_rwlock);  // 释放读锁

write_lock(&my_rwlock);   // 获取写锁
write_unlock(&my_rwlock); // 释放写锁

// 非阻塞尝试
read_trylock(&my_rwlock);
write_trylock(&my_rwlock);

内核真实用例 —— netdevice 操作表读写锁:

// net/core/dev.c
static DEFINE_RWLOCK(dev_base_lock);

// 遍历设备列表(读端,高频)
read_lock(&dev_base_lock);
for_each_netdev_rcu(net, dev) { ... }
read_unlock(&dev_base_lock);

// 注册设备(写端,低频)
write_lock(&dev_base_lock);
list_add_tail_rcu(&dev->list, &dev_base_head);
write_unlock(&dev_base_lock);

五、Seqlock —— 无锁读的科学

Seqlock(Sequential Lock)由 Stephen Tweedie 提出,专为读极多写极少、少量数据设计。读端完全无锁(对读者零开销),写端自旋,读者通过序列号检测冲突。

// 写端
write_seqlock(&seq_lock);
/* 临界区写操作 */
write_sequnlock(&seq_lock);

// 读端
unsigned int seq;
do {
    seq = read_seqbegin(&seq_lock);
    /* 临界区读操作 */
} while (read_seqretry(&seq_lock, seq));

jiffies 获取就是经典的 seqlock 应用:

// kernel/time/timekeeping.c
u64 get_jiffies_64(void)
{
    unsigned int seq;
    u64 ret;
    do {
        seq = read_seqbegin(&jiffies_lock);
        ret = jiffies_64;
    } while (read_seqretry(&jiffies_lock, seq));
    return ret;
}

限制:不能保护指针内容。因为读者读指针时写者可能释放,读者访问已释放内存导致 use-after-free。对含指针的数据应使用 RCU。

六、Mutex —— 可睡眠的进程同步之王

6.1 为什么 mutex 比 semaphore 更好

Mutex 是 kernel 4.x 后推荐的睡眠互斥锁,相比传统 semaphore:

  • 严格互斥(count = 1),不做计数信号量
  • 严格 owner 语义(谁持有谁释放,不允许递归解锁)
  • DEBUG 模式开启死锁检测(lockdep 集成)
  • 比 semaphore 更快的慢路径(fast path 直接 CAS)
// 初始化
struct mutex my_mutex = __MUTEX_INITIALIZER(my_mutex);
// 或
mutex_init(&my_mutex);

// 获取/释放(可能睡眠)
mutex_lock(&my_mutex);
/* 临界区——可以睡眠、可以copy_to_user */
mutex_unlock(&my_mutex);

// 尝试获取(不睡眠)
if (!mutex_trylock(&my_mutex))
    return -EBUSY;

// 带超时获取
mutex_lock_interruptible(&my_mutex);    // 可中断(推荐)
mutex_lock_killable(&my_mutex);          // 仅致命信号可中断

与 spinlock 选择规则:

场景选 spinlock选 mutex
可能睡眠❌✅
非原子上下文❌✅
中断上下文✅❌
临界区短且不可睡眠✅✅(轻微开销)
临界区长或需拷贝❌✅

6.2 独占等待——优化 contention

Linux 6.1+ 引入了mutex_optimistic_spin优化:mutex 被持有时,高优先级 waiter 乐观自旋等待(类似 qspinlock),避免立即睡眠的上下文切换开销。

七、读写信号量 (rw_semaphore) —— 可睡眠的读者写者锁

static DECLARE_RWSEM(my_rwsem);

// 读端(可并发)
down_read(&my_rwsem);
up_read(&my_rwsem);

// 写端(独占)
down_write(&my_rwsem);
up_write(&my_rwsem);

// 尝试版本
down_read_trylock(&my_rwsem);
down_write_trylock(&my_rwsem);

// 可中断版本
down_read_interruptible(&my_rwsem);

内核用例 —— VFS inode 操作: 文件读路径持有 inode 的 i_rwsem,写路径持有独占锁,允许并发读同一文件偏移的更新互不干扰。

自 kernel 5.13 起,rw_semaphore 同样支持 optimistic spinning。在 CONFIG_PREEMPT_RT 实时内核中被映射为 rt_mutex 以完全消除优先级反转。

八、Completion —— 线程间的一击唤醒

Completion 并非互斥机制,而是事件通知:一个线程等待某个事件完成,另一个线程触发通知。常用于驱动初始化同步、模块卸载等待 I/O 结束。

DECLARE_COMPLETION(my_completion);
// 或
init_completion(&my_completion);

// 等待(多种变体)
wait_for_completion(&my_completion);            // 不可中断
wait_for_completion_interruptible(&my_completion);
wait_for_completion_timeout(&my_completion, HZ);
wait_for_completion_killable(&my_completion);

// 触发完成(唤醒所有等待者)
complete(&my_completion);

// 只唤醒一个等待者
complete_all(&my_completion);

设备初始化同步示例:

static DECLARE_COMPLETION(firmware_loaded);

static int firmware_wait_thread(void *data)
{
    wait_for_completion_interruptible(&firmware_loaded);
    /* 加载完成后继续初始化... */
    return 0;
}

static irqreturn_t firmware_irq(int irq, void *dev)
{
    complete(&firmware_loaded);
    return IRQ_HANDLED;
}

九、Per-CPU 变量 —— 用空间换无锁

Per-CPU 变量是每个 CPU 拥有独立副本的全局变量,访问天然不需要同步(每个 CPU 只改自己的副本)。但在读取其他 CPU 副本时需要抢占保护。

// 静态定义
DEFINE_PER_CPU(u64, my_counter);
// 动态分配
alloc_percpu(u64);

// 访问自己的副本(无需同步)
get_cpu_var(my_counter)++;
// 或 API 形式
this_cpu_inc(my_counter);           // 原子增量
__this_cpu_add(my_counter, val);    // 非原子(仅你独占时使用)

// 安全读取其他 CPU 副本
per_cpu_sum = 0;
for_each_possible_cpu(cpu)
    per_cpu_sum += per_cpu(my_counter, cpu);

// 带抢占保护的读其他 CPU
u64 val;
preempt_disable();
val = per_cpu(my_counter, target_cpu);
preempt_enable();

内核热点案例: vm_event_state(VM 事件统计)、网络协议栈的 CPU 局部统计(softnet_data)、slab 分配器的 CPU 缓存。

十、Read-Copy-Update (RCU) —— 读端零开销的终极答案

10.1 核心原理

RCU 由 Paul McKenney 发明,适用读极多写极少场景。核心思想:

  1. 读端完全无锁:rcu_read_lock()/rcu_read_unlock()在抢占内核中是抢占关闭,无抢占内核中是空操作
  2. 写端不原地更新:copy 出副本→修改副本→原子替换指针→等待Grace Period→释放旧数据
// 读端(零开销!)
rcu_read_lock();
p = rcu_dereference(head);  // 安全取指针
if (p)
    do_something(p->data);
rcu_read_unlock();

// 写端
new_node = kmalloc(sizeof(*new_node), GFP_KERNEL);
new_node->data = value;
new_node->next = head->next;

// 原子替换——读者可能看到新旧两个版本
rcu_assign_pointer(head->next, new_node);

// 等待所有已有读者退出Grace Period后释放旧结点
synchronize_rcu();            // 同步版(睡眠等待)
call_rcu(&old_node->rcu_head, free_node_callback); // 异步版(推荐)

10.2 Grace Period 与 Quiescent State

Grace Period(GP)定义:从写端替换指针开始,到所有读端在替换前进入的 rcu_read_lock() 区间都已退出。内核通过检测每个 CPU 都经过一个 Quiescent State(QS,通常为上下文切换、用户态执行或空闲)来判断 GP 结束。

call_rcu 流程:写者将回调注册到 per-cpu 链表 → 调度器/定时器中断检测到该 CPU 已过渡 QS → 所有 CPU 都经过 QS 后触发 callback → 在软中断或 worker 线程中执行 callback。

10.3 RCU 在驱动中的真实模式

struct dev_priv {
    struct rcu_head rcu;
    u32             reg_val;
    char            name[32];
};

static struct dev_priv __rcu *g_dev_priv;  // RCU 保护的指针

// 读端(如中断中、数据读)
static u32 read_reg_val(void)
{
    struct dev_priv *priv;
    u32 val;
    rcu_read_lock();
    priv = rcu_dereference(g_dev_priv);
    val = priv->reg_val;
    rcu_read_unlock();
    return val;
}

// 写端(如IOCTL配置)
static int update_priv(struct dev_priv *new_priv)
{
    struct dev_priv *old = rcu_dereference_protected(g_dev_priv,
                              lockdep_is_held(&my_mutex));
    // 原子替换——读者零阻塞
    rcu_assign_pointer(g_dev_priv, new_priv);
    // 异步释放旧数据——经过GP后自动调用 free
    call_rcu(&old->rcu, free_old_priv);
    return 0;
}

static void free_old_priv(struct rcu_head *rcu)
{
    struct dev_priv *old = container_of(rcu, struct dev_priv, rcu);
    kfree(old);
}

十一、Lockdep —— 内核的运行时锁验证器

Lockdep 是 Linux 最独特的子系统之一。它在运行时构建 lockdep map(锁类图),检测以下问题:

  • 死锁(Deadlock):A→B 与 B→A 环路
  • 中断反转:关中断锁在不关中断上下文被获取
  • 递归锁:同一线程对非递归锁两次获取
  • 虚假 irq-safe:spin_lock 而非 spin_lock_irqsave 在中断路径共用
  • 锁类混淆
// 为每个锁类注册独立map(避免不同语义锁合并)
static struct lock_class_key my_lock_key;
spin_lock_init(&my_lock);    // 首次初始化自动注册

开启方式:CONFIG_PROVE_LOCKING=y。触发死锁时打印完整的 acquire-after-release 顺序 warn 信息。

十二、同步机制选择决策树


临界区是否可能睡眠(copy_to_user/kmalloc(GFP_KERNEL)/mutex_lock)?
  ├── 否 ── 是否在中断上下文?
  │         ├── 是 ──▶ spin_lock_irqsave
  │         └── 否 ──▶ 临界区极短?
  │                   ├── 是 ──▶ spin_lock
  │                   └── 否(仍可能睡眠?)▶ 用 mutex 更省事
  └── 是 ──▶ 是否读多写少?
            ├── 是 ──▶ 读者极高频且无指针?
            │         ├── 是 ──▶ seqlock
            │         └── 否 ──▶ RCU(读端零开销最佳)
            └── 否 ──▶ 需要多个读者并发?
                      ├── 是 ──▶ rw_semaphore
                      └── 否 ──▶ mutex

十三、生产级最佳实践

  1. 最小临界区原则:只在锁保护下访问共享数据;数据拷贝等耗时操作在锁外完成
  2. 锁顺序一致性:系统中所有锁形成偏序关系,按固定顺序获取
  3. 中断路径优先 irqsave:不明确时永远选 spin_lock_irqsave
  4. 能 RCU 就 RCU:读端无锁是性能最优解(lkml 传统智慧)
  5. 避免信号量:用 mutex 替代二进制 semaphore:更好的语义和调试
  6. percpu 是第一优化:能用 percpu 解决就别加锁
  7. 启用 lockdep 调试:所有新开发的驱动必须经过 lockdep 过一遍
  8. 用 might_sleep() 标注:临界区调用的 API 必须可能睡眠

十四、总结

Linux 内核同步机制并非"越高级越好",而是根据并发模型精确匹配。Spinlock 是中断上下文与极短临界区的硬通货;mutex 是进程上下文可睡眠场景的万金油;rwlock/rw_semaphore 让多读者解放并发;seqlock 用序列号消除读端开销;percpu 用副本打破共享、用空间换零阻塞;RCU 则将读端优化推向极致。

掌握这六个同步原语,加上它们与内存屏障、lockdep 的深度配合,就掌握了驱动与子系统开发的半壁江山。

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论
立即
投稿

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部
0.374940s