一文搞懂 | Linux 同步管理（下）

Love The Way You Lie 2022-09-16 11:15 176阅读 0赞

[上面][Link 1]讲的自旋锁，信号量和互斥锁的实现，都是使用了原子操作指令。由于原子操作会 lock，当线程在多个 CPU 上争抢进入临界区的时候，都会操作那个在多个 CPU 之间共享的数据 lock。CPU 0 操作了 lock，为了数据的一致性，CPU 0 的操作会导致其他 CPU 的 L1 中的 lock 变成 invalid，在随后的来自其他 CPU 对 lock 的访问会导致 L1 cache miss（更准确的说是communication cache miss），必须从下一个 level 的 cache 中获取。

这就会使缓存一致性变得很糟，导致性能下降。所以内核提供一种新的同步方式：RCU（读-复制-更新）。

## RCU 解决了什么 ##

RCU 是读写锁的高性能版本，它的核心理念是读者访问的同时，写者可以更新访问对象的副本，但写者需要等待所有读者完成访问之后，才能删除老对象。读者没有任何同步开销，而写者的同步开销则取决于使用的写者间同步机制。

RCU 适用于需要频繁的读取数据，而相应修改数据并不多的情景，例如在文件系统中，经常需要查找定位目录，而对目录的修改相对来说并不多，这就是 RCU 发挥作用的最佳场景。

## RCU 例子 ##

RCU 常用的接口如下图所示：

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>rcu_read_lock</td> 
   <td>标记读者进入读端临界区</td> 
  </tr> 
  <tr> 
   <td>rcu_read_unlock</td> 
   <td>标记读者退出临界区</td> 
  </tr> 
  <tr> 
   <td>synchronize_rcu</td> 
   <td>同步RCU，即所有的读者已经完成读端临界区，写者才可以继续下一步操作。由于该函数将阻塞写者，只能在进程上下文中使用</td> 
  </tr> 
  <tr> 
   <td>call_rcu</td> 
   <td>把回调函数 func 注册到RCU回调函数链上，然后立即返回</td> 
  </tr> 
  <tr> 
   <td>rcu_assign_pointer</td> 
   <td>用于RCU指针赋值</td> 
  </tr> 
  <tr> 
   <td>rcu_dereference</td> 
   <td>用于RCU指针取值</td> 
  </tr> 
  <tr> 
   <td>list_add_rcu</td> 
   <td>向RCU注册一个链表结构</td> 
  </tr> 
  <tr> 
   <td>list_del_rcu</td> 
   <td>从RCU移除一个链表结构</td> 
  </tr> 
 </tbody> 
</table>

为了更好的理解，在剖析 RCU 之前先看一个例子：

