絮絮叨

这两节主要介绍 xv6 中的线程切换，首先预警说明，这节课程的容量和第 5/6 节：进程的用户态到内核态的切换一样，细节多到爆炸，连我自己复习时都有点懵，看来以后不能偷懒了，学完课程之后要马上写博客总结。但是不要怕，这节的内容真的特别有趣！

同时，再次强调这个系列的博客（其实包括我所有的博客）都不是课程的中文翻译或者简单摘抄，是我学完课程之后的思考和总结，是综合了视频、xv6 book、xv6 源码以及大量资料后的思考和总结的成果，写博客真的好累啊，点收藏的小伙伴如果觉得俺写的还行顺便点个赞呗，这个点赞-收藏比如此悬殊让我有点绷不住 233333，刚从知乎上学了个金句：“反正收藏了你也不看，点赞意思下得了。。。”~如果想要看课程中文翻译的童鞋点[这里](11.1 线程（Thread）概述 - MIT6.S081 (gitbook.io))

ps：xv6 中一个进程只包含一个线程，所以老师在讲课时并没有特意区分进程和线程（值得吐槽），本文中也是混用的，，但是要记住本节讲的是进程的切换，如果出现线程字样，也请理解为进程。

引言

说到进程切换，我相信即使是计算机的初学者，都能说上一句：

“ 就是上下文切换嘛！保存 A 线程的各种寄存器信息等，然后恢复 B 进程的相应信息，这样就由 A 进程切换到了 B 进程。”

这样的回答没有任何问题，上下文切换确实是线程切换最核心的思想，但是线程切换的一个重要特点是：思想很简单，工程实现十分晦涩，连 xv6 book 都承认这部分代码是整个 xv6 中 最晦涩难懂的一部分代码：

the implementation is some of the most opaque code in xv6.

但还是那句话，魔鬼隐藏在细节中，之所以实现如此晦涩，是因为线程切换面临以下几个难点：

1. 切换线程时，保存线程的哪些信息？在哪保存它们？
2. 什么时候切换线程？是当前线程自愿让出 cpu 还是 cpu 强制其让出？如何自愿？如何强制？
3. 线程切换对于用户进程如何实现透明？比如你有一个单核 cpu，需要运行 2 个进程，那么这 2 个进程一定是都认为自己独占 cpu（就像认为自己独占内存一样），如何做到这一点？
4. 和上一个情况相反，如果有多个 cpu 核，但是只有一个待调度的进程，如何加锁来防止这个进程在多个 cpu 上运行？

以上几点如果要在工程代码中全部解决，确实需要花一番心思的，下面来看细节。

进程切换的细节

以两个计算密集型进程为例，我们讨论在不主动 yield（出让）cpu 的情况下，进程是如何切换的。

当一个进程运行的时间足够久，以至于硬件产生周期性的定时器中断，该中断信号传入内核，程序运行的控制权从用户空间代码切换到内核中的中断处理程序（注，因为中断处理程序优先级更高）

usertrap() 函数在第5/6节已经讲过，这是 xv6 内核空间中的一个函数，如果有中断、异常、系统调用发生，就会跳转到这里，在这里进行进一步判断并且运行相应的处理程序

//
// handle an interrupt, exception, or system call from user space.
// called from trampoline.S
//
void usertrap(void)
{
  int which_dev = devintr()
  // save user program counter.
  struct proc *p = myproc();
  // 由于中断发生时处于用户空间，而且中断发生时 PC-> SEPC，所以这里 p->trapframe->epc 被赋予了用户空间中某条指令的地址
  p->trapframe->epc = r_sepc();

    // some code ignore
  // ...
  // ...
  // ...

  // give up the CPU if this is a timer interrupt.
  if(which_dev == 2)
    yield();

  usertrapret();
}

RISC-V 中规定了如果是定时器中断 which_dev的值会被置为 2，所以如果是定时器中断，就会运行函数yield()，yield()的作用是该内核线程自愿地将 cpu 出让（yield）给线程调度器，并告诉线程调度器：你可以让一些其他的线程运行了，下面是yield()的实现，可以看到核心函数是 sched()，并且将旧线程的状态由"RUNNING"改为"RUNABLE"，将一个正在运行的线程转换成了一个当前不在运行但随时可以再运行的线程。：

// Give up the CPU for one scheduling round.
void
yield(void)
{
  struct proc *p = myproc();
  acquire(&p->lock);
  p->state = RUNNABLE;
  sched();
  release(&p->lock);
}

