当前位置：网站首页>ucore lab7 同步互斥实验报告

ucore lab7 同步互斥实验报告

2022-07-06 09:24:00 【芜湖韩金轮】

实验目的

理解操作系统的同步互斥的设计实现；
理解底层支撑技术：禁用中断、定时器、等待队列；
在ucore中理解信号量（semaphore）机制的具体实现；
理解管程机制，在ucore内核中增加基于管程（monitor）的条件变量（condition variable）的支持；
了解经典进程同步问题，并能使用同步机制解决进程同步问题。

实验内容

实验六完成了用户进程的调度框架和具体的调度算法，可调度运行多个进程。如果多个进程需要协同操作或访问共享资源，则存在如何同步和有序竞争的问题。本次实验，主要是熟悉ucore的进程同步机制—信号量（semaphore）机制，以及基于信号量的哲学家就餐问题解决方案。然后掌握管程的概念和原理，并参考信号量机制，实现基于管程的条件变量机制和基于条件变量来解决哲学家就餐问题。

在本次实验中，在kern/sync/check_sync.c中提供了一个基于信号量的哲学家就餐问题解法。同时还需完成练习，即实现基于管程（主要是灵活运用条件变量和互斥信号量）的哲学家就餐问题解法。哲学家就餐问题描述如下：有五个哲学家，他们的生活方式是交替地进行思考和进餐。哲学家们公用一张圆桌，周围放有五把椅子，每人坐一把。在圆桌上有五个碗和五根筷子，当一个哲学家思考时，他不与其他人交谈，饥饿时便试图取用其左、右最靠近他的筷子，但他可能一根都拿不到。只有在他拿到两根筷子时，方能进餐，进餐完后，放下筷子又继续思考。

练习

练习0：填写已有实验

本实验依赖实验1/2/3/4/5/6。请把你做的实验1/2/3/4/5/6的代码填入本实验中代码中有“LAB1”/“LAB2”/“LAB3”/“LAB4”/“LAB5”/“LAB6”的注释相应部分。并确保编译通过。注意：为了能够正确执行lab7的测试应用程序，可能需对已完成的实验1/2/3/4/5/6的代码进行进一步改进。

使用meld软件将lab6和lab7文件进行比较后，发现下列文件需要复制：

proc.c
default_pmm.c
pmm.c
swap_fifo.c
vmm.c
trap.c
sched.c

其他文件不需要修改直接使用即可

练习1: 理解内核级信号量的实现和基于内核级信号量的哲学家就餐问题（不需要编码）

完成练习0后，建议大家比较一下（可用meld等文件diff比较软件）个人完成的lab6和练习0完成后的刚修改的lab7之间的区别，分析了解lab7采用信号量的执行过程。执行make grade，大部分测试用例应该通过。

请在实验报告中给出内核级信号量的设计描述，并说明其大致执行流程。

请在实验报告中给出给用户态进程/线程提供信号量机制的设计方案，并比较说明给内核级提供信号量机制的异同。

基于内核级信号量的哲学家就餐问题

哲学家就餐问题在实验内容中以给出，下面分析一下此问题的代码，主要在check_sync.c实现

该问题的主体函数如下，内容就是每个哲学家拿起叉子，进食，然后放下叉子，无限循环下去：

int state_sema[N]; /* 记录每个人状态的数组 */
/* 信号量是一个特殊的整型变量 */
semaphore_t mutex; /* 临界区互斥 */
semaphore_t s[N]; /* 每个哲学家一个信号量 */

struct proc_struct *philosopher_proc_sema[N];

int philosopher_using_semaphore(void * arg) /* i：哲学家号码，从0到N-1 */
{
    
    int i, iter=0;
    i=(int)arg;
    cprintf("I am No.%d philosopher_sema\n",i);
    while(iter++<TIMES)// 模拟多次实验
    {
     /* 无限循环 */
        cprintf("Iter %d, No.%d philosopher_sema is thinking\n",iter,i); /* 哲学家正在思考 */
        do_sleep(SLEEP_TIME);
        phi_take_forks_sema(i);
        /* 需要两只叉子，或者阻塞 */
        cprintf("Iter %d, No.%d philosopher_sema is eating\n",iter,i); /* 进餐 */
        do_sleep(SLEEP_TIME);
        phi_put_forks_sema(i);
        /* 把两把叉子同时放回桌子 */
    }
    cprintf("No.%d philosopher_sema quit\n",i);
    return 0;
}

