一个操作系统运行的进程可能比这个计算机的CPU数量多，因此操作系统需要规划如何在进程间共享CPU。理想状态下要让这个过程对于用户进程来说是透明的，让每个进程有一种自己单独拥有一个CPU的错觉。

xv6-book Chapter 7 部分内容

一、多路复用

xv6通过在两种情况下切换进程来实现多路复用：

1. 等待IO或事件：当进程需要等待磁盘、管道或子程序退出时，会调用sleep主动让出CPU，等到条件满足时被wakeup唤醒。

2. 时间片到期：一个进程不能一直持有CPU，否则其他进程将发生饥饿，因此有时间片机制；xv6会周期性强制一个占用CPU较长时间的进程放弃CPU（yield）从而实现抢占式多任务。

二、上下文切换

上下文切换是调度的核心机制。xv6不直接从一个用户进程切换到另一个用户进程，而是遵循以下路径（如图6-1所示）：

$$\begin{aligned} \text{old (Process A)} \to \text{current CPU scheduler process (Scheduler)}\to \text{new (Process B)} \end{aligned}$$

swtch（kernel/swtch.S）汇编指令为内核线程切换执行保存和恢复功能，这也是这一章的重点：

swtch的唯一任务就是把当前CPU寄存器的快照保存到旧进程的上下文结构（context）中，然后将新进程先前保存好的快照恢复到CPU寄存器中。

它只保存被调用者保存的寄存器（Callee-saved registers），包括：

• ra：返回地址
• sp：栈指针
• s0-s11：被调用者保存寄存器

这里不需要保存所有的32个寄存器，因为gcc编译器会自动在进行函数调用时保存部分寄存器，这些寄存器可以叫做Caller-saved registers（如a0-a7,t0-t6），在进入这个函数之前，这些寄存器已经被编译器自动保存到了内核栈中。

如果在swtch中重复保存这些已经被保存过的寄存器，将造成资源的浪费。

这里不需要保存pc寄存器的值：因为pc寄存器对于用户是不可见的，它一直指向下一个指令的地址，其本身是无法被汇编指令直接读写的；这里只需要保存ra寄存器的值即可，因为在C语言中调用一个函数，ra寄存器会自动写入返回地址，调用ret指令后会将ra中的地址加载到pc中，这个地址也就是执行完被调用函数后返回的指令地址即进程被重新调度后应该执行的位置。

三、调度

在xv6中，swtch不仅仅是寄存器的交换，也是一次信任的交付。p->lock就像一根接力棒，在进程内核栈与CPU调度器栈之间传递，打破了传统的加锁规范。

1. Coroutine relationship between sched and scheduler

   • 内核线程不是从A直接切换到B，它总是：进程A的内核线程->scheduler->进程B的内核线程
   • 协程（Coroutine）：sched和scheduler不是简单的函数调用，他们是彼此的协程，执行流程在(kernel/proc.c:456),(kernel/proc.c:490)之间反复跳，这样故意将控制权通过线程切换转移的程序称为协程。
   • 在创建新进程时scheduler对swtch的调用并不会在sched中结束，allocproc会把ra寄存器设置为forkret，因此调用scheduler时获取到的p->lock会在forkret中释放。forkret存在的目的就是为了释放p->lock，来让控制流能够不被中断；如果不考虑这一点，新进程可以直接调用usertrapret返回用户空间。

2. 跨线程的p->lock传递

   通常在用到锁时会遵循“谁加锁，谁解锁”的准则。但在线程调度中，p->lock由一个线程获取（如yield,sleep,exit），再由另一个线程释放（scheduler）。

   锁的持有权随着swtch将控制权转移一起发生了转移。

   这样做可以保护不变性：当进程从RUNNING变成RUNNABLE时，这个状态改变是不完整的；如果没有这把跨越swtch的锁，另一个CPU可能直接看到这个进程的状态变成了RUNNABLE，然后尝试运行它，但此时进程还在swtch过程中，就可能导致两个CPU同时操作一个内核栈从而引发系统错误。

3. 调度器的不变性（Invariants）

状态	必须满足的不变性
RUNNING	1. CPU寄存器持有进程的真实值；2. c->proc指向该进程。
RUNNABLE	1.寄存器存在p->context中；2. 没有CPU在该线程上运行；3. 没有CPU的c->proc指向它。

四、Sleep & wakeup

