1. ucore lab6介绍
ucore在lab5中实现了较为完整的进程/线程机制，能够创建和管理位于内核态或用户态的多个线程，让不同的线程通过上下文切换并发的执行，最大化利用CPU硬件资源。ucore在lab5中使用FIFO的形式进行线程调度，不同的线程按照先来先服务的策略，直到之前创建的线程完全执行完毕并退出，后续的线程才能获得执行机会。
FIFO的策略固然简单，但实际效果却非常差。在非抢占的FIFO调度策略中，如果之前的线程任务耗时很长，将导致后续的线程迟迟得不到执行机会而陷入饥饿；即使后续的线程是短任务、能很快的执行完，也会由于被迫等待前面长任务线程的执行而导致系统整体的任务吞吐量大幅下降。如果前面线程出现了bug陷入死循环，则整个系统将会被阻塞。
为此，计算机科学家提出了很多线程调度策略来解决这一问题。例如在批处理操作系统中除了FIFO先来先服务策略，还有短任务优先、最短剩余时间优先等多种非抢占式调度算法；而在交互式操作系统中又提出了时间片轮转调度、优先级调度、多级队列调度等基于抢占的调度算法(在视频公开课的原理篇以及《现代操作系统》中的调度一节对此都有着详细介绍)。
ucore在lab6实现了可以满足上述不同调度算法的的线程调度框架。lab6中采用了和之前内存调度算法框架类似的方式，通过函数指针集合，以面向对象的方式抽象出了一个线程调度器框架。并在参考答案中实现了一个基于线程优先级、时间片轮转的、抢占式的stride调度算法。
通过lab6的学习，使得原本枯燥乏味的关于各种调度算法的纯理论知识有了实践的机会，可以更深入的了解操作系统线程调度算法的工作机制。
lab6是建立在之前实验的基础之上的，需要先理解之前的实验内容才能顺利理解lab6的内容。
可以参考一下我关于前面实验的博客：
1. ucore操作系统学习(一) ucore lab1系统启动流程分析
2. ucore操作系统学习(二) ucore lab2物理内存管理分析
3. ucore操作系统学习(三) ucore lab3虚拟内存管理分析
4. ucore操作系统学习(四) ucore lab4内核线程管理
5. ucore操作系统学习(五) ucore lab5用户进程管理
2. ucore lab6实验细节分析
ucore在lab6中的改进大致可以分为几个部分：
1. 为了支持基于优先级的stride调度算法，改进线程控制块，加入了相关的新字段。
2. 以面向对象的方式实现了基本的线程调度器框架， 在对应的地方以接口的形式(函数指针)进行访问；不同的调度算法只需要实现对应的调度器框架接口即可简单的接入ucore。
3. 实现了stride线程调度算法。
2.1 lab6中线程控制块的变化
lab6中为了能够实现线程调度框架，需要在线程控制块中额外引入例如就绪队列、线程调度优先级、时间片等属性，用于兼容多种调度算法。
lab6中的proc_struct：
/** * 进程控制块结构（ucore进程和线程都使用proc_struct进行管理） * */ struct proc_struct { 。。。 只列出了lab6中新增加的属性项 // 包含该线程的就绪队列(多级多列调度时，系统中存在多个就绪队列) struct run_queue *rq; // running queue contains Process // 就绪队列节点 list_entry_t run_link; // the entry linked in run queue // 线程能够占有的CPU时间片 int time_slice; // time slice for occupying the CPU // lab6中支持stride算法的斜堆节点 skew_heap_entry_t lab6_run_pool; // FOR LAB6 ONLY: the entry in the run pool // lab6中支持stride算法的当前线程stride步长 uint32_t lab6_stride; // FOR LAB6 ONLY: the current stride of the process // 线程的特权级 uint32_t lab6_priority; // FOR LAB6 ONLY: the priority of process, set by lab6_set_priority(uint32_t) };
就绪队列介绍：
为了能够支持不同的线程调度算法，lab6中引入了就绪队列的概念。就绪队列（run_queue）是一个包含了所有就绪态线程集合的队列结构，能够在有就绪线程出现时令其高效的入队，当线程脱离就绪态时高效的将其从就绪队列中移除。
就绪队列是一个抽象的队列，其底层实现可以是双向链表，平衡二叉树或是堆等等。由于堆结构中的元素只需要满足堆序性，而不像平衡二叉树一样需要满足全局的有序性，因此其整体效率还要略高于平衡二叉树，很适合用来实现优先级队列。这也是lab6中引入斜堆skew_heap作为stride调度算法中就绪队列的底层实现的原因。
2.2 线程调度器框架介绍
前面提到过，ucore抽象出了一系列的调度器的行为，并通过函数指针以面向对象的形式提供服务。调度器本身比较简单，主要包括以下几部分(位于/kern/schedule/sched.[ch])：
1. 就绪队列的初始化、入队、出队操作(init、enqueue、dequeue)
2. 由特定的就绪算法处理器实现，从就绪队列中挑选下一个进行调度的线程(pick_next)
3. 当时钟中断发生时，令调度器感知并修改对应数据的逻辑(proc_tick)。例如在时间片轮转算法中每当发生时钟中断时，减少当前线程对应的时间片。
线程调度器框架：
// The introduction of scheduling classes is borrrowed from Linux, and makes the // core scheduler quite extensible. These classes (the scheduler modules) encapsulate // the scheduling policies. struct sched_class { // the name of sched_class // 调度类的名字 const char *name; // Init the run queue // 初始化就绪队列 void (*init)(struct run_queue *rq); // put the proc into runqueue, and this function must be called with rq_lock // 将一个线程加入就绪队列(调用此方法一定要使用队列锁调用(lab6中直接关中断来避免并发)) void (*enqueue)(struct run_queue *rq, struct proc_struct *proc); // get the proc out runqueue, and this function must be called with rq_lock // 将一个线程从就绪队列中移除(调用此方法一定要使用队列锁调用(lab6中直接关中断来避免并发)) void (*dequeue)(struct run_queue *rq, struct proc_struct *proc); // choose the next runnable task // 调度框架从就绪队列中选择出下一个可运行的线程 struct proc_struct *(*pick_next)(struct run_queue *rq); // dealer of the time-tick // 发生时钟中断时，调度器的处理逻辑 void (*proc_tick)(struct run_queue *rq, struct proc_struct *proc); };
2.3 线程调度器的工作原理
通过分析线程调度器具体在哪些地方被调用，可以更好的理解ucore中线程调度器的工作原理。
sched_class.init使用分析：
在总控函数kern_init中，调用了sched_init函数用于初始化和调度器相关的数据结构。其中设置当前系统对应的调度框架(sched_class = &default_sched_class)，并通过调度器框架的init函数进行了调度器的初始化。
/** * 初始化任务调度器 * */ void sched_init(void) { // 清空定时器队列 list_init(&timer_list); // 令当前的调度框架为default_sched_class(stride_schedule) sched_class = &default_sched_class; rq = &__rq; // 设置最大的时间片 rq->max_time_slice = MAX_TIME_SLICE; // 初始化全局就绪队列 sched_class->init(rq); cprintf("sched class: %s\n", sched_class->name); }
schedule线程调度函数分析：
相比于lab5，在lab6中由于引入了线程调度器，因此schedule函数的线程调度逻辑进行了一定的改动。
schedule函数被调用时，意味着需要当前线程让出CPU而令另一个就绪线程获得CPU。lab6实现了调度框架后，调度函数需要完成几个步骤：
1. 如果当前线程依然是就绪态，将其放入就绪队列(enqueue)，令其有机会再度被选中获得CPU(比如特权级很高，很可能下一次调度依然被选中)。
2. 由调度器根据特定的调度算法实现，选出下一个需要获取CPU的线程(pick_next)。
3. 将挑选出的线程从就绪队列中移除(dequeue)。(这么做的目的我认为是因为被选中调度的线程将要变成运行态了，语义上不适合继续放在就绪队列中，造成理解上的困难)
4. 被挑选的线程作为proc_run的参数，与当前线程进行上下文切换(与lab5中的逻辑一样)。
/** * 就绪线程进行CPU调度 * */ void schedule(void) { bool intr_flag; struct proc_struct *next; // 暂时关闭中断，避免被中断打断，引起并发问题 local_intr_save(intr_flag); { // 令current线程处于不需要调度的状态 current->need_resched = 0; if (current->state == PROC_RUNNABLE) { // 如果当前线程依然是就绪态，将其置入就绪队列(有机会再被调度算法选出来) sched_class_enqueue(current); } // 通过调度算法筛选出下一个需要被调度的线程 if ((next = sched_class_pick_next()) != NULL) { // 如果选出来了，将其从就绪队列中出队 sched_class_dequeue(next); } if (next == NULL) { // 没有找到任何一个就绪线程，则由idleproc获得CPU next = idleproc; } // 被选中进行调度执行的线程，被调度执行次数+1 next->runs ++; if (next != current) { // 如果被选出来的线程不是current当前正在执行的线程，进行线程上下文切换，令被选中的next线程获得CPU proc_run(next); } } local_intr_restore(intr_flag); } static inline void sched_class_enqueue(struct proc_struct *proc) { if (proc != idleproc) { // 如果不是idleproc，令proc线程加入就绪队列 sched_class->enqueue(rq, proc); } } static inline void sched_class_dequeue(struct proc_struct *proc) { // 将proc线程从就绪队列中移除 sched_class->dequeue(rq, proc); } static inline struct proc_struct * sched_class_pick_next(void) { // 有调度框架从就绪队列中挑选出下一个进行调度的线程 return sched_class->pick_next(rq); }
sched_class.proc_tick使用分析：
在lab6中，时钟中断的处理逻辑中主动调用了调度器的proc_tick函数，使得调度器能感知到时钟中断的产生并调整调度相关的数据结构。
例如在基于时间片轮转调度的抢占式线程调度算法中，当周期性的时钟中断发生时就减少当前线程所分配的的时间片。当发现为当前线程分配的时间片用完时，便强制进行一次线程调度，令其它的线程可以获得CPU，避免长时间的饥饿。
trap_dispatch时间中断处理逻辑：
static void trap_dispatch(struct trapframe *tf) { char c; int ret=0; switch (tf->tf_trapno) { 。。。 省略其它中断处理逻辑 case IRQ_OFFSET + IRQ_TIMER: ticks ++; assert(current != NULL); // 令调度框架得以监听到时钟中断，修改对应的调度状态 sched_class_proc_tick(current); break; 。。。 省略其它中断处理逻辑 ｝ ｝ void sched_class_proc_tick(struct proc_struct *proc) { if (proc != idleproc) { // 处理时钟中断，令调度框架更新对应的调度参数 sched_class->proc_tick(rq, proc); } else { // idleproc处理时钟中断，需要进行调度 proc->need_resched = 1; } }
2.4 stride调度算法实现
在了解了ucore的线程调度器是如何工作之后，下面分析线程调度器的实现。
线程调度器是一个抽象的接口框架，可以简单的接入不同的调度算法实现。在ucore lab6的参考代码示例中实现了名为stride的线程调度算法(大致工作原理在实验公开课视频中有提到，比较容易理解)。
stride调度算法就绪队列初始化：
/* * stride_init initializes the run-queue rq with correct assignment for * member variables, including: * * - run_list: should be a empty list after initialization. * - lab6_run_pool: NULL * - proc_num: 0 * - max_time_slice: no need here, the variable would be assigned by the caller. * * hint: see proj13.1/libs/list.h for routines of the list structures. */ static void stride_init(struct run_queue *rq) { /* LAB6: YOUR CODE */ // 初始化就绪队列 list_init(&(rq->run_list)); rq->lab6_run_pool = NULL; rq->proc_num = 0; } /** * 就绪队列 * */ struct run_queue { list_entry_t run_list; unsigned int proc_num; int max_time_slice; // For LAB6 ONLY 斜堆堆顶节点 skew_heap_entry_t *lab6_run_pool; };
stride调度算法入队、出队和挑选下一个就绪线程：
stride调度算法的入队、出队和挑选下一个就绪线程的逻辑比较类似，因此放到一起说明。
就绪队列作为一个抽象数据结构，底层可以由各种常用的数据结构实现，在lab6给出的代码实例中通过USE_SKEN_HEAP宏来决定就绪队列的底层实现。如果USE_SKEN_HEAP为真则使用斜堆，如果USE_SKEN_HEAP不为真则使用普通的双向链表来实现。
斜堆结构实现的就绪队列其入队、出队操作能达到O(logn)的对数复杂度，比其双向链表实现的就绪队列入队、出队效率O(n)要高出一个数量级(限于篇幅就不在这里展开关于斜堆的内容了)。
/* * stride_enqueue inserts the process ``proc'' into the run-queue * ``rq''. The procedure should verify/initialize the relevant members * of ``proc'', and then put the ``lab6_run_pool'' node into the * queue(since we use priority queue here). The procedure should also * update the meta date in ``rq'' structure. * * proc->time_slice denotes the time slices allocation for the * process, which should set to rq->max_time_slice. * * hint: see proj13.1/libs/skew_heap.h for routines of the priority * queue structures. */ static void stride_enqueue(struct run_queue *rq, struct proc_struct *proc) { /* LAB6: YOUR CODE */ #if USE_SKEW_HEAP // 使用斜堆实现就绪队列(lab6中默认USE_SKEW_HEAP为真) // 将proc插入就绪队列，并且更新就绪队列的头元素 rq->lab6_run_pool = skew_heap_insert(rq->lab6_run_pool, &(proc->lab6_run_pool), proc_stride_comp_f); #else // 不使用斜堆实现就绪队列，而是使用双向链表实现就绪队列 assert(list_empty(&(proc->run_link))); // 将proc插入就绪队列 list_add_before(&(rq->run_list), &(proc->run_link)); #endif if (proc->time_slice == 0 || proc->time_slice > rq->max_time_slice) { // 入队时，如果线程之前时间片被用完进行过调度则time_slice会为0，再次入队时需要重置时间片(或者时间片未正确设置，大于了就绪队列的max_time_slice) // 令其time_slice=rq->max_time_slice(最大分配的时间片) proc->time_slice = rq->max_time_slice; } // 令线程和就绪队列进行关联 proc->rq = rq; // 就绪队列中的就绪线程数加1 rq->proc_num ++; } /* * stride_dequeue removes the process ``proc'' from the run-queue * ``rq'', the operation would be finished by the skew_heap_remove * operations. Remember to update the ``rq'' structure. * * hint: see proj13.1/libs/skew_heap.h for routines of the priority * queue structures. */ static void stride_dequeue(struct run_queue *rq, struct proc_struct *proc) { /* LAB6: YOUR CODE */ #if USE_SKEW_HEAP // 使用斜堆实现就绪队列(lab6中默认USE_SKEW_HEAP为真) // 将proc移除出就绪队列，并且更新就绪队列的头元素 rq->lab6_run_pool = skew_heap_remove(rq->lab6_run_pool, &(proc->lab6_run_pool), proc_stride_comp_f); #else // 不使用斜堆实现就绪队列，而是使用双向链表实现就绪队列 assert(!list_empty(&(proc->run_link)) && proc->rq == rq); // 将proc移除出就绪队列，并且更新就绪队列的头元素 list_del_init(&(proc->run_link)); #endif // 移除完成之后，就绪队列所拥有的线程数减1 rq->proc_num --; } /* * stride_pick_next pick the element from the ``run-queue'', with the * minimum value of stride, and returns the corresponding process * pointer. The process pointer would be calculated by macro le2proc, * see proj13.1/kern/process/proc.h for definition. Return NULL if * there is no process in the queue. * * When one proc structure is selected, remember to update the stride * property of the proc. (stride += BIG_STRIDE / priority) * * hint: see proj13.1/libs/skew_heap.h for routines of the priority * queue structures. */ static struct proc_struct * stride_pick_next(struct run_queue *rq) { /* LAB6: YOUR CODE */ #if USE_SKEW_HEAP // 使用斜堆实现就绪队列(lab6中默认USE_SKEW_HEAP为真) if (rq->lab6_run_pool == NULL) return NULL; // 就绪队列为空代表没找到，返回null // 获取就绪队列的头结点，转换为所关联的线程返回 struct proc_struct *p = le2proc(rq->lab6_run_pool, lab6_run_pool); #else // 不使用斜堆实现就绪队列，而是使用双向链表实现就绪队列 // 获取双向链表的头结点 list_entry_t *le = list_next(&(rq->run_list)); if (le == &rq->run_list) // 双向链表为空代表没找到，返回null return NULL; struct proc_struct *p = le2proc(le, run_link); le = list_next(le); // 遍历整个双向链表，找到p->lab6_stride最小的那一个（p） while (le != &rq->run_list) { struct proc_struct *q = le2proc(le, run_link); if ((int32_t)(p->lab6_stride - q->lab6_stride) > 0){ // 如果线程q的lab6_stride小于当前lab6_stride最小的线程p // 令p=q，即q成为当前找到的lab6_stride最小的那一个线程 p = q; } // 指向双向链表的下一个节点，进行遍历 le = list_next(le); } #endif // 最终找到的线程指针p指向的是lab6_stride最小的那一个线程，即按照stride调度算法被选中的那一个线程 if (p->lab6_priority == 0){ // 特权级为0比较特殊代表最低权限，一次的步进为BIG_STRIDE p->lab6_stride += BIG_STRIDE; }else{ // 否则一次的步进为BIG_STRIDE / p->lab6_priority // 即lab6_priority(正整数)越大，特权级越高，一次步进的就越小 // 更容易被stride调度算法选中，相对而言被执行的次数也就越多，因此满足了线程特权级越高，被调度越频繁的需求 p->lab6_stride += BIG_STRIDE / p->lab6_priority; } return p; }
stride调度算法时钟中断处理：
stride调度算法是抢占式的。在stride_enqueue中，每当就绪队列入队时都会为其分配一定的时间片，当线程运行的过程中发生时钟中断时则会通过stride_proc_tick函数扣减对应的时间片。当为线程分配的时间片扣减为0时，则会将线程的need_resched设置为1。
在trap中断处理函数中，当对应中断号的处理例程返回时会单独的检查need_resched的值，当发现为1时，则会触发schedule函数进行一次强制的线程调度，从而令当前时间片扣减为0的线程得以让出CPU，使其它的就绪线程能得到执行的机会。这也是stride调度算法被称为抢占式调度算法的原因：无论当前执行的线程是否主动的让出cpu，在分配的时间片用完之后，操作系统将会强制的撤下当前线程，进行一次调度。
/* * stride_proc_tick works with the tick event of current process. You * should check whether the time slices for current process is * exhausted and update the proc struct ``proc''. proc->time_slice * denotes the time slices left for current * process. proc->need_resched is the flag variable for process * switching. */ static void stride_proc_tick(struct run_queue *rq, struct proc_struct *proc) { /* LAB6: YOUR CODE */ if (proc->time_slice > 0) { // 如果线程所分配的时间片还没用完(time_slice大于0)，则将所拥有的的时间片减1 proc->time_slice --; } if (proc->time_slice == 0) { // 当时间片减为0时，说明为当前线程分配的时间片已经用完，需要重新进行一次线程调度 proc->need_resched = 1; } }
trap函数(中断处理)：
/* * * trap - handles or dispatches an exception/interrupt. if and when trap() returns, * the code in kern/trap/trapentry.S restores the old CPU state saved in the * trapframe and then uses the iret instruction to return from the exception. * */ void trap(struct trapframe *tf) { // dispatch based on what type of trap occurred // used for previous projects if (current == NULL) { trap_dispatch(tf); } else { // keep a trapframe chain in stack struct trapframe *otf = current->tf; current->tf = tf; bool in_kernel = trap_in_kernel(tf); trap_dispatch(tf); current->tf = otf; if (!in_kernel) { if (current->flags & PF_EXITING) { // 如果当前线程被杀了(do_kill),将自己退出（被唤醒之后发现自己已经被判了死刑，自我了断） do_exit(-E_KILLED); } if (current->need_resched) { // 可能执行了阻塞的系统调用等情况，need_resched为真，进行线程调度切换 schedule(); } } } }
3. 总结
lab6是一个承上启下的实验，ucore在lab4、lab5中实现了线程机制，而lab6在线程调度切换上做了拓展。正是因为抢占式线程调度机制的出现，使得线程可能在执行时的任意时刻被打断。并发的线程执行流在提高cpu利用率的同时也带来了线程安全的问题，也引出了后续lab7中将要介绍的线程同步概念。
通过lab6实验的学习，可以更深刻的理解操作系统中线程调度的工作原理，也有机会在ucore中亲手实现操作系统书籍上介绍的各种调度算法，使得其不再是抽象的纯理论知识，理解时印象会更深刻。
这篇博客的完整代码注释在我的github上：https://github.com/1399852153/ucore_os_lab (fork自官方仓库)中的lab6_answer。
希望我的博客能帮助到对操作系统、ucore os感兴趣的人。存在许多不足之处，还请多多指教。