Linux内核同步机制与内存排序深度实战：从自旋锁到RCU的无锁世界

2.3 CMPXCHG：无锁编程的基础原语

比较并交换（Compare-And-Swap, CAS）是实现无锁数据结构的基石：

// 内核CAS实现
static inline int atomic_cmpxchg(atomic_t *v, int old, int new)
{
    return cmpxchg(&v->counter, old, new);
}

// x86实现
#define cmpxchg(ptr, old, new) ({                                    \
    __typeof__(ptr) __ptr = (ptr);                                    \
    __typeof__(*ptr) __old = (old);                                   \
    __typeof__(*ptr) __new = (new);                                   \
    asm volatile(LOCK_PREFIX "cmpxchg %2,%1"                         \
                 : "=a" (__old), "+m" (*__ptr)                       \
                 : "r" (__new), "0" (__old)                          \
                 : "memory");                                         \
    __old;                                                            \
})

// 典型用法：无锁引用计数增减
struct refcount_struct {
    atomic_t refs;
};

bool refcount_inc_not_zero(refcount_t *r)
{
    unsigned int new, val = atomic_read(&r->refs);
    do {
        if (unlikely(val == UINT_MAX))
            return false;
        new = val + 1;
    } while (!atomic_try_cmpxchg_relaxed(&r->refs, &val, new));
    return true;
}

CAS操作需要注意ABA问题——即变量值从A变为B再变回A时，CAS无法检测到中间的变化。解决方案通常是使用带标签的指针（tagged pointer）或双宽度CAS。

三、自旋锁：短临界区的首选

3.1 自旋锁的实现原理

自旋锁（Spinlock）是最基本的内核同步原语，当获取锁失败时，CPU会循环忙等待而不是睡眠。自旋锁适用于锁持有时间极短的场景（如修改几个字段的自增、链表头指针交换等）。

// 内核自旋锁结构（无配置PREEMPT_RT时）
typedef struct raw_spinlock {
    arch_spinlock_t raw_lock;
} raw_spinlock_t;

// x86架构的arch_spinlock_t
typedef struct qspinlock {
    union {
        atomic_t val;        // 4字节锁值
        
        struct {
            u8 locked;        // 锁标志位
            u8 pending;       // 待处理标志
        };
        
        struct {
            u16 locked_pending; // 锁+待处理合并
            u16 tail;           // 队列尾部信息
        };
    };
} arch_spinlock_t;

3.2 MCS锁：自旋锁的公平化演进

Linux 4.2引入了排队自旋锁（Queued Spinlock, qspinlock），解决了经典TAS（Test-And-Set）自旋锁在多核竞争时的缓存行乒乓问题：

经典锁的问题：所有等待者在同一个原子变量上自旋，每次CAS成功都会导致其他核心的缓存行失效，产生O(n)总线流量。

MCS排队锁方案：每个等待者在本地的CPU上自旋，通过一个链表排队：

// MCS锁节点
struct mcs_spinlock {
    struct mcs_spinlock *next;    // 链表中下一个等待者
    int locked;                   // 当前持有者是否已释放
    int count;                    // 嵌套计数
};

// qspinlock获取流程
void queued_spin_lock(struct qlock *lock)
{
    u32 val = atomic_xchg(&lock->val, _Q_LOCKED_VAL);  // 尝试快速获取
    
    if (likely(val == 0))
        return;  // 快速路径：直接获得锁
    
    // 慢速路径：加入MCS排队队列
    mcs_spin_lock(&lock->mcs, &node);
    // 等待：locked位为0时获得锁
}

队列机制保证先到先服务（FIFO公平性），避免了饥饿问题。

3.3 自旋锁与中断处理

在中断处理函数中使用自旋锁时，必须禁用本地中断，防止死锁：

// 错误做法（可能死锁）
spin_lock(&my_lock);
/* 如果此时中断到来，同一个CPU再次尝试获取my_lock → 死锁 */

// 正确做法
unsigned long flags;
spin_lock_irqsave(&my_lock, flags);  // 获取锁并保存中断状态，禁用中断
/* 临界区操作 */
spin_unlock_irqrestore(&my_lock, flags);  // 释放锁并恢复中断状态