注意：拿起 / 放下叉子时，由于需要修改当前哲学家的状态，同时该状态是全局共享变量，所以需要获取锁来防止条件竞争。

可以看到两个重要的函数phi_take_forks_sema和phi_put_forks_sema

void phi_take_forks_sema(int i) /* i：哲学家号码从0到N-1 */
{
    
        down(&mutex); /* 进入临界区 */
        state_sema[i]=HUNGRY; /* 记录下哲学家i饥饿的事实 */
        phi_test_sema(i); /* 试图得到两只叉子 */
        up(&mutex); /* 离开临界区 */
        down(&s[i]); /* 如果得不到叉子就阻塞 */
}
void phi_put_forks_sema(int i) /* i：哲学家号码从0到N-1 */
{
    
        down(&mutex); /* 进入临界区 */
        state_sema[i]=THINKING; /* 哲学家进餐结束 */
        phi_test_sema(LEFT); /* 看一下左邻居现在是否能进餐 */
        phi_test_sema(RIGHT); /* 看一下右邻居现在是否能进餐 */
        up(&mutex); /* 离开临界区 */
}

将叉子放回桌上时，如果当前哲学家左右两边的两位哲学家处于饥饿状态，即准备进餐但没有刀叉时，如果条件符合，则唤醒这两位哲学家并让其继续进餐。

phi_test_sema函数用于设置哲学家的进食状态。如果当前哲学家满足进食条件，则更新哲学家状态，执行哲学家锁所对应的V操作，以唤醒等待叉子的哲学家所对应的线程。

void phi_test_sema(i) /* i：哲学家号码从0到N-1 */
{
    
    if(state_sema[i]==HUNGRY&&state_sema[LEFT]!=EATING
            &&state_sema[RIGHT]!=EATING)
    {
    
        state_sema[i]=EATING;
        up(&s[i]);
    }
}

在试图获得筷子的时候，函数的传入参数为i，即为哲学家编号，此时，他自己为HUNGRY，而且试图检查旁边两位是否都在吃。如果都不在吃，那么可以获得EATING的状态。

内核级信号量的设计描述及大致执行流程

信号量的核心用以下的简短代码可以大致表示：

struct semaphore {
    
     int count;
     queueType queue;
 };
 void semWait(semaphore s)
 {
    
     s.count--;
     if (s.count < 0) {
    
     /* place this process in s.queue */;
     /* block this process */;
 }
 }
 void semSignal(semaphore s)
 {
    
     s.count++;
     if (s.count<= 0) {
    
     /* remove a process P from s.queue */;
     /* place process P on ready list */;
}
}

当多个 (>1)进程可以进行互斥或同步合作时，一个进程会由于无法满足信号量设置的某条件而在某一位置停止,直到它接收到一个特定的信号(表明条件满足了)。为了发信号，需要使用一个称作信号量的特殊变量。为通过信号量s传送信号,信号量的V操作采用进程可执行原语semSignal(s)；为通过信号量s接收信号，信号量的P操作采用进程可执行原语semWait(s)；如果相应的信号仍然没有发送，则进程被阻塞或睡眠，直到发送完为止。

下面分析一下如何实现内核信号量：

信号量的结构体定义如下：

typedef struct {
    
    int value;                    //信号量的当前值
    wait_queue_t wait_queue;     //信号量对应的等待队列
} semaphore_t;

等待队列wait_queue定义在另一个结构体中：

// 用于等待队列，存放了当前等待的线程的PCB和唤醒原因和等待队列和用于还原结构体的等待队列标志
typedef  struct {
    
    struct proc_struct *proc;     //等待进程的指针
    uint32_t wakeup_flags;        //进程被放入等待队列的原因标记
    wait_queue_t *wait_queue;     //指向此wait结构所属于的wait_queue
    list_entry_t wait_link;       //用来组织wait_queue中wait节点的连接
} wait_t;

