qinghuan
← 回到知识探索

计算机趣闻 / 可下载小项目

优先级最高的任务,为什么会被一个没抢它锁的任务拖住?

低优先级拿着锁,高优先级等锁,中优先级一直占CPU:用单锁时间表看懂优先级反转与临时继承。

展开本文目录
  1. 下载完整练习包
  2. 先动手看看
  3. 优先级高,也必须先处于可以运行的状态
  4. 没有继承时,中优先级多算多久,高优先级就多等多久
  5. 临时抬高持锁者,是为了让它早点交回资源
  6. 运行完整时间表,别把虚拟格子读成毫秒
  7. 它调整谁先完成,不会凭空减少总工作
  8. 只看最高优先级任务,会漏掉真正阻塞它的人
  9. 记住这三点
  10. 别把例子用过头
  11. 自己试一小步
  12. 查原始资料
开始前,会这些就够了
  • 知道同一把互斥锁同一时刻只能由一方持有
  • 理解任务可能等待而不能继续运行
  • 会运行Python文件
  • 需要补基础时:先补线程和共享状态
遇到不熟悉的词,先看这 2 个解释
阻塞
任务正在等某个条件,当前还不能继续执行,因此不参加本轮可运行任务选择。
优先级继承
持锁者临时借用高优先级等待者的优先级,争取尽快完成并解锁。
打开完整术语小抄 ↗

高优先级任务 H 需要一把锁,但锁在低优先级任务 L 手里。H 只能先等 L 做完。如果这时中优先级任务 M 开始大量计算,调度器会先运行 M,L 没机会释放锁,H 也就跟着一直等。

M 根本没有拿那把锁,却间接拖住了 H。本文用虚拟的一格一格时间把依赖画出来。所有任务都是数据记录,程序不会改变你电脑上的线程优先级。

把例子带到自己的电脑上

下载完整练习包

需要 Python 3.10 或更新版本,无第三方依赖;制作时实测 Windows / Python 3.14.0。使用标准库,不需要安装第三方包。下载 ZIP 后选择“全部提取”,进入能看到 README.md 的文件夹。

下载 ZIP · 5 个文件
包里有什么
  • priority_inversion.py:完整可运行示例
  • README.md:运行说明与独立练习
  • observed.stdout.txt:本次实际输出
  • verification.json:源码和产物验证记录
  • SHA256SUMS.txt:内容摘要

在这个文件夹打开 PowerShell,逐行运行

python --version
python -B priority_inversion.py

解压到独立练习文件夹,在含 priority_inversion.py 的目录打开终端运行。单CPU、单锁、三任务、固定整数优先级与虚拟tick的调度教学模型;无真实线程,无系统调度接口,不包含IO、多锁传播、死锁预防或实时保证。

找不到命令或文件时,先按 README 检查环境和所在目录。文中标出的阅读时间只估算文章篇幅,安装环境和动手练习可按自己的节奏进行。

谁先完成,取决于谁在等谁

只有一个 CPU、一把锁。L 已持锁,优先级为 1;H 和 M 在时刻 1 到达,优先级分别为 3、2。H 必须拿到锁才能执行最后一格,M 不用这把锁。

0—1L优先级 1
1—2M优先级 2
2—3M优先级 2
3—4M优先级 2
4—5M优先级 2
5—6L优先级 1
6—7L优先级 1
7—8H优先级 3
H 等锁时间
6 格
H 完成时刻
8
全部工作总量
8 格

当前关闭继承,H 在时刻 7 拿到锁,时刻 8 完成。切换后,H 会在时刻 4 完成。

继承让持锁者先完成,H 便能更早继续;M 的工作被推后,总工作量没有减少。

逐格检查 H 的状态与锁主人
时间格运行任务H 状态锁主人L 是否继承
0—1L尚未到达L否
1—2M等 L 解锁L否
2—3M等 L 解锁L否
3—4M等 L 解锁L否
4—5M等 L 解锁L否
5—6L等 L 解锁L否
6—7L等 L 解锁L否
7—8H正在运行H否

时间格是虚拟 tick,不是毫秒。每格结束先完成工作和解锁,再处理下一边界的到达;拿锁、阻塞与唤醒不消耗 CPU 格。本模型没有操作真实线程优先级,不含多锁、I/O 与切换成本,也不证明系统实时性。

