Linux rt_mutex实时互斥锁优先级继承与pi链

Linux rt_mutex实时互斥锁优先级继承与pi链

rt_mutex是Linux内核专为实时任务设计的互斥锁,其核心机制是优先级继承(Priority Inheritance, PI),用于解决经典的无优先级继承互斥锁导致的优先级反转问题.

一、优先级反转问题

当低优先级任务持有锁,中优先级任务抢占CPU,高优先级任务等待锁时:

```c
/* 优先级: T_HIGH(99) > T_MID(50) > T_LOW(1) */

/* T_LOW持有锁L */
/* T_HIGH试图获取锁L -> 被阻塞 */
/* T_MID抢占T_LOW -> T_LOW无法释放锁L */
/* T_HIGH无限期等待 = 优先级反转 */
```

rt_mutex通过优先级继承解决:当T_HIGH被T_LOW持有的锁阻塞时,T_LOW临时继承T_HIGH的优先级(99),从而不会被T_MID抢占,可以尽快释放锁.

二、rt_mutex数据结构

```c
struct rt_mutex {
raw_spinlock_t wait_lock; /* 保护pi_chain的自旋锁 */
struct rb_root_cached waiters; /* 红黑树组织等待者,按优先级排序 */
struct task_struct *owner; /* 当前所有者, NULL表示未锁定 */
/* 低2位用作标志 */
};

/* rt_mutex的等待者 */
struct rt_mutex_waiter {
struct rb_node tree_entry; /* 在rt_mutex等待树中的节点 */
struct rb_node pi_tree_entry;/* 在pi链(任务pi_waiters)中的节点 */
struct task_struct *task; /* 等待任务 */
struct rt_mutex *lock; /* 等待的锁 */
int prio; /* 等待者的优先级 */
};

/* 任务结构体中的PI相关字段 */
struct task_struct {
...
struct rt_mutex_waiter *pi_blocked_on; /* 当前任务被哪个waiter阻塞 */
struct rb_root_cached pi_waiters; /* 该任务阻塞了哪些waiter(按优先级) */
int prio; /* 当前动态优先级(受PI影响) */
int normal_prio; /* 正常优先级(不受PI影响) */
...
};
```

三、pi_chain(优先级继承链)

pi链是被阻塞任务构成的树状结构.Task A等待锁L1(被Task B持有),Task B等待锁L2(被Task C持有),构成A->B->C的pi链.

```c
/* pi链的建立过程 */
static int task_blocks_on_rt_mutex(struct rt_mutex *lock,
struct rt_mutex_waiter *waiter,
struct task_struct *task)
{
struct task_struct *owner = rt_mutex_owner(lock);
struct rt_mutex_waiter *top_waiter;

/* 将waiter加入锁的等待红黑树 */
rt_mutex_enqueue(lock, waiter);

/* 将waiter加入owner的pi_waiters红黑树 */
rt_mutex_enqueue_pi(owner, waiter);

/* 获取owner的pi_waiters中优先级最高的waiter */
top_waiter = rt_mutex_top_waiter(lock);

/* 递归调整pi链上所有任务的优先级 */
rt_mutex_adjust_prio_chain(owner, top_waiter, lock, waiter, task);

/* 递归调用: 如果owner也被其他锁阻塞, 继续向上调整 */
if (rt_mutex_owner(owner->pi_blocked_on->lock))
task_blocks_on_rt_mutex(owner->pi_blocked_on->lock,
owner->pi_blocked_on,
owner);
}
```

四、优先级传播的核心算法

rt_mutex_adjust_prio_chain是pi链中最复杂的函数:

```c
static int rt_mutex_adjust_prio_chain(struct task_struct *task,
struct rt_mutex_waiter *orig_waiter,
struct task_struct *top_task)
{
struct rt_mutex *lock;
struct rt_mutex_waiter *waiter, *top_waiter;
int depth = 0;

retry:
depth++;

/* 获取任务当前持有的锁 */
lock = task->pi_blocked_on->lock;

/* 计算该锁上的最高优先级等待者 */
top_waiter = rt_mutex_top_waiter(lock);

/* 如果task的优先级已经适配, 无需调整 */
if (task->prio == min(top_waiter->prio, task->normal_prio))
goto out_unlock_pi;

/* 调整task的优先级到需要继承的值 */
task->prio = min(top_waiter->prio, task->normal_prio);

/* 如果task也被阻塞,递归向pi链上游调整 */
if (rt_mutex_owner(lock)) {
struct task_struct *next_owner = rt_mutex_owner(lock);

/* 递归调用, depth限制避免无限循环 */
raw_spin_lock(&next_owner->pi_lock);
waiter = next_owner->pi_blocked_on;
if (waiter) {
raw_spin_unlock(&lock->wait_lock);
/* 向链上游推进 */
rt_mutex_adjust_prio_chain(next_owner, waiter, top_task);
raw_spin_lock(&lock->wait_lock);
}
raw_spin_unlock(&next_owner->pi_lock);
}

out_unlock_pi:
return depth;
}
```