#include <linux/kernel.h>
    #include <linux/module.h>
    #include <linux/init.h>
    #include <linux/slab.h>
    #include <linux/spinlock.h>
    #include <linux/rcupdate.h>
    #include <linux/kthread.h>
    #include <linux/delay.h>
    
    struct foo {
            int a;
            struct rcu_head rcu;
    };
    
    static struct foo *g_ptr;
    
    static int myrcu_reader_thread1(void *data) //读者线程1
    {
            struct foo *p1 = NULL;
    
            while (1) {
                    if(kthread_should_stop())
                            break;
                    msleep(20);
                    rcu_read_lock();
                    mdelay(200);
                    p1 = rcu_dereference(g_ptr);
                    if (p1) 
                            printk("%s: read a=%d\n", __func__, p1->a);
                    rcu_read_unlock();
            }
     
            return 0;
    }
    
    static int myrcu_reader_thread2(void *data) //读者线程2
    {
            struct foo *p2 = NULL;
    
            while (1) {
                    if(kthread_should_stop())
                            break;
                    msleep(30);
                    rcu_read_lock();
                    mdelay(100);
                    p2 = rcu_dereference(g_ptr);
                    if (p2)
                            printk("%s: read a=%d\n", __func__, p2->a);
             
                    rcu_read_unlock();
            }
     
            return 0;
    }
    
    static void myrcu_del(struct rcu_head *rh) //回收处理操作
    {
            struct foo *p = container_of(rh, struct foo, rcu);
            printk("%s: a=%d\n", __func__, p->a);
            kfree(p);
    }
    
    static int myrcu_writer_thread(void *p) //写者线程
    {
            struct foo *old;
            struct foo *new_ptr;
            int value = (unsigned long)p;
    
            while (1) {
                    if(kthread_should_stop())
                            break;
                    msleep(250);
                    new_ptr = kmalloc(sizeof (struct foo), GFP_KERNEL);
                    old = g_ptr;
                    *new_ptr = *old;
                    new_ptr->a = value;
                    rcu_assign_pointer(g_ptr, new_ptr);
                    call_rcu(&old->rcu, myrcu_del);
                    printk("%s: write to new %d\n", __func__, value);
                    value++;
            }
    
            return 0;
    }
    
    static struct task_struct *reader_thread1;
    static struct task_struct *reader_thread2;
    static struct task_struct *writer_thread;
    
    static int __init my_test_init(void)
    {
            int value = 5;
    
            printk("figo: my module init\n");
            g_ptr = kzalloc(sizeof (struct foo), GFP_KERNEL);
    
            reader_thread1 = kthread_run(myrcu_reader_thread1, NULL, "rcu_reader1");
            reader_thread2 = kthread_run(myrcu_reader_thread2, NULL, "rcu_reader2");
            writer_thread = kthread_run(myrcu_writer_thread, (void *)(unsigned long)value, "rcu_writer");
    
            return 0;
    }
    static void __exit my_test_exit(void)
    {
            printk("goodbye\n");
            kthread_stop(reader_thread1);
            kthread_stop(reader_thread2);
            kthread_stop(writer_thread);
            if (g_ptr)
                    kfree(g_ptr);
    }
    MODULE_LICENSE("GPL");
    module_init(my_test_init);
    module_exit(my_test_exit);

执行结果是：

> myrcu\_reader\_thread2: read a=0  
> myrcu\_reader\_thread1: read a=0  
> myrcu\_reader\_thread2: read a=0  
> myrcu\_writer\_thread: write to new 5  
> myrcu\_reader\_thread2: read a=5  
> myrcu\_reader\_thread1: read a=5  
> myrcu\_del: a=0

## RCU 原理 ##

可以用下面一张图来总结，当写线程 myrcu\_writer\_thread 写完后，会更新到另外两个读线程 myrcu\_reader\_thread1 和 myrcu\_reader\_thread2。读线程像是订阅者，一旦写线程对临界区有更新，写线程就像发布者一样通知到订阅者那里，如下图所示。

![53d2990f8fc3485dc37e5de6c9bb3c2a.png][]

写者在拷贝副本修改后进行 update 时，首先把旧的临界资源数据移除（**Removal**）；然后把旧的数据进行回收（**Reclamation**）。结合 API 实现就是，首先使用 **rcu\_assign\_pointer** 来移除旧的指针指向，指向更新后的临界资源；然后使用 **synchronize\_rcu** 或 **call\_rcu** 来启动 Reclaimer，对旧的临界资源进行回收（其中 synchronize\_rcu 表示同步等待回收，call\_rcu 表示异步回收）。

为了确保没有读者正在访问要回收的临界资源，Reclaimer 需要等待所有的读者退出临界区，这个等待的时间叫做宽限期（Grace Period）。

### Grace Period ###

![c7fbca384faccc09bbd0dd98c72dcb64.png][]

中间的黄色部分代表的就是 Grace Period，中文叫做宽限期，从 Removal 到 Reclamation，中间就隔了一个宽限期，只有当宽限期结束后，才会触发回收的工作。宽限期的结束代表着 Reader 都已经退出了临界区，因此回收工作也就是安全的操作了。

宽限期是否结束，与 CPU 的执行状态检测有关，也就是检测静止状态 Quiescent Status。

