Linux 内核 RCU 深度实战:从基础原理到生产级应用的完全工程指南

RCU (Read-Copy-Update) 是 Linux 内核中最精妙的同步机制之一。它在读多写少场景下实现零开销读取,通过延迟回收保证数据一致性。本文从硬件内存模型说起,完整拆解 RCU 的核心原理、宽限期机制、内核 API 使用、经典陷阱与性能调优,结合内核源码级分析和生产环境最佳实践。

一、为什么需要 RCU?

在现代多核系统中,读多写少的数据结构无处不在:路由表、文件描述符表、进程描述符链表、网络连接跟踪表、文件系统中 dentry 缓存。这些数据的读取频率可能是写入的上万倍。

传统同步机制的困境:

机制 读者开销 写者开销 读写互斥 适用场景
互斥锁 (mutex) 高(原子操作+缓存同步) 高 是 通用读写信号量
rwlock 中(原子递增) 高 是(写优先) 读多写少(已废弃)
rwsem 中 高 是 读多写少(可睡眠)
seqlock 极低(读-重试) 中 否(读写并行) 少量数据的频繁读
RCU 极低(无原子操作) 中(内存屏障+延迟回收) 否(读写并行) 读多写少的复杂结构

RCU 的核心优势:读者无需任何原子操作或内存屏障,在 x86 上读取 RCU 保护的指针就是一个普通的 load 指令。这意味着读者之间的扩展性近乎线性——100 个 CPU 并发读同一个 RCU 保护的数据,每个 CPU 的缓存行不需要在总线上反复 bouncing。

二、RCU 的核心原理

2.1 三个基本概念

RCU 建立在三个核心概念之上:

Read-Side Critical Section(读侧临界区):读者通过 rcu_read_lock() 和 rcu_read_unlock() 包围代码。在临界区内,读者可以安全地访问 RCU 保护的数据,但不能睡眠或阻塞。

Grace Period(宽限期):一个时间窗口,在这个窗口结束后,所有的读侧临界区都必然已经退出。宽限期结束意味着所有曾经指向旧数据的读者指针都不存在了,可以安全释放旧数据。

Publish-Subscription Mechanism(发布-订阅机制):写者通过 rcu_assign_pointer() 发布新数据(StoreStore 屏障),读者通过 rcu_dereference() 订阅新数据(LoadLoad 屏障)。

2.2 数据更新流程


写者更新流程(Read-Copy-Update 名称的由来):

1. Read(读)
   ├── 分配新数据结构副本: new_node = kmalloc(sizeof(*new_node), GFP_KERNEL)
   └── 复制旧数据内容到副本: memcpy(new_node, old_node, sizeof(*new_node))

2. Copy(拷贝修改)
   └── 在副本上应用更新: new_node->value = new_value

3. Update(更新指针)
   └── rcu_assign_pointer(gp, new_node)  // 原子更新全局指针
       └── 内置 StoreStore 屏障,保证先前的修改对读者可见

2.3 为什么读者不能睡眠?

RCU 的核心假设是:读者在 rcu_read_lock 到 rcu_read_unlock 之间的代码是无自愿切换(voluntary context switch)的。如果读者睡眠,CPU 可能调度走,导致读侧临界区被拉长到不确定时限,宽限期也就会被无限延迟。

因此,在 RCU 读侧临界区内,以下操作是禁止的:

  • schedule() / cond_resched()
  • 获取可能睡眠的互斥锁(mutex_lock)
  • 可能触发缺页的内存访问
  • copy_to_user / copy_from_user(可能缺页)

三、宽限期机制深度解析

3.1 宽限期的定义与检测

宽限期结束后,所有曾经持有对旧数据引用的读者都已经"离开"了它们各自的读侧临界区。内核通过 quiescent state(静止状态) 来检测宽限期结束:


quiescent state 的定义:
- 一个 CPU 发生了 context switch(表明它不在 RCU 读侧临界区内)
- 或者一个 CPU 在用户态运行(对于 CONFIG_RCU_FAST_NO_HZ 类配置)

宽限期完成的判定:
- 所有 CPU 都经历过至少一次 quiescent state

// kernel/rcu/update.c
void call_rcu(struct rcu_head *head, rcu_callback_t func)
{
    // 宽限期结束后回调 func,func 负责释放旧数据
    head->func = func;
    __call_rcu(&head->next, head->func, rcu_data);
}

3.2 宽限期状态机


                    CPU 进入读侧临界区
                    v
   ┌─────────────────────────────────┐
   │         RCU_READER_ACTIVE        │
   └────────────────┬────────────────┘
                    │ rcu_read_unlock()
                    │ OR CPU 报告 quiescent state
                    v
   ┌─────────────────────────────────┐
   │    RCU_READER_QUIESCENT         │ ← 宽限期完成的条件
   └─────────────────────────────────┘

