一条普遍存在于嵌入式实时系统(RTOS)中的嵌套使用互斥量的bug分析

caozilong LV 5

这篇文章尝试阐明一条普遍存在于嵌入式实时系统(RTOS)中的bug,　包括ucosii, rt-thread, nuttx在内的大多数RTOS系统中都存在本文要说的问题．虽说是一条bug，但是在实际的项目应用中，只有很小概率会造成严重的后果，因为只有在比较苛刻的几个条件同时满足的时候，它才会表现的比较致命.但伟人墨菲曾经曰过 “如果觉得事件有可能发生,那它就一定会发生”，所以呢，为了保证有不幸遇到这个问题的同学能跨过这个坎儿，或者让大家装X的时候有货，我们就来分析一下这个问题。

首先从优先级反转讲起．

优先级反转

这个世界中，有些东西只属于某一个人，或者说在某一个时间段只能属于某个人，映射到多任务系统中，不同任务之间存在共享资源，操作系统一般会提供mutex等同步机制来保证数据同步．有时候低优先级的任务已经持有了某个共享资源，因此，如果一个高优先级的任务想要访问该共享资源，需要等待低优先级的任务释放该资源，这种高优先级等待低优先级完成后才可以执行的现象被称为优先级翻转(Priority Inversion).下图是一个发生优先级反转的例子，系统一开始任务 C 执行 P(S) 申请访问共享资源 S 并获得 S 的占有权，然后高优先级任务 A 被创建并调度执行，此时任务 A 也需要访问资源 S，但是因为 C 已经占有了资源 S，所以 A 被迫等待资源 S 直到任务 C执行 V(S) 释放资源 S。除非能保证所有任务在访问共享资源时，共享资源都未被其他低优先级的任务占有，否则为了保证数据同步，优先级反转的现象很难避免。一般能够接受系统出现下图中所示的情况，因为任务对共享资源的访问一般都会在较短时间结束，下图中任务 A 虽然在等待一个低优先级的任务 C，但等待的时间是有限的。因为我们基于这样一个前提，所有持有锁的任务对系统来说是有责任的，它必须保证创建的临界区最小，资源使用完毕，立即释放,如果线程没有意识到这种责任，比如在持有阻塞高优先级资源的锁临界区内去睡眠，这就是设计者的责任了．另外关于这个责任的问题，值得一讲的是，互斥量(mutex)和信号量(semaphore)的责任是不一样的，mutex具有属主属性，所以对它的要求更高，拥有mutex的任务更倾向于在尽可能短的时间内释放锁。而semaphore由于没有属主特性，释放锁的时限要求没有那么高。

下图给出了一个更糟糕的优先级反转的例子，假设任务优先级 A>B>C，一开始任务 C 执行 P(S) 申请访问共享资源 S 并获得 S 的占有权；然后任务 A 被创建并调度执行，此时任务 A 也需要访问资源 S，但是因为 C 已经占有了资源S，所以 A 被迫等待资源 S 直到任务 C 执行 V(S) 释放资源 S；但是 C 在执行时，任务 B 被创建并调度执行，此时任务 A 不仅仅需要等待任务 C 还需要等待任务 B，但任务 B 不在临界区中，所以执行时间是不确定的。这种情况对于实时系统来说是致命的，因为高优先级的任务 A 可能永远无法被调度执行，这种高优先级任务可能无限期等待低优先级任务执行的现象被称为无边界优先级反转（unbounded-priority-inversion）。

上面第一幅图是有边界的优先级反转，下面一幅图是无边界优先级反转．

当面对更多任务的场景，优先级反转情况会更复杂，更难于辨别，下图中任意一处的task mid 任务都可以反转系统的优先级:

有些同学可能觉得，只要保证任务B能够自觉的每隔一段时间释放处理器（对应任务中周期性的调用sleep)就可以避免无边界的优先级反转，或者让B也拿一个锁，保证它可以在一段时间内释放处理机，但这无疑又让高优先级的任务A多了一层时序依赖，从一开始的依赖C ,到后面又增加了对B的时序依赖。最重要的是，后者不满足硬实时的截止时间要求. 由于C的临界区指令数目一定是固定的，所以执行时间必然固定,所以P对C的时序依赖是可控的，不会超过某个固定值。但B的横插一杠就不同了，首先是给执行时间引入不确定性，其次是，如果放开限制，如果系统存在中级优先级任务B，那是否还有可能存在另一个中等优先级任务序列B1，B2，...，来使依赖链增长？最终造成总的截至时间不确定。

