这些小活动你都参加了吗?快来围观一下吧!>>
电子产品世界 » 论坛首页 » 嵌入式开发 » MCU » RCU 机制

共1条 1/1 1 跳转至

RCU 机制

工程师
2014-11-05 20:39:56     打赏

《RCU是什么?》第一部分


概述

Read-copy update (RCU) 是一种 2002 年 10 月被引入到内核当中的同步机制。通过允许在更新的同时读数据,RCU 提高了同步机制的可伸缩性(scalability)。相对于传统的在并发线程间不区分是读者还是写者的简单互斥性锁机制,或者是哪些允许并发读但同时不 允许写的读写锁,RCU 支持同时一个更新线程和多个读线程的并发。RCU 通过保存对象的多个副本来保障读操作的连续性,并保证在预定的读方临界区没有完成之前不会释放这个对象。RCU定义并使用高效、可伸缩的机制来发布并读取 对象的新版本,并延长旧版本们的寿命。这些机制将工作分发到了读和更新路径上,以保证读路径可以极快地运行。在某些场合(非抢占内核),RCU 的读方没有任何性能负担。


问题1:seqlock 不是也允许读线程和更新线程并发工作么?


这个问题可以归结到 “确切地说,什么是RCU?” 这个问题,或许还是 “RCU 可能是如何工作的?” (再或者,不太可能的情况下,问题会变为什么情况下 RCU 不太可能工作)。本文从几个基本的出发点来回答这些问题;之后还会分批地从使用的角度和 API 的角度来看这些问题。最后一篇连载还会给出一组参考文献。


RCU 由三个基本机制组成,第一个用于插入,第二个用于删除,而第三个则用于让读线程可以承受并发的插入或删除。这三个机制将在下面的三节中介绍,讲述如何将 RCU 转化为链表:



  1. 订阅发布机制 (用于插入) 
  2. 等待已有的RCU读者完成 (用于删除) 
  3. 维护多个最近更新的对象的版本 (为读者维护)

这三个章节之后还有上重点回顾与快速问题答案。


订阅发布机制

RCU的一个关键特性是它可以安全地扫描数据,即使数据正被同时改写也没问题。要提供这种并发插入的能力,RCU使用了一种订阅发布机制。举例说,考虑一 个被初始化为 NULL 的全局指针变量 gp 将要被修改为新分配并初始化的数据结构。下面这段代码(使用附加的合适的锁机制)可以用于这个目的:


1 struct foo {
2 int a;
3 int b;
4 int c;
5 };
6 struct foo *gp = NULL;

8 /* . . . */

10 p = kmalloc(sizeof(*p), GFP_KERNEL);
11 p->a = 1;
12 p->b = 2;
13 p->c = 3;
14 gp = p;


不幸的是,没有方法强制保证编译器和CPU能顺序执行最后四条语句。如果gp的赋值早于p的各个域的初始化的话,那么并发的读操作将访问到未初始化的变 量。内存屏障(barrier)可以用于保障操作的顺序,但内存屏障以难以使用而闻名。这样我们将他们封装到具有发布语义的 rcu_assign_pointer() 原语之中。最后的四条将成为这样:


1 p->a = 1;
2 p->b = 2;
3 p->c = 3;
4 rcu_assign_pointer(gp, p);


rcu_assign_pointer() 将会发布新的结构,强制编译器和CPU在给p的各个域赋值之后再把指针赋值给gp。然而,仅仅强制更新操作的顺序是不够的,读者也必须强制使用恰当的顺序。考虑下面的这段代码:


1 p = gp;
2 if (p != NULL) {
3 do_something_with(p->a, p->b, p->c);
4 }


尽管这段代码看起来不会受到顺序错乱的影响,不过十分不幸,DEC Alpha CPU 和投机性编译器优化可能会引发问题,不论你是否相信,这的确有可能会导致 p->a, p->b, p->c 的读取会在读取 p 之前!这种情况在投机性编译器优化的情况中最有可能会出现,编译器会揣测p的值,取出 p->a, p->b 和 p->c,之后取出 p 的真实值来检查拽侧的正确性。这种优化非常激进,或者说疯狂,不过在确实会在profile-driven优化时发生。


毫无疑问,我们需要在CPU和编译器上阻止这种情况的发生。rcu_dereference() 原语使用了必要的内存屏障指令和编译器指令来达到这一目的:


1 rcu_read_lock();
2 p = rcu_dereference(gp);
3 if (p != NULL) {
4 do_something_with(p->a, p->b, p->c);
5 }
6 rcu_read_unlock();


rcu_dereference() 原语可以被看作是订阅了指针指向的值,保证接下来的取值操作将会看到对应的发布操作(rcu_assign_pointer())发生之前被初始化的值。 rcu_read_lock() 和 rcu_read_unlock() 绝对是必须的:他们定义了 RCU 读方临界区的范围。他们的目的将在下一节 解释,不过,他们不会自旋或阻塞,也不阻止 list_add_rcu() 的并发执行。事实上,对于非抢占内核,它们不产生任何代码。


虽然 rcu_assign_pointer() 和 rcu_dereference() 在理论上可以用于构建任意 RCU 保护的数据结构,但实际上,使用高层构造常常更好。因此,rcu_assign_pointer() 和 rcu_dereference() 原语被嵌入到了 Linux 的链表维护 API 中的特殊 RCU 变量之中了。Linux 有两个双向链表的变种,循环链表 struct list_head 和线性链表 struct hlist_head/struct hlist_node。前者的结构如下图所示,绿色的方块表示表头,蓝色的是链表中的元素。



点击图片可在新窗口打开


将上面的指针发布例子放到链表的场景中来就是这样:


1 struct foo {
2 struct list_head list;
3 int a;
4 int b;
5 int c;
6 };
7 LIST_HEAD(head);

9 /* . . . */
10 
11 p = kmalloc(sizeof(*p), GFP_KERNEL);
12 p->a = 1;
13 p->b = 2;
14 p->c = 3;
15 list_add_rcu(&p->list, &head);


第15行被使用某种同步机制保护住了,通常是某种所,以组织多个 list_add() 实例并发执行。然而,这些同步不能组织同时发生的RCU读者。订阅一个 RCU 保护的链表非常直接:


1 rcu_read_lock();
2 list_for_each_entry_rcu(p, head, list) {
3 do_something_with(p->a, p->b, p->c);
4 }
5 rcu_read_unlock();


list_add_rcu() 原语发布一个节点到制定的链表中去,保证对应的 list_for_each_entry_rcu() 调用都正确的订阅到同一个节点上。


