Linux内核RCU深度实战：从宽限期机制到生产级无锁读取

RCU（Read-Copy-Update）是 Linux 内核中最精妙的同步机制之一，它能让读取端实现真正的零开销——无需原子操作、无需内存屏障、无需锁。本文将从底层原理出发，深入剖析 RCU 的宽限期机制、内核实现、生产级使用模式以及性能优化策略。

核心思想：RCU 将"更新"拆分为"移除"和"释放"两个阶段，通过等待所有读者完成来安全回收旧数据，从而让读者享受无锁读写的极致性能。

一、RCU 核心原理

1.1 读者优先哲学

传统同步机制（如读写锁）本质上是对共享资源的互斥访问——写者阻塞读者，读者也阻塞写者。RCU 打破了这一范式：读者永远不阻塞，写者批量阻塞。

1.2 宽限期（Grace Period）机制

宽限期是 RCU 的核心概念。所谓宽限期，是指"所有当前活跃的 RCU 读者完成读取"的一段时间。

时间线：
  读者A  |---rcu_read_lock--->[读取数据]---rcu_read_lock--->|
  读者B         |---rcu_read_lock--->[读取数据]---rcu_read_unlock--->|
  写者  |--更新指针--|--synchronize_rcu()等待---|--释放旧数据--|
                                        |
                                  宽限期边界

Linux 内核通过 quiescent state（静默状态）机制跟踪宽限期：

1. 每个 CPU 在两种情况下处于静默状态：上下文切换、或用户态执行
2. 全局计数器 gpnum 记录当前的宽限期编号
3. 每个 CPU 维护自己的 passed_quiesce 标志，标记是否经历过静默状态
4. 当所有 CPU 都经历过静默状态，宽限期结束

1.3 发布-订阅协议（Publish-Subscribe）

RCU 的数据访问严格遵循发布-订阅语义：

// 写者：发布新数据
new_node->data = 42;
new_node->next = old_head->next;
rcu_assign_pointer(list_head, new_node);  // 发布：Store-Store 屏障

// 读者：订阅数据
rcu_read_lock();
p = rcu_dereference(list_head);  // 订阅：Load-Load 屏障
if (p)
    do_something(p->data);
rcu_read_unlock();

rcu_assign_pointer() 保证所有先前的写入对其他 CPU 可见，之后指针才被更新——这是 Store-Store 屏障。rcu_dereference() 保证指针读取先于后续的数据访问——这是 Load-Load 屏障。

二、内核 RCU 实现深度剖析

2.1 数据结构体系

// RCU 状态节点（层次化 RCU 的核心）
struct rcu_node {
    raw_spinlock_t lock;         // 保护本节点
    unsigned long gpnum;        // 本节点已知的最新宽限期
    unsigned long completed;     // 本节点完成的最新宽限期
    unsigned long qsmask;       // 子节点静默位图
    unsigned long qsmaskinit;   // 初始静默位图
    int grplo;                  // 本节点管理的最低 CPU
    int grphi;                  // 本节点管理的最高 CPU
    u8 grpmask;                 // 本节点在父节点中的位掩码
    u8 level;                   // 在树中的层级
    struct rcu_node *parent;    // 父节点
};

// 每 CPU RCU 数据
struct rcu_data {
    unsigned long gpnum;        // 本 CPU 已响应的最新 GP
    unsigned long completed;     // 本 CPU 完成的最新 GP
    bool passed_quiesce;         // 是否经历过静默状态
    bool qs_pending;             // 是否有待处理的 QS 报告
    // ...
};

// RCU 状态（全局）
struct rcu_state {
    struct rcu_node node[MAX_RCU_LVLS];  // RCU 节点树
    struct rcu_data __percpu *rda;       // 每 CPU 数据
    unsigned long gpnum;                // 当前全局 GP 编号
    unsigned long completed;             // 已完成的 GP 编号
    // ...
};

Linux 内核采用层次化 RCU 树来管理宽限期状态。在 256 CPU 的系统中，RCU 节点组织为多叉树，叶子节点管理少量 CPU，内部节点合并子节点的静默状态。这种设计使得宽限期等待的时间复杂度从 O(N) 降低到 O(log N)。

2.2 宽限期检测流程

// 简化版宽限期检测流程

