【RT-Thread内核详解系列】基于优先级的全抢占式调度算法...
本帖最后由 sunsili 于 2022-2-13 21:09 编辑【RT-Thread内核详解系列】基于优先级的全抢占式调度算法的实现
The unexamined life is not worth living.
未经审视的人生是不值得过的。
-- 苏格拉底
一、原理概述
RT-Thread 是一款嵌入式实时操作系统（RTOS），同时也是一款优秀的物联网操作系统，相对于裸机的轮询调度算法，它使用的线程（任务）调度算法是基于优先级的全抢占式多线程调度算法，该算法大大增强了系统的实时响应，大大扩展了系统的应用场景。
该调度算法在每次调度任务时，总会选择优先级最高的就绪任务执行，保证优先级高的任务得到最及时的响应。
下面，我们来详细讲解该调度算法的原理和具体实现。
注：本文基于 RT-Thread 的版本为 RT-Thread v4.0.5 released， 其发布于 2021-12-31。
首先，RT-Thread 中定义了一个双向链表数组 rt_thread_priority_table，用来挂载就绪的线程，代码如下所示：
/* Double List structure*/
struct rt_list_node{
struct rt_list_node *next;
/**< point to next node. */
struct rt_list_node *prev;
/**< point to prev node. */
};
typedef struct rt_list_node rt_list_t;
/**< Type for lists. */
rt_list_t rt_thread_priority_table; //线程优先级双向链表数组
其实实现的就是如下的结构：
其中 RT_THREAD_PRIORITY_MAX 代表的是该系统配置的优先级数目，RT-Thread 的调度算法只支持 RT_THREAD_PRIORITY_MAX <= 256，这个数目绝对够绝大多数的项目需求了。而 rt_thread_priority_table 这个数组从 0 到 RT_THREAD_PRIORITY_MAX-1的数组元素则是分别对应任务的优先级从 0 到 RT_THREAD_PRIORITY_MAX-1。
当有线程就绪时，线程会挂载到其优先级对应的线程优先级表(rt_thread_priority_table)元素的位置，假设现在有两个线程，分别是优先级为 0 的 thread1 和优先级为 1 的 thread2，当它们都进入就绪状态等待调度时，效果如下：
此时问题来了，当任务们都进入就绪状态，挂载在这个线程优先级表(rt_thread_priority_table)上时，我们如何查找到所有就绪任务中拥有最高优先级的那个呢？
首先，我们想到可以从线程优先级表(rt_thread_priority_table)的位置 0 开始查找，一旦发现哪个位置有挂载线程，则说明该优先级下有对应的任务就绪了，执行该线程即可。当下一次调度任务时，又按刚才的步骤从 0 开始查找即可，这不失为一个能解决问题的方法。
但是细细想来，如果就绪任务中的最高优先级比较靠后，这样的查找算法几乎要遍历一遍，明显很耗时间，一点都配不上实时操作系统（RTOS）中 “实时”这两个字。
为了更高效地调度任务，RT-Thread 使用了线程就绪表（rt_thread_ready_table）映射来实现的，简单的说，就是维护一个记录已就绪任务的表，通过查找这个表，我们就能够知道当前就绪任务中最高优先级的任务是哪一个，然后直接执行该优先级下的任务即可。
具体来说，这个线程就绪表（rt_thread_ready_table）的位数和优先级是一一对应的，当这个表的某一位为1时，代表这个优先级下有任务就绪。
RT-Thread 作为多任务调度系统，各个任务由于外部条件的触发，要经常在就绪状态和非就绪状态中切换（对应到数据结构就是：任务的双向指针插入或者脱离 rt_thread_priority_table ），所以作为映射的线程就绪表（rt_thread_ready_table）也要根据任务们的状态随时更新，对应到这个表，无非就是当某个优先级下没有任务时，将这个表对应的位清0，当有任务挂载在该优先级下时，将这个就绪表的位置1。
我们通过这个表，能够在 O(1) 的时间复杂度上找到最高优先级的任务，实现了快速地调度最高优先级任务。在切换任务的状态时，实时更新这张表的内容，确保线程就绪表（rt_thread_ready_table）和线程优先级表(rt_thread_priority_table)相互对应的结果是正确的。
接下来到了 show the code 时间了，可能会耗点脑细胞了。
二、实现细节
先来看看线程就绪表（rt_thread_ready_table）的代码实现形式：
/* 线程优先级就绪组，当RT_THREAD_PRIORITY_MAX<=32时，每一位直接对应32个优先级，否则分别映射到rt_thread_ready_table）*/
rt_uint32_t rt_thread_ready_priority_group;
#if RT_THREAD_PRIORITY_MAX > 32
/* Maximum priority level, 256 */
rt_uint8_t rt_thread_ready_table;
//对应 32*8(位) = 256个优先级
#endif
根据 RT_THREAD_PRIORITY_MAX这个宏定义的不同，线程就绪表（rt_thread_ready_table）的实现方式有两种：
[*]当优先级数量小于等于 32 时，定义一个 32 位的变量 rt_thread_ready_priority_group 即可实现线程就绪表，实现起来最简单，省空间、省时间（查找效率会提高）。所以如果不需要太多的优先级，建议优先级数量设置到小于等于 32。
[*]对应于RT_THREAD_PRIORITY_MAX > 32的情况，你可以想象目的是想要一个数据长度为 256 位的变量，奈何目前跑 RTOS 的MCU大多是 32 位的，所以这里只能采取这种折中的方式，相当于是双重映射实现出了那个梦寐以求的数据长度为 256 位的变量。
在谈到具体如何使用、更新线程就绪表（rt_thread_ready_table）之前，看看具体某个优先级任务的结构体有哪些是和调度算法相关的。如下是任务线程的结构体的定义，我将无关的变量删除，只留下所有相关的变量：
/** * Thread structure */
struct rt_thread{
.......
rt_list_t tlist;
/**< the thread list 用于挂载在线程优先级表中*/
/* priority */
rt_uint8_tcurrent_priority;
/**< current priority 当前优先级*/
rt_uint8_tinit_priority;
/**< initialized priority 初始优先级*/
#ifRT_THREAD_PRIORITY_MAX > 32
rt_uint8_tnumber;
//标记为是rt_thread_ready_table中的哪个元素
rt_uint8_thigh_mask;//具体元素对应的掩码,用于加快操作速度
#endif
rt_uint32_t number_mask; //rt_thread_ready_priority_group对应的掩码，用于加快操作速度
.......
};
typedef struct rt_thread *rt_thread_t;
简单提及一个这里的掩码的作用，目的是为了方便后面对线程优先级就绪表进行操作。举个例子，假设有一个优先级(priority)为 20 的任务，那么其对应的线程结构的掩码情况如下：
对于 RT_THREAD_PRIORITY_MAX <= 32：
number_mask = 1<<20(priority);即number_mask：000100000000000000000000，就是让 number_mask 的第 19 位为1。
话说回来，为什么要设置 number_mask 这个变量呢？遇到要使用这个变量的时候直接使用 1<<priority 代替 number_mask 不就可以了吗？其实也不是不可以，只是这样使用的话，需要多进行一次位移操作，这其实也是一种用空间换时间的策略，付出一个变量的代价获取更快的运算速度，增加了实时系统的调度任务的速度，也算是有利于系统的实时性。
对于 RT_THREAD_PRIORITY_MAX > 32，原理是一样的，就不赘述了。
RT-Thread是如何初始化上述的变量的呢？
在线程的初始化函数中（无论是动态创建线程还是静态初始化线程），根据传入的优先级参数进行操作，主要是用作初始化：
rt_err_t _thread_init(...,priority)
{
......
/* priority init */
RT_ASSERT(priority < RT_THREAD_PRIORITY_MAX);
thread->init_priority = priority;
thread->current_priority = priority;
thread->number_mask = 0;
#if RT_THREAD_PRIORITY_MAX > 32
thread->number = 0;
thread->high_mask = 0;
#endif
......
}
当启动线程时，更新线程对应的数据：
rt_err_t rt_thread_startup(rt_thread_t thread)
{
......
/* calculate priority attribute */
#if RT_THREAD_PRIORITY_MAX > 32
thread->number = thread->current_priority >> 3; /* 相当于除以8，得到对应所处的是哪个数组 */
thread->number_mask = 1L << thread->number; /* 得到对应所处的数组位置的掩码 */
thread->high_mask = 1L << (thread->current_priority & 0x07);/* 具体到对应的数组元素的掩码 */
#else
thread->number_mask = 1L << thread->current_priority;
/* 掩码 *
#endif
......
}
一切准备就绪，接下来看看当调度器调度任务时是怎么操作优先级就绪表的。
2.1、如何查找当前就绪任务中的最高优先级
其实就是查找优先级就绪表最靠近前面的1所处的位置，每次进行任务调度的时候会用到这个表去获得最高优先级，然后根据最高优先级得到此时就绪的最高优先级线程。
//该函数用于获取当前已就绪的最高优先级任务
static struct rt_thread* _scheduler_get_highest_priority_thread(rt_ubase_t *highest_prio)
{
......
#if RT_THREAD_PRIORITY_MAX > 32
register rt_ubase_t number; //得到是数组rt_thread_ready_table的哪个元素
number = __rt_ffs(rt_thread_ready_priority_group) - 1; //右移3位则是*8的意思，得到该数组的基地址，最后加上在该组的位置，就可以得到真正的优先级
highest_ready_priority = (number << 3) + __rt_ffs(rt_thread_ready_table) - 1;
#else
//直接得到具体的优先级
highest_ready_priority = __rt_ffs(rt_thread_ready_priority_group) - 1;
#endif
/* RT_THREAD_PRIORITY_MAX > 32 */
......
}
重点说一下这里的 __rt_ffs()函数，该函数实现的功能：得到传入的 32 位变量中 1 所在的最低位位置，也就是所谓的实现查找字节最低非0位。该函数能够在 O(1)的时间复杂度下实现查找到最低优先级，其使用的是位图查找算法，还是挺精妙的，由于篇幅有限，请移步到这篇文章：《RT-Thread的位图调度算法分析（最新版）》。
2.2、当任务就绪时，如何更新线程就绪列表？
如果某个任务就绪，根据该任务的优先级更新线程就绪表rt_thread_ready_table对应位置。
//该函数用于将新的任务加入就绪状态
void rt_schedule_insert_thread(struct rt_thread *thread)
{
......
#if
RT_THREAD_PRIORITY_MAX > 32 rt_thread_ready_table |= thread->high_mask; //让对应的优先级所在分组的具体位置1，表明就绪
#endif
rt_thread_ready_priority_group |= thread->number_mask; //让对应的优先级所在分组位置置1
......
}
2.3、当任务脱离就绪状态时，如何更新线程就绪列表？
当某个任务脱离就绪状态时（被删除、挂起等），我们也需要更新更新线程就绪表rt_thread_ready_table，由于RT-Thread支持多个任务可以拥有相同的优先级，所以我们在更新线程就绪表rt_thread_ready_table需要考虑该优先级下是否有别的任务还处于就绪状态。
//该函数用于将某个任务从就绪状态中移除
void rt_schedule_remove_thread(struct rt_thread *thread)
{
......
if (rt_list_isempty(&(rt_thread_priority_table)))//说明该优先级下没有别的就绪任务了
{
#if RT_THREAD_PRIORITY_MAX > 32
rt_thread_ready_table &= ~thread->high_mask;//让对应的优先级所在分组的具体位 置0，表明脱离就绪状态 if (rt_thread_ready_table == 0) //当这个数组元素的每个bit都是0，才能够置0，因为可能所处的组还有别的任务就绪
{
rt_thread_ready_priority_group &= ~thread->number_mask;//将该表的对应位置清0
}
#else
rt_thread_ready_priority_group &= ~thread->number_mask; //将该表的对应位置清0
#endif /* RT_THREAD_PRIORITY_MAX > 32 */
}
.....
}
三、一些有用的建议
[*]由于代码实现的原因，RT_THREAD_PRIORITY_MAX必须配置为小于等于 256，否则系统工作会异常。
[*]除了空闲任务，每一个优先级任务都应该有让出CPU使用权的操作（暂时脱离就绪状态，比如使用延时），让优先级更低的线程能够有机会执行。
[*]如果使用的芯片的资源有限（RAM、ROM等），建议尽量将 RT_THREAD_PRIORITY_MAX 设置小一点，最好是小于等于 32，不仅省资源，还能加快调度速度。
四、参考资料
[*]《RT-Thread的位图调度算法分析（最新版）》
[*]《RT-Thread文档中心 -- 线程管理》
页:
[1]