问题2:如果在 list_for_each_entry_rcu() 运行时,刚好进行了一次 list_add_rcu(),如何防止 segfault 的发生呢?


Linux 中的另一个双向链表,hlist,是一个线性表,也就是说,它的头部仅需要一个指针,而不是向循环链表一样需要两个指针。这样,使用 hlist 作为大型哈希表的 hash-bucket 数组的容器将仅消耗一半的内存空间。



点击图片可在新窗口打开


将一个新元素添加到一个 RCU 保护的 hlist 里面与添加到循环链表里非常类似:


1 struct foo {
2 struct hlist_node *list;
3 int a;
4 int b;
5 int c;
6 };
7 HLIST_HEAD(head);

9 /* . . . */
10 
11 p = kmalloc(sizeof(*p), GFP_KERNEL);
12 p->a = 1;
13 p->b = 2;
14 p->c = 3;
15 hlist_add_head_rcu(&p->list, &head);


和上面一样,第15行一定使用了锁或其他某种同步机制。


订阅一个 RCU 保护的 hlist 也和循环链表非常接近。


1 rcu_read_lock();
2 hlist_for_each_entry_rcu(p, q, head, list) {
3 do_something_with(p->a, p->b, p->c);
4 }
5 rcu_read_unlock();


问题3:为什么我们需要传递两个指针给 hlist_for_each_entry_rcu(), list_for_each_entry_rcu() 可是只需要一个指针的啊?


RCU 发布与订阅原语在如下表中列出,同时给出了 “取消发布”或是撤回的原语


类别
发布
撤销
订阅


类别 
发布 
撤销 
订阅


指针 
rcu_assign_pointer()
rcu_assign_pointer(…, NULL)
rcu_dereference()


循环链表 
list_add_rcu()
list_add_tail_rcu()
list_replace_rcu()
list_del_rcu()
list_for_each_entry_rcu()


双向链表 
hlist_add_after_rcu()
hlist_add_before_rcu()
hlist_add_head_rcu()
hlist_replace_rcu()


hlist_del_rcu()


hlist_for_each_entry_rcu()


注意,list_replace_rcu(), list_del_rcu(), hlist_replace_rcu(), 以及 hlist_del_rcu() 增加了一些复杂度。什么时候释放被替换或删除掉的数据元素才是安全的呢?具体地说,我们怎么能知道所有的读者都释放了他们手中对数据元素的引用呢?


这些问题将在下面的章节中得到回答。


等待已经存在的RCU读者完成

RCU的最基本的功能就是等待一些事情的完成。当然,还有很多其他方法也是用于等待事情完成的,包括引用计数、读写锁、事件等。RCU最大的好处在于它可 以等待所有(比如说)两万件不同点事情,而无需显式地跟踪它们中的每一个,也不需要担心性能的下降、可伸缩性限制、复杂度死锁场景,以及内存泄露等所有这 些显式跟踪手法所固有的问题。


RCU 中,被等待的东西被叫做“RCU读方临界区”。一个RCU读方临界区始于 rcu_read_lock() 原语,止于 rcu_read_unlock() 原语。RCU 读方临界区可以嵌套,也可以放入很多代码,只要这些代码显式阻塞或睡眠即可(有一种称为“SRCU”的特殊RCU允许在它的读方临界区中睡眠)。只要你遵守这些约定,你就可以使用RCU来等待任何期望的代码段的完成。


正如其他地方对经典RCU实时RCU的描述,RCU 通过间接确定这些其他事情的完成时间来达到这一目的。


具体地说,如下图所示,RCU是一种等待已经存在的RCU读方临界区结束的方法,包括这些临界区中执行的内存操作。



点击图片可在新窗口打开


注意,开始于一个给定宽限期开始之后的RCU读方临界区能够、并可以延续到该宽限期结束之后。


下面的伪码展示了使用RCU等待读者的基本算法形式:



  1. 进行改动,比如,替换链表中的一个元素。 
  2. 等待所有已经存在的RCU读方临界区完成(比如,使用synchronize_rcu()原语)。关键点是接下来的RCU读方临界区将无法得到新近删除的元素的引用了。 
  3. 清理,比如,释放上述所有被替换的元素。

下面的代码段是从前一节修改而得的,用于说明这一过程,这里面的域a是这个搜索的键值。


1 struct foo {
2 struct list_head list;
3 int a;
4 int b;
5 int c;
6 };
7 LIST_HEAD(head);

9 /* . . . */
10 
11 p = search(head, key);
12 if (p == NULL) {
13 /* Take appropriate action, unlock, and return. */
14 }
15 q = kmalloc(sizeof(*p), GFP_KERNEL);
16 *q = *p;
17 q->b = 2;
18 q->c = 3;
19 list_replace_rcu(&p->list, &q->list);
20 synchronize_rcu();
21 kfree(p);


第19、20 和 21 行实现了上面所说的三个步骤。第 16-19行展现了 RCU 的名字(读-复制-更新):在允许进行并发读操作的同时,第16行进行了复制,而第17-19行进行了更新。


乍一看会觉得 synchronize_rcu() 原语显得比较神秘。毕竟它必须等所有读方临界区完成,而且,正如我们前面看到的,用于限制RCU读方临界区的rcu_read_lock() 和 rcu_read_unlock() 原语在非抢占内核中甚至什么代码都不会生成。


这里有一个小伎俩,经典RCU通过 rcu_read_lock() 和 rcu_read_unlock() 界定的读方临界区是不允许阻塞和休眠的。因此,当一个给定的CPU要进行上下文切换的时候,我们可以确定任何已有的RCU读方临界区都已经完成了。也就是说,只要每个CPU都至少进行了一次上下文切换,那么所有先前的 RCU 读方临界区也就保证都完成了,即 synchronize_rcu() 可以安全返回了。


因此,经典RCU的 synchronize_rcu() 从概念上说可以被简化成这样:


1 for_each_online_cpu(cpu)
2 run_on(cpu);


这里,run_on() 将当前线程切换到指定 CPU,来强制该 CPU 进行上下文切换。而 for_each_online_cpu() 循环强制对每个 CPU 进行一次上下文切换。虽然这个简单的方法可以在一个不支持抢占的内核上工作,换句话说,对 non-CONFIG_PREEMPT 和 CONFIG_PREEMPT,但对 CONFIG_PREEMPT_RT 实时 (-rt) 内核无效。因此,实时RCU使用了一个(松散地)基于引用计数的方法。


