RTOS开发学习记录(6):信号量与互斥量

成元 2026-04-22 { 教程, 笔记 } [ RTOS ]

前言

上一篇文章记录了 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;

其中:

初始化时需要设置最大计数值和初始计数值:

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( &current_task->timer, wait_ticks);
    yr_timer_start( &current_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;

其中:

初始化时,锁没有持有者,递归深度为 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,
                          &current_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,
                          &current_task->init_priority,
                          &need_switch);
}

也就是说,当前实现采用的是比较保守的恢复策略:只要任务还持有任意互斥量,就暂时不恢复初始优先级;只有当它不再持有任何互斥量时,才恢复初始优先级

这种方式的好处是实现简单,不需要为每个任务维护复杂的持锁关系,也不需要在每次释放锁时做大量扫描。对于一个轻量级 RTOS 来说,这能明显减少代码复杂度、RAM 占用和运行开销。

它的缺点也很明显:任务可能会比理论上需要的时间更久地保持一个被提升后的优先级。换句话说,它能避免高优先级任务被长期饿死,但不一定能做到“每一刻的任务优先级都完全精确”。

为什么最终选择这种方式

之所以当前选择这种惰性继承,主要是因为这个项目的目标是轻量化。

如果为了追求完美的优先级继承,就需要保存更多关系。例如任务需要知道自己当前持有哪些锁,锁也需要知道有哪些任务在等待它。释放任意一个锁时,还要重新计算该任务当前应该继承的最高优先级。

这当然更加准确,但代价也更高:

对于当前 Yuan RTOS 的定位来说,先实现一个简单、可理解、能覆盖常见场景的优先级继承更合适。因此这里选择了惰性继承,而不是一开始就把互斥量做得非常复杂。

后续可以怎么改进

如果后续希望把优先级继承做得更准确,可以考虑给任务控制块增加一个“已持有互斥量链表”。大概思路如下:

这样就能做到更接近“真正的优先级继承”。任务不会因为已经释放了相关互斥量而继续保持不必要的高优先级。

不过这个方案也有局限。首先,任务控制块要增加链表节点或链表头,互斥量也可能需要更多辅助字段,资源占用会增加。其次,每次释放锁都可能要扫描当前任务持有的锁,再检查锁上的等待队列,执行时间会变慢。最后,实现复杂度提高后,还要认真处理任务删除、互斥量删除、超时等待、优先级动态修改等情况。

所以这里本质上是一个取舍:当前实现牺牲了一部分精确性,换来了更小的资源占用、更短的关键路径和更容易维护的代码。对于一个轻量化 RTOS 的早期实现来说,这是可以接受的。

信号量和互斥量的区别

信号量和互斥量都能让任务阻塞,也都能唤醒等待任务,但它们的语义不同:

因此不要因为“二值信号量看起来像锁”就把两者完全混用。保护共享资源时,互斥量通常更合适。

总结

Yuan RTOS 中的信号量和互斥量都建立在 IPC 基础层之上:

这两个同步对象覆盖了嵌入式 RTOS 中非常常见的任务同步和临界区保护需求。