来看核心函数是 sched()，其中核心函数是swtch()，注意这里不是拼写错误，因为 switch 是 C 语言的关键字，不能作为函数名，所以采用 swtch 作为函数名：

void
sched(void)
{
  // ignore some check code
  // ...
  // ...
  // ...
  struct proc *p = myproc();
  swtch(&p->context, &mycpu()->context);

}

swtch函数会保存用户进程P1对应内核线程的寄存器至P1的 context 对象。然后将 cpu 的 context 对象恢复到相应的寄存器中，因为需要直接和寄存器打交道，所以 swtch 函数的代码是用汇编写的:

可以看到所谓的内核线程寄存器就是指 ra、sp、s0~s11 这 14 个寄存器。a0 寄存器对应着 swtch 函数的第一个参数，也就是当前线程的 context 对象地址；a1 寄存器对应着 swtch 函数的第二个参数，也就是即将要切换到的调度器线程的 context 对象地址

为什么RISC-V中有32个寄存器，但是swtch函数中只保存并恢复了14个寄存器?

因为swtch函数是从C代码调用的，所以我们知道Caller Saved Register会被C编译器保存在当前的栈上。Caller Saved Register大概有15-18个，而我们在swtch函数中只需要处理C编译器不会保存，但是对于swtch函数又有用的一些寄存器。所以在切换线程的时候，我们只需要保存Callee Saved Register。

# Save current registers in old. Load from new.
.globl swtch
swtch:
       # 上半部分，将 14 个寄存器的值保存到当前线程的 context 对象中
        sd ra, 0(a0)
        sd sp, 8(a0)
        sd s0, 16(a0)
        sd s1, 24(a0)
        sd s2, 32(a0)
        sd s3, 40(a0)
        sd s4, 48(a0)
        sd s5, 56(a0)
        sd s6, 64(a0)
        sd s7, 72(a0)
        sd s8, 80(a0)
        sd s9, 88(a0)
        sd s10, 96(a0)
        sd s11, 104(a0)
             # 下半部分，从 cpu 的 context 对象中恢复调度器线程的 14 个寄存器的值
        ld ra, 0(a1)
        ld sp, 8(a1)
        ld s0, 16(a1)
        ld s1, 24(a1)
        ld s2, 32(a1)
        ld s3, 40(a1)
        ld s4, 48(a1)
        ld s5, 56(a1)
        ld s6, 64(a1)
        ld s7, 72(a1)
        ld s8, 80(a1)
        ld s9, 88(a1)
        ld s10, 96(a1)
        ld s11, 104(a1)

        # 指令 ret 被调用的时候，指令寄存器 pc 会被重置到 ra 所保存的地址。 经打印 ra 的值可以发现是返回到 scheduler 函数中
        ret

这里要特别注意 ra 和 sp 寄存器的值，这也是理解整个线程切换的关键的关键！！！

打印出保存之前的 ra 的值：可以发现是 sched 函数中的地址，这也符合逻辑，因为我们在 sched 函数中调用了 swtch 函数，在 swtch 函数中做的第一件事就是保存 ra 寄存器，这时 ra 寄存器的值是当前进程中 swtch 函数执行完毕后的地址，也就是 sched 函数中 swtch 函数的下一行的地址

再打印出恢复之后 ra 的值，现在 ra 寄存器的值是 0x80001f2e，

打印出这个地址的指令，发现这个地址在 scheduler 函数中，意味着在 swtch 函数的后半部分：切换到调度器线程执行完毕后，函数会返回到 scheduler 函数中

完成 swtch 函数的后半部分：恢复从 cpu 中 14 个寄存器的值后，虽然依旧在 swtch 函数中，但已经不是 usertrap() -> yield() -> sched() -> swtch 这个链条上的 swtch 函数了，而是 scheduler() -> swtch 这个链条上的 swtch 函数；

至于 scheduler 函数什么时候运行并且调用了 swtch 函数，这里先大概说一下，下面会详解：scheduler 函数属于调度器线程，而调度器线程 是 xv6 系统启动的最后一环，所以调度器线程早就随着 xv6 的启动而启动了。

