引言

这节课和上一节xv6进程切换是一个完整的的进程切换专题，上一节主要讨论进程切换过程中的细节，而这一节主要讨论进程切换过程中锁的使用，所以本节的两大关键词就是"Coordination"（协调）和 "lost wakeup"

Coordination 就是有关出让CPU，直到等待的事件发生再恢复执行。人们发明了很多不同的 Coordination 的实现方式，但是与许多Unix风格操作系统一样，xv6 使用的是 Sleep&Wakeup 这种方式。而几乎所有的Coordination机制都需要处理 lost wakeup 的问题。

"busy wait" vs "coordination"

busy wait 其实和 lectrue10中讲过的 spin lock 思想是一样的，在一个进程运行期间，可能会遇到一写需要等待的场景：

• 等待从磁盘上读取数据
• 读取 pipe 中的数据，但是 pipe 为空了，需要等待 pipe 中再次来数据
• unix 中经典的 wait 函数（父进程使用 wait 函数来等待子进程 exit）

典型的 busy wait 模型如下：

while(事件未发生) {
     ; // 空语句，表示事件未发生就一直循环等待
}

busy wait 很明显是一种"笨办法"，因为如果等待的事件不能很快发生，那么该进程在等待期间进程还是一直占用着 cpu，直到期待的事件发生，这对于追求高效的计算机来说有些无法忍受，毕竟现代的主流计算机在 1ms 内都可以执行上百万条指令。

与 busy wait 相对的就是 Coordination，即进程发现自己在等待，就让主动出让 cpu，当等待的事件发生时，再恢复执行。Coordination 是一个很大的话题，这一节只讨论主流的一种实现 Coordination 的方式：Sleep&Wakeup

所以这里的关键技术点有三个：

• 怎么发现自己在等待？
• 怎么出让？
• 怎么恢复执行？

会在下面给出示例以及说明。

一个设计良好的 Sleep&Wakeup 示例

以 Robert 教授重写的uartwrite()和 uartintr()函数为例：

如果需要往 console 中写字符，需要调用uartwrite()函数，这个函数会在循环中将 buf 中的字符一个一个的向UART硬件写入，UART硬件一次只能接受一个字符的传输，所以在两个字符之间的等待时间可能会很长，所以这里就采用了 sleep&wakeup 的方式：

• 先将 buf 中的一个字符写入THR 寄存器中，然后将标志位 tx_done 置为 0（初始值为 1），开始循环，检查 tx_done 是否为 1（发送完成），若未发送完成则 sleep，出让 cpu
• THR 寄存器中的数据会由 uart 硬件写入到 console 中，uart 硬件会在完成传输一个字符后，触发中断，从而进入中断处理程序，在中断处理程序中将 tx_done 设置为完成，并且 wakeup 之前uartwrite()中 sleep 的进程
• 接着写下一个字符

以上就是 Sleep&Wakeup 工作的方式，这里需要注意 sleep 和 wakeup 共同的参数：tx_chan，这是一个64bit的值，用来标识这个 sleep 以及 wakeup 是一对，或者说 wakeup 会唤醒具有相同标识的 sleep。

Sleep&Wakeup 原理

sleep 和 wakeup 的原理也很简单，尤其是学了上一节课程之后：sleep 修改进程的状态为 SLEEPING，然后将进程打上 sleep channel 的标签，最后调用 sched 切换到调度器线程，由调度器选择其他合适的进程运行。

// Atomically release lock and sleep on chan.
// Reacquires lock when awakened.
void
sleep(void *chan, struct spinlock *lk)
{
  struct proc *p = myproc();
  
  // Must acquire p->lock in order to
  // change p->state and then call sched.
  // Once we hold p->lock, we can be
  // guaranteed that we won't miss any wakeup
  // (wakeup locks p->lock),
  // so it's okay to release lk.

  acquire(&p->lock);  //DOC: sleeplock1
  release(lk);

  // Go to sleep.
  p->chan = chan;
  p->state = SLEEPING;

  sched();

  // Tidy up.
  p->chan = 0;

  // Reacquire original lock.
  release(&p->lock);
  acquire(lk);
}

