锁系列第二篇。互斥锁解决"互斥",但不解决"公平"。本篇从一次真实的航天事故讲起,把优先级反转完整推演一遍,然后对比三种解法、讲清等待队列策略的配套关系,最后给出可落地的排查与防御手段。
文中代码为伪代码,事故细节引自公开资料。


一、开场:火星上的反复重启

1997 年,火星探路者号着陆器在火星表面执行任务期间,出现了周期性复位。地面收到的遥测显示:

  • 系统反复重启,每次重启后又能正常工作一段时间
  • 复位原因指向看门狗超时
  • 日志里没有任何"错误",没有异常码,没有崩溃堆栈

工程团队在地面搭建了复现环境,反复加压后终于复现:一个低优先级的气象数据采集任务,间接把高优先级的总线管理任务饿死了。

根因不是死锁——系统一直在跑,任务都在"活着",只是高优先级任务得不到执行机会。而看门狗只认时间,不认原因,于是到点就复位。

这个案例后来被反复引用,成为实时系统里最经典的教训之一:"我用了互斥锁"并不等于"我不会被反转"。


二、三步推演:反转到底怎么发生

设定三个任务,优先级 H > M > L,一把互斥锁保护共享资源:

H(高优先级):需要访问共享资源,持锁时间极短
M(中优先级):完全不碰这把锁,但很吃 CPU
L(低优先级):大量使用共享资源,临界区较长

正常情况没问题,H 来了就抢,L 拿锁时 H 等着也不会太久。问题出现在特定交错时序:

时刻HML锁状态说明
T0——运行L 持有L 进入长临界区
T1就绪—运行L 持有H 要这把锁,被阻塞
T2阻塞就绪被抢占L 持有M 抢占 L(L 优先级仍低于 M)
T3阻塞运行就绪L 持有M 一直跑,L 拿不到 CPU
T4阻塞运行就绪L 持有H 的等待时间被 M 的执行时间无限拉长
T5阻塞结束运行L 持有M 终于结束,L 继续
T6阻塞—释放空闲L 退出临界区
T7运行——由 H 持有H 终于拿到锁

关键在 T2~T4:H 的等待时间 = L 剩余临界区 + 所有中优先级任务的执行时间总和。

如果 M 是周期性任务、每次跑几毫秒,H 的等待时间可能达到几十甚至上百毫秒——对实时系统来说,这已经是灾难。

这不是"锁慢",是"优先级被间接降级"

用一句话概括机制:

H 因为等待 L 持有的锁,实际上被降到了 L 的优先级;而 L 又会被任何中优先级任务抢占,于是 H 就被所有中优先级任务"排队"了。

这种阻塞叫间接阻塞(indirect blocking),它和"直接阻塞(等锁)"是两回事:

类型来源可控性
直接阻塞H 等待 L 释放锁可控(临界区多长就等多久)
间接阻塞H 等待时,L 被 M 抢占不可控(取决于系统里有多少 M)

这才是优先级反转真正可怕的地方:等待时长与你自己的代码无关,取决于整个系统里其他任务的活跃度。这也是为什么它"偶发"——系统负载轻的时候完全正常,一旦有后台任务跑起来就出问题。


三、几种变体,都值得知道

变体一:多级链式反转

H 等 L1 持有的锁A
L1 在临界区里又要锁B,而锁B 被 L2 持有
L2 又被 M 抢占

此时 H 的等待时间 = L2 的临界区 + L1 的剩余临界区 + 所有 M 的执行时间。链越长,最坏情况越不可控。

推论:临界区内不要再去获取另一把锁。这是"锁层级"设计的铁律。

变体二:中断参与的版本

L 持锁 → ISR 唤醒 H → H 就绪等待锁 → 更高优先级的中断持续到来

即使没有 M,只要 H 在等锁期间被"其他就绪任务"排队,反转就已经成立。

变体三:多核上的版本

多核场景下,持锁者 L 可能在某核上跑,H 在另一个核上自旋等待——如果 L 所在的核被其他任务占满,H 的空转就是纯浪费。多核通常需要自旋锁 + 优先级继承的组合策略,复杂度更高。


四、三种解法,各有代价

方案机制优点代价适用
优先级继承持锁者临时提升到等待者的优先级通用、无需预先知道所有使用者实现复杂(链式、恢复、多锁)互斥量的标准做法
优先级天花板持锁即提升到"该锁所有使用者的最高优先级"最坏情况可静态分析需要静态分析出每个锁的天花板硬实时、安全关键系统
临界区极短化不依赖机制,靠纪律零运行时开销依赖评审与纪律,容易退化任何系统,作为基础

优先级继承的语义

