Linux内核 RCU锁(一)

好久没有更文，上次更文时西安天气还很热，现在“寒气”它还真来了。在前一阶段经历了一些公司的面试，经常会问到RCU锁的原理，其实在跟对方口述表达时才真正能体现出来自己到底懂不懂，关于RCU锁的原理与使用，我打算分若干个次文章整理出来， 本次就先从一个大概的原理上进行讲解。

Read-Copy Update，简称RCU，中文对应"读取-拷贝-更新"，先给出一个解释：对于被RCU保护的共享数据结构，读者不需要获得任何锁就可以直接访问，但写者在访问它时首先要拷贝一个副本，然后对副本进行修改，最后使用一个回调机制在适当的时机把指向数据的指针重新指向新的被修改的数据。

简化来说：

1. 记录指针：记录所有的指向“共享数据”的指针。
2. 读取-拷贝： " 指针持有者 “ 修改该 ” 共享数据 " ,：先创建一个共享数据 " 副本 " , 然后在副本中修改 。
3. 更新数据：读者读取”共享数据“，离开”读者临界区“后，指向原来 " 共享数据 " 的 指针 重新指向 " 副本 " , 然后再回收处理旧的 " 共享数据 " 。

RCU锁优劣：

• 读者不需要承担同步开销(同步开销：1、获取锁；2、执行”原子指令“；3、执行”内存屏障“），因为读端不需要锁，不使用原子指令，故不会导致锁竞争。
• 写者承担很大的同步开销，需要读取并复制共享数据，还有使用互斥锁机制等。

关于具体场景：RCU锁是 Linux 内核实现的一种针对“读多写少”的共享数据的同步机制。RCU主要针对的数据对象是链表，目的是提高遍历读取数据的效率，为了达到目的使用RCU机制读取数据的时候不对链表进行耗时的加锁操作。RCU机制极大提高"链表"数据结构的读取效率，多个线程同时读取链表时，使用rcu_read_lock()即可，在多线程读取的同时还允许有1个线程修改链表。

在Linux内核中专门提供了头文件：include/linux/rculist.h定义了一些宏函数用于RCU处理链表，如下表中是该头文件中的宏定义.在内核编程时可根据需要查询该头文件中源码选择，如list_entry_rcu与list_for_each_entry_rcu：

#define list_entry_rcu(ptr, type, member) \
 container_of(READ_ONCE(ptr), type, member)

#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))

list_for_each_entry_rcu用于遍历由RCU保护的链表head，只要在读端临界区使用该函数，它就可以安全地和其它_rcu链表操作函数（如list_add_rcu）并发运行。

RCU链表遍历操作相关宏：

RCU list 遍历

list_entry_rcu list_first_entry_rcu list_next_rcu list_for_each_entry_rcu list_for_each_entry_continue_rcu list_for_each_entry_from_rcu hlist_first_rcu hlist_next_rcu hlist_pprev_rcu hlist_for_each_entry_rcu hlist_for_each_entry_rcu_bh hlist_for_each_entry_from_rcu hlist_for_each_entry_continue_rcu hlist_for_each_entry_continue_rcu_bh hlist_nulls_first_rcu hlist_nulls_for_each_entry_rcu hlist_bl_first_rcu hlist_bl_for_each_entry_rcu

直接上代码来个简单的Demo:仅创建一个读者和写者去感受一下RCU锁的使用,下面的例子通过RCU机制保护rcu_test_init()函数分配的共享数据结构struct foo *test ;并创建一个读者和一个写者来模拟同步场景。

#include <linux/module.h>
#include <linux/init.h>
#include <linux/slab.h>
#include <linux/kthread.h>
#include <linux/rcupdate.h>
#include <linux/delay.h>
#include <linux/string.h>

struct foo {
 int a;
 int b;
 int c;
 struct rcu_head rcu;
 struct list_head list;
};
static struct foo *test = NULL;

LIST_HEAD(g_rcu_list);

struct task_struct *rcu_reader;
struct task_struct *rcu_updater;

static int rcu_reader_list(void *data)
{
 struct foo *p = NULL;
 int cnt = 100;

 while (cnt-->=0) {
  msleep(100);
  rcu_read_lock();
  list_for_each_entry_rcu(p, &g_rcu_list, list) {
   pr_info("%s: a = %d, b = %d, c = %d\n",
     __func__, p->a, p->b, p->c);
  }
  rcu_read_unlock();
 }

 return 0;
}

static int rcu_updater_list(void *data)
{
 int cnt = 100;
 int value = 1000;
 while (cnt-->=0) {
  msleep(100);
  struct foo *p = list_first_or_null_rcu(&g_rcu_list, struct foo, list);
  struct foo *q = (struct foo *)kzalloc(sizeof(struct foo), GFP_KERNEL);
  *q = *p;
  q->a = value;
  q->b = value + 1;
  q->c = value + 2;

  list_replace_rcu(&p->list, &q->list);

  pr_info("%s: a = %d, b = %d, c = %d\n",
    __func__, q->a, q->b, q->c);
  synchronize_rcu();
  kfree(p);
  value++; 
 }

 return 0;
}

static int rcu_test_init(void)
{
 struct foo *p;
 rcu_reader = kthread_run(rcu_reader_list, NULL, "rcu_reader_list");
 rcu_updater = kthread_run(rcu_updater_list, NULL, "rcu_updater_list"); 
 test = (struct foo *)kzalloc(sizeof(struct foo), GFP_KERNEL);

 p = (struct foo *)kzalloc(sizeof(struct foo), GFP_KERNEL);
 list_add_rcu(&p->list, &g_rcu_list);

 return 0;
}

static void rcu_test_exit(void)
{ 
 pr_info("test-end\n");
}

module_init(rcu_test_init);
module_exit(rcu_test_exit);

MODULE_AUTHOR("dongxu");
MODULE_LICENSE("GPL");

对于读线程：

• 通过rcu_read_lock()函数和rcu_read_unlock()函数来构建一个读者临界区，rcu_read_lock() 和 rcu_read_unlock()，是 RCU “随意读” 的关键，它们的效果是声明了一个读端的临界区。读者在临界区中,不能发生进程上下文切换，否则,因为写者需要要等待读者完成,写者进程也会一直被阻塞。
• 调用list_for_each_entry_rcu宏函数，遍历获取被保护数据，此时P指向被保护的数据

对于写线程：

• 调用list_first_or_null_rcu宏函数，读取元素，然后开始复制一份副本。
• 对副本进行修改操作。
• 调用list_replace_rcu宏函数，用新节点替换掉旧节点，实际也是调用了rcu_assign_pointer()更新了元素，rcu_assign_pointer用来为被RCU保护的指针分配一个新的值，这样是为了安全更改其值，这个原语保护并发读不受更新操作的影响。写者调用rcu_assign_pointer后，对于读者就"可见"了，调用rcu_assign_pointer前就已经开始读取旧值的依然可以访问旧值。
• 调用synchronize_rcu宏函数，为了确保没有读者正在访问要回收的临界资源，需要等待所有的读者退出临界区，该宏函数通过阻塞来做到这一点，直到所有cpu上所有预先存在的RCU读端临界区都完成，相当于给读者一个安全退出的宽限区。
• kfree释放旧数据。

以上分析与记录的相关概念都比较简单，RCU的实现很复杂，本文对一些细节没有展开，如回收旧资源时的宽限区等，所以 本次只是一个原理层面一个大概的分析，后面有时间会继续分析。