其中信号量结构体中value的含义如下：

value>0，表示共享资源的空闲数
vlaue<0，表示该信号量的等待队列里的进程数
value=0，表示等待队列为空

信号量初始化。
将value设置为输入的值，将wq初始化为空链表。

void sem_init(semaphore_t *sem, int value) 
{
    
    //将value设为特定值
    sem->value = value;
    //将等待队列初始化
    wait_queue_init(&(sem->wait_queue));
}

在phi_take_forks_sema和phi_put_forks_sema中调用了两个核心函数：up和down，通过这两个函数，可以进入或离开临界区。

_down函数会递减当前信号量的value值。如果value在递减前为0，则将其加入至等待队列wait_queue中，并使当前线程立即放弃CPU资源，调度至其他线程。注意其中的原子操作。该函数的源码如下：

//表示请求一个该信号量对应的资源
static __noinline uint32_t __down(semaphore_t *sem, uint32_t wait_state) {
    
    bool intr_flag;
    local_intr_save(intr_flag);
    //查询整型变量来了解是否存在多余的可分配的资源，
    //如果有多余可分配资源，取出资源（整型变量减1），之后当前进程便可以正常进行；
    if (sem->value > 0) {
    
        sem->value --;
        local_intr_restore(intr_flag);
        return 0;
    }
    wait_t __wait, *wait = &__wait;
    wait_current_set(&(sem->wait_queue), wait, wait_state);
    local_intr_restore(intr_flag);
//如果没有可用的资源，整型变量不是正数，当前进程的资源需求得不到满足，
//因此将其状态改为SLEEPING态，然后将其挂到对应信号量的等待队列中，
    wait_t __wait, *wait = &__wait;
    wait_current_set(&(sem->wait_queue), wait, wait_state);
    local_intr_restore(intr_flag);
   //调用schedule函数来让出CPU
    schedule();
    local_intr_save(intr_flag);
    //在资源得到满足，重新被唤醒之后，将自身从等待队列上删除掉；
    wait_current_del(&(sem->wait_queue), wait);
    local_intr_restore(intr_flag);

    if (wait->wakeup_flags != wait_state) {
    
        return wait->wakeup_flags;
    }
    return 0;
}

_up函数实现的功能稍微简单一点：如果没有等待线程则value++，否则唤醒第一条等待线程。

//表示释放一个该信号量对应的资源，
static __noinline void __up(semaphore_t *sem, uint32_t wait_state) 
{
    
    bool intr_flag;
    local_intr_save(intr_flag);
    {
    
        wait_t *wait;
        查询等待队列是否为空，如果是空的话，给整型变量加1；
        if ((wait = wait_queue_first(&(sem->wait_queue))) == NULL) {
    
            sem->value ++;
        }
        //如果等待队列非空，取出其中的一个进程唤醒；
        else 
        {
    
            assert(wait->proc->wait_state == wait_state);
            wakeup_wait(&(sem->wait_queue), wait, wait_state, 1);
        }
    }
    local_intr_restore(intr_flag);
}

回答问题

请在实验报告中给出给用户态进程/线程提供信号量机制的设计方案，并比较说明给内核级提供信号量机制的异同。

用户态的进程/线程的信号量的数据结构与内核态相同。用户态进程/线程的信号量的相关操作通过系统调用来完成。每当用户进程调用信号量相关函数时，都会进入系统调用，由内核进行处理，之后再返回到用户态继续执行。相比于为内核提供的信号量机制，用户态进程/线程由于要执行中断操作等特权指令，需要通过系统调用进入内核态使用内核信号量机制。

因此想要在用户态完成信号量机制设计，其实只需要在完成内核态信号量机制设计的基础上，增添一些系统调用，包括：

申请创建一个信号量的系统调用
对某一信号量进行P操作
对某一信号量进行V操作
将指定信号量释放

相同点：提供信号量机制的代码实现逻辑是相同的

不同点：