当然,在真实内核中的实现要复杂得多了,因为它需要管理终端,NMI,CPU热插拔和其他实际内核中的可能有的风险,而且还要维护良好的性能和可伸缩性。RCU的实时实现还必须拥有良好的实时响应能力,这就使得(像上面两行那样)直接禁止抢占变得不可能了。


虽然我们了解到了 synchronize_rcu() 的简单实现原理,不过还有很多其它问题呢。比如,RCU读者们在读一个正在被并发地更新的链表的时候究竟读到了什么呢?这个问题将在下一节讲到。


维护多个版本的近期更新的对象

本节将展示 RCU 如何为多个不需要同步的读者维护不同版本的链表。我们使用两个例子来展示一个可能被给定的读者引用的元素必须在该读者处于读方临界区的整个过程中保持完好无损。第一个例子展示了链表元素的删除,而第二个例子则展示了元素的替换。


例1:在删除时维护多个版本

要开始这个“删除”的例子,我们先把上节这个例子的 11-21行改成如下的形式:


1 p = search(head, key);
2 if (p != NULL) {
3 list_del_rcu(&p->list);
4 synchronize_rcu();
5 kfree(p);
6 }


这个链表以及指针p的最初情况是这样的:



点击图片可在新窗口打开


表中每个元素的三元组分别代表域a, b, c。红色的便捷表明读者可以获取它们的指针,而且因为读操作和更新操作不是直接同步的,读者可以在这个删除的过程中同时发生。这里我们为了清晰没有画出双向链表的反向指针。


在第三行的 list_del_rcu() 完成的时候,5,6,7 这个元素已经被从链表中删除了(如下图)。由于读者并不直接和更新操作同步,读者可能同时正在扫描这个链表。由于访问时间不同,这些并发读者可能看到、也 可能没看到新近删除的元素。不过,那些在获取指针之后延迟了读操作的读者(比如因为中断、ECC内存错误,或在CONFIG_PREEMPT_RT内核中因为抢占而延迟了的)可能仍然会在删除之后的一段时间内看到那个老的链表的版本。下图中 5,6,7 元素的边框仍然是红色的,这意味着仍然有读者可能会引用它。



点击图片可在新窗口打开


这里注意,在退出读方临界区之后,读者们就不能再持有 5,6,7 这个元素的引用了。所以,一旦第4行的 synchronize_rcu() 完成了,所有已有读者也就保证都完成了,这样就没有读者会访问这个元素了,下图中,这个元素的边框也变黑了。我们的链表也回到了一个单一的版本了。



点击图片可在新窗口打开


这之后,5,6,7 这个元素就可以被安全的释放了:



After deletion.


这里,我们完成了删除 5,6,7 这个元素的操作,下一小节将介绍替换操作。


例2:在替换的过程中维护数据的多个不同版本

在开始替换的例子钱,我们再修改一下前面例子的最后几行:


1 q = kmalloc(sizeof(*p), GFP_KERNEL);
2 *q = *p;
3 q->b = 2;
4 q->c = 3;
5 list_replace_rcu(&p->list, &q->list);
6 synchronize_rcu();
7 kfree(p);


这个链表的初始状态和指针p和删除的那个例子是完全一样的:



Initial list state.


和之前一样,每个元素里面的三元组分别代表域 a, b 和 c。红色的边框代表了读者可能会持有这个元素的引用,因为读者和更新者没有直接的同步,读者可能会和整个替换过程并发进行。再次说明,这里我们为了清晰,再次省略了反向指针。


第一行的 kmalloc() 生成了一个替换元素,如下:



点击图片可在新窗口打开


第二行把旧的元素的内容拷贝给新的元素:



点击图片可在新窗口打开


第三行,将 q->b 更新为2:



点击图片可在新窗口打开


第四行,将 q->c 更新为3:



点击图片可在新窗口打开


现在,第5行进行替换操作,这里,新元素最终对读者可见了。到了这里,如下所示,我们有了这个链表的两个版本。先前已经存在的读者可以看到 5,6,7 元素,而新读者将看到 5,2,3 元素。不过,任何读者都被保证可以看到一个完整的链表。



点击图片可在新窗口打开


第6行的 synchronize_rcu() 返回后,宽限期将完成,所有在 list_replace_rcu() 之前开始的读者都将完成。具体地说,任何可能持有 5,6,7 的读者都已经退出了他们的读方临界区,这就保证他们不再持有一个引用。因而也在没有任何读者持有老元素的引用了,途中,5,6,7 元素的边框也就变黑了。对于读者来说,目前又只有一个单一的链表版本了,只是新的元素已经替代了旧元素的位置。



点击图片可在新窗口打开


第七行的 kfree() 完成后,链表旧成为了如下的样子:



点击图片可在新窗口打开


尽管 RCU 是以替换而命名的,但内核中的大多数使用都是前面小节 中的简单删除的情况。


讨论

这个例子假设在更新操作的过程中保存着一个互斥量,也就是说,这个链表在一个给定时间最多有两种版本。


问题4:如何修改删除的例子,来允许超过两个版本的链表可以同时存在?


问题5:在某一时刻,RCU最多可以有多少个链表的版本?


这组例子显示了RCU使用多个版本来保障在存在并发读者的情况下的安全更改数据。当然,一些算法是无法很好地支持多个版本的。有一个参考文献 介绍了如何对这些算法进行改造以使用RCU,不过,这超出了本文的讨论范围了。


小结

本文介绍了基于RCU的算法的三个基本部分:



  1. 对与添加新数据的发布-订阅机制 
  2. 等待已有RCU读者完成,以及 
  3. 维护多个版本以便在不顺坏或严重延迟RCU读者的情况下,允许更改。

问题6:如果 rcu_read_lock() 与 rcu_read_unlock() 之间没有自旋锁或阻塞,RCU更新者会怎样延迟RCU读者?


这三个RCU的组成部分允许数据在并发读者访问的同时更新数据,并可以以多种方式实现基于RCU的算法,一些算法将会在接下来的“What is RCU, Really?”系列中继续介绍。


RCU 机制 http://www.360doc.com/content/10/0106/17/701200_12820595.shtml
Linux RCU机制详解
分类: RCU  311人阅读 评论(0) 收藏 举报

原文网址:http://blog.chinaunix.net/u1/51562/showart_1341707.html


