翻 uC/OS-II 的内核源码这件事我给自己定的规矩是别贪多一次只啃一两个文件。到第 8 篇前面任务控制块、就绪表、调度器、时钟节拍、事件控制块、信号量和互斥量都理过一遍了接下来绕不开的就是任务之间怎么传数据。做 RTOS 项目的人心里都清楚任务之间真正高频的交互不是抢资源而是把手里的东西交给下一个环节——串口收到的帧、ADC 采样的结果、上位机下发的命令这些都得有个地方放、有个地方取。uC/OS-II 给的两条路就是消息邮箱 OSMbox 和消息队列 OSQ。这一篇我们就盯着 OSMbox.c 和 OSQ.c 这两个文件看看这六千多行内核代码里这一小块到底怎么写的为什么非得这么写。1. 为什么第 8 篇要啃消息通信这块硬骨头1.1 信号量管的是“资格”消息管的是“内容”前面几篇把信号量和互斥量翻来覆去讲了不少但它们解决的都是同一类问题某个资源现在能不能被访问。任务 Pend 一个信号量本质上是在问“轮到我了没有”拿到的只是一个许可不携带任何数据。这在保护共享外设、控制并发访问的时候够用可一旦涉及数据流转就明显不够了。举个例子。一个采集任务从 UART 里扒出一帧数据另一个协议解析任务等着处理它。你用信号量去通知解析任务“有新数据了”那这条数据放哪儿只能开一个全局数组然后加个标志位或者用互斥量保护。伪代码大概长这样static INT8U g_uart_buf[64]; static INT8U g_uart_len; static OS_EVENT *g_uart_sem; /* 采集任务 */ Uart_Read(g_uart_buf, g_uart_len); OSSemPost(g_uart_sem); /* 解析任务 */ OSSemPend(g_uart_sem, 0, err); Parse(g_uart_buf, g_uart_len); /* 这里其实还有隐患 */看着能用问题是这份数据的所有权没交代清楚。采集任务 Post 完之后能不能继续往 g_uart_buf 里写下一帧解析任务读到一半采集任务又覆盖了怎么办信号量只保证了“串行访问”的那一瞬间数据缓冲区的生命周期完全靠人肉约定。任务一多这种约定基本守不住。消息邮箱和消息队列把这件事彻底改了个方向传递的不是“有没有”的标记而是数据本身的指针。谁 Pend 到谁就拿到这份数据的处置权用完自己决定释放还是归还内存池。所有权跟着指针走逻辑就干净了。1.2 OSMbox 和 OSQ 是同一个爹生的很多人第一次看这两个模块会觉得邮箱是队列的特例——容量为 1 的队列嘛。概念上这么说没错但在 uC/OS-II 的源码里它们是两个独立实现共用同一套底层机制。真正共用的东西是OS_EVENT 事件控制块和等待任务表这套东西前面讲信号量的时候已经见过一回了。它们的差异集中在两个字段上特性消息邮箱 OSMbox消息队列 OSQ事件类型 OSEventTypeOS_EVENT_TYPE_MBOXOS_EVENT_TYPE_QOSEventPtr 指向消息本身void *队列控制块OS_Q *容量1 条创建时指定FIFO满时行为返回OS_ERR_MBOX_FULL返回OS_ERR_Q_FULL额外控制块不需要每个队列一个OS_Q内存开销只占一个事件块事件块 队列块 用户数组这张表里有几个点值得单独拎出来说。第一邮箱不会覆盖。2.86 版本里OSMboxPost()在邮箱已满时直接返回错误码不会像有些 RTOS 那样把旧消息丢掉。这个设计有人嫌它麻烦但它逼着你在设计阶段就想清楚消息的生产速率和消费速率是否匹配。第二队列比你想象的多占一块内存。用户数组之外每个队列还要从OS_MAX_QS里领一个OS_Q结构这个结构在 32 位机器上是 24 字节。那实际项目里怎么选我的经验是任务之间是一对一的点对点握手用邮箱一对多、或者生产速率可能短时超过消费速率、需要缓冲的用队列。中断服务程序往任务里投数据这种场景基本无脑用队列因为中断来的时候没人等你。2. OS_EVENT 事件块的复用设计2.1 OSEventType 与 OSEventPtr 的双重身份打开uCOS_II.HOS_EVENT 的定义大概是这样typedef struct os_event { INT8U OSEventType; /* 事件类型 */ void *OSEventPtr; /* 消息 / 队列控制块 */ INT16U OSEventCnt; /* 计数器信号量用 */ OS_PRIO OSEventGrp; /* 等待任务优先级组 */ OS_PRIO OSEventTbl[OS_EVENT_TBL_SIZE];/* 等待任务优先级表 */ #if OS_EVENT_NAME_EN 0u INT8U *OSEventName; #endif } OS_EVENT;OSEventType决定了后面几个字段怎么解释OSEventType 取值对应对象OSEventPtr 的用途OS_EVENT_TYPE_UNUSED(0)未使用兼作空闲链表 next 指针OS_EVENT_TYPE_MBOX(1)邮箱指向消息OS_EVENT_TYPE_Q(2)队列指向OS_QOS_EVENT_TYPE_SEM(3)信号量未使用OS_EVENT_TYPE_MUTEX(4)互斥量指向持有者的 TCB这里有个挺精妙的复用空闲事件块和已使用事件块共用同一块内存。OSInit()的时候所有事件块被串成一条单向链表OSEventPtr当 next 指针用从OSEventFreeList里摘一个出来之后OSEventPtr立刻换岗去指消息或者队列。这种“同一片内存不同阶段干不同事”的做法在嵌入式里非常常见好处是零额外开销代价是调试的时候不能想当然地看某个字段。顺带说一句早期版本里事件标志组也是塞在 OS_EVENT 里的OS_EVENT_TYPE_FLAG2.86 之后事件标志独立成了OS_FLAG_GRP结构。所以你在一些老代码里看到OS_EVENT_TYPE_FLAG别以为是笔误那是历史遗留。2.2 等待任务表位图比链表快在哪OSEventGrp加OSEventTbl[]这套东西跟内核里的就绪表OSRdyGrp/OSRdyTbl[]是同一个套路8 位分组每组 8 位一共支撑 64 个优先级OS_LOWEST_PRIO默认 63。挂上一个任务、摘掉一个任务都是位运算OS_EventTaskWait()里就三行pevent-OSEventTbl[OSTCBCur-OSTCBY] | OSTCBCur-OSTCBBitX; /* 置位 */ pevent-OSEventGrp | OSTCBCur-OSTCBBitY; /* 同时把自己从就绪表里摘掉 */ OSRdyTbl[y] (OS_PRIO)~OSTCBCur-OSTCBBitX; if (OSRdyTbl[y] 0u) { OSRdyGrp (OS_PRIO)~OSTCBCur-OSTCBBitY; }为什么不用链表因为事件唤醒的时候要立刻找到等待者中优先级最高的那一个。用链表就得从头遍历最坏 O(n)用位图配合OSUnMapTbl[]查表两步查表拿到最高优先级是常数时间。RTOS 里事件唤醒这件事可能发生在中断上下文时间必须可预期链表那点便利换不来这个确定性。代价也很直白每个事件块固定吃掉 9 个字节的等待表1 字节 Grp 8 字节 Tbl跟这个事件实际挂了多少个任务无关。所以OS_MAX_EVENTS别乱开大一个 20 字节左右的事件块乘上几十个RAM 里就是小一千字节在 GD32F103 这种 48KB SRAM 的片子上不是小数目。3. OSMbox 源码逐行拆3.1 OSMboxCreate创建时就允许带一条消息OS_EVENT *OSMboxCreate (void *pmsg) { OS_EVENT *pevent; OS_ENTER_CRITICAL(); pevent OSEventFreeList; /* 从空闲链表摘一个 */ if (OSEventFreeList ! (OS_EVENT *)0) { OSEventFreeList (OS_EVENT *)OSEventFreeList-OSEventPtr; } OS_EXIT_CRITICAL(); if (pevent ! (OS_EVENT *)0) { pevent-OSEventType OS_EVENT_TYPE_MBOX; pevent-OSEventCnt 0u; pevent-OSEventPtr pmsg; /* 初始消息 */ OS_EventWaitListInit(pevent); /* 清空等待表 */ } return (pevent); }这段代码有两个地方新手容易忽略。一是创建失败返回 NULL因为没有空闲事件块了。很多人的代码里OSMboxCreate()的返回值从来不判断跑着跑着OS_MAX_EVENTS不够拿回来一个空指针后面一 Post 就崩在事件类型校验上。二是参数 pmsg 可以不为 NULL。这意味着你可以在OSInit()之后、多任务还没启动之前先把第一条消息塞进去。系统启动初期任务按优先级依次调度第一个跑起来的任务 Pend 就能直接拿到不需要额外搞一个“初始化完成”标志。提示OS_EventWaitListInit()做的事就是把OSEventGrp清零、OSEventTbl[]全清、OSEventCnt归零。它不碰OSEventPtr和OSEventType所以必须在这之前把这两个字段设好顺序不能反。3.2 OSMboxPend 的三条返回路径OSMboxPend()是邮箱模块里最长的一个函数逻辑上分成三段void *OSMboxPend (OS_EVENT *pevent, INT16U timeout, INT8U *perr) { void *pmsg; /* ...省略参数检查... */ OS_ENTER_CRITICAL(); pmsg pevent-OSEventPtr; if (pmsg ! (void *)0) { /* 路径一邮箱里有货直接拿走 */ pevent-OSEventPtr (void *)0; /* 取走即清空 */ OS_EXIT_CRITICAL(); *perr OS_ERR_NONE; return (pmsg); } OSTCBCur-OSTCBStat | OS_STAT_MBOX; /* 路径二没货挂起自己 */ OSTCBCur-OSTCBStatPend OS_STAT_PEND_OK; OSTCBCur-OSTCBDly timeout; OS_EventTaskWait(pevent); OS_EXIT_CRITICAL(); OS_Sched(); /* 让出 CPU */ OS_ENTER_CRITICAL(); switch (OSTCBCur-OSTCBStatPend) { case OS_STAT_PEND_OK: /* 被别人 Post 唤醒 */ pmsg OSTCBCur-OSTCBMsg; *perr OS_ERR_NONE; break; case OS_STAT_PEND_ABORT: /* 被强制取消等待 */ pmsg (void *)0; *perr OS_ERR_PEND_ABORT; break; case OS_STAT_PEND_TO: default: /* 路径三超时 */ OS_EventTaskRemove(OSTCBCur, pevent); pmsg (void *)0; *perr OS_ERR_TIMEOUT; break; } OSTCBCur-OSTCBStat OS_STAT_RDY; OSTCBCur-OSTCBStatPend OS_STAT_PEND_OK; OSTCBCur-OSTCBEventPtr (OS_EVENT *)0; OS_EXIT_CRITICAL(); return (pmsg); }这里最值得琢磨的是路径二为什么不直接读OSEventPtr而要读OSTCBMsg。答案在于邮箱只有一个槽位。假设OSMboxPost()的时候把消息写进pevent-OSEventPtr那么被唤醒的任务还没被调度上 CPU第二个任务调OSMboxPend()就会看到邮箱里“有消息”又拿走了同一条。所以 uC/OS-II 的做法是有任务在等的时候消息直接塞进那个任务的 TCBOSTCBMsg字段邮箱本身保持“空”的状态直到下次有人 Post。这个设计在OS_EventTaskRdy()里能看到对应代码。另外注意timeout参数传 0 的含义是无限等待不是立即返回。想要非阻塞查询得用OSMboxAccept()如果这个配置项打开了。3.3 OSMboxPost 的两条投递分支INT8U OSMboxPost (OS_EVENT *pevent, void *pmsg) { if (OSIntNesting 0u) { return (OS_ERR_POST_ISR); /* 中断里禁止调用 */ } if (pevent-OSEventType ! OS_EVENT_TYPE_MBOX) { return (OS_ERR_EVENT_TYPE); } if (pmsg (void *)0) { return (OS_ERR_POST_NULL_PTR); /* 不允许投递空指针 */ } OS_ENTER_CRITICAL(); if (pevent-OSEventGrp ! 0u) { /* 有人在等 */ (void)OS_EventTaskRdy(pevent, pmsg, OS_STAT_MBOX, OS_STAT_PEND_OK); OS_EXIT_CRITICAL(); OS_Sched(); /* 唤醒的是高优先级就要切换 */ return (OS_ERR_NONE); } if (pevent-OSEventPtr ! (void *)0) { /* 没人在等且邮箱已满 */ OS_EXIT_CRITICAL(); return (OS_ERR_MBOX_FULL); } pevent-OSEventPtr pmsg; /* 没人在等先存着 */ OS_EXIT_CRITICAL(); return (OS_ERR_NONE); }三个校验里OS_ERR_POST_NULL_PTR这条特别容易踩。因为邮箱用“OSEventPtr NULL”来表示“空”你要是真往里面 Post 一个空指针邮箱的状态就乱了——下次 Pend 会认为没货继续等等到天亮。所以内核干脆在入口就把这条路堵死。OS_ERR_POST_ISR这条限制也值得说道。中断里想投递邮箱得改用OSMboxPostOpt()它的实现里没有这个 ISR 检查同时多了两个可选行为OS_POST_OPT_NOWAIT不触发任务切换留给OSIntExit()统一调度和OS_POST_OPT_BROADCAST广播给所有等待者只对事件标志有意义用在邮箱上相当于只唤醒一个。写串口中断服务程序的时候这个区别就是能不能正常收发数据的分水岭。4. OSQ 源码逐行拆4.1 OS_Q 控制块与那个哨兵地址typedef struct os_q { struct os_q *OSQPtr; /* 空闲队列链表的 next 指针 */ void **OSQStart; /* 用户数组起始地址 */ void **OSQEnd; /* 用户数组结束地址哨兵 */ void **OSQIn; /* 入队指针 */ void **OSQOut; /* 出队指针 */ INT16U OSQSize; /* 队列容量 */ INT16U OSQEntries; /* 当前条目数 */ } OS_Q;这个结构在 32 位机上是 24 字节前 20 字节是五个指针后面 4 字节是两个INT16U。它和 OS_EVENT 一样玩“空闲时当链表节点”的把戏OSInit()阶段OS_MAX_QS个OS_Q被串成OSQFreeListOSQPtr当 next 用OSQCreate()摘一个出来之后OSQPtr就不管了。OSQEnd的含义要特别注意pq-OSQStart start; pq-OSQEnd start[size]; /* 注意是 start[size]不是 start[size-1] */ pq-OSQIn start; pq-OSQOut start; pq-OSQSize size; pq-OSQEntries 0u;OSQEnd指向的是数组最后一个元素的下一个位置也就是个哨兵。有了它OSQIn和OSQOut推进时只需要一句if (OSQIn OSQEnd) OSQIn OSQStart;就能完成回绕不需要在每次移动时做取模运算。取模在 Cortex-M3 上要几十个周期而这个判断只要一两个周期。中断里跑的代码省下来的都是实打实的响应时间。所以数组该怎么开static void *UartQTbl[32]; /* 32 个指针占 128 字节 */ UartQ OSQCreate(UartQTbl[0], 32); /* size 传 32 */数组开 32 个元素size 也传 32这是对的。别自作聪明开成 33 个多出来的那格永远不会被写到。4.2 OSQPost 的入队与溢出判断INT8U OSQPost (OS_EVENT *pevent, void *pmsg) { OS_Q *pq; /* ...参数校验... */ OS_ENTER_CRITICAL(); if (pevent-OSEventGrp ! 0u) { /* 有任务在等直接给它 */ (void)OS_EventTaskRdy(pevent, pmsg, OS_STAT_Q, OS_STAT_PEND_OK); OS_EXIT_CRITICAL(); OS_Sched(); return (OS_ERR_NONE); } pq (OS_Q *)pevent-OSEventPtr; if (pq-OSQEntries pq-OSQSize) { /* 判断在前 */ OS_EXIT_CRITICAL(); return (OS_ERR_Q_FULL); } *pq-OSQIn pmsg; /* 写入并推进指针 */ pq-OSQEntries; if (pq-OSQIn pq-OSQEnd) { /* 回绕在后 */ pq-OSQIn pq-OSQStart; } OS_EXIT_CRITICAL(); return (OS_ERR_NONE); }这段代码的四个动作顺序是固定的先判满、再写入、再计数、最后回绕。顺序换任何一个都会出问题。比如把回绕挪到写入之前写入的地址就错了把计数放到回绕之后中断打断时读到的OSQEntries和实际指针状态对不上。判满用而不是是因为OSQEntries的最大合法值就是OSQSize两者相等表示满。还有个细节OSEventPtr指向队列控制块而不是消息本身。所以队列的“空/满”状态由OSQEntries决定跟OSEventPtr没关系。这跟邮箱的设计是反过来的也是邮箱和队列没法直接互相替代的根本原因。4.3 OSQPend 的出队与超时收尾void *OSQPend (OS_EVENT *pevent, INT16U timeout, INT8U *perr) { void *pmsg; OS_Q *pq; /* ...参数校验... */ OS_ENTER_CRITICAL(); pq (OS_Q *)pevent-OSEventPtr; if (pq-OSQEntries 0u) { /* 队列非空 */ pmsg *pq-OSQOut; /* 读出并推进 */ pq-OSQEntries--; if (pq-OSQOut pq-OSQEnd) { pq-OSQOut pq-OSQStart; } OS_EXIT_CRITICAL(); *perr OS_ERR_NONE; return (pmsg); } OSTCBCur-OSTCBStat | OS_STAT_Q; /* 队列空挂起 */ OSTCBCur-OSTCBStatPend OS_STAT_PEND_OK; OSTCBCur-OSTCBDly timeout; OS_EventTaskWait(pevent); OS_EXIT_CRITICAL(); OS_Sched(); OS_ENTER_CRITICAL(); /* 醒来后同样是三选一POST / ABORT / TIMEOUT */ /* ...与 OSMboxPend 结构完全一致... */ }跟邮箱对比一下就能看出Pend 阶段两者的骨架几乎一模一样差别只在“非空判断”的依据——邮箱看指针是不是 NULL队列看OSQEntries是不是大于 0。这也是为什么我一直建议先把邮箱搞透队列就是多了一个控制块和一层环形缓冲的算术。有一个细节容易看漏出队时pmsg是从*pq-OSQOut读出来的然后才把OSQOut往前推。这一步如果被中断打断只要临界区保护到位就不会出问题因为进入OS_ENTER_CRITICAL()之后所有任务和大部分中断都被挡住了除非有更高优先级的中断没被屏蔽那种情况要靠OSIntEnter()/OSIntExit()配合。5. 消息投递背后的枢纽等待与唤醒5.1 OS_EventTaskWait把自己从就绪表挪到等待表void OS_EventTaskWait (OS_EVENT *pevent) { INT8U y; OSTCBCur-OSTCBEventPtr pevent; pevent-OSEventTbl[OSTCBCur-OSTCBY] | OSTCBCur-OSTCBBitX; pevent-OSEventGrp | OSTCBCur-OSTCBBitY; y OSTCBCur-OSTCBY; OSRdyTbl[y] (OS_PRIO)~OSTCBCur-OSTCBBitX; if (OSRdyTbl[y] 0u) { OSRdyGrp (OS_PRIO)~OSTCBCur-OSTCBBitY; } }这个函数干的事就四行但蕴含了一个关键约定任务的OSTCBEventPtr字段只允许指向一个事件。也就是说一个任务同一时刻只能等待一个邮箱、或者一个队列、或者一个信号量。你要是想让某个任务同时等三个事件源uC/OS-II 的邮箱和队列帮不了你得靠事件标志组或者在应用层用多个中间任务做转发。这个约束在源码里没有任何运行时检查全靠OSTCBEventPtr被反复覆盖来“自然生效”。如果哪个任务的等待逻辑写岔了OSTCBEventPtr指着一个已经被删除的事件超时唤醒的时候OS_EventTaskRemove()就会去操作一块已经还给空闲链表的内存——这类野指针在嵌入式上极难查因为崩的地方往往离出问题的地方很远。5.2 OS_EventTaskRdy唤醒谁、消息怎么给INT8U OS_EventTaskRdy (OS_EVENT *pevent, void *pmsg, INT8U msk, INT8U pend_stat) { OS_TCB *ptcb; INT8U y, x, prio; y OSUnMapTbl[pevent-OSEventGrp]; /* 找最高优先级组 */ x OSUnMapTbl[pevent-OSEventTbl[y]]; /* 找组内最高位 */ prio (INT8U)((y 3u) x); ptcb OSTCBPrioTbl[prio]; ptcb-OSTCBEventPtr (OS_EVENT *)0; ptcb-OSTCBMsg pmsg; /* 消息塞进 TCB */ ptcb-OSTCBStat (INT8U)~(INT8U)msk; ptcb-OSTCBStatPend pend_stat; if ((ptcb-OSTCBStat OS_STAT_SUSPEND) OS_STAT_RDY) { OSRdyGrp | ptcb-OSTCBBitY; /* 重新挂回就绪表 */ OSRdyTbl[y] | ptcb-OSTCBBitX; } else { ptcb-OSTCBDly 0u; /* 被挂起了不再计时 */ } OS_EventTaskRemove(ptcb, pevent); /* 从等待表摘掉 */ return (prio); }注意ptcb-OSTCBMsg pmsg;这一行。这就是邮箱和队列能传递数据的唯一通道。不管你是 Post 邮箱还是 Post 队列消息指针最终都是通过这个字段交到被唤醒任务的OSMboxPend()/OSQPend()手里的。所以前面那条“有任务在等时消息不走OSEventPtr”的规则在这里得到了解释。另外注意那个else分支如果被唤醒的任务正好处于挂起状态OSSuspend过它不能被挂回就绪表但OSTCBDly会被清零表示它的等待条件已经满足超时计时不再生效。等它被OSResume()恢复之后读到的OSTCBStatPend就是OS_STAT_PEND_OK能正常拿到消息。这个细节在看OSTaskSuspend()/OSTaskResume()那部分源码时能对上。OS_EventTaskRemove()就是OS_EventTaskWait()的逆操作把任务的位从OSEventTbl[]里清掉如果整组空了再把OSEventGrp对应位清零。三行代码但组清零那个判断不能漏否则最高优先级查找会命中一个空组。6. 在 GD32F103 上跑一个生产者消费者6.1 先把配置项理清楚移植层的东西前面几篇讲过了这里只说和消息通信相关的配置。OS_CFG.H里这几个宏必须对配置项建议值说明OS_MAX_EVENTS8~16所有事件块总数邮箱、队列、信号量、互斥量共用OS_MAX_QS4OS_Q控制块数量只统计队列OS_Q_EN1队列总开关OS_Q_DEL_EN0不需要运行时删队列就关掉省代码OS_Q_QUERY_EN1调试时看队列里剩几条很好用OS_MBOX_EN1邮箱总开关OS_MBOX_POST_EN1关掉的话OSMboxPost()编不进来OS_MBOX_POST_OPT_EN1中断投递必须开OS_MBOX_DEL_EN0同上不用就关OS_MBOX_QUERY_EN1调试用OS_TASK_DEL_EN1OSQDel/OSMboxDel里要用到OSTaskDel相关逻辑OS_EVENT_NAME_EN1给事件起名用调试器看变量时一目了然OS_MAX_EVENTS和OS_MAX_QS是两本账别搞混。OSQCreate()要同时从两个池子里各领一个块任意一个池子空了都会返回 NULL。我见过有人把OS_MAX_QS设成 8、OS_MAX_EVENTS设成 4然后创建第 5 个队列的时候失败查了半天以为是内存不够。6.2 完整的生产消费骨架下面这段代码我在 GD32F103 上跑过串口 1 收数据、串口 2 转发中间加了两级缓冲。#include includes.h #define UART_Q_SIZE 32 #define FRAME_POOL_SIZE 8 #define FRAME_BUF_SIZE 16 #define PROD_PRIO 10 #define CONS_PRIO 11 #define TASK_STK_SIZE 256 static OS_STK ProdStk[TASK_STK_SIZE]; static OS_STK ConsStk[TASK_STK_SIZE]; static void *UartQTbl[UART_Q_SIZE]; /* 队列用的指针数组 */ static OS_EVENT *UartQ; /* 队列事件 */ typedef struct { INT8U len; INT8U buf[FRAME_BUF_SIZE]; } UartFrame; static UartFrame FramePool[FRAME_POOL_SIZE]; static OS_MEM *FrameMem; /* 内存池第 9 篇细讲 */ /* 生产者从串口 FIFO 取一帧装箱后投递 */ void ProdTask (void *p_arg) { INT8U err; UartFrame *pf; (void)p_arg; for (;;) { pf (UartFrame *)OSMemGet(FrameMem, err); if (err OS_ERR_NONE) { pf-len Uart_PopFrame(pf-buf, FRAME_BUF_SIZE); if (pf-len 0u) { OSQPost(UartQ, (void *)pf); /* 投递指针不拷贝数据 */ } else { (void)OSMemPut(FrameMem, (void *)pf); /* 空帧直接还回去 */ } } OSTimeDlyHMSM(0, 0, 0, 20); /* 20ms 轮询一次 */ } } /* 消费者拿到就用用完归还内存池 */ void ConsTask (void *p_arg) { INT8U err; UartFrame *pf; (void)p_arg; for (;;) { pf (UartFrame *)OSQPend(UartQ, 0u, err); /* 0 表示无限等 */ if (err OS_ERR_NONE) { Uart_SendFrame(pf-buf, pf-len); /* 转发到串口 2 */ (void)OSMemPut(FrameMem, (void *)pf); /* 关键谁拿谁还 */ } } } int main (void) { INT8U err; BSP_Init(); OSInit(); FrameMem OSMemCreate(FramePool[0], FRAME_POOL_SIZE, sizeof(UartFrame), err); UartQ OSQCreate(UartQTbl[0], UART_Q_SIZE); if ((FrameMem (OS_MEM *)0) || (UartQ (OS_EVENT *)0)) { while (1) { } /* 初始化失败死等 */ } OSTaskCreate(ProdTask, (void *)0, ProdStk[TASK_STK_SIZE - 1], PROD_PRIO); OSTaskCreate(ConsTask, (void *)0, ConsStk[TASK_STK_SIZE - 1], CONS_PRIO); OSStart(); return 0; }这段代码里有三个设计决策值得解释一下。为什么传指针而不是拷贝整个结构体。OSQPost()存的只是一个void *队列本身不关心指针指向什么。传指针的好处是零拷贝、速度快中断里投递一个 4 字节指针代价可以忽略。代价是缓冲区的生命周期管理转移给了应用层——这个责任必须落到“拿到消息的那个任务”头上。为什么用内存池而不是 malloc。每次 Post 前 malloc 一块、Pend 后 free在裸机上跑不了多久就会碎片化尤其是有大小不一的结构体混在一起的时候。用固定大小的内存池分配释放都是 O(1)时间可预期且不会产生碎片。第 9 篇会专门拆OSMemGet/OSMemPut这里先按“它就是一组固定大小的块”理解就行。为什么消费者优先级比生产者低。生产者是 10消费者是 11。串口数据来的速度可能是突发的生产者优先级高一点能及时把数据抢进来。但这里有个隐患如果生产者一直来数据消费者永远排不上队列就会满。所以生产者的循环末尾加了OSTimeDlyHMSM(0,0,0,20)主动让出 CPU。这个 20ms 不是拍脑袋定的是算出来的队列 32 个槽每 20ms 最多产生一帧理论上 640ms 内不会满而消费者处理一帧大概 1ms 以内完全跟得上。队列深度和轮询周期的关系就是这么倒推出来的。7. 常见问题与排查技巧实录7.1 故障速查表现象最可能的原因排查动作OSQCreate()返回 NULLOS_MAX_EVENTS或OS_MAX_QS耗尽统计全工程创建了多少事件和队列查有没有创建失败没处理的分支OSMboxCreate()返回 NULLOS_MAX_EVENTS不够或者OSInit()没调用确认OSInit()在创建之前执行OSQPost()一直返回OS_ERR_Q_FULL消费者卡住或队列太小用OSQQuery()看OSNMsgs是不是长期顶格查消费者任务状态OSMboxPost()返回OS_ERR_POST_ISR在中断里调了OSMboxPost换成OSMboxPostOpt(pevent, pmsg, OS_POST_OPT_NOWAIT)OSMboxPost()返回OS_ERR_POST_NULL_PTR投递了空指针上游指针没赋值成功检查OSMemGet的返回值Pend 永远拿不到数据消息投递到了别的事件或者等待的是不同的队列用调试器看OSTCBEventPtr指向哪个事件块偶发死机、栈溢出队列传指针后多个任务同时改了同一块内存确认所有权唯一或者改用值传递系统跑一段时间后创建事件失败有事件被 Del 了但没回收检查OS_EVENT_TYPE_UNUSED的块有没有回到空闲链表7.2 几条只能靠踩坑得到的经验第一条邮箱满了不覆盖这件事一定要在协议设计阶段就考虑。我在一个项目里用邮箱传“紧急停机”命令正常状态下邮箱是空的Post 进去等着即将被 Pend。结果有一次 Post 完之后负责 Pend 的任务因为另一个事件被阻塞了 200ms第二个紧急命令又来了OSMboxPost()返回OS_ERR_MBOX_FULL调用方没检查返回值命令就丢了。这类“高优先级消息被丢”的故障在测试环境里几乎复现不出来因为时序太特殊。后来我改成队列容量给 4丢包变成了排队才彻底解决。第二条OSQPend的 timeout 参数单位是时钟节拍不是毫秒。默认OS_TICKS_PER_SEC是 100也就是 10ms 一个节拍。你想等 100ms 得传 10。我在早期代码里直接写OSQPend(Q, 100, err)心里想的是 100ms实际等了 1 秒调试的时候一脸懵。要么养成用OS_TICKS_PER_SEC做换算的习惯要么写个宏包一层。第三条事件块删除之后全局指针一定要置 NULL。OSQDel()会把OS_Q块和OS_EVENT块分别还给空闲链表。如果全局变量UartQ还留着那个地址理论上后面创建别的事件可能复用同一块内存你就OSQPost(UartQ, ...)就投跑到别人的队列里去了或者更糟投到一个信号量上。OSQDel()的实现里会检查事件类型所以大概率会返回OS_ERR_EVENT_TYPE但要是那块内存恰好被复用成了新队列类型校验就过了然后你就看着一个完全不相干的队列里冒出数据来。第四条调试时打开OS_Q_QUERY_EN和OS_MBOX_QUERY_EN太值了。OSQQuery()返回的OS_Q_DATA里有OSNMsgs当前条目数、OSQSize容量、OSMsg如果是在等的话指向任务拿到的消息。用调试器把某个队列的这几个值加进 watch 窗口生产的速率和消费的速率一眼就看出来了。比打印日志方便也不影响实时性。第五条别在一个任务里同时 Pend 两个队列。前面说过了OSTCBEventPtr只能存一个事件指针。你要是写了OSQPend(Q1, ...)然后紧接着OSQPend(Q2, ...)这是顺序等待没问题但如果你想“谁先来等谁”得用别的手段。这个约束在 uC/OS-II 里没有编译期检查纯粹靠开发者记住。第六条队列数组建议加static并且初始化为零。队列数组里的指针在取出之前是未定义的如果你在OSQCreate()之前就有人 Pend内核不会去读数组所以不会出事但调试器里看到的是一堆随机值容易让人误判。加上static至少保证了启动时是零。这个习惯在排查问题时能省不少时间。第七条中断里投递队列之后别自己算优先级切换。uC/OS-II 的中断处理流程是这样进中断先OSIntEnter()投递用OSQPost()或者带 Opt 的版本出中断调OSIntExit()。OSIntExit()会判断有没有更高优先级的任务变成就绪如果有就触发任务切换。你不能在中断里自己去操作就绪表那是内核的活。这条规则在串口 DMA 接收中断里特别要注意很多人的第一版代码就是在中断里直接调OSTaskResume()跑起来看着正常压测就崩。这一篇把OSMbox.c和OSQ.c两个文件从头到尾过了一遍算下来有效代码量不到 400 行但里面每一个判断、每一行临界区都有它的道理。我个人读源码的习惯是先把结构体定义和那几个宏翻明白再去读函数实现因为结构体决定了数据怎么组织函数只是在对这个组织做操作。消息通信这块的结构其实就两个——OS_EVENT管等待队列OS_Q管环形缓冲剩下全是围绕它们的增删改查。第 9 篇打算接着往下走拆内存管理OSMem.c那个文件里有一个特别容易搞错的细节内存池的分区链表在分配和释放时怎么维护以及为什么OSMemCreate()的块大小要按 4 字节对齐。如果你在项目里正好被动态分配碎片化折腾过那篇应该对你胃口。 uC/OS-II 消息邮箱 OSMbox 与消息队列 OSQ 源码剖析
翻 uC/OS-II 的内核源码这件事我给自己定的规矩是别贪多一次只啃一两个文件。到第 8 篇前面任务控制块、就绪表、调度器、时钟节拍、事件控制块、信号量和互斥量都理过一遍了接下来绕不开的就是任务之间怎么传数据。做 RTOS 项目的人心里都清楚任务之间真正高频的交互不是抢资源而是把手里的东西交给下一个环节——串口收到的帧、ADC 采样的结果、上位机下发的命令这些都得有个地方放、有个地方取。uC/OS-II 给的两条路就是消息邮箱 OSMbox 和消息队列 OSQ。这一篇我们就盯着 OSMbox.c 和 OSQ.c 这两个文件看看这六千多行内核代码里这一小块到底怎么写的为什么非得这么写。1. 为什么第 8 篇要啃消息通信这块硬骨头1.1 信号量管的是“资格”消息管的是“内容”前面几篇把信号量和互斥量翻来覆去讲了不少但它们解决的都是同一类问题某个资源现在能不能被访问。任务 Pend 一个信号量本质上是在问“轮到我了没有”拿到的只是一个许可不携带任何数据。这在保护共享外设、控制并发访问的时候够用可一旦涉及数据流转就明显不够了。举个例子。一个采集任务从 UART 里扒出一帧数据另一个协议解析任务等着处理它。你用信号量去通知解析任务“有新数据了”那这条数据放哪儿只能开一个全局数组然后加个标志位或者用互斥量保护。伪代码大概长这样static INT8U g_uart_buf[64]; static INT8U g_uart_len; static OS_EVENT *g_uart_sem; /* 采集任务 */ Uart_Read(g_uart_buf, g_uart_len); OSSemPost(g_uart_sem); /* 解析任务 */ OSSemPend(g_uart_sem, 0, err); Parse(g_uart_buf, g_uart_len); /* 这里其实还有隐患 */看着能用问题是这份数据的所有权没交代清楚。采集任务 Post 完之后能不能继续往 g_uart_buf 里写下一帧解析任务读到一半采集任务又覆盖了怎么办信号量只保证了“串行访问”的那一瞬间数据缓冲区的生命周期完全靠人肉约定。任务一多这种约定基本守不住。消息邮箱和消息队列把这件事彻底改了个方向传递的不是“有没有”的标记而是数据本身的指针。谁 Pend 到谁就拿到这份数据的处置权用完自己决定释放还是归还内存池。所有权跟着指针走逻辑就干净了。1.2 OSMbox 和 OSQ 是同一个爹生的很多人第一次看这两个模块会觉得邮箱是队列的特例——容量为 1 的队列嘛。概念上这么说没错但在 uC/OS-II 的源码里它们是两个独立实现共用同一套底层机制。真正共用的东西是OS_EVENT 事件控制块和等待任务表这套东西前面讲信号量的时候已经见过一回了。它们的差异集中在两个字段上特性消息邮箱 OSMbox消息队列 OSQ事件类型 OSEventTypeOS_EVENT_TYPE_MBOXOS_EVENT_TYPE_QOSEventPtr 指向消息本身void *队列控制块OS_Q *容量1 条创建时指定FIFO满时行为返回OS_ERR_MBOX_FULL返回OS_ERR_Q_FULL额外控制块不需要每个队列一个OS_Q内存开销只占一个事件块事件块 队列块 用户数组这张表里有几个点值得单独拎出来说。第一邮箱不会覆盖。2.86 版本里OSMboxPost()在邮箱已满时直接返回错误码不会像有些 RTOS 那样把旧消息丢掉。这个设计有人嫌它麻烦但它逼着你在设计阶段就想清楚消息的生产速率和消费速率是否匹配。第二队列比你想象的多占一块内存。用户数组之外每个队列还要从OS_MAX_QS里领一个OS_Q结构这个结构在 32 位机器上是 24 字节。那实际项目里怎么选我的经验是任务之间是一对一的点对点握手用邮箱一对多、或者生产速率可能短时超过消费速率、需要缓冲的用队列。中断服务程序往任务里投数据这种场景基本无脑用队列因为中断来的时候没人等你。2. OS_EVENT 事件块的复用设计2.1 OSEventType 与 OSEventPtr 的双重身份打开uCOS_II.HOS_EVENT 的定义大概是这样typedef struct os_event { INT8U OSEventType; /* 事件类型 */ void *OSEventPtr; /* 消息 / 队列控制块 */ INT16U OSEventCnt; /* 计数器信号量用 */ OS_PRIO OSEventGrp; /* 等待任务优先级组 */ OS_PRIO OSEventTbl[OS_EVENT_TBL_SIZE];/* 等待任务优先级表 */ #if OS_EVENT_NAME_EN 0u INT8U *OSEventName; #endif } OS_EVENT;OSEventType决定了后面几个字段怎么解释OSEventType 取值对应对象OSEventPtr 的用途OS_EVENT_TYPE_UNUSED(0)未使用兼作空闲链表 next 指针OS_EVENT_TYPE_MBOX(1)邮箱指向消息OS_EVENT_TYPE_Q(2)队列指向OS_QOS_EVENT_TYPE_SEM(3)信号量未使用OS_EVENT_TYPE_MUTEX(4)互斥量指向持有者的 TCB这里有个挺精妙的复用空闲事件块和已使用事件块共用同一块内存。OSInit()的时候所有事件块被串成一条单向链表OSEventPtr当 next 指针用从OSEventFreeList里摘一个出来之后OSEventPtr立刻换岗去指消息或者队列。这种“同一片内存不同阶段干不同事”的做法在嵌入式里非常常见好处是零额外开销代价是调试的时候不能想当然地看某个字段。顺带说一句早期版本里事件标志组也是塞在 OS_EVENT 里的OS_EVENT_TYPE_FLAG2.86 之后事件标志独立成了OS_FLAG_GRP结构。所以你在一些老代码里看到OS_EVENT_TYPE_FLAG别以为是笔误那是历史遗留。2.2 等待任务表位图比链表快在哪OSEventGrp加OSEventTbl[]这套东西跟内核里的就绪表OSRdyGrp/OSRdyTbl[]是同一个套路8 位分组每组 8 位一共支撑 64 个优先级OS_LOWEST_PRIO默认 63。挂上一个任务、摘掉一个任务都是位运算OS_EventTaskWait()里就三行pevent-OSEventTbl[OSTCBCur-OSTCBY] | OSTCBCur-OSTCBBitX; /* 置位 */ pevent-OSEventGrp | OSTCBCur-OSTCBBitY; /* 同时把自己从就绪表里摘掉 */ OSRdyTbl[y] (OS_PRIO)~OSTCBCur-OSTCBBitX; if (OSRdyTbl[y] 0u) { OSRdyGrp (OS_PRIO)~OSTCBCur-OSTCBBitY; }为什么不用链表因为事件唤醒的时候要立刻找到等待者中优先级最高的那一个。用链表就得从头遍历最坏 O(n)用位图配合OSUnMapTbl[]查表两步查表拿到最高优先级是常数时间。RTOS 里事件唤醒这件事可能发生在中断上下文时间必须可预期链表那点便利换不来这个确定性。代价也很直白每个事件块固定吃掉 9 个字节的等待表1 字节 Grp 8 字节 Tbl跟这个事件实际挂了多少个任务无关。所以OS_MAX_EVENTS别乱开大一个 20 字节左右的事件块乘上几十个RAM 里就是小一千字节在 GD32F103 这种 48KB SRAM 的片子上不是小数目。3. OSMbox 源码逐行拆3.1 OSMboxCreate创建时就允许带一条消息OS_EVENT *OSMboxCreate (void *pmsg) { OS_EVENT *pevent; OS_ENTER_CRITICAL(); pevent OSEventFreeList; /* 从空闲链表摘一个 */ if (OSEventFreeList ! (OS_EVENT *)0) { OSEventFreeList (OS_EVENT *)OSEventFreeList-OSEventPtr; } OS_EXIT_CRITICAL(); if (pevent ! (OS_EVENT *)0) { pevent-OSEventType OS_EVENT_TYPE_MBOX; pevent-OSEventCnt 0u; pevent-OSEventPtr pmsg; /* 初始消息 */ OS_EventWaitListInit(pevent); /* 清空等待表 */ } return (pevent); }这段代码有两个地方新手容易忽略。一是创建失败返回 NULL因为没有空闲事件块了。很多人的代码里OSMboxCreate()的返回值从来不判断跑着跑着OS_MAX_EVENTS不够拿回来一个空指针后面一 Post 就崩在事件类型校验上。二是参数 pmsg 可以不为 NULL。这意味着你可以在OSInit()之后、多任务还没启动之前先把第一条消息塞进去。系统启动初期任务按优先级依次调度第一个跑起来的任务 Pend 就能直接拿到不需要额外搞一个“初始化完成”标志。提示OS_EventWaitListInit()做的事就是把OSEventGrp清零、OSEventTbl[]全清、OSEventCnt归零。它不碰OSEventPtr和OSEventType所以必须在这之前把这两个字段设好顺序不能反。3.2 OSMboxPend 的三条返回路径OSMboxPend()是邮箱模块里最长的一个函数逻辑上分成三段void *OSMboxPend (OS_EVENT *pevent, INT16U timeout, INT8U *perr) { void *pmsg; /* ...省略参数检查... */ OS_ENTER_CRITICAL(); pmsg pevent-OSEventPtr; if (pmsg ! (void *)0) { /* 路径一邮箱里有货直接拿走 */ pevent-OSEventPtr (void *)0; /* 取走即清空 */ OS_EXIT_CRITICAL(); *perr OS_ERR_NONE; return (pmsg); } OSTCBCur-OSTCBStat | OS_STAT_MBOX; /* 路径二没货挂起自己 */ OSTCBCur-OSTCBStatPend OS_STAT_PEND_OK; OSTCBCur-OSTCBDly timeout; OS_EventTaskWait(pevent); OS_EXIT_CRITICAL(); OS_Sched(); /* 让出 CPU */ OS_ENTER_CRITICAL(); switch (OSTCBCur-OSTCBStatPend) { case OS_STAT_PEND_OK: /* 被别人 Post 唤醒 */ pmsg OSTCBCur-OSTCBMsg; *perr OS_ERR_NONE; break; case OS_STAT_PEND_ABORT: /* 被强制取消等待 */ pmsg (void *)0; *perr OS_ERR_PEND_ABORT; break; case OS_STAT_PEND_TO: default: /* 路径三超时 */ OS_EventTaskRemove(OSTCBCur, pevent); pmsg (void *)0; *perr OS_ERR_TIMEOUT; break; } OSTCBCur-OSTCBStat OS_STAT_RDY; OSTCBCur-OSTCBStatPend OS_STAT_PEND_OK; OSTCBCur-OSTCBEventPtr (OS_EVENT *)0; OS_EXIT_CRITICAL(); return (pmsg); }这里最值得琢磨的是路径二为什么不直接读OSEventPtr而要读OSTCBMsg。答案在于邮箱只有一个槽位。假设OSMboxPost()的时候把消息写进pevent-OSEventPtr那么被唤醒的任务还没被调度上 CPU第二个任务调OSMboxPend()就会看到邮箱里“有消息”又拿走了同一条。所以 uC/OS-II 的做法是有任务在等的时候消息直接塞进那个任务的 TCBOSTCBMsg字段邮箱本身保持“空”的状态直到下次有人 Post。这个设计在OS_EventTaskRdy()里能看到对应代码。另外注意timeout参数传 0 的含义是无限等待不是立即返回。想要非阻塞查询得用OSMboxAccept()如果这个配置项打开了。3.3 OSMboxPost 的两条投递分支INT8U OSMboxPost (OS_EVENT *pevent, void *pmsg) { if (OSIntNesting 0u) { return (OS_ERR_POST_ISR); /* 中断里禁止调用 */ } if (pevent-OSEventType ! OS_EVENT_TYPE_MBOX) { return (OS_ERR_EVENT_TYPE); } if (pmsg (void *)0) { return (OS_ERR_POST_NULL_PTR); /* 不允许投递空指针 */ } OS_ENTER_CRITICAL(); if (pevent-OSEventGrp ! 0u) { /* 有人在等 */ (void)OS_EventTaskRdy(pevent, pmsg, OS_STAT_MBOX, OS_STAT_PEND_OK); OS_EXIT_CRITICAL(); OS_Sched(); /* 唤醒的是高优先级就要切换 */ return (OS_ERR_NONE); } if (pevent-OSEventPtr ! (void *)0) { /* 没人在等且邮箱已满 */ OS_EXIT_CRITICAL(); return (OS_ERR_MBOX_FULL); } pevent-OSEventPtr pmsg; /* 没人在等先存着 */ OS_EXIT_CRITICAL(); return (OS_ERR_NONE); }三个校验里OS_ERR_POST_NULL_PTR这条特别容易踩。因为邮箱用“OSEventPtr NULL”来表示“空”你要是真往里面 Post 一个空指针邮箱的状态就乱了——下次 Pend 会认为没货继续等等到天亮。所以内核干脆在入口就把这条路堵死。OS_ERR_POST_ISR这条限制也值得说道。中断里想投递邮箱得改用OSMboxPostOpt()它的实现里没有这个 ISR 检查同时多了两个可选行为OS_POST_OPT_NOWAIT不触发任务切换留给OSIntExit()统一调度和OS_POST_OPT_BROADCAST广播给所有等待者只对事件标志有意义用在邮箱上相当于只唤醒一个。写串口中断服务程序的时候这个区别就是能不能正常收发数据的分水岭。4. OSQ 源码逐行拆4.1 OS_Q 控制块与那个哨兵地址typedef struct os_q { struct os_q *OSQPtr; /* 空闲队列链表的 next 指针 */ void **OSQStart; /* 用户数组起始地址 */ void **OSQEnd; /* 用户数组结束地址哨兵 */ void **OSQIn; /* 入队指针 */ void **OSQOut; /* 出队指针 */ INT16U OSQSize; /* 队列容量 */ INT16U OSQEntries; /* 当前条目数 */ } OS_Q;这个结构在 32 位机上是 24 字节前 20 字节是五个指针后面 4 字节是两个INT16U。它和 OS_EVENT 一样玩“空闲时当链表节点”的把戏OSInit()阶段OS_MAX_QS个OS_Q被串成OSQFreeListOSQPtr当 next 用OSQCreate()摘一个出来之后OSQPtr就不管了。OSQEnd的含义要特别注意pq-OSQStart start; pq-OSQEnd start[size]; /* 注意是 start[size]不是 start[size-1] */ pq-OSQIn start; pq-OSQOut start; pq-OSQSize size; pq-OSQEntries 0u;OSQEnd指向的是数组最后一个元素的下一个位置也就是个哨兵。有了它OSQIn和OSQOut推进时只需要一句if (OSQIn OSQEnd) OSQIn OSQStart;就能完成回绕不需要在每次移动时做取模运算。取模在 Cortex-M3 上要几十个周期而这个判断只要一两个周期。中断里跑的代码省下来的都是实打实的响应时间。所以数组该怎么开static void *UartQTbl[32]; /* 32 个指针占 128 字节 */ UartQ OSQCreate(UartQTbl[0], 32); /* size 传 32 */数组开 32 个元素size 也传 32这是对的。别自作聪明开成 33 个多出来的那格永远不会被写到。4.2 OSQPost 的入队与溢出判断INT8U OSQPost (OS_EVENT *pevent, void *pmsg) { OS_Q *pq; /* ...参数校验... */ OS_ENTER_CRITICAL(); if (pevent-OSEventGrp ! 0u) { /* 有任务在等直接给它 */ (void)OS_EventTaskRdy(pevent, pmsg, OS_STAT_Q, OS_STAT_PEND_OK); OS_EXIT_CRITICAL(); OS_Sched(); return (OS_ERR_NONE); } pq (OS_Q *)pevent-OSEventPtr; if (pq-OSQEntries pq-OSQSize) { /* 判断在前 */ OS_EXIT_CRITICAL(); return (OS_ERR_Q_FULL); } *pq-OSQIn pmsg; /* 写入并推进指针 */ pq-OSQEntries; if (pq-OSQIn pq-OSQEnd) { /* 回绕在后 */ pq-OSQIn pq-OSQStart; } OS_EXIT_CRITICAL(); return (OS_ERR_NONE); }这段代码的四个动作顺序是固定的先判满、再写入、再计数、最后回绕。顺序换任何一个都会出问题。比如把回绕挪到写入之前写入的地址就错了把计数放到回绕之后中断打断时读到的OSQEntries和实际指针状态对不上。判满用而不是是因为OSQEntries的最大合法值就是OSQSize两者相等表示满。还有个细节OSEventPtr指向队列控制块而不是消息本身。所以队列的“空/满”状态由OSQEntries决定跟OSEventPtr没关系。这跟邮箱的设计是反过来的也是邮箱和队列没法直接互相替代的根本原因。4.3 OSQPend 的出队与超时收尾void *OSQPend (OS_EVENT *pevent, INT16U timeout, INT8U *perr) { void *pmsg; OS_Q *pq; /* ...参数校验... */ OS_ENTER_CRITICAL(); pq (OS_Q *)pevent-OSEventPtr; if (pq-OSQEntries 0u) { /* 队列非空 */ pmsg *pq-OSQOut; /* 读出并推进 */ pq-OSQEntries--; if (pq-OSQOut pq-OSQEnd) { pq-OSQOut pq-OSQStart; } OS_EXIT_CRITICAL(); *perr OS_ERR_NONE; return (pmsg); } OSTCBCur-OSTCBStat | OS_STAT_Q; /* 队列空挂起 */ OSTCBCur-OSTCBStatPend OS_STAT_PEND_OK; OSTCBCur-OSTCBDly timeout; OS_EventTaskWait(pevent); OS_EXIT_CRITICAL(); OS_Sched(); OS_ENTER_CRITICAL(); /* 醒来后同样是三选一POST / ABORT / TIMEOUT */ /* ...与 OSMboxPend 结构完全一致... */ }跟邮箱对比一下就能看出Pend 阶段两者的骨架几乎一模一样差别只在“非空判断”的依据——邮箱看指针是不是 NULL队列看OSQEntries是不是大于 0。这也是为什么我一直建议先把邮箱搞透队列就是多了一个控制块和一层环形缓冲的算术。有一个细节容易看漏出队时pmsg是从*pq-OSQOut读出来的然后才把OSQOut往前推。这一步如果被中断打断只要临界区保护到位就不会出问题因为进入OS_ENTER_CRITICAL()之后所有任务和大部分中断都被挡住了除非有更高优先级的中断没被屏蔽那种情况要靠OSIntEnter()/OSIntExit()配合。5. 消息投递背后的枢纽等待与唤醒5.1 OS_EventTaskWait把自己从就绪表挪到等待表void OS_EventTaskWait (OS_EVENT *pevent) { INT8U y; OSTCBCur-OSTCBEventPtr pevent; pevent-OSEventTbl[OSTCBCur-OSTCBY] | OSTCBCur-OSTCBBitX; pevent-OSEventGrp | OSTCBCur-OSTCBBitY; y OSTCBCur-OSTCBY; OSRdyTbl[y] (OS_PRIO)~OSTCBCur-OSTCBBitX; if (OSRdyTbl[y] 0u) { OSRdyGrp (OS_PRIO)~OSTCBCur-OSTCBBitY; } }这个函数干的事就四行但蕴含了一个关键约定任务的OSTCBEventPtr字段只允许指向一个事件。也就是说一个任务同一时刻只能等待一个邮箱、或者一个队列、或者一个信号量。你要是想让某个任务同时等三个事件源uC/OS-II 的邮箱和队列帮不了你得靠事件标志组或者在应用层用多个中间任务做转发。这个约束在源码里没有任何运行时检查全靠OSTCBEventPtr被反复覆盖来“自然生效”。如果哪个任务的等待逻辑写岔了OSTCBEventPtr指着一个已经被删除的事件超时唤醒的时候OS_EventTaskRemove()就会去操作一块已经还给空闲链表的内存——这类野指针在嵌入式上极难查因为崩的地方往往离出问题的地方很远。5.2 OS_EventTaskRdy唤醒谁、消息怎么给INT8U OS_EventTaskRdy (OS_EVENT *pevent, void *pmsg, INT8U msk, INT8U pend_stat) { OS_TCB *ptcb; INT8U y, x, prio; y OSUnMapTbl[pevent-OSEventGrp]; /* 找最高优先级组 */ x OSUnMapTbl[pevent-OSEventTbl[y]]; /* 找组内最高位 */ prio (INT8U)((y 3u) x); ptcb OSTCBPrioTbl[prio]; ptcb-OSTCBEventPtr (OS_EVENT *)0; ptcb-OSTCBMsg pmsg; /* 消息塞进 TCB */ ptcb-OSTCBStat (INT8U)~(INT8U)msk; ptcb-OSTCBStatPend pend_stat; if ((ptcb-OSTCBStat OS_STAT_SUSPEND) OS_STAT_RDY) { OSRdyGrp | ptcb-OSTCBBitY; /* 重新挂回就绪表 */ OSRdyTbl[y] | ptcb-OSTCBBitX; } else { ptcb-OSTCBDly 0u; /* 被挂起了不再计时 */ } OS_EventTaskRemove(ptcb, pevent); /* 从等待表摘掉 */ return (prio); }注意ptcb-OSTCBMsg pmsg;这一行。这就是邮箱和队列能传递数据的唯一通道。不管你是 Post 邮箱还是 Post 队列消息指针最终都是通过这个字段交到被唤醒任务的OSMboxPend()/OSQPend()手里的。所以前面那条“有任务在等时消息不走OSEventPtr”的规则在这里得到了解释。另外注意那个else分支如果被唤醒的任务正好处于挂起状态OSSuspend过它不能被挂回就绪表但OSTCBDly会被清零表示它的等待条件已经满足超时计时不再生效。等它被OSResume()恢复之后读到的OSTCBStatPend就是OS_STAT_PEND_OK能正常拿到消息。这个细节在看OSTaskSuspend()/OSTaskResume()那部分源码时能对上。OS_EventTaskRemove()就是OS_EventTaskWait()的逆操作把任务的位从OSEventTbl[]里清掉如果整组空了再把OSEventGrp对应位清零。三行代码但组清零那个判断不能漏否则最高优先级查找会命中一个空组。6. 在 GD32F103 上跑一个生产者消费者6.1 先把配置项理清楚移植层的东西前面几篇讲过了这里只说和消息通信相关的配置。OS_CFG.H里这几个宏必须对配置项建议值说明OS_MAX_EVENTS8~16所有事件块总数邮箱、队列、信号量、互斥量共用OS_MAX_QS4OS_Q控制块数量只统计队列OS_Q_EN1队列总开关OS_Q_DEL_EN0不需要运行时删队列就关掉省代码OS_Q_QUERY_EN1调试时看队列里剩几条很好用OS_MBOX_EN1邮箱总开关OS_MBOX_POST_EN1关掉的话OSMboxPost()编不进来OS_MBOX_POST_OPT_EN1中断投递必须开OS_MBOX_DEL_EN0同上不用就关OS_MBOX_QUERY_EN1调试用OS_TASK_DEL_EN1OSQDel/OSMboxDel里要用到OSTaskDel相关逻辑OS_EVENT_NAME_EN1给事件起名用调试器看变量时一目了然OS_MAX_EVENTS和OS_MAX_QS是两本账别搞混。OSQCreate()要同时从两个池子里各领一个块任意一个池子空了都会返回 NULL。我见过有人把OS_MAX_QS设成 8、OS_MAX_EVENTS设成 4然后创建第 5 个队列的时候失败查了半天以为是内存不够。6.2 完整的生产消费骨架下面这段代码我在 GD32F103 上跑过串口 1 收数据、串口 2 转发中间加了两级缓冲。#include includes.h #define UART_Q_SIZE 32 #define FRAME_POOL_SIZE 8 #define FRAME_BUF_SIZE 16 #define PROD_PRIO 10 #define CONS_PRIO 11 #define TASK_STK_SIZE 256 static OS_STK ProdStk[TASK_STK_SIZE]; static OS_STK ConsStk[TASK_STK_SIZE]; static void *UartQTbl[UART_Q_SIZE]; /* 队列用的指针数组 */ static OS_EVENT *UartQ; /* 队列事件 */ typedef struct { INT8U len; INT8U buf[FRAME_BUF_SIZE]; } UartFrame; static UartFrame FramePool[FRAME_POOL_SIZE]; static OS_MEM *FrameMem; /* 内存池第 9 篇细讲 */ /* 生产者从串口 FIFO 取一帧装箱后投递 */ void ProdTask (void *p_arg) { INT8U err; UartFrame *pf; (void)p_arg; for (;;) { pf (UartFrame *)OSMemGet(FrameMem, err); if (err OS_ERR_NONE) { pf-len Uart_PopFrame(pf-buf, FRAME_BUF_SIZE); if (pf-len 0u) { OSQPost(UartQ, (void *)pf); /* 投递指针不拷贝数据 */ } else { (void)OSMemPut(FrameMem, (void *)pf); /* 空帧直接还回去 */ } } OSTimeDlyHMSM(0, 0, 0, 20); /* 20ms 轮询一次 */ } } /* 消费者拿到就用用完归还内存池 */ void ConsTask (void *p_arg) { INT8U err; UartFrame *pf; (void)p_arg; for (;;) { pf (UartFrame *)OSQPend(UartQ, 0u, err); /* 0 表示无限等 */ if (err OS_ERR_NONE) { Uart_SendFrame(pf-buf, pf-len); /* 转发到串口 2 */ (void)OSMemPut(FrameMem, (void *)pf); /* 关键谁拿谁还 */ } } } int main (void) { INT8U err; BSP_Init(); OSInit(); FrameMem OSMemCreate(FramePool[0], FRAME_POOL_SIZE, sizeof(UartFrame), err); UartQ OSQCreate(UartQTbl[0], UART_Q_SIZE); if ((FrameMem (OS_MEM *)0) || (UartQ (OS_EVENT *)0)) { while (1) { } /* 初始化失败死等 */ } OSTaskCreate(ProdTask, (void *)0, ProdStk[TASK_STK_SIZE - 1], PROD_PRIO); OSTaskCreate(ConsTask, (void *)0, ConsStk[TASK_STK_SIZE - 1], CONS_PRIO); OSStart(); return 0; }这段代码里有三个设计决策值得解释一下。为什么传指针而不是拷贝整个结构体。OSQPost()存的只是一个void *队列本身不关心指针指向什么。传指针的好处是零拷贝、速度快中断里投递一个 4 字节指针代价可以忽略。代价是缓冲区的生命周期管理转移给了应用层——这个责任必须落到“拿到消息的那个任务”头上。为什么用内存池而不是 malloc。每次 Post 前 malloc 一块、Pend 后 free在裸机上跑不了多久就会碎片化尤其是有大小不一的结构体混在一起的时候。用固定大小的内存池分配释放都是 O(1)时间可预期且不会产生碎片。第 9 篇会专门拆OSMemGet/OSMemPut这里先按“它就是一组固定大小的块”理解就行。为什么消费者优先级比生产者低。生产者是 10消费者是 11。串口数据来的速度可能是突发的生产者优先级高一点能及时把数据抢进来。但这里有个隐患如果生产者一直来数据消费者永远排不上队列就会满。所以生产者的循环末尾加了OSTimeDlyHMSM(0,0,0,20)主动让出 CPU。这个 20ms 不是拍脑袋定的是算出来的队列 32 个槽每 20ms 最多产生一帧理论上 640ms 内不会满而消费者处理一帧大概 1ms 以内完全跟得上。队列深度和轮询周期的关系就是这么倒推出来的。7. 常见问题与排查技巧实录7.1 故障速查表现象最可能的原因排查动作OSQCreate()返回 NULLOS_MAX_EVENTS或OS_MAX_QS耗尽统计全工程创建了多少事件和队列查有没有创建失败没处理的分支OSMboxCreate()返回 NULLOS_MAX_EVENTS不够或者OSInit()没调用确认OSInit()在创建之前执行OSQPost()一直返回OS_ERR_Q_FULL消费者卡住或队列太小用OSQQuery()看OSNMsgs是不是长期顶格查消费者任务状态OSMboxPost()返回OS_ERR_POST_ISR在中断里调了OSMboxPost换成OSMboxPostOpt(pevent, pmsg, OS_POST_OPT_NOWAIT)OSMboxPost()返回OS_ERR_POST_NULL_PTR投递了空指针上游指针没赋值成功检查OSMemGet的返回值Pend 永远拿不到数据消息投递到了别的事件或者等待的是不同的队列用调试器看OSTCBEventPtr指向哪个事件块偶发死机、栈溢出队列传指针后多个任务同时改了同一块内存确认所有权唯一或者改用值传递系统跑一段时间后创建事件失败有事件被 Del 了但没回收检查OS_EVENT_TYPE_UNUSED的块有没有回到空闲链表7.2 几条只能靠踩坑得到的经验第一条邮箱满了不覆盖这件事一定要在协议设计阶段就考虑。我在一个项目里用邮箱传“紧急停机”命令正常状态下邮箱是空的Post 进去等着即将被 Pend。结果有一次 Post 完之后负责 Pend 的任务因为另一个事件被阻塞了 200ms第二个紧急命令又来了OSMboxPost()返回OS_ERR_MBOX_FULL调用方没检查返回值命令就丢了。这类“高优先级消息被丢”的故障在测试环境里几乎复现不出来因为时序太特殊。后来我改成队列容量给 4丢包变成了排队才彻底解决。第二条OSQPend的 timeout 参数单位是时钟节拍不是毫秒。默认OS_TICKS_PER_SEC是 100也就是 10ms 一个节拍。你想等 100ms 得传 10。我在早期代码里直接写OSQPend(Q, 100, err)心里想的是 100ms实际等了 1 秒调试的时候一脸懵。要么养成用OS_TICKS_PER_SEC做换算的习惯要么写个宏包一层。第三条事件块删除之后全局指针一定要置 NULL。OSQDel()会把OS_Q块和OS_EVENT块分别还给空闲链表。如果全局变量UartQ还留着那个地址理论上后面创建别的事件可能复用同一块内存你就OSQPost(UartQ, ...)就投跑到别人的队列里去了或者更糟投到一个信号量上。OSQDel()的实现里会检查事件类型所以大概率会返回OS_ERR_EVENT_TYPE但要是那块内存恰好被复用成了新队列类型校验就过了然后你就看着一个完全不相干的队列里冒出数据来。第四条调试时打开OS_Q_QUERY_EN和OS_MBOX_QUERY_EN太值了。OSQQuery()返回的OS_Q_DATA里有OSNMsgs当前条目数、OSQSize容量、OSMsg如果是在等的话指向任务拿到的消息。用调试器把某个队列的这几个值加进 watch 窗口生产的速率和消费的速率一眼就看出来了。比打印日志方便也不影响实时性。第五条别在一个任务里同时 Pend 两个队列。前面说过了OSTCBEventPtr只能存一个事件指针。你要是写了OSQPend(Q1, ...)然后紧接着OSQPend(Q2, ...)这是顺序等待没问题但如果你想“谁先来等谁”得用别的手段。这个约束在 uC/OS-II 里没有编译期检查纯粹靠开发者记住。第六条队列数组建议加static并且初始化为零。队列数组里的指针在取出之前是未定义的如果你在OSQCreate()之前就有人 Pend内核不会去读数组所以不会出事但调试器里看到的是一堆随机值容易让人误判。加上static至少保证了启动时是零。这个习惯在排查问题时能省不少时间。第七条中断里投递队列之后别自己算优先级切换。uC/OS-II 的中断处理流程是这样进中断先OSIntEnter()投递用OSQPost()或者带 Opt 的版本出中断调OSIntExit()。OSIntExit()会判断有没有更高优先级的任务变成就绪如果有就触发任务切换。你不能在中断里自己去操作就绪表那是内核的活。这条规则在串口 DMA 接收中断里特别要注意很多人的第一版代码就是在中断里直接调OSTaskResume()跑起来看着正常压测就崩。这一篇把OSMbox.c和OSQ.c两个文件从头到尾过了一遍算下来有效代码量不到 400 行但里面每一个判断、每一行临界区都有它的道理。我个人读源码的习惯是先把结构体定义和那几个宏翻明白再去读函数实现因为结构体决定了数据怎么组织函数只是在对这个组织做操作。消息通信这块的结构其实就两个——OS_EVENT管等待队列OS_Q管环形缓冲剩下全是围绕它们的增删改查。第 9 篇打算接着往下走拆内存管理OSMem.c那个文件里有一个特别容易搞错的细节内存池的分区链表在分配和释放时怎么维护以及为什么OSMemCreate()的块大小要按 4 字节对齐。如果你在项目里正好被动态分配碎片化折腾过那篇应该对你胃口。