内核态的信号量机制可以直接调用内核的服务，而用户态的则需要通过内核提供的接口来访问内核态服务，这其中涉及到了用户态转内核态的相关机制。
内核态的信号量存储于内核栈中；但用户态的信号量存储于用户栈中。

练习2: 完成内核级条件变量和基于内核级条件变量的哲学家就餐问题（需要编码）

首先掌握管程机制，然后基于信号量实现完成条件变量实现，然后用管程机制实现哲学家就餐问题的解决方案（基于条件变量）。

执行：make grade 。如果所显示的应用程序检测都输出ok，则基本正确。如果只是某程序过不去，比如matrix.c，则可执行

make run-matrix

命令来单独调试它。大致执行结果可看附录。

请在实验报告中给出内核级条件变量的设计描述，并说明其大致执行流程。

请在实验报告中给出给用户态进程/线程提供条件变量机制的设计方案，并比较说明给内核级提供条件变量机制的异同。

请在实验报告中回答：能否不用基于信号量机制来完成条件变量？如果不能，请给出理由，如果能，请给出设计说明和具体实现。

管程

查看实验指导书，可得知一个管程定义了一个数据结构和能为并发进程所执行（在该数据结构上）的一组操作，这组操作能同步进程和改变管程中的数据

管程由四部分组成：

管程内部的共享变量；
管程内部的条件变量；
管程内部并发执行的进程；
对局部于管程内部的共享数据设置初始值的语句。

管程相当于一个隔离区，它把共享变量和对它进行操作的若干个过程围了起来，所有进程要访问临界资源时，都必须经过管程才能进入，而管程每次只允许一个进程进入管程,从而需要确保进程之间互斥。

但在管程中仅仅有互斥操作是不够用的。进程可能需要等待某个条件C为真才能继续执行。

所谓条件变量，即将等待队列和睡眠条件包装在一起，就形成了一种新的同步机制，称为条件变量。一个条件变量CV可理解为一个进程的等待队列，队列中的进程正等待某个条件C变为真。每个条件变量关联着一个断言 “断言” PC。当一个进程等待一个条件变量，该进程不算作占用了该管程，因而其它进程可以进入该管程执行，改变管程的状态，通知条件变量CV其关联的断言Pc在当前状态下为真。

因而条件变量两种操作如下：

wait_cv：被一个进程调用,以等待断言Pc被满足后该进程可恢复执行. 进程挂在该条件变量上等待时,不被认为是占用了管程。如果条件不能满足，就需要等待。
signal_cv：被一个进程调用,以指出断言Pc现在为真,从而可以唤醒等待断言Pc被满足的进程继续执行。如果条件可以满足，那么可以运行。

管程数据结构被定义在monitor.h中：

typedef struct monitor{
    
    semaphore_t mutex; // 二值信号量 用来互斥访问管程
    semaphore_t next; // 用于条件同步 用于发出signal操作的进程等条件为真之前进入睡眠
    int next_count; // 记录睡在 signal 操作的进程数
    condvar_t *cv; // 条件变量
} monitor_t;

条件变量cv的数据结构被定义在monitor.h中：

typedef struct condvar{
    
    semaphore_t sem; // 用于条件同步 用于发出wait操作的进程等待条件为真之前进入睡眠
    int count; // 记录睡在 wait 操作的进程数(等待条件变量成真)
    monitor_t * owner; // 所属管程
} condvar_t;

条件变量机制的实现主要是cond_signal, cond_wait两个函数中，分别表示提醒等待在这个条件变量上的进程恢复执行，以及等待在这个条件变量上，直到有其他进行将其唤醒为止

cond_signal: 将指定条件变量上等待队列中的一个线程进行唤醒，并且将控制权转交给这个进程。该操作实现了对共享变量访问的互斥性，代码如下：

