ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

手写RTOS信号量:从原理到代码实现

手写RTOS信号量:从原理到代码实现 1. 点灯进阶从“我全都要”到资源不够分写这个系列写到第8篇前面几篇我们手搓了任务控制块、就绪队列、调度器、时间片轮转、延时与休眠机制跑通了“两个任务轮流闪烁LED”这种基础demo。当时评论区不少人说这跟裸机写个延时翻转IO口有啥区别说实话区别确实不大因为那些例子里每个任务各干各的互不打扰压根没碰过“资源竞争”这个真正的内核难题。但点灯这件事往前深挖一步就全是坑。我举个实际场景你现在有A、B两个任务A任务负责在串口上打印调试日志B任务也偶尔打印一条错误信息。如果不做任何保护A打印到一半B抢占了CPU两条日志在同一行里搅成一团输出变成“Err[INF]or: time”这种鬼样子。再比如一个按键消抖任务和一个LED闪烁任务如果不同步按下去之后虚线得等好几个毫秒才能响应体验很怪。这些问题的本质就是多个任务同时要访问同一份资源——串口是资源、LED的全局变量是资源、按键按下这件事本身也是一个需要传递的信息。裸机时代我们靠关中断、靠简单标志位硬扛但在RTOS里任务什么时候切换是由调度器决定的你没法预测任务会在哪条指令上被打断所以必须有一套内核级别的机制让大家排队、等待、互相打招呼。信号量Semaphore就是干这个的。它是操作系统里历史最悠久的同步原语由Dijkstra在1965年提出到今天RTOS、Linux、各种嵌入式系统里依然是绝对主力。这套机制不复杂但理解得够不够深直接决定你能不能写出稳定可靠的多任务程序。这篇就把信号量从原理到实现一层层拆开最后落到我们自研内核的源码里。2. 信号量的内功心法一个计数器加一条等待队列2.1 为什么裸变量做不到先想一个最朴素的问题为什么不能用普通全局变量来计数比如int count1然后每个任务想用资源就if(count0){count--;}用完count单任务下没问题但多任务下几乎必挂。假设两个任务都要执行count--在C语言里这行代码编译完至少是三条汇编指令先从内存把count读到寄存器、寄存器减一、再把寄存器写回内存。如果任务A刚读完count值此时count1还没写回去调度器切换到任务BB也读到count1减完写回变成0。等切回AA把寄存器里的0写回count还是0。两次减一结果只减了一次。这就是教科书里说的“读-改-写”竞态条件。有人会说那我给这个加减操作加关中断不就行了没错这是单核MCU上最简单的解决办法我们前面的内核其实也是靠关中断来保护临界区的。但程序员不能每次操作都手动开关中断——太容易出错了而且你想保护的是“检查-获取-使用-释放”这一整段逻辑手动管理根本守不住。信号量就是把这件事封装成一个标准接口内核保证它的操作原子性你调用就行不用管底层细节。2.2 数据结构设计信号量的核心说白了就两样东西一个非负整数计数器加一条阻塞任务的等待队列。计数器表示当前还有多少资源可用当计数器为0时任务再想去拿资源就得停下来被挂到等待队列里直到有人释放资源、把它唤醒。这个设计可以类比成餐厅门口的叫号机。取号的时候看看手里的号要是当前空位还剩几个对应的就是计数器的值。没空位了你不是死等在那儿而是先找个凳子坐下来等广播叫你——这对应任务进入阻塞态被放进等待队列。餐厅每翻台一桌广播叫一个号被叫到的人起身进去对应take操作从等待队列里唤醒一个任务。在实现层面我给每个信号量定义这样一个结构体typedef struct { uint32_t count; // 资源计数 uint32_t max_count; // 计数上限防止无意义溢出 list_t wait_list; // 等待该信号量的任务队列 } sem_t;#undef 注意一个细节wait_list里挂的节点是任务控制块TCB本身还是单独的一个节点在FreeRTOS里TCB里直接内嵌了一个xGenericListItem用来挂进各种等待链表。我们自己的内核早期版本也这么做TCB里留一个ready_list_node用于就绪队列、一个wait_list_node用于各种阻塞等待。这样一个TCB同一时刻只能挂在一条链上逻辑清晰也方便判断任务当前状态。计数器用uint32_t而不是int进位溢出问题先不说关键是信号量计数不能为负——所有需要负数表达的语义都该用另一个叫“互斥量”的机制去处理我在下一节会讲为什么。API方面信号量的核心操作就四个sem_init(sem, max_count, init_count)初始化设定计数上限和初始值。sem_take(sem, timeout)获取信号量拿不到就阻塞等待支持超时。sem_give(sem)释放信号量唤醒等待队列里的一个任务。sem_delete(sem)删除信号量并唤醒所有还在等待的任务让它们返回错误码。take和give就是我们的P操作和V操作名字不同而已。所有RTOS都绕不开这一对。3. 三种信号量形态与选型二值、计数、互斥量3.1 二值信号量事件通知的利器二值信号量是计数器上限为1、初始值为0或1的信号量。它有两种用法第一种用法当互斥锁用。初始值为1表示资源可用谁take到谁就持有资源用完give。这种情况下它只有两个状态有/无。第二种用法是更常见的“事件通知”。初始化计数为0任务A负责等待某个事件具体表现就是阻塞在sem_take()上任务B在事件发生时调用sem_give()把计数变为1A被唤醒。注意这种场景下A拿到信号量后不需要再give回来因为它的语义不是“占有一个资源”而是“收到一个信号”。这两种用法的混用是新手最容易踩的坑。把二值信号量当锁用时如果任务A拿到锁后执行到一半崩了或迟迟不give其他任务会永久阻塞把二值信号量当事件通知时如果任务B连续give两次而A只take了一次第二次give的信号就积压在计数器里变成“事件提前发生”的假通知。所以用之前一定要想清楚我到底是锁还是信号3.2 计数信号量资源池管理计数信号量的计数器上限大于1比如N。它管理的是“同一时刻最多允许N个任务访问”的资源。最典型的场景是串口DMA收发缓冲池。假设你有4个DMA描述符4个任务要并发发送数据。计数器初始值为4每个任务take一个描述符用完give。如果同时来了5个任务第5个就得等着直到前面的任务释放描述符。另一个经典场景是“生产者-消费者”模型。生产者和消费者共享一个环形缓冲区用两个信号量管理empty信号量初始值为缓冲区大小表示还有几个空位full信号量初始值为0表示有几条数据。生产者每次放入数据前takeempty放入后givefull消费者takefull取出后giveempty。这套组合拳在无数嵌入式产品和Linux内核代码里都能看到。3.3 优先级翻转与互斥量的优先级继承二值信号量用来做互斥时有个著名的坑叫“优先级翻转”。场景是这样的任务H优先级最高任务M中等任务L最低。L先拿到了锁正在执行H想take同一把锁没拿到阻塞了此时调度器继续运行其他就绪任务发现M就绪了M抢占L的CPU——但问题是M的优先级并不比L高多少L没执行完就不会释放锁H就算优先级再高也只能干等。结果就是中等优先级任务M把高优先级任务H“饿死”了。真正的互斥量Mutex就是为了解决这个问题它具备“优先级继承”能力。当高优先级任务H因拿不到锁而阻塞时内核会把持有锁的低优先级任务L的优先级临时提升到与H相同甚至更高让L尽快被调度执行、尽快释放锁从而缩短H的等待时间。锁释放后L的优先级再恢复原样。我们手写内核时要做的不是一个二值信号量和优先级继承的叠加而是一个原生支持这一机制的独立类型。我在第6节会展开代码。3.4 三种形态选择对照表类型计数器上限典型用途核心特点注意事项二值信号量1事件通知、简单互斥状态量语义简单注意区分锁与事件语义计数信号量N资源池、生产者消费者允许多个任务同时访问小心溢出和配对关系互斥量1保护共享资源支持优先级继承只能由持有者释放不能用于中断注意互斥量的另一个限制是它必须在任务上下文中使用不能在中断服务函数里give——中断里没有“持有者”的概念该用二值信号量去通知任务。4. 从0实现信号量初始化、获取、释放的完整代码这一节直接看代码。我们延续前几篇的风格用标准C 关临界区的办法手写一套轻量级信号量。目标平台是cortex-M单核MCU编译器用gcc或armcc都行。4.1 实现文件整体结构我在工程里新建sema.c和sema.h结构如下// sema.h #ifndef __SEMA_H__ #define __SEMA_H__ #include kernel.h typedef struct { uint32_t count; uint32_t max_count; list_t wait_list; } sem_t; #define SEM_FAILED ((sem_t*)0) void sem_init(sem_t* sem, uint32_t max_count, uint32_t init_count); int sem_take(sem_t* sem, uint32_t timeout_ms); void sem_give_from_irq(sem_t* sem); int sem_take_from_irq(sem_t* sem, uint32_t timeout_ticks); /* 一般不实现 */ void sem_delete(sem_t* sem); #endif注意我在API里区分了sem_give和sem_give_from_irq。在中断里调用give和Task里调用give最大的区别是不能阻塞、不能切换到其他任务所以中断版本实现更简——只自增计数并唤醒等待队列中最高优先级任务设置一个“上下文切换请求”标志位真正的切换在退出中断的末尾由调度器完成。4.2 初始化设置上限是关键的防御性编程void sem_init(sem_t* sem, uint32_t max_count, uint32_t init_count) { if (sem NULL) return; /* 上限必须大于0 */ if (max_count 0) max_count 1; if (init_count max_count) init_count max_count; sem-max_count max_count; sem-count init_count; list_init(sem-wait_list); }这里有个很实际的问题为什么要设max_count如果不设上限一个bug导致give被调用一万次计数器涨到一万语义就彻底变了——两个任务本来只能同时用一把锁结果变成一万个都能进。防御性编程的力量就在这种细节里。4.3 take关中断、递减与阻塞一步都不能少take的核心逻辑分三步关中断快查有没有资源有就直接拿走没有就登记当前任务到wait_list然后触发调度器切换到其他任务。这里有个容易被忽视的细节处理完“资源不足”的逻辑后必须重新打开中断否则整个系统就死了。int sem_take(sem_t* sem, uint32_t timeout_ms) { uint32_t saved_irq; if (sem NULL || current_task NULL) return -1; saved_irq disable_irq(); if (sem-count 0) { sem-count--; enable_irq(saved_irq); return 0; } /* 资源不足把当前任务挂到等待队列 */ current_task-state TASK_BLOCKED; current_task-blocked_on sem; list_add_tail(sem-wait_list, current_task-wait_list_node); if (timeout_ms 0) { /* 设置超时控制块进入延时队列 */ task_delay_set(current_task, timeout_ms); } enable_irq(saved_irq); /* 触发一次调度切换到其他就绪任务 */ schedule(); /* 被唤醒后回到这里 */ if (current_task-blocked_on ! NULL) { /* 说明被超时唤醒了还没拿到信号量 */ return -2; } return 0; }被唤醒后为什么还要检查blocked_on因为任务的唤醒可能来自两种途径一是别人give了这时持有者会清掉task的blocked_on二是超时触发了内核在超时处理里把任务移出wait_list但blocked_on还留着。如果这里不判断任务以为拿到信号量了但实际上没有资源计数器也没变这就是经典的“假成功”。对了我这里的disable_irq/enable_irq在原子里是用CPSID/CPSIE汇编实现的也可以用__disable_irq()/__enable_irq()替代。但注意中断状态要保存——如果调用take之前中断是关闭的返回后要恢复原状否则可能破坏调用者的关中断状态。4.4 give最少操作但要小心唤醒哪个任务void sem_give(sem_t* sem) { uint32_t saved_irq; if (sem NULL) return; saved_irq disable_irq(); if (!list_is_empty(sem-wait_list)) { /* 有任务等待不增加计数直接把最高优先级任务唤醒 */ tcb_t* task list_first_entry(sem-wait_list, tcb_t, wait_list_node); list_remove(task-wait_list_node); task-state TASK_READY; task-blocked_on NULL; /* 取消超时计时 */ task_delay_cancel(task); schedule(); } else { /* 没有任务等待计数值加一最多加满 */ if (sem-count sem-max_count) { sem-count; } } enable_irq(saved_irq); }这里有个值得反复琢磨的点为什么有任务等待时give不增加count因为计数值不变等待队列里任务被take走了信号量但计数本来就该是0否则它不会等待被唤醒的任务带着这个“名义上的0”继续执行。如果你在give里既把count又唤醒一个任务那信号量等于同一份资源分给了两个人肯定出问题。唤醒哪个任务也有讲究。我在代码里用list_first_entry这是一个链表的第一个节点。我建议等待队列按优先级排序插入这样每次唤醒都是当前等待队列里最高优先级的任务效率最优也够公平。如果按FIFO顺序插入会出现低优先级任务长期霸占信号量、高优先级任务饿死的情况。4.5 中断版givevoid sem_give_from_irq(sem_t* sem) { uint32_t saved_irq; if (sem NULL) return; saved_irq disable_irq(); if (!list_is_empty(sem-wait_list)) { tcb_t* task list_first_entry(sem-wait_list, tcb_t, wait_list_node); list_remove(task-wait_list_node); task-state TASK_READY; task-blocked_on NULL; task_delay_cancel(task); /* 不在中断里直接调度只是标记一下 */ irq_pending_switch 1; } else { if (sem-count sem-max_count) { sem-count; } } enable_irq(saved_irq); }中断版和任务版的区别就在最后一个schedule()中断退出时PendSV会检查irq_pending_switch只有置位了才做上下文切换。这个细节保证了中断里面不调用任何可能阻塞的函数也就是“零延迟中断响应”。5. 优先级反转实战一个bug演示的调度危机5.1 复现场景与三个任务的血案为了验证我们内核里互斥量优先级继承的效果我在开发板上设计了这样一个实验三个任务优先级分别是高、中、低共享一把互斥锁用二值信号量模拟也完全一样。低优先级任务L先take锁执行一段比较长的计算我printf模拟500ms然后release锁。高优先级任务H延时100ms后尝试take锁拿不到就阻塞。中优先级任务M延时150ms后进入一个死循环持续占用CPU。结果不用跑都能猜到H到了100ms要拿锁锁在L手上阻塞。150ms时M就绪M优先级高于L立刻抢占L的CPU。L迟迟没执行完锁永远不释放H就一直等——这就是优先级反转高优先级任务H被中优先级任务M“间接阻塞”了。实测日志[L] take lock, start working... [H] try lock... blocked! [M] running, occupy cpu... [L] ... still working (preempted by M) [M] running, occupy cpu... ... (一直这样循环)如果不加优先级继承H可能要等M退出循环才能拿到锁而M的延时循环很可能是个bug导致的无限循环直接让系统卡死。真实产品中这种问题会表现为“看门狗复位”、“外设超时”等诡异现象排查起来非常痛苦。5.2 互斥锁的优先级继承实现解决思路前面说过了H阻塞在锁上时把持有者L的优先级临时提升到H的优先级。实现方式不复杂主要是在take失败和give释放时做两次优先级调整。先看互斥锁结构体typedef struct { tcb_t* owner; // 当前持有者 uint32_t orig_prio; // 持有者原始优先级 list_t wait_list; } mutex_t;take逻辑中当mutex被占用、任务需要阻塞时先检查持有者owner的优先级如果低于当前任务就立刻提升int mutex_take(mutex_t* mutex, uint32_t timeout_ms) { saved_irq disable_irq(); if (mutex-owner NULL) { mutex-owner current_task; mutex-orig_prio 0; /* 暂时无意义 */ enable_irq(saved_irq); return 0; } /* 锁被占用 */ if (mutex-owner current_task) { /* 重复获取直接返回成功还是死锁看你的设计 */ enable_irq(saved_irq); return -3; } /* 优先级继承核心把持有者的优先级提升到当前任务的优先级 */ if (current_task-priority mutex-owner-priority) { /* 数字越小优先级越高 */ mutex-save_owner_prio mutex-owner-priority; mutex-owner-priority current_task-priority; /* 重新调整持有者在就绪队列中的位置 */ reschedule_task(mutex-owner); } current_task-state TASK_BLOCKED; list_add_tail(mutex-wait_list, current_task-wait_list_node); enable_irq(saved_irq); schedule(); return 0; }give时就要把优先级恢复原状int mutex_give(mutex_t* mutex) { saved_irq disable_irq(); if (mutex-owner ! current_task) { enable_irq(saved_irq); return -1; /* 谁持有谁释放 */ } /* 恢复持有者优先级 */ if (mutex-owner-priority ! mutex-save_owner_prio) { mutex-owner-priority mutex-save_owner_prio; reschedule_task(mutex-owner); } /* 唤醒等待队列中最高优先级任务作为新的持有者 */ if (!list_is_empty(mutex-wait_list)) { tcb_t* next list_first_entry(mutex-wait_list, tcb_t, wait_list_node); list_remove(next-wait_list_node); mutex-owner next; mutex-save_owner_prio next-priority; next-state TASK_READY; next-blocked_on NULL; enable_irq(saved_irq); schedule(); return 0; } mutex-owner NULL; enable_irq(saved_irq); return 0; }还有个情况必须考虑如果锁在L手上H和更高级的J都在等L得同时继承两个高优先级任务里最高的那个。我上面的简单实现只存一个save_owner_prio遇到多级继承时会出问题。工业级RTOS一般维护一个优先级继承链任务被多个锁阻塞时持有者要继承所有等待任务中的最高优先级。我们作为学习型内核先实现单级继承把原理搞清楚多级继承可以在这个框架上扩展。加了优先级继承之后重现实验[L] take lock, start working... [H] try lock... blocked! [L] priority boosted to 0 (Hs priority) [L] finish, give lock, priority restore to 2 [H] get lock now, continue [M] running...日志显示L被临时提到极高优先级的瞬间M没法再抢占它L一气呵成跑完释放锁H因此只多等了很短的时间。这个机制在FreeRTOS的互斥锁里是这样在我们自己写的内核里也能做到跑通那一瞬间我还是挺有成就感的。6. 信号量使用中的经典坑与完整排查过程6.1 坑一二值信号量当事件通知连续两次give丢失事件一位朋友的项目里按键中断里给任务发信号需求是“按键按一下亮灯一下”。他写的是中断里sem_give(key_sem)任务里sem_take(key_sem)。表面看没错但实际有一个bug如果按键中断快速按了两次间隔小于任务被调度的时间第二个give来临时任务还没take过第一个信号量等待队列是空的计数器从0变成1第二个give进来时就只能把计数器置1而不是变成2。结果两次按键被当成一次用户感觉按键“吞了”。解决方案有两个一是改用队列消息队列而不是信号量每次按键压一个事件进队列任务逐个取出二是如果你必须用信号量那就维护一个独立的事件计数变量在中断里递增、任务里消耗信号量只用于唤醒。现实中第一种方案最常见所以这里也是我们下一篇文章“消息队列”的引子。6.2 坑二take/give不配对导致计数失衡在资源池场景里take和give必须严格按照“take几次就give几次”来配对。我在一个DMA收发程序里犯过这个错发送任务take了空闲DMA描述符DMA完成后在中断里give但我在任务里又加了一个“万一DMA失败就give一次”的分支结果两个give对应一个take计数虚增1。这个bug的可怕之处在于它不会立刻报错——计数器上限是4多出来的那个“幽灵资源”可能导致第5个任务同时拿到DMA描述符两个任务写同一个DMA缓冲数据互相踩踏查半天查不出来。排查这类问题建议在调试版代码里加两个全局统计变量sem_take_count和sem_give_count跑一段时间看两者差是否和当前计数值一致。不一致就说明有人多吃多占或少还。6.3 坑三在中断里调用sem_take信号量take是可能阻塞的中断上下文里调用等于自杀。我见过一个新手写UART接收中断每收到一字节就sem_take(rx_sem, 10)想实现“中断里等待某个事件”。这直接违反了RTOS最基本的使用规则程序跑起来就hard fault。正确做法中断里只用sem_give_from_irq向任务发信号任何take、任何阻塞、任何耗时操作都只能在任务里做。如果真的需要“中断里等一个标志”那应该用硬件信号量、事件标志组或者重新设计架构。6.4 坑四死锁——两个任务互相持有对方的锁两把锁、两个任务A持有锁1想拿锁2B持有锁2想拿锁1双方都在对方释放之前不放手系统僵住。排查这种问题最直接的办法是用调试器查看每个任务的blocked_on指针如果发现两个任务的blocked_on互相指向对方持有的锁基本就实锤了。预防死锁有几个经验多个锁必须按固定顺序获取所有任务都遵循同样的加锁顺序。尽量用sem_take的超时参数不要无限等待超时后主动回滚所有已持锁。把持锁时间压到最短锁里的操作精简到极致耗时操作移出临界区。6.5 坑五信号量计数作为标志位未考虑优先级反转很多人在任务里用计数器做“是否初始化完成”的标志比如if (sem_take(init_sem, 0) 0) { // 非阻塞尝试 // 已初始化 }这种用法本身没问题但它不是信号量的主要场景。如果这个标志经常被高优先级任务查询而初始化任务优先级很低存在和前面一样的优先级反转风险。更轻量的做法是直接用原子标志位或事件标志组没有挂起/唤醒的额外开销。7. 实测验证在开发板上跑通信号量点灯理论说了一大堆最终还是要落到开发板上。我用一块常见的STM32F103板子两个LED两个按键验证三种信号量场景。7.1 场景一二值信号量实现按键通知任务1按键扫描任务每10ms读一次按键检测到按下就sem_give(key_sem)。任务2LED控制任务阻塞在sem_take(key_sem, timeout1000)上一收到信号就翻转LED。结果按键每次按下LED都翻转。我特意快速连按十几次没有丢失因为按键中断或扫描里我没有用信号量直接做事件计数而是通过一个软件计数器和信号量组合实现。这也验证了我前面说的“事件计数需要配合队列或独立计数器”的结论。7.2 场景二计数信号量限制并发访问用一个全局数组模拟“共享缓冲区”把它包在信号量保护里。三个任务同时往缓冲区写数据计数信号量上限为1这里等价于二值信号量做互斥。串口打印出来的缓冲区数据没有出现交错的脏数据说明保护有效。7.3 场景三互斥锁优先级继承的实际效果我在5.1已经演示了。这里只补充一个测量数据不加优先级继承时H从尝试取锁到真正拿到锁的耗时是1200ms加了继承之后这个值降到80ms——90%以上的等待时间被消灭了。这个对比很直观值得在自己的板子上跑一跑你会真切感受到调度器设计一个看似微小的改动对实时性的影响有多大。8. 写在最后从手搓到理解差的不是代码量而是问题视角回顾这一篇的内容信号量在RTOS里的地位相当于“人与人之间的握手”——不只管资源使用还管事件通知和任务协作。你如果只停留在“会用xSemaphoreTake和xSemaphoreGive”的层面遇到优先级反转、事件丢失、死锁这些问题会毫无头绪但当你自己从数据结构开始写一遍亲手把任务塞进等待队列、再从等待队列里拽出来很多看似玄学的坑会变得无比清晰。我在调试优先级继承那两天最大的感受是每个API接口背后都藏着设计者对并发场景的假设。比如give里“有任务等待时不自增计数”这个决定如果没有自己实现一遍很难注意到差别注意到之后你就明白为什么生产者和消费者必须用两个信号量而不是一个。下一步我打算把消息队列加进来这玩意儿可以和信号量配合出很多花样。到时候继续用点灯的例子把“按键产生事件队列、多个任务消费”这个最经典的场景完整跑一遍。录这个系列的初衷就是希望每个读者都能从“复制代码”进阶到“敢自己改调度器”。信号量这块拧开了后面再看队列、互斥量、事件标志组会轻松很多。
返回列表