实验文档链接：Lab1: Threads

我的实现（更新至Lab 2）：Altair-Alpha/pintos

开始之前

如文档所述，在开始编写代码之前需要先阅读掌握 GETTING STARTED 部分和附录中 Code Guide 部分的前四节（重点是 Synchronization）、4.4BSD Scheduler 部分以及 C Standards 部分内容。

博主自己整理的一些本节中你可能会用到的命令行：

# docker运行容器
docker run -it --rm --name pintos --mount type=bind,source="你的Pintos路径",target=/home/PKUOS/pintos pkuflyingpig/pintos bash
# 启动另一个命令行
docker exec -it pintos bash
# 启动gdb
cd pintos/src/threads/build && pintos-gdb kernel.o
# 编译并运行单个测试（根据需要更换测试名，注意要有.result后缀）
make && make tests/threads/priority-donate-multiple.result
# 编译并调试单个测试（根据需要更换测试名）
make && pintos --gdb -- run priority-donate-multiple

# 打印list的第一个元素的某成员，这里以打印ready_list的第一个线程的名称为例
# 使用原因是访问列表元素的list_entry是宏，无法直接打印（除非改编译选项），这里的命令是手动展开后的结果
# 可以将list_begin(&ready_list)换成任何返回值为list_elem的表达式
((struct thread *) ((uint8_t *) &(list_begin(&ready_list))->next - ((size_t)&(((struct thread*)0)->elem.next))))->name

Task 1: Alarm Clock

Exercise 1.1

• Reimplement timer_sleep(), defined in devices/timer.c.
  • Although a working implementation is provided, it “busy waits,” that is, it spins in a loop checking the current
  • time and calling thread_yield() until enough time has gone by. Reimplement it to avoid busy waiting.

本节实验的热身环节，要求将 timer.c 中已经用忙等待实现的线程睡眠函数 timer_sleep 改为用非忙等实现。timer_sleep 接受一个参数，为请求睡眠的 tick 数。timer.c 中其它基于物理时间的睡眠函数最终也是通过该函数实现。原代码为：

/** Sleeps for approximately TICKS timer ticks. Interrupts must
   be turned on. */
void
timer_sleep (int64_t ticks) 
{
  int64_t start = timer_ticks ();

  ASSERT (intr_get_level () == INTR_ON);
  while (timer_elapsed (start) < ticks) 
    thread_yield ();
}

timer_ticks 函数返回自开机以来经历的 tick 数，time_elapsed 则是将当前 tick 数与参数 tick 数相减得到 tick 的时间差，原实现通过一个 while 循环询问已经历的时间直到 ticks，如果没到达则调用 thread_yield，该函数让出 CPU 但仍将自身设为随时可被调度的 THREAD_READY 状态，因此是 busy wait 的。

分析

实现的要点在于将不断主动询问的行为改为先将自身阻塞，然后等待一个管理者被动地唤醒。如此，就必须要有这样一个高频掌握时间更新的管理者，其是否存在呢？当然。考虑时间系统如何更新，关注到 timer.c 中有一个全局变量 ticks 记录了开机至今的 tick 数。它是如何被更新的？答案是在 timer_interrupt 函数中，该函数正是时间中断的 handler，也就是 CPU 在每个时间片都会调用该函数通知时间的更新。时间中断是一种外部中断，我们在 timer_init 初始化函数中也可以看到该 handler 函数的注册：

intr_register_ext (0x20, timer_interrupt, "8254 Timer");

于是，我们可以在 timer_sleep 中将线程和睡眠时间记录起来，然后转为阻塞状态。在被高频调用的 timer_interrupt 中，更新所有睡眠线程的时间，如果降至 0，则将其唤醒。

实现

首先从设计上，我不希望为了睡眠这样的非核心功能而向 thread 结构体添加成员，因此我选择仅在 timer.c 中添加代码。定义如下结构：

/** Record sleeping threads. */
struct sleep_thread_entry
{
  struct thread *t;
  int64_t sleep_ticks;
  struct list_elem elem;
};
static struct list sleep_list;

• 关于 list 的实现请详细阅读源码，使用上与 C++ 的 std::list 有明显区别