void 
cond_signal (condvar_t *cvp) {
    
   //LAB7 EXERCISE1: YOUR CODE
   cprintf("cond_signal begin: cvp %x, cvp->count %d, cvp->owner->next_count %d\n", cvp, cvp->count, cvp->owner->next_count);  
//判断当前的条件变量的等待队列上是否有正在等待的进程，如果没有则不需要进行任何操作；
//如果有正在等待的进程
     if(cvp->count>0)
     {
    
        //将其中的一个唤醒
        up(&(cvp->sem));
        //所属管程的next计数加1，表示当前进程会被等待者堵塞
        cvp->owner->next_count ++;
        //阻塞，等待条件同步
        down(&(cvp->owner->next));
        //当前进程被唤醒，恢复next上的等待进程计数
        cvp->owner->next_count --;
      }
   cprintf("cond_signal end: cvp %x, cvp->count %d, cvp->owner->next_count %d\n", cvp, cvp->count, cvp->owner->next_count);
}

总的来说：执行cond_signal函数时，首先当前进程测试cv.count，如果不大于0，则表示当前没有执行cond_wait而进入休眠状态的进程，函数直接返回。如果cv.count大于0，这表示当前有执行cond_wait而进入休眠状态的进程，因此需要唤醒在cv.sem上等待出入休眠状态的进程。由于只允许一个进程在管程中执行，所以一旦当前进程唤醒了其他进程，自身就要进入休眠状态，即增加monitor.next_count，让当前进程在信号量monitor.next上进入休眠状态，直到被唤醒时，减少monitor.next_count。

cond_wait：该函数的功能为将当前进程等待在指定信号量上，其操作过程为将等待队列的计数加1，然后释放管程的锁或者唤醒一个next上的进程来释放锁，然后把自己等在条件变量的等待队列上，直到有signal信号将其唤醒，正常退出函数；

void
cond_wait (condvar_t *cvp) {
    
    //LAB7 EXERCISE1: YOUR CODE
    cprintf("cond_wait begin: cvp %x, cvp->count %d, cvp->owner->next_count %d\n", cvp, cvp->count, cvp->owner->next_count);
   cvp->count ++; // 修改等待在条件变量的等待队列上的进程计数
  //当管程的 next_count 大于0，说明有进程睡在了 signal 操作上我们将其唤醒
   if (cvp->owner->next_count > 0)
   {
     // 释放锁
    up(&cvp->owner->next);
   }
   else  //当前没有进程睡在 signal操作数 只需要释放互斥体
   {
    
    up(&cvp->owner->mutex);
   }
   //将自身阻塞，等待条件变量的条件为真，被唤醒后将条件不成立而睡眠的进程计数减1
   down(&cvp->sem); // 将自己等待在条件变量上 
   cvp->count --; // 被唤醒，修正等待队列上的进程计数
    cprintf("cond_wait end: cvp %x, cvp->count %d, cvp->owner->next_count %d\n", cvp, cvp->count, cvp->owner->next_count);
}

因此，若某个进程执行了cond_wait函数，表明该进程需因要的某个条件变量不满足需求而进入休眠状态。等待这个条件变量的休眠进程数cv.count增加。此时如果monitor.next_count大于0，表示至少有1个进程执行cond_signal函数后进入休眠状态，等待monitor.next信号量时，则需要唤醒等待该条件量的另一个进程，然后该进程在cv.sem上休眠。当进程A被唤醒时，减少cv.count，表示等待此条件变量的睡眠进程个数减少了一个，可继续执行。如果monitor.next_count小于等于0，表明目前没有进程执行cond_signal函数进入休眠，则需要唤醒的是由于互斥条件限制而无法进入管程的进程，即唤醒在monitor.mutex上休眠的进程。然后当前进程在cv.sem上休眠，直到被唤醒时，减少cv.count，表示等待此条件的睡眠进程个数减少，可继续执行。

基于条件变量和管程的哲学家就餐问题的实现

实现哲学家问题主要在check_sync.c中

这里创建了5个线程表示5个哲学家，哲学家尝试4次思考->拿叉子->吃饭->放下叉子。

具体分析实现:主要是phi_take_forks_condvar函数以及phi_put_forks_condvar函数

phi_take_forks_condvar函数表示指定的哲学家尝试获得自己所需要进餐的两把叉子，如果不能获得则阻塞，具体实现流程为：