// Per-CPU process scheduler.
// Each CPU calls scheduler() after setting itself up.
// Scheduler never returns.  It loops, doing:
//  - choose a process to run.
//  - swtch to start running that process.
//  - eventually that process transfers control
//    via swtch back to the scheduler.
void
scheduler(void)
{
  struct proc *p;
  struct cpu *c = mycpu();

  c->proc = 0;
  for(;;){
    // Avoid deadlock by ensuring that devices can interrupt.
    intr_on();

    for(p = proc; p < &proc[NPROC]; p++) {
      acquire(&p->lock);
      if(p->state == RUNNABLE) {
        // Switch to chosen process.  It is the process's job
        // to release its lock and then reacquire it
        // before jumping back to us.
        p->state = RUNNING;
        c->proc = p;
        swtch(&c->context, &p->context);
        // ---------------特别注意，这里就是地址 0x80001f2e 处

        // Process is done running for now.
        // It should have changed its p->state before coming back.
        c->proc = 0;
      }
      release(&p->lock);
    }
  }
}

现在捋一下这个过程：

有两个进程 P1 和 P2， 1 个 CPU core， 先运行 P2，然后运行 P1，然后再运行 P2，那么 P1-> P2 是怎么切换的？

答：xv6 运行 P1 一段时间后，定时器中断被周期性触发，进程 P1 陷入内核态，在内核态中，保存公用寄存器（14 个）的状态到 P1->context 结构体中，然后恢复 cpu->context 结构体的数据到公用寄存器中，这样就把切换到了调度器线程，调度器线程会寻找一个进程状态为 RUNABLE 的进程（即 P2），将其状态修改为 RUNNING，然后调用 swtch 切换公用寄存器的状态为 P2->context，此时由调度器线程切换到 P2 的内核进程，接着返回到用户态，便完成了 P1->P2 的切换，如下图演示的这样：

这个过程中最妙的地方在于对于 ra 寄存器的巧妙使用，使 swtch 函数巧妙地返回到了调度器进程中，然后又巧妙地返回到另一个用户进程中，多到爆炸的细节见下图：尤其注意图中橙色的箭头就是 swtch 函数返回的路径。即 P1 的 shced 函数的 swtch 函数执行完毕后，就跳转到 scheduler 函数的 c->proc = 0 这一行开始执行。当 shceduler() 函数的 swtch 执行完毕后，就跳转到 P2 的 sched 函数的 swtch 函数的下一行开始执行。

具体细节见下图：

这里最妙的地方在于调度器进程是怎么保持连续性的，如下面的代码所示，scheduler 函数最核心的部分就是调用 swtch 函数，当进程 P1 由于终端切换到调度器进程时，会从地点 1 开始执行（因为 ra 指向地点 1），经过循环后到达地点 2，然后执行 swtch 后，离开 scheduler 函数，返回到 P2 的内核进程（因为 ra 指向该地址），下次又有中断需要切换进程时，又会从地点 1 开始执行，所以 scheduler 函数就是连贯的，遍历时 p 的值一直保存在调度器进程的 stack 中，并不会丢失。

void
scheduler(void)
{
  for(;;){
    // ......ignore some code

    for(p = proc; p < &proc[NPROC]; p++) {

      //地点 2，执行完 swtch 后离开 scheduler
        swtch(&c->context, &p->context);
      // 地点 1，进入 scheduler

    // ......ignore some code
    }
  }
}

线程除了寄存器以外的还有很多其他状态，它有变量，堆中的数据等等，但是所有的这些数据都在内存中，并且会保持不变。我们没有改变线程的任何栈或者堆数据。所以线程切换的过程中，cpu 中的寄存器是唯一的不稳定状态，且需要保存并恢复。而所有其他在内存中的数据会保存在内存中不被改变，所以不用特意保存并恢复。我们只是保存并恢复了cpu 中的寄存器，因为我们想在新的线程中也使用这组寄存器。

刚刚的过程我们已经看到了，当调用swtch函数的时候，实际上是从 P1 对于 swtch 的调用切换到了 P2 对于 switch 的调用（实际上是从 P1 对于 swtch 的调用切换到 调度器进程对于 swtch 的调用；从 调度器进程对于 swtch 的调用切换到 P2 对于 swtch 的调用，这里这么说只是为了宏观上的理解），为什么能从 cpu 的调度器线程切换到 P2 的内核进程呢？关键就是P2 的 context -> ra 的值是 P2 的 sched 函数的 swtch 函数的下一行，当这个地址被从 context 中恢复到 ra 寄存器中后，就会根据该地址返回到 P2 进程的 swtch 函数的下一行。

这里有一个关键问题就是如果 P2 进程是第一次被调度，那么 context->ra 的值就不会是 P2 的 sched 函数的 swtch 函数的下一行，原因也很简单啊，因为之前 P2 一定是在运行，然后主动或者被动调用了 yield() 函数，出让了 cpu，所以 ra 的值就保存了出让的那一刻的地址，也就是 P2 的 sched 函数的 swtch 函数的下一行，但是如果之前 P2 没有运行，而是第一次被调度，就需要我们手动设置 P2->context->ra 的值了，在 xv6 中，这个值在 allocproc() 函数中被设置为 p->context.ra = (uint64)forkret;这个函数如下：

