2026/9/10 4:16:11
FEATURED · 精选文章 
1. 为什么点灯点着点着就要跟“同步”较劲了先交代一下背景这个系列走到第8篇前面几篇其实已经把手搓RTOS最关键的一块骨头——任务调度——啃完了。系统里跑起来了几个任务延时队列、就绪队列、上下文切换都能转了LED也能以不同的频率闪烁了。按照很多教程的节奏到这一步好像就可以收工了。但你在实际测试中很快会发现一个现象任务一多程序就开始“不讲道理”了。举个例子我有两个任务任务A负责往缓冲区里写数据任务B负责从缓冲区里取数据。写的一方速度快、读的一方速度慢或者反过来只要没有协调机制缓冲区很快就会出现“空读”或者“覆盖写”的问题。还有更隐蔽的两个任务共享一个全局变量比如一个计数器任务A和任务B都会对它做自增操作。在裸机下一条count看起来是原子操作但在RTOS里这一条语句在汇编层面可能是“读内存—加一—写回”三步。任务A读到值、还没写回就被调度器切走了任务B进来也读到了旧值两边各自加一写回最终count只加了一次。这就是典型的竞态条件Race Condition。很多人第一次遇到这种情况时第一反应是“我的调度器是不是写错了”“上下文切换代码是不是有bug”。排查了一圈发现调度本身没有错错的是任务的思维方式——它们太“自我”了各干各的完全不理会别人。要想让多个任务在同一个系统里和睦相处就得引入一个老朋友信号量。信号量这个名字听起来有点学术但它的核心思想其实特别朴素就是一个“资源计数器”加上“等待队列”。你要用资源先拿信号量资源没了你就排队等着别人用完释放信号量把排队的人叫醒一个。说白了就是给任务之间立一个交通规则谁先绿灯谁走红灯亮了就老老实实等着。这篇咱们就延续“从手搓操作系统开始”的思路不调库、不抄现成代码从数据结构到内核API一步步把信号量实现出来再用几个点灯实验说明同步和互斥到底是怎么回事。2. 信号量的核心结构不就是一个计数器加一条等待链表吗2.1 先想清楚信号量到底要记录哪些东西在设计信号量之前最好先把它的“职责边界”画出来。信号量本身不是一个功能模块它是一套内核服务要支撑起两个基本语义同步Synchronization一个任务做完某件事通过信号量通知另一个任务“你可以继续了”。互斥Mutual Exclusion多个任务竞争同一个共享资源同一时刻只允许一个任务进临界区。为了实现这两种语义信号量对象最少要包含两个成员一个计数器count表示当前可用的资源数量一个等待队列wait_list用来挂起那些“想要资源但暂时拿不到”的任务。我在设计的时候还加了一个max_count用来限制计数器的上限。为什么要有上限因为信号量支持“释放”操作如果不限制最大值释放次数异常多的时候计数器会不断往上加甚至可以溢出成负数这种问题一旦发生排查起来非常痛苦。对于二值信号量max_count就是1对于计数信号量max_count就是资源池的大小。内核对象这块我建议先定义一个通用的结构体为后面实现互斥量、事件组这些东西留好扩展空间。在C语言里这一步可以这么设计typedef struct { uint32_t count; /* 当前可用资源数 */ uint32_t max_count; /* 信号量最大计数值 */ list_t wait_list; /* 等待该信号量的任务链表 */ } k_sem_t;wait_list的类型list_t就是前面几篇已经在用的双向链表不要在信号量里单独发明一套链表复用它就好。2.2 初始化二值信号量、计数信号量、互斥量怎么共用一套代码信号量的初始化函数需要支持三种使用方式二值信号量、计数信号量、互斥锁互斥锁本质上就是初始值为1的二值信号量但后面会看到它有额外的优先级继承需求所以我会单独区分开。初始化逻辑很简单把传进来的init_count赋值给countmax_count存起来然后把等待链表初始化成空链表void k_sem_init(k_sem_t *sem, uint32_t init_count, uint32_t max_count) { sem-count init_count; sem-max_count max_count; list_init(sem-wait_list); }这段代码没啥高深的但有一点要提醒init_count不能大于max_count。比如一个只允许3把钥匙的资源池初始却给了5把钥匙这就是明显的配置错误。我通常在k_sem_init里加一个断言调试阶段直接用assert拦下来。这种问题越早暴露越好千万别等到运行几小时后才出诡异故障。2.3 两个核心原语拿take和给give信号量的两大操作各大RTOS的叫法不太一样uC/OS叫OSSemPend和OSSemPostFreeRTOS叫xSemaphoreTake和xSemaphoreGive名字无所谓语义都一样。takeP操作等待信号量/* 返回值0成功-1超时-2参数错误 */ int k_sem_take(k_sem_t *sem, uint32_t timeout_ms) { if (sem NULL) { return -2; } uint32_t flags cpu_enter_critical(); if (sem-count 0) { /* 资源充足直接拿走一个 */ sem-count--; cpu_exit_critical(flags); return 0; } /* 资源不足但是允许不等待超时时间为0 */ if (timeout_ms 0) { cpu_exit_critical(flags); return -1; } /* 将当前任务挂入等待队列并触发调度 */ k_task_t *current get_current_task(); list_add_tail(sem-wait_list, current-wait_node); k_task_block(current, timeout_ms); /* 这个函数内部会切换到其他任务 */ cpu_exit_critical(flags); return 0; }注意几个细节第一cpu_enter_critical进入临界区是为了保护信号量自身的count和wait_list不被并发修改。进入临界区用的是关中断的方式这一点在单核MCU上是最直接有效的。第二当资源不足时当前任务不会被“立刻删除”而是被挂到wait_list上去同时把状态从就绪改为阻塞。第三任务被唤醒后函数是从k_task_block返回的返回后必须重新检查count是否还大于0——因为唤醒可能由超时触发而不是由信号量释放触发。完整的代码还应该判断唤醒原因这里先省略后面排查章节再讲。giveV操作释放信号量int k_sem_give(k_sem_t *sem) { if (sem NULL) { return -2; } uint32_t flags cpu_enter_critical(); if (!is_list_empty(sem-wait_list)) { /* 有人等着直接唤醒队首任务 */ k_task_t *task list_entry(sem-wait_list.next, k_task_t, wait_node); list_remove(task-wait_node); k_task_wakeup(task); } else if (sem-count sem-max_count) { /* 没人等就把资源数加回去 */ sem-count; } else { /* 已经到上限了这个是异常情况 */ } cpu_exit_critical(flags); return 0; }这里的处理顺序很关键先检查有没有等待者如果有直接把资源交给等待者而不是先count再让等待者下次take的时候拿。这样做的效率更高也避免了一次无谓的调度延迟。打个比方食堂打饭窗口空出来的那一刻如果后面已经排了队就直接让下一个人上前而不是把“空位”这个虚拟资源先存下来、再让人来取。3. 让任务“睡”与“醒”信号量背后的调度器配合逻辑3.1 任务为什么要“睡”而不是“死等”如果你在裸机开发里遇到“资源没准备好”最常见的做法是什么轮询。也就是while循环不断检查标志位。这种忙等待在单任务裸机里没问题但在RTOS里是灾难——它占着CPU不放其他任务全都卡死。所以在设计RTOS时当一个任务拿不到信号量时应该主动“睡觉”把CPU让给其他任务这就是阻塞Block状态。我之前在实现k_sem_take的时候最关键的调用是k_task_block(current, timeout_ms)。这个函数做的事情把它拆开看就是三步把当前任务从就绪队列中摘除设置当前任务的延时唤醒时间如果timeout是0则永远不唤醒调用调度器切换到下一个就绪任务。等调度器切回来的时候函数返回然后重新判断自己为什么被唤醒——是因为拿到了信号量还是因为超时。如果是超时k_sem_take要向调用者返回一个超时错误码调用者根据错误码决定是重试还是放弃操作。3.2 唤醒路径上最容易忽略的“优先级问题”k_sem_give里唤醒队首任务的那一行代码看起来简单但实现时有一个容易被忽视的细节被唤醒的任务必须重新插入就绪队列而插入就绪队列的位置不是随便放的。如果一个高优先级任务正在等待信号量而当前给出信号量的是一个低优先级任务那么唤醒高优先级任务后应该立刻触发一次任务调度让高优先级任务马上运行。否则低优先级任务可能会继续执行很久导致高优先级任务的延迟加大。我在实际实现中k_task_wakeup函数内部会在插入就绪队列之后检查如果被唤醒任务的优先级高于当前任务优先级就设置一个need_sched标志。这个标志在中断退出或者系统调用返回的时候会被检查从而触发一次抢占调度。这种“延迟调度”的做法是为了避免在临界区里直接切任务导致栈混乱。3.3 关中断的粒度越小越好写RTOS内核关中断是一把双刃剑。关的时间太长中断响应就会变差关的时间太短共享数据又得不到保护。信号量的take和give操作理想情况下只在“检查并修改count / wait_list”这几条指令期间关中断其余时间保持中断开启。但有一个点要特别注意在k_sem_take中从“检查count”到“把当前任务挂入等待队列并调用调度器”这段过程必须全程关中断。如果你中途开了中断一个中断服务程序里也可能调用k_sem_give那这个中断处理程序会看到当前任务还没挂上等待链表、就绪队列里也没有它整个等待关系就乱了。这是新手做信号量时最容易出的bug等任务挂到了队列上才允许中断来唤醒。4. 三大经典实验同步、互斥、计数信号量的实际验证4.1 实验一二值信号量做任务间同步——“先做完再通知”“点灯大师”的传统艺能在这里派上用场了。第一个实验我设计的是任务A负责翻转LED但它不能一上来就翻必须等任务B先完成一个耗时操作、通过信号量发出“我干完了”的通知之后才能开始翻。这种场景在现实项目里非常常见比如传感器数据采集任务和数据处理任务之间一个采完了另一个才能处理。实验代码骨架如下static k_sem_t sync_sem; void task_b_entry(void *arg) { /* 模拟耗时操作比如点亮另一个LED 500ms */ led_on(LED_GREEN); k_task_delay(500); /* 干完了发信号 */ k_sem_give(sync_sem); /* 然后这个任务休眠 */ k_task_delay(1000); } void task_a_entry(void *arg) { while (1) { /* 等信号量没有就阻塞在这 */ k_sem_take(sync_sem, 2000); /* 拿不到就继续等拿到就翻转LED一次 */ led_toggle(LED_RED); } }运行后的现象很直观红灯一开始一动不动等绿灯亮起并熄灭之后红灯才开始以200ms周期翻转。如果把timeout从2000改成0也就是无限等待那行为也类似但要注意在任务A里不能加任何自己的逻辑delay否则任务B得等它让出CPU才能运行。这个实验证明了二值信号量的同步语义。注意这里的信号量初始值设置了0意思是“一开始没有资源”任务A第一次take一定被阻塞。这种“初始为0”的二值信号量本质上是事件标志。4.2 实验二计数信号量管理共享资源池——多个资源、多把钥匙第二个实验用计数信号量来模拟“一个资源池同时最多有N个访问者”。我用它来管理板的串口打印权限系统里有一个串口同一时刻只允许一个任务打印否则打印内容会交错乱掉。如果想做得更有意思可以把资源池改成N个虚拟设备比如N个可用的USB端口或N个内存块每个任务用之前take用完give。static k_sem_t pool_sem; void pool_init(void) { k_sem_init(pool_sem, 3, 3); /* 资源池里有3个可用资源 */ } void worker_task_entry(void *arg) { int id (int)arg; while (1) { k_task_delay(id * 100); /* 申请资源 */ if (k_sem_take(pool_sem, 1000) 0) { /* 占用资源 */ dbg_print(Task %d enter pool\n, id); k_task_delay(300); dbg_print(Task %d leave pool\n, id); /* 释放资源 */ k_sem_give(pool_sem); } else { dbg_print(Task %d timeout\n, id); } } }运行三个worker任务观察打印内容你会看到“enter pool”和“leave pool”之间是成对出现的而且任意时刻最多只有三条“enter”记录没有被对应的“leave”抵消。这就是计数信号量最典型的应用。如果把sem_init的第三个参数改成1这个计数信号量就退化成了互斥锁同一时刻只允许一个任务占用资源。4.3 实验三互斥量保护临界区——共享变量的自增不再丢数据第三个实验直接复现开头提到的计数器问题。两个任务同时对同一个全局变量g_count做10000次自增。如果没有互斥保护最终的计数值一定小于20000而且每次跑的结果还不一样。加上互斥量之后结果稳定在20000。static k_sem_t mutex_sem; static uint32_t g_count; void inc_task_entry(void *arg) { int i; for (i 0; i 10000; i) { k_sem_take(mutex_sem, 1000); g_count; k_sem_give(mutex_sem); } }这里我要多说一句用二值信号量做互斥在功能上是可行的但工程上不推荐。二值信号量没有“所有权”的概念也就是说任务A可以give一个自己没有take过的信号量这会导致互斥保护名存实亡。真正的互斥量mutex应该要求只有持有者才能释放并且在任务被删除时自动释放它持有的锁。后续的章节里我会单独实现一个互斥量这次先用二值信号量顶一下但“互斥量必须单独设计”这个意识得留在脑子里。5. 优先级反转信号量实现中最容易踩的坑与对策5.1 一个反直觉的实验现象如果你照着上面的实验三跑一阵子大概率会觉得“信号量已经搞定一切了”。但别急我再加一个实验三个任务高优先级任务H中优先级任务M低优先级任务L。L先抢到互斥量进入临界区但还没执行完就被调度走了。这时候H就绪了H想拿互斥量拿不到被阻塞。然后M就绪了因为M的优先级高于L所以M抢占CPU运行。M运行完后L才继续。L执行完毕释放互斥量H才重新运行。这个过程的问题在于H明明优先级最高最后完成的时间却比L还晚。M这个“路人甲”任务反而抢在了H前面运行。这就是教科书上说的“优先级反转”Priority Inversion。“反转”不是说优先级数字变了而是任务的调度行为背离了优先级的设计意图——最高优先级的任务实际等待时间最长。5.2 对策一优先级继承解决优先级反转最常见的方法是优先级继承Priority Inheritance。核心思想是当一个低优先级任务持有互斥量、而一个高优先级任务正在等待这个互斥量时系统把低优先级任务的优先级临时提升到与高优先级任务相同或者至少提升到一定高度。这样M中优先级任务就没有机会抢占L了L能快速跑完临界区、释放互斥量然后恢复原来的优先级H才能尽早执行。优先级继承的实现需要在互斥量的结构体里额外记录三个信息当前持有者是谁、持有者的原始优先级、当前等待互斥量的最高优先级任务的优先级。当k_mutex_take发现锁被占用时就把持有者的优先级抬升当k_mutex_give释放锁时把持有者的优先级还原。typedef struct { k_task_t *owner; /* 当前持有者 */ uint8_t owner_prio; /* 持有者的原始优先级 */ uint8_t inherit_prio; /* 需要继承到的优先级 */ list_t wait_list; /* 等待该互斥量的任务链表 */ uint8_t locked; /* 是否被锁定 */ } k_mutex_t;注意这个地方有个细节继承的不是固定值而是要随着等待者优先级的增加而动态变化。比如L持有锁H和M同时在等但H优先级更高那L应该继承到H的优先级等H先被唤醒、拿到锁之后L的优先级可能又要降回M的优先级因为现在M是剩余等待者里优先级最高的了。实现这一步需要在每次take被阻塞的时候扫描wait_list里所有任务的优先级取最大值然后动态更新owner的优先级。5.3 对策二优先级天花板另一种方案是优先级天花板Priority Ceiling它规定任务在持有某互斥量期间直接运行在“所有可能使用这个互斥量的任务的最高优先级”之上。这个方案在实现上更简单系统只需要在创建互斥量时指定一个天花板优先级take的时候直接把当前任务抬到天花板give的时候再降回原级别。它的缺点是灵活性差——如果互斥量被很多不同优先级的任务使用天花板优先级会变得很高导致系统整体调度灵活性下降。我在实际项目中优先使用优先级继承方案只有在系统任务数量少、而且互斥量的使用场景可以预先确定时才考虑天花板。这个对照关系可以整理成一张表便于后面选型时参考方案实现难度系统开销适用场景优先级继承较高每次拖动都要调整任务数量多、优先级关系复杂的通用系统优先级天花板较低锁的临界区拉长任务结构固定、互斥量使用场景可预先分析的场景注意我们这里的实时性讨论都基于单核MCU和可抢占内核的前提。如果你是拿这个示例跑在PC或者别的环境上调度器的行为细节会不一样但信号量的语义和优先级反转问题是通用的。5.4 调试优先级继承时的“自锁”陷阱最后分享一个我在实现优先级继承时真实踩过的坑。第一次实现完成跑实验三一切正常但一旦把H、M、L三个任务组合在一起跑系统就在某个随机时刻卡死。查了很久之后发现问题出在k_mutex_give里面释放锁后要恢复owner的原始优先级但我调用的k_task_set_priority函数里又去拿了一遍互斥量……结果锁还没完全释放自己先把自己卡死了。所以这里面有一个重要的编码原则**内核内部函数之间的调用次序要格外小心尤其不能在内核API还没有完成状态更新前又去调用同一个API的其它变体。**我的做法是把k_task_set_priority拆成两个版本一个内部版本__k_task_set_priority假设调用者已经进入了临界区一个外部版本k_task_set_priority先关中断再调内部版本。这样每个内核服务都遵循“关中断—改状态—开中断”的规范自锁问题就彻底解决了。6. 信号量的边界情况与下一步扩展写到这里这篇的核心内容其实已经覆盖完整了从信号量的数据结构、take/give的实现到调度器的配套逻辑再到优先级继承的对策。但在收尾之前还有三个边界情况值得拿出来单独提醒一下——它们在实际项目中一定会遇到。超时返回的多义性。k_sem_take返回0时表示成功获取了信号量返回-1时表示超时。但是超时本身有两种原因一种是真的等满了时间还没等到资源另一种是时间虽然到了但中间被打断过比如被高优先级任务抢占实际等待时间远远大于传进去的参数。如果你的代码把timeout当作严格的最后期限这种多义性会出问题。解决方法是多返回一个“剩余等待时间”或者“是否真正超时”的标志这个细节可以根据实际应用决定要不要做。中断服务程序里的give。信号量的give操作经常在中断里调用用来唤醒一个任务去处理数据。这种情况需要考虑一个问题中断里调用give时当前执行的并不是任务上下文不能直接触发任务切换。正确的姿势是只做“把任务从等待队列摘除、插入就绪队列、设置need_sched标志”这几件事等中断退出时由汇编代码统一检查need_sched再决定要不要切任务。如果你在中断里直接调用调度器大概率会把栈搞崩。内存序的问题。这一条在Cortex-M等现代ARM核上尤其重要。信号量的count和wait_list被修改后要确保对其它核或者同一核上的中断上下文是立即可见的。如果你的编译器优化级别开得比较高或者目标平台是多核需要在进入临界区时加上合适的内存屏障。目前我只用了关中断来保证原子性在单核MCU上够用如果后面要把这套代码移植到多核平台这里得补上dmb之类的指令。下一步如果要继续扩展我建议按这个顺序走先实现真正的互斥量带所有权和优先级继承然后是事件标志组再后面可以试试消息队列和邮箱。这几个东西一旦有了一个迷你RTOS的基本IPC就齐活了。下一篇我打算直接从互斥量补全开始写把今天提到的优先级继承完整落地成代码。你如果已经照着这个系列板调到了第八篇不妨先把信号量移植到自己的开发板上跑一遍文章里的三个实验亲眼确认一下竞态问题的存在和解决过程这个过程比任何源码都更有说服力。
手搓RTOS：从零实现信号量与优先级反转对策
1. 为什么点灯点着点着就要跟“同步”较劲了先交代一下背景这个系列走到第8篇前面几篇其实已经把手搓RTOS最关键的一块骨头——任务调度——啃完了。系统里跑起来了几个任务延时队列、就绪队列、上下文切换都能转了LED也能以不同的频率闪烁了。按照很多教程的节奏到这一步好像就可以收工了。但你在实际测试中很快会发现一个现象任务一多程序就开始“不讲道理”了。举个例子我有两个任务任务A负责往缓冲区里写数据任务B负责从缓冲区里取数据。写的一方速度快、读的一方速度慢或者反过来只要没有协调机制缓冲区很快就会出现“空读”或者“覆盖写”的问题。还有更隐蔽的两个任务共享一个全局变量比如一个计数器任务A和任务B都会对它做自增操作。在裸机下一条count看起来是原子操作但在RTOS里这一条语句在汇编层面可能是“读内存—加一—写回”三步。任务A读到值、还没写回就被调度器切走了任务B进来也读到了旧值两边各自加一写回最终count只加了一次。这就是典型的竞态条件Race Condition。很多人第一次遇到这种情况时第一反应是“我的调度器是不是写错了”“上下文切换代码是不是有bug”。排查了一圈发现调度本身没有错错的是任务的思维方式——它们太“自我”了各干各的完全不理会别人。要想让多个任务在同一个系统里和睦相处就得引入一个老朋友信号量。信号量这个名字听起来有点学术但它的核心思想其实特别朴素就是一个“资源计数器”加上“等待队列”。你要用资源先拿信号量资源没了你就排队等着别人用完释放信号量把排队的人叫醒一个。说白了就是给任务之间立一个交通规则谁先绿灯谁走红灯亮了就老老实实等着。这篇咱们就延续“从手搓操作系统开始”的思路不调库、不抄现成代码从数据结构到内核API一步步把信号量实现出来再用几个点灯实验说明同步和互斥到底是怎么回事。2. 信号量的核心结构不就是一个计数器加一条等待链表吗2.1 先想清楚信号量到底要记录哪些东西在设计信号量之前最好先把它的“职责边界”画出来。信号量本身不是一个功能模块它是一套内核服务要支撑起两个基本语义同步Synchronization一个任务做完某件事通过信号量通知另一个任务“你可以继续了”。互斥Mutual Exclusion多个任务竞争同一个共享资源同一时刻只允许一个任务进临界区。为了实现这两种语义信号量对象最少要包含两个成员一个计数器count表示当前可用的资源数量一个等待队列wait_list用来挂起那些“想要资源但暂时拿不到”的任务。我在设计的时候还加了一个max_count用来限制计数器的上限。为什么要有上限因为信号量支持“释放”操作如果不限制最大值释放次数异常多的时候计数器会不断往上加甚至可以溢出成负数这种问题一旦发生排查起来非常痛苦。对于二值信号量max_count就是1对于计数信号量max_count就是资源池的大小。内核对象这块我建议先定义一个通用的结构体为后面实现互斥量、事件组这些东西留好扩展空间。在C语言里这一步可以这么设计typedef struct { uint32_t count; /* 当前可用资源数 */ uint32_t max_count; /* 信号量最大计数值 */ list_t wait_list; /* 等待该信号量的任务链表 */ } k_sem_t;wait_list的类型list_t就是前面几篇已经在用的双向链表不要在信号量里单独发明一套链表复用它就好。2.2 初始化二值信号量、计数信号量、互斥量怎么共用一套代码信号量的初始化函数需要支持三种使用方式二值信号量、计数信号量、互斥锁互斥锁本质上就是初始值为1的二值信号量但后面会看到它有额外的优先级继承需求所以我会单独区分开。初始化逻辑很简单把传进来的init_count赋值给countmax_count存起来然后把等待链表初始化成空链表void k_sem_init(k_sem_t *sem, uint32_t init_count, uint32_t max_count) { sem-count init_count; sem-max_count max_count; list_init(sem-wait_list); }这段代码没啥高深的但有一点要提醒init_count不能大于max_count。比如一个只允许3把钥匙的资源池初始却给了5把钥匙这就是明显的配置错误。我通常在k_sem_init里加一个断言调试阶段直接用assert拦下来。这种问题越早暴露越好千万别等到运行几小时后才出诡异故障。2.3 两个核心原语拿take和给give信号量的两大操作各大RTOS的叫法不太一样uC/OS叫OSSemPend和OSSemPostFreeRTOS叫xSemaphoreTake和xSemaphoreGive名字无所谓语义都一样。takeP操作等待信号量/* 返回值0成功-1超时-2参数错误 */ int k_sem_take(k_sem_t *sem, uint32_t timeout_ms) { if (sem NULL) { return -2; } uint32_t flags cpu_enter_critical(); if (sem-count 0) { /* 资源充足直接拿走一个 */ sem-count--; cpu_exit_critical(flags); return 0; } /* 资源不足但是允许不等待超时时间为0 */ if (timeout_ms 0) { cpu_exit_critical(flags); return -1; } /* 将当前任务挂入等待队列并触发调度 */ k_task_t *current get_current_task(); list_add_tail(sem-wait_list, current-wait_node); k_task_block(current, timeout_ms); /* 这个函数内部会切换到其他任务 */ cpu_exit_critical(flags); return 0; }注意几个细节第一cpu_enter_critical进入临界区是为了保护信号量自身的count和wait_list不被并发修改。进入临界区用的是关中断的方式这一点在单核MCU上是最直接有效的。第二当资源不足时当前任务不会被“立刻删除”而是被挂到wait_list上去同时把状态从就绪改为阻塞。第三任务被唤醒后函数是从k_task_block返回的返回后必须重新检查count是否还大于0——因为唤醒可能由超时触发而不是由信号量释放触发。完整的代码还应该判断唤醒原因这里先省略后面排查章节再讲。giveV操作释放信号量int k_sem_give(k_sem_t *sem) { if (sem NULL) { return -2; } uint32_t flags cpu_enter_critical(); if (!is_list_empty(sem-wait_list)) { /* 有人等着直接唤醒队首任务 */ k_task_t *task list_entry(sem-wait_list.next, k_task_t, wait_node); list_remove(task-wait_node); k_task_wakeup(task); } else if (sem-count sem-max_count) { /* 没人等就把资源数加回去 */ sem-count; } else { /* 已经到上限了这个是异常情况 */ } cpu_exit_critical(flags); return 0; }这里的处理顺序很关键先检查有没有等待者如果有直接把资源交给等待者而不是先count再让等待者下次take的时候拿。这样做的效率更高也避免了一次无谓的调度延迟。打个比方食堂打饭窗口空出来的那一刻如果后面已经排了队就直接让下一个人上前而不是把“空位”这个虚拟资源先存下来、再让人来取。3. 让任务“睡”与“醒”信号量背后的调度器配合逻辑3.1 任务为什么要“睡”而不是“死等”如果你在裸机开发里遇到“资源没准备好”最常见的做法是什么轮询。也就是while循环不断检查标志位。这种忙等待在单任务裸机里没问题但在RTOS里是灾难——它占着CPU不放其他任务全都卡死。所以在设计RTOS时当一个任务拿不到信号量时应该主动“睡觉”把CPU让给其他任务这就是阻塞Block状态。我之前在实现k_sem_take的时候最关键的调用是k_task_block(current, timeout_ms)。这个函数做的事情把它拆开看就是三步把当前任务从就绪队列中摘除设置当前任务的延时唤醒时间如果timeout是0则永远不唤醒调用调度器切换到下一个就绪任务。等调度器切回来的时候函数返回然后重新判断自己为什么被唤醒——是因为拿到了信号量还是因为超时。如果是超时k_sem_take要向调用者返回一个超时错误码调用者根据错误码决定是重试还是放弃操作。3.2 唤醒路径上最容易忽略的“优先级问题”k_sem_give里唤醒队首任务的那一行代码看起来简单但实现时有一个容易被忽视的细节被唤醒的任务必须重新插入就绪队列而插入就绪队列的位置不是随便放的。如果一个高优先级任务正在等待信号量而当前给出信号量的是一个低优先级任务那么唤醒高优先级任务后应该立刻触发一次任务调度让高优先级任务马上运行。否则低优先级任务可能会继续执行很久导致高优先级任务的延迟加大。我在实际实现中k_task_wakeup函数内部会在插入就绪队列之后检查如果被唤醒任务的优先级高于当前任务优先级就设置一个need_sched标志。这个标志在中断退出或者系统调用返回的时候会被检查从而触发一次抢占调度。这种“延迟调度”的做法是为了避免在临界区里直接切任务导致栈混乱。3.3 关中断的粒度越小越好写RTOS内核关中断是一把双刃剑。关的时间太长中断响应就会变差关的时间太短共享数据又得不到保护。信号量的take和give操作理想情况下只在“检查并修改count / wait_list”这几条指令期间关中断其余时间保持中断开启。但有一个点要特别注意在k_sem_take中从“检查count”到“把当前任务挂入等待队列并调用调度器”这段过程必须全程关中断。如果你中途开了中断一个中断服务程序里也可能调用k_sem_give那这个中断处理程序会看到当前任务还没挂上等待链表、就绪队列里也没有它整个等待关系就乱了。这是新手做信号量时最容易出的bug等任务挂到了队列上才允许中断来唤醒。4. 三大经典实验同步、互斥、计数信号量的实际验证4.1 实验一二值信号量做任务间同步——“先做完再通知”“点灯大师”的传统艺能在这里派上用场了。第一个实验我设计的是任务A负责翻转LED但它不能一上来就翻必须等任务B先完成一个耗时操作、通过信号量发出“我干完了”的通知之后才能开始翻。这种场景在现实项目里非常常见比如传感器数据采集任务和数据处理任务之间一个采完了另一个才能处理。实验代码骨架如下static k_sem_t sync_sem; void task_b_entry(void *arg) { /* 模拟耗时操作比如点亮另一个LED 500ms */ led_on(LED_GREEN); k_task_delay(500); /* 干完了发信号 */ k_sem_give(sync_sem); /* 然后这个任务休眠 */ k_task_delay(1000); } void task_a_entry(void *arg) { while (1) { /* 等信号量没有就阻塞在这 */ k_sem_take(sync_sem, 2000); /* 拿不到就继续等拿到就翻转LED一次 */ led_toggle(LED_RED); } }运行后的现象很直观红灯一开始一动不动等绿灯亮起并熄灭之后红灯才开始以200ms周期翻转。如果把timeout从2000改成0也就是无限等待那行为也类似但要注意在任务A里不能加任何自己的逻辑delay否则任务B得等它让出CPU才能运行。这个实验证明了二值信号量的同步语义。注意这里的信号量初始值设置了0意思是“一开始没有资源”任务A第一次take一定被阻塞。这种“初始为0”的二值信号量本质上是事件标志。4.2 实验二计数信号量管理共享资源池——多个资源、多把钥匙第二个实验用计数信号量来模拟“一个资源池同时最多有N个访问者”。我用它来管理板的串口打印权限系统里有一个串口同一时刻只允许一个任务打印否则打印内容会交错乱掉。如果想做得更有意思可以把资源池改成N个虚拟设备比如N个可用的USB端口或N个内存块每个任务用之前take用完give。static k_sem_t pool_sem; void pool_init(void) { k_sem_init(pool_sem, 3, 3); /* 资源池里有3个可用资源 */ } void worker_task_entry(void *arg) { int id (int)arg; while (1) { k_task_delay(id * 100); /* 申请资源 */ if (k_sem_take(pool_sem, 1000) 0) { /* 占用资源 */ dbg_print(Task %d enter pool\n, id); k_task_delay(300); dbg_print(Task %d leave pool\n, id); /* 释放资源 */ k_sem_give(pool_sem); } else { dbg_print(Task %d timeout\n, id); } } }运行三个worker任务观察打印内容你会看到“enter pool”和“leave pool”之间是成对出现的而且任意时刻最多只有三条“enter”记录没有被对应的“leave”抵消。这就是计数信号量最典型的应用。如果把sem_init的第三个参数改成1这个计数信号量就退化成了互斥锁同一时刻只允许一个任务占用资源。4.3 实验三互斥量保护临界区——共享变量的自增不再丢数据第三个实验直接复现开头提到的计数器问题。两个任务同时对同一个全局变量g_count做10000次自增。如果没有互斥保护最终的计数值一定小于20000而且每次跑的结果还不一样。加上互斥量之后结果稳定在20000。static k_sem_t mutex_sem; static uint32_t g_count; void inc_task_entry(void *arg) { int i; for (i 0; i 10000; i) { k_sem_take(mutex_sem, 1000); g_count; k_sem_give(mutex_sem); } }这里我要多说一句用二值信号量做互斥在功能上是可行的但工程上不推荐。二值信号量没有“所有权”的概念也就是说任务A可以give一个自己没有take过的信号量这会导致互斥保护名存实亡。真正的互斥量mutex应该要求只有持有者才能释放并且在任务被删除时自动释放它持有的锁。后续的章节里我会单独实现一个互斥量这次先用二值信号量顶一下但“互斥量必须单独设计”这个意识得留在脑子里。5. 优先级反转信号量实现中最容易踩的坑与对策5.1 一个反直觉的实验现象如果你照着上面的实验三跑一阵子大概率会觉得“信号量已经搞定一切了”。但别急我再加一个实验三个任务高优先级任务H中优先级任务M低优先级任务L。L先抢到互斥量进入临界区但还没执行完就被调度走了。这时候H就绪了H想拿互斥量拿不到被阻塞。然后M就绪了因为M的优先级高于L所以M抢占CPU运行。M运行完后L才继续。L执行完毕释放互斥量H才重新运行。这个过程的问题在于H明明优先级最高最后完成的时间却比L还晚。M这个“路人甲”任务反而抢在了H前面运行。这就是教科书上说的“优先级反转”Priority Inversion。“反转”不是说优先级数字变了而是任务的调度行为背离了优先级的设计意图——最高优先级的任务实际等待时间最长。5.2 对策一优先级继承解决优先级反转最常见的方法是优先级继承Priority Inheritance。核心思想是当一个低优先级任务持有互斥量、而一个高优先级任务正在等待这个互斥量时系统把低优先级任务的优先级临时提升到与高优先级任务相同或者至少提升到一定高度。这样M中优先级任务就没有机会抢占L了L能快速跑完临界区、释放互斥量然后恢复原来的优先级H才能尽早执行。优先级继承的实现需要在互斥量的结构体里额外记录三个信息当前持有者是谁、持有者的原始优先级、当前等待互斥量的最高优先级任务的优先级。当k_mutex_take发现锁被占用时就把持有者的优先级抬升当k_mutex_give释放锁时把持有者的优先级还原。typedef struct { k_task_t *owner; /* 当前持有者 */ uint8_t owner_prio; /* 持有者的原始优先级 */ uint8_t inherit_prio; /* 需要继承到的优先级 */ list_t wait_list; /* 等待该互斥量的任务链表 */ uint8_t locked; /* 是否被锁定 */ } k_mutex_t;注意这个地方有个细节继承的不是固定值而是要随着等待者优先级的增加而动态变化。比如L持有锁H和M同时在等但H优先级更高那L应该继承到H的优先级等H先被唤醒、拿到锁之后L的优先级可能又要降回M的优先级因为现在M是剩余等待者里优先级最高的了。实现这一步需要在每次take被阻塞的时候扫描wait_list里所有任务的优先级取最大值然后动态更新owner的优先级。5.3 对策二优先级天花板另一种方案是优先级天花板Priority Ceiling它规定任务在持有某互斥量期间直接运行在“所有可能使用这个互斥量的任务的最高优先级”之上。这个方案在实现上更简单系统只需要在创建互斥量时指定一个天花板优先级take的时候直接把当前任务抬到天花板give的时候再降回原级别。它的缺点是灵活性差——如果互斥量被很多不同优先级的任务使用天花板优先级会变得很高导致系统整体调度灵活性下降。我在实际项目中优先使用优先级继承方案只有在系统任务数量少、而且互斥量的使用场景可以预先确定时才考虑天花板。这个对照关系可以整理成一张表便于后面选型时参考方案实现难度系统开销适用场景优先级继承较高每次拖动都要调整任务数量多、优先级关系复杂的通用系统优先级天花板较低锁的临界区拉长任务结构固定、互斥量使用场景可预先分析的场景注意我们这里的实时性讨论都基于单核MCU和可抢占内核的前提。如果你是拿这个示例跑在PC或者别的环境上调度器的行为细节会不一样但信号量的语义和优先级反转问题是通用的。5.4 调试优先级继承时的“自锁”陷阱最后分享一个我在实现优先级继承时真实踩过的坑。第一次实现完成跑实验三一切正常但一旦把H、M、L三个任务组合在一起跑系统就在某个随机时刻卡死。查了很久之后发现问题出在k_mutex_give里面释放锁后要恢复owner的原始优先级但我调用的k_task_set_priority函数里又去拿了一遍互斥量……结果锁还没完全释放自己先把自己卡死了。所以这里面有一个重要的编码原则**内核内部函数之间的调用次序要格外小心尤其不能在内核API还没有完成状态更新前又去调用同一个API的其它变体。**我的做法是把k_task_set_priority拆成两个版本一个内部版本__k_task_set_priority假设调用者已经进入了临界区一个外部版本k_task_set_priority先关中断再调内部版本。这样每个内核服务都遵循“关中断—改状态—开中断”的规范自锁问题就彻底解决了。6. 信号量的边界情况与下一步扩展写到这里这篇的核心内容其实已经覆盖完整了从信号量的数据结构、take/give的实现到调度器的配套逻辑再到优先级继承的对策。但在收尾之前还有三个边界情况值得拿出来单独提醒一下——它们在实际项目中一定会遇到。超时返回的多义性。k_sem_take返回0时表示成功获取了信号量返回-1时表示超时。但是超时本身有两种原因一种是真的等满了时间还没等到资源另一种是时间虽然到了但中间被打断过比如被高优先级任务抢占实际等待时间远远大于传进去的参数。如果你的代码把timeout当作严格的最后期限这种多义性会出问题。解决方法是多返回一个“剩余等待时间”或者“是否真正超时”的标志这个细节可以根据实际应用决定要不要做。中断服务程序里的give。信号量的give操作经常在中断里调用用来唤醒一个任务去处理数据。这种情况需要考虑一个问题中断里调用give时当前执行的并不是任务上下文不能直接触发任务切换。正确的姿势是只做“把任务从等待队列摘除、插入就绪队列、设置need_sched标志”这几件事等中断退出时由汇编代码统一检查need_sched再决定要不要切任务。如果你在中断里直接调用调度器大概率会把栈搞崩。内存序的问题。这一条在Cortex-M等现代ARM核上尤其重要。信号量的count和wait_list被修改后要确保对其它核或者同一核上的中断上下文是立即可见的。如果你的编译器优化级别开得比较高或者目标平台是多核需要在进入临界区时加上合适的内存屏障。目前我只用了关中断来保证原子性在单核MCU上够用如果后面要把这套代码移植到多核平台这里得补上dmb之类的指令。下一步如果要继续扩展我建议按这个顺序走先实现真正的互斥量带所有权和优先级继承然后是事件标志组再后面可以试试消息队列和邮箱。这几个东西一旦有了一个迷你RTOS的基本IPC就齐活了。下一篇我打算直接从互斥量补全开始写把今天提到的优先级继承完整落地成代码。你如果已经照着这个系列板调到了第八篇不妨先把信号量移植到自己的开发板上跑一遍文章里的三个实验亲眼确认一下竞态问题的存在和解决过程这个过程比任何源码都更有说服力。 