一:前言
RCU机制出现的比较早,只是在linux kernel中一直到2.5版本的时候才被采用.关于RCU机制,这里就不做过多的介绍了,网上有很多有关RCU介绍和使用的文档.请自行查阅.本文主要是从linux kernel源代码的角度.来分析RCU的实现.
在讨论RCU的实现之前.有必要重申以下几点:
1:RCU使用在读者多而写者少的情况.RCU和读写锁相似.但RCU的读者占锁没有任何的系统开销.写者与写写者之间必须要保持同步,且写者必须要等它之前的读者全部都退出之后才能释放之前的资源.
2:RCU保护的是指针.这一点尤其重要.因为指针赋值是一条单指令.也就是说是一个原子操作.因它更改指针指向没必要考虑它的同步.只需要考虑cache的影响.
3:读者是可以嵌套的.也就是说rcu_read_lock()可以嵌套调用.
4:读者在持有rcu_read_lock()的时候,不能发生进程上下文切换.否则,因为写者需要要等待读者完成,写者进程也会一直被阻塞.
以下的代码是基于linux kernel 2.6.26
二:使用RCU的实例
Linux kernel中自己附带有详细的文档来介绍RCU,这些文档位于linux-2.6.26.3/Documentation/RCU. 这些文档值得多花点时间去仔细研读一下.
下面以whatisRCU.txt中的例子作为今天分析的起点:
struct foo {
        int a;
        char b;
        long c;
};
DEFINE_SPINLOCK(foo_mutex);


struct foo *gbl_foo;
void foo_update_a(int new_a)
{
        struct foo *new_fp;
        struct foo *old_fp;


        new_fp = kmalloc(sizeof(*new_fp), GFP_KERNEL);
        spin_lock(&foo_mutex);
        old_fp = gbl_foo;
        *new_fp = *old_fp;
        new_fp->a = new_a;
        rcu_assign_pointer(gbl_foo, new_fp);
        spin_unlock(&foo_mutex);
        synchronize_rcu();
        kfree(old_fp);
}


int foo_get_a(void)
{
        int retval;


        rcu_read_lock();
        retval = rcu_dereference(gbl_foo)->a;
        rcu_read_unlock();
        return retval;
}
如上代码所示,RCU被用来保护全局指针struct foo *gbl_foo. foo_get_a()用来从RCU保护的结构中取得gbl_foo的值.而foo_update_a()用来更新被RCU保护的gbl_foo的值.
另外,我们思考一下,为什么要在foo_update_a()中使用自旋锁foo_mutex呢?
假设中间没有使用自旋锁.那foo_update_a()的代码如下:


void foo_update_a(int new_a)
{
        struct foo *new_fp;
        struct foo *old_fp;


        new_fp = kmalloc(sizeof(*new_fp), GFP_KERNEL);
       
        old_fp = gbl_foo;
        1:-------------------------    
        *new_fp = *old_fp;
        new_fp->a = new_a;
        rcu_assign_pointer(gbl_foo, new_fp);
       
        synchronize_rcu();
        kfree(old_fp);
}
假设A进程在上图----标识处被B进程抢点.B进程也执行了goo_ipdate_a().等B执行完后,再切换回A进程.此时,A进程所持的old_fd实际上已经被B进程给释放掉了.此后A进程对old_fd的操作都是非法的.


另外,我们在上面也看到了几个有关RCU的核心API.它们为别是:
rcu_read_lock()
rcu_read_unlock()
synchronize_rcu()
rcu_assign_pointer()
rcu_dereference()
其中,rcu_read_lock()和rcu_read_unlock()用来保持一个读者的RCU临界区.在该临界区内不允许发生上下文切换.
rcu_dereference():读者调用它来获得一个被RCU保护的指针.
Rcu_assign_pointer():写者使用该函数来为被RCU保护的指针分配一个新的值.这样是为了安全从写者到读者更改其值.这个函数会返回一个新值


三:RCU API实现分析
Rcu_read_lock()和rcu_read_unlock()的实现如下:
#define rcu_read_lock() __rcu_read_lock()
#define rcu_read_unlock() __rcu_read_unlock()


#define __rcu_read_lock() /
    do { /
        preempt_disable(); /
        __acquire(RCU); /
        rcu_read_acquire(); /
    } while (0)
#define __rcu_read_unlock() /
    do { /
        rcu_read_release(); /
        __release(RCU); /
        preempt_enable(); /
    } while (0)
其中__acquire(),rcu_read_read_acquire(),rcu_read_release(),rcu_read_release()都是一些选择编译函数,可以忽略不可看.因此可以得知.rcu_read_lock(),rcu_read_unlock()只是禁止和启用抢占.因为在读者临界区,不允许发生上下文切换.


rcu_dereference()和rcu_assign_pointer()的实现如下:
#define rcu_dereference(p)     ({ /
                typeof(p) _________p1 = ACCESS_ONCE(p); /
                smp_read_barrier_depends(); /
                (_________p1); /
                })
#define rcu_assign_pointer(p, v) /
    ({ /
        if (!__builtin_constant_p(v) || /
            ((v) != NULL)) /
            smp_wmb(); /
        (p) = (v); /
    })
它们的实现也很简单.因为它们本身都是原子操作.因为只是为了cache一致性,插上了内存屏障.可以让其它的读者/写者可以看到保护指针的最新值.


synchronize_rcu()在RCU中是一个最核心的函数,它用来等待之前的读者全部退出.我们后面的大部份分析也是围绕着它而进行.实现如下:
void synchronize_rcu(void)
{
    struct rcu_synchronize rcu;


    init_completion(&rcu.completion);
    /* Will wake me after RCU finished */
    call_rcu(&rcu.head, wakeme_after_rcu);


    /* Wait for it */
    wait_for_completion(&rcu.completion);
}
我们可以看到,它初始化了一个本地变量,它的类型为struct rcu_synchronize.调用call_rcu()之后.一直等待条件变量rcu.competion的满足.


在这里看到了RCU的另一个核心API,它就是call_run().它的定义如下:
void call_rcu(struct rcu_head *head,
                void (*func)(struct rcu_head *rcu))


它用来等待之前的读者操作完成之后,就会调用函数func.
我们也可以看到,在synchronize_rcu()中,读者操作完了要调用的函数就是wakeme_after_rcu().
另外,call_rcu()用在不可睡眠的条件中,如果中断环境,禁止抢占环境等.而synchronize_rcu()用在可睡眠的环境下.先跟踪看一下wakeme_after_rcu():
static void wakeme_after_rcu(struct rcu_head  *head)
{
    struct rcu_synchronize *rcu;


    rcu = container_of(head, struct rcu_synchronize, head);
    complete(&rcu->completion);
}
我们可以看到,该函数将条件变量置真,然后唤醒了在条件变量上等待的进程.


