优先级继承：rtmutex 与 PI 链

4.1优先级反转：从有界到无界

三个任务，优先级 H > M > L，H 和 L 共享一把锁，M 与锁无关：

有界反转（H 等 L 的临界区）是任何互斥锁都无法避免的，只要临界区短就可接受。无界反转（H 等 M，而 M 可以是任意多个、任意长的中优先级任务）才是灾难。

为什么 RT 让问题变严重

非 RT 内核里 spinlock_t 持有者关着抢占，M 根本插不进来——反转天然有界（代价是 H 也插不进来）。原生 mutex 虽可能无界反转，但持 mutex 的路径相对少。RT 把所有 spinlock 都变成了可抢占的睡眠锁，相当于在全内核上万个位置同时打开了无界反转的口子。所以 PI 不是 RT 的可选优化，而是"spinlock 可睡眠"成立的前提。

几种对策

Linux 选择 PI：它是动态的、不需要预先知道使用关系，而且无争用时零开销。

4.2数据结构：两棵树、三个指针

struct rt_waiter_node {          /* 排序键的一份拷贝 */
	struct rb_node	entry;
	int		prio;
	u64		deadline;
};

struct rt_mutex_waiter {         /* 分配在阻塞任务自己的内核栈上 */
	struct rt_waiter_node	tree;      /* 挂进 lock->waiters，受 lock->wait_lock 保护 */
	struct rt_waiter_node	pi_tree;   /* 挂进 owner->pi_waiters，受 owner->pi_lock 保护 */
	struct task_struct	*task;
	struct rt_mutex_base	*lock;
	unsigned int		wake_state;  /* TASK_NORMAL 或 TASK_RTLOCK_WAIT */
	struct ww_acquire_ctx	*ww_ctx;
};

task_struct 里与 PI 有关的字段（include/linux/sched.h:1195 起）：

	raw_spinlock_t			pi_lock;        /* 保护下面这些 */
	...
	struct rb_root_cached		pi_waiters;     /* 我持有的所有锁的 top waiter，按优先级排序 */
	struct task_struct		*pi_top_task;   /* 当前给我"捐"优先级的任务 */
	struct rt_mutex_waiter		*pi_blocked_on; /* 我正阻塞在哪个 waiter 上（一次只能一个） */

有效优先级的计算（kernel/sched/sched.h:3848）：

static inline int __rt_effective_prio(struct task_struct *pi_task, int prio)
{
	if (pi_task)
		prio = min(prio, pi_task->prio);
	return prio;
}

注意它取的是 pi_task->prio（捐赠者的有效优先级），所以继承是可传递的：捐赠者自己被提升了，被捐赠者也跟着提升。还有一个细节：__waiter_prio()（kernel/locking/rtmutex.c:348）把所有非 RT/DL 任务的排队优先级统一成 DEFAULT_PRIO——nice 值不参与 PI，普通任务之间不互相提升。

4.3加锁：从快路径到 PI

flowchart TD
  A["rt_mutex_lock / mutex_lock (RT)"] --> B{"cmpxchg owner NULL→current"}
  B -- 成功 --> OK["持有锁，零 PI 开销"]
  B -- 失败 --> C["rt_mutex_slowlock：拿 wait_lock"]
  C --> D{"try_to_take_rt_mutex<br/>（先置 HAS_WAITERS 位）"}
  D -- 拿到 --> OK
  D -- 没拿到 --> E["task_blocks_on_rt_mutex"]
  E --> E1["填 waiter，入 lock->waiters<br/>current->pi_blocked_on = waiter"]
  E1 --> F{"我是这把锁的新 top waiter?"}
  F -- 否 --> W["等待循环"]
  F -- 是 --> G["owner->pi_waiters：换掉旧 top waiter<br/>rt_mutex_adjust_prio(owner) → rt_mutex_setprio"]
  G --> H{"owner 自己也阻塞在别的锁上?"}
  H -- 否 --> W
  H -- 是 --> I["释放 wait_lock<br/>rt_mutex_adjust_prio_chain 沿链传播"]
  I --> W
  W --> J{"try_to_take_rt_mutex"}
  J -- 拿到 --> OK
  J -- 否 --> K["自适应自旋或 rt_mutex_schedule 睡眠"]
  K --> J
  classDef rt fill:#f6e3dc,stroke:#b5371c,color:#1d1d1f
  class G,I rt
      