// CPU 经历静默状态时调用
void rcu_qs(void)
{
    struct rcu_data *rdp = this_cpu_ptr(rcu_state.rda);
    struct rcu_node *rnp = rdp->mynode;

    rdp->passed_quiesce = true;
    
    // 如果当前 GP 已经全部完成，向父节点报告
    if (!rdp->qs_pending)
        return;

    // 将本 CPU 的位图清零
    raw_spin_lock(rnp->lock);
    rnp->qsmask &= ~rdp->grpmask;
    
    if (rnp->qsmask != 0) {
        raw_spin_unlock(rnp->lock);
        return;  // 兄弟 CPU 还未全部静默
    }
    
    // 所有子节点已完成，向父节点报告
    raw_spin_unlock(rnp->lock);
    report_qs_upward(rnp->parent, rnp);  // 递归向上传播
}

// 顶层节点检测到宽限期完成
void rcu_gp_end(struct rcu_data *rdp)
{
    // 使用回调机制释放旧数据
    rcu_do_batch(rdp);
}

2.3 回调机制：call_rcu 实现

synchronize_rcu() 会阻塞等待宽限期结束，但在软中断或 NMI 上下文中不能阻塞。call_rcu() 提供异步版本，将回调注册到队列中，宽限期结束时异步执行：

// call_rcu 回调链表结构
struct rcu_head {
    struct rcu_head *next;
    void (*func)(struct rcu_head *head);
};

// 每 CPU 的回调队列（按优先级分四个段）
// wait: 等待当前 GP 完成的回调
// done: 当前 GP 刚完成，等待下一 GP 释放
// next: 未来 GP 的回调（cascade）

void call_rcu(struct rcu_head *head, rcu_callback_t func)
{
    head->func = func;
    head->next = NULL;
    
    // 将回调加入当前 CPU 的 next 队列
    local_irq_save(flags);
    // 加入环形缓冲区...
    local_irq_restore(flags);
    
    // 触发 RCU 软中断
    raise_softirq(RCU_SOFTIRQ);
}

三、生产级 RCU 使用模式

3.1 单向链表的 RCU 保护

// 经典场景：RCU 保护的单向链表

struct my_node {
    int data;
    struct my_node __rcu *next;
};

struct my_node __rcu *list_head;

// ===== 读者 =====
struct my_node *p;
rcu_read_lock();
p = rcu_dereference(list_head);
while (p) {
    if (p->data == target) {
        // 访问 p 是安全的（p 不会被释放）
        rcu_read_unlock();
        return p;
    }
    p = rcu_dereference(p->next);
}
rcu_read_unlock();
return NULL;

// ===== 写者：插入节点 =====
struct my_node *new = kmalloc(sizeof(*new), GFP_KERNEL);
new->data = val;
raw_spin_lock(&list_lock);           // 写者之间仍需互斥
new->next = list_head;
rcu_assign_pointer(list_head, new);  // 原子发布新头
raw_spin_unlock(&list_lock);

// ===== 写者：删除节点 =====
raw_spin_lock(&list_lock);
old = list_head;
rcu_assign_pointer(list_head, old->next);
raw_spin_unlock(&list_lock);

call_rcu(&old->rcu_head, free_node);  // 异步释放
// 或者 synchronize_rcu(); kfree(old);  // 同步释放

3.2 哈希表的 RCU 保护

哈希表是 RCU 的典型应用场景。内核中的 pid_hash、dentry_hashtable 等都使用 RCU 实现无锁读取：

#define HASH_SIZE 256

struct hash_node {
    int key;
    void __rcu *value;
    struct hlist_node hlist;
    struct rcu_head rcu;
};

struct hlist_head buckets[HASH_SIZE];

// 读者：无锁读取
void *hash_lookup(int key) {
    struct hash_node *node;
    unsigned int hash = hash_int(key) % HASH_SIZE;
    void *val;
    
    rcu_read_lock();
    hlist_for_each_entry_rcu(node, &buckets[hash], hlist) {
        if (node->key == key) {
            val = rcu_dereference(node->value);
            rcu_read_unlock();
            return val;
        }
    }
    rcu_read_unlock();
    return NULL;
}

// 写者：RCU 安全的插入
void hash_insert(int key, void *value) {
    struct hash_node *node = kmalloc(sizeof(*node), GFP_KERNEL);
    node->key = key;
    raw_spin_lock(&hash_lock);
    rcu_assign_pointer(node->value, value);
    hlist_add_head_rcu(&node->hlist, &buckets[hash]);
    raw_spin_unlock(&hash_lock);
}

// 写者：RCU 安全的删除
void hash_delete(int key) {
    struct hash_node *node;
    struct hlist_node *n;
    unsigned int hash = hash_int(key) % HASH_SIZE;
    
    raw_spin_lock(&hash_lock);
    hlist_for_each_entry_safe(node, n, &buckets[hash], hlist) {
        if (node->key == key) {
            hlist_del_init_rcu(&node->hlist);
            raw_spin_unlock(&hash_lock);
            call_rcu(&node->rcu, hash_free_cb);
            return;
        }
    }
    raw_spin_unlock(&hash_lock);
}

