RCU 读者无锁直读，Linux 内核机制在用户态怎么复刻？

一、RCU 是什么

写多线程代码时，要保护共享数据，第一反应通常是加锁。

但有一类场景很特别：读得特别多，写得特别少。

比如内核的路由表、文件系统的目录缓存，一秒钟可能被读几百万次，却难得改一次。如果每次读都要拿锁、放锁，性能就全耗在锁竞争上了。

Linux 内核专门为这种场景设计了一套机制，叫 RCU。

全称 Read-Copy-Update，翻译过来就是“读-复制-更新”。

核心思想一句话：

读操作几乎零开销、不加锁；写操作通过“复制-修改-替换”完成，旧数据等所有读者都走了再回收。

（图1：RCU 核心思想——读者无锁直读，写者复制副本、替换指针、旧数据待回收）

读者无需加锁，只是进入临界区时标记一下“我进去了”，出来时再标记一下“我出来了”。这个动作，比拿一把锁便宜太多。

那开销去哪了？转嫁给写者了。写者怎么干活，下一节说。

二、写者在做什么

写者的思路是：绝不在原处改，另起炉灶。

一共五步：

1. 复制：把要改的数据节点复制一份。
2. 修改：在副本上完成所有修改。
3. 发布：用一次原子操作，把旧指针换成新副本的指针。
4. 等宽限期：等所有在发布前就开始读的读者离开。
5. 回收：确认没人再碰旧数据后，安全释放。

（图2：写者五步流程——复制、修改、发布、等宽限期、回收）

第 3、4 步是 RCU 最精巧的地方。

发布之后，新读者马上能看到新数据；正在读的老读者，还拿着旧数据。新旧两版共存一段时间，直到最后一个老读者离开。这个窗口，就叫宽限期（Grace Period）。

这里得说清一个前提：第 3 步的“马上看到”，并不是简单的指针赋值。发布动作的本质是“写屏障 + 原子替换”。写者发布前，要保证副本里的数据都写完了；读者拿到新指针后，也要保证读到的是完整新数据。

内核里对应两个 API：rcu_assign_pointer() 和 rcu_dereference()。

有个坑要记住：x86 是强内存模型，硬件基本兜底；但 ARM 这类嵌入式多核是弱内存模型，读者侧的屏障省不得。否则可能出现“拿到新指针，却读到旧数据的半截”。

三、用读写锁模拟

原理听着不难，真要在用户态实现，宽限期的检测是个大麻烦。

先用读写锁模拟下，看清 RCU 的接口和生命周期。

思路是这样的：读者持读锁，写者持写锁。 写锁天然排他，写者一拿到写锁，就说明没有读者还待在临界区里，拿它来充当“宽限期”。

严格说这不是真 RCU：读者仍要加锁，写者会阻塞读者。但“复制-修改-发布-回收”的骨架一模一样，足够理解 RCU 怎么工作。

完整代码 rcu_sim.c：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

/* 被保护的数据结构 */
struct config {
    int timeout;
    char message[64];
};

/* 全局共享指针，读者通过它访问数据 */
struct config *g_config = NULL;

/* 保护指针替换和旧数据回收的读写锁 */
pthread_rwlock_t g_rwlock = PTHREAD_RWLOCK_INITIALIZER;

/* ---- RCU 风格的接口（模拟版） ---- */

/* 读者：获取一个不可变的引用 */
struct config *rcu_read_lock(void){
    pthread_rwlock_rdlock(&g_rwlock);
    return g_config;   /* 在锁内读取，保证可见性 */
}

void rcu_read_unlock(void){
    pthread_rwlock_unlock(&g_rwlock);
}