该结构由一个线程和其请求的睡眠时间构成，然后以一个 sleep_list 记录所有的项。

在 timer_init 中，初始化 sleep_list：

list_init(&sleep_list);

在新版的 timer_sleep 中，构建一个记录项，放入 sleep_list 中，然后转为阻塞状态即可：

void
timer_sleep (int64_t ticks) 
{
  if (ticks < 0) {
    return;
  }

  enum intr_level old_level = intr_disable();

  struct sleep_thread_entry *entry = (struct sleep_thread_entry *)
                                      malloc(sizeof(struct sleep_thread_entry));
  entry->t = thread_current();
  entry->sleep_ticks = ticks;
  list_push_back(&sleep_list, &entry->elem);
  thread_block();

  intr_set_level(old_level);
}

• 当 ticks < 0 时，测试期望的行为是正常返回而非 Kernel Panic，所以不能使用 ASSERT
• 当 ticks = 0 时，参考 Linux 编程 sleep(0) 的意义，应该也执行后续代码让出 CPU，此时相当于执行了 thread_yield
• thread_block 需要在关闭中断状态下被调用

编写一个新函数 sleep_tick，遍历更新 sleep_list，并唤醒到时间的线程：

/** Tick the sleep list. Threads whose sleep tick becomes zero 
 * will be unblocked. This process shouldn't be interrupted. */
static void
sleep_tick(void)
{
  struct list_elem *e = list_begin(&sleep_list);
  while (e != list_end(&sleep_list)) {
    struct sleep_thread_entry *entry = list_entry(e, struct sleep_thread_entry, elem);
    if (--(entry->sleep_ticks) <= 0) {
      e = list_remove(e);
      thread_unblock(entry->t);
    } else {
      e = list_next(e);
    }
  }
}

• Core Guide / Interrupt Handling 文档最后部分说明了 “The handler will run with interrupts disabled”，因此这里无需再手动关闭中断
• 注意循环移除列表元素的写法（for 循环也可）

在 timer_interrupt 中，调用该函数即可：

/** Timer interrupt handler. */
static void
timer_interrupt (struct intr_frame *args UNUSED)
{
  ticks++;
  sleep_tick();
  thread_tick ();
}

运行 make check，此时与 alarm 相关的 6 个测试中除了 alarm-priority 外应该全部通过。

Task 2: Priority Scheduling

Exercise 2.1

• Implement priority scheduling in Pintos.
  • When a thread is added to the ready list that has a higher priority than the currently running thread, the current thread should immediately yield the processor to the new thread.
  • Similarly, when threads are waiting for a lock, semaphore, or condition variable, the highest priority waiting thread should be awakened first.
  • A thread may raise or lower its own priority at any time, but lowering its priority such that it no longer has the highest priority must cause it to immediately yield the CPU.

接下来要实现的是基础的优先级调度。上一小节 alarm-priority 用例不通过的原因，根据名称也能猜出来是线程唤醒的顺序不对，没有根据优先级。在 sleep_tick 中调用 thread_unblock 时，线程被通过 list_push_back 加入了一个就绪队列 ready_list 中（thread_yield 中亦如此）。在这两个函数中，schedule 被调用，使得新的线程被调度。schedule 中又通过 next_thread_to_run 来找到下一个调度的线程，其实现为：

next_thread_to_run (void) 
{
  if (list_empty (&ready_list))
    return idle_thread;
  else
    return list_entry (list_pop_front (&ready_list), struct thread, elem);
}

这只是从 ready_list 中取了第一个元素，而又因为在插入时没有考虑优先级，所以无法通过测试。于是这里我们有两种方案：

• 方案 1：修改出端，next_thread_to_run 选择时线性扫描一次 ready_list，取出优先级最高的线程。
• 方案 2：修改入端，thread_yield 和 thread_unblock 插入 ready_list 时使队列保持按优先级降序，于是取出时可以选择第一个元素即优先级最高的线程。

显然，在需要多次进行调度时，方案 2 的效率是更高的。幸运的是在 list.c 中正提供了我们需要的函数 list_insert_ordered，需要我们提供列表元素比较的函数。于是写出：

