实时操作系统的任务调度示例之优先级反转

1 什么是优先级反转？
目前市面流行的实时操作系统都是采用可抢占式的基于优先级的调度方式，其保证了处于就绪状态的优先级高的任务可以先于优先级低的任务而执行。但这并不是绝对的，优先级反转是实时系统中的一个经典特例。其大体流程如下：
假设系统中有3个task，优先级分别为高、中、低，以task_high,task_middle,task_low代称
1) 一开始，task_high,task_middle处于挂起状态，task_low在运行
2) task_low要访问共享资源，获取了一个信号量semaphore
3) 这之后task_high等待的事件到来，task_high抢占了task_low开始运行，在运行过程中，也需要访问同一共享资源，所以也要去拿semaphore，由于它已经被task_low取得，所以task_high再次被挂起。
4) task_low继续运行

5) task_middle等待的事件这时候到来，它从task_low手里抢来了CPU，开始运行
6) task_middle运行结束或再次挂起，让出CPU
7) task_low继续运行，完成共享资源的访问，释放semaphore
8) task_high获得semaphore，得以继续运行。

task_middle的优先级是低于task_high的，原则上应该是task_high优先获得CPU得到执行。但是在这个过程中，由于共享资源的竞争访问，导致了高优先级的任务并不能被优先调度
执行的顺序看起来好像是高优先级任务task_high和中优先级任务task_middle，两者之间的优先级调换了，反转了一样。

2 优先级反转会有什么危害？

乍一看上面的这个运行流程，很多同学可能觉得，不就是有个task本来想让它早点执行，结果因为一个意外，晚一点点执行了吗，能有什么后果？
是的，如果上面案例中task_middle占用的时间较短，可能问题还不大，但是如果运气不好，task_middle执行的是一个耗时的操作，或者task_middle不是一个任务，而是一堆优先级处于task_low和task_high之间，正好在这期间需要运行呢？那那task_low始终无法释放信号量，task_high也要被delay很久才能获得CPU。
嵌入式实时操作系统，最重要的指标就是：确保重要的任务执行时间是可预测的，有一个不能容忍的deadline，要确保任何时刻都不能超过某个时间。考虑一下汽车的电子系统里控制受撞击后弹出安全气囊的task，如果执行时间被delay会有什么后果。
在有些场景下，会导致整个系统崩溃，有一个高大上的典型案例，美国的火星探路者号在一次执行任务时，就是因为重要任务被有delay,结果导致了系统复位、数据丢失。有兴趣的朋友可以google一下 What
Happened on Mars?

3 优先级反转的示例
本节参考第一节的描述设计了一个简单的实验，来演示优先级反转的运行过程，使用的rtos是Nucleus。
应用创建了3个task，其优先级分别为149,150,151，对应的task名分别为task_high，task_middle和task_low
创建2个消息队列，让task_high和task_middle可以阻塞在上面并可在适当时间将它们唤醒
还有一个信号量，模拟task_high和task_low访问共享资源时的互斥

NU_Create_Queue(&sync_queue1,"queue1",
queue_Buf1, QUEUE_SIZE, NU_FIXED_SIZE, 1, NU_FIFO);

NU_Create_Queue(&sync_queue2,"queue2", queue_Buf2,
QUEUE_SIZE, NU_FIXED_SIZE, 1, NU_FIFO);

NU_Create_Semaphore(&sync_sema,"sync_sema",
1, NU_PRIORITY);

NU_Create_Task(&highTask, ""，task_high,
0, NU_NULL, Task_Stack1, Stack_Size, 149, 0,NU_PREEMPT, NU_START);

NU_Create_Task(&middleTask, "" ,task_middle, 0, NU_NULL, Task_Stack2,Stack_Size, 150, 0,NU_PREEMPT, NU_START);

NU_Create_Task(&lowTask, ""，task_low, 0, NU_NULL, Task_Stack3, Stack_Size, 151, 0,NU_PREEMPT, NU_START);

static int a = 100,b = 50,tmp;
void swap_tmp() //纯粹是为了演示延迟的函数，因为如果就写for循环空转可能会被编译器优化掉
{
tmp = a; a = b; b = tmp;
}
void task_high(UNSIGNED argc, VOID* argv)
{
u32 msg,msg_len;
printf("task_high start: priority %d",((TC_TCB*)TCT_Current_Thread())->tc_priority);
NU_Receive_From_Queue(&sync_queue1,&msg,1,&msg_len,NU_SUSPEND);
printf("task_high. obtain semaphore: %d",tc_priority);
NU_Obtain_Semaphore(&sync_sema,NU_SUSPEND);

for (i=0;i<0xffffff;i++)
swap_tmp();
printf("task_high. do works");

}