3.3 多版本数据（MV-RCU）模式

对于频繁读取、偶尔更新的配置数据，MV-RCU（Multi-Version RCU）是最优模式——读者直接访问当前版本，无需任何同步：

// 多版本数据结构
struct config_snapshot {
    unsigned long version;
    int param_a;
    int param_b;
    struct rcu_head rcu;
};

struct config_snapshot __rcu *current_config;

// 读者：读取当前快照（零开销）
const struct config_snapshot *get_config(void)
{
    return rcu_dereference(current_config);
}

// 写者：创建新版本 + 原子切换
void update_config(int a, int b)
{
    struct config_snapshot *new, *old;

    old = rcu_dereference_protected(current_config, lockdep_is_held(&config_lock));
    new = kmalloc(sizeof(*new), GFP_KERNEL);
    new->version = old->version + 1;
    new->param_a = a;
    new->param_b = b;

    raw_spin_lock(&config_lock);
    rcu_assign_pointer(current_config, new);
    raw_spin_unlock(&config_lock);

    call_rcu(&old->rcu, free_old_config);
}

四、内核中的 RCU 实战案例

4.1 路由缓存（dst_entry）的消亡

Linux 4.16 之前，路由子系统使用 RCU 保护 dst_entry 的读写。虽然由于 GC 问题已改用 refdst，但 RCU 在路由查找的热路径中仍然有广泛应用：

// 路由查找中的 RCU 使用（简化）
const struct fib_table *tb;
struct fib_result res;
struct fib_nh_common *nhc;

// fib_table_lookup 内部使用 RCU 保护 FIB 遍历
rcu_read_lock();
err = fib_table_lookup(net, tb, fl4, &res, FIB_LOOKUP_NOREF);
if (err) {
    rcu_read_unlock();
    return err;
}

nhc = rcu_dereference(res.nhc);
if (!nhc) {
    rcu_read_unlock();
    return -EINVAL;
}
// 使用 nhc...
rcu_read_unlock();

4.2 进程 PID 管理

内核中的 pid_hash 使用 RCU 实现无锁的 PID 到 struct pid 映射，这对 kill()、waitpid() 等系统调用的性能至关重要：

// kernel/pid.c: pid_nr_ns 的读取路径
struct pid *find_pid_ns(int nr, struct pid_namespace *ns)
{
    return idr_find(&pid_idr, nr);  // 使用 RCU 保护
}

// kernel/pid.c: forget_pid 的删除路径
void detach_pid(struct task_struct *task, enum pid_type type)
{
    // 使用 RCU 安全移除
    idr_replace(&task_active_pid_ns(task)->idr, 
                NULL, task_pid_nr_ns(task));
    call_rcu(&task->rcu, delayed_put_task_struct);
}

4.3 网络协议栈中的 RCU

TCP 的监听哈希表、conntrack 的连接跟踪表都使用 RCU 保护，以支持高并发的连接建立：

// net/ipv4/tcp_ipv4.c: 使用 RCU 的 TCP 连接查找
struct sock *tcp_v4_lookup(struct net *net,
                           __be32 saddr, __be16 sport,
                           __be32 daddr, __be16 dport,
                           int dif)
{
    struct sock *sk;
    
    // 使用 RCU 保护监听哈希表的遍历
    rcu_read_lock();
    sk = __inet_lookup_listener(net, &tcp_hashinfo,
                                saddr, sport, daddr, dport, dif);
    if (!sk)
        sk = __inet_lookup(net, &tcp_hashinfo,
                           saddr, sport, daddr, dport, dif);
    if (sk)
        sock_hold(sk);
    rcu_read_unlock();
    
    return sk;
}

4.4 调度器中的 RCU

调度器在读取任务列表和 cgroup 配置时大量使用 RCU。sched_ext（BPF 调度器）的 CPU 选择逻辑也依赖 RCU 读取任务组信息：

// 示例：读取任务组的 RCU 保护
struct task_group;

void update_task_group_bandwidth(struct task_group *tg, u64 new_bw)
{
    struct task_group *old_tg, *new_tg;
    
    // 创建新的 task_group 副本
    new_tg = dup_task_group(tg);
    new_tg->bandwidth = new_bw;
    
    // 原子切换
    rcu_assign_pointer(per_cpu_ptr(&root_task_group, cpu), new_tg);
    synchronize_rcu();
    free_task_group(old_tg);
}