每个 CPU 维护一个嵌套计数器 rcu_read_lock_nesting。当此计数器为 0 时,CPU 被视为处于 quiescent state。

3.3 Non-Blocking 与 Blocking 读者

Linux 内核有两类 RCU 变体:

Tree RCU(CONFIG_PREEMPT_RCU=y 时启用):

  • 每个 CPU 维护 rcu_node 构成的树形结构
  • 宽限期的完成通过根节点收集所有 leaf 节点的 quiescent state
  • 支持抢占式内核中的 RCU 读侧临界区

Tiny RCU(CONFIG_PREEMPT_RCU=n 时启用):

  • 极简实现,所有 CPU 共享位图
  • 仅 40 行核心代码,适合嵌入式系统

3.4 宽限期与上下文切换的交互


时间线:

CPU 0 (读者):    [rcu_read_lock] ... 数据读取 ... [rcu_read_unlock]
CPU 1 (写者):                          [调用 call_rcu]
CPU 2 (读者):         [rcu_read_lock] ... 读取 ...
                         ↓ context switch (quiescent state 报告)
CPU 3 (读者):                            [rcu_read_unlock]

                |<---------- 宽限期 ---------->|
                写者调用 call_rcu              所有 CPU 都 QS,宽限期结束
                                               call_rcu 回调执行

四、RCU API 完整使用指南

4.1 基本读取 API


// 读取 RCU 保护的指针(发布-订阅模式)
struct my_data *p;

rcu_read_lock();
p = rcu_dereference(gp);  // 带内存屏障保护的引用获取
if (p) {
    // 安全读取 p 指向的数据
    printk("value: %d\n", p->value);
}
rcu_read_unlock();

rcu_dereference() 在不同架构上有不同的屏障保证:

  • x86: 由于 TSO 模型,实际上只是防止编译器重排序(barrier())
  • ARM64: 插入 dmb ishld 屏障,防止 LoadLoad 重排序
  • DEC Alpha: 需要最强屏障(Alpha 是唯一允许 Invisible 行为的平台)

4.2 保护链表操作


// 链表的 RCU 安全遍历
struct node {
    int value;
    struct node *next;  // RCU 保护
};

struct node *head;  // RCU 全局保护指针

// 读者遍历链表
void read_list(void)
{
    struct node *p;
    rcu_read_lock();
    p = rcu_dereference(head);
    while (p) {
        printk("%d ", p->value);
        p = rcu_dereference(p->next);
    }
    rcu_read_unlock();
}

// 写者插入新节点
void insert_node(int value)
{
    struct node *new = kmemdup(...);

    new->value = value;
    new->next = head;
    rcu_assign_pointer(head, new);  // 新节点成为 head
    // 旧 head 不会被释放,等待宽限期后由 call_rcu 回收
}

// 写者删除节点(标记删除)
void delete_node(int value)
{
    struct node *p, *prev;
    
    // 找到要删除的节点
    for (prev = NULL, p = rcu_dereference(head); 
         p != NULL; prev = p, p = rcu_dereference(p->next)) {
        if (p->value == value)
            break;
    }
    if (!p) return;
    
    if (prev)
        rcu_assign_pointer(prev->next, p->next);
    else
        rcu_assign_pointer(head, p->next);
    
    call_rcu(&p->rcu_head, free_node);  // 延迟释放
}

static void free_node(struct rcu_head *rh)
{
    struct node *node = container_of(rh, struct node, rcu_head);
    kfree(node);
}

4.3 同步等待宽限期

synchronize_rcu() 阻塞调用者直到当前宽限期结束:


// 适用于不能将释放操作放在 call_rcU 回调中的场景
// 注意:不能在 RCU 读侧临界区内调用!

void teardown_old(struct my_data *old)
{
    synchronize_rcu();  // 等待所有读者退出
    kfree(old);         // 此时没有任何读者持有 old 指针
}

4.4 srcu(Sleepable RCU)

如果读者需要睡眠,使用 synchronize_srcu() 系列:


// 声明
static DEFINE_SRCU(my_srcu);

// 读者(可以睡眠)
int idx = srcu_read_lock(&my_srcu);
// ... 此处可以睡眠 ...
srcu_read_unlock(&my_srcu, idx);

// 写者
void srcu_update(new_data)
{
    struct my_data *old = rcu_dereference_protected(gp, ...);
    rcu_assign_pointer(gp, new_data);
    synchronize_srcu(&my_srcu);  // 等待所有读者
    kfree(old);
}