// 软中断安全版本
spin_lock_bh(&my_lock);   // 获取锁并禁用软中断
/* 临界区操作 */
spin_unlock_bh(&my_lock);  // 释放锁并恢复软中断

3.4 自旋锁性能指标

x86_64平台上自旋锁的延迟特征：

• 无争用时：约50ns（纯原子操作开销）
• 单竞争者：约200-500ns（缓存一致性协议交互）
• 高争用（8+核心）：可达数微秒（排队等待）

因此自旋锁仅适用于锁持有时间在几十到几百纳秒级别的场景，超出应使用信号量或互斥锁。

四、信号量与互斥锁：长等待的选择

4.1 计数信号量

信号量允许一定数量的持有者同时进入临界区，典型的应用场景是资源池限制：

struct semaphore {
    raw_spinlock_t lock;        // 保护count和list
    unsigned int count;         // 可用资源数
    struct list_head wait_list; // 等待队列
};

// 获取信号量（不可中断）
void down(struct semaphore *sem)
{
    unsigned long flags;
    raw_spin_lock_irqsave(&sem->lock, flags);
    if (likely(sem->count > 0)) {
        sem->count--;
        raw_spin_unlock_irqrestore(&sem->lock, flags);
    } else {
        /* 加入等待队列并调度出去 */
        DECLARE_WAITQUEUE(wait, current);
        list_add_tail(&wait.list, &sem->wait_list);
        set_current_state(TASK_UNINTERRUPTIBLE);
        raw_spin_unlock_irqrestore(&sem->lock, flags);
        schedule();  // 让出CPU
    }
}

// 获取信号量（仅当count>0时成功，否则返回失败）
int down_trylock(struct semaphore *sem)
{
    unsigned long flags;
    int count;
    raw_spin_lock_irqsave(&sem->lock, flags);
    count = sem->count - 1;
    if (likely(count >= 0))
        sem->count = count;
    raw_spin_unlock_irqrestore(&sem->lock, flags);
    return count >= 0;
}

4.2 互斥锁：带有优化语义的二值信号量

互斥 locks（struct mutex）是专门优化的二值信号量，相比原始信号量有额外特性：

• 严格互斥：同一时间只能有一个持有者
• 仅持有者释放：防止非持有者释放锁（调试时可检测）
• 不可递归：同一线程不能重复获取
• 快速路径优化：无争用时使用原子操作直接获取，避免加锁开销

struct mutex {
    atomic_long_t owner;       // 持有者线程ID（0表示空闲）
    raw_spinlock_t wait_lock;  // 保护等待队列
    struct list_head wait_list;
};

// mutex_lock快速路径
void __sched mutex_lock(struct mutex *lock)
{
    // 尝试快速路径：原子操作设置owner
    if (likely(__mutex_trylock_fast(lock)))
        return;
    
    // 慢速路径：进入睡眠等待
    __mutex_lock_slowpath(lock);
}

static inline bool __mutex_trylock_fast(struct mutex *lock)
{
    unsigned long curr = (unsigned long)current;
    unsigned long zero = 0;
    
    // 尝试将owner从0原子交换为当前线程ID
    return atomic_long_try_cmpxchg_acquire(&lock->owner, &zero, curr);
}

4.3 互斥锁与信号量的选择指南

五、读写锁：读优先与写优先的权衡

5.1 读写自旋锁

读写锁允许并发读取，但写入时独享。读取操作可以并行，写入操作必须独占：

// 读写锁：1 = 写锁定; 0 = 未锁定; 每读者使计数器加 _Q_READ_OFFSET
// 读者快速路径
static __always_inline void queued_read_lock(struct qrwlock *lock)
{
    atomic_add(_Q_RD_BIAS, &lock->cnts);
    // 快速路径：无写者，直接成功
}

// 写者快速路径
static __always_inline void queued_write_lock(struct qrwlock *lock)
{
    // 快速路径：无读者且无写者
    if (likely(atomic_cmpxchg_acquire(&lock->cnts, 0, _Q_LOCKED_VAL) == 0))
        return;
    // 慢速路径：排队等待
    ...
}