看下call_rcu():
void call_rcu(struct rcu_head *head,
                void (*func)(struct rcu_head *rcu))
{
    unsigned long flags;
    struct rcu_data *rdp;


    head->func = func;
    head->next = NULL;
    local_irq_save(flags);
    rdp = &__get_cpu_var(rcu_data);
    *rdp->nxttail = head;
    rdp->nxttail = &head->next;
    if (unlikely(++rdp->qlen > qhimark)) {
        rdp->blimit = INT_MAX;
        force_quiescent_state(rdp, &rcu_ctrlblk);
    }
    local_irq_restore(flags);
}
该函数也很简单,就是将head加在了per_cpu变量rcu_data的tail链表上.
Rcu_data定义如下:
DEFINE_PER_CPU(struct rcu_data, rcu_data) = { 0L };
由此,我们可以得知,每一个CPU都有一个rcu_data.每个调用call_rcu()/synchronize_rcu()进程所代表的head都会挂到rcu_data的tail链表上.


那究竟怎么去判断当前的写者已经操作完了呢?我们在之前看到,不是读者在调用rcu_read_lock()的时候要禁止抢占么?因此,我们只需要判断如有的CPU都进过了一次上下文切换,就说明所有读者已经退出了.
引用<< Linux 2.6内核中新的锁机制--RCU >>( (http://www.ibm.com/developerworks/cn/linux/l-rcu/)中有关这个过程的描述:
“等待适当时机的这一时期称为grace period,而CPU发生了上下文切换称为经历一个quiescent state,grace period就是所有CPU都经历一次quiescent state所需要的等待的时间。垃圾收集器就是在grace period之后调用写者注册的回调函数来完成真正的数据修改或数据释放操作的”


要彻底弄清楚这个问题,我们得从RCU的初始化说起.
四:从RCU的初始化说起
RCU的初始化位于start_kernel()àrcu_init().代码如下:
void __init rcu_init(void)
{
    __rcu_init();
}


void __init __rcu_init(void)
{
    rcu_cpu_notify(&rcu_nb, CPU_UP_PREPARE,
            (void *)(long)smp_processor_id());
    /* Register notifier for non-boot CPUs */
    register_cpu_notifier(&rcu_nb);
}
Reqister_cpu_notifier()是关于通知链表的操作,可以忽略不看.
跟进rcu_cpu_notify():
static int __cpuinit rcu_cpu_notify(struct notifier_block *self,
                unsigned long action, void *hcpu)
{
    long cpu = (long)hcpu;


    switch (action) {
    case CPU_UP_PREPARE:
    case CPU_UP_PREPARE_FROZEN:
        rcu_online_cpu(cpu);
        break;
    case CPU_DEAD:
    case CPU_DEAD_FROZEN:
        rcu_offline_cpu(cpu);
        break;
    default:
        break;
    }
    return NOTIFY_OK;
}
注意到,在__rcu_init()中是以CPU_UP_PREPARE为参数调用此函数,对应流程转入rcu_online_cpu中:
static void __cpuinit rcu_online_cpu(int cpu)
{
    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
    struct rcu_data *bh_rdp = &per_cpu(rcu_bh_data, cpu);


    rcu_init_percpu_data(cpu, &rcu_ctrlblk, rdp);
    rcu_init_percpu_data(cpu, &rcu_bh_ctrlblk, bh_rdp);
    open_softirq(RCU_SOFTIRQ, rcu_process_callbacks, NULL);
}
我们从这里又看到了另一个per_cpu变量,rcu_bh_data.有关bh的部份之后再来分析.在这里略过这些部份.
Rcu_init_percpu_data()如下:
static void rcu_init_percpu_data(int cpu, struct rcu_ctrlblk *rcp,
                        struct rcu_data *rdp)
{
    memset(rdp, 0, sizeof(*rdp));
    rdp->curtail = &rdp->curlist;
    rdp->nxttail = &rdp->nxtlist;
    rdp->donetail = &rdp->donelist;
    rdp->quiescbatch = rcp->completed;
    rdp->qs_pending = 0;
    rdp->cpu = cpu;
    rdp->blimit = blimit;
}
调用这个函数的第二个参数是一个全局变量rcu_ctlblk.定义如下:
static struct rcu_ctrlblk rcu_ctrlblk = {
    .cur = -300,
    .completed = -300,
    .lock = __SPIN_LOCK_UNLOCKED(&rcu_ctrlblk.lock),
    .cpumask = CPU_MASK_NONE,
};
static struct rcu_ctrlblk rcu_bh_ctrlblk = {
    .cur = -300,
    .completed = -300,
    .lock = __SPIN_LOCK_UNLOCKED(&rcu_bh_ctrlblk.lock),
    .cpumask = CPU_MASK_NONE,
};
在rcu_init_percpu_data中,初始化了三个链表,分别是taillist,curlist和donelist.另外, 将rdp->quiescbatch 赋值为 rcp->completed.这个是一个很重要的操作.
Rdp-> quiescbatch表示rcu_data已经完成的grace period序号(在代码中也被称为了batch),rcp->completed表示全部变量rcu_ctrlblk计数已经完成的grace period序号.将rdp->quiescbatch = rcp->completed;,表示不需要等待grace period.


回到rcu_online_cpu()中:
open_softirq(RCU_SOFTIRQ, rcu_process_callbacks, NULL);
初始化了RCU_SOFTIRQ类型的软中断.但这个软中断什么时候被打开,还需要之后来分析.
之后,每个CPU的初始化都会经过start_kernel()->rcu_init().相应的,也为每个CPU初始化了RCU的相关结构.


五:等待RCU读者操作完成
之前,我们看完了RCU的初始化,现在可以来看一下RCU如何来判断当前的RCU读者已经退出了.
在每一次进程切换的时候,都会调用rcu_qsctr_inc().如下代码片段如示:
asmlinkage void __sched schedule(void)
{
    ......
    ......
    rcu_qsctr_inc(cpu);
    ......
}
Rcu_qsctr_inc()代码如下:
static inline void rcu_qsctr_inc(int cpu)
{
    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
    rdp->passed_quiesc = 1;
}
该函数将对应CPU上的rcu_data的passed_quiesc成员设为了1.
或许你已经发现了,这个过程就标识该CPU经过了一次quiescent state.没错:-)


另外,在时钟中断中,会进行以下操作:
void update_process_times(int user_tick)
{
    ......
    ......


    if (rcu_pending(cpu))
        rcu_check_callbacks(cpu, user_tick);
    ......
    ......
}
在每一次时钟中断,都会检查是否有需要更新的RCU需要处理,如果有,就会为其调用rcu_check_callbacks().


Rcu_pending()的代码如下:
int rcu_pending(int cpu)
{
    return __rcu_pending(&rcu_ctrlblk, &per_cpu(rcu_data, cpu)) ||
        __rcu_pending(&rcu_bh_ctrlblk, &per_cpu(rcu_bh_data, cpu));
}
同上面一样,忽略bh的部份.
static int __rcu_pending(struct rcu_ctrlblk *rcp, struct rcu_data *rdp)
{
    /* This cpu has pending rcu entries and the grace period
     * for them has completed.
     */
    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch))
        return 1;


    /* This cpu has no pending entries, but there are new entries */
    if (!rdp->curlist && rdp->nxtlist)
        return 1;


    /* This cpu has finished callbacks to invoke */
    if (rdp->donelist)
        return 1;


    /* The rcu core waits for a quiescent state from the cpu */
    if (rdp->quiescbatch != rcp->cur || rdp->qs_pending)
        return 1;


    /* nothing to do */
    return 0;
}
上面有四种情况会返回1,分别对应:
1:该CPU上有等待处理的回调函数,且已经经过了一个batch(grace period).rdp->datch表示rdp在等待的batch序号
2:上一个等待已经处理完了,又有了新注册的回调函数.
3:等待已经完成,但尚末调用该次等待的回调函数.
4:在等待quiescent state.
关于rcp和rdp结构中成员的含义,我们等用到的时候再来分析.