SRCU 的开销比标准 RCU 大(需要原子操作维护嵌套计数),但是允许读者睡眠。

4.5 RCU 保护的 hash table 模式


struct hash_table {
    struct hlist_head *heads;
    size_t size;
};

struct entry {
    struct hlist_node hlist;  // hash 链表节点
    struct rcu_head rcu;      // RCU 回调
    u32 key;
    void *value;
};

struct hash_table *ht __rcu;  // RCU 保护的 hash 表

// 查找
void *hash_lookup(struct hash_table *ht, u32 key)
{
    struct entry *e;
    void *val = NULL;
    
    rcu_read_lock();
    hash_for_each_possible(ht->heads, e, hlist, key % ht->size) {
        if (e->key == key) {
            val = e->value;
            break;
        }
    }
    rcu_read_unlock();
    return val;
}

// 插入
int hash_insert(struct hash_table *ht, u32 key, void *value)
{
    struct entry *new = kmalloc(sizeof(*new), GFP_KERNEL);
    new->key = key;
    new->value = value;
    
    rcu_read_lock();
    hash_add(ht->heads, &new->hlist, key % ht->size);
    rcu_read_unlock();
    return 0;
}

// 替换并延迟释放
void hash_replace(struct hash_table *ht, u32 key, void *newval,
                  struct entry **old_entry_ptr)
{
    ...
}

五、内存序与编译器屏障

5.1 为什么需要 StoreStore 屏障


// rcu_assign_pointer() 的实现(简化版)
#define rcu_assign_pointer(p, v) \
    smp_storeptr_deprecated(p, v, __ATOMIC_REL (__rcu))

// 实际展开(以 x86 为例):
static __always_inline void __write_once_size(volatile void *p,
                                                void *data,
                                                int size)
{
    switch (size) {
    case 1: *(volatile u8 *)p = *(u8 *)data; break;
    case 2: *(volatile u16 *)p = *(u16 *)data; break;
    case 4: *(volatile u32 *)p = *(u32 *)data; break;
    case 8: *(volatile u64 *)p = *(u64 *)data; break;
    }
}

在弱内存序架构(如 ARM)上:

写者可能执行:

  1. 填充新节点的 fields
  2. rcu_assign_pointer 更新指针

如果没有 StoreStore 屏障,ARM CPU 可能重排序:

  1. 更新指针(让读者看到新指针)
  2. 填充新节点的 fields(读者看到的节点内容是未初始化的垃圾数据)

5.2 弱内存序下的 LoadLoad 屏障

rcu_dereference() 的 LoadLoad 屏障确保:

  1. 先获取指针 p
  2. 再读取 p->field

在 Alpha 架构上,如果没有 LoadLoad 屏障,可能发生:

  1. 读取 p->field(旧节点的值)
  2. 读取 p(读者的指针获取晚于字段读取)

这会导致读者可能同时看到旧节点的字段和新节点的指针组合——一个完全错乱的数据状态。

5.3 与 C11 内存模型的对应关系


发布-订阅模式的 C11 原子操作等价:

rcu_assign_pointer (写者发布):
    atomic_store_explicit(&gp, new_ptr, memory_order_release);

rcu_dereference (读者订阅):
    atomic_load_explicit(&gp, memory_order_consume);
    // 注意:当前 C12++ 标准中 memory_order_consume 被弃用
    // 内核使用更底层架构相关的实现

六、RCU 在内核中的典型应用

6.1 进程 PID 管理

Linux 的 PID hash 表使用 RCU 保护:


// kernel/pid.c
struct pid *find_pid_ns(int nr, struct pid_namespace *ns)
{
    struct pid *pid;
    
    rcu_read_lock();
    pid = idr_find(&pid_idr, nr);  // RCU-safe hash 查找
    rcu_read_unlock();
    return pid;
}

6.2 路由表查找(FIB)

Linux 内核的 FIB (Forwarding Information Base) 使用 RCU 保护前缀树:


// 路由查找是网络栈最频繁的操作之一
// RCU 保证了在并发路由更新期间查找无需加锁

struct fib_alias *fib_find_alias(...)
{
    rcu_read_lock();
    // 遍历 fib_alias 链表
    rcu_read_unlock();
    ...
}

6.3 VFS dentry 缓存

dentry 缓存是 RCU 在 VFS 中最成功的应用之一:


// fs/dcache.c 中的 d_lookup()
struct dentry *d_lookup(const struct dentry *parent, const struct qstr *name)
{
    struct dentry *dentry;
    unsigned seq;
    