优先级高,也必须先处于可以运行的状态

这份模型每格只让一个可运行任务执行,并在其中选择有效优先级最高的任务。等待锁的任务处于阻塞状态,不参加这次选择。H 的数字再高,也不能越过互斥规则直接使用 L 尚未释放的资源。

设置 L、M、H 的基础优先级为 1、2、3,这里明确约定数字越大越优先。这个数字方向只是教学规则,不对应某个系统的 nice 值或调度接口。L 在时刻 0 之前已拿到唯一一把锁,持锁工作需要 3 格;H 和 M 在时刻 1 到达。

没有继承时,中优先级多算多久,高优先级就多等多久

L 先执行第 0 格,还剩 2 格。到时刻 1,H 尝试拿锁而阻塞,M 则有 4 格普通计算。可运行者只有 L 和 M,M 优先级更高,于是从时刻 1 算到 5。L 随后才有机会完成余下两格,并在时刻 7 解锁。H 最后执行一格,在时刻 8 完成。

时间表因此是 L、M、M、M、M、L、L、H。H 等锁用了 6 格,其中有 4 格来自不需要这把锁的 M。这种间接拖延就是本例要解释的反转;如果中优先级工作持续到来,低优先级持锁者还可能长期得不到执行机会。

临时抬高持锁者,是为了让它早点交回资源

加入优先级继承后,H 等待 L 的锁时,L 暂时采用 H 的优先级 3。M 的 2 便不能再抢在 L 前面。L 在时刻 1、2 连续完成余下工作,时刻 3 解锁,H 紧接着执行,在时刻 4 完成。

时间表变成 L、L、L、H、M、M、M、M。不是把锁硬抢给 H,而是让阻塞链上的持锁者先完成必要工作。在这个单锁模型里,L 解锁后恢复基础优先级;真实系统的多把锁、多个等待者与优先级传播要按具体实现分析。

运行完整时间表,别把虚拟格子读成毫秒

保存为 priority_inversion.py,执行 python -B priority_inversion.py。输出列出每格运行谁、当时有效优先级,以及 H 阻塞和拿锁的时刻。两组 H 完成时刻分别为 8 和 4,等锁时间分别为 6 和 2;所有数字的单位都是模型里的 tick。

程序再枚举 L 工作量为2—6格、M为0—8格的45种组合,并检查没有等锁、非法工作量等边界,共142项核对。锁的获取、阻塞与释放在模型边界瞬间完成,不计切换成本;这不是对Windows、Linux或Python线程的实测。

展开收起代码 · 小实验python · 88 行

长行可在代码区域内左右滚动,按 Tab 可把焦点移入代码区。

小实验 · python
"""Deterministic single-CPU/single-lock model; no OS priority APIs or threads."""
import json


def simulate(medium_work=4, inherit=False, low_work=3):
    if type(medium_work) is not int or medium_work < 0 or type(low_work) is not int or low_work < 1:
        raise ValueError("invalid work units")
    base = {"L": 1, "M": 2, "H": 3}  # Larger means higher in this teaching model.
    remaining = {"L": low_work, "M": medium_work, "H": 1}
    owner = "L"  # L has acquired the only lock just before tick 0.
    blocked = False
    high_arrived = False
    waiting_since = None
    acquired_at = None
    high_finished = None
    timeline = []
    events = [{"at": 0, "event": "L already owns lock"}]
    tick = 0
    while any(remaining.values()):
        if tick == 1 and not high_arrived:
            high_arrived = True
            if owner is not None:
                blocked, waiting_since = True, tick
                events.append({"at": tick, "event": "H blocks on lock owned by L"})
        ready = []
        for name in ("L", "M", "H"):
            if remaining[name] <= 0 or (name in ("M", "H") and tick < 1):
                continue
            if name == "H" and blocked:
                continue
            priority = base[name]
            if inherit and name == owner and blocked:
                priority = base["H"]
            ready.append((priority, name))
        if not ready:
            raise AssertionError("unexpected deadlock in this single-lock workload")
        priority, running = max(ready)
        if running == "H" and owner is None:
            owner, acquired_at = "H", tick
            events.append({"at": tick, "event": "H acquires lock"})
        timeline.append({"tick": tick, "task": running, "effective_priority": priority})
        remaining[running] -= 1
        tick += 1
        if running == "L" and remaining["L"] == 0:
            owner, blocked = None, False
            events.append({"at": tick, "event": "L unlocks; L effective priority returns to 1"})
        if running == "H" and remaining["H"] == 0:
            owner, high_finished = None, tick
        if tick > low_work + medium_work + 5:
            raise AssertionError("model failed to finish")
    return {"inheritance": inherit, "timeline": timeline, "events": events,
            "H_finished_at": high_finished,
            "H_lock_wait_ticks": 0 if waiting_since is None else acquired_at - waiting_since,
            "total_ticks": tick}


