手机版欢迎访问it开发者社区(www.mfbz.cn)网站

当前位置： > 开发

Linux内核原语（十）——RCU机制

时间：2021/5/18 16:36:55|来源：|点击：次

Linux内核原语（十）——RCU机制

小狼@http://blog.csdn.net/xiaolangyangyang

RCU 的全称是（Read-Copy-Update），意在读写-复制-更新，在 Linux 提供的所有内核互斥的设施当中属于一种免锁机制。在之前讨论过的读写自旋锁（rwlock）、顺序锁（seqlock）一样，RCU 的适用模型也是读写共存的系统。

读写自旋锁：读者和写者互斥，读者和读者共存，写者和写者互斥。（偏向读者）
顺序锁：写者和写者互斥，写者直接打断读者（偏向写者）

上述两种都是基于 spinlock 的一种用于特定场景的锁机制。

RCU 与他们不同，它的读取和写入操作无需考虑两者之间的互斥问题。

之前的锁分析中，可以知道，加锁、解锁都涉及内存操作，同时伴有内存屏障引入，这些都会导致锁操作的系统开销变大，在此基础之上，内核在 Kernel 的 2.5 版本引入了 RCU 的免锁互斥访问机制。

什么叫免锁机制呢？对于被RCU保护的共享数据结构，读者不需要获得任何锁就可以访问它，但写者在访问它时首先拷贝一个副本，然后对副本进行修改，最后使用一个回调（callback）机制在适当的时机把指向原来数据的指针重新指向新的被修改的数据。这个时机就是所有引用该数据的CPU都退出对共享数据的操作。

因此RCU实际上是一种改进的 rwlock，读者几乎没有什么同步开销，它不需要锁，不使用原子指令。不需要锁也使得使用更容易，因为死锁问题就不需要考虑了。写者的同步开销比较大，它需要延迟数据结构的释放，复制被修改的数据结构，它也必须使用某种锁机制同步并行的其它写者的修改操作。读者必须提供一个信号给写者以便写者能够确定数据可以被安全地释放或修改的时机。有一个专门的垃圾收集器来探测读者的信号，一旦所有的读者都已经发送信号告知它们都不在使用被RCU保护的数据结构，垃圾收集器就调用回调函数完成最后的数据释放或修改操作。

RCU与rwlock的不同之处是：它既允许多个读者同时访问被保护的数据，又允许多个读者和多个写者同时访问被保护的数据（注意：是否可以有多个写者并行访问取决于写者之间使用的同步机制），读者没有任何同步开销，而写者的同步开销则取决于使用的写者间同步机制。但RCU不能替代rwlock，因为如果写比较多时，对读者的性能提高不能弥补写者导致的损失。

读者在访问被RCU保护的共享数据期间不能被阻塞，这是RCU机制得以实现的一个基本前提，也就说当读者在引用被RCU保护的共享数据期间，读者所在的CPU不能发生上下文切换，spinlock和rwlock都需要这样的前提。写者在访问被RCU保护的共享数据时不需要和读者竞争任何锁，只有在有多于一个写者的情况下需要获得某种锁以与其他写者同步。写者修改数据前首先拷贝一个被修改元素的副本，然后在副本上进行修改，修改完毕后它向垃圾回收器注册一个回调函数以便在适当的时机执行真正的修改操作。等待适当时机的这一时期称为宽限期 grace period，而CPU发生了上下文切换称为经历一个quiescent state，grace period就是所有CPU都经历一次quiescent state所需要的等待的时间（读的时候禁止了内核抢占，也就是上下文切换，如果在某个 CPU 上发生了进程切换，那么所有对老指针的引用都会结束之后）。垃圾收集器就是在grace period之后调用写者注册的回调函数（call_rcu 函数注册回调）来完成真正的数据修改或数据释放操作的。

总的来说，RCU的行为方式：

1、随时可以拿到读锁，即对临界区的读操作随时都可以得到满足

2、某一时刻只能有一个人拿到写锁，多个写锁需要互斥，写的动作包括拷贝--修改--宽限窗口到期后删除原值

3、临界区的原始值为m1,如会有人拿到写锁修改了临界区为m2,则在写锁修改临界区之后拿到的读锁获取的临界区的值为m2,之前获取的为m1,这通过原子操作保证