    do {
        seq = read_seqcount_begin(&parent->d_seq);
        dentry = __d_lookup(parent, name);  // RCU 模式遍历
        } while (read_seqcount_retry(&parent->d_seq, seq));
    return dentry;
}

fast_d_lookup 和 lookup_fast() 中的 dentry 路径行走完全在 RCU 模式下进行——不需要任何锁。

6.4 网络连接跟踪(conntrack)

nf_conntrack 是 RCU 在 netfilter 中的经典用例:


// net/netfilter/nf_conntrack_core.c
struct nf_conn *nf_conntrack_find_get(...)
{
    struct nf_conn *ct;
    
    rcu_read_lock();
    hlist_nulls_for_each_entry_rcu(ct, n, hash, ...) {
        if (nf_ct_tuple_equal(tuple, &ct->tuple)) {
            if (atomic_notifier_call_chain(...) == NOTIFY_DONE)
                goto found;
        }
    }
    rcu_read_unlock();
    return NULL;
found:
    // 获取引用计数后仍保持 RCU 锁
    ...
}

七、RCU 的调试与追踪

7.1 RCU stall 检测

当 RCU 读侧临界区过长时,内核会打印 RCU stall 警告:


# 查看 RCU stall 阈值(默认 21 秒)
$ sysctl kernel.rcu_cpu_stall_timeout
21

# 在 /var/log/messages 中搜索 RCU stall 日志
$ dmesg | grep -E "RCU.*stall|RCU.*blocked"

典型输出:


rcu_sched self-detected stall on CPU 3

7.2 LOCKDEP 对 RCU 的支持

开启 CONFIG_PROVE_RCU 后,lockdep 会检测:

  • 在 RCU 读侧临界区内睡眠
  • 在 IRQ 上下文中使用 synchronize_rcu()
  • 未持 RCU 锁时直接读取 RCU 保护的数据

7.3 BPF 追踪 RCU 事件


# 追踪 RCU 回调调用频率
$ bpftrace -e 'kprobe:__call_rcu / @rcu = count(); } interval:s:5 { print(@rcu); clear(@rcu); }'

# 追踪宽限期事件
$ bpftrace -e 'tracepoint:rcu:rcu_grace_period { printf("GP %s\n", args->s); }'

# 追踪 RCU 利用率
$ bpftool prog show | grep rcu

7.4 RCU 运行时统计


# 查看 RCU 状态
$ cat /sys/kernel/debug/rcu/rcu_preempt/rcudata

# 查看宽限期统计
$ dmesg | grep -i "rcu.*grac"
[    0.062547] rcu: Setting dynamically calculated FQS delay to 10 jiffies.
[    0.135827] rcu:     CPUs: 4.

八、常见陷阱与解决方案

陷阱一:在 RCU 读侧临界区内睡眠


// ❌ 错误:在 RCU 临界区内申请可能睡眠的锁
rcu_read_lock();
p = rcu_dereference(gp);
mutex_lock(&p->lock);  // 陷阱:p->lock 可能导致睡眠!
// ... 使用 p ...
mutex_unlock(&p->lock);
rcu_read_unlock();