读写锁的固有问题是写者饥饿——持续到来的读者可能导致写者永远等待。为解决这个问题引入了读写信号量（rw_semaphore），支持公平的排队机制。

5.2 读写信号量（rw_semaphore）

rw_semaphore在内核中被大量使用（如mmap_lock内存映射锁、i_rwsem inode读写锁）。Linux 5.19将其优化为MCS排队公平锁：

struct rw_semaphore {
    atomic_long_t count;        // 读者计数和写标志
    struct rwsem_waiter *wait_list;  // 等待队列
    struct task_struct *owner;  // 写者（仅RCU_STARVATIONChecker使用）
};

// count字段编码：
// - 位[30:0]：读者计数
// - RWSEM_FLAG_WAITERS (位31)：有等待者
// - RWSEM_FLAG_HANDOFF (位0)：写者让渡标志

rw_semaphore的关键优化——当有写者在排队时，新读者会被阻塞（而非插队），避免了经典的读者-写者饥饿问题。

六、RCU：无锁读取的艺术

6.1 RCU的核心思想

Read-Copy-Update（RCU）是Linux内核中最精妙的同步机制，它允许多个读者完全不加锁地访问共享数据。其核心思想是：

1. 读取无锁：读者直接访问指针，无需原子操作或内存屏障（在多数架构上）
2. 写时复制：写者复制数据，修改副本，然后原子替换指针
3. 延迟回收：旧数据在所有现有读者完成后才被释放

// RCU读取侧临界区
void read_example(struct my_data *ptr)
{
    rcu_read_lock();       // 开始读取临界区（仅禁用抢占）
    
    ptr = rcu_dereference(global_ptr);  // 安全获取指针
    
    if (ptr)
        printk("value = %d\n", ptr->value);  // 无锁访问
    
    rcu_read_unlock();     // 结束读取临界区
}

// RCU写入侧
void write_example(struct my_data *new_data)
{
    struct my_data *old;
    
    old = rcu_dereference_protected(global_ptr, lockdep_is_held(&my_lock));
    
    // 复制并修改
    new_data = kmemdup(old, sizeof(*old), GFP_KERNEL);
    new_data->value = 42;
    
    // 原子替换指针
    rcu_assign_pointer(global_ptr, new_data);
    
    // 等待所有现有读者完成，然后释放旧数据
    synchronize_rcu();  // 或 kfree_rcu(old, rcu_head) 异步版本
    kfree(old);
}

6.2 宽限期（Grace Period）的实现

RCU的核心挑战是判断何时所有读者都已退出临界区（宽限期结束）。Linux内核采用被动检测策略：

// 宽限期检测：基于静止状态（quiescent state）
// 每个CPU在以下时机处于静止状态：
// 1. 执行上下文切换（说明已退出RCU临界区）
// 2. 进入idle循环
// 3. 执行用户空间代码（用户态不可能持有RCU读锁）

// 宽限期推进
static void rcu_gp_init(struct rcu_state *rsp)
{
    // 1. 记录当前所有CPU的状态（如最近一次静止状态）
    // 2. 等待每个CPU经历一次静止状态
    for_each_possible_cpu(cpu) {
        if (rcu_report_qs_rdp(cpu, rsp)) {
            // 该CPU已报告静止状态
        }
    }
    // 3. 所有CPU都经历了静止状态 → 宽限期结束
    // 4. 调用等待的回调函数（释放旧数据）
}

6.3 RCU的三个主要变体

6.4 无睡眠RCU（Sleepable RCU, SRCU）

经典RCU的rcu_read_lock()禁止抢占，因此不能睡眠。但某些场景需要RCU保护的同时允许睡眠（如读取需要阻塞获取的数据）。synchronize_srcu()解决了这个问题：

// SRCU：读者可以睡眠
void srcu_read_lock(struct srcu_struct *ssp)
{
    int idx = __srcu_read_lock(ssp);  // 获取当前计数器索引
    // 不需要禁用抢占
    // 读者可以睡眠！
}

void srcu_read_unlock(struct srcu_struct *ssp, int idx)
{
    __srcu_read_unlock(ssp, idx);
}