对比发现RCU读操作随时都会得到满足，但写锁之后的写操作所耗费的系统资源就相对比较多了，并且只有在宽限期之后删除原资源。

针对对象

RCU 保护的对象是指针。这一点尤其重要.因为指针赋值是一条单指令.也就是说是一个原子操作.因它更改指针指向没必要考虑它的同步.只需要考虑cache的影响。

内核中所有关于 RCU 的操作都应该使用内核提供的 RCU 的 APIs 函数完成，这些 APIs 主要集中在指针和链表的操作。

使用场景

RCU 使用在读者多而写者少的情况.RCU和读写锁相似.但RCU的读者占锁没有任何的系统开销.写者与写写者之间必须要保持同步,且写者必须要等它之前的读者全部都退出之后才能释放之前的资源

读者是可以嵌套的.也就是说rcu_read_lock()可以嵌套调用

从 RCU 的特性可知，RCU 的读取性能的提升是在增加写入者负担的前提下完成的，因此在一个读者与写者共存的系统中，按照设计者的说法，如果写入者的操作比例在 10% 以上，那么久应该考虑其他的互斥方法，反正，使用 RCU 的话，能够获取较好的性能。

使用限制

读者在访问被RCU保护的共享数据期间不能被阻塞。

在读的时候，会屏蔽掉内核抢占。

RCU 的实现原理

在RCU的实现过程中，我们主要解决以下问题：

1，在读取过程中，另外一个线程删除了一个节点。删除线程可以把这个节点从链表中移除，但它不能直接销毁这个节点，必须等到所有的读取线程读取完成以后，才进行销毁操作。RCU中把这个过程称为宽限期（Grace period）。

2，在读取过程中，另外一个线程插入了一个新节点，而读线程读到了这个节点，那么需要保证读到的这个节点是完整的。这里涉及到了发布-订阅机制（Publish-Subscribe Mechanism）。

3，保证读取链表的完整性。新增或者删除一个节点，不至于导致遍历一个链表从中间断开。但是RCU并不保证一定能读到新增的节点或者不读到要被删除的节点。

1. 宽限期

通过例子，方便理解这个内容。以下例子修改于Paul的文章：

struct foo {
    int a;
    char b;
    long c;
};

DEFINE_SPINLOCK(foo_mutex);

struct foo *gbl_foo;

void foo_read (void)
{
    foo *fp = gbl_foo;
    if ( fp != NULL )
    dosomething(fp->a, fp->b , fp->c );
}

void foo_update( foo* new_fp )
{
    spin_lock(&foo_mutex);
    foo *old_fp = gbl_foo;
    gbl_foo = new_fp;
    spin_unlock(&foo_mutex);
    kfee(old_fp);
}

如上的程序，是针对于全局变量 gbl_foo 的操作。假设以下场景。有两个线程同时运行 foo_ read 和 foo_update 的时候，当 foo_ read 执行完赋值操作后，线程发生切换；此时另一个线程开始执行 foo_update 并执行完成。当 foo_ read运行的进程切换回来后，运行 dosomething 的时候，fp 已经被删除，这将对系统造成危害。为了防止此类事件的发生，RCU里增加了一个新的概念叫宽限期（Grace period）。如下图所示：

图中每行代表一个线程，最下面的一行是删除线程，当它执行完删除操作后，线程进入了宽限期。宽限期的意义是，在一个删除动作发生后，它必须等待所有在宽限期开始前已经开始的读线程结束，才可以进行销毁操作。这样做的原因是这些线程有可能读到了要删除的元素。图中的宽限期必须等待1和2结束；而读线程5在宽限期开始前已经结束，不需要考虑；而3,4,6也不需要考虑，因为在宽限期结束后开始后的线程不可能读到已删除的元素。为此RCU机制提供了相应的API来实现这个功能。

用 RCU：

void foo_read(void)
{
    rcu_read_lock();
    foo *fp = gbl_foo;
    if ( fp != NULL )
    dosomething(fp->a,fp->b,fp->c);
    rcu_read_unlock();
}