/ *简化版优先级继承(伪代码,不追求可直接编译) */
void mutex_take(mutex_t *m)
{
    task_t *self = task_self();

    if (m->owner == NULL) {
        m->owner = self;
        return;
    }

    /* 1) 沿"所有权链"提升所有间接阻塞者:
          如果持锁者自己也在等另一把锁,要一路提上去 */
    for (task_t *t = m->owner; t != NULL; t = t->blocked_on_owner) {
        if (self->prio < t->prio) {        /* 数值小 = 优先级高(按平台约定) */
            t->prio = self->prio;
        }
    }

    /* 2) 挂到该锁的等待队列 */
    wait_queue_insert(&m->waiters, self);

    /* 3) 让出 CPU */
    schedule();
}

void mutex_release(mutex_t *m)
{
    task_t *self = task_self();

    m->owner = wait_queue_pop(&m->waiters);   /* 交给下一个等待者 */
    restore_priority(self);                   /* 恢复原始优先级 */
    if (m->owner != NULL) {
        wake(m->owner);
    }
}

三个实现细节,是"看起来对、其实有 bug"的高发区:

  1. 恢复优先级要取最大值:一个任务可能同时被多把锁"继承提升",释放其中一把时不能直接恢复成 base_prio,而要取"仍在等待我的所有任务中的最高优先级"
  2. 链式提升要及时:如果提升只做一层,多级反转依然存在
  3. 提升是可传递的副作用:L 的优先级被临时改高后,它可能又去抢别的锁,系统调度行为随之变化——这是继承机制"侵入性"的体现

优先级天花板

做法:预先分析出"每把锁可能被哪些任务使用",取其中最高优先级作为该锁的天花板。任何任务拿到这把锁,立刻升到天花板优先级,释放后恢复。

优点:最坏情况是可静态分析的(阻塞时间上界 = 最长临界区 + 一次天花板阻塞),这在安全关键系统里非常重要。
代价:天花板必须准确——分析错一个使用者,保护就失效;且所有使用者都以最高优先级运行,可能拖慢其他任务。

临界区极短化

前两种是机制,这一种是纪律。它永远有效,而且应该先做:

/* 反例:临界区里做了三件不该做的事 */
lock_take(&m);
record = find_node(&table, key);      /* 遍历,可以移出去 */
log_write("found");                   /* 日志可能有锁、可能阻塞 */
data = malloc(sizeof(*data));         /* 内存分配有锁、可能慢 */
lock_release(&m);

/* 正例:锁内只做最小状态变更 */
lock_take(&m);
node->value = new_value;              /* 一条赋值 */
lock_release(&m);
log_write("updated");                 /* 日志移到锁外 */

五、等待队列策略:FIFO 与 PRIO

这是一个极易被忽略的维度。很多 RTOS 的互斥量创建时可以指定等待队列策略:

策略行为优点缺点
FIFO先来先得,严格排队公平、无饥饿、行为可预测高优先级任务可能排在低优先级后面
PRIO等待队列按优先级排序,高优先级先拿减少高优先级任务的等待极端情况下低优先级可能长期拿不到(饥饿)

两个关键认知

认知一:PRIO 策略并不消除反转

反转的本质是"持锁者被打断",而 PRIO 只决定"锁释放时交给谁"。如果持锁者 L 依然能被 M 抢占,H 依然在等——PRIO 只是让 H 在锁释放后能更快拿到,不解决等待期间的饿死问题。

认知二:策略必须与"是否开启优先级继承"配套

四种组合的实际效果:

等待策略优先级继承实际效果
FIFO关闭最差:既可能反转,又要排队,高优先级任务最坏情况极长
FIFO开启可用:继承避免了饿死,FIFO 保证公平
PRIO关闭半吊子:锁释放后高优先级先拿,但等待期间照样被 M 拖死
PRIO开启通常最优:等待期间有继承保护,释放后高优先级优先

所以:如果你看到工程里两把锁用了不同策略,先别急着统一——要问的是"这两把锁是否都开启了继承、使用者优先级跨度多大"。

一个真实的判断方法

对每把锁问三个问题:

  1. 这把锁的使用者优先级跨度大吗?(如果都同优先级,策略无所谓)
  2. 是否开启了优先级继承?(没有的话,策略再讲究也防不住反转)
  3. 是否存在"低优先级长临界区"?(有的话,优先缩短临界区,而不是调策略)

六、超时:最后的防线

前面所有机制都是"防反转",超时则是"限制损害"。

现象:大多数工程里,互斥量几乎清一色使用无限等待。

lock_take(&m, WAIT_FOREVER);   /* 工程里的绝对主流 */

风险链条很清晰:

某处持锁逻辑出错(或死锁)
   → 任务永久阻塞在等锁上
   → 该任务负责的功能全部停摆
   → 看门狗发现"心跳没喂" → 整机复位
   → 现场数据丢失 / 用户感知为"设备自己重启了"