def main():
    plain, inherited = simulate(), simulate(inherit=True)
    assert [x["task"] for x in plain["timeline"]] == list("LMMMMLLH")
    assert [x["task"] for x in inherited["timeline"]] == list("LLLHMMMM")
    assert (plain["H_finished_at"], inherited["H_finished_at"]) == (8, 4)
    assert (plain["H_lock_wait_ticks"], inherited["H_lock_wait_ticks"]) == (6, 2)
    checks = 4
    for low_work in range(2, 7):
        for medium_work in range(9):
            p = simulate(medium_work, False, low_work)
            i = simulate(medium_work, True, low_work)
            assert p["H_finished_at"] == low_work + medium_work + 1
            assert i["H_finished_at"] == low_work + 1
            assert p["total_ticks"] == i["total_ticks"] == low_work + medium_work + 1
            checks += 3
    # If L has already released the lock when H arrives, no lock wait occurs.
    for inherit in (False, True):
        no_wait = simulate(4, inherit, 1)
        assert no_wait["H_lock_wait_ticks"] == 0 and no_wait["H_finished_at"] == 2
        checks += 1
    try:
        simulate(-1)
        raise AssertionError("negative workload accepted")
    except ValueError:
        checks += 1
    print(json.dumps({"checks_passed": checks, "without_inheritance": plain,
                      "with_inheritance": inherited,
                      "boundary": "fixed virtual ticks; one CPU, one lock, no IO or lock chains"}, indent=2))


if __name__ == "__main__":
    main()

怎样运行:在独立练习文件夹中执行 python -B priority_inversion.py。2026-10-17 已在 Windows、Python 3.14.0 运行此份源码并通过自带检查。单CPU、单锁、三任务、固定整数优先级与虚拟tick的调度教学模型;无真实线程,无系统调度接口,不包含IO、多锁传播、死锁预防或实时保证。

运行后对照结果
{
  "checks_passed": 142,
  "without_inheritance": {
    "inheritance": false,
    "timeline": [
      {
        "tick": 0,
        "task": "L",
        "effective_priority": 1
      },
      {
        "tick": 1,
        "task": "M",
        "effective_priority": 2
      },
      {
        "tick": 2,
        "task": "M",
        "effective_priority": 2
      },
      {
        "tick": 3,
        "task": "M",
        "effective_priority": 2
      },
      {
        "tick": 4,
        "task": "M",
        "effective_priority": 2
      },
      {
        "tick": 5,
        "task": "L",
        "effective_priority": 1
      },
      {
        "tick": 6,
        "task": "L",
        "effective_priority": 1
      },
      {
        "tick": 7,
        "task": "H",
        "effective_priority": 3
      }
    ],
    "events": [
      {
        "at": 0,
        "event": "L already owns lock"
      },
      {
        "at": 1,
        "event": "H blocks on lock owned by L"
      },
      {
        "at": 7,
        "event": "L unlocks; L effective priority returns to 1"
      },
      {
        "at": 7,
        "event": "H acquires lock"
      }
    ],
    "H_finished_at": 8,
    "H_lock_wait_ticks": 6,
    "total_ticks": 8
  },
  "with_inheritance": {
    "inheritance": true,
    "timeline": [
      {
        "tick": 0,
        "task": "L",
        "effective_priority": 1
      },
      {
        "tick": 1,
        "task": "L",
        "effective_priority": 3
      },
      {
        "tick": 2,
        "task": "L",
        "effective_priority": 3
      },
      {
        "tick": 3,
        "task": "H",
        "effective_priority": 3
      },
      {
        "tick": 4,
        "task": "M",
        "effective_priority": 2
      },
      {
        "tick": 5,
        "task": "M",
        "effective_priority": 2
      },
      {
        "tick": 6,
        "task": "M",
        "effective_priority": 2
      },
      {
        "tick": 7,
        "task": "M",
        "effective_priority": 2
      }
    ],
    "events": [
      {
        "at": 0,
        "event": "L already owns lock"
      },
      {
        "at": 1,
        "event": "H blocks on lock owned by L"
      },
      {
        "at": 3,
        "event": "L unlocks; L effective priority returns to 1"
      },
      {
        "at": 3,
        "event": "H acquires lock"
      }
    ],
    "H_finished_at": 4,
    "H_lock_wait_ticks": 2,
    "total_ticks": 8
  },
  "boundary": "fixed virtual ticks; one CPU, one lock, no IO or lock chains"
}