void foo_update( foo* new_fp )
{
    spin_lock(&foo_mutex);
    foo *old_fp = gbl_foo;
    gbl_foo = new_fp;
    spin_unlock(&foo_mutex);
    ynchronize_rcu();
    kfee(old_fp);
}

其中 foo_ read 中增加了 rcu_read_lock 和 rcu_read_unlock，这两个函数用来标记一个RCU读过程的开始和结束。其实作用就是帮助检测宽限期是否结束。foo_update 增加了一个函数 synchronize_rcu()，调用该函数意味着一个宽限期的开始，而直到宽限期结束，该函数才会返回。我们再对比着图看一看，线程1和2，在 synchronize_rcu 之前可能得到了旧的 gbl_foo，也就是 foo_update 中的 old_fp，如果不等它们运行结束，就调用 kfee(old_fp)，极有可能造成系统崩溃。而3,4,6在synchronize_rcu 之后运行，此时它们已经不可能得到 old_fp，此次的kfee将不对它们产生影响。

宽限期是RCU实现中最复杂的部分,原因是在提高读数据性能的同时，删除数据的性能也不能太差。

2. 订阅——发布机制

当前使用的编译器大多会对代码做一定程度的优化，CPU也会对执行指令做一些优化调整,目的是提高代码的执行效率，但这样的优化，有时候会带来不期望的结果。如例：

void foo_update( foo* new_fp )
{
    spin_lock(&foo_mutex);
    foo *old_fp = gbl_foo;

    new_fp->a = 1;
    new_fp->b = ‘b’;
    new_fp->c = 100;

    gbl_foo = new_fp;
    spin_unlock(&foo_mutex);
    synchronize_rcu();
    kfee(old_fp);
}

这段代码中，我们期望的是6，7，8行的代码在第10行代码之前执行。但优化后的代码并不对执行顺序做出保证。在这种情形下，一个读线程很可能读到 new_fp，但new_fp的成员赋值还没执行完成。当读线程执行dosomething(fp->a, fp->b , fp->c ) 的时候，就有不确定的参数传入到dosomething，极有可能造成不期望的结果，甚至程序崩溃。可以通过优化屏障来解决该问题，RCU机制对优化屏障做了包装，提供了专用的API来解决该问题。这时候，第十行不再是直接的指针赋值，而应该改为 :

rcu_assign_pointer(gbl_foo,new_fp);

<include/linux/rcupdate.h> 中 rcu_assign_pointer的实现比较简单，如下：

#define rcu_assign_pointer(p, v) \
__rcu_assign_pointer((p), (v), __rcu)

#define __rcu_assign_pointer(p, v, space) \
do { \
    smp_wmb(); \
    (p) = (typeof(*v) __force space *)(v); \
} while (0)

我们可以看到它的实现只是在赋值之前加了优化屏障 smp_wmb来确保代码的执行顺序。另外就是宏中用到的__rcu，只是作为编译过程的检测条件来使用的。

<include/linux/rcupdate.h>

#define rcu_dereference(p) rcu_dereference_check(p, 0)


#define rcu_dereference_check(p, c) \
__rcu_dereference_check((p), rcu_read_lock_held() || (c), __rcu)

#define __rcu_dereference_check(p, c, space) \
({ \
    typeof(*p) *_________p1 = (typeof(*p)*__force )ACCESS_ONCE(p); \
    rcu_lockdep_assert(c, "suspicious rcu_dereference_check()" \
    " usage"); \
    rcu_dereference_sparse(p, space); \
    smp_read_barrier_depends(); \
    ((typeof(*p) __force __kernel *)(_________p1)); \
})

static inline int rcu_read_lock_held(void)
{
    if (!debug_lockdep_rcu_enabled())
        return 1;
    if (rcu_is_cpu_idle())
        return 0;
    if (!rcu_lockdep_current_cpu_online())
        return 0;
    return lock_is_held(&rcu_lock_map);
}

这段代码中加入了调试信息，去除调试信息，可以是以下的形式（其实这也是旧版本中的代码）：

#define rcu_dereference(p) ({ \
    typeof(p) _________p1 = p; \
    smp_read_barrier_depends(); \
    (_________p1); \
})