/* 写者：更新全局配置 */
int rcu_update_config(int new_timeout, const char *new_msg){
    /* 1. 复制 */
    struct config *new_cfg = malloc(sizeof(struct config));
    if (!new_cfg) {
        fprintf(stderr, "malloc failed\n");
        return -1;
    }

    /* 若旧配置存在，先复制过来（模拟部分更新） */
    /* 注意：仅单写者演示下安全，多写者需把复制移入锁内 */
    if (g_config) {
        memcpy(new_cfg, g_config, sizeof(struct config));
    } else {
        memset(new_cfg, 0, sizeof(struct config));
    }

    /* 2. 修改副本 */
    new_cfg->timeout = new_timeout;
    strncpy(new_cfg->message, new_msg, sizeof(new_cfg->message) - 1);
    new_cfg->message[sizeof(new_cfg->message) - 1] = '\0';

    /* 3. 发布：获取写锁，原子替换指针，释放旧数据 */
    pthread_rwlock_wrlock(&g_rwlock);
    struct config *old_cfg = g_config;
    g_config = new_cfg;        /* 原子发布（在锁保护下） */
    /* 4. 宽限期：写锁被持有 → 没有读者持有旧指针，可以安全回收 */
    if (old_cfg) {
        free(old_cfg);
    }
    pthread_rwlock_unlock(&g_rwlock);
    return 0;
}

/* ---- 测试线程 ---- */

void *reader_thread(void *arg){
    long id = (long)arg;
    int i;
    for (i = 0; i < 5; i++) {
        struct config *cfg = rcu_read_lock();
        /* 临界区：安全读取数据 */
        printf("Reader %ld: timeout=%d, msg='%s'\n",
               id, cfg->timeout, cfg->message);
        /* 模拟处理时间 */
        usleep(100000);
        rcu_read_unlock();
        usleep(50000);
    }
    return NULL;
}

void *writer_thread(void *arg){
    int i;
    for (i = 0; i < 3; i++) {
        char new_msg[64];
        snprintf(new_msg, sizeof(new_msg), "update-%d", i);
        printf("Writer: updating to timeout=%d, msg='%s'\n", 100 + i, new_msg);
        rcu_update_config(100 + i, new_msg);
        sleep(1);  /* 给读者时间观察变化 */
    }
    return NULL;
}

int main(){
    pthread_t readers[3], writer;
    long i;

    /* 初始化第一个配置 */
    struct config *init_cfg = malloc(sizeof(struct config));
    init_cfg->timeout = 30;
    strcpy(init_cfg->message, "initial");
    g_config = init_cfg;

    /* 创建线程 */
    for (i = 0; i < 3; i++) {
        pthread_create(&readers[i], NULL, reader_thread, (void*)i);
    }
    pthread_create(&writer, NULL, writer_thread, NULL);

    /* 等待结束 */
    for (i = 0; i < 3; i++) {
        pthread_join(readers[i], NULL);
    }
    pthread_join(writer, NULL);

    /* 清理最后一份数据 */
    free(g_config);
    pthread_rwlock_destroy(&g_rwlock);
    return 0;
}

这样写者每发布一次新值，读者下一轮就能读到新数据，而正在读的老读者手上还是旧数据。这就是 RCU 的“新旧并存”。

如果和内核中的 RCU 对比，差异如下：

四、用户态能用吗？

代价不小。

那为什么内核态又相对比较容易呢？

内核 RCU 之所以简单，靠的是特权。打个比方：

• 内核像一家纪律严明的公司内部图书馆。读者进门必须打卡且严禁打瞌睡（禁止抢占），因此必定很快离开。管理员（调度器）只需扫一眼系统，就能瞬间知道“旧书已没人看了”（宽限期结束），回收旧书毫无风险。
• 用户态则像完全开放的公共图书馆。读者随时可能看一半就出门接电话（被系统暂停），你却无权阻止。管理员管不到每个人，写者只能自己费力挨个去问“谁还在看旧书”，而这偏偏极不可靠。

技术上看，内核的读者只需轻量级的 preempt_disable，调度器在每次线程切换时都能天然捕捉到“不在读”的时刻，宽限期检测直接内建。而用户态失去了这种全局监控能力，必须让读者主动打报告（显式静默态），并用复杂的内存屏障或信号来猜测状态，实现自然变得困难重重。

真 RCU 到了用户态，宽限期怎么检测、内存顺序怎么保证、旧内存怎么安全回收，全得自己来。每一件都不省心。

好在有现成的轮子：liburcu（Userspace RCU）。它提供了 QSBR、内存屏障、信号等多种后端，直接调它的 API 就行。

想深入研究的话，可以从 liburcu 的源码看起。如果你对这类内核同步机制与用户态并发编程话题感兴趣，欢迎常来云栈社区逛逛，后续还会有更多源码级实战解析。