前言
上一篇文章记录了 Yuan RTOS 中 IPC 基础层的实现。它主要解决的是“任务如何阻塞”和“任务如何被唤醒”的问题。
这篇文章继续记录两个基于 IPC 基础层实现的同步对象:信号量和互斥量。
完整代码请访问我的Github仓库,网址为:yuan-rtos。
值得注意是本文的示例代码在你阅读时可能已经进行过修改了,具体实现以仓库中代码为主。
信号量
信号量可以理解为一个带上限的计数器。它适合用于资源计数,也可以用于任务之间的简单同步。
Yuan RTOS 中信号量结构体如下:
typedef struct yr_semaphore_t {
yr_ipc_base_t ipc_base;
yr_uint16_t max_count;
yr_uint16_t current_count;
} yr_semaphore_t;
其中:
ipc_base:用于管理阻塞任务max_count:最大计数值current_count:当前计数值
初始化时需要设置最大计数值和初始计数值:
yr_err_t yr_semaphore_init( yr_semaphore_t* sem,
yr_uint16_t max_count,
yr_uint16_t init_count,
yr_uint32_t flag)
{
sem->max_count = max_count;
sem->current_count = init_count;
return yr_ipc_init( &sem->ipc_base, flag);
}
获取信号量
获取信号量的函数是:
yr_err_t yr_semaphore_take( yr_semaphore_t* sem,
yr_uint32_t wait_ticks)
如果当前计数大于 0,说明资源可用,直接把计数减 1 后返回:
if( sem->current_count > 0 ) {
sem->current_count--;
return YR_OK;
}
如果当前计数为 0,并且 wait_ticks == 0,说明调用者不愿意等待,那么直接返回失败。
如果允许等待,则当前任务会进入信号量的阻塞队列:
yr_task_set_msg( current_task,
(void*)&sem->ipc_base,
NULL,
YR_TASK_MR_IPC,
YR_TASK_MN_NONE);
yr_ipc_block_task( &sem->ipc_base, current_task);
如果 wait_ticks 不是 YR_WAIT_FOREVER,还会启动当前任务的定时器,用于处理等待超时:
if( wait_ticks != YR_WAIT_FOREVER ) {
yr_timer_set_ticks( ¤t_task->timer, wait_ticks);
yr_timer_start( ¤t_task->timer);
}
任务被唤醒后,需要检查等待结果:
if( current_task->msg_info.notify == YR_TASK_MN_WAIT_OK )
return YR_OK;
return YR_ERR;
也就是说,任务从阻塞中恢复运行不一定表示获取成功,也可能是超时或者信号量被删除。
释放信号量
释放信号量的函数是:
yr_err_t yr_semaphore_give( yr_semaphore_t* sem)
如果没有任务等待信号量,就直接增加 current_count:
sem->current_count++;
如果已经达到 max_count,则返回错误。正常使用时,信号量计数不应该超过上限。
如果有任务正在等待,那么资源不会简单地让 current_count++,而是直接交给阻塞队列中的第一个任务:
task = YR_LIST_ENTRY( sem->ipc_base.blocked_list.next,
yr_task_t,
list_node);
yr_list_delete_self( &task->list_node);
task->status = YR_TASK_STATUS_READY;
yr_task_set_msg( task,
NULL,
NULL,
YR_TASK_MR_NONE,
YR_TASK_MN_WAIT_OK);
yr_timer_stop(&task->timer);
yr_sched_insert_task( task);
这样可以避免资源刚释放又被其它任务抢走,语义上更接近“把资源交给等待者”。
信号量的中断接口
信号量提供了两个中断接口:
yr_semaphore_take_from_isr(...)
yr_semaphore_give_from_isr(...)
中断上下文下有一个重要原则:不能阻塞。
因此 take_from_isr 只能在当前计数大于 0 时成功,否则直接返回错误。
give_from_isr 可以释放信号量,并在有等待任务时唤醒一个任务。如果被唤醒任务优先级高于当前任务,则通过 need_switch 告诉外部需要进行任务切换。
互斥量
互斥量用于保护临界资源。它和二值信号量有一点像,但互斥量多了“所有者”的概念,也支持递归持有。
Yuan RTOS 中互斥量结构体如下:
typedef struct yr_mutex_t {
yr_ipc_base_t ipc_base;
yr_task_t *owner;
yr_uint8_t hold;
} yr_mutex_t;
其中:
ipc_base:用于管理等待锁的任务owner:当前持有锁的任务hold:递归持有深度
初始化时,锁没有持有者,递归深度为 0:
mutex->owner = NULL;
mutex->hold = 0;
yr_ipc_init( &mutex->ipc_base, flag);
获取互斥量
如果当前任务已经持有锁,那么再次获取锁只会增加递归深度:
if( mutex->owner == current_task ) {
mutex->hold++;
return YR_OK;
}
如果锁没有被任何任务持有,则当前任务直接成为锁的拥有者:
mutex->hold = 1;
mutex->owner = current_task;
current_task->hold_mutex_count++;
如果锁已经被其它任务持有,并且调用者不愿意等待,则直接返回失败。
如果允许等待,当前任务会被挂到互斥量的阻塞队列上,并可根据 wait_ticks 启动超时定时器。
释放互斥量
释放互斥量时,首先要检查当前任务是不是锁的拥有者:
if( mutex->owner != current_task ) {
return YR_ERR;
}
只有锁的拥有者才能释放锁。
如果锁被递归持有多次,则释放时只减少递归深度:
if( mutex->hold > 0 )
mutex->hold--;
只有当 hold 变为 0 时,锁才真正释放。
如果此时阻塞队列中有等待任务,则取出一个任务唤醒,并把锁直接交给它:
task = YR_LIST_ENTRY( mutex->ipc_base.blocked_list.next,
yr_task_t,
list_node);
yr_list_delete_self( &task->list_node);
task->status = YR_TASK_STATUS_READY;
yr_task_set_msg( task,
NULL,
NULL,
YR_TASK_MR_NONE,
YR_TASK_MN_WAIT_OK);
yr_timer_stop(&task->timer);
yr_sched_insert_task( task);
mutex->owner = task;
mutex->hold = 1;
task->hold_mutex_count++;
这里和信号量类似,资源会直接交给等待任务,而不是先释放成一个“无主资源”。
优先级继承
互斥量中最需要注意的是优先级反转问题。
假设低优先级任务持有锁,高优先级任务等待锁,如果此时中优先级任务一直运行,就可能导致高优先级任务迟迟拿不到锁。
为缓解这个问题,Yuan RTOS 在互斥量中实现了简单的优先级继承:
if( mutex->owner &&
current_task->current_priority <
mutex->owner->current_priority ) {
yr_task_ctrl_current( mutex->owner,
YR_TASK_CTL_SET_CUR_PRIORITY,
¤t_task->current_priority,
NULL);
}
当高优先级任务等待低优先级任务持有的锁时,会临时提升锁拥有者的当前优先级。
惰性继承
不过这里需要说明一点:Yuan RTOS 当前实现的是一种比较简单的惰性优先级继承。
所谓惰性,主要体现在它并不会精确记录“当前任务持有了哪些互斥量、每个互斥量上又分别有哪些任务在等待”。当一个高优先级任务等待某个互斥量时,内核只会把该互斥量的拥有者提升到等待者的优先级。
这解决了最常见的一类优先级反转问题,但它不是一个非常完美的实现。比如一个任务同时持有多个互斥量时,它因为其中一个互斥量被提升了优先级。之后它释放其中某个锁时,内核并没有重新扫描它仍然持有的所有锁,也没有重新计算“它当前还应该继承到哪个最高优先级”。
释放锁后,如果任务不再持有任何互斥量,并且当前优先级和初始优先级不同,则恢复初始优先级:
if( current_task->hold_mutex_count == 0 &&
current_task->current_priority !=
current_task->init_priority ) {
yr_task_ctrl_current( current_task,
YR_TASK_CTL_SET_CUR_PRIORITY,
¤t_task->init_priority,
&need_switch);
}
也就是说,当前实现采用的是比较保守的恢复策略:只要任务还持有任意互斥量,就暂时不恢复初始优先级;只有当它不再持有任何互斥量时,才恢复初始优先级。
这种方式的好处是实现简单,不需要为每个任务维护复杂的持锁关系,也不需要在每次释放锁时做大量扫描。对于一个轻量级 RTOS 来说,这能明显减少代码复杂度、RAM 占用和运行开销。
它的缺点也很明显:任务可能会比理论上需要的时间更久地保持一个被提升后的优先级。换句话说,它能避免高优先级任务被长期饿死,但不一定能做到“每一刻的任务优先级都完全精确”。
为什么最终选择这种方式
之所以当前选择这种惰性继承,主要是因为这个项目的目标是轻量化。
如果为了追求完美的优先级继承,就需要保存更多关系。例如任务需要知道自己当前持有哪些锁,锁也需要知道有哪些任务在等待它。释放任意一个锁时,还要重新计算该任务当前应该继承的最高优先级。
这当然更加准确,但代价也更高:
- 任务控制块需要增加额外链表或数组,RAM 占用会上升
- 互斥量控制块可能也要保存更多等待关系
take和give路径会变复杂,执行时间变长- 需要处理更多边界情况,比如嵌套持锁、多锁交叉等待、任务删除时的清理
- 代码复杂度上升后,调试难度也会明显增加
对于当前 Yuan RTOS 的定位来说,先实现一个简单、可理解、能覆盖常见场景的优先级继承更合适。因此这里选择了惰性继承,而不是一开始就把互斥量做得非常复杂。
后续可以怎么改进
如果后续希望把优先级继承做得更准确,可以考虑给任务控制块增加一个“已持有互斥量链表”。大概思路如下:
- 每个任务维护自己当前持有的互斥量链表
- 每个互斥量维护阻塞等待它的任务队列
- 任务释放某个互斥量后,扫描自己仍然持有的所有互斥量
- 从这些互斥量的等待队列中找到最高优先级等待者
- 将任务当前优先级恢复到
init_priority与所有继承优先级中的最高者
这样就能做到更接近“真正的优先级继承”。任务不会因为已经释放了相关互斥量而继续保持不必要的高优先级。
不过这个方案也有局限。首先,任务控制块要增加链表节点或链表头,互斥量也可能需要更多辅助字段,资源占用会增加。其次,每次释放锁都可能要扫描当前任务持有的锁,再检查锁上的等待队列,执行时间会变慢。最后,实现复杂度提高后,还要认真处理任务删除、互斥量删除、超时等待、优先级动态修改等情况。
所以这里本质上是一个取舍:当前实现牺牲了一部分精确性,换来了更小的资源占用、更短的关键路径和更容易维护的代码。对于一个轻量化 RTOS 的早期实现来说,这是可以接受的。
信号量和互斥量的区别
信号量和互斥量都能让任务阻塞,也都能唤醒等待任务,但它们的语义不同:
- 信号量是计数资源,可以由一个任务获取,也可以由另一个任务释放
- 互斥量有所有者,只有持有锁的任务才能释放锁
- 信号量适合资源计数和任务同步
- 互斥量适合保护临界区
- 互斥量支持递归持有和优先级继承
因此不要因为“二值信号量看起来像锁”就把两者完全混用。保护共享资源时,互斥量通常更合适。
总结
Yuan RTOS 中的信号量和互斥量都建立在 IPC 基础层之上:
- 信号量通过
current_count和max_count表示资源数量 - 信号量获取失败时可以阻塞等待,释放时可以唤醒等待任务
- 互斥量记录
owner和hold,支持递归持有 - 互斥量释放时会把锁直接交给等待任务
- 互斥量实现了简单的优先级继承,用于缓解优先级反转
- 有限等待时间通过任务定时器实现超时
这两个同步对象覆盖了嵌入式 RTOS 中非常常见的任务同步和临界区保护需求。