static bool
thread_priority_greater(const struct list_elem *lhs,
                    	const struct list_elem* rhs, void *aux UNUSED) {
  return list_entry(lhs, struct thread, elem)->priority
          > list_entry(rhs, struct thread, elem)->priority;
}

然后将 list_push_back 调用换为 list_insert_ordered 即可。运行测试发现 alarm-priority 已通过。

回到 2.1 的要求，这里的重点是无论有线程是因为状态变化还是优先级变化，都要立刻尝试进行调度，也就是抢占式（preemptive）调度。优先级变化对应的是 thread_set_priority。该函数设置的是当前线程的优先级，所以只有当优先级降低时才可能发生调度。修改如下：

/** Sets the current thread's priority to NEW_PRIORITY. */
void
thread_set_priority (int new_priority) 
{
  int old_priority = thread_current()->priority;
  thread_current ()->priority = new_priority;
  if (new_priority < old_priority) {
    thread_yield();
  }
}

状态变化就是有其它线程变为了就绪状态。这里的其它线程可能是新创建的线程，也可能是原有的线程，在代码中对应 thread_create 和 timer 唤醒等情况。虽然它们最终都会落到 thread_unblock，但是注意原代码注释说明了该函数不应该抢占当前线程，因为它可能在开中断和关中断两种情况下被调用：

对于开中断情况（thread_create），在 thread_unblock 后和上面一样直接调用 thread_yield 即可：

tid_t
thread_create (const char *name, int priority,
               thread_func *function, void *aux) 
{
  ...
  /* Add to run queue. */
  thread_unblock (t);
  if (priority > thread_current()->priority) {
    thread_yield();
  }

  return tid;
}

对于关中断情况，不能像上面那样调用 thread_yield（该函数中也进行了 ASSERT(!intr_context ()) 的断言）。讲义中给了我们提示：

• To yield the CPU in the interrupt context, you can take a look at functions in threads/interrupt.c.

阅读 interrupt.c，可以看到这样一段描述。红色划线部分告诉我们，在外部中断中可以通过调用 intr_yield_on_return 使得在中断返回前调度一个新的进程，这正是我们需要的！

考察该函数，可以发现其作用是将 yield_on_return 设为了 true。而这使得 intr_handler 函数能够在处理完中断即将返回时触发一次 thread_yield，完成调度：

void
intr_handler (struct intr_frame *frame) 
{
  ...
  /* Complete the processing of an external interrupt. */
  if (external) 
    {
      ASSERT (intr_get_level () == INTR_OFF);
      ASSERT (intr_context ());

      in_external_intr = false;
      pic_end_of_interrupt (frame->vec_no); 

      if (yield_on_return) 
        thread_yield (); 
    }
}

于是对于这类情况（如 timer 唤醒），应该在 thread_unblock 后使用 intr_yield_on_return：

static void
sleep_tick(void)
{
  struct list_elem *e = list_begin(&sleep_list);
  while (e != list_end(&sleep_list)) {
    struct sleep_thread_entry *entry = list_entry(e, struct sleep_thread_entry, elem);
    if (--(entry->sleep_ticks) <= 0) {
      e = list_remove(e);
      thread_unblock(entry->t);
      if (entry->t->priority > thread_current()->priority) {
        intr_yield_on_return();
      }
    } else {
      e = list_next(e);
    }
  }
}

博主认为这类如 timer 唤醒导致抢占式调度的情况是测试用例中没有覆盖到的。

Exercise 2.2.1 + 2.2.2

以上的优先级调度存在一个问题：例如分别有一个高优先级（H）、一个中优先级（M）和一个低优先级（L）的线程，假设 H 请求一个资源，L 持有该资源，而 M 又处于就绪态，就会使得高优先级的 H 一直无法运行。解决方法是 priority donation，即 H 先将优先级“捐献”给 L，使得 L 成为高优先级得到调度，待其释放资源后再回收优先级，使自身能够运行。这里的捐献只需提高对方的优先级，无需将自身优先级降低。2.2.1 和 2.2.2 小节就是要实现该机制：

• Implement priority donation.
  • You will need to account for all different situations in which priority donation is required.
  • You must implement priority donation for locks. You need not implement priority donation for the other Pintos synchronization constructs.
  • You do need to implement priority scheduling in all cases.
  • Be sure to handle multiple donations, in which multiple priorities are donated to a single thread.