wakeup 就更简单了，根据进程的 state 以及 sleep channel 的数值就可以寻找到之前 sleep 的进程，将其状态修改为 RUNABLE，以便调度器随时调用。

// Wake up all processes sleeping on chan.
// Must be called without any p->lock.
void
wakeup(void *chan)
{
  struct proc *p;

  for(p = proc; p < &proc[NPROC]; p++) {
    if(p != myproc()){
      acquire(&p->lock);
      if(p->state == SLEEPING && p->chan == chan) {
        p->state = RUNNABLE;
      }
      release(&p->lock);
    }
  }
}

lost wakeup

注意上一节的 sleep 函数有两个参数，一个是 void *chan，也就是标识 sleep-wakeup 对的 slep channel，另一个参数是一把锁，：struct spinlock *lk。第一个参数好理解，必须要有一个标识来把 sleep 和 wakeup 对应起来，以确保 wakeup 唤醒的是争取的进程；但是第二个参数看上去就有点"丑陋了"，怎么会把锁传入 sleep 函数中呢？

要解释这件事情，最好是使用反证法，假如我们的 sleep 函数的参数中没有这个锁，程序运行是否会出问题？答案是肯定的，而且出的问题就是 lost wakeup。

通俗第解释一下什么是 lost wakeup，比如一个进程在 sleep channel 为 233 的数值上 sleep 了，然后调用函数 wakeup(233)，就可以唤醒这个特定的进程，这是正常的 sleep-wakeup 使用方式，但是如果由于编码的疏漏，造成 wakeup(233)在 sleep(233) 之前运行了，这样 wakeup(233)不会唤醒任何进程，因为对应的进程还没有 sleep，之后再运行 sleep(233) ，但是这一次，不会有 wakeup 来唤醒了，该进程就会一直 sleep，这就是 lost wakeup 的问题

首先看一下原始的、正确的 uartwrite()函数和 uartintr()函数：

接着，我们想象去掉 sleep 函数的中的 lock 参数，这个代码要怎么修改：

• 首先，锁肯定还是需要的，因为 uartwrite()函数和 uartintr()函数都操作了变量 tx_done，而这两个函数是可能被不同的 cpu core 运行的，所以为了保护共享数据的正确性，锁还是需要的
• 其次，需要在uartwrite() 函数中、sleep 函数之前添加解锁语句，因为如果 sleep 函数之前没有解锁的语句的话，一旦运行 uartwrite 函数的进程带着锁 sleep 了，即使之后成功写入数据、发生了中断，也会因为没办法获取锁而卡在uartintr()函数开头，从而无法修改tx_done的值为 1，也无法 wakeup 之前 sleep 的进程，所以还是需要在 uartwrite()函数中添加加解锁语句，修改后的uartwrite()函数代码如下：
  

好了，现在我们为了优雅，或者说为了探究 sleep 函数为什么要传入 lock 参数，把代码修改为了更容易理解的版本，那么现在问题就出来了，为什么会造成 lost wakeup 的问题呢？

比起复杂的语言解释，下面这幅图更加清晰，如果按照图中 1~14 的顺序执行代码，就会发生 lost wakeup 的情况，因为第 1 步 release 解锁后，可能立即发生中断，然后执行uartintr()函数，并运行 wakeup，所以这里 wakeup 就会在第 5 步运行，而 broken_sleep 则在第 11 步运行，wakeup 发生在了 sleep 之前，即 lost wakeup 的情况：

所以这也回答了为什么 sleep 函数第 2 个参数是 lock，因为不传入这个 lock，在 sleep 外加解锁的话，会发生 lost wakeup 的情况。

解决 lost wakeup

明白了 lost wakeup 是如何发生的，结局方案似乎就有一些眉目了，我们需要把 lock 传入 sleep 函数中，想一个办法确保 sleep 发生在 wakeup 之前，现在来仔细分析" sleep 发生在 wakeup 之前 "的含义，我们之所以想让 sleep 发生在 wakeup 之前，是因为可能存在以下运行顺序：