// ✅ 正确做法 1:使用 refcount + 二次检查
get_object(p);  // 原子递增引用计数
rcu_read_unlock();
mutex_lock(&p->lock);
// 此时即使宽限期结束,对象也不会被释放
mutex_unlock(&put_object(p);

// ✅ 正确做法 2:使用 srcu
idx = srcu_read_lock(&my_srcu);
p = srcu_dereference(gp, &my_srcu);
mutex_lock(&p->lock);  // srcu 允许睡眠
// ...
mutex_unlock(&p->lock);
srcu_read_unlock(&my_srcu, idx);

陷阱二:在 synchronize_rcu 中嵌套 RCU 读侧临界区


// ❌ 错误:死锁
rcu_read_lock();
synchronize_rcu();  // 等待读者退出,但自己就是读者!→ 死锁
rcu_read_unlock();

// ✅ 正确:必须在 RCU 临界区外调用 synchronize_rcu()
rcu_read_unlock();
synchronize_rcu();
// 需要时再重新获取
rcu_read_lock();

陷阱三:使用 kfree 替代 call_rcu


// ❌ 错误:写者立即 kfree 后,旧读者可能访问已释放内存
old = rcu_dereference_protected(gp, ...);
new = kmalloc(...);
rcu_assign_pointer(gp, new);
kfree(old);  // 有读者可能仍持有 old 指针 → use-after-free!

// ✅ 正确:通过 call_rcu 延迟释放
old = rcu_dereference_protected(gp, ...);
new = kmalloc(...);
rcu_assign_pointer(gp, new);
call_rcu(&old->rcu_head, free_old);  // 宽限期结束后安全释放

陷阱四:删除节点时未使用 RCU API


// ❌ 错误:链表删除直接修改 rcu_assign_pointer
old = rcu_dereference(head);
prev->next = old->next;  // 不是 RCU 安全的写法!

// ✅ 正确:使用 RCU 感知的删除
list_for_each_entry_rcu(pos, head, member) { ... }
list_del_rcu(&entry->list);  // 专门的 RCU 删除 API

陷阱五:多写者场景下缺乏外部同步


// ❌ 错误:两个写者同时更新 RCU 保护的同一数据
// writer A:
rcu_assign_pointer(gp, new_a);
// writer B(同时执行):
rcu_assign_pointer(gp, new_b);  // A 的更新被 B 覆盖!

// ✅ 正确:写者之间需额外自旋锁
spin_lock(&writer_lock);
old = rcu_dereference_protected(gp, lockdep_is_held(&writer_lock));
rcu_assign_pointer(gp, new_data);
spin_unlock(&writer_lock);
call_rcu(&old->rcu, free_rcu);

九、生产环境性能调优

9.1 批量 RCU 回调

当需要频繁更新 RCU 保护数据时,考虑批量提交回调:


// 单次操作触发一次 call_rcu 效率低
// 可以批量收集后一次性提交

struct rcu_callbacks {
    struct rcu_head rcu;
    struct list_head list;
    int count;
};

static void batch_release(struct rcu_head *rh)
{
    struct rcu_callbacks *batch = container_of(rh, struct rcu_callbacks, rcu);
    struct free_entry *entry, *tmp;
    
    list_for_each_entry_safe(entry, tmp, &batch->list, node) {
        kfree(entry->data);
        kfree(entry);
    }
    kfree(batch);
}

9.2 `rcu_read_lock` 的开销分析

在 x86 上,rcu_read_lock()/rcu_read_unlock() 的实际指令序列:


# CONFIG_PREEMPT=n 时(非抢占内核):
rcu_read_lock()  →  空操作(no-op)
rcu_read_unlock() →  空操作(no-op)

# CONFIG_PREEMPT=y 时:
rcu_read_lock()  →  仅 preempt_disable() 的汇编代码
rcu_read_unlock() →  仅 preempt_enable()

即使在抢占式内核上,RCU 读侧也只需要 preempt_disable/enable,远轻量于原子操作。

9.3 RCU 与 CPU 热插拔

当 CPU 离线时,内核需要将待处理的 RCU 回调迁移到其他 CPU:


# 查看 RCU 热插拔状态
$ cat /sys/devices/system/cpu/cpu0/rcu_expedited

# 强制 expedited 宽限期(加速 RCU 回调处理)
$ echo 1 > /proc/sys/kernel/rcu_normal

# 设置 RCU 回调高优先级处理
$ echo 1 > /sys/module/rcutree/parameters/rcu_kick_kthreads

十、RCU 发展趋势与高级主题

10.1 RCU 与内存回收(Reclaim)

SLUB/SLAB 分配器中与 RCU 配合的 rcu_free 技术在不断进化。Linux 6.x 引入了 kvfree_rcu() 系列,允许在 VLA 或 vmalloc 分配的数据上使用 RCU 延迟释放。

10.2 Polled Grace Period

Linux 6.4+ 引入了 RCU polling API,在中断上下文或不能阻塞的代码中检查宽限期状态:


// 不可阻塞上下文宽限期等待
void poll_state_snapshot(struct rcu_poll_state *ps);

10.3 RCU 与 Rust for Linux

随着 Rust 代码进入 Linux 内核,RCU 的 Rust 封装正在开发中,利用 Rust 的 ownership 系统提供更安全的读取 API:


// 未来可能的 Rust 风格 API
let guard = rcu::read_lock();
let data = gp.read(&guard);  // 借用检查器保证生命周期
// 自动释放 RCU 读锁

总结:RCU 是 Linux 内核中最具创新性的同步原语,它以精妙的宽限期机制实现了读多写少场景下的极致性能。理解 RCU 的内存模型、正确使用其 API 并避开常见陷阱,是编写高性能内核代码和维护生产系统的关键技能。在 VFS dentry、网络连接跟踪、路由表查找等关键路径中,RCU 让 Linux 内核实现了近线性的横向扩展能力——这是现代高并发系统不可或缺的架构基石。

点赞(0) 打赏

评论列表 共有 0 条评论

暂无评论
立即
投稿

微信公众账号

微信扫一扫加关注

发表
评论
返回
顶部