linux 内核经典RCU

如果不关心使用的RCU是不可抢占RCU还是可抢占RCU，应该使用经典RCU的编程接口。最初的经典RCU是不可抢占RCU，后来实现了可抢占RCU，经典RCU的意思发生了变化：如果内核编译了可抢占RCU，那么经典RCU的编程接口被实现为可抢占RCU，否则被实现为不可抢占RCU。
读者使用函数rcu_read_lock()标记进入读端临界区，使用函数rcu_read_unlock()标记退出读端临界区。读端临界区可以嵌套。
在读端临界区里面应该使用宏rcu_dereference(p)访问指针，这个宏封装了数据依赖屏障，即只有阿尔法处理器需要的读内存屏障。
写者可以使用下面4个函数。
（1）使用函数synchronize_rcu()等待宽限期结束，即所有读者退出读端临界区，然后写者执行下一步操作。这个函数可能睡眠。
（2）使用函数synchronize_rcu_expedited()等待宽限期结束。和函数synchronize_rcu()的区别是：该函数会向其他处理器发送处理器间中断（Inter-Processor Interrupt，IPI）请求，强制宽限期快速结束。我们把强制快速结束的宽限期称为加速宽限期（expedited grace period），把没有强制快速结束的宽限期称为正常宽限期（normal grace period）。
（3）使用函数call_rcu()注册延后执行的回调函数，把回调函数添加到RCU回调函数链表中，立即返回，不会阻塞。函数原型如下：
void call_rcu(struct rcu_head *head, rcu_callback_t func);
struct callback_head {
struct callback_head *next;
void (*func)(struct callback_head *head);
} __attribute__((aligned(sizeof(void *))));
#define rcu_head callback_head
typedef void (*rcu_callback_t)(struct rcu_head *head);
（4）使用函数rcu_barrier()等待使用call_rcu注册的所有回调函数执行完。这个函数可能睡眠。
现在举例说明使用方法，假设链表节点和头节点如下：
typedef struct {
struct list_head link;
struct rcu_head rcu;
int key;
int val;
} test_entry;
struct list_head test_head;
成员“struct rcu_head rcu”：调用函数call_rcu把回调函数添加到RCU回调函数链表的时候需要使用。
读者访问链表的方法如下：
int test_read(int key, int *val_ptr)
{
test_entry *entry;
int found = 0;
rcu_read_lock();
list_for_each_entry_rcu(entry, &test_head, link) {
if (entry->key == key) {
*val_ptr = entry->val;
found = 1;
break;
}
}
rcu_read_unlock();
return found;
}
如果只有一个写者，写者不需要使用锁，添加、更新和删除3种操作的实现方法如下。
（1）写者添加一个节点到链表尾部。
void test_add_node(test_entry *entry)
{
list_add_tail_rcu(&entry->link, &test_head);
}
（2）写者更新一个节点。
更新的过程是：首先把旧节点复制更新，然后使用新节点替换旧节点，最后使用函数call_rcu注册回调函数，延后释放旧节点。
int test_update_node(int key, int new_val)
{
test_entry *entry, *new_entry;
int ret;
ret = -ENOENT;
list_for_each_entry(entry, &test_head, link) {
if (entry->key == key) {
new_entry = kmalloc(sizeof(test_entry), GFP_ATOMIC);
if (new_entry == NULL) {
ret = -ENOMEM;
break;
}
*new_entry = *entry;
new_entry->val = new_val;
list_replace_rcu(&entry->list, &new_entry->list);
call_rcu(&entry->rcu, test_free_node);
ret = 0;
break;
}
}
return ret;
}
static void test_free_node(struct rcu_head *head)
{
test_entry *entry = container_of(head, test_entry, rcu);
kfree(entry);
}
（3）写者删除一个节点的第一种实现方法是：首先把节点从链表中删除，然后使用函数call_rcu注册回调函数，延后释放节点。
int test_del_node(int key)
{
test_entry *entry;
int found = 0;
list_for_each_entry(entry, &test_head, link) {
if (entry->key == key) {
list_del_rcu(&entry->link);
call_rcu(&entry->rcu, test_free_node);
found = 1;
break;
}
}
return found;
}
static void test_free_node(struct rcu_head *head)
{
test_entry *entry = container_of(head, test_entry, rcu);
kfree(entry);
}
（4）写者删除一个节点的第二种实现方法是：首先把节点从链表中删除，然后使用函数synchronize_rcu()等待宽限期结束，最后释放节点。
int test_del_node(int key)
{
test_entry *entry;
int found = 0;
list_for_each_entry(entry, &test_head, link) {
if (entry->key == key) {
list_del_rcu(&entry->link);
synchronize_rcu();
kfree(entry);
found = 1;
break;
}
}
return found;
}
如果有多个写者，那么写者之间必须使用锁互斥，添加、更新和删除3种操作的实现方法如下。
（1）写者添加一个节点到链表尾部。假设使用自旋锁“test_lock”保护链表。
void test_add_node(test_entry *entry)
{
spin_lock(&test_lock);
list_add_tail_rcu(&entry->link, &test_head);
spin_unlock(&test_lock);
}
（2）写者更新一个节点。
int test_update_node(int key, int new_val)
{
test_entry *entry, *new_entry;
int ret;
ret = -ENOENT;
spin_lock(&test_lock);
list_for_each_entry(entry, &test_head, link) {
if (entry->key == key) {
new_entry = kmalloc(sizeof(test_entry), GFP_ATOMIC);
if (new_entry == NULL) {
ret = -ENOMEM;
break;
}
*new_entry = *entry;
new_entry->val = new_val;
list_replace_rcu(&entry->list, &new_entry->list);
call_rcu(&entry->rcu, test_free_node);
ret = 0;
break;
}
}
spin_unlock(&test_lock);
return ret;
}
static void test_free_node(struct rcu_head *head)
{
test_entry *entry = container_of(head, test_entry, rcu);
kfree(entry);
}
（3）写者删除一个节点。
int test_del_node(int key)
{
test_entry *entry;
int found = 0;
spin_lock(&test_lock);
list_for_each_entry(entry, &test_head, link) {
if (entry->key == key) {
list_del_rcu(&entry->link);
call_rcu(&entry->rcu, test_free_node);
found = 1;
break;
}
}
spin_unlock(&test_lock);
return found;
}
static void test_free_node(struct rcu_head *head)
{
test_entry *entry = container_of(head, test_entry, rcu);
kfree(entry);
}