或许还有同学要问，如果优先级最高的任务不止有一个，该怎么办？其实这个问题本身就已经给出了答案，既然是 “最” 高，就一定只有一个，多于一个就不叫最了，哪怕几个任务都很致命，你也要选出一个来作为肯綮，去解决与其它任务之间的逻辑依赖,具体操作就和应用场景有关了。

如果说你非要有两个“最”高优先级的任务需要处理怎么办？它们是如此紧急以至于哪怕有一个处理不及时你就要完蛋。那对不起了，真不行，这个世界上，不是所有问题都有答案的不是么？在这种情况下，单核的CPU救不了你，你需要多核SMP或者AMP了，有几个“最”，给几个核，单核只能有一个"最“存在，在单核上，还要啥自行车！

而且，选择“最”重要的任务，本身就带有主观色彩. 重要与不重要，是相对于具体的场景来说的，和平年代，一点皮外伤都要去医院包扎一下，还要打破伤风，防感染。但如果换做在战场上，可能缺胳膊断腿都不算什么了，只有保住脑袋才算是最高优先级的任务，所以这就是优先级的相对性，它是人定义的，三观不同，定义也不同,选择面前，有的人舍生取义，有的人卖国求荣，并没有人逼迫他们.

不要以为这种case很难遇到，实际上，现代很多消费电子中都存在这种场景，音频视频的偶然卡顿，某些事件响应的偶然不及时，可能都是后台发生了优先级反转，只是这种事故即便对最苛刻的用户来讲，也是无关痛痒（你可能会因为视频卡顿而错过某粒进球，但是绝对不会为此而跳楼)，所以根本就不会去重视。但是换一种场景就截然不同了。最著名的优先级反转事件，当属美国的好奇号火星车（用的还是大名鼎鼎的VxWorks)，大家可以网上搜一下，简直可以作为无边界优先级反转的经典案例写入教科书。下图就是火星车上发生的情况，它只不过是把上面的图用其它工具重绘了一遍。

避免优先级反转的两种常见方法是：优先级继承协议(Priority Inheritance Protocol, PIP)和优先级上限协议(Priority Ceiling Protocol, PCP)，其中优先级上限协议又被称为优先级天花板协议。两种方法的前提都是优先级可改变，而且是可以动态改变的。

优先级上限协议：
为每个mutex设置一个“优先级顶”，“优先级顶”定义为要调用该mutex的所有任务中最高优先级, 一个任务要想对共享资源操作开始一个新的临界区时，它的优先级必须严格高于当前其他试图获取该mutex的任务的“优先级顶”，否则它将被具有中等优先级的任务所阻塞。每个信号量的“优先级顶”是在任务集执行之前通过预分析得到的。

优先级继承(也叫优先级捐赠)协议：
优先级继承协议(Priority Inheritance Protocol, PIP)的基本思想是当更高优先级的进程A被进程B阻塞时，B暂时继承A的优先级，这将防止中间优先级的进程抢占进程B使高优先级进程A的阻塞时间延长。

优先级继承协议应对优先级反转

优先级天花板协议应对优先级反转

优先级天花板和优先级继承的异同
区别：

优先级继承需要将资源owner任务的低优先级提升为和试图获取资源的任务的同样的高优先级，所以，对于实现优先级继承的系统来说，支持任务的同等优先级是必要条件．

所以，在不支持同等优先级的系统中，无法实现优先级继承，只能实现优先级天花板机制．这样的系统有ucosii等等．

所以，区别1,优先级继承只能实现在支持同等优先级的系统中，而且优先级天花板则与系统能否支持同等优先级无关．

联系：

优先级继承是优先级天花板的特殊情况，当天花板优先级和owner任务的优先级相同时，优先级天花板就变成了优先级继承．
Musl库中对优先级天花板和优先级继承的支持
musl的支持策略貌似是不支持!