给管程上锁，将哲学家的状态修改为HUNGER；
判断相邻的哲学家是否正在进餐；
如果能够进餐，将自己的状态修改成EATING，然后释放锁，离开管程即可；
如果不能进餐，等待在自己对应的条件变量上，等待相邻的哲学家释放资源的时候将自己唤醒；

void phi_take_forks_condvar(int i) {
    
    //通过P操作进入临界区
     down(&(mtp->mutex));  
      //记录下哲学家i是否饥饿，即处于等待状态拿叉子
      state_condvar[i]=HUNGRY; 
      phi_test_condvar(i);
      while (state_condvar[i] != EATING) 
      {
    
          cprintf("phi_take_forks_condvar: %d didn't get fork and will wait\n",i);
          cond_wait(&mtp->cv[i]);//如果得不到叉子就睡眠
      }
      //如果存在睡眠的进程则那么将之唤醒
      if(mtp->next_count>0)
         up(&(mtp->next));
      else
         up(&(mtp->mutex));
}

phi_put_forks_condvar函数表示释放当前哲学家占用的叉子，并且唤醒相邻的因为得不到资源而进入等待的哲学家：

首先获取管程的锁，将自己的状态修改成THINKING；
检查相邻的哲学家是否在自己释放了叉子的占用之后满足了进餐的条件，如果满足，将其从等待中唤醒
释放锁，离开管程

void phi_put_forks_condvar(int i) {
    
   	 //通过P操作进入临界区
     down(&(mtp->mutex));
    //记录进餐结束的状态
      state_condvar[i]=THINKING;
    //看一下左边哲学家现在是否能进餐
      phi_test_condvar(LEFT);
    //看一下右边哲学家现在是否能进餐
      phi_test_condvar(RIGHT);
      //如果有哲学家睡眠就予以唤醒
     if(mtp->next_count>0)
        up(&(mtp->next));
     else
        up(&(mtp->mutex));
}

由于每个哲学家只可能占有所有需要的资源或者完全不占用资源，因此不会出现部分占有资源的现象，从而避免了死锁的产生；最终必定所有哲学将都能成功就餐

运行make qemu查看运行结果：
在这里插入图片描述
运行make grade 查看成绩：

扩展练习 Challenge ：在ucore中实现简化的死锁和重入探测机制

在ucore下实现一种探测机制，能够在多进程/线程运行同步互斥问题时，动态判断当前系统是否出现了死锁产生的必要条件，是否产生了多个进程进入临界区的情况。如果发现，让系统进入monitor状态，打印出你的探测信息。

扩展练习 Challenge ：参考Linux的RCU机制，在ucore中实现简化的RCU机制

在ucore 下实现下Linux的RCU同步互斥机制。可阅读相关Linux内核书籍或查询网上资料，可了解RCU的设计实现细节，然后简化实现在ucore中。要求有实验报告说明你的设计思路，并提供测试用例。下面是一些参考资料：

RCU 的全称是（Read-Copy-Update），意在读写-复制-更新，在 Linux 提供的所有内核互斥的设施当中属于一种免锁机制。在之前讨论过的读写自旋锁（rwlock）、顺序锁（seqlock）一样，RCU 的适用模型也是读写共存的系统。

读写自旋锁：读者和写者互斥，读者和读者共存，写者和写者互斥。（偏向读者）
顺序锁：写者和写者互斥，写者直接打断读者（偏向写者）

RCU 与他们不同，它的读取和写入操作无需考虑两者之间的互斥问题。

之前的锁分析中，可以知道，加锁、解锁都涉及内存操作，同时伴有内存屏障引入，这些都会导致锁操作的系统开销变大，在此基础之上，内核在 Kernel 的 2.5 版本引入了 RCU 的免锁互斥访问机制。因此RCU实际上是一种改进的 rwlock，读者几乎没有什么同步开销，它不需要锁，不使用原子指令。不需要锁也使得使用更容易，因为死锁问题就不需要考虑了。写者的同步开销比较大，它需要延迟数据结构的释放，复制被修改的数据结构，它也必须使用某种锁机制同步并行的其它写者的修改操作。读者必须提供一个信号给写者以便写者能够确定数据可以被安全地释放或修改的时机。有一个专门的垃圾收集器来探测读者的信号，一旦所有的读者都已经发送信号告知它们都不在使用被RCU保护的数据结构，垃圾收集器就调用回调函数完成最后的数据释放或修改操作。