// 等待宽限期
void synchronize_srcu(struct srcu_struct *ssp)
{
    // 1. 阻塞新的读者获取旧计数器
    // 2. 等待现有读者退出
    // 3. 释放旧数据
}

SRCU保证了每个读者看到的是一致的旧版本或新版本，但其开销高于经典RCU（计数器需原子操作）。

七、内存排序：看不见的障碍

7.1 编译期屏障（Compiler Barrier）

编译器可能会对无关的内存操作进行重排优化，barrier()和READ_ONCE()宏防止这种重排：

// barrier()：仅阻止编译器重排，不影响CPU
#define barrier() __asm__ __volatile__("" : : : "memory")

// READ_ONCE()：确保每次读取都从内存获取（非寄存器缓存）
#define READ_ONCE(x) (*(const volatile typeof(x) *)&(x))

// WRITE_ONCE(x, val)：确保写入不被优化掉或重排
#define WRITE_ONCE(x, val) (*(volatile typeof(x) *)&(x) = (val))

// 场景示例：配置硬件寄存器
void configure_reg(void __iomem *base)
{
    u32 ctrl = ioread32(base + REG_CTRL);
    ctrl |= CTRL_ENABLE;    // 使能设备
    iowrite32(ctrl, base + REG_CTRL);
    
    // 在设备启动后设置DMA地址
    iowrite32(dma_addr, base + REG_DMA_ADDR);
    iowrite32(size, base + REG_DMA_SIZE);
    
    // 注意：在x86上IO-MMIO访问天然有序
    // 但在PCIe等总线上可能需要显式写屏障
}

7.2 CPU内存屏障

CPU级别的内存屏障阻止硬件层面的指令重排。x86 vs ARM64屏障对比：

// x86内存屏障（TSO模型，大多数序天然保证）
#define mb()  asm volatile("mfence" : : : "memory")   // 全屏障：LoadLoad|LoadStore|StoreLoad|StoreStore
#define rmb() asm volatile("lfence" : : : "memory")   // 读屏障：LoadLoad|LoadStore
#define wmb() asm volatile("sfence" : : : "memory")   // 写屏障：StoreStore
// 注：x86的读操作不会重排到写之前，因此rmb()实际为空操作

// ARM64内存屏障（弱序模型，显式屏障必须）
#define mb()  asm volatile("dmb sy" : : : "memory")      // 全数据内存屏障
#define rmb() asm volatile("dmb ld" : : : "memory")       // 读内存屏障
#define wmb() asm volatile("dmb st" : : : "memory")       // 写内存屏障
#define dma_mb() asm volatile("dmb osh" : : : "memory")   // 仅Outer Shareable域

// ARM64获取/释放语义（更精细，用于锁的获取与释放）
#define smp_load_acquire(p)  ({ ... ldar ... })   // 加载-获取
#define smp_store_release(p, v) ({ ... stlr ... }) // 存储-释放

7.3 获取-释放语义（Acquire-Release）

Lock获取（acquire）和释放（release）操作隐含的内存序：

• Acquire（获取锁）：之后的所有读写不会被重排到获取操作之前。确保锁内能看到锁释放前的所有修改
• Release（释放锁）：之前的所有读写不会被重排到释放操作之后。确保锁释放前所有修改对下一个获取者可见
• 不是全屏障：Acquire不阻挡后面的Store，Release不阻挡前面的Load

// 发布-订阅模式的实现
struct message {
    int data;
    int ready;
};

struct message *global_msg = NULL;

void producer(void)
{
    struct message *m = kmalloc(sizeof(*m), GFP_KERNEL);
    m->data = 42;
    WRITE_ONCE(m->ready, 1);
    
    // smp_store_release确保m->data=42和m->ready=1的写入
    // 在指针发布之前对其他CPU可见
    smp_store_release(&global_msg, m);
}

void consumer(void)
{
    struct message *m = smp_load_acquire(&global_msg);
    
    if (m) {
        // 由于acquire语义，这里保证能读到data=42
        // 因为data写入在ready之前，ready写入在指针发布之前
        // 指针发布在指针获取之前 → data读取必定为42
        int d = READ_ONCE(m->data);
        printk("data = %d\n", d);
    }
}