// A fork child's very first scheduling by scheduler()
// will swtch to forkret.
void
forkret(void)
{
  static int first = 1;

  // Still holding p->lock from scheduler.
  release(&myproc()->lock);

  if (first) {
    // File system initialization must be run in the context of a
    // regular process (e.g., because it calls sleep), and thus cannot
    // be run from main().
    first = 0;
    fsinit(ROOTDEV);
  }

  usertrapret();
}

这个函数其实做的工作很简单，当调用 fork 函数分配的子进程准备好后，会先在池子中等待 scheduler() 函数调度，当呗调度后，就会返回到 forkret 函数中，在这个函数中返回到用户空间，这里其实也解释了为什么 fork() 函数可以一次调用，两次返回。

在第一节中我们就了解到，fork()函数是一次调用两次返回，在父进程中返回子进程 pid，在子进程中返回 0，所以fork 的典型用法就是：

//pid_t fork(void);
pid_t pid = fork();
if(pid) // parent process
{
    //do something in parent process
} else
{
    //do something in child process
}

这好像和我们 c 语言的是相反的，怎么可能一个函数调用一次，有两个返回值呢？？？

别急，学完前面的知识，我们就能理解这件事了，来看 fork 函数的实现：

int fork(void)
{
    struct proc *child_process = allocproc();
    // copy memory page table...
    // copy fp and other properties
    child_process->state = RUNNABLE;
    child_process->trapframe->a0 = 0;// return value is 0 for child_process fork()
    return child_process->pid;
}

刚刚说过，在 allocproc() 函数设置为 p->context.ra = (uint64)forkret;根据 ra 的值，所以子进程将来被调度后，会返回到 forkret 函数中，进而返回到用户空间，并且子进程保存返回值的 trapframe->a0 被设置为 0，而父进程的 trapframe->a0 被设置为子进程的 pid

void syscall(void)
{
    //...
    p->trapframe->a0 = fork(); //return value for parent is child_pid
    //...
}

父进程遵循 c 语言的直觉，调用了 fork 函数，然后把自己的内存复制给子进程，并且返回了子进程的 pid，而子进程没有立即返回，而是等待 scheduler 调度，调度后返回到 forkret 函数进而返回到用户空间，并且由于父子进程的 trapframe page 是一样的，下一行的代码地址是由 trapframe page 的 epc 变量保存的，所以 pid_t pid = fork();这行代码在父进程中被执行后，子进程也会执行这行代码，给 pid 重新赋值为 0。

所以总结一下就是两个要点：

1. 由于复制了trapframe page ，所以 pid_t pid = fork();会被父子进程都执行
2. fork() 是系统调用，进入内核态后父进程会新建一个子进程，父子进程会分别从内核态返回到用户空间，父进程是系统调用的正常返回到用户空间，子进程由于生在内核空间，是由 scheduler 调用返回到 forkret 函数然后返回到用户空间

lab6 Thread 心得

这一节的三个小 lab 都是 morerate 级别的，思路和实现都比较简单

设计并实现一个用户级别的线程切换机制，我理解就是为 xv6 实现多线程机制，其实相当于在用户态重新实现一遍 xv6 中的 scheduler() 和 swtch() 的功能，所以大多数代码都是可以借鉴的。

而且由于是“用户级”的线程，所以无需 trap 到内核态，只需要在进程中设置 n 个 thread 结构体，每个结构体都有空间保存自己的 context 即可。

也无需使用时钟中断来强制执行调度，由线程主动调用 yield() 来出让 cpu 、重新调度，这里的代码比较简单，就直接都贴出来了

#include "kernel/types.h"
#include "kernel/stat.h"
#include "user/user.h"

/* Possible states of a thread: */
#define FREE        0x0
#define RUNNING     0x1
#define RUNNABLE    0x2

#define STACK_SIZE  8192
#define MAX_THREAD  4

// Saved registers for thread context switches.
struct context {
  uint64 ra;
  uint64 sp;

  // callee-saved
  uint64 s0;
  uint64 s1;
  uint64 s2;
  uint64 s3;
  uint64 s4;
  uint64 s5;
  uint64 s6;
  uint64 s7;
  uint64 s8;
  uint64 s9;
  uint64 s10;
  uint64 s11;
};