它调整谁先完成,不会凭空减少总工作

默认两组时间表都需要 8 格才能把全部工作做完。继承把 H 提前完成的同时,也把 M 排到了后面。它解决的是这个依赖关系下的执行顺序,不是让 CPU 突然多了一倍算力。

继承也不能让慢IO瞬间结束、把过长临界区缩短,或自动消除多把锁之间的循环等待。本篇故意只有一把锁,没有建立死锁场景。Linux rt-mutex 的文档还讨论沿多锁链传播优先级,但这里没有实现,不能凭这张时间表宣称整个系统具备实时性保证。

只看最高优先级任务,会漏掉真正阻塞它的人

排查任务迟迟不完成时,除了看它自身是否在运行,还要看它在等哪个资源、资源由谁持有,以及持有者为什么得不到执行机会。H 的高优先级只描述它自己的排序,无法独立解释整个依赖链。

把临界区做得短而可预测,明确资源协议,并查看目标平台是否支持相应互斥锁属性,才是后续工程工作的入口。本文只帮助你读懂一个最小反例,没有替真实应用配置调度策略。

记住这三点

  • 高优先级但在等锁的任务,不属于当前可运行候选。
  • 中优先级任务能通过拖延低优先级持锁者,间接拖延最高优先级。
  • 优先级继承帮助持锁者尽早解锁,不等于抢锁、消除全部延迟或预防全部死锁。

别把例子用过头

  • 把阻塞任务当作随时可被调度的任务。
  • 把教学优先级1/2/3直接映射到某个操作系统参数。
  • 把虚拟tick的改善写成真实毫秒测量或整个系统的实时保证。

自己试一小步

如果 H 到达时 L 早已解锁,优先级继承还会改变这一次等待吗?

  1. 自行修改到达时刻或持锁工作量,先画出何时发生真正的锁等待。
  2. 比较开启与关闭继承时的任务完成顺序,同时记录全部任务完成时刻。
  3. 写出本实验遗漏的一个现实因素,并说明需要什么额外证据才能讨论它。

做到这些,就算完成

  • 继承只在有等待依赖时触发,而不是永久抬高L。
  • 区分H完成时间、锁等待时间与全部工作总时间。
  • 保留教学模型标识;本节不提供时间表答案。

查原始资料

想确认一个细节,先看这份资料对应的部分,不必从头读完所有文档。

  • Linux内核:RT-mutex implementation design (新标签页)↗Linux Kernel / Steven Rostedt · 英文

    三任务间的无界优先级反转例子,以及优先级继承针对这条依赖的作用。

    2026-10-17 实际打开核对:Unbounded Priority Inversion、Priority Inheritance 两节;后面的多锁链只作为范围提示,未在本文实现。

  • Linux内核:RT-mutex subsystem with PI support (新标签页)↗Linux Kernel development community · 英文

    持锁者临时继承等待者优先级及解锁后的调整,不能将其视为所有应用问题的万能修复。

    2026-10-17 实际打开核对:Basic principles;本篇没有调用或复现内核rt-mutex实现。

  • POSIX Programmer's Manual:mutex protocol (新标签页)↗POSIX Programmer's Manual / The Open Group,man7镜像 · 英文

    PTHREAD_PRIO_NONE与PTHREAD_PRIO_INHERIT的协议含义及显式属性。

    2026-10-17 实际打开核对:DESCRIPTION中协议默认值与继承规则,并阅读PROLOG关于规范与平台实现可能不同的提示;没有执行pthread实验。