在赋值后加入优化屏障smp_read_barrier_depends()。

我们之前的第四行代码改为 foo *fp = rcu_dereference(gbl_foo);，就可以防止上述问题。

3. 数据读取的完整性

还是通过例子来说明这个问题：

如图我们在原list中加入一个节点new到A之前，所要做的第一步是将new的指针指向A节点，第二步才是将Head的指针指向new。这样做的目的是当插入操作完成第一步的时候，对于链表的读取并不产生影响，而执行完第二步的时候，读线程如果读到new节点，也可以继续遍历链表。如果把这个过程反过来，第一步head指向new，而这时一个线程读到new，由于new的指针指向的是Null，这样将导致读线程无法读取到A，B等后续节点。从以上过程中，可以看出RCU并不保证读线程读取到new节点。如果该节点对程序产生影响，那么就需要外部调用做相应的调整。如在文件系统中，通过RCU定位后，如果查找不到相应节点，就会进行其它形式的查找，相关内容等分析到文件系统的时候再进行叙述。

我们再看一下删除一个节点的例子：

如图我们希望删除B，这时候要做的就是将A的指针指向C，保持B的指针，然后删除程序将进入宽限期检测。由于B的内容并没有变更，读到B的线程仍然可以继续读取B的后续节点。B不能立即销毁，它必须等待宽限期结束后，才能进行相应销毁操作。由于A的节点已经指向了C，当宽限期开始之后所有的后续读操作通过A找到的是C，而B已经隐藏了，后续的读线程都不会读到它。这样就确保宽限期过后，删除B并不对系统造成影响。

如何使用 RCU

除了上一节使用 RCU 的例子，这一节在贴一个使用 RCU 的例子：

// 假设 struct shared_data 是一个在读者和写者之间共享的数据结构
struct shared_data {
    int a;
    int b;
    struct rcu_head rcu;
};

Reader Code :

// 读者的代码。
// 读者调用 rcu_read_lock 和 rcu_read_unlock 来构建并访问临界区
// 所有对指向被保护资源指针的引用都应该只在临界区出现，而且临界区代码不能睡眠
static void demo_reader(struct shared_data *ptr)
{
    struct shared_data *p = NULL;
    rcu_read_lock();
    // call rcu_dereference to get the ptr pointer
    p = rcu_dereference(ptr);
    if (p)
        do_somethings(p);
    rcu_read_unlock();
}

Writer Code :

// 写入侧的代码
// 写入者提供的回调函数，用于释放老的数据指针
static void demo_del_oldptr(struct rcu_head *rh)
{
    struct shared_data *p = container_of(rh, struct rcu_head, rcu);
    kfree(p);
}

static void demo_writer(struct shared_data *ptr)
{
    struct shared_data *new_ptr = kmalloc(...);
    ....
    new_ptr->a = 10;
    new_ptr->b = 20;
    // 用新指针更新老指针
    rcu_assign_pointer(ptr, new_ptr);
    // 调用 call_rcu 让内核在确保所有对老指针 ptr 的引用都解锁后，回调到 demo_del_oldptr 释放老指针
    call_rcu(ptr->rcu, demo_del_oldptr);
}

在上面的例子，写者调用 rcu_assign_pointer 更新老指针后，使用 call_rcu 接口，向系统注册了一会回调函数，系统在确定没有对老指针引用之后，调用这个函数。另一个类似的函数是上一节遇到的 synchronize_rcu 调用，这个函数可能会阻塞，因为他要等待所有对老指针的引用全部结束才返回，函数返回的时候意味着系统所有对老指针的引用都消失，此时在释放老指针才是安全的。如果在中断上下文执行写入者的操作，那么就应该使用 call_rcu ，不能使用 synchronize_rcu。

对于这 call_rcu 和 synchronize_rcu 的分析如下：