struct thread {
  char       stack[STACK_SIZE]; /* the thread's stack */
  int        state;             /* FREE, RUNNING, RUNNABLE */
  struct context ctx; // 在 thread 中添加 context 结构体
};
struct thread all_thread[MAX_THREAD];
struct thread *current_thread;
extern void thread_switch(uint64, uint64);

void
thread_init(void)
{
  // main() is thread 0, which will make the first invocation to
  // thread_schedule().  it needs a stack so that the first thread_switch() can
  // save thread 0's state.  thread_schedule() won't run the main thread ever
  // again, because its state is set to RUNNING, and thread_schedule() selects
  // a RUNNABLE thread.
  current_thread = &all_thread[0];
  current_thread->state = RUNNING;
}

void
thread_schedule(void)
{
  struct thread *t, *next_thread;

  /* Find another runnable thread. */
  next_thread = 0;
  t = current_thread + 1;
  for(int i = 0; i < MAX_THREAD; i++){
    if(t >= all_thread + MAX_THREAD)
      t = all_thread;
    if(t->state == RUNNABLE) {
      next_thread = t;
      break;
    }
    t = t + 1;
  }

  if (next_thread == 0) {
    printf("thread_schedule: no runnable threads\n");
    exit(-1);
  }

  if (current_thread != next_thread) {         /* switch threads?  */
    next_thread->state = RUNNING;
    t = current_thread;
    current_thread = next_thread;
    /* YOUR CODE HERE
     * Invoke thread_switch to switch from t to next_thread:
     * thread_switch(??, ??);
     */
    thread_switch((uint64)&t->ctx, (uint64)&next_thread->ctx);
  } else
    next_thread = 0;
}

void
thread_create(void (*func)())
{
  struct thread *t;

  for (t = all_thread; t < all_thread + MAX_THREAD; t++) {
    if (t->state == FREE) break;
  }
  t->state = RUNNABLE;
  // YOUR CODE HERE
  t->ctx.ra = (uint64)func;       // 返回地址
  // thread_switch 的结尾会返回到 ra，从而运行线程代码
  t->ctx.sp = (uint64)&t->stack + STACK_SIZE ;  // 栈指针
}

void
thread_yield(void)
{
  current_thread->state = RUNNABLE;
  thread_schedule();
}

volatile int a_started, b_started, c_started;
volatile int a_n, b_n, c_n;

void
thread_a(void)
{
  int i;
  printf("thread_a started\n");
  a_started = 1;
  while(b_started == 0 || c_started == 0)
    thread_yield();

  for (i = 0; i < 100; i++) {
    printf("thread_a %d\n", i);
    a_n += 1;
    thread_yield();
  }
  printf("thread_a: exit after %d\n", a_n);

  current_thread->state = FREE;
  thread_schedule();
}

void
thread_b(void)
{
  int i;
  printf("thread_b started\n");
  b_started = 1;
  while(a_started == 0 || c_started == 0)
    thread_yield();

  for (i = 0; i < 100; i++) {
    printf("thread_b %d\n", i);
    b_n += 1;
    thread_yield();
  }
  printf("thread_b: exit after %d\n", b_n);

  current_thread->state = FREE;
  thread_schedule();
}

void
thread_c(void)
{
  int i;
  printf("thread_c started\n");
  c_started = 1;
  while(a_started == 0 || b_started == 0)
    thread_yield();

  for (i = 0; i < 100; i++) {
    printf("thread_c %d\n", i);
    c_n += 1;
    thread_yield();
  }
  printf("thread_c: exit after %d\n", c_n);

  current_thread->state = FREE;
  thread_schedule();
}

int
main(int argc, char *argv[])
{
  a_started = b_started = c_started = 0;
  a_n = b_n = c_n = 0;
  thread_init();
  thread_create(thread_a);
  thread_create(thread_b);
  thread_create(thread_c);
  thread_schedule();
  exit(0);
}

这个可以说是整个课程中最简单的 lab 了，要做的就是两点：

1. 为 hashtable 加大表保证多线程操作的正确性；
2. 降低锁的粒度，为每个 hashtable 的 bucket 加锁以提高并发性。

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

简单来说，一段代码被多个线程执行，如何保证多个线程都到了其中某一点之后，才能继续往下执行？但是由于这个 lab 涉及到 lost wake-up 问题，所以我打算放到下一节一起复习~

ok，本节就到这里，本门课程最难的一节就复习完啦，按照 lab 的线索，接下来再写 3 篇，这个系列就收工~

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

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

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