task_blocks_on_rt_mutex()（kernel/locking/rtmutex.c:1202）是 PI 的入口，关键片段：

	if (owner == task && !(build_ww_mutex() && ww_ctx))
		return -EDEADLK;                /* 自己等自己：最简单的死锁 */

	raw_spin_lock(&task->pi_lock);
	waiter->task = task;
	waiter->lock = lock;
	waiter_update_prio(waiter, task);
	waiter_clone_prio(waiter, task);

	if (rt_mutex_has_waiters(lock))
		top_waiter = rt_mutex_top_waiter(lock);
	rt_mutex_enqueue(lock, waiter);
	task->pi_blocked_on = waiter;
	raw_spin_unlock(&task->pi_lock);
	...
	raw_spin_lock(&owner->pi_lock);
	if (waiter == rt_mutex_top_waiter(lock)) {
		rt_mutex_dequeue_pi(owner, top_waiter);   /* 旧第一名出 owner 的 PI 树 */
		rt_mutex_enqueue_pi(owner, waiter);       /* 我进去 */
		rt_mutex_adjust_prio(lock, owner);        /* 重新算 owner 的有效优先级 */
		if (owner->pi_blocked_on)
			chain_walk = 1;                       /* owner 也在等别人 → 要传播 */
	} else if (rt_mutex_cond_detect_deadlock(waiter, chwalk)) {
		chain_walk = 1;
	}
	next_lock = task_blocked_on_lock(owner);
	raw_spin_unlock(&owner->pi_lock);

	if (!chain_walk || !next_lock)
		return 0;

	get_task_struct(owner);
	preempt_disable();
	raw_spin_unlock_irq(&lock->wait_lock);
	wake_up_q(wake_q);
	wake_q_init(wake_q);
	preempt_enable();

	res = rt_mutex_adjust_prio_chain(owner, chwalk, lock, next_lock, waiter, task);

一级提升在持有 wait_lock 的情况下就地完成；只有当 owner 自己也阻塞在别的锁上（pi_blocked_on != NULL）时才需要链遍历，而且链遍历是在释放 wait_lock 之后进行的——为了可抢占。

4.4PI 链遍历：rt_mutex_adjust_prio_chain()

一条三级链

算法：每步最多持两把锁

函数顶部注释（kernel/locking/rtmutex.c:625 起）把算法写成了带编号的 13 步，并标出每步受什么保护：[R] = 持有 task 引用计数，[P] = 持有 task->pi_lock，[L] = 持有 lock->wait_lock。翻译并精简如下：

again:
  if (++depth > max_lock_depth) return -EDEADLK      // 链长上限，默认 1024
retry:
  [1]  lock(task->pi_lock)                     [R] → [P1]
  [2]  waiter = task->pi_blocked_on             链尾？→ 结束
  [3]  检查链是否在我们放锁期间变了（next_lock 对比）
  [4]  lock = waiter->lock
  [5]  trylock(lock->wait_lock)                 失败则放掉 pi_lock 从 [1] 重来
  [6]  死锁检测：lock == orig_lock 或 owner(lock) == top_task → -EDEADLK
  [7]  在 lock->waiters 里按 task 的新优先级重新排队 waiter
  [8]  unlock(task->pi_lock); put_task(task)
  [9]  lock 没有 owner？→ 唤醒新 top waiter，结束
  [10] task = owner(lock); get_task; lock(task->pi_lock)
  [11] 若 top waiter 变了：在 owner->pi_waiters 中替换，rt_mutex_adjust_prio(owner)
  [12] next_lock = owner 正阻塞的锁（只记地址，不解引用）
  [13] unlock(task->pi_lock); unlock(lock->wait_lock)
       owner 没阻塞 → 结束；否则 goto again（owner 成为下一轮的 task）