// 调度热路径中读取
static void pick_next_task(struct rq *rq)
{
    struct task_group *tg;
    rcu_read_lock();
    tg = rcu_dereference(rq->curr_tg);
    // 读取 tg 配置...
    rcu_read_unlock();
}

五、进阶主题：SRCU、Tasks RCU 与 BPF 集成

5.1 Sleepable RCU (SRCU)

标准 RCU 要求读者在非抢占上下文中执行（不能休眠），但在许多场景中读者需要持有休眠型锁或做内存分配。SRCU 解决了这个问题：

// SRCU 读者：允许在 RCU 读侧临界区休眠
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);
}

// SRCU 宽限期（比常规 RCU 慢一次上下文切换）
synchronize_srcu(ssp);  // 等待所有 srcu_read_lock 完成

// 使用场景：需要持有信号量时的 RCU 读取
struct srcu_struct my_srcu;

int read_data_with_lock(void)
{
    int idx;
    idx = srcu_read_lock(&my_srcu);
    // 可以在这里休眠（持有互斥锁）
    mutex_lock(&data_mutex);
    do_read(data);
    mutex_unlock(&data_mutex);
    srcu_read_unlock(&my_srcu, idx);
    return 0;
}

5.2 Tasks RCU

针对包含大量长睡眠进程的特殊场景（如调试器），常规 RCU 可能无法正常检测静默状态。Tasks RCU 专门追踪任务级宽限期：

// tasks_rcu 特殊处理 TASK_UNINTERRUPTIBLE 状态
// 常用于 ftrace 的栈回溯追踪
void task_rcu_read_lock(void)
{
    current->rcu_read_lock_nesting++;
    barrier();
}

void task_rcu_read_unlock(void)
{
    barrier();
    current->rcu_read_lock_nesting--;
}

// synchronize_tasks_rcu() 等待所有持有锁的任务切换

5.3 BPF 与 RCU 的协同

eBPF 程序中访问内核数据结构时需要正确的 RCU 语义。libbpf 提供了封装函数：

// BPF 程序中的安全读取（BPF 已自动处理 rcu_read_lock）
SEC("kprobe/tcp_v4_connect")
int BPF_KPROBE(trace_tcp_connect, struct sock *sk)
{
    struct inet_sock *inet = (struct inet_sock *)sk;
    // BPF 程序在 tracepoint/kprobe 上下文中自动持有 RCU 读锁
    u16 sport = BPF_CORE_READ(inet, inet_sport);
    u32 saddr = BPF_CORE_READ(inet, inet_saddr);
    bpf_printk("TCP connect from %pI4:%d\n", &saddr, bpf_ntohs(sport));
    return 0;
}

对于 cgroup 级别的 BPF 程序，可以使用 bpf_rcu_read_lock()/bpf_rcu_read_unlock() 显式管理 RCU 区域：

// 显式 RCU 锁用于 cgroup BPF
SEC("cgroup_skb/ingress")
int handle_ingress(struct __sk_buff *skb)
{
    struct cgroup *cgrp;
    
    bpf_rcu_read_lock();
    cgrp = BPF_CORE_READ(skb, sk, sk_cgrp_data, cgroup);
    u64 cgrp_id = cgrp->kn->id;
    bpf_rcu_read_unlock();
    
    // cgrp_id 可以在 RCU 之外使用
    return pass_or_drop(skb, cgrp_id);
}

六、性能优化与最佳实践

6.1 RCU vs 读写锁性能对比

在 64 核 ARM 服务器上的微基准测试数据（读取占比 99%）：

结论：RCU 在读取优势场景下吞吐量是读写锁的 6.4 倍。但写延迟更高，不适合写入频繁的场景。

6.2 宽限期优化策略

// 策略1：批量更新减少宽限期次数
// ❌ 不好：每次插入都同步等待
for (i = 0; i < N; i++) {
    add_element(list, data[i]);
    synchronize_rcu();  // 每次等待 ~30ms！
}

// ✅ 好：批量插入后统一等待
raw_spin_lock(&batch_lock);
for (i = 0; i < N; i++)
    raw_add_element(list, data[i]);
raw_spin_unlock(&batch_lock);
synchronize_rcu(); // 只等待一次

// 策略2：使用 call_rcu 避免读端阻塞
// ❌ 不好：阻塞读线程
void update_config(void) {
    synchronize_rcu();  // 持有大锁等待 30ms
    kfree(old);
}

// ✅ 好：异步释放
void update_config(void) {
    call_rcu(&old->rcu, free_config);  // 立即返回
}