改用带超时的等待,至少能把"整机复位"降级为"单个功能报错":

if (lock_take(&m, WAIT_MS(100)) != OK) {
    /* 超时:记录现场,便于事后分析 */
    log_warn("lock timeout: holder=%s waiters=%d held_for=%llu us",
             task_name(lock_owner(&m)),
             lock_waiter_count(&m),
             lock_held_duration_us(&m));
    return ERR_BUSY;          /* 降级返回,而不是永久挂死 */
}

三条使用纪律:

  1. 不要盲目重试:超时后立刻重试会放大问题(尤其是持锁者已经卡死时),应配合退避或直接上报
  2. 超时值要有依据:取"正常最坏持有时间 × 3~5",而不是随手写个 100
  3. 超时必须能观测:超时日志要带持锁者信息,否则排查时只知道"超时了",不知道"被谁卡住"

七、如何发现优先级反转

反转的排查难点在于:它不留痕迹。没有错误码,没有崩溃,只有"偶发变慢"。

现象特征(出现两条以上就要怀疑)

特征说明
高优先级任务偶发延迟,且无法用自身代码解释最典型
延迟与低优先级任务的活跃度正相关系统空闲时正常,后台任务一跑就慢
CPU 占用不高,但响应慢大家在等,不是在算
延迟出现时,中优先级任务正在运行抓现场时的关键证据
问题不可稳定复现,加压后概率上升时序相关

排查手段

手段一:把"最大等待时间"变成监控指标

不要只看平均值。为每把锁记录 wait_max 和 hold_max(见第一篇的统计封装),超过阈值就打印现场。反转的特征正是"最大值离谱、平均值正常"。

手段二:记录持锁者

/* 带调试信息的锁:谁持有、持有多久、谁在等 */
typedef struct {
    lock_t   lock;
    task_t  *owner;
    uint64_t held_since;
    uint32_t waiter_cnt;
} dbg_lock_t;

超时或长等待时,一次性打印:持锁者是谁、持了多久、当前在什么状态、谁在等。这四个信息基本能定位问题。

手段三:任务状态快照

出现延迟时抓一次全任务状态表,看"H 是不是在 WAIT_MUTEX、L 是不是在 RUNNING、还有谁在跑"。如果 H 在等锁而 L 没在跑,那就是典型的反转现场。

手段四:时间线对齐

把"H 开始等待"与"L 恢复执行"两个时刻对齐,中间发生了什么——这段时间里跑的是谁,答案就是罪魁祸首。


八、最小复现(可用于教学与验证)

/* 三任务最小复现:验证你的系统是否真的开启了优先级继承 */

sem_t g_low_started;         /* 让 L 先启动 */
mutex_t g_lock;              /* 关键:这把锁是否带优先级继承 */

void task_low(void *arg)
{
    for (;;) {
        sem_wait(&g_low_started);
        mutex_take(&g_lock);
        busy_wait_ms(50);    /* 模拟长临界区 */
        mutex_release(&g_lock);
        sleep_ms(10);
    }
}

void task_mid(void *arg)
{
    for (;;) {
        busy_wait_ms(200);   /* 不碰锁,纯吃 CPU */
        sleep_ms(5);
    }
}

void task_high(void *arg)
{
    for (;;) {
        uint64_t t0 = now_us();
        mutex_take(&g_lock);
        uint64_t wait = now_us() - t0;
        mutex_release(&g_lock);

        if (wait > 10000) {  /* 10ms 以上就报警 */
            log_error("high wait=%llu us -> inversion suspected", wait);
        }
        sleep_ms(100);
    }
}

判读方法:

  • 关闭继承:wait 会出现几十毫秒的尖峰,且尖峰与 M 的运行周期相关
  • 开启继承:wait 应该收敛到"L 的剩余临界区"量级,尖峰消失或大幅下降

这个实验成本很低(半小时),但能验证一个非常关键的事实:你系统里的"互斥量",到底有没有继承。很多工程用了多年才发现,自己用的"互斥量"其实是个二值信号量。


九、自查清单

  1. 保护共享数据用的是互斥量还是二值信号量?后者没有优先级继承
  2. 互斥量是否真的开启了继承?(查创建属性,不要假设)
  3. 等待队列策略与继承是否配套?
  4. 是否存在"低优先级任务持有长临界区"?这是反转的必要条件
  5. 临界区内是否获取了第二把锁?会形成链式反转
  6. 是否全部使用无限等待?至少关键路径要设超时
  7. 是否统计了"最大等待时间"?平均值会骗你
  8. 是否有"持锁者记录",便于超时后定位?

Logo

社区规范:仅讨论OpenHarmony相关问题。

更多推荐