UCOSII优先级天花板的实现:
OS_EVENT *OSMutexCreate (INT8U prio, INT8U *perr);
OS_EVENT *OSMutexDel (OS_EVENT *pevent, INT8U opt, INT8U *perr);
void OSMutexPend (OS_EVENT *pevent, INT16U timeout, INT8U *perr);
INT8U OSMutexPost (OS_EVENT *pevent);
INT8U OSMutexQuery (OS_EVENT *pevent, OS_MUTEX_DATA *p_mutex_data);
static void OSMutex_RdyAtPrio (OS_TCB *ptcb, INT8U prio);
BOOLEAN OSMutexAccept (OS_EVENT *pevent, INT8U perr)
OS_EVENT结构体定义：
388 #if (OS_EVENT_EN) && (OS_MAX_EVENTS > 0)
389 typedef struct os_event {
390 INT8U OSEventType; / Type of event control block (see OS_EVENT_TYPE_xxxx) */
391 void OSEventPtr; / Pointer to message or queue structure /
392 INT16U OSEventCnt; / Semaphore Count (not used if other EVENT type) /
393 #if OS_LOWEST_PRIO <= 63
394 INT8U OSEventGrp; / Group corresponding to tasks waiting for event to occur /
395 INT8U OSEventTbl[OS_EVENT_TBL_SIZE]; / List of tasks waiting for event to occur /
396 #else
397 INT16U OSEventGrp; / Group corresponding to tasks waiting for event to occur /
398 INT16U OSEventTbl[OS_EVENT_TBL_SIZE]; / List of tasks waiting for event to occur */
399 #endif
400
401 #if OS_EVENT_NAME_SIZE > 1
402 INT8U OSEventName[OS_EVENT_NAME_SIZE];
403 #endif
404 } OS_EVENT;
405 #endif
OS_EVENT *OSMutexCreate (INT8U prio, INT8U *perr)中，第一个参数prio表示的访问互斥量时使用的优先级。换句话说，当信号量被获取时，一个更高优先级的任务尝试获取该信号量，那么拥有该信号量的任务的优先级将被提升到该优先级。这里需要指定一个优先级，其优先级高于任何竞争互斥锁的任务。其值被记录再pevent->OSEventCnt中．

void OSMutexPend (OS_EVENT *pevent, INT16U timeout, INT8U *perr)函数中，如果mutex当前处于有资源的状态，则无论任务的优先级是高还是低，都不改变当前优先级（因为不存在其它优先级的竞争，优先级天花板协议就可以推迟生效,假如竞争不存在，优先级提升就没有必要了）．获取后，将owner线程记录再pevent->OSEventPtr中．并且抹去avaliable状态，变为owner任务的优先级．这里还要进行一项检查，确保PIP优先级是所有会竞争mutex的任务中最高的．

如果mutex当前无资源，则调整owner的优先级到pip级．

这是ucosii的做法，看似没有问题，实际上，这里存在一个bug.
UCOSII Mutex嵌套使用时出现的 bug
下图展示了一个UCOSII Mutex在嵌套使用时出现的Bug,连ＰＩＰ机制都救不了．

图中包含 4 个任务 A、B、C、D 以及两个互斥信号量 S1、S2。用户设置的原始优先级满足下述关系：PrD < PrC < PrS2 < PrB < PrA < PrS1。假设一开始任务 A、B、C 被延迟并且正在等待时钟中断唤醒，D 是唯一正在运行的任务。首先任务 D 获取信号量S2，然后 C 被唤醒并抢占了 D。C 接着获取了信号量 S1, 然后尝试获取 S2，但因为 S2 已经被 D 占有，所以 C 会被阻塞，因为 PrD < PrC，所以根据协议将D 的优先级提升到 PrS2，并且 D 被调度执行。然后任务 A 被唤醒，因为 A 的优先级最高，所以 A 抢占了 D。此时 A 尝试获取被 C 占有的 S1 ，所以 A 被阻塞，并且 C 的优先级被提升到 PrS1。最后任务 B 被唤醒并执行，但此时已经发生了无边界优先级反转，因为任务 A 可能无限期等待任务 B。并且从图中可以看出，此时任务 B 不占有任何资源，并且原始优先级低于 A。前者何时出让处理器，不可预期。图中括号内的二元组表示该任务当前优先级和状态．