• Support nested priority donation:
  • if H is waiting on a lock that M holds and M is waiting on a lock that L holds, then both M and L should be boosted to H’s priority.
  • If necessary, you may impose a reasonable limit on depth of nested priority donation, such as 8 levels.

提示信息中告诉我们要考虑两种“多”捐献情况：（1）一个线程可能持有多个锁，使得多个线程同时向一个线程捐献。（3）H 向 M，M 向 L 等多层级的捐献，应该添加合理的深度限制。

分析 & 实现

这部分是本 Lab 最烧脑的，我的实现和网上大部分 blog 差不多，讲一下自己的思考历程：

1. 首先，为了记录 donation 导致的优先级变化，thread 类中要添加另一个优先级成员。为保证原有的 priority 成员的语义不变，添加一个成员 base_priority，代表线程自身的优先级属性，而原 priority 则为可能收到捐献而提升，向外表现出的优先级。
2. 考虑多个捐献的情况，一方面，一个线程持有的一个锁可能被多个其它线程请求（并捐献），针对这个锁只需关注其中捐献最大的优先级，所以要为 lock 类添加一个成员 max_priority，初始化为 PRI_MIN。另一方面，一个线程可能持有多个锁，释放掉一个锁但还有其它锁的时候，其 priority 不能立刻被调回 base_priority，而是先下降到其它锁中的最高优先级（即 max_priority 最大的）。于是还应该在 thread 类中添加一个持有的 lock 的链表 lockhold_list，并可以将其做成优先级队列，使得从链表取头即得到最大优先级。
3. 考虑多层捐献的情况，当 H 找到 M 时，必须要能知道 M 是否正被一个锁卡住以及那个锁和持有锁的线程的信息，所以需要在 thread 类中添加一个 lock 指针 waiting_lock。

先进行以上 1 和 2 的实现。priority donation 一定发生在一个线程请求一个锁时，也就是 lock_acquire 函数中。函数分为两部分：获取到锁之前和获取到锁之后，对应 sema_down 调用之上和之下。获取到锁之前，要判断锁是否被其它线程持有，如果是且该线程优先级较低，则应进行 donate，同时也尝试进行锁的 max_priority 更新。获取到锁之后，将锁有序插入线程的 lockhold_list 中。这里顺便也把 waiting_lock 的更新加上了，后面多级捐献会用到。

void
lock_acquire (struct lock *lock)
{
  ASSERT (lock != NULL);
  ASSERT (!intr_context ());
  ASSERT (!lock_held_by_current_thread (lock));

  struct thread *t_cur = thread_current();
  
  if (lock->holder != NULL) {
    t_cur->waiting_lock = lock;
    // Try to update lock priority and holder priority.
    lock->max_priority = t_cur->priority > lock->max_priority ? t_cur->priority : lock->max_priority;
    lock->holder->priority = t_cur->priority > lock->holder->priority ? t_cur->priority : lock->holder->priority;
  }

  sema_down (&lock->semaphore);

  // Now we have acquired the lock.
  lock->holder = t_cur;
  list_insert_ordered(&t_cur->lockhold_list, &lock->elem, lock_priority_greater, NULL);
  t_cur->waiting_lock = NULL;
}

static bool
lock_priority_greater(const struct list_elem *lhs,
                           const struct list_elem* rhs, void *aux UNUSED) {
  return list_entry(lhs, struct lock, elem)->max_priority
       > list_entry(rhs, struct lock, elem)->max_priority;
}

相应的，当线程释放锁，也就是 lock_release 函数中，可能发生 donation 优先级的还原。还原后应该为 base_priority 和其它锁中 max_priority 最大的。由于 lockhold_list 的有序性，取头即可。

void
lock_release (struct lock *lock) 
{
  ASSERT (lock != NULL);
  ASSERT (lock_held_by_current_thread (lock));

  struct thread *t_cur = thread_current();
  
  list_remove(&lock->elem); // remove lock from thread's holding list

  t_cur->priority = t_cur->base_priority;
  if (!list_empty(&t_cur->lockhold_list)) {
    struct lock *lock_front = list_entry(list_front(&t_cur->lockhold_list), struct lock, elem);
    t_cur->priority = lock_front->max_priority > t_cur->priority ? lock_front->max_priority : t_cur->priority;
  }

  lock->holder = NULL;
  sema_up (&lock->semaphore);
}

