
RTOS设计与开发(9):互斥锁与优先级继承,把同步真正接到调度器里
RTOS设计与开发(9):互斥锁与优先级继承,把同步真正接到调度器里
信号量和消息队列打通之后,RTOS 的对象层已经能做“同步”和“通信”了,但还缺最关键的一块:互斥锁。
互斥锁和信号量最大的区别,从来不只是 API 名字不同,而是它自带 owner 语义。谁拿到锁,谁才能解锁;谁阻塞在锁上,谁的优先级变化还可能继续影响持锁者。这意味着 mutex 不能只靠对象层自己玩,它必须真正接进调度器。
这一版代码做成了一个很典型的 non-recursive mutex,并把优先级继承也一起接上了。
先看对象模型:为什么 mutex 一定要记录 owner
当前互斥锁对象定义如下:
typedef struct os_mutex {
uint32_t magic;
tcb_t *owner;
list_t wait_list;
list_node_t owner_node;
} os_mutex_t;
和信号量相比,这里多出来的两个东西特别关键:
ownerowner_node
owner 负责表达“当前这把锁属于谁”,这决定了 mutex 能实现“只有 owner 才能 unlock”的语义。
owner_node 更有意思。它不是给 mutex 自己排队用的,而是为了把“这把 mutex”挂进 owner->owned_mutex_list。也就是说,当前 RTOS 已经开始显式维护“一个任务手上持有哪些锁”这件事了。
这是优先级继承能成立的前提。
为什么这版 mutex 明确做成 non-recursive
os_mutex_lock() 里有一个很明确的限制:
owner = mutex->owner;
if (owner == current_task)
{
os_port_exit_critical(primask);
return OS_STATUS_RECURSIVE_LOCK;
}
这段代码说明当前实现故意不做 recursive mutex。
为什么这是个合理选择?
因为递归锁一旦引入,就必须同时处理至少三件事:
- 重入计数。
- 最后一层 unlock 才真正释放 ownership。
- 递归持锁期间的优先级继承如何与计数协同。
对于一个正在稳步长大的 RTOS,先把 non-recursive mutex 做稳,比一开始就把递归语义也一起带进来更靠谱。
当前实现通过显式返回 OS_STATUS_RECURSIVE_LOCK,把这个边界讲得非常清楚。
为什么 ownership 要挂进 owner 的持锁链表
ownership 建立时,不只是简单写一个 mutex->owner = current_task,而是先挂链:
static os_status_t os_mutex_owner_link_locked(os_mutex_t *mutex, tcb_t *owner)
{
if ((os_mutex_is_valid(mutex) == 0U) || (owner == NULL))
{
return OS_STATUS_INVALID_STATE;
}
if ((mutex->owner != NULL) || (mutex->owner_node.owner != NULL))
{
return OS_STATUS_INVALID_STATE;
}
if (list_insert_tail(&owner->owned_mutex_list, &mutex->owner_node) == 0U)
{
return OS_STATUS_INSERT_FAILED;
}
mutex->owner = owner;
return OS_STATUS_OK;
}
这样做的意义,不在于“方便查一下我现在持有几把锁”,而在于后面能基于这条链表重算 effective priority。
换句话说,这版 mutex 的设计不是“阻塞时临时 raise 一下 owner 优先级”这么简单,而是已经把优先级继承做成了一套可回算的状态模型。
lock 的慢路径为什么要同时做挂链和继承
快速路径很简单:没 owner,直接拿锁。
真正体现设计取舍的是慢路径:
status = task_wait_list_insert_priority_ordered(&mutex->wait_list, current_task);
if (status != OS_STATUS_OK)
{
os_port_exit_critical(primask);
return status;
}
status = task_priority_inheritance_raise_locked(owner, current_task->priority);
if (status != OS_STATUS_OK)
{
task_wait_list_remove_task(&mutex->wait_list, current_task);
os_port_exit_critical(primask);
return status;
}
status = task_block_current(mutex, timeout_ticks, os_mutex_wait_cleanup_locked);
这里顺序非常重要:
- 先把 waiter 挂进 mutex 的等待链表。
- 再根据 waiter 的优先级提升 owner。
- 最后把当前任务真正阻塞起来。
这个顺序的核心目的是:一旦 waiter 成为“合法等待者”,owner 的生效优先级就必须立刻反映这件事。否则系统虽然知道有人在等锁,但调度器还在按旧优先级看 owner,就会留下优先级反转的窗口。
清理回调为什么是这版 mutex 的关键补丁
这版 mutex 还有一个很容易被忽略、但实际上非常关键的设计:把 cleanup hook 接进了任务层等待框架。
static void os_mutex_wait_cleanup_locked(tcb_t *task)
{
os_mutex_t *mutex = NULL;
if ((task == NULL) || (task->wait_obj == NULL))
{
return;
}
mutex = (os_mutex_t *)task->wait_obj;
if (os_mutex_is_valid(mutex) == 0U)
{
return;
}
if (mutex->owner != NULL)
{
(void)task_priority_inheritance_refresh_locked(mutex->owner);
}
}
为什么这个 hook 必须存在?
因为 waiter 不一定只会通过 unlock 被唤醒。它也可能:
- timeout 离开等待队列。
- 因删除而离开等待队列。
如果这两条路径不顺手刷新 owner 的继承优先级,那么 owner 就可能继续背着一个已经不存在的“高优先级等待者”运行,导致 effective priority 长时间偏高。
这说明当前 RTOS 的对象层已经开始认真处理“对象等待之外的退出路径”了,而不是只写 happy path。
为什么优先级继承不是一次性 raise,而是“base + 全部持锁 waiter”重算
任务层真正负责继承重算的函数是这个:
os_status_t task_priority_inheritance_refresh_locked(tcb_t *task)
{
list_node_t *node = NULL;
os_mutex_t *mutex = NULL;
tcb_t *waiter = NULL;
uint8_t new_priority = 0U;
new_priority = task->base_priority;
node = task->owned_mutex_list.head;
while (node != NULL)
{
mutex = LIST_CONTAINER_OF(node, os_mutex_t, owner_node);
waiter = task_wait_list_peek_head_task(&mutex->wait_list);
if ((waiter != NULL) && (waiter->priority < new_priority))
{
new_priority = waiter->priority;
}
node = node->next;
}
return task_effective_priority_update_locked(task, new_priority);
}
这段代码揭示了当前实现的真正语义:
base_priority是任务原始优先级。priority是当前生效优先级。- 生效优先级不是“记住上一次 raise 到多少”,而是每次按“base priority + 全部持锁 mutex 的队头 waiter”重新计算。
这就避免了优先级继承最常见的一类 bug:只会升,不会降;或者一把锁释放后忘了看另外几把锁上是否还有更高优先级 waiter。
为什么 unlock 要先转交 ownership,再唤醒 waiter
这版 unlock 采用的是“直接 ownership 交接”:
task_wait_list_remove_task(&mutex->wait_list, waiter);
os_mutex_owner_unlink_locked(mutex);
status = os_mutex_owner_link_locked(mutex, waiter);
...
status = task_priority_inheritance_refresh_locked(waiter);
...
status = task_priority_inheritance_refresh_locked(old_owner);
...
status = task_unblock(waiter, TASK_WAIT_RESULT_OBJECT);
也就是说,当前实现不是先把 waiter 唤醒,然后让它自己再来竞争这把锁,而是在 unlock 的临界区里直接把 ownership 原子转交给它。
这有两个明显好处:
- waiter 被唤醒后,语义已经是“你拿到锁了”,不需要再重试一次
lock()。 - ownership 交接和优先级继承刷新都发生在同一原子区段里,不会出现锁已经释放但新 owner 还没成型的模糊窗口。
这也是为什么 os_mutex_lock() 在恢复运行后可以直接根据 wait_result 返回 OS_STATUS_OK,而不需要像队列那样进入重试循环。
链式继承为什么必须回传到“正在等待另一把 mutex 的任务”
任务层里有一段专门处理链式等待的逻辑:
if ((task->state == TASK_BLOCKED) && (waiting_mutex != NULL) && (waiting_mutex->magic == OS_MUTEX_MAGIC) && (waiting_mutex->owner != NULL))
{
return task_priority_inheritance_refresh_locked(waiting_mutex->owner);
}
这段代码的意思是:
如果一个任务自己正在等待另一把 mutex,那么它优先级的变化不能只停留在它本人这里,还必须继续向上游的 owner 传播。
这就是优先级继承最难的一块:不是单把锁本地 raise 一下就完事,而是等待链和持锁链交织起来之后,优先级变化可能需要继续传递。
当前这版代码已经把这个洞补上了,说明实现者不是只做了一个“看起来像继承”的 demo,而是真的把继承关系当成调度状态的一部分在维护。
小结
这一版 mutex 最有价值的地方,不只是“多了一个锁”,而是它第一次把同步对象和调度器真正接在了一起。
具体来说,它把这几件事同时做成了:
- owner 语义明确成立。
- non-recursive 边界明确成立。
- waiter 超时/删除离队时能刷新继承优先级。
- ownership 交接和 waiter 唤醒在同一临界区内完成。
- effective priority 由
base priority + 全部持锁 waiter统一回算。
这说明当前 RTOS 已经不是简单的“对象会阻塞线程”了,而是开始有能力维护跨对象、跨任务的调度约束。
下一篇会换个角度,讲这一阶段新引入的 panic/assert、栈填充、栈哨兵和 HardFault/UsageFault 收口,看看这个内核是怎么让错误暴露得更一致、更可诊断的。