从就绪队列的角度看，最高优先级的任务A的优先级，并没有传染给就绪队列的其他任务，因为虽然任务Ｃ受到了A的影响，被提升了优先级，但是由于任务Ｃ睡眠在了S2上．S1的优先级并没有间接影响到S2的owner任务，也就是Ｄ．导致无边界优先级反转的情况再次出现．

这个bug,在rt-thread操作系统中同样存在, 口说无凭，程序为证,下面这段程序没有用法上面的问题，但却可以把一个活活的RTT系统跑死(同样的逻辑我在Nuttx和UCOSII中都跑过，同样有问题，所以这个例子不能说明RTT有问题，只能说它follow的是通常的设计实现,RTT仍然是我向大家首推的系统).

#include <typedef.h>
#include <rtthread.h>
#include <waitqueue.h>
#include <rthw.h>
#include <arch.h>
#include <log.h>
 
void udelay(unsigned int us);
static rt_mutex_t mutex_s1, mutex_s2;
static void pri_task_a(void *ARG_UNUSED(para))
{
    while (1)
    {
        rt_thread_delay(2000);
        __log("before got s1");
        rt_mutex_take(mutex_s1, RT_WAITING_FOREVER);
        __log("got s1");
        udelay(100 * 1000 * 1000);
        __log("after delay 100s");
        rt_mutex_release(mutex_s1);
        __log("release s1");
    }
}
 
static void pri_task_b(void *ARG_UNUSED(para))
{
    rt_thread_delay(5000);
 
    while (1)
    {
        udelay(1);
    }
}
 
static void pri_task_c(void *ARG_UNUSED(para))
{
    while (1)
    {
        rt_thread_delay(1000);
        __log("before got s1");
        rt_mutex_take(mutex_s1, RT_WAITING_FOREVER);
        __log("got s1");
        rt_mutex_take(mutex_s2, RT_WAITING_FOREVER);
        __log("got s2");
        udelay(100 * 1000 * 1000);
        __log("delay 100s");
        rt_mutex_release(mutex_s2);
        __log("release s2");
        rt_mutex_release(mutex_s1);
        __log("release s1");
    }
}
 
static void pri_task_d(void *ARG_UNUSED(para))
{
    while (1)
    {
        __log("before got s2");
        rt_mutex_take(mutex_s2, RT_WAITING_FOREVER);
        __log("got s2");
        udelay(100 * 1000 * 1000);
        __log("after delay 100s.");
        rt_mutex_release(mutex_s2);
        __log("release s2.");
    }
}
 
/* ----------------------------------------------------------------------------*/
/*
 *  @brief  schedule_in_irqlock <test pattern, Thread swith out when iterrupt
 *          disable>
 */
/* ----------------------------------------------------------------------------*/
void priority_entry(void)
{
    rt_thread_t thread;
 
    mutex_s1 = rt_mutex_create("mutexs1", RT_IPC_FLAG_PRIO /*RT_IPC_FLAG_FIFO*/);
    mutex_s2 = rt_mutex_create("mutexs2", RT_IPC_FLAG_PRIO /*RT_IPC_FLAG_FIFO*/);
    if (mutex_s1 == RT_NULL || mutex_s2 == RT_NULL)
    {
        __err("fatal error");
        return;
    }
 
    thread = rt_thread_create("pri_a", pri_task_a, RT_NULL, 0x1000, 0, 10);
    rt_thread_startup(thread);
 
    thread = rt_thread_create("pri_b", pri_task_b, RT_NULL, 0x1000, 1, 10);
    rt_thread_startup(thread);
 
    thread = rt_thread_create("pri_c", pri_task_c, RT_NULL, 0x1000, 2, 10);
    rt_thread_startup(thread);
 
    thread = rt_thread_create("pri_d", pri_task_d, RT_NULL, 0x1000, 3, 10);
    rt_thread_startup(thread);
 
    return;
}

————————————————
版权声明：本文为CSDN博主「tugouxp」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/tugouxp/article/details/116428458

左手右手一个慢动作 LV 3

@caozilong
非常好的文章，想请教一下有什么工具或者方法对死锁问题进行debug吗