1. sleep 中先 release(tx_lock)
2. uartintr()中 acquire(tx_lock)，然后调用 wakeup 修改 p->state = RUNNABLE
3. sleep 中继续 p->chan = chan; p->state = SLEEPING

由于 p->state = RUNNABLE发生在 p->chan = chan; p->state = SLEEPING 之前，所以就有了 lost wakeup，所以现在的问题就转换成了怎么保证以上三步骤中第 1 步和第 3 步之间的原子性；或者说保证共享数据 p->state的安全性，答案就很明显了，我们还需要一把锁，而且这把锁 xv6 已经实现了，就是每个进程自带的进程锁p->lock，用来保护进程自身的数据安全，加锁方式见下图：

所以总结来看，要想进入 wakeup 函数修改 p->state的状态，需要获取两把锁，一把是 tx_lock，用来进入uartintr()函数、保护 tx_done，一把是 p->lock，用来保护 p->state，这里巧妙地方在于释放 tx_lock之前需要先获取p->lock，颇有一种"交换人质"的感觉（在放你之前先把他抓过来 23333）

lab6 Thread 心得

lab6 的前两个 part 见上一篇博客

Barrier

这个 lab 但是挺有趣的，可以了解到了计算机中同步屏障机制是如何实现的。

简单来说，一段代码被多个线程执行，如何保证多个线程都到了其中某一点之后，才能继续往下执行？或者说如何"拦住"执行的较快的线程，让他们都到达 barrier 之后再继续?

题目的 hint 已经给出了重要的工具：pthread，尤其是以下两个函数：

pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
pthread_cond_broadcast(&cond);     // wake up every thread sleeping on cond

童鞋们，考验英语水平的时候来了！查看 unix 中关于pthread_cond_wait的描述：

This functions atomically release mutex and cause the calling thread to block on the condition variable cond; atomically here means "atomically with respect to access by another thread to the mutex and then the condition variable". That is, if another thread is able to acquire the mutex after the about-to-block thread has released it, then a subsequent call to pthread_cond_signal() or pthread_cond_broadcast() in that thread behaves as if it were issued after the about-to-block thread has blocked.

我先来直译一下：

这个函数（pthread_cond_wait）原子地释放互斥锁，并在条件变量 cond 上阻塞调用线程；这里的原子性指的是“相对于另一个线程对互斥锁和条件变量的访问而言是原子的”。也就是说，如果另一个线程能够在即将阻塞的线程释放互斥锁后获取互斥锁，那么在该线程中对pthread_cond_signal（）或pthread_cond_broadcast（）的后续调用的行为就像它是在即将阻塞的线程阻塞后发出的一样。

看完这段直译是不是 cpu 有点发烫？我来把他翻译为人话：

• pthread_cond_wait 接收两个参数，第一个参数是条件变量，第二个参数是保护条件变量的锁，调用 pthread_cond_wait 的线程一定要提前持有了该条件变量的锁（官方要求，否则会发生未定义的行为）。
• 进入 pthread_cond_wait 函数后，在 pthread_cond_wait 中可以原子性地释放锁：因为 pthread_cond_wait 做的主要工作就是阻塞当前线程，但是由于当前线程还持有条件变量的锁，所以 pthread_cond_wait 还应该负责释放该锁，这样其他线程才能操作该条件变量。所以 pthread_cond_wait 中要做两件事：
  1. 要释放锁
  2. 还要阻塞线程