void task_middle(UNSIGNED argc, VOID* argv)
{
u32 msg,msg_len;
u32 i;
printf("task_middle start: priority %d",((TC_TCB*)TCT_Current_Thread())->tc_priority);
NU_Receive_From_Queue(&sync_queue2,&msg,1,&msg_len,NU_SUSPEND);

for (i=0;i<0xffffff;i++)
swap_tmp();
printf("task_middle. do works");

}

void task_low(UNSIGNED argc, VOID* argv)
{
int cnt;
printf("task_low start: priority %d,obtain semaphone",((TC_TCB*)TCT_Current_Thread())->tc_priority);
NU_Obtain_Semaphore(&sync_sema,NU_SUSPEND);
printf("task_low. do 1-round works");

for (i=0;i<0xffffff;i++)
swap_tmp();
printf("task_low. send message to wakeup task_high");
NU_Send_To_Queue(&sync_queue1,&cnt,1,NU_SUSPEND);
printf("task_low. do 2-round works");

for (i=0;i<0xffffff;i++)
swap_tmp();
printf("task_low. send message to wakeup task_middle");
printf("task_low. do 3-round works");

for (i=0;i<0xffffff;i++)
swap_tmp();
printf("task_low. release semaphore");
NU_Release_Semaphore(&sync_sema);
}

输出的log结果如下所示，很显然task_high的do works比task_middle的do works要晚得到执行：

[0:0:16:373] task_high start: priority 149
[0:0:16:383] task_middle start: priority 150
[0:0:16:388] task_low start: priority 151,obtain semaphore
[0:0:17:033] task_low. do 1-round works
[0:0:18:323] task_low. send message to wakeup task_high
[0:0:18:323] task_high. obtain semaphore: 149
[0:0:18:968] task_low. do 2-round works
[0:0:20:257] task_low. send message to wakeup task_middle
[0:0:20:902] task_middle. do works
[0:0:22:837] task_low. do 3-round things
[0:0:24:127] task_low. release semaphore
[0:0:24:773] task_high. do works

4 怎么避免优先级反转？

楼主手上有3个rtos的运行环境:FreeRtos,ucosii,Nucleus，它们都没有内置避免优先级反转的方法。大名鼎鼎的VxWorks据称是天然支持，可惜它是不开源的。楼主在Nucleus里做了一个简单的workround，以优先级继承的方式避免了发生优先级反转。（对“优先级继承”不了解的同学请自行百度）
在信号量的结构体SM_SCB里增加一个字段TC_TCB* sm_own_task。用来保存当前拥有信号量的task指针
NU_Obtain_Semaphore的函数实现里，如果能成功获得信号的分支，加上一句

semaphore->sm_own_task = (TC_TCB*)TCT_Current_Thread();
task_high函数修改如下,在获取信号量之前，将task_low的优先级暂时提升至与task_high一样，获得信号量之后再将其恢复。

void task_high(UNSIGNED argc, VOID* argv)
{
u32 msg,msg_len;
TC_TCB* ptcb = NULL;
DATA_ELEMENT old_priority;

printf("task_high start: priority %d",((TC_TCB*)TCT_Current_Thread())->tc_priority);
NU_Receive_From_Queue(&sync_queue1,&msg,1,&msg_len,NU_SUSPEND);
printf("task_high. obtain semaphore:");
if (sync_sema.sm_own_task->tc_priority > ((TC_TCB*)TCT_Current_Thread())->tc_priority)
{
printf("task_high.Improve temporary priority");
ptcb = sync_sema.sm_own_task;
old_priority = ptcb ->tc_priority;
TCS_Change_Priority(ptcb,((TC_TCB*)TCT_Current_Thread())->tc_priority);
}
NU_Obtain_Semaphore(&sync_sema,NU_SUSPEND);
if (ptcb)
{
printf("task_high.restore temporary priority");

TCS_Change_Priority(ptcb,old_priority);
}

for (i=0;i<0xffffff;i++)
swap_tmp();
printf("task_high. do works");

}

运行起来输出的log如下：

[0:0:16:389] task_high start: priority149
[0:0:16:399] task_middle start: priority 150
[0:0:16:405] task_low start: priority 151,obtain semaphone
[0:0:17:050] task_low. do 1-round works
[0:0:18:339] task_low. send message to wakeup task_high
[0:0:18:339] task_high. obtain semaphone: 149
[0:0:18:339] task_high. Improve temporary priority //临时提升了task_low的优先级
[0:0:18:984] task_low. do 2-round works
[0:0:20:274] task_low. send message to wakeup task_middle
[0:0:20:918] task_low. do 3-round things
[0:0:22:208] task_low. release semaphore
[0:0:22:208] task_high. restore temporary priority //task_low的优先级恢复
[0:0:22:853] task_high. do works 1-times
[0:0:26:080] task_middle. do 3-round works //task_middle的执行顺序排在了task_high之后