flowchart TD
  S(["进入：task = 持有者, 持有其引用"]) --> D{"++depth 超过 max_lock_depth?"}
  D -- 是 --> DL["-EDEADLK（打印一次警告）"]
  D -- 否 --> P1["[1] 拿 task->pi_lock"]
  P1 --> P2{"[2][3] task 还阻塞在<br/>预期的 next_lock 上?"}
  P2 -- 否：链已变化或到头 --> END(["结束"])
  P2 -- 是 --> P5{"[5] trylock lock->wait_lock"}
  P5 -- 失败 --> R1["放 pi_lock, cpu_relax"] --> P1
  P5 -- 成功 --> P6{"[6] lock 是起点锁, 或其 owner<br/>是发起者 top_task?"}
  P6 -- 是 --> DL2["-EDEADLK"]
  P6 -- 否 --> P7["[7] waiter 在 lock 的等待树中按新优先级重排"]
  P7 --> P8["[8] 放 task->pi_lock 与引用"]
  P8 --> P9{"[9] lock 有 owner?"}
  P9 -- 无 --> WK["若 top waiter 变了则唤醒它"] --> END
  P9 -- 有 --> P10["[10] task = owner, 拿其 pi_lock"]
  P10 --> P11["[11] 更新 owner->pi_waiters,<br/>rt_mutex_adjust_prio(owner)"]
  P11 --> P12["[12] next_lock = owner 阻塞的锁"]
  P12 --> P13["[13] 放两把锁"]
  P13 --> Q{"next_lock 为空,<br/>或无需继续提升?"}
  Q -- 是 --> END
  Q -- 否 --> D
  classDef rt fill:#f6e3dc,stroke:#b5371c,color:#1d1d1f
  classDef lock fill:#f5e1eb,stroke:#8b3a62,color:#1d1d1f
  class P7,P11 rt
  class P1,P5,P10 lock
      

四个设计要点

1. 可抢占。注释原话："We want this to be preemptible and we want hold a maximum of two locks per step."链可能很长（尤其用户态 PI futex 可以构造任意长的链），如果整个遍历都持锁关中断，PI 本身就成了延迟源。代价是每轮开始时都要假设"一切都可能已经变了"，所以有 [3] 的 next_lock 校验——它只存地址用来比较，从不解引用（放锁后那把锁可能已被释放）。
2. 反序加锁用 trylock。正常的锁序是 wait_lock → pi_lock，而链遍历在 [5] 是持着 pi_lock 去拿 wait_lock，反了。所以只能 raw_spin_trylock，失败就全部放开从 [1] 重来，避免 ABBA 死锁。
3. 提前终止。如果本轮的 waiter 不是 owner 的 top PI waiter，或者优先级没变（rt_waiter_node_equal），说明再往上传播也不会改变任何人的优先级——在不做死锁检测时直接结束。这让绝大多数链遍历只走一两步。
4. 深度上限。max_lock_depth 默认 1024（kernel/locking/rtmutex_api.c:14），可通过 /proc/sys/kernel/max_lock_depth 调整（kernel/sysctl.c:1939）。超过就返回 -EDEADLK——用户态可以构造很深的 PI futex 链来消耗内核时间，这个上限是防御手段。

死锁检测

第 [6] 步的判据：沿链走到的锁就是起点锁，或者某把锁的持有者就是发起这次遍历的任务——环出现了。是否执行完整检测由 chwalk 决定（kernel/locking/rtmutex.c:590 rt_mutex_cond_detect_deadlock()）：

rt_mutex_setprio：真正改优先级

链遍历的每一级最终调到 rt_mutex_setprio()（kernel/sched/core.c:7123）。它做的事比"改个数"复杂得多：

• 只改 p->prio（有效优先级），不碰 normal_prio 和 policy——去 boost 时才能恢复；
• 根据新优先级重新选择调度类（__setscheduler_class()）：一个 SCHED_OTHER 任务被 FIFO 90 的等待者提升后，会从 fair 类挪到 rt 类，真正以 RT 身份参与调度；
• 若在运行队列上，要先 dequeue 再以新类 enqueue；正在运行的任务要 put_prev_task/set_next_task；
• SCHED_DEADLINE 的继承：被 DL 任务阻塞时，持有者借用捐赠者的 DL 参数（p->dl.pi_se = pi_task->dl.pi_se）；
• 特例：idle 任务不允许被提升（注释解释了 RT + NOHZ 下 idle 持有 timer base 锁的情形可以安全忽略）。

还有一条反向路径：如果一个正被阻塞的任务被 sched_setscheduler() 改了优先级，那么它下游的持有者们的继承优先级也要跟着变。kernel/sched/syscalls.c:753 会调用 rt_mutex_adjust_pi()（kernel/locking/rtmutex_api.c:459），以 orig_lock = NULL 发起一次链遍历。

调用栈：内核 mutex 上的 PI 提升

4.5解锁与去 boost