如果rcu_pending返回1,就会进入到rcu_check_callbacks().代码如下:
void rcu_check_callbacks(int cpu, int user)
{
    if (user ||
        (idle_cpu(cpu) && !in_softirq() &&
                hardirq_count() <= (1 << HARDIRQ_SHIFT))) {
        rcu_qsctr_inc(cpu);
        rcu_bh_qsctr_inc(cpu);
    } else if (!in_softirq())
        rcu_bh_qsctr_inc(cpu);
    raise_rcu_softirq();
}
如果已经CPU中运行的进程是用户空间进程或者是CPU空闲且不处于中断环境,那么,它也已经进过了一次切换.注意,RCU只能在内核空间使用.
最后调用raise_rcu_softirq()打开了软中断处理.相应的,也就调用RCU的软中断处理函数.结合上面分析的初始化流程,软中断的处理函数为rcu_process_callbacks().
代码如下:
static void rcu_process_callbacks(struct softirq_action *unused)
{
    __rcu_process_callbacks(&rcu_ctrlblk, &__get_cpu_var(rcu_data));
    __rcu_process_callbacks(&rcu_bh_ctrlblk, &__get_cpu_var(rcu_bh_data));
}


在阅读__rcu_process_callbacks()之前,先来了解一下rdp中几个链表的含义:
每次新注册的回调函数,都会链入到rdp->taillist.
当前等待grace period完成的函数都会链入到rdp->curlist上.
到等待的grace period已经到来,就会将curlist上的链表移到donelist上.
当一个grace period过了之后,就会将taillist上的数据移到rdp->curlist上.之后加册的回调函数又会将其加到rdp->taillist上.


__rcu_process_callbacks()代码分段分析如下:
static void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,
                    struct rcu_data *rdp)
{
    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch)) {
        *rdp->donetail = rdp->curlist;
        rdp->donetail = rdp->curtail;
        rdp->curlist = NULL;
        rdp->curtail = &rdp->curlist;
    }


    如果有需要处理的回调函数,且已经经过了一次grace period.就将curlist上的数据移到donetlist上.
其中,crp->completed表示已经完成的grace period.rdp->batch表示该CPU正在等待的grace period序号.


    if (rdp->nxtlist && !rdp->curlist) {
        local_irq_disable();
        rdp->curlist = rdp->nxtlist;
        rdp->curtail = rdp->nxttail;
        rdp->nxtlist = NULL;
        rdp->nxttail = &rdp->nxtlist;
        local_irq_enable();


        /*
         * start the next batch of callbacks
         */


        /* determine batch number */
        rdp->batch = rcp->cur + 1;
        /* see the comment and corresponding wmb() in
         * the rcu_start_batch()
         */
        smp_rmb();


        if (!rcp->next_pending) {
            /* and start it/schedule start if it's a new batch */
            spin_lock(&rcp->lock);
            rcp->next_pending = 1;
            rcu_start_batch(rcp);
            spin_unlock(&rcp->lock);
        }
    }
如果上一个等待的回调函数处理完了,而且又有了新注册的回调函数.就将taillist上的数据移动到curlist上.并开启新的grace period等待.
注意里面几个变量的赋值:
rdp->batch = rcp->cur + 1表示该CPU等待的grace period置为当前已发生grace period序号的下一个.
每次启动一个新的grace period等待之后,就会将rcp->next_pending.在启动的过程中,也就是rcu_start_batch()的过程中,会将rcp->next_pending置为1.设置这个变量主要是防止多个写者竞争的情况


    //更新相关信息
    rcu_check_quiescent_state(rcp, rdp);
    //处理等待完成的回调函数
    if (rdp->donelist)
        rcu_do_batch(rdp);
}
接着,更新相关的信息,例如,判断当前CPU是否进行了quiescent state.或者grace period是否已经完成.
最后再处理挂在rdp->donelist上的链表.


这里面有几个子函数值得好好分析,分别分析如下:
第一个要分析的是rcu_start_batch():
static void rcu_start_batch(struct rcu_ctrlblk *rcp)
{
    if (rcp->next_pending &&
            rcp->completed == rcp->cur) {
        rcp->next_pending = 0;
        smp_wmb();
        rcp->cur++;
        smp_mb();
        cpus_andnot(rcp->cpumask, cpu_online_map, nohz_cpu_mask);


        rcp->signaled = 0;
    }
}
这个函数的代码虽然很简单,但隐藏了很多玄机.
每次启动一个新的grace period等待的时候就将rcp->cur加1,将rcp->cpumask中,将存在的CPU的位置1.
其中,if判断必须要满足二个条件:
第一:rcp->next_pending必须为1.我们把这个函数放到__rcu_process_callbacks()这个大环境中看一下:
static void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,
                    struct rcu_data *rdp)
{
    ......
    ......
    if (rdp->nxtlist && !rdp->curlist) {
        ......
        if (!rcp->next_pending) {
            /* and start it/schedule start if it's a new batch */
            spin_lock(&rcp->lock);
            rcp->next_pending = 1;
            rcu_start_batch(rcp);
            spin_unlock(&rcp->lock);
        }
    }
}
首先,rcp->next_pending为0才会调用rcu_start_batch()启动一个新的进程.然后,将rcp->next_pending置为1,再调用rcu_start_batch().在这里要注意中间的自旋锁.然后在rcu_start_batch()中,再次判断rcp->next_pending为1后,再进行后续操作,并将rcp->next_pending置为0.
为什么这里需要这样的判断呢? 如果其它CPU正在开启一个新的grace period等待,那就用不着再次开启一个新的等待了,直接返回即可.