另外需要注意，在调用 thread_set_priority 时如果设置较低的优先级，而此时存在较高的 donate 优先级，则设置只应该影响 base_priority 而暂时不影响 priority（直至锁被释放）：

void
thread_set_priority (int new_priority) 
{
  struct thread *t_cur = thread_current();
  int old_priority = t_cur->priority;
  t_cur->base_priority = new_priority;
  // If current thread has donated priority that is higher than new priority,
  // then priority shouldn't change
  if (list_empty(&t_cur->lockhold_list) || new_priority > t_cur->priority) {
    t_cur->priority = new_priority;
  }
  if (t_cur->priority < old_priority) {
    thread_yield();
  }
}

材料说捐献只需要考虑锁，但按优先级调度是要考虑 semaphore 和 condvar 的。先看信号量，一个信号量有多个 waiters，唤醒时应该选择优先级最大的。考虑到等待线程的优先级随时可能发生变化，这里 waiters 不做成有序链表，而是在取出时线性扫描一次（前面已经写过一个线程优先级比较函数，因为那个是 greater，所以应该结合 list_min）。于是 sema_down 无需修改，sema_up 注意唤醒后可能需要重新调度，最后要添加 thread_yield。

void
sema_up (struct semaphore *sema) 
{
  enum intr_level old_level;

  ASSERT (sema != NULL);

  old_level = intr_disable ();

  if (!list_empty(&sema->waiters)) {
    // since the compare function is "greater", we
    // should use list_min to find thread with highest priority
    struct list_elem *e = list_min(&sema->waiters, thread_priority_greater, NULL);
    struct thread *t = list_entry(e, struct thread, elem);
    list_remove(e);
    thread_unblock(t);
  }
  sema->value++;
  intr_set_level (old_level);
  thread_yield();
}

条件变量也是一样的道理，多写一个 semaphore_elem 的比较函数，然后修改 cond_signal 使其取优先级最大的 waiter 即可。

接下来考虑多级捐献，其实就是在 lock_acquire 中，除了向等待的锁的持有者捐献外，还要递归判断该持有者是否也被一个更低优先级线程持有的锁阻塞。当然这里并不是必须写成递归函数，用一个 while 循环即可，注意还要加一个递归的深度限制。实现如下（省略部分与之前相同）：

void
lock_acquire (struct lock *lock)
{
  ...
  if (lock->holder != NULL) {
    t_cur->waiting_lock = lock;
    
    lock->max_priority = t_cur->priority > lock->max_priority ? t_cur->priority : lock->max_priority;
    struct thread *t_holder = lock->holder;
    t_holder->priority = t_cur->priority > t_holder->priority ? t_cur->priority : t_holder->priority;

    size_t depth = 0;
    while (t_holder->waiting_lock != NULL && depth < PRI_DONATE_MAX_DEPTH) { // PRI_DONATE_MAX_DEPTH = 8
      struct thread *t_next = t_holder->waiting_lock->holder; // waiting lock must has a holder
      if (t_next->priority > t_holder->priority) {
        // no need to continue donation because their priority is higher
        break;
      }
      t_holder->waiting_lock->max_priority = t_holder->priority;
      t_next->priority = t_holder->priority;
      t_holder = t_next;
      depth++;
    }
  }
  ...
}

现在，除了 mlfqs 相关的所有其他测试样例应该均能通过。

Task3: Advanced Scheduler

Exercise 3.1

• Implement a multilevel feedback queue scheduler similar to the 4.4BSD scheduler to reduce the average response time for running jobs on your system. See section 4.4BSD Scheduler, for detailed requirements.

本节实现多级反馈队列（Multi-Level Feedback Queue, MLFQ）调度，由布尔值 thread_mlfqs 控制是否启用，具体规则在附录 4.4BSD Scheduler 中。

首先，MLFQ 中线程的优先级由调度器计算得到，thread_set_priority() 的调用应该被忽略。MLFQ 也不使用优先级捐献，lockhold_list，waiting_lock 等我们为 thread 类添加的成员都无需考虑。先在之前的代码中加入相关判断。