解锁时持有者必须恢复原优先级，并把锁"交给"top waiter。mark_wakeup_next_waiter()（kernel/locking/rtmutex.c:1317）：

	raw_spin_lock(&current->pi_lock);
	waiter = rt_mutex_top_waiter(lock);

	/* 从自己的 pi_waiters 中移除并去 boost —— 必须在被唤醒者运行之前完成，
	 * 保证 p->pi_top_task 永远指向一个仍在阻塞的任务 */
	rt_mutex_dequeue_pi(current, waiter);
	rt_mutex_adjust_prio(lock, current);

	/* 锁暂时无主但有等待者：强迫所有新来者进慢路径，
	 * 防止比 top waiter 优先级低的任务偷锁 */
	lock->owner = (void *) RT_MUTEX_HAS_WAITERS;

	/* 去 boost 后、唤醒前不能被抢占，否则又是一次优先级反转 */
	preempt_disable();
	rt_mutex_wake_q_add(wqh, waiter);
	raw_spin_unlock(&current->pi_lock);

注意 rtmutex 不做直接交接（handoff）：owner 被置为"无主 + 有等待者"，被唤醒的 top waiter 自己回到 try_to_take_rt_mutex() 去拿。这样如果在它醒来之前来了一个优先级更高的任务，后者可以直接拿走锁（steal），符合"永远让最高优先级的先走"。

时序

sequenceDiagram
  participant L as 持有者 L（被提升到 90）
  participant M as rt_mutex
  participant H as 等待者 H（90）
  participant S as 调度器
  L->>M: unlock：cmpxchg 失败（HAS_WAITERS）
  L->>M: rt_mutex_slowunlock 拿 wait_lock
  L->>L: mark_wakeup_next_waiter：去 boost，prio 回到 120
  Note over L: preempt_disable 防止此刻被抢占
  L->>M: owner = HAS_WAITERS（无主）
  L->>M: 释放 wait_lock
  L->>H: rt_mutex_wake_up_q：唤醒 H
  L->>S: preempt_enable → H 优先级更高，立即抢占 L
  S->>H: 切换到 H
  H->>M: try_to_take_rt_mutex 成功，成为 owner
      

4.6用户态 PI：PI futex

用户态 RT 程序同样面临优先级反转——而且更常见，因为应用线程间共享 mutex 是常态。Linux 让用户态复用同一套内核 rtmutex，入口是 PI futex（2.6.18 合入主线，早于 RT 的其它大部分改造）：

pthread_mutexattr_t a;
pthread_mutexattr_init(&a);
pthread_mutexattr_setprotocol(&a, PTHREAD_PRIO_INHERIT);   /* ← 这一行 */
pthread_mutex_init(&m, &a);

futex 字的编码

PI futex 是用户空间里的一个 32 位整数（include/uapi/linux/futex.h:152）：

 bit 31          bit 30             bits 29..0
+---------------+------------------+--------------------------------+
| FUTEX_WAITERS | FUTEX_OWNER_DIED |  owner TID (FUTEX_TID_MASK)    |
+---------------+------------------+--------------------------------+
  0x80000000      0x40000000          0x3fffffff

• 无竞争加锁完全在用户态：cmpxchg(word, 0, my_tid)，成功即持有，不进内核。
• 无竞争解锁：cmpxchg(word, my_tid, 0)。只有当 FUTEX_WAITERS 被置位时这个 cmpxchg 才会失败、才需要进内核。
• 内核通过 TID 就能找到持有者的 task_struct，从而知道要提升谁——这正是普通 futex 做不到 PI 的原因（普通 futex 字里没有 owner）。

内核侧：代理加锁

持有者从未进过内核，内核里没有任何 rtmutex 记录"它持有锁"。所以第一个阻塞者进内核时，内核要替持有者创建一个已被它持有的 rtmutex——这就是 proxy lock：