五、锁获取和释放

锁获取:

```c
static int rt_mutex_fastlock(struct rt_mutex *lock, int state)
{
if (likely(rt_mutex_cmpxchg_acquire(lock, NULL, current)))
return 0; /* 快路径: 无竞争, 直接获取 */

return rt_mutex_slowlock(lock, state); /* 慢路径: 进入pi链 */
}

static noinline int rt_mutex_slowlock(struct rt_mutex *lock, int state)
{
struct rt_mutex_waiter waiter;
int ret = 0;

/* 初始化waiter并加入等待树 */
rt_mutex_init_waiter(&waiter, false);
waiter.task = current;

raw_spin_lock(&lock->wait_lock);

/* 加入pi链 */
if (unlikely(!rt_mutex_owner(lock))) {
/* 锁已释放, 获取锁 */
rt_mutex_set_owner(lock, current);
raw_spin_unlock(&lock->wait_lock);
return 0;
}

/* 加入等待队列并建立pi链 */
task_blocks_on_rt_mutex(lock, &waiter, current);

/* 循环直到锁可用 */
for (;;) {
set_current_state(state);

if (rt_mutex_owner(lock) == current) {
/* 成功获取 */
break;
}

/* 睡眠 */
raw_spin_unlock(&lock->wait_lock);
schedule();
raw_spin_lock(&lock->wait_lock);
}

__set_current_state(TASK_RUNNING);
/* 从pi链中移除 */
task_unblocks_on_rt_mutex(lock, &waiter, current);

raw_spin_unlock(&lock->wait_lock);
return ret;
}
```

锁释放:

```c
static void __rt_mutex_unlock(struct rt_mutex *lock)
{
struct task_struct *owner = rt_mutex_owner(lock);

/* 清除所有者 */
rt_mutex_set_owner(lock, NULL);

/* 如果有等待者, 将锁传递给最高优先级的等待者 */
if (!rt_mutex_has_waiters(lock))
return;

/* 选择优先级最高的等待者 */
struct rt_mutex_waiter *top = rt_mutex_top_waiter(lock);
struct task_struct *next_task = top->task;

/* 唤醒下一个等待者 */
rt_mutex_set_owner(lock, next_task);
wake_up_process(next_task);
}
```

六、rt_mutex与RCU PI boosting

RT内核中RCU也依赖rt_mutex的PI机制解决RCU read-side的优先级反转:

```c
/* RCU read-side critical section中, 如果rcu_gp_task被低优先级任务阻塞 */
void rcu_boost(void)
{
struct task_struct *t;
struct rb_node *rbp;

/* 遍历所有等待RCU GP的任务 */
for_each_rcu_boost_task(t) {
/* 通过rt_mutex提升任务优先级 */
rt_mutex_lock(&t->rcu_boost_mutex);
/* t的优先级被提升, GP可以完成 */
rt_mutex_unlock(&t->rcu_boost_mutex);
}
}
```

七、PI链的复杂性

pi链的最坏情况复杂度:

- pi链深度限制: 内核通过MAX_RT_PRIO(100)间接限制,但实际pi链可能跨多个锁形成树状结构
- pi链调整的O(n)复杂度: 每次锁获取/释放都需要从叶子节点到根节点的优先级传播
- 锁嵌套导致pi链交织: Task A持有L1等待L2, Task B持有L2等待L1 -> 死锁检测通过pi链的循环检测实现

```c
/* 死锁检测: 在pi链中检查是否存在环 */
static int rt_mutex_check_deadlock(struct rt_mutex_waiter *waiter)
{
struct task_struct *task = waiter->task;
struct rt_mutex *lock = waiter->lock;

/* 沿着pi链向上遍历 */
while (task) {
if (task == current) {
/* 发现自身, 死锁! */
return -EDEADLK;
}
if (!task->pi_blocked_on)
break;
task = rt_mutex_owner(task->pi_blocked_on->lock);
}
return 0;
}
```

rt_mutex的优先级继承机制解决了实时系统中的优先级反转问题,pi链的递归传播保证了系统中所有受影响的锁相关的任务都能及时调整到正确的优先级.但pi链的维护带来了显著的复杂度,这是为实时性付出的必要代价.