Userspace RCU原理
##总览
urcu
全称user-space read-copy update即用户态RCU,它提供了与内核RCU相似的功能,使得在多核多线程并发访问共享数据时,reader
线程不用阻塞于writer
线程的操作,从而使reader
线程运行的更快,非常适合于读多写少的场景。
urcu特性
针对不同的应用场景,urcu
提供了以下5种不同的flavors
- urcu
- QSBR(quiescent-state-based RCU)
- Memory-barrier-base RCU
- “Bullet-proof” RCU
- Signal-based RCU
其中qsbr是5种flavor中读性能最好的,它也称为显式静默期声明模式。这种模式下,rcu_read_lock()和rcu_read_unlock()为空操作,对于reader
来说负担为零(这一点是另外4种flavor
不具备的),但这带来的代价是,每个reader
线程必须周期性的调用rcu_quiescent_state()来声明自己进入静默期。不过,需要注意的是,并非每次读操作完成后都需要做此声明,考虑到读操作的性能和应用读写操作次数的不平衡性,通常的做法是每进行一定次数(如1024)的读操作之后声明进入一次静默期。还有一点,每个会进入RCU-read-side critical sections的线程都需要事先通过rcu_register_thread()接口进行注册,退出时调用rcu_unregister_thread()接口进行去注册。
实现
全局计数器
RCU机制是用于多核系统中,保持每个核上的线程所看到的全局数据一致性的一种机制,所以需要一种手段可以判断当writer
线程进行数据更新后,reader
线程看到的数据是否已经最新,为此urcu维护了一个全局的计数器rcu_gp.ctr
,每次writer
进行同步操作(synchronize),都会使计数器加1,表示数据我已经更新了,reader你需要更新
struct rcu_gp rcu_gp;
struct rcu_gp {unsigned long ctr;...
} __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
每个reader
线程也持有一个线程内部的计数器ctx
,如果这个ctx
与rcu_gp.ctr
一致,就表明本reader
线程的数据已经最新(ACTIVE_CURRENT),反之则不是最新(ACTIVE_OLD),
struct rcu_reader {unsigned long ctr;...
};
DECLARE_URCU_TLS(struct rcu_reader, rcu_reader)
读者 注册(register)\去注册(unregister)\上线(online)\下线(offline)
qsbr rcu
的实现中,reader
线程必须进行显式注册, 将自己挂接在全局链表registry
上,通俗地说就是将自己置于全局管理之下,这样当writer在进行同步(synchronize)时,才能知道哪些线程需要同步(只有注册过的线程才需要)。线程的上线状态分为在线(online)和离线(offline),其中处于offline的线程虽然在registry
链表上,但在synchronized时,writer会忽略这些线程。线程注册会默认置于online状态。
void rcu_register_thread(void)
{URCU_TLS(rcu_reader).tid = pthread_self();mutex_lock(&rcu_registry_lock);URCU_TLS(rcu_reader).registered = 1;cds_list_add(&URCU_TLS(rcu_reader).node, ®istry);_rcu_thread_online();
}
线程上线(online)的本质,就是将rcu_gp.ctr
的值存储到本线程的ctr中
static inline void _rcu_thread_online(void)
{_CMM_STORE_SHARED(URCU_TLS(rcu_reader).ctr, CMM_LOAD_SHARED(rcu_gp.ctr));
}
而线程下线(offline),则是将本线程的ctr
清零
static inline void _rcu_thread_offline(void)
{CMM_STORE_SHARED(URCU_TLS(rcu_reader).ctr, 0);wake_up_gp();
}
写者 同步(synchronize)
rcu
机制的一个典型场景: 全局指针gp_ptr
指向内存区域A,writer
在申请了一份新的内存区域B后,使全局指针gp_ptr
指向B。在多核系统中,writer
在更新后并不知道有没有reader
正在区域A的数据,所以它需要阻塞等待所有的reader
线程已经更新(即reader.ctx
等于gp.ctr
),这个操作便是同步(synchronize),其简化版实现代码片段如下
void synchronize_rcu(void)
{CDS_LIST_HEAD(qsreaders);DEFINE_URCU_WAIT_NODE(wait, URCU_WAIT_WAITING);......urcu_wait_add(&gp_waiters, &wait) /* 将writer自身置于wait状态 */......wait_for_readers(®istry, &cur_snap_readers, &qsreaders); /* writer阻塞在这里 */.....
}static void wait_for_readers(struct cds_list_head *input_readers,struct cds_list_head *cur_snap_readers,struct cds_list_head *qsreaders)
{unsigned int wait_loops = 0;struct rcu_reader *index, *tmp;/** Wait for each thread URCU_TLS(rcu_reader).ctr to either* indicate quiescence (offline), or for them to observe the* current rcu_gp.ctr value.*/for (;;) { /* 直到所有reader.ctr已经到最新才跳出循环 */uatomic_set(&rcu_gp.futex, -1);cds_list_for_each_entry(index, input_readers, node) {_CMM_STORE_SHARED(index->waiting, 1);/* 遍历所有输入的reader */cds_list_for_each_entry_safe(index, tmp, input_readers, node) {switch (rcu_reader_state(&index->ctr)) {case RCU_READER_ACTIVE_CURRENT: /* reader.ctr已经最新 */case RCU_READER_INACTIVE: /* reader处于offline状态 */cds_list_move(&index->node, qsreaders); /* 从遍历列表中移除 */break;case RCU_READER_ACTIVE_OLD: /* reader.ctr不是最新 */break;}}if (cds_list_empty(input_readers)) {uatomic_set(&rcu_gp.futex, 0); /* 列表空了,表示所有reader已更新 跳出循环 */break;}}
}
读者 静默(quiescent)
从上面writer
synchronize的过程可知,要使writer
结束阻塞状态,reader
必须将其ctr
更新到最新(除非它处于offline状态),更新到最新是通过reader
调用rcu_quiescent_state()
接口声明静默期完成的.
static inline void _rcu_quiescent_state(void)
{unsigned long gp_ctr;if ((gp_ctr = CMM_LOAD_SHARED(rcu_gp.ctr)) == URCU_TLS(rcu_reader).ctr)return;_rcu_quiescent_state_update_and_wakeup(gp_ctr);
}
static inline void _rcu_quiescent_state_update_and_wakeup(unsigned long gp_ctr)
{_CMM_STORE_SHARED(URCU_TLS(rcu_reader).ctr, gp_ctr); /* 将本线程ctr更新为gp_ctr */wake_up_gp(); /* 唤醒writer */
}
example
这个example节选自urcu官方代码的测试例程test_urcu_qsbr
测试例程根据用户输入创建若干个writer和reader,writer不断申请释放内存资源,并用全局指针test_rcu_pointer
记录资源,reader不断读取test_rcu_pointer
指向资源的值,并且每1024次声明静默期,最后统计reader和writer的次数。
void *thr_writer(void *_count)
{unsigned long long *count = _count;int *new, *old;for (;;) {new = malloc(sizeof(int));assert(new);*new = 8;old = rcu_xchg_pointer(&test_rcu_pointer, new);synchronize_rcu();if (old)*old = 0;free(old);URCU_TLS(nr_writes)++;}printf_verbose("thread_end %s, tid %lu\n","writer", urcu_get_thread_id());*count = URCU_TLS(nr_writes);return ((void*)2);
}
void *thr_reader(void *_count)
{unsigned long long *count = _count;int *local_ptr;rcu_register_thread();rcu_thread_offline();rcu_thread_online();for (;;) {rcu_read_lock();local_ptr = rcu_dereference(test_rcu_pointer);if (local_ptr)assert(*local_ptr == 8);rcu_read_unlock();URCU_TLS(nr_reads)++;/* QS each 1024 reads */if (caa_unlikely((URCU_TLS(nr_reads) & ((1 << 10) - 1)) == 0))rcu_quiescent_state();}rcu_unregister_thread();*count = URCU_TLS(nr_reads);printf_verbose("thread_end %s, tid %lu\n","reader", urcu_get_thread_id());return ((void*)1);
}
perfomance
flavor | total read | totol write |
---|---|---|
urcu | 7826237468 | 91 |
qsbr | 10427746859 | 1746176 |
memory-barrir | 365980233 | 10333212 |
bullet-proof | 568170476 | 5226213 |
signal-based | 9522616041 | 2329 |
call rcu
前面writer
的例子中,当writer
进行数据更新后需要释放旧资源,而这要在synchronize_rcu()
接触阻塞后才能进行(否则reader
还在使用呐),但还有的时候,我们希望提高writer
的效率,‘释放’过程不要阻塞,再reader进行了更新后,再进行资源释放,urcu提供了call_rcu()
接口来完成这一功能
struct rcu_head {struct cds_wfcq_node next;void (*func)(struct rcu_head *head);
};
void call_rcu(struct rcu_head *head,void (*func)(struct rcu_head *head);
一般得,将要延迟释放的数据结构内嵌一个rcu_head
结构,在需要延迟释放时调用
struct global_foo {struct rcu_head rcu_head;......
};
struct global_foo g_foo;
在writer
更新后,需要释放旧的资源时时,调用call_rcu()
,之后当所有reader都更新完成后,设置的回调函数free_func
被自动调用
call_rcu(&g_foo.rcu_head, free_func);
那么urcu是如何实现这个功能的呢?
既然不能阻塞将writer
阻塞在synchronize_rcu()
,那总得有一个线程阻塞在synchronize_rcu()
等待所有reader
更新,于是urcu内部创建一个线程,称为call_rcu_thread
,这个线程专门用于writer
call_rcu()
(这个线程只会在第一次call_rcu()
被创建,之后的call_rcu()
均使用这个线程),以下是call_rcu_thread
创建时的代码片段
/* 第一次call_rcu()会调用到 call_rcu_data_init() */
static void call_rcu_data_init(struct call_rcu_data **crdpp,unsigned long flags,int cpu_affinity)
{struct call_rcu_data *crdp;int ret;crdp = malloc(sizeof(*crdp));if (crdp == NULL)urcu_die(errno);memset(crdp, '\0', sizeof(*crdp));cds_wfcq_init(&crdp->cbs_head, &crdp->cbs_tail);......ret = pthread_create(&crdp->tid, NULL, call_rcu_thread, crdp); /* 创建call_rcu_thread */if (ret)urcu_die(ret);
}static void *call_rcu_thread(void *arg)
{struct call_rcu_data *crdp = (struct call_rcu_data *) arg;rcu_register_thread();URCU_TLS(thread_call_rcu_data) = crdp;for (;;) {......synchronize_rcu(); /* 在这里完成同步 */rhp->func(rhp); /* 执行回调 */......}rcu_unregister_thread();return NULL;
}
Userspace RCU原理相关推荐
- RCU 原理与实现分析
RCU锁谈起来许多人云里雾里,听着也觉得高大上,竟然可以不加锁就实现加锁的效果.此外,网上关于RCU实现原理的完整闭环逻辑分析,比较少,大多停留在阻塞写者,等待读者退出临界区的程度.而对如何实现延迟调 ...
- 一文深入分析|RCU原理
Linux 内核设计了多种锁机制,比如 读写锁.自旋锁 和 信号量 等.为什么要设计这么多锁机制呢?这是因为不同的锁机制适用于不同的场景,比如 读写锁 适用于读多写少的场景:而 信号量 适用于进程长时 ...
- rcu锁原理以及rcu example学习
rcu参考资料: https://airekans.github.io/c/2016/05/10/dive-into-liburcu https://lwn.net/Articles/262464/ ...
- linux kernel rcu 读复制更新 并发控制机制 简介
目录 RCU原理 普通RCU 变种RCU的特点 可睡眠RCU(SRCU) RCU原理 RCU特点:RCU全称Read Copy Update 读复制更新,是一种完全不同于锁的并发控制机制. 主要保护的 ...
- 内核并发控制---RCU (来自网易)
定义在头文件linux/rcupdate.h中; 一.RCU原理 RCU,全名Read-Copy-Update(读-拷贝-更新);对于被RCU所保护的共享资源,读执行单元不需要获得任何锁就可以访问到它 ...
- 深入理解并行编程原理与实践
本文是<Is Parallel Programming Hard, And, If So, What Can You Do About It?>的中文翻译版<深入理解并行编程> ...
- linux rcu stall 分析
linux4.9 aarch32 被rcu 折腾过几次后决心查明rcu原理和发生cpu stall warning怎么分析问题,这篇文章的由来是遇到了一个类似如下的kernel cpu stall ...
- 【高阶】一个用户态的RCU实现liburcu
在无锁编程的世界里,ABA问题是一个没有办法回避的实现问题.我们有Read-Copy Update(RCU)这个法宝,帮助我们方便的实现很多的无锁算法数据结构. 本文会首先简略介绍RCU的基本概念,然 ...
- 深入理解 RCU 实现
http://blog.jobbole.com/106856/ 深入理解RCU实现 --基于内核2.6.21 RCU实现(lvyilong316) RCU(Read-Copy Update),顾名思义 ...
最新文章
- C# 2进制、8进制、10进制、16进制...各种进制间的轻松转换
- StringBuilder、StringBuffer、String区别
- 这就是编程的终极难题? | 每日趣闻
- ios利用block实现回调示例
- 看bilibili的财经博主,分析博主的频道
- Code Quality
- 如何解决90%的报表设计难题?300张报表模板任君挑选
- Linux 命令之 pico -- 文本编辑器
- C语言:运行中获取宏名字的技巧
- bcp 不能调用where 子句_三、p18-28条件查询、分组聚合、排序where/group by/having/order by...
- 2020年中国电力线载波通信行业发展现状及竞争格局分析,国家电网持续推进电网转型升级,配电自动化覆盖率达到90%「图」
- eclipse无法识别ftl文件解决
- ECharts 饼图的合并显示
- tkinter教学(五)tkinter内置变量类别 set,get用法
- 为什么用MongoDB而不用Redis
- 基于TCP的网络实时聊天室(socket通信案例)
- 漫谈唯一设备ID,android开发工程师
- 3d稀疏卷积——spconv源码剖析(三)
- fixed脱离文档流 不占位置
- python 模拟键盘自动打字敲英语文章