// 策略3：RCU tiered 批量处理回调
// 在低负载时积累回调，达到阈值后批量执行

6.3 NUMA 感知的 RCU

在大型 NUMA 系统中，全局 RCU 状态可能成为瓶颈。现代内核提供了 NUMA 节点级别的 RCU 回调隔离：

// RCU stall 典型日志
[  +0.000000] rcu: INFO: rcu_preempt self-detected stall on CPU
[  +0.000000] rcu:     0-....: (21001 ticks this GP) idle=...
[  +0.000000] rcu:     rcu_scheduler_active=1
[  +0.000000]  rcu:  Stack:
[  +0.000000]  [] my_buggy_function+0x42/0x78
[  +0.000000]  [] some_caller+0x15/0x2a

// 排查步骤：
// 1. 检查栈顶函数是否在 RCU 读侧临界区内休眠
// 2. 检查是否有中断被长时间屏蔽
// 3. 增大 stall 阈值：rcupdate.rcu_cpu_stale=60

// lockdep 检测 RCU 休眠
BUG: sleeping function called from invalid context 
 at include/linux/rcupdate.h:...rcu_dereference_check()

// 常见错误代码：
rcu_read_lock();
p = rcu_dereference(head);
// ❌ 在 RCU 读侧 GFP_KERNEL 分配（可能休眠）
copy = kmemdup(p, sizeof(*p), GFP_KERNEL);
rcu_read_unlock();

// ✅ 正确做法：先复制再读
rcu_read_lock();
p = rcu_dereference(head);
copy = kmalloc(sizeof(*p), GFP_ATOMIC);  // 不触发休眠
if (copy)
    memcpy(copy, p, sizeof(*p));
rcu_read_unlock();

// 使用 BPF 追踪 RCU 宽限期
$ bpftrace -e 'tracepoint:rcu:rcu_grace_period { 
    printf("GP start on CPU %d, state: %d\n", args->cpu, args->gp_state); 
}'
$ bpftrace -e 'tracepoint:rcu:rcu_grace_period { 
    printf("GP end on CPU %d, completed: %lu\n", args->cpu, args->completed); 
}'
$ bpftrace -e 'tracepoint:rcu:rcu_callback { 
    printf("callback on CPU %d\n", args->cpu); 
}'

// conn_counter.c — RCU 保护的连接统计

#include <linux/module.h>
#include <linux/rcupdate.h>
#include <linux/slab.h>
#include <linux/hashtable.h>

struct conn_entry {
    u32 saddr;
    u32 daddr;
    u16 sport;
    u16 dport;
    u64 bytes;
    u64 packets;
    struct hlist_node hlist;
    struct rcu_head rcu;
};

static DEFINE_HASHTABLE(conn_table, 10);
static DEFINE_SPINTABLE(conn_lock);

// 查找连接（RCU 读取）
static struct conn_entry *conn_lookup(u32 saddr, u32 daddr, 
                                       u16 sport, u16 dport)
{
    struct conn_entry *entry;
    u32 key = hash_32(saddr ^ daddr ^ (sport << 16 | dport), 32);
    
    rcu_read_lock();
    hash_for_each_possible_rcu(conn_table, entry, hlist, key) {
        if (entry->saddr == saddr && entry->daddr == daddr &&
            entry->sport == sport && entry->dport == dport) {
            rcu_read_unlock();
            return entry;
        }
    }
    rcu_read_unlock();
    return NULL;
}

// 更新连接（RCU 写入，零阻塞读取）
static void conn_update(struct conn_entry *entry, u64 bytes, u64 pkts)
{
    // 创建新版本
    struct conn_entry *new = kmemdup(entry, sizeof(*entry), GFP_KERNEL);
    new->bytes += bytes;
    new->packets += pkts;
    
    // 原子替换
    spin_lock(&conn_lock);
    hash_del_rcu(&entry->hlist);
    hash_add_rcu(conn_table, &new->hlist, 
                 hash_32(new->saddr ^ new->daddr, 32));
    spin_unlock(&conn_lock);
    
    // 异步释放旧版本
    call_rcu(&entry->rcu, conn_free);
}

static void conn_free(struct rcu_head *head)
{
    struct conn_entry *entry = container_of(head, struct conn_entry, rcu);
    kfree(entry);
}

static int __init conn_counter_init(void)
{
    pr_info("RCU connection counter loaded\n");
    return 0;
}

static void __exit conn_counter_exit(void)
{
    synchronize_rcu();  // 等待所有读者退出
    // 清理残余条目...
}

module_init(conn_counter_init);
module_exit(conn_counter_exit);
MODULE_LICENSE("GPL");