### Quiescent Status ###

Quiescent Status，用于描述 CPU 的执行状态。当某个 CPU 正在访问 RCU 保护的临界区时，认为是活动的状态，而当它离开了临界区后，则认为它是静止的状态。当所有的 CPU 都至少经历过一次 Quiescent Status 后，宽限期将结束并触发回收工作。

因为 rcu\_read\_lock 和 rcu\_read\_unlock 分别是关闭抢占和打开抢占，如下所示：

static inline void __rcu_read_lock(void)
    {
     preempt_disable();
    }

static inline void __rcu_read_unlock(void)
    {
     preempt_enable();
    }

所以发生抢占，就说明不在 rcu\_read\_lock 和 rcu\_read\_unlock 之间，即已经完成访问或者还未开始访问。

## Linux 同步方式的总结 ##

<table> 
 <thead> 
  <tr> 
   <th>机制</th> 
   <th>等待机制</th> 
   <th>优缺</th> 
   <th>场景</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>原子操作</td> 
   <td>无；ldrex 与 strex 实现内存独占访问</td> 
   <td>性能相当高；场景受限</td> 
   <td>资源计数</td> 
  </tr> 
  <tr> 
   <td>自旋锁</td> 
   <td>忙等待；唯一持有</td> 
   <td>多处理器下性能优异；临界区时间长会浪费</td> 
   <td>中断上下文</td> 
  </tr> 
  <tr> 
   <td>信号量</td> 
   <td>睡眠等待（阻塞）；多数持有</td> 
   <td>相对灵活，适用于复杂情况；耗时长</td> 
   <td>情况复杂且耗时长的情景；比如内核与用户空间的交互</td> 
  </tr> 
  <tr> 
   <td>互斥锁</td> 
   <td>睡眠等待（阻塞）；优先自旋等待；唯一持有</td> 
   <td>较信号量高效，适用于复杂场景；存在若干限制条件</td> 
   <td>满足使用条件下，互斥锁优先于信号量</td> 
  </tr> 
  <tr> 
   <td>RCU</td> 
   <td><br></td> 
   <td>绝大部分为读而只有极少部分为写的情况下，它是非常高效的；但延后释放内存会造成内存开销，写者阻塞比较严重</td> 
   <td>读多写少的情况下，对内存消耗不敏感的情况下，满足 RCU 条件的情况下，优先于读写锁使用；对于动态分配数据结构这类引用计数的机制，也有高性能的表现。</td> 
  </tr> 
 </tbody> 
</table>

![4aac3674bf6e83d89f012a9f21b7dcb6.png][]

5T技术资源大放送！包括但不限于：C/C++，Arm, Linux，Android，人工智能，单片机，树莓派，等等。在上面的【人人都是极客】公众号内回复「peter」，即可免费获取！！

![72f24eadca8e8544929a932ea333b2c8.gif][] **记得点击****分享****、****赞****和****在看****，给我充点儿电吧**

[Link 1]: http://mp.weixin.qq.com/s?__biz=MzIxMjE1MzU4OA==&mid=2648927521&idx=1&sn=7c36d4b87a86a987bc9920fcc2d29396&chksm=8f5da8bab82a21ac26f4ebd4b6c2025f96044c65b4524dc29b76e76e44aefa1c2162ba72dc1d&scene=21#wechat_redirect
[53d2990f8fc3485dc37e5de6c9bb3c2a.png]: /images/20220828/2db494df1eb1487cbc7c5b82270baa9d.png
[c7fbca384faccc09bbd0dd98c72dcb64.png]: /images/20220828/e1b4072dff1d4702bde500bcc3cdeda4.png
[4aac3674bf6e83d89f012a9f21b7dcb6.png]: /images/20220828/288af97d6f7344e2b86cc558e991cb14.png
[72f24eadca8e8544929a932ea333b2c8.gif]: /images/20220828/e84e8102cb994016835fbeb07c3941f4.png