第二: rcu_start_batch()中if要满足的第二个条件为rcp->completed == rcp->cur.也就是说前面的grace period全部都完成了.每次开启新等待的时候都会将rcp->cur加1.每一个等待完成之后,都会将rc-> completed等于rcp->cur.


第二个要分析的函数是rcu_check_quiescent_state().代码如下:
static void rcu_check_quiescent_state(struct rcu_ctrlblk *rcp,
                    struct rcu_data *rdp)
{
    if (rdp->quiescbatch != rcp->cur) {
        /* start new grace period: */
        rdp->qs_pending = 1;
        rdp->passed_quiesc = 0;
        rdp->quiescbatch = rcp->cur;
        return;
    }


    /* Grace period already completed for this cpu?
     * qs_pending is checked instead of the actual bitmap to avoid
     * cacheline trashing.
     */
    if (!rdp->qs_pending)
        return;


    /*
     * Was there a quiescent state since the beginning of the grace
     * period? If no, then exit and wait for the next call.
     */
    if (!rdp->passed_quiesc)
        return;
    rdp->qs_pending = 0;


    spin_lock(&rcp->lock);
    /*
     * rdp->quiescbatch/rcp->cur and the cpu bitmap can come out of sync
     * during cpu startup. Ignore the quiescent state.
     */
    if (likely(rdp->quiescbatch == rcp->cur))
        cpu_quiet(rdp->cpu, rcp);


    spin_unlock(&rcp->lock);
}
首先,如果rdp->quiescbatch != rcp->cur.则说明又开启了一个新的等待,因此需要重新处理这个等待,首先将rdp->quiescbatch 更新为rcp->cur.然后,使rdp->qs_pending为1.表示有等待需要处理. passed_quiesc也被清成了0.
然后,再判断rdp->passed_quiesc是否为真,记得我们在之前分析过,在每次进程切换或者进程切换的时候,都会调用rcu_qsctr_inc().该函数会将rdp->passed_quiesc置为1.
因此,在这里判断这个值是为了检测该CPU上是否发生了上下文切换.
之后,就是一段被rcp->lock保护的一段区域.如果还是等待没有发生改变,就会调用cpu_quiet(rdp->cpu, rcp)将该CPU位清零.如果是一个新的等待了,就用不着清了,因为需要重新判断该CPU上是否发生了上下文切换.


cpu_quiet()函数代码如下:
static void cpu_quiet(int cpu, struct rcu_ctrlblk *rcp)
{
    cpu_clear(cpu, rcp->cpumask);
    if (cpus_empty(rcp->cpumask)) {
        /* batch completed ! */
        rcp->completed = rcp->cur;
        rcu_start_batch(rcp);
    }
}
它清除当前CPU对应的位,如果CPMMASK为空,对应所有的CPU都发生了进程切换,就会将rcp->completed = rcp->cur.并且根据需要是否开始一个grace period等待.


最后一个要分析的函数是rcu_do_batch().它进行的是清尾的工作.如果等待完成了,那就必须要处理donelist链表上挂载的数据了.代码如下:
static void rcu_do_batch(struct rcu_data *rdp)
{
    struct rcu_head *next, *list;
    int count = 0;


    list = rdp->donelist;
    while (list) {
        next = list->next;
        prefetch(next);
        list->func(list);
        list = next;
        if (++count >= rdp->blimit)
            break;
    }
    rdp->donelist = list;


    local_irq_disable();
    rdp->qlen -= count;
    local_irq_enable();
    if (rdp->blimit == INT_MAX && rdp->qlen <= qlowmark)
        rdp->blimit = blimit;


    if (!rdp->donelist)
        rdp->donetail = &rdp->donelist;
    else
        raise_rcu_softirq();
}
它遍历处理挂在链表上的回调函数.在这里,注意每次调用的回调函数有最大值限制.这样做主要是防止一次调用过多的回调函数而产生不必要系统负载.如果donelist中还有没处理完的数据,打开RCU软中断,在下次软中断到来的时候接着处理.


五:几种RCU情况分析
1:如果CPU 1上有进程调用rcu_read_lock进入临界区,之后退出来,发生了进程切换,新进程又通过rcu_read­_lock进入临界区.由于RCU软中断中只判断一次上下文切换,因此,在调用回调函数的时候,仍然有进程处于RCU的读临界区,这样会不会有问题呢?
这样是不会有问题的.还是上面的例子:
        spin_lock(&foo_mutex);
        old_fp = gbl_foo;
        *new_fp = *old_fp;
        new_fp->a = new_a;
        rcu_assign_pointer(gbl_foo, new_fp);
        spin_unlock(&foo_mutex);
        synchronize_rcu();
        kfree(old_fp);


使用synchronize_rcu ()只是为了等待持有old_fd(也就是调用rcu_assign_pointer ()更新之前的gbl_foo)的进程退出.而不需要等待所有的读者全部退出.这是因为,在rcu_assign_pointer ()之后的读取取得的保护指针,已经是更新好的新值了.


2:上面分析的似乎是针对有挂载链表的CPU而言的,那对于只调用rcu_read_lock()的CPU,它们是怎么处理的呢?
首先,每次启动一次等待,肯定是会更新rcp->cur的.因此,在rcu_pending()的判断中,下面语句会被满足:
    if (rdp->quiescbatch != rcp->cur || rdp->qs_pending)
        return 1;
因此会进入到RCU的软中断.在软中断处理中:
rcu_process_callbacks() à __rcu_process_callbacks() àrcu_check_quiescent_state()
中,如果该CPU上有进程切换,就会各新rcp中的CPU 掩码数组.


3:如果一个CPU连续调用synchronize_rcu()或者call_rcu()它们会有什么影响呢?
如果当前有请求在等待,就会新请提交的回调函数挂到taillist上,一直到前一个等待完成,再将taillist的数据移到curlist,并开启一个新的等待,因此,也就是说,在前一个等待期间提交的请求,都会放到一起处理.也就是说,他们会共同等待所有CPU切换完成.
举例说明如下:
假设grace period时间是12ms.在12ms内,先后有A,B,C进程提交请求.
那系统在等待处理完后,交A,B,C移到curlist中,开始一个新的等待.