在释放老指针方面，Linux内核提供两种方法供使用者使用，一个是调用call_rcu,另一个是调用synchronize_rcu。前者是一种异步方式，call_rcu会将释放老指针的回调函数放入一个结点中，然后将该结点加入到当前正在运行call_rcu的处理器的本地链表中，在时钟中断的 softirq部分（RCU_SOFTIRQ）， rcu软中断处理函数rcu_process_callbacks会检查当前处理器是否经历了一个休眠期(quiescent，此处涉及内核进程调度等方面的内容)，rcu的内核代码实现在确定系统中所有的处理器都经历过了一个休眠期之后(意味着所有处理器上都发生了一次进程切换，因此老指针此时可以被安全释放掉了)，将调用call_rcu提供的回调函数。
synchronize_rcu的实现则利用了等待队列，在它的实现过程中也会向call_rcu那样向当前处理器的本地链表中加入一个结点，与 call_rcu不同之处在于该结点中的回调函数是wakeme_after_rcu，然后synchronize_rcu将在一个等待队列中睡眠，直到系统中所有处理器都发生了一次进程切换，因而wakeme_after_rcu被rcu_process_callbacks所调用以唤醒睡眠的 synchronize_rcu，被唤醒之后，synchronize_rcu知道它现在可以释放老指针了。

所以我们看到，call_rcu返回后其注册的回调函数可能还没被调用，因而也就意味着老指针还未被释放，而synchronize_rcu返回后老指针肯定被释放了。所以，是调用call_rcu还是synchronize_rcu，要视特定需求与当前上下文而定，比如中断处理的上下文肯定不能使用 synchronize_rcu函数了。

基本RCU操作 APIs

对于reader，RCU的操作包括：

（1）rcu_read_lock：用来标识RCU read side临界区的开始。

（2）rcu_dereference：该接口用来获取RCU protected pointer。reader要访问RCU保护的共享数据，当然要获取RCU protected pointer，然后通过该指针进行dereference的操作。

（3）rcu_read_unlock：用来标识reader离开RCU read side临界区

对于writer，RCU的操作包括：

（1）rcu_assign_pointer：该接口被writer用来进行removal的操作，在witer完成新版本数据分配和更新之后，调用这个接口可以让RCU protected pointer指向RCU protected data。

（2）synchronize_rcu：writer端的操作可以是同步的，也就是说，完成更新操作之后，可以调用该接口函数等待所有在旧版本数据上的reader线程离开临界区，一旦从该函数返回，说明旧的共享数据没有任何引用了，可以直接进行reclaimation的操作。

（3）call_rcu：当然，某些情况下（例如在softirq context中），writer无法阻塞，这时候可以调用call_rcu接口函数，该函数仅仅是注册了callback就直接返回了，在适当的时机会调用callback函数，完成reclaimation的操作。这样的场景其实是分开removal和reclaimation的操作在两个不同的线程中：updater和reclaimer。

RCU 的链表操作：

在 Linux kernel 中还专门提供了一个头文件（include/linux/rculist.h），提供了利用 RCU 机制对链表进行增删查改操作的接口。

（1） list_add_rcu ：该函数把链表项new插入到RCU保护的链表head的开头。使用内存栅保证了在引用这个新插入的链表项之前，新链表项的链接指针的修改对所有读者是可见的。

static inline void list_add_rcu(struct list_head *new, struct list_head *head)

（2） list_add_tail_rcu：该函数类似于list_add_rcu，它将把新的链表项new添加到被RCU保护的链表的末尾。

static inline void list_add_tail_rcu(struct list_head *new, struct list_head *head)

（3） list_del_rcu：该函数从RCU保护的链表中移走指定的链表项entry，并且把entry的prev指针设置为LIST_POISON2，但是并没有把entry的next指针设置为LIST_POISON1，因为该指针可能仍然在被读者用于便利该链表。

static inline void list_del_rcu(struct list_head *entry)

（4） list_replace_rcu：该函数是RCU新添加的函数，并不存在非RCU版本。它使用新的链表项new取代旧的链表项old，内存栅保证在引用新的链表项之前，它的链接指针的修正对所有读者可见

static inline void list_replace_rcu(struct list_head *old, struct list_head *new)

（5） list_for_each_entry_rcu ：该宏用于遍历由RCU保护的链表head

#define list_for_each_entry_rcu(pos, head, member) \
    for (pos = list_entry_rcu((head)->next, typeof(*pos), member); \
    &pos->member != (head); \
    pos = list_entry_rcu(pos->member.next, typeof(*pos), member))