开始前,会这些就够了
- 知道同一把互斥锁同一时刻只能由一方持有
- 理解任务可能等待而不能继续运行
- 会运行Python文件
- 需要补基础时:先补线程和共享状态
高优先级任务 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 不用这把锁。
- H 等锁时间
- 6 格
- H 完成时刻
- 8
- 全部工作总量
- 8 格
当前关闭继承,H 在时刻 7 拿到锁,时刻 8 完成。切换后,H 会在时刻 4 完成。
继承让持锁者先完成,H 便能更早继续;M 的工作被推后,总工作量没有减少。
逐格检查 H 的状态与锁主人
| 时间格 | 运行任务 | H 状态 | 锁主人 | L 是否继承 |
|---|---|---|---|---|
| 0—1 | L | 尚未到达 | L | 否 |
| 1—2 | M | 等 L 解锁 | L | 否 |
| 2—3 | M | 等 L 解锁 | L | 否 |
| 3—4 | M | 等 L 解锁 | L | 否 |
| 4—5 | M | 等 L 解锁 | L | 否 |
| 5—6 | L | 等 L 解锁 | L | 否 |
| 6—7 | L | 等 L 解锁 | L | 否 |
| 7—8 | H | 正在运行 | 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线程的实测。
展开收起代码 · 小实验
长行可在代码区域内左右滚动,按 Tab 可把焦点移入代码区。
"""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 早已解锁,优先级继承还会改变这一次等待吗?
- 自行修改到达时刻或持锁工作量,先画出何时发生真正的锁等待。
- 比较开启与关闭继承时的任务完成顺序,同时记录全部任务完成时刻。
- 写出本实验遗漏的一个现实因素,并说明需要什么额外证据才能讨论它。
做到这些,就算完成
- 继承只在有等待依赖时触发,而不是永久抬高L。
- 区分H完成时间、锁等待时间与全部工作总时间。
- 保留教学模型标识;本节不提供时间表答案。
查原始资料
想确认一个细节,先看这份资料对应的部分,不必从头读完所有文档。
- Linux内核:RT-mutex implementation design (新标签页)↗
三任务间的无界优先级反转例子,以及优先级继承针对这条依赖的作用。
- Linux内核:RT-mutex subsystem with PI support (新标签页)↗
持锁者临时继承等待者优先级及解锁后的调整,不能将其视为所有应用问题的万能修复。
- POSIX Programmer's Manual:mutex protocol (新标签页)↗
PTHREAD_PRIO_NONE与PTHREAD_PRIO_INHERIT的协议含义及显式属性。