八、无锁数据结构实战

8.1 无锁队列（Lock-Free Queue）

经典的M&S无锁队列实现：

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

struct lockless_queue {
    struct node *head;
    struct node *tail;
};

// 初始化哨兵节点
void queue_init(struct lockless_queue *q)
{
    q->head = q->tail = calloc(1, sizeof(struct node));
    q->head->next = NULL;
}

// 入队（无锁，多生产者安全）
void enqueue(struct lockless_queue *q, void *data)
{
    struct node *node = malloc(sizeof(*node));
    node->data = data;
    node->next = NULL;
    
    struct node *t, *next;
    while (true) {
        t = READ_ONCE(q->tail);
        next = READ_ONCE(t->next);
        
        // 验证一致性：tail未变
        if (t == READ_ONCE(q->tail)) {
            if (next == NULL) {
                // 尝试将node链接到尾部
                if (cmpxchg(&t->next, next, node) == next)
                    break;  // 成功
            } else {
                // tail落后了，推进tail
                cmpxchg(&q->tail, t, next);
            }
        }
    }
    // 尝试推进tail（即使失败也没关系，入队者会帮忙）
    cmpxchg(&q->tail, t, node);
}

8.2 内核kref：优化引用计数

引用计数是内核中最常见的模式，但传统的atomic_inc/dec在多核上性能差（缓存行独占）。kref被kobject整合，但更优的实践是：

// 使用percpu_ref实现引用计数（读多写少场景）
struct percpu_ref {
    atomic_long_t count;
    percpu_counter fast;
    refcount_t *data;
    percpu_ref_func_t *release;
};

// 读操作：在本地CPU计数器上增减（零争用）
void percpu_ref_get(struct percpu_ref *ref)
{
    this_cpu_inc(ref->fast);  // 仅本地CPU操作，无缓存行乒乓
}

// 写操作（释放阶段）：合并所有CPU计数器
void percpu_ref_kill(struct percpu_ref *ref)
{
    // 1. 切换模式为"退出中"
    // 2. 等待所有percpu计数器归零
    // 3. 合并总计数
    // 4. 如果计数为0，调用release回调释放资源
}

九、内核同步的调试与检测

9.1 Lockdep：死锁检测器

Linux内核的Lockdep（Lock Dependency Checker）在运行时检测潜在的锁依赖问题：

// Lockdep能检测的死锁场景：
// 1. 经典的ABBA死锁：线程A先锁L1后请求L2，线程B先锁L2后请求L1
// 2. 中断在已持有锁的CPU上尝试获取同一锁
// 3. 锁在递归场景中的重入
// 4. 锁类交叉依赖（跟踪锁类而非实例）

// 启用Lockdep的配置
CONFIG_PROVE_LOCKING=y
CONFIG_DEBUG_LOCK_ALLOC=y
CONFIG_DEBUG_LOCKDEP=y

// Lockdep输出示例：
// [    3.141592] INFO: possible circular locking dependency detected
// [    3.141592] 5.15.0-test #1 Not tainted
// [    3.141592] ------------------------------------------------------
// [    3.141592] swapper/0/0 is trying to acquire lock:
// [    3.141592] ffff888000123456 (&dev->mutex){+.+.}-{3:3}, at: __device_attach+0x3c
// [    3.141592] but task is already holding lock:
// [    3.141592] ffff888000789abc (&driver->lock){+.+.}-{3:3}, at: really_probe+0x1a
// [    3.141592] which lock already depends on the new lock.

9.2 KCSAN：数据竞争检测器

KCSAN（Kernel Concurrency Sanitizer）是编译器插桩实现的数据竞争运行时检测器，能发现未同步的数据访问：

// 启用KCSAN的配置
CONFIG_KCSAN=y

// KCSAN能检测的场景：
// 1. 两个CPU同时写同一内存位置（无锁保护）
// 2. 一个CPU写、另一个CPU读同一内存位置（无锁保护）
// 3. 代码编译时使用 -fsanitize=thread（内核上为选择性插桩）