MLFQS 为了计算 priority 引入了 nice，recent_cpu 和 load_avg 的概念，后两者为实数。然而 Pintos 内核中不支持浮点数类型，所以文档中为我们介绍了一种用整型模拟浮点数的方法 fixed-point，将整型的一部分位拿来表示小数部分（位数的分配自行规定，文档中介绍的是 17.14 格式）。添加一个头文件 fixed_point.h，因为比较简单函数就直接在里面定义了。其中 fixed_point 之间的加减法、fixed_point 和 int 的乘除法无需特殊处理。可以用宏也可以用函数实现，如果是后者记得将其标记为 static 否则会导致函数重复定义错误。

#ifndef THREADS_FIXED_POINT_H
#define THREADS_FIXED_POINT_H

#include <stdint.h>

static int f = 16384; // f value for 17.14 fixed point
typedef int fixed_point;

static inline fixed_point itof(int n) {
  return n * f;
}

static inline int ftoi(fixed_point x) {
  return x / f;
}

static inline int ftoi_round(fixed_point x) {
  return x >= 0 ? (x + f / 2) / f : (x - f / 2) / f;
}

static inline fixed_point add_fi(fixed_point x, int n) {
  return x + n * f;
}

static inline fixed_point sub_fi(fixed_point x, int n) {
  return x - n * f;
}

static inline fixed_point mul_ff(fixed_point x, fixed_point y) {
  return ((int64_t)x) * y / f;
}

static inline fixed_point div_ff(fixed_point x, fixed_point y) {
  return ((int64_t)x) * f / y;
}

#endif /**< threads/fixed_point.h */

thread.c 中，添加 3 个函数分别计算 priority，recent_cpu 和 load_avg 的更新，其中前二者为每个线程所有的值，后者为系统中统一的值：

static void update_priority_mlfqs(struct thread *t, void *aux UNUSED)
{
  if (t != idle_thread) {
    t->priority = PRI_MAX - ftoi(t->recent_cpu / 4) - 2 * t->nice;
    t->priority = (t->priority > PRI_MAX) ? PRI_MAX : ((t->priority < PRI_MIN) ? PRI_MIN : t->priority);
  }
}
static void update_recent_cpu_mlfqs(struct thread *t, void *aux UNUSED)
{
  if (t != idle_thread) {
    t->recent_cpu = mul_ff(div_ff(2 * load_avg, 2 * load_avg + itof(1))
                           , t->recent_cpu)
                           + itof(t->nice);
  }
}
static void update_load_avg_mlfqs()
{
  // coefficient 59/60 in update formula
  fixed_point coef = div_ff(itof(59), itof(60));
  int ready_thread_cnt = list_size(&ready_list);
  if (thread_current() != idle_thread) {
    ready_thread_cnt++;
  }
  load_avg = mul_ff(coef, load_avg) + itof(ready_thread_cnt) / 60;
}

update_load_avg_mlfqs 中的 $\frac{59}{60}$ 系数实际是一个常量，然而因为 C 语言的性质，coef 无法被标为 static（C++ 中是可以的），导致每次调用都会重复计算。如果用宏则可以。

这些值的更新在 thread_tick 中进行，根据文档，priority 每 4 个 tick 更新一次，recent_cpu 和 load_avg 每秒（即 TIMER_FREQ 个 tick）更新一次。注意前面的更新函数是按照 thread_action_func 的格式写的，所以可以利用 thread_foreach 进行所有线程的更新。

void
thread_tick (void) 
{
  ...
  if (thread_mlfqs) {
    int64_t ticks = timer_ticks();
    if (t != idle_thread) {
      t->recent_cpu += itof(1);
    }
    if (ticks % 4 == 0) { // update priority on every fourth tick
      thread_foreach(update_priority_mlfqs, NULL);
    }
    if (ticks % TIMER_FREQ == 0) { // update recent_cpu and load_avg on every second
      update_load_avg_mlfqs();
      thread_foreach(update_recent_cpu_mlfqs, NULL);
    }
  }
  ...
}

到了这里恭喜你，你将收获到：

下个 Lab 见：）