为了实现进程间协作，xv6提供了sleep和wakeup机制： 例如当一个进程等待磁盘读取时，如果占用CPU进行轮询是十分浪费CPU资源的，此时应该将CPU让给其他需要计算的进程，这就是sleep机制；如果磁盘读取完成，应该告诉先前等待读取的进程这个消息，原本的进程已经可以继续执行了，但是此时它还处于sleep中，因此需要wakeup机制去唤醒这个进程。

丢失唤醒（Lost Wake-up Problem）问题：

这是并发编程中的经典BUG，如果sleep在检查条件和真正进入睡眠之间被中断，此时wakeup发生了，那么wakeup就不起作用，等到wakeup结束后才真正进入睡眠状态，那么该进程将处于永久的休眠中。

解决方案：xv6要求调用sleep时一定要获取一个底层锁（Condition Lock）： sleep(chan, lock)：进程在持有底层锁的情况下检查条件，进入sleep后，内核会原子性地释放该锁并将进程标记为SLEEPING。这样确保了wakeup只有在进程真正睡眠之后才能获得那把锁并发出通知。

五、管道

xv6中管道使用上面提到的sleep和wakeup机制来进行同步，当 $pi\!\rightarrow\!nwrite = pi \! \rightarrow\!nread$ 时管道为空，当 $pi\!\rightarrow\!nwrite = pi\!\rightarrow\!nread+PIPESIZE$ 则管道为满。

以pipewrite为例，想要写入管道的buf前，一定要获取pi->lock，如果管道的buf还有空间，就将用户空间的数据不断写入管道，直到写完数据或者将管道写满；如果管道满了则先调用wakeup唤醒读进程，然后调用sleep进入休眠状态，在sleep中会释放pi->lock以便读进程能够获取到管道的控制权限。同样，piperead和pipewrite几乎一致。

六、终止和清理（Wait,exit,and kill）

• wait

父进程等待子进程退出并返回子进程的pid。首先获取wait_lock，这把锁是条件锁，防止父进程错过一个exiting子进程的wakeup。然后扫描进程表，如果找到一个子进程处于ZOMBIE状态，它释放这个子进程的资源和proc结构体，将子进程的退出状态码复制到wait提供的地址，然后返回这个子进程的pid。

如果没找到一个处于ZOMBIE状态的子进程，父进程调用sleep等待子进程exit，接着再次扫描进程表。

wait通常持有两把锁，wait_lock和np->lock，死锁避免顺序是先获取wait_lock再获取np->lock。

• exit

exit记录退出状态，释放一些资源，调用reparent将子进程传递给init进程（父进程一旦退出子进程也需要销毁），如果父进程正在wait中就唤醒它，将调用exit的进程标记为ZOMBIE并永远放弃CPU。

exit需要持有wait_lock，因为它需要wakeup父进程，持有wait_lock防止丢失唤醒。同时为了防止在exit的过程中（还未调用swtch），处在wait中的父进程被唤醒直接看到p->state为ZOMBIE，exit必须持有p->lock。

exit为了避免死锁，获取这两把锁的顺序和wait一样。

• kill

kill让一个进程请求终止另一个进程。如果直接让kill去杀死目标进程是十分复杂的，因为目标进程可能正在其他的CPU上运行，也许正处于修改内核数据结构的敏感行为中。

因此kill只做非常简单的事情：设置p->killed标志，如果目标进程处于SLEEPING中，将它唤醒。最终目标进程会进入或者退出内核，在usertrap中如果p->killed被检测到会调用exit退出。即使被kill后的目标进程在用户空间运行，很快它会因为系统调用、时钟中断或其他硬件中断进入usertrap然后销毁。

kill唤醒目标进程直接修改p->state为RUNNABLE，而不调用wakeup，原因是此时目标进程等待的条件并没有被满足，如果直接唤醒可能会造成危险的后果。

有些情况下在循环中调用sleep会检查p->killed，并且放弃当前的行为如果p->killed被设置了，只有当这种放弃是正确的才会这样做（比如等待键盘按下，防止死等用户敲键盘）。

在xv6中，一些sleep循环不检查p->killed，因为这些代码正处于应该是原子操作的多步骤系统调用的中间，比如对磁盘的操作。即使设置了p->killed标志，也需要等待当前系统调用完成，然后回到usertrap中看到这个killed标志才会进行销毁。

6.s081 Lab6: Multithreading 实现

一、Uthread: switching between threads (moderate)

添加用户线程切换机制，需要实现两个目标：

1. 当thread_schedule()第一次运行给定线程时，这个线程要在自己的栈上运行传递给thread_create(void (*func))的函数。
2. 确保swtch能够保存和恢复寄存器，能让即将返回的线程继续执行下一条上一次离开时的指令。