// KCSAN输出示例：
// [   12.345] BUG: KCSAN: data race in foo_write / bar_read
// [   12.345] write to 0xffff888000a00000 of 4 bytes by task 123 on cpu 2:
// [   12.345]  foo_write+0x5a/0x120
// [   12.345] read to 0xffff888000a00000 of 4 bytes by task 456 on cpu 5:
// [   12.345]  bar_read+0x3c/0x80
// [   12.345] no locks held by either task

十、实战场景：选择合适的同步机制

10.1 内核模块开发同步选择决策树

需要同步？
├── 读写比例？
│   ├── 读极多，写极少 → RCU（rcu_dereference + synchronize_rcu）
│   ├── 读写均衡 → rw_semaphore（避免写饥饿）
│   └── 写多于读 → mutex/semaphore
├── 临界区耗时？
│   ├── 极短（<1μs） → spinlock
│   ├── 短（1-100μs） → mutex
│   └── 可能很长（>100μs或阻塞） → semaphore/mutex
│        （注意：不要持有锁睡眠）
├── 能否睡眠？
│   ├── 不能（中断上下文） → spinlock_irqsave
│   └── 可以（进程上下文） → mutex/semaphore
└── 是否需要等待事件发生？
    ├── 是 → completion（完成量）
    └── 否 → 以上同步机制

10.2 性能对比数据

以下数据基于Intel Xeon 8380处理器，启用超线程，内核5.15：

10.3 典型内核子系统中的同步选择案例

• VFS文件操作：inode使用rwsem保护（i_rwsem）；dentry缓存使用RCU保护查找（srcu/rcu-walk）
• 内存管理：mmap_lock使用rw_semaphore保护内存映射；percpu_ref管理LRU页框回收
• 网络协议栈：连接哈希表使用RCU+spinlock（读端RCU，写端spinlock+RCU延迟释放）
• 调度器：运行队列使用per_cpu（减少全局锁竞争）；负载均衡使用RCU遍历任务组
• 块层：request_queue使用自旋锁保护合并；tag_set使用每CPU变量减少锁争用

十一、内核同步的最新演进

11.1 Mutex的乐观自旋（Optimistic Spinning）

Linux 3.16引入了mutex的乐观自旋机制——获取mutex失败时先自旋一段时间（相当锁持有者正在运行），若持有者已切换出再睡眠。这避免了mutex → 睡眠 → 调度 → 唤醒的短持有场景开销。

11.2 qspinlock与qrwlock

Linux 4.2引入的排队自旋锁和读写字自旋锁已验证显著优于传统ticket锁，尤其在8+核心系统上。排队锁避免了全局缓存行乒乓问题。

11.3 BPF自旋锁

eBPF引入了bpf_spin_lock()，在BPF程序中安全地使用自旋锁保护map元素，扩展了eBPF的表达能力。

11.4 Hazard Pointer

虽未进入主线，但Hazard Pointer方案在许多场景替代RCU可以实现更精确的内存回收。Linux主线已引入refscale（Refcount Scalability测试集）和hrtimer中的HP实现作为评估。

十二、总结

Linux内核的同步机制经过二十余年的发展，形成了一套从简单到复杂、从粗粒度到无锁的完整体系：

• 原子操作是一切同步的基础，理解内存序是内核开发者的必备素养
• 自旋锁适合极短临界区，qspinlock解决了可扩展性问题
• 互斥锁是长临界区场景的主力，优化路径无争用时接近原子操作
• 信号量和读写锁解决多值信号和读写分离需求
• RCU实现了极致的读扩展性，但写者有宽限期延迟代价
• 内存屏障是底层保证，正确使用Acquire-Release语义可以避免全屏障开销
• Lockdep和KCSAN提供了强大的运行时检测能力

选择合适的同步机制需要综合考虑临界区长度、争用程度、读写比例和中断上下文。在性能关键路径上，优先考虑RCU和无锁数据结构；在开发友好性优先的场景下，mutex和rw_semaphore是稳妥的选择。内核同步领域仍在继续演进，特别是随着非易失内存（NVM）和异构计算（GPU/FPGA协处理）的发展，新的同步挑战会不断涌现。