Linux rcu机制
RCU是一种同步机制,在Linux内核2.5开发阶段被引入,主要针对读多写少的场景进行优化。
RCU 的基本思想是将更新操作拆分为“移除”和“回收”两个阶段。他允许更新者立即执行移除阶段,并将回收阶段推迟到所有在移除阶段活跃的读取者完成操作之后。任何在移除阶段之后开始的读取者都将无法获取已移除数据项的引用,因此不会受到回收阶段的干扰。RCU 更新流程大致如下:
- 移除指向数据结构的指针,使后续读取者无法获取其引用。
- 等待所有先前存在的读取者完成其 RCU 读端临界区。
- 此时,已不存在任何持有该数据结构引用的读取者,因此现在可以安全地回收该结构。
发布订阅机制
RCU的一个关键特性是能够安全地扫描数据,即使这些数据正在被并发修改。为实现并发插入的这一能力,RCU采用了可被视为发布-订阅机制的方案。
1 p->a = 1;2 p->b = 2;3 p->c = 3;4 gp = p;
没有任何机制强制编译器和CPU按顺序执行四个赋值语句。如果对 gp 的赋值发生在 p 字段初始化之前,那么并发读取者可能会看到未初始化的值。此时需要内存屏障来维持执行顺序,但内存屏障的使用难度是出了名的高。因此我们将其封装成具有发布语义的原始操作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的赋值。
核心API
rcu_read_lock()和rcu_read_unlock()
rcu_read_lock:标记进入一个RCU读临界区, 在读取侧临界区内访问的任何受RCU保护的数据结构,都能保证在临界区的整个持续时间内不会被回收。他的核心逻辑就是调用preempt_disable()关闭内核抢占。
static inline void rcu_read_lock(void)
{__rcu_read_lock(); // 关闭内核抢占......
}
rcu_read_unlock:标记退出一个RCU读临界区。
static inline void rcu_read_unlock(void)
{__rcu_read_unlock(); // 开启内核抢占......
}
synchronize_rcu()
阻塞当前执行流,直到所有CPU上所有已存在的RCU读端临界区都执行完成(这段等待时间被称为"宽限期");需要注意的是,它不一定会等待后续出现的RCU读端临界区完成。还有一个函数叫做 call_rcu(),他会在宽限期结束后,使用指定参数调用回调函数而不会阻塞当前执行流。
static inline void synchronize_rcu(void)
{synchronize_sched();
}void synchronize_sched(void)
{// 检查是否在RCU读临界区内被调用,防止死锁RCU_LOCKDEP_WARN(lock_is_held(&rcu_bh_lock_map) ||lock_is_held(&rcu_lock_map) ||lock_is_held(&rcu_sched_lock_map),"Illegal synchronize_sched() in RCU-sched read-side critical section");// 对于单CPU环境,因为禁止抢占,如果允许到这里宽限期已经结束,直接返回if (rcu_blocking_is_gp())return;// 根据是否启用加速模式走不同的路径等待宽限期结束if (rcu_gp_is_expedited())synchronize_sched_expedited(); // 加速模式:更快但开销更大elsewait_rcu_gp(call_rcu_sched); // 普通模式
}
rcu_assign_pointer()
更新者使用此宏为受RCU保护的指针赋予新值,以便安全的将数值变更从更新者传递给读取者。
static inline void list_add_rcu(struct list_head *new, struct list_head *head)
{__list_add_rcu(new, head, head->next);
}static inline void __list_add_rcu(struct list_head *new,struct list_head *prev, struct list_head *next)
{if (!__list_add_valid(new, prev, next))return;new->next = next;new->prev = prev;// 此时读侧通过prev->next能够到达new节点,所以需要发布rcu_assign_pointer(list_next_rcu(prev), new); // 等价于prev->next = new,确保其他cpu在读取prev->next时看到的一定是new的值next->prev = new; //读侧不依赖prev指针,所以不需要发布
}#define rcu_assign_pointer(p, v)
({uintptr_t _r_a_p__v = (uintptr_t)(v);if (__builtin_constant_p(v) && (_r_a_p__v) == (uintptr_t)NULL)WRITE_ONCE((p), (typeof(p))(_r_a_p__v)); // 若 v 在编译期是常量且为 NULL,直接用 WRITE_ONCE 写入elsesmp_store_release(&p, RCU_INITIALIZER((typeof(p))_r_a_p__v)); // 确保其他cpu在读取p时一定是看到p = _r_a_p__v的写入_r_a_p__v;
})
rcu_dereference()
读取者使用此宏来获取受RCU保护的指针。若要从RCU保护的结构中获取多个字段,使用局部变量是更可取的,重复调用rcu_dereference()会导致关键段内发生更新时无法保证返回相同指针。需要注意的是,rcu_dereference()返回的值仅在包含它的 RCU 读端临界区内有效,例如下面用法是非法的
rcu_read_lock();
p = rcu_dereference(head.next);
rcu_read_unlock();
x = p->address; /* BUG!!! */
rcu_read_lock();
y = p->data; /* BUG!!! */
rcu_read_unlock();