为了实现以上两个目标，需要添加以下代码：

• 为thread数据结构增添上下文context结构体：

struct context {
  uint64 ra;
  uint64 sp;

  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_create中初始化ctx字段，并设置ra和sp，使线程第一次执行时能够在本身的栈上执行传过去的函数：

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
  memset(&t->ctx, 0, sizeof(t->ctx));
  t->ctx.ra = (uint64)func;
  t->ctx.sp = (uint64)(t->stack + STACK_SIZE);
}

• swtch要能够完成寄存器的存储和恢复：

 .text

 /*
         * save the old thread's registers,
         * restore the new thread's registers.
         */

 .globl thread_switch
thread_switch:
 /* YOUR CODE HERE */
        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)

        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    /* return to ra */

• thread_schedule添加一行进行线程上下文切换：

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)&current_thread->ctx);
  } else
    next_thread = 0;
}

二、Using threads (moderate)

使用锁来同步哈希表的操作，这里十分简单，只需要运用以往学过的锁知识进行同步就可以了，不涉及过多xv6的知识，需要使用pthread库。

对每个table entry设置一个锁，当要访问该entry时申请获得对应的锁，然后再进行操作。

static
void
put(int key, int value)
{
  int i = key % NBUCKET;

  pthread_mutex_lock(&lock[i]);
  // is the key already present?
  struct entry *e = 0;
  for (e = table[i]; e != 0; e = e->next) {
    if (e->key == key)
      break;
  }
  if(e){
    // update the existing key.
    e->value = value;
  } else {
    // the new is new.
    insert(key, value, &table[i], table[i]);
  }

  pthread_mutex_unlock(&lock[i]);
}

这里就放一个put操作的代码，在此测试中，get操作与put操作是分离的，先将所有的put进行完再进行get操作，因此get无需上锁。

三、Barrier (moderate)

编写一个barrier函数，当所有线程到达这个barrier后才会一起放行，否则在最后一个线程到达barrier前都进行等待，这个任务也要用到pthread库。这里用到的条件变量类似于xv6中的sleep和wakeup机制。

条件变量介绍：

条件变量一般伴随着三个成员：互斥锁、条件变量、逻辑条件。

互斥锁负责保护共享数据，确保一个时刻最多只有一个线程在检查或修改条件；

条件变量是线程被挂起和唤醒的等待队列。

逻辑条件是一个真正的布尔表达式，只有这个条件满足，线程才能真正被唤醒。

判断条件时必须用while而不是if，被唤醒后的线程还没来得及加锁，条件已经被其他线程修改了，这时候就需要继续等待；还有一种情况，即使没有接收到信号，线程也可能因为硬件或不可控外部因素醒来，这叫做虚假唤醒（Spurious Wakeup）。虚假唤醒的原因是许多操作系统（如Linux）可能出现一个信号唤醒多个线程的情况，其原因是为了达到更高的性能（不是Bug,是Feature），这种做法十分常见，以至于POSIX标准明确规定这是允许的。

wait函数内部是原子性的，pthread_cond_wait执行了非常精密的操作：释放锁+线程入队休眠。如果这两个操作不是原子性的（比如先释放锁，再让线程入队休眠），可能会出现丢失唤醒的问题（Lost Wake-up）。

这一个任务要求多轮次的barrier，要求不同轮次的barrier不能互相影响，因此这里的条件我选择了使用bstate.round == myround，每个线程在进入barrier时先记录下当前轮次，如果最后一个线程进入到barrier时，会将bstate.round++，唤醒所有正在等待的线程，并将bstate.nthread设为0，开始下一次的barrier。

static void
barrier()
{
  // YOUR CODE HERE
  //
  // Block until all threads have called barrier() and
  // then increment bstate.round.
  //
  pthread_mutex_lock(&bstate.barrier_mutex);

  int myround = bstate.round;

  bstate.nthread++;
  if (bstate.nthread < nthread) {
    while (bstate.round == myround) {
      pthread_cond_wait(&bstate.barrier_cond, &bstate.barrier_mutex);
    }
  } else {
    bstate.round++;
    bstate.nthread = 0;
    pthread_cond_broadcast(&bstate.barrier_cond);
  }

  pthread_mutex_unlock(&bstate.barrier_mutex);
}

本次任务结束，感觉xv6-book对多线程的内容写的十分丰富，反而是lab中涉及的并不是很多。