六:有关rcu_read_lock_bh()/rcu_read_unlock_bh()/call_rcu_bh().
在上面的代码分析的时候,经常看到带有bh的RCU代码.现在来看一下这些带bh的RCU是什么样的.
#define rcu_read_lock_bh() __rcu_read_lock_bh()
#define rcu_read_unlock_bh() __rcu_read_unlock_bh()


#define __rcu_read_lock_bh() /
    do { /
        local_bh_disable(); /
        __acquire(RCU_BH); /
        rcu_read_acquire(); /
    } while (0)
#define __rcu_read_unlock_bh() /
    do { /
        rcu_read_release(); /
        __release(RCU_BH); /
        local_bh_enable(); /
    } while (0)
根据上面的分析:bh RCU跟普通的RCU相比不同的是,普通RCU是禁止内核抢占,而bh RCU是禁止下半部.
其实,带bh的RCU一般在软中断使用,不过计算quiescent state并不是发生一次上下文切换.而是发生一次softirq.我们在后面的分析中可得到印证.


Call_rcu_bh()代码如下:
void call_rcu_bh(struct rcu_head *head,
                void (*func)(struct rcu_head *rcu))
{
    unsigned long flags;
    struct rcu_data *rdp;


    head->func = func;
    head->next = NULL;
    local_irq_save(flags);
    rdp = &__get_cpu_var(rcu_bh_data);
    *rdp->nxttail = head;
    rdp->nxttail = &head->next;


    if (unlikely(++rdp->qlen > qhimark)) {
        rdp->blimit = INT_MAX;
        force_quiescent_state(rdp, &rcu_bh_ctrlblk);
    }


    local_irq_restore(flags);
}
它跟call_rcu()不相同的是,rcu是取per_cpu变量rcu__data和全局变量rcu_ctrlblk.而bh RCU是取rcu_bh_data,rcu_bh_ctrlblk.他们的类型都是一样的,这样做只是为了区分BH和普通RCU的等待.


对于rcu_bh_qsctr_inc
static inline void rcu_bh_qsctr_inc(int cpu)
{
    struct rcu_data *rdp = &per_cpu(rcu_bh_data, cpu);
    rdp->passed_quiesc = 1;
}
它跟rcu_qsctr_inc()机同,也是更改对应成员.


所不同的是,调用rcu_bh_qsctr_inc()的地方发生了变化.
asmlinkage void __do_softirq(void)
{
    ......
        do {
        if (pending & 1) {
            h->action(h);
            rcu_bh_qsctr_inc(cpu);
        }
        h++;
        pending >>= 1;
    } while (pending);
    ......
}
也就是说,在发生软中断的时候,才会认为是经过了一次quiescent state.


七:小结
基本上,RCU的代码架构不难,难的是对它的临界条件分析.作者的思路也相当缜密.值得花时间仔细研读.
RCU机制是Linux2.6之后提供的一种数据一致性访问的机制,从RCUread-copy-update)的名称上看,我们就能对他的实现机制有一个大概的了解,在修改数据的时候,首先需要读取数据,然后生成一个副本,对副本进行修改,修改完成之后再将老数据update成新的数据,此所谓RCU

 


       在操作系统中,数据一致性访问是一个非常重要的部分,通常我们可以采用锁机制实现数据的一致性访问。例如,semaphorespinlock机制,在访问共享数据时,首先访问锁资源,在获取锁资源的前提下才能实现数据的访问。这种原理很简单,根本的思想就是在访问临界资源时,首先访问一个全局的变量(锁),通过全局变量的状态来控制线程对临界资源的访问。但是,这种思想是需要硬件支持的,硬件需要配合实现全局变量(锁)的读-修改-写,现代CPU都会提供这样的原子化指令。采用锁机制实现数据访问的一致性存在如下两个问题:


1、  效率问题。锁机制的实现需要对内存的原子化访问,这种访问操作会破坏流水线操作,降低了流水线效率。这是影响性能的一个因素。另外,在采用读写锁机制的情况下,写锁是排他锁,无法实现写锁与读锁的并发操作,在某些应用下回降低性能。


2、  扩展性问题。当系统中CPU数量增多的时候,采用锁机制实现数据的同步访问效率偏低。并且随着CPU数量的增多,效率降低,由此可见锁机制实现的数据一致性访问扩展性差。


为了解决上述问题,Linux中引进了RCU机制。该机制在多CPU的平台上比较适用,对于读多写少的应用尤其适用。RCU的思路实际上很简单,下面对其进行描述:


从上述分析来看,RCU思想是比较简单的,其核心内容紧紧围绕“写时拷贝”,采用RCU机制,能够保证在读写操作共享资源时,基本不需要取锁操作,能够在一定程度上提升性能。但是该机制的应用是有条件的,对于读多写少的应用,机制的开销比较小,性能会大幅度提升,但是如果写操作较多时,开销将会增大,性能不一定会有所提升。总体来说,RCU机制是对rw_lock的一种优化。




1、             对于读操作,可以直接对共享资源进行访问,但是前提是需要CPU支持访存操作的原子化,现代CPU对这一点都做了保证。但是RCU的读操作上下文是不可抢占的(这一点在下面解释),所以读访问共享资源时可以采用read_rcu_lock(),该函数的工作是停止抢占。


2、             对于写操作,其需要将原来的老数据作一次备份(copy),然后对备份数据进行修改,修改完毕之后再用新数据更新老数据,更新老数据时采用了rcu_assign_pointer()宏,在该函数中首先屏障一下memory,然后修改老数据。这个操作完成之后,需要进行老数据资源的回收。操作线程向系统注册回收方法,等待回收。采用数据备份的方法可以实现读者与写者之间的并发操作,但是不能解决多个写着之间的同步,所以当存在多个写者时,需要通过锁机制对其进行互斥,也就是在同一时刻只能存在一个写者。


3、             RCU机制中存在一个垃圾回收的daemon,当共享资源被update之后,可以采用该daemon实现老数据资源的回收。回收时间点就是在update之前的所有的读者全部退出。由此可见写者在update之后是需要睡眠等待的,需要等待读者完成操作,如果在这个时刻读者被抢占或者睡眠,那么很可能会导致系统死锁。因为此时写者在等待读者,读者被抢占或者睡眠,如果正在运行的线程需要访问读者和写者已经占用的资源,那么死锁的条件就很有可能形成了


共1条 1/1 1 跳转至

回复

匿名不能发帖!请先 [ 登陆 注册 ]