sequenceDiagram
  participant A as 线程 A（OTHER，持锁）
  participant W as futex 字（用户内存）
  participant K as 内核 futex/rtmutex
  participant B as 线程 B（FIFO 80）
  A->>W: cmpxchg 0 → TID_A（用户态，成功）
  B->>W: cmpxchg 0 → TID_B 失败
  B->>K: futex(FUTEX_LOCK_PI)
  K->>W: futex_lock_pi_atomic：置 FUTEX_WAITERS 位
  K->>K: attach_to_pi_owner：由 TID 找到 A，新建 pi_state
  K->>K: rt_mutex_init_proxy_locked(pi_mutex, A)：替 A 持有
  K->>K: __rt_mutex_start_proxy_lock(FULL_CHAINWALK)
  K-->>A: rt_mutex_setprio：A 被提升到 FIFO 80
  K->>K: rt_mutex_wait_proxy_lock：B 睡眠
  A->>W: 解锁 cmpxchg TID_A → 0 失败（有 WAITERS 位）
  A->>K: futex(FUTEX_UNLOCK_PI)
  K->>W: wake_futex_pi：写入 FUTEX_WAITERS 或 TID_B
  K->>K: __rt_mutex_futex_unlock：A 去 boost，交给 B
  K-->>B: 唤醒，B 从 FUTEX_LOCK_PI 返回，已持有锁
      

关键数据结构 struct futex_pi_state（kernel/futex/futex.h:124）就是"一个 rt_mutex_base + owner + 引用计数 + futex key"。futex_lock_pi()（kernel/futex/pi.c:918）里有一段 RT 专属的注释值得一读：

	/*
	 * On PREEMPT_RT, when hb->lock becomes an rt_mutex, we must not
	 * hold it while doing rt_mutex_start_proxy(), because then it will
	 * include hb->lock in the blocking chain, even through we'll not in
	 * fact hold it while blocking. This will lead it to report -EDEADLK
	 * and BUG when futex_unlock_pi() interleaves with this.
	 *
	 * Therefore acquire wait_lock while holding hb->lock, but drop the
	 * latter before calling __rt_mutex_start_proxy_lock(). ...
	 */
	raw_spin_lock_irq(&q.pi_state->pi_mutex.wait_lock);
	spin_unlock(q.lock_ptr);
	ret = __rt_mutex_start_proxy_lock(&q.pi_state->pi_mutex, &rt_waiter, current, &wake_q);

这是 RT 改造"牵一发动全身"的典型：futex 哈希桶锁 hb->lock 是 spinlock_t，RT 下它本身也是 rtmutex。一个任务同一时刻只能阻塞在一个 rtmutex 上（只有一个 pi_blocked_on），所以 futex 代码必须小心地把"持 hb 锁"和"阻塞在 pi_mutex 上"错开，用 wait_lock 做交接。

调用栈：ARM64 上 pthread_mutex_lock 阻塞

条件变量与 requeue PI

pthread_cond_wait() 配合 PI mutex 时有一个经典难题：被 broadcast 唤醒的一群线程都要重新拿 mutex，如果全部唤醒，会形成惊群并让 PI 失效。FUTEX_WAIT_REQUEUE_PI / FUTEX_CMP_REQUEUE_PI 让内核把等待者从条件变量的 futex 直接迁移到 PI mutex 的 rtmutex 等待树上，只唤醒一个。

RT 下这里又多一层麻烦：kernel/futex/requeue.c 开头的注释说，被 requeue 的任务可能正因超时/信号醒来，要去拿 hb->lock（RT 下是 rtmutex），而它同时可能正被代理地挂在 PI rtmutex 上——"As a task can only be blocked on ONE rtmutex at a time, the proxy lock blocking and the hash bucket lock blocking would collide and corrupt state."解决办法是一个原子状态机 Q_REQUEUE_PI_{NONE, IGNORE, IN_PROGRESS, WAIT, DONE, LOCKED}，让等待方和 requeue 方通过 cmpxchg 协商谁先谁后（kernel/futex/requeue.c:51）。

4.7PI 的边界

4.8变化与代价

• 收益：反转被限制在"一条阻塞链上各临界区之和"，且这些临界区都以最高等待者的优先级运行；无争用时开销为零（cmpxchg 快路径）。
• 代价 1：争用路径重——每次阻塞都要维护两棵红黑树、可能修改调度类、可能链遍历；去 boost 也要走一遍 rt_mutex_setprio()。
• 代价 2：被提升的任务会"带着高优先级"执行临界区，挤占了其它 RT 任务的 CPU 时间——这是 PI 的设计本意，但做容量分析时要计入。
• 代价 3：PI 让 RT 调度行为更难推理。实践中最好的锁仍然是"不争用的锁"：缩小临界区、per-CPU 化、无锁结构。

4.9本章小结

下一章看另一半改造：把硬中断 handler 搬进线程，让它们也成为能被 PI 和调度器管理的普通任务。