我在这里写第八篇不是来讲调度器的——那玩意儿下一章再说。我先把DSH内核里一个藏得比较深、但几乎每天都会碰到的机制拆开AgentLoop。简单说它就是驱动所有Agent的那只循环手。在DSH里Agent不是传统意义上的“用户进程”而是一段被纳入统一管理的内核服务单元比如控制台、块设备、定时器、甚至未来的网络栈。你会在很多嵌入式内核里看到类似设计叫法可能是shell driver或者task runner。这篇文章适合谁如果你正在从零写一个微内核、裸机任务框架或者被一堆while(1)的模块互相抢占折磨过这篇正好对症。在DSH内核里AgentLoop承担三件事记录有哪些Agent活着、决定下一个该哪个Agent跑、以及把消息从一个Agent搬到另一个Agent。它本身不干具体业务更像内核里的包工头所有脏活累活都由Agent自己干但谁干、什么时候干、消息怎么递到手里由AgentLoop统一调度。前七篇我们已经把内存、中断、定时器这些基础件搭起来了这一篇补上的就是让这些零件能协作起来的那根轴。1. 为什么DSH内核需要一个AgentLoop从裸奔的Agent到统一调度1.1 早期原型里的野蛮生长在AgentLoop出现之前DSH内核的各个模块各自为政。最典型的例子是定时器服务和串口输出定时器模块在while(1)里轮询待超时的节点串口模块也在自己的while(1)里等着环形缓冲区里有新字节。表面上它们各跑各的但两个循环都会修改同一个全局变量——比如时钟节拍计数于是一旦串口阻塞定时器立刻跟着卡住。我最早调试时撞见过一个特别玄学的现象内核一启动打印到某一行就停住按键还有反应但定时器不再触发。后来发现串口Agent里面的发送循环死死占着CPU定时器Agent根本没有机会推进。这就是裸奔Agent的经典问题——你以为自己在做模块化实际上只是把几个死循环拼在了一起它们之间靠运气协作。这种野蛮生长在原型阶段不算致命因为功能少每次只验证一条链路。可一旦我要同时跑console、ext2文件系统、CMD解析器就立刻变得不可控谁先谁后、等待多久、消息怎么转交全部要靠人为编排。任何一个模块里的delay都会拖垮全系统。1.2 Agent到底指什么进程还是服务先澄清一个边界问题DSH里的Agent不是用户态进程它没有独立的进程地址空间也不依赖MMU做隔离。一个Agent就是一个c函数入口 一份私有状态 一个信箱。它运行在内核态共享同一个栈空间由AgentLoop逐个调用。为什么要叫Agent而不是任务或线程因为它的运行模型更接近事件驱动Agent平时不占CPU只有两种情况会被唤醒——要么收到了消息要么注册的定时器到期。每次被调度时AgentLoop只调用一次它的entry函数entry处理完当下事件后返回Agent继续回到休眠状态。这种模型和Reactor模式、协程调度很像但放在内核里它让我能非常克制地管理运行时间而不是让一个Agent无限循环下去。所以Agent的驱动包含两层Loop把运行权交给它同时Loop也负责把事件送到它面前。Agent是一个被驱动者驱动者永远是那个Loop。1.3 引入AgentLoop的三个直接原因第一多个Agent需要有序地共享CPU。哪怕只是console和timer两个Agent也需要有人决定现在跑谁。AgentLoop引入优先级队列让高优先级的实时Agent比如时钟维护不会被低优先级的打印Agent拖住。第二Agent之间必须通信。文件系统写完数据要通知块设备刷盘块设备完成后又要通知文件系统收尾。如果让它们直接互相调用函数调用关系会乱成一团而且依赖链一深就很容易发生重入。AgentLoop提供一个邮局所有通信变成发消息发送方不需要知道接收方此刻在哪个上下文。第三需要在不重启内核的情况下增加或移除Agent。调试内核的时候我经常要临时挂一个统计Agent或者替换一个故障的驱动Agent。AgentLoop的注册表让Agent可以动态加载和卸载只要保证当前没有消息停留在它信箱里。2. AgentLoop的三件套注册表、调度器、消息管道2.1 注册表AgentControlBlock的诞生每个Agent在注册时对应一个控制块在DSH里叫dsh_agent_t。这个ACB是所有调度决策的数据基础我把它设计成固定大小、放在全局数组里而不是动态分配——因为内核启动阶段内存分配器还不完全可靠静态上限能帮我提前算出需要多少内存。#define DSH_MAX_AGENTS 32 #define DSH_AGENT_NAME_LEN 16 #define DSH_AGENT_STATE_MASK 0x0F typedef struct dsh_agent { int id; char name[DSH_AGENT_NAME_LEN]; int (*entry)(struct dsh_agent *self, const struct dsh_msg *msg); void *private_data; uint8_t priority; /* 0~255数值越小优先级越高 */ uint8_t state; /* READY/RUNNING/BLOCKED/EXIT等 */ uint16_t tick_quota; /* 单次执行允许消耗的tick上限 */ uint32_t tick_used; /* 本次执行已经用掉的tick */ uint64_t wait_until; /* 阻塞唤醒的时间点 */ struct dsh_mailbox *mbox; uint64_t total_runs; uint64_t total_ticks; uint32_t last_ret; } dsh_agent_t;字段不是随手加的每一个都有针对性。priority解决调度先后tick_quota和tick_used是给协作式调度加软限制防止一个Agent在entry里待太久wait_until用于定时阻塞比如一个延迟任务可以设置等待10ms后再加入ready队列total_runs和total_ticks则是调试和性能剖析的关键数据。2.2 调度器从轮询所有Agent到按紧急程度跑如果每个Agent按ID轮询最直接的问题就是低编号Agent里有高优先级任务时后面的Agent可能要等很久。于是我把下一个该跑谁从线扫描改成了优先级位图。假设Agent的数组下标就是优先级顺序下标0优先级最高。维护一个32位的ready_bitmap第N位为1表示第N个Agent当前可运行。查找下一个可运行Agent只需要一条指令static uint32_t ready_bitmap; static dsh_agent_t *agent_table[DSH_MAX_AGENTS]; static dsh_agent_t *pick_next_agent(void) { unsigned long idx __builtin_ctz(ready_bitmap); if (idx DSH_MAX_AGENTS) return NULL; return agent_table[idx]; }注意__builtin_ctz是GCC内建的count trailing zeros它直接算出最低位1的位置时间复杂度O(1)。ARM上会编译成CLZ相关指令RISC-V上有类似指令非常划算。比遍历数组快一个数量级在真实硬件上调度开销可以压到几十个周期。但位图本身不解决优先级高的Agent永远霸占CPU的问题这个后面在第4章会配合老化机制一起说。2.3 消息管道Agent之间的邮局Agent之间如果直接调用函数相当于A把工作塞给B去执行B的上下文就不是由Loop控制了。DSH里我规定Agent之间只能通过消息发生关系。每个Agent有一个dsh_mailbox本质是固定容量的环形缓冲。消息本体不是动态分配的而是从一个全局消息池msg_pool里取出预定义大小的节点。typedef struct dsh_msg { uint16_t src; uint16_t dst; uint16_t type; uint16_t rlen; uint32_t cookie; /* 同步会话ID */ uint32_t payload[DSH_MSG_MAX_PAYLOAD]; } dsh_msg_t;我在早期想过零拷贝传指针这样大payload不用复制。但后来否了如果发送者把数据用完就释放接收者还没来得及处理引用就悬空了如果接收者用完再释放又得引入引用计数和所有权转移复杂度直线上升。拷贝一代价值的处理牺牲一小部分内存换来消息生命周期完全明确。在绝大多数内核实务场景里Agent间消息体都不大几十字节的拷贝完全可接受。3. 核心API设计从注册到退出3.1 注册和注销一个ACB的生命周期注册一个Agent非常简单把ACB填充好之后交给系统static dsh_agent_t my_agent { .name mycom, .entry mycom_entry, .priority 40, .private_data my_stuff, }; int dsh_agent_register(dsh_agent_t *acb);注册函数会做三件事从agent_table找一个空位分配id把状态置为READY并把对就位图里对应位置1。这里有个我一开始忽略的细节注册时entry函数里千万不要立刻访问全局资源因为Agent可能在任意时刻被调度而依赖的子系统可能还没完成初始化。我的建议是在注册函数里只初始化private_data真正的资源获取放到第一次进入entry时通过state机做延迟初始化。注销比注册麻烦因为必须保证Agent此刻没有正在运行、没有堆积消息。int dsh_agent_unregister(int id);实现时Loop先设置一个DSH_AGENT_STATE_STOPPING标志然后把该Agent从ready位图里清掉最后把mailbox里的旧消息全部回收。如果Agent正好在RUNNING状态不能立刻注销。我的处理是记录一个quiesce_request等该Agent本次entry返回后再由Loop替它执行清理。这个机制避免了一个正在跑的Agent突然把栈释放掉的惨案。3.2 Agent的状态机状态迁移是整个AgentLoop的逻辑核心用一张表就能看明白状态进入条件离开条件READY注册完成或被消息/定时器唤醒被pick_next_agent选中进入RUNNINGRUNNINGLoop完成上下文交接entry返回若是STOP则EXIT否则回到READYBLOCKEDentry调用dsh_agent_sleep_ms等待定时器定时到Loop复位状态为READYWAIT_REPLYentry调用dsh_agent_call等待同步回复收到reply或超时转为READYEXITentry返回DSH_AGENT_STOP且Loop完成注销清理ACB可以被复用每个Agent的entry函数返回值表达自己的意愿#define DSH_AGENT_CONTINUE 0 /* 处理完毕可以再次调度 */ #define DSH_AGENT_SLEEP 1 /* 主动让出等待定时器或消息 */ #define DSH_AGENT_STOP 2 /* 请求退出 */如果Agent想等待下一次消息它不需要故意阻塞自己。只要返回CONTINUELoop就会检查它的mailbox队列如果空则自动把它从ready位图清掉。这个细节非常关键一个Agent在调度后若没有工作必须被移出ready集合否则会浪费一次调度循环。3.3 一个最简Agent实现下面这个hello_timerAgent每隔100ms向调试台打印一次heartbeat它完整展示了注册、入口函数、定时等待的写法static int heartbeat_entry(dsh_agent_t *self, const dsh_msg_t *msg) { static uint32_t counter 0; if (msg msg-type DSH_MSG_TIMEOUT) { debug_printf([%u] HB %u\n, dsh_uptime_ms(), counter); dsh_agent_sleep_ms(self, 100); } else if (msg msg-type DSH_MSG_CALL) { dsh_agent_reply(self, msg, NULL, 0); } return DSH_AGENT_CONTINUE; } static dsh_agent_t heartbeat_agent { .name heartbeat, .entry heartbeat_entry, .priority 120, }; void heartbeat_init(void) { dsh_agent_register(heartbeat_agent); dsh_agent_post(DSH_AGENT_ID_LOOP, heartbeat_agent.id, DSH_MSG_TIMEOUT, NULL, 0); }注意我第一次就是这么写的结果发现了两个问题如果Agent刚注册时自己给自己发一条TIMEOUT消息这条消息会在注册前被放到mailbox里注册时不会丢但如果你在dsh_agent_register之后才postAgent可能在post之前已经被调度跑了一次那次它看到空mailbox会什么都不做。所以DSH里约定初始化消息要在注册前post靠消息队列暂存这个约定不算优雅但很实用。4. 调度轮转时间配额、优先级老化与空转节电4.1 协作式主循环AgentLoop的日常AgentLoop本身是一个永不返回的while(1)但它的责任不是占用CPU而是快速决定下一步void dsh_agent_loop_run(void) { while (dsh_loop_running) { dsh_agent_t *next pick_next_agent(); if (!next) { agent_loop_idle(); /* 没有任务进入低功耗 */ continue; } ready_bitmap ~(1UL next-id); next-state DSH_AGENT_STATE_RUNNING; next-tick_used 0; uint64_t t0 dsh_timer_get_us(); int ret next-entry(next, dsh_agent_peek(next-id)); uint64_t dt dsh_timer_get_us() - t0; next-total_runs; next-total_ticks dt; if (ret DSH_AGENT_STOP) { next-state DSH_AGENT_STATE_EXIT; dsh_agent_cleanup(next-id); } else if (ret DSH_AGENT_CONTINUE) { if (next-mbox-count 0 || next-wait_until dsh_uptime_ms()) { ready_bitmap | (1UL next-id); } } next-state DSH_AGENT_STATE_READY; } }这段逻辑里最关键的是Loop调用entry时并没有刻意关中断所以一个Agent运行期间中断照常发生。这样做的好处是定时器中断仍然会维护dsh_uptime_ms坏处是Agent的entry必须做到可重入安全——如果它在运行中被自己的中断回调调用就会出问题。我的约定是Agent的entry内不能访问同一个Agent的mailbox锁否则会产生死锁。4.2 时间配额给协作式调度加个软约束DSH的AgentLoop是协作式的没有完整的上下文切换因此Loop无法像抢占式调度器那样从外部打断一个Agent。那怎么防止Agent在entry里while(1)我采用的方法是时间配额记账tick quota。在每次进入entry前记下开始时间t0entry返回后计算耗时累加到tick_used。如果tick_used超过配额Loop就把这次运行标成超限并把该Agent的优先级临时降低一个档位。如果连续超过三次系统打印告警并强制在下一轮调度中跳过它一次。这个机制不是真正的强制隔离但它能把长期占用CPU的Agent暴露出来。配合第6章的观察手段很容易定位到是哪一个Agent在耍流氓。真正要硬割断一个正在死循环的Agent需要完整的任务上下文切换那个复杂度放到进程调度那篇再展开目前AgentLoop的软约束已经够用。4.3 优先级老化别让低优先级Agent饿死位图调度有个天然缺陷如果高优先级Agent每轮都ready低优先级可能永远选不中。DSH用老化解决。实现很粗暴但有效Loop每10ms扫描一次把所有state READY但优先级高于一定阈值的Agent的priority临时加1数值变大优先级变低超过高水位再重置。如果某个低优先级Agent等了很久还没被选到它的实际优先级会逐渐上升直到获得运行机会。void agent_loop_aging(void) { static uint32_t aging_clock; if (aging_clock % AGING_PERIOD_TICKS) return; for (int i 0; i DSH_MAX_AGENTS; i) { dsh_agent_t *a agent_table[i]; if (!a || a-state ! DSH_AGENT_STATE_READY) continue; if (a-age DSH_AGING_THRESHOLD) a-priority MIN(a-priority 1, 255); } }这样一个等待超过阈值的Agent会升级到接近高优先级最终被调度到。代价是位图不再一劳永逸因为优先级变了以后需要更新位图。我为此在AgentLoop里增加了一个rebuild_ready_bitmap()在老化步骤之后调用一次保证位图和实际优先级一致。这部分代码量不大但容易写错调试时一定要打印每个Agent的历史等待时间。4.4 空转与低功耗没有任务时就halt当所有Agent都没有ready调度器不能原地空转浪费CPU否则内核功耗和发热都难看。DSH里用一条指令进入低功耗等待static void agent_loop_idle(void) { irq_disable(); if (!ready_bitmap !dsh_loop_pending_messages) { cpu_halt(); /* 等待中断唤醒 */ } irq_enable(); }这里有个坑cpu_halt之前必须确认中断已经启用否则任何中断都来不了系统就永远睡过去了。代码里我先把中断关了检查完状态再重新打开然后马上halt。由于irq_enable后和halt之间可能被中断打断需要用dsb/wfi组合或者处理器的enable_irq_and_wfi原子指令来避免竞态。如果你在自己的内核里实现空转务必查芯片手册里WFI/WFE的原子用法而不是盲写关中断再开中断。5. 消息投递同步与异步邮局不发空信5.1 异步投递post与post_defer异步消息是AgentLoop的主干API很直接int dsh_agent_post(uint16_t dst, uint16_t type, const void *payload, uint16_t rlen);内部流程从消息池取一个空闲dsh_msg_t填上本Agent id作为src拷入payload把消息挂到dst的mailbox环形队列尾然后把dst的ready位图置1。这个过程中mailbox操作必须加锁或者在临界区执行。在单核DSH上我直接采用关中断保护临界区比自旋锁简单。如果你在中断上下文里调dsh_agent_post比如串口收到一个字节后通知串口处理Agent我会推荐你走post_deferint dsh_agent_post_defer(uint16_t dst, uint16_t type, const void *payload, uint16_t rlen);post和post_defer的区别在于post立即把消息塞进目标信箱post_defer只把消息放入一个pending队列等中断退出后由Loop在主上下文统一投递。为什么要区分因为串口中断里如果直接postmailbox锁的持有者可能正是一个被中断打断的Agent同一个锁在中断上下文中被再次获取就会死锁。post_defer把真正的锁竞争推迟到主Loop中断里只做一个入队动作风险小得多。5.2 同步调用call/reply会等待的结果有些场景适合异步但文件系统读块必须等待结果。如果要让AgentA等AgentB返回数据DSH提供dsh_agent_calluint32_t dsh_agent_call(uint16_t target_id, uint16_t type, const void *req, uint16_t req_len, void *resp, uint16_t *resp_len, uint32_t timeout_ms);调用后当前Agent把自己的cookie、caller_id记录到调试表状态变为WAIT_REPLY从ready位图清除。Loop于是去调度其他Agent。目标Agent收到消息后如果想把结果送回来调用dsh_agent_reply(self, msg, resp, resp_len)Loop会根据cookie找到呼叫者把响应数据拷到呼叫者给的buffer再把呼叫者置回READY。这个机制等价于阻塞/唤醒但因为Agent本来就在Loop控制下不需要保存大量寄存器只是改状态和拷贝数据开销非常小。注意一点同步调用期间呼叫者的栈是保留的所以被唤醒后还能继续之前的执行流。这算半个上下文切换但不改变特权级。5.3 超时和环形等待防止邮局变成死结任何同步调用都可能遇到对方挂死。所以dsh_agent_call必须带timeout_ms。Loop里会注册一个单次的定时器事件超时后如果呼叫者还在WAIT_REPLY就直接把它唤醒并返回错误码DSH_ERR_TIMEOUT。测试时我特意做过手术让一个Agent收到调用后永远不回复然后观察超时是否生效结果第一次就发现一个Bug——定时器事件没有和cookie绑定导致超时唤醒的是上一个刚收到回复的Agent。修正方式是在超时事件里记录target cookie唤醒前检查调用者当前是否真的还在等待这个cookie。环形等待是更隐蔽的问题AgentA调用AgentBAgentB调用AgentCAgentC又调用AgentA。大家全变成WAIT_REPLY谁也不会回复。我在AgentLoop里加了一个简单的循环检测器每次call记录wait_graph[caller] target新调用发起前做一次DFS如果发现路径上又回到了当前Agent就打印调用链并强制让最晚一个调用返回超时。这个检测器不算性能友好但它只在CONFIG_AGENTLOOP_DEBUG_CALL_GRAPH开启时工作平时不影响速度。6. 调试与踩坑如何定位AgentLoop里的死循环和竞态6.1 用状态输出快速定位内核卡死定位内核问题是热搜词但这确实是AgentLoop最痛苦的一环。内核卡死最常见的现象是串口最后一行打印了一个Agent编号然后整个世界安静了。这时候先不要急着打断而是想办法把运行时信息榨出来。DSH的调试命令行里有一个命令dsh loop dump [agent] name pri state runs ticks mbox [00] heartbeat 120 READY 120 1345 0 [01] console 10 RUNNING 5 12344 2 [02] blockdev 50 BLOCKED 0 0 0这个dump能让我立刻确认是console把自己饿死了还是blockdev从来没有被调度。如果连dump都打印不出来说明死循环已经发生在更底层此时我会在AgentLoop入口放一个硬件看门狗注入让它复位后从保存的寄存器里读出最后一次运行的Agent id。保存现场的方法是每次调度前把current_agent_id写到一个固定的全局地址这个地址在崩溃分析工具里可以直接读。看起来朴素但真的救过我好几次。有一次调试ext2文件系统挂载后立即卡死用这个老办法定位到是blockdev Agent里有个while (wait_for_dma)没设置超时才把AgentLoop整个拖住。6.2 我在实际开发中踩过的三个坑第一个坑Agent的entry里调用dsh_agent_sleep并等待自己的事件完成。听起来很合理但Sleep会把自己的Agent从ready里清掉如果唤醒条件依赖自己继续执行就永远没人来唤醒它了。我的解决思路是把“等待事件完成”拆成两步——先向设备Agent发送一个读请求然后返回SLEEP等设备Agent发来完成消息时Loop再通过消息唤醒自己。Agent不能自己等自己这是AgentLoop编程里最基本的觉悟。第二个坑注销Agent时还持有mailbox锁。注销函数要回收mailbox里的消息而那时另一个Agent可能正在post。我一开始直接调post发一个请退出消息但注销Agent等post返回时锁已经被自己占着直接死锁。最后统一了注销协议注销必须由Loop执行且先清空ready位再锁mailbox再回收。锁的获取顺序全局一致才止住这类死锁。第三个坑在中断上下文里调用了dsh_agent_call。我在写网卡驱动时图省事想在中断里同步查询dma描述符状态结果那个Agent从此不回话系统直接panic。原因很直接AgentLoop自身的调度上下文在中断里根本没准备好不能做阻塞切换。规范是中断里只能用post_defer不能碰call。如果想做中断里拿结果就把请求post到Agent再让Agent主动查询结果并post回来。6.3 性能剖析把AgentLoop的每一步都变成数字我还在AgentLoop里加了一个可选的Profile模式。注册时记录total_runs和total_ticks每次entry返回后累加。收集样本dsh loop profile agent name avg_us runs wakeups max_us 00 heartbeat 4.2 1098 1098 18 01 console 72.1 203 1098 410 02 blockdev 14.5 40 40 33从这个表就能看出不少问题console的唤醒次数和heartbeat一样多但heartbeat每100ms才醒来一次console怎么会跟着醒来那么多次后来发现是console注册的定时器错误地设成了每个tick唤醒而不是只有输入时才唤醒。这种问题只看代码很难发现但Profile一打出来就无处遁形。另一个典型特征是某个Agent的max_us远超平均值说明它偶尔会遇到慢路径比如持锁等待或DMA超时。7. 从单核到多核AgentLoop的扩展思路DSH目前跑在单核上所以AgentLoop里所有数据结构都假设同一时刻只有一个Agent在运行。将来要上SMP这套设计需要动几次手术。每个CPU核心要有一个独立的run queue和ready_bitmap不能共享一个全局位图否则两个核心可能同时选中同一个Agent。消息队列需要加mpmc安全保护或者采用per-CPU存放消息IPI唤醒的模式。当AgentA在CPU0上要给CPU2上的目标Agent发消息时post函数必须写入目标CPU的消息队列然后向CPU2发送一个IPI中断让它唤醒对应Agent。这一来AgentLoop就从单邮局变成了分布式邮局。但现在单核版本的AgentLoop已经帮我把主路径跑得很顺。等真上了多核优先要改的不是调度算法而是消息投递的并发安全——调度逻辑本身可以照搬锁才是大头。现在如果你也正在写自己的内核我的建议是先别急着实现抢占式任务试着把所有模块都收敛到一个AgentLoop下面跑通一周再说。你会遇到很多早知道就设计清楚的问题但这些问题在Loop模型里都容易暴露、容易修补。我一开始也觉得AgentLoop是多余的一层直到它在调试时帮我抓住了三次死循环我才承认内核里最不性感的那块组件往往才是撑起一切的那根骨架。