• 那么是先释放锁？还是先阻塞线程呢？
  1. 假如先阻塞线程，那么锁就无法被释放了，因为线程一旦被阻塞，就失去了 cpu 的控制权，只能被动等待 schedule，所以不能先阻塞线程
  2. 假如先释放锁，那么就会发生 lost wakeup 的情况，因为存在这样的顺序：A 线程调用了 pthread_cond_wait，然后先释放了锁，B 线程拿到锁，执行一些业务逻辑后发现 cond 满足要求，于是调用 pthread_cond_broadcast 唤醒了所有线程，最后 A 线程进入才进入 wait （sleep）阶段
  3. 所以不管先释放锁、先阻塞线程都有问题啊，但不用担心，这就是 pthread_cond_wait 存在的意义，他可以原子性地释放锁
  4. 这里的原子性就是指，一旦 B 线程有机会拿到锁，就意味着 A 线程已经阻塞完成了，作为 pthread_cond_wait 的使用者，你不必考虑A 线程是先释放锁还是先阻塞线程，你只需要知道 pthread_cond_wait 可以原子性地释放锁就好
• 所以 pthread_cond_wait 作为 UNIX 提供的工具（其实是 POSIX 要求 UNIX 实现的接口规范），和本节讲的 xv6 的 Sleep&Wakeup 一样，也是一种实现 Coordination 的方式，也解决了 lost wakeup 的问题。

了解了 pthread_cond_wait 后，pthread_cond_broadcast 就更简单了，是用来 unblocked的，当条件变量满足时，就主动调用这个函数解除阻塞。

所以有了 pthread_cond_wait 和 pthread_cond_broadcast 这两个这么好用的工具，写出 Barrier 函数是很简单的：

struct barrier {
  pthread_mutex_t barrier_mutex;
  pthread_cond_t barrier_cond;
  int nthread;      // Number of threads that have reached this round of the barrier
  int round;     // Barrier round
} bstate;

static void 
barrier()
{
  // YOUR CODE HERE
  //
  // Block until all threads have called barrier() and
  // then increment bstate.round.
  //
  pthread_mutex_lock(&bstate.barrier_mutex); // line 16
  bstate.nthread++;
  if(bstate.nthread < nthread){
    pthread_cond_wait(&bstate.barrier_cond, &bstate.barrier_mutex);
  } else {
    bstate.nthread = 0;
    // printf("round=%d\n", bstate.round);
    pthread_cond_broadcast(&bstate.barrier_cond);
    bstate.round++; 
  }
  pthread_mutex_unlock(&bstate.barrier_mutex);  // line 26
  
}

这里再次强调，上方代码注释中 line16 和 line 26 的代码，他们可不是一对对应的加解锁啊！因为在 pthread_cond_wait 中会先解锁再加锁， 这其实是两对加解锁，伪代码如下，其中的 A1&A2 和 B1&B2 才是正确的两对加解锁，A1 和 B2 只是表面上像一对而已：

static void 
barrier()
{
  pthread_mutex_lock(&bstate.barrier_mutex); // 加锁----------A1
  
    pthread_cond_wait(&bstate.barrier_cond, &bstate.barrier_mutex) {
				// 原子性解锁------------------------------------------A2
        //  some code
        // 再次加锁------------------------------------------B1
    }
  
  pthread_mutex_unlock(&bstate.barrier_mutex);  // 解锁------B2
  
}

总结

ok，以上就是 MIT6.s081 中关于进程的 scheduling 的所有内容了，在课程中一共占据两节内容，第一节是细节繁多的进程切换过程，主要是 swtch 函数中对于 ra 的巧妙使用使得进程之间完成切换，第二节是关于 lost wakeup 问题的解决，虽然把锁作为参数传入 sleep 函数是一个比较丑陋的实现，是效果却很好，而且也是一种通用的写法~教授也提到有一些更加优雅的解决如 semaphore，无需知道任何锁的信息，但使用场景有限。

对了，我目前在寻找工作机会，本人计算机基础比较扎实，独立完成了 CSAPP（计算机组成）、MIT6.s081（操作系统）、MIT6.824（分布式）、Stanford CS144 NetWorking（计算机网络）、CMU15-445（数据库基础，leaderboard 可查） 等硬核课程的所有 lab，如果有内推名额的大佬可私信我，我来发简历。

获得更好的阅读体验，这里是我的博客，欢迎访问：byFMH - 博客园

所有代码见：我的GitHub实现（记得切换到相应分支）