总的来说，RCU的行为方式：

随时可以拿到读锁，即对临界区的读操作随时都可以得到满足
某一时刻只能有一个人拿到写锁，多个写锁需要互斥，写的动作包括拷贝–修改–宽限窗口到期后删除原值
临界区的原始值为m1,如会有人拿到写锁修改了临界区为m2,则在写锁修改临界区之后拿到的读锁获取的临界区的值为m2,之前获取的为m1,这通过原子操作保证

对比发现RCU读操作随时都会得到满足，但写锁之后的写操作所耗费的系统资源就相对比较多了，并且只有在宽限期之后删除原资源。

针对对象：

RCU 保护的对象是指针。这一点尤其重要.因为指针赋值是一条单指令.也就是说是一个原子操作.因它更改指针指向没必要考虑它的同步.只需要考虑cache的影响。

内核中所有关于 RCU 的操作都应该使用内核提供的 RCU 的 APIs 函数完成，这些 APIs 主要集中在指针和链表的操作。

实现原理：

在RCU的实现过程中，我们主要解决以下问题：

在读取过程中，另外一个线程删除了一个节点。删除线程可以把这个节点从链表中移除，但它不能直接销毁这个节点，必须等到所有的读取线程读取完成以后，才进行销毁操作。RCU中把这个过程称为宽限期（Grace period）。
在读取过程中，另外一个线程插入了一个新节点，而读线程读到了这个节点，那么需要保证读到的这个节点是完整的。这里涉及到了发布-订阅机制（Publish-Subscribe Mechanism）。
保证读取链表的完整性。新增或者删除一个节点，不至于导致遍历一个链表从中间断开。但是RCU并不保证一定能读到新增的节点或者不读到要被删除的节点。

1. 宽限期

void foo_read(void)
{
    
    rcu_read_lock();
    foo *fp = gbl_foo;
    if ( fp != NULL )
        dosomething(fp->a,fp->b,fp->c);
    rcu_read_unlock();
}
 
void foo_update( foo* new_fp )
{
    
	spin_lock(&foo_mutex);
	foo *old_fp = gbl_foo;
	gbl_foo = new_fp;
	spin_unlock(&foo_mutex);
	synchronize_rcu();
	kfee(old_fp);
}

其中 foo_ read 中增加了 rcu_read_lock 和 rcu_read_unlock，这两个函数用来标记一个RCU读过程的开始和结束。其实作用就是帮助检测宽限期是否结束。foo_update 增加了一个函数 synchronize_rcu()，调用该函数意味着一个宽限期的开始，而直到宽限期结束，该函数才会返回。我们再对比着图看一看，线程1和2，在 synchronize_rcu 之前可能得到了旧的 gbl_foo，也就是 foo_update 中的 old_fp，如果不等它们运行结束，就调用 kfee(old_fp)，极有可能造成系统崩溃。而3,4,6在synchronize_rcu 之后运行，此时它们已经不可能得到 old_fp，此次的kfee将不对它们产生影响。

2. 订阅——发布机制

#define rcu_assign_pointer(p, v) 
         __rcu_assign_pointer((p), (v), __rcu)
 
#define __rcu_assign_pointer(p, v, space) 
         do {
     
                 smp_wmb(); 
                 (p) = (typeof(*v) __force space *)(v); 
         } while (0)

我们可以看到它的实现只是在赋值之前加了优化屏障 smp_wmb来确保代码的执行顺序。另外就是宏中用到的__rcu，只是作为编译过程的检测条件来使用的。

3、数据读取的完整性
为了解决读取完整性的问题，我们调用一个发布语义的原生接口，rcu_assign_ pointer() ：

#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));
        }
    else {
    			//关闭编译器在赋值时的非顺序编译优化，保证赋值时已经初始化了。
        	smp_store_release(&p, RCU_INITIALIZER((typeof(p))_r_a_p__v));
    		}
}