写内核这事儿越写到后面越会发现最难的不是某个精妙的算法而是控制流。DSH内核写完中断管理、内存分配和任务调度之后我面对的是一个很实际的问题——外设越来越多每个外设都有自己的一套事件处理逻辑这个模块要中断通知那个模块要定时查询代码开始往“蜘蛛网”方向失控。第八篇落地的AgentLoop模块就是专门来收拾这个局面的。它的核心思路很简单把所有“事件驱动型”的逻辑统一收编成Agent代理由AgentLoop这个循环统一负责接收、调度和分发。文章先从设计动机讲起然后逐步拆解核心数据结构和实现代码最后把我踩过的坑和排查方法一并整理出来给正在写OS、搞RTOS或者研究内核事件机制的朋友一个可以直接参考的样本。1. AgentLoop到底解决什么问题先把痛点摆到桌面上没有AgentLoop的DSH内核其实也能跑但只有亲身经历过那段时间才知道什么叫“维护噩梦”。在介绍AgentLoop之前我想先花点篇幅把当时的痛点讲清楚因为很多设计决策都是被这些痛点逼出来的。1.1 旧写法的三宗罪DSH内核当时的控制流主要分成三路外部中断定时器、串口、键盘、内部软中断/系统调用、还有各类设备寄存器状态轮询。早期写法很朴素——每个中断服务程序里直接改写全局标志位然后主循环里一个接一个地查这些标志位查到哪个就去处理哪个。这种写法最直接的问题是“标志位地狱”。键盘按下设置一个key_pressed串口收到数据设置uart_rx_ready定时器到期设置timer_tick……全局变量越来越多谁在什么地方修改了哪个标志只能靠记忆力。我有一回排查一个串口丢字符的问题查了两天才发现是另一个模块的中断处理函数顺手把uart_rx_ready给清了。第二宗罪是“优先级缺失”。所有标志在同一个循环里被查询处理顺序完全由代码顺序决定。键盘驱动写得靠前它就比串口驱动优先哪天调整了代码顺序整个行为就可能变。内核里事件处理是有实时性要求的比如串口接收缓冲区快满了这时候必须优先处理串口事件而不是先去处理什么无关紧要的按键事件。第三宗罪是“轮询浪费”。有些外设并没有中断能力或者中断被屏蔽了只能靠周期性读取状态寄存器。早期做法是在主循环里固定每圈都去读一遍但有些设备的状态变化频率极低这种轮询纯属白烧CPU。1.2 AgentLoop想达到的目标设计AgentLoop的时候我给自己定了几个明确目标统一入口所有“事件驱动型”逻辑不管来源是中断、软中断还是轮询最终都转换成统一的事件结构体投递到同一个循环里处理。可配置优先级事件分发时按优先级排序高优先级事件先处理而不是靠代码书写顺序“侥幸”获得优先级。解耦生产与消费中断处理器只负责把事件塞进队列具体怎么处理由Agent决定。中断里不做重活Agent里不碰硬件寄存器。1.3 为什么是“Agent”而不是“Handler”最初我也管这个模块叫EventHandlerLoop但写了几版之后发现“Handler”这个词的语义太窄了它暗示的是一次性的、被动的处理过程。而DSH内核里这些处理逻辑往往是“有状态”的——键盘代理要维护按键组合状态串口代理要管理接收缓冲区的水位网络代理要维护连接上下文。Agent这个词强调的恰恰是这种“有状态的行为主体”每个Agent有自己的上下文Context、自己的状态机、自己的处理函数Loop则负责在合适的时间把合适的事件交给合适的Agent。这个概念后来也被用到了DSH内核的软中断和定时器管理里相当于把整个事件处理框架统一到了同一套抽象之下。2. AgentLoop核心设计数据结构先于行为确定要写一个AgentLoop之后我做的第一件事不是写循环而是先定义数据结构。因为我始终觉得内核模块的设计质量一大半取决于数据结构选型行为逻辑反而是水到渠成的事。2.1 Agent结构体一个代理到底要带什么一个Agent在DSH内核里长这样struct dsh_agent { uint32_t agent_id; /* 全局唯一代理ID */ uint32_t prio; /* 代理优先级数值越小优先级越高 */ uint32_t flags; /* 状态标志位如AGENT_READY / AGENT_SUSPENDED */ dsh_agent_fn handler; /* 事件处理函数指针 */ void *ctx; /* 代理私有上下文由注册方维护 */ struct dsh_agent *next; }; typedef int (*dsh_agent_fn)(struct dsh_agent *agent, struct dsh_event *evt);为什么每个Agent要自带一个ctx指针这是我踩过坑之后加上的。最初的版本里Agent处理函数是纯函数所有状态都放在全局变量里结果两个实例共用一个逻辑时就完全没法搞。比如我想同时跑两个不同配置的串口代理第二个代理一注册处理函数里引用的全局缓冲区就冲突了。给每个Agent挂一个私有上下文这个问题自然就化解了这也是“有状态主体”在设计上的落点。2.2 Event结构体事件该长什么样事件结构是Agent全部的信息来源它必须同时满足两个要求携带足够的信息以及高效。DSH内核里事件结构如下struct dsh_event { uint32_t type; /* 事件类型如EVT_KEY_PRESS、EVT_UART_RX */ uint32_t src; /* 来源标识如中断号、软中断号 */ uint64_t timestamp; /* 事件产生时间戳用于超时处理和统计 */ size_t data_len; /* 附带数据长度 */ void *data; /* 附带数据指针 */ struct dsh_event *next; };这里有个设计决策值得多说一句data到底应该用固定大小的内联数组还是指针。固定数组不需要动态分配但浪费空间——一个按键事件可能只需要一个字节的扫描码却要占满64字节的事件槽指针方式省内存但需要管理事件数据的生命周期。DSH内核最终用了指针因为内存分配器已经在前几篇写好了而且事件数据多数来自设备缓冲区本来就是现成的没必要拷贝一份。不过这也引出了另一个问题——数据的生命周期管理。每次处理完事件之后谁负责释放data我的约定是谁投递谁拥有谁释放。Loop在处理完一个事件之后会把evt-data原样交还给来源模块来源模块自己决定是释放还是回收复用。这个约定在文档里被反复加粗标注因为这块出了问题就是内存泄漏和野指针的温床。2.3 就绪队列与心跳计数的设计考量AgentLoop的核心调度结构是一个基于优先级数组的环形队列而不是一个简单的FIFO。这里的取舍很有代表性如果只用FIFO实现起来确实简单push到尾部、pop从头取但高优先级事件会被低频事件持续阻塞。用完全有序链表又太浪费每次入队都要做插入排序。最后我采用了最简单但能满足需求的折衷方案——一个包含8个优先级的环形队列数组。投递事件时根据优先级放入不同队列Loop消费时从高优先级队列往下扫。优先级量化到8档这个数字不是拍脑袋定的而是结合DSH内核的实际需求算出来的中断类事件占2档高优先级软中断和定时器占3档中优先级设备轮询和后台任务占3档低优先级。#define AGENTLOOP_LEVELS 8 static struct dsh_event *loop_queue[AGENTLOOP_LEVELS]; static uint32_t loop_queue_head[AGENTLOOP_LEVELS]; static uint32_t loop_queue_tail[AGENTLOOP_LEVELS];每个队列方向的头尾指针都放在独立数组里处理环形结构时用的是经典的取模运算。在这里我要特别提醒一点环形队列的容量必须是2的幂这样取模才能用位掩码替代除法。DSH内核事件队列的容量是256头尾索引的推进只需要idx (idx 1) 0xFF效率高一个量级。这个优化极其微小但在每秒可能处理几万个事件的内核循环里每一次除法省下的CPU周期都是在给整体延迟做贡献。3. 从零实现AgentLoop核心代码一步步拆开看第2节梳理的是设计蓝图这一节进入实操。我在这里展示的代码是DSH内核里可以编译运行的真实代码为了文章可读性稍微精简了注释和错误处理但逻辑保持一致。3.1 注册与注销代理的生命周期管理任何Agent想接收事件第一步必须是注册。注册接口的完整实现如下int agentloop_register(struct dsh_agent *agent) { struct dsh_agent *iter; if (agent NULL || agent-handler NULL) return -EINVAL; if (agent-agent_id 0) agent-agent_id agentloop_new_id(); /* 防止重复注册同一ID */ iter agent_list; while (iter) { if (iter-agent_id agent-agent_id) return -EEXIST; iter iter-next; } agent-next agent_list; agent_list agent; agent-flags | AGENT_READY; return 0; }这个接口有两个细节值得展开。一是agent-agent_id的分配策略这里用了一个全局原子计数器每分配一个ID就自增1。之所以不用内存地址当作ID是因为地址在每次重启后都可能变化而且不方便打印调试。调试日志里看到agent[3]被调用远比看到agent[0x80123456]直观得多。二是把新Agent插入链表头部而不是尾部。链表头插入是O(1)操作尾部需要遍历同时新注册的Agent往往是刚初始化的外设驱动让它先处理事件能在启动阶段更早地排错。注销接口则是注册的逆过程int agentloop_unregister(uint32_t agent_id) { struct dsh_agent *prev NULL, *iter agent_list; while (iter) { if (iter-agent_id agent_id) { if (prev) prev-next iter-next; else agent_list iter-next; iter-flags ~AGENT_READY; return 0; } prev iter; iter iter-next; } return -ENOENT; }注销的时候我并没有立刻释放Agent结构体本身只是把它标记为!READY并从链表摘除。这是为了保证“正在等待事件处理”的Agent不会在处理到一半的时候被释放——内核模块卸载时这种并发问题特别容易导致崩溃。3.2 事件投递中断侧唯一能做的事事件投递是AgentLoop对外最关键的接口它格外强调“快速返回”原则。因为投递函数经常在中断上下文里被调用它不能发生阻塞也尽可能不要动态分配内存。int agentloop_post(uint32_t prio, struct dsh_event *evt) { uint32_t idx; unsigned long iflags; if (evt NULL || prio AGENTLOOP_LEVELS) return -EINVAL; idx loop_queue_tail[prio]; if (((idx 1) 0xFF) loop_queue_head[prio]) { /* 队列满记录丢弃统计直接返回 */ agloop_stats.dropped; return -EAGAIN; } loop_queue_buf[prio][idx] evt; loop_queue_tail[prio] (idx 1) 0xFF; return 0; }这里有个取舍必须当面说清楚**队列满的时候DSH内核选择丢弃并计数而不是阻塞等待。**整个过程里最微妙的地方是中断上下文中的锁保护问题。我在写这一版的时候用的是spin_lock_irqsave就是在关中断的前提下拿自旋锁。后来做了优化如果确认某个投递调用只会发生在非中断上下文可以使用更轻量的spin_lock但为了安全起见投递接口统一用了带irqsave版本——代价是每次投递都要多一次保存和恢复中断标志的寄存器操作但换来的是“任何上下文里调用都安全”的承诺。定时器中断真正发生的那一刻全局变量只能标记“有事件待处理”把它转换成具体的Agent调用则发生在Loop主循环的运行期间。这个设计给主循环让出了很大的空闲窗口。3.3 主循环实现Loop的骨架主循环是整个AgentLoop的心脏它其实非常朴素void agentloop_run(void) { struct dsh_event *evt; struct dsh_agent *agent; uint32_t level; for (;;) { evt NULL; /* 从高到低扫描优先级队列 */ for (level 0; level AGENTLOOP_LEVELS; level) { if (loop_queue_head[level] ! loop_queue_tail[level]) { evt loop_queue_buf[level][loop_queue_head[level]]; loop_queue_head[level] (loop_queue_head[level] 1) 0xFF; break; } } if (evt NULL) { /* 所有队列空闲触发idle hook */ if (agloop_idle_hook) agloop_idle_hook(); continue; } /* 根据事件源查找对应Agent */ agent agentloop_lookup(evt-src); if (agent (agent-flags AGENT_READY)) { /* 真正执行Agent逻辑 */ agent-handler(agent, evt); } else { agloop_stats.unhandled; } /* 事件数据所有权归还给来源模块 */ evt-data NULL; } }主循环为什么是for (;;)加continue而不是等事件到来再唤醒这其实是DSH内核当前架构下的合理选择——它运行在一个专门的内核线程/执行流里没有操作系统的“休眠/唤醒”原语可以依赖所以只能采取“忙查询空闲钩子”的方式。在真实的多任务内核里这个循环会把“队列空闲”翻译成“睡眠等待信号量”但原理是相同的不断寻找可执行的事件找不到就让自己停一停。agentloop_lookup在事件结构里的src字段和Agent的agent_id之间建立映射。这个查表操作我用了最简单的一维数组映射因为DSH内核的Agent总数上限是128直接用Agent[src]就能定位。在这里必须承认一个经验教训我从第三版才开始在evt里记录src字段。早期版本的事件只带着类型Loop拿到事件之后根本不知道该把事件投递给谁只能把所有Agent挨个问一遍“你要不要处理这个事件”。这样做的正确性没有问题但每次投递的开销从O(1)退化成了O(N)。后来加了src字段之后整体吞吐量翻了几倍。3.4 实战写一个键盘控制器代理理论说了一堆不如看一个具体的Agent实现。DSH内核里最早期注册的Agent之一就是键盘控制器代理逻辑非常典型static int kbd_agent_handler(struct dsh_agent *agent, struct dsh_event *evt) { struct kbd_state *ks (struct kbd_state *)agent-ctx; if (evt-type EVT_KBD_SCANCODE) { uint8_t scancode *(uint8_t *)evt-data; if (scancode 0x80) { /* 按键释放 */ ks-key_up 1; } else { /* 按键按下 */ ks-pending_keys[ks-pending_cnt] scancode; ks-pending_cnt; } } return 0; }这个Agent在注册时通过ctx字段绑定了一个kbd_state结构专门用来记录哪些按键是按下状态、哪些是释放状态。因为中断处理程序里只负责投递事件按键的语义解析全放在了Agent里所以中断服务程序可以做到几个微秒内就返回而整个系统的按键响应能力完全由AgentLoop的调度延迟决定。键盘代理这个例子还有一层意义它展示了Agent状态机的价值。如果没有ctx按键的组合逻辑比如CtrlC就必须依赖全局变量而全局变量是无法支撑多实例并存的。4. 踩坑实录AgentLoop最容易出错的地方AgentLoop作为一个框架模块它的坑往往不在自身逻辑而在与其他模块交互的边缘地带。这一节我整理了实际开发中遇到并解决过的四类典型问题每一条都是从调试器里捞出来的真实教训。4.1 事件“莫名其妙”地丢失我遇到的第一个严重问题是串口中断明明触发了但AgentLoop里对应的事件就是不存在。查了两天最终发现是环形队列的容量太小。串口中断在115200波特率下接收一个字节的时间大约86.8微秒中断处理程序每收到一个字节就投递一个事件。如果Agent处理速度稍慢或者高优先级的定时器事件占据了太多循环时间队列很容易满。队列满的时候我在代码里选择了dropped并返回但没有做任何补救。解决思路分两步第一步把串口事件队列容量从64扩大到256第二步更关键——给AgentLoop加了一条背压规则当某个优先级的队列满时投递方要主动暂停该设备的中断响应能力直到队列水位降下来。说到底内核里没有多少场景能容忍事件无限堆积与其在这个地方拼命扩容不如让源头节流。4.2 Agent阻塞导致整个Loop卡死这是第二个深坑。有一个调试用的Agent处理函数里顺手调用了一个print函数而那个输出函数本身是轮询串口的。串口忙碌时print函数会自己忙等串口空闲。问题在于这个Agent处理的事件本身就来自串口串口还没处理完又被Agent输出占用形成了内部的自锁死循环。AgentLoop对这样的场景没有任何保护机制因为它在设计上就假设了“Agent是快速且非阻塞的”。这是所有事件循环类框架的通用隐含前置条件。我的实际修复办法是在Agent处理函数的文档里加了一级硬性约束Agent不允许调用任何可能阻塞的接口需要做重活的Agent应该把任务切到独立的任务列表里或者投递一个“延迟事件”来延后处理。4.3 高优先级事件的“意外饥饿”优先级用得越久越容易发现一种不太直观的现象只要高优先级队列有事件低优先级队列就永远得不到执行。乍一看这似乎没问题——高优先级优先嘛——但后来我发现定时器中断被配置成1毫秒触发一次它是高优先级队列的常客结果就是低优先级的磁盘刷新Agent几乎从未运行过。我对这个问题的处理是给每个优先级添加一个简单的“配额计数”机制Loop在处理完N个高优先级事件后强制切换去处理一个低优先级事件。N根据系统负载可以动态调整DSH内核的默认值是8。这个机制不是完美的公平调度但足以保证低优先级Agent在极端负载下仍然能获得基础处理时间。4.4 事件数据的生命周期混乱最后一个问题纯粹是代码管理层面的。DSH内核的约定是“谁投递谁拥有”但代码写得久了这个约定就容易被忘记。有一版代码里某个Agent在处理完事件后顺手free了evt-data而投递它的模块在事件处理完后又会对同一指针做一次free——结果就是双重释放。这类问题在QEMU环境里比较隐蔽真实硬件上往往表现为随机崩溃。我最后的解决方式是给事件结构体增加了一个owner_id字段记录投递者的ID。投递者有义务在Agent处理完后归还数据而整个归还动作就是检查owner_id是否与当前处理循环所关联的来源模块一致。这种方式足够简单也足够直观。4.5 常见问题速查表为了让你排查问题时顺一点我把上面这些问题的现象、根因和解决方向整理成了一个表格现象可能根因排查方法解决方案事件丢失、计数骤增队列容量太小检查agloop_stats.dropped扩容队列加背压机制系统“死机”无响应Agent阻塞或自锁看CPU是否仍执行Loop主循环Agent内禁止阻塞调用低优先级Agent长期不执行高优先级抢占过重统计各优先级处理次数加配额强制轮转随机崩溃或内存异常事件数据所有权混乱查谁在释放evt-data用owner_id做审计注册多个同ID Agent成功注册接口未做去重检查打印Agent表注册时检查唯一性5. AgentLoop的进一步扩展把框架用起来AgentLoop的价值不只是“一个循环”它其实是DSH内核里事件驱动能力的地基。地基打完之后我很快在它上面长出了几个新东西这里挑两个比较有代表性的聊聊。5.1 用AgentLoop承载软中断DSH内核里的软中断机制最初是独立写的后来我干脆把它重构成AgentLoop的一个用途软中断本身就是一个高优先级的Agent系统调用通过一个特殊事件类型来触发它。这样一来软中断需要的优先级管理、排队能力、超时统计全都白捡到了——因为它们本来就在AgentLoop里实现了。这个重构让我对“框架”有了更深的理解真正值得抽象的不是某个具体功能而是那一层稳定的调度语义。软中断和外部中断在语义上都是“来了一个事件要在合适的时间处理”差异只在来源和优先级而这两点AgentLoop都已经参数化了。5.2 代理之间的消息总线另一个扩展方向是Agent之间的通信。最开始Agent是彼此孤立的键盘Agent处理完扫描码之后如果想通知任务调度器“有按键事件”只能通过全局变量——又回到了我第1节就批判的“标志位地狱”。后来我做了一条轻量的消息总线本质上就是一个广播Agent任何Agent都可以注册自己为“感兴趣的接收方”总线把事件复制或者引用转发给所有订阅者。在断言的严格性上总线比AgentLoop主路径更宽松一点因为广播不处理背压问题订阅者的处理速度如果跟不上消息会积压。但在DSH内核的教学场景中这层直观的“观察者模式”价值远大于性能损失。5.3 AgentLoop与你自己的项目怎么结合如果你不是在内核项目里而是在做嵌入式裸机开发、RTOS应用甚至只是写一个网络服务器的事件循环AgentLoop的这套思想同样可以直接移植。核心要迁移的不是代码而是三句话把“事件源”和“事件处理者”解耦来源只管投递Agent只管处理用统一的优先级队列替代散落的标志位轮询对事件数据的生命周期定下明确约定并且通过结构体字段把它固化下来。这三句话听起来平淡无奇但真正写代码的时候每一条都需要靠踩坑来真正理解。AgentLoop的这五千多行代码包括测试可以说就是用这些朴素原则锤炼出来的。最后再分享一个小经验。每次往AgentLoop里加新功能之前我会先在纸面上画出事件流谁投递、谁消费、中间经过几个队列、谁释放数据。这套流程坚持了八个系列文章如今已经成了我写DSH内核任何模块的固定习惯。你如果刚开始做类似的事件循环建议也试试——画出那条链路比你想象中更能找出隐藏的并发冲突。 
长河君 · 2026/10/8 9:52:48 · 阅读 1.2k AgentLoop：内核事件驱动架构的统一调度循环设计
写内核这事儿越写到后面越会发现最难的不是某个精妙的算法而是控制流。DSH内核写完中断管理、内存分配和任务调度之后我面对的是一个很实际的问题——外设越来越多每个外设都有自己的一套事件处理逻辑这个模块要中断通知那个模块要定时查询代码开始往“蜘蛛网”方向失控。第八篇落地的AgentLoop模块就是专门来收拾这个局面的。它的核心思路很简单把所有“事件驱动型”的逻辑统一收编成Agent代理由AgentLoop这个循环统一负责接收、调度和分发。文章先从设计动机讲起然后逐步拆解核心数据结构和实现代码最后把我踩过的坑和排查方法一并整理出来给正在写OS、搞RTOS或者研究内核事件机制的朋友一个可以直接参考的样本。1. AgentLoop到底解决什么问题先把痛点摆到桌面上没有AgentLoop的DSH内核其实也能跑但只有亲身经历过那段时间才知道什么叫“维护噩梦”。在介绍AgentLoop之前我想先花点篇幅把当时的痛点讲清楚因为很多设计决策都是被这些痛点逼出来的。1.1 旧写法的三宗罪DSH内核当时的控制流主要分成三路外部中断定时器、串口、键盘、内部软中断/系统调用、还有各类设备寄存器状态轮询。早期写法很朴素——每个中断服务程序里直接改写全局标志位然后主循环里一个接一个地查这些标志位查到哪个就去处理哪个。这种写法最直接的问题是“标志位地狱”。键盘按下设置一个key_pressed串口收到数据设置uart_rx_ready定时器到期设置timer_tick……全局变量越来越多谁在什么地方修改了哪个标志只能靠记忆力。我有一回排查一个串口丢字符的问题查了两天才发现是另一个模块的中断处理函数顺手把uart_rx_ready给清了。第二宗罪是“优先级缺失”。所有标志在同一个循环里被查询处理顺序完全由代码顺序决定。键盘驱动写得靠前它就比串口驱动优先哪天调整了代码顺序整个行为就可能变。内核里事件处理是有实时性要求的比如串口接收缓冲区快满了这时候必须优先处理串口事件而不是先去处理什么无关紧要的按键事件。第三宗罪是“轮询浪费”。有些外设并没有中断能力或者中断被屏蔽了只能靠周期性读取状态寄存器。早期做法是在主循环里固定每圈都去读一遍但有些设备的状态变化频率极低这种轮询纯属白烧CPU。1.2 AgentLoop想达到的目标设计AgentLoop的时候我给自己定了几个明确目标统一入口所有“事件驱动型”逻辑不管来源是中断、软中断还是轮询最终都转换成统一的事件结构体投递到同一个循环里处理。可配置优先级事件分发时按优先级排序高优先级事件先处理而不是靠代码书写顺序“侥幸”获得优先级。解耦生产与消费中断处理器只负责把事件塞进队列具体怎么处理由Agent决定。中断里不做重活Agent里不碰硬件寄存器。1.3 为什么是“Agent”而不是“Handler”最初我也管这个模块叫EventHandlerLoop但写了几版之后发现“Handler”这个词的语义太窄了它暗示的是一次性的、被动的处理过程。而DSH内核里这些处理逻辑往往是“有状态”的——键盘代理要维护按键组合状态串口代理要管理接收缓冲区的水位网络代理要维护连接上下文。Agent这个词强调的恰恰是这种“有状态的行为主体”每个Agent有自己的上下文Context、自己的状态机、自己的处理函数Loop则负责在合适的时间把合适的事件交给合适的Agent。这个概念后来也被用到了DSH内核的软中断和定时器管理里相当于把整个事件处理框架统一到了同一套抽象之下。2. AgentLoop核心设计数据结构先于行为确定要写一个AgentLoop之后我做的第一件事不是写循环而是先定义数据结构。因为我始终觉得内核模块的设计质量一大半取决于数据结构选型行为逻辑反而是水到渠成的事。2.1 Agent结构体一个代理到底要带什么一个Agent在DSH内核里长这样struct dsh_agent { uint32_t agent_id; /* 全局唯一代理ID */ uint32_t prio; /* 代理优先级数值越小优先级越高 */ uint32_t flags; /* 状态标志位如AGENT_READY / AGENT_SUSPENDED */ dsh_agent_fn handler; /* 事件处理函数指针 */ void *ctx; /* 代理私有上下文由注册方维护 */ struct dsh_agent *next; }; typedef int (*dsh_agent_fn)(struct dsh_agent *agent, struct dsh_event *evt);为什么每个Agent要自带一个ctx指针这是我踩过坑之后加上的。最初的版本里Agent处理函数是纯函数所有状态都放在全局变量里结果两个实例共用一个逻辑时就完全没法搞。比如我想同时跑两个不同配置的串口代理第二个代理一注册处理函数里引用的全局缓冲区就冲突了。给每个Agent挂一个私有上下文这个问题自然就化解了这也是“有状态主体”在设计上的落点。2.2 Event结构体事件该长什么样事件结构是Agent全部的信息来源它必须同时满足两个要求携带足够的信息以及高效。DSH内核里事件结构如下struct dsh_event { uint32_t type; /* 事件类型如EVT_KEY_PRESS、EVT_UART_RX */ uint32_t src; /* 来源标识如中断号、软中断号 */ uint64_t timestamp; /* 事件产生时间戳用于超时处理和统计 */ size_t data_len; /* 附带数据长度 */ void *data; /* 附带数据指针 */ struct dsh_event *next; };这里有个设计决策值得多说一句data到底应该用固定大小的内联数组还是指针。固定数组不需要动态分配但浪费空间——一个按键事件可能只需要一个字节的扫描码却要占满64字节的事件槽指针方式省内存但需要管理事件数据的生命周期。DSH内核最终用了指针因为内存分配器已经在前几篇写好了而且事件数据多数来自设备缓冲区本来就是现成的没必要拷贝一份。不过这也引出了另一个问题——数据的生命周期管理。每次处理完事件之后谁负责释放data我的约定是谁投递谁拥有谁释放。Loop在处理完一个事件之后会把evt-data原样交还给来源模块来源模块自己决定是释放还是回收复用。这个约定在文档里被反复加粗标注因为这块出了问题就是内存泄漏和野指针的温床。2.3 就绪队列与心跳计数的设计考量AgentLoop的核心调度结构是一个基于优先级数组的环形队列而不是一个简单的FIFO。这里的取舍很有代表性如果只用FIFO实现起来确实简单push到尾部、pop从头取但高优先级事件会被低频事件持续阻塞。用完全有序链表又太浪费每次入队都要做插入排序。最后我采用了最简单但能满足需求的折衷方案——一个包含8个优先级的环形队列数组。投递事件时根据优先级放入不同队列Loop消费时从高优先级队列往下扫。优先级量化到8档这个数字不是拍脑袋定的而是结合DSH内核的实际需求算出来的中断类事件占2档高优先级软中断和定时器占3档中优先级设备轮询和后台任务占3档低优先级。#define AGENTLOOP_LEVELS 8 static struct dsh_event *loop_queue[AGENTLOOP_LEVELS]; static uint32_t loop_queue_head[AGENTLOOP_LEVELS]; static uint32_t loop_queue_tail[AGENTLOOP_LEVELS];每个队列方向的头尾指针都放在独立数组里处理环形结构时用的是经典的取模运算。在这里我要特别提醒一点环形队列的容量必须是2的幂这样取模才能用位掩码替代除法。DSH内核事件队列的容量是256头尾索引的推进只需要idx (idx 1) 0xFF效率高一个量级。这个优化极其微小但在每秒可能处理几万个事件的内核循环里每一次除法省下的CPU周期都是在给整体延迟做贡献。3. 从零实现AgentLoop核心代码一步步拆开看第2节梳理的是设计蓝图这一节进入实操。我在这里展示的代码是DSH内核里可以编译运行的真实代码为了文章可读性稍微精简了注释和错误处理但逻辑保持一致。3.1 注册与注销代理的生命周期管理任何Agent想接收事件第一步必须是注册。注册接口的完整实现如下int agentloop_register(struct dsh_agent *agent) { struct dsh_agent *iter; if (agent NULL || agent-handler NULL) return -EINVAL; if (agent-agent_id 0) agent-agent_id agentloop_new_id(); /* 防止重复注册同一ID */ iter agent_list; while (iter) { if (iter-agent_id agent-agent_id) return -EEXIST; iter iter-next; } agent-next agent_list; agent_list agent; agent-flags | AGENT_READY; return 0; }这个接口有两个细节值得展开。一是agent-agent_id的分配策略这里用了一个全局原子计数器每分配一个ID就自增1。之所以不用内存地址当作ID是因为地址在每次重启后都可能变化而且不方便打印调试。调试日志里看到agent[3]被调用远比看到agent[0x80123456]直观得多。二是把新Agent插入链表头部而不是尾部。链表头插入是O(1)操作尾部需要遍历同时新注册的Agent往往是刚初始化的外设驱动让它先处理事件能在启动阶段更早地排错。注销接口则是注册的逆过程int agentloop_unregister(uint32_t agent_id) { struct dsh_agent *prev NULL, *iter agent_list; while (iter) { if (iter-agent_id agent_id) { if (prev) prev-next iter-next; else agent_list iter-next; iter-flags ~AGENT_READY; return 0; } prev iter; iter iter-next; } return -ENOENT; }注销的时候我并没有立刻释放Agent结构体本身只是把它标记为!READY并从链表摘除。这是为了保证“正在等待事件处理”的Agent不会在处理到一半的时候被释放——内核模块卸载时这种并发问题特别容易导致崩溃。3.2 事件投递中断侧唯一能做的事事件投递是AgentLoop对外最关键的接口它格外强调“快速返回”原则。因为投递函数经常在中断上下文里被调用它不能发生阻塞也尽可能不要动态分配内存。int agentloop_post(uint32_t prio, struct dsh_event *evt) { uint32_t idx; unsigned long iflags; if (evt NULL || prio AGENTLOOP_LEVELS) return -EINVAL; idx loop_queue_tail[prio]; if (((idx 1) 0xFF) loop_queue_head[prio]) { /* 队列满记录丢弃统计直接返回 */ agloop_stats.dropped; return -EAGAIN; } loop_queue_buf[prio][idx] evt; loop_queue_tail[prio] (idx 1) 0xFF; return 0; }这里有个取舍必须当面说清楚**队列满的时候DSH内核选择丢弃并计数而不是阻塞等待。**整个过程里最微妙的地方是中断上下文中的锁保护问题。我在写这一版的时候用的是spin_lock_irqsave就是在关中断的前提下拿自旋锁。后来做了优化如果确认某个投递调用只会发生在非中断上下文可以使用更轻量的spin_lock但为了安全起见投递接口统一用了带irqsave版本——代价是每次投递都要多一次保存和恢复中断标志的寄存器操作但换来的是“任何上下文里调用都安全”的承诺。定时器中断真正发生的那一刻全局变量只能标记“有事件待处理”把它转换成具体的Agent调用则发生在Loop主循环的运行期间。这个设计给主循环让出了很大的空闲窗口。3.3 主循环实现Loop的骨架主循环是整个AgentLoop的心脏它其实非常朴素void agentloop_run(void) { struct dsh_event *evt; struct dsh_agent *agent; uint32_t level; for (;;) { evt NULL; /* 从高到低扫描优先级队列 */ for (level 0; level AGENTLOOP_LEVELS; level) { if (loop_queue_head[level] ! loop_queue_tail[level]) { evt loop_queue_buf[level][loop_queue_head[level]]; loop_queue_head[level] (loop_queue_head[level] 1) 0xFF; break; } } if (evt NULL) { /* 所有队列空闲触发idle hook */ if (agloop_idle_hook) agloop_idle_hook(); continue; } /* 根据事件源查找对应Agent */ agent agentloop_lookup(evt-src); if (agent (agent-flags AGENT_READY)) { /* 真正执行Agent逻辑 */ agent-handler(agent, evt); } else { agloop_stats.unhandled; } /* 事件数据所有权归还给来源模块 */ evt-data NULL; } }主循环为什么是for (;;)加continue而不是等事件到来再唤醒这其实是DSH内核当前架构下的合理选择——它运行在一个专门的内核线程/执行流里没有操作系统的“休眠/唤醒”原语可以依赖所以只能采取“忙查询空闲钩子”的方式。在真实的多任务内核里这个循环会把“队列空闲”翻译成“睡眠等待信号量”但原理是相同的不断寻找可执行的事件找不到就让自己停一停。agentloop_lookup在事件结构里的src字段和Agent的agent_id之间建立映射。这个查表操作我用了最简单的一维数组映射因为DSH内核的Agent总数上限是128直接用Agent[src]就能定位。在这里必须承认一个经验教训我从第三版才开始在evt里记录src字段。早期版本的事件只带着类型Loop拿到事件之后根本不知道该把事件投递给谁只能把所有Agent挨个问一遍“你要不要处理这个事件”。这样做的正确性没有问题但每次投递的开销从O(1)退化成了O(N)。后来加了src字段之后整体吞吐量翻了几倍。3.4 实战写一个键盘控制器代理理论说了一堆不如看一个具体的Agent实现。DSH内核里最早期注册的Agent之一就是键盘控制器代理逻辑非常典型static int kbd_agent_handler(struct dsh_agent *agent, struct dsh_event *evt) { struct kbd_state *ks (struct kbd_state *)agent-ctx; if (evt-type EVT_KBD_SCANCODE) { uint8_t scancode *(uint8_t *)evt-data; if (scancode 0x80) { /* 按键释放 */ ks-key_up 1; } else { /* 按键按下 */ ks-pending_keys[ks-pending_cnt] scancode; ks-pending_cnt; } } return 0; }这个Agent在注册时通过ctx字段绑定了一个kbd_state结构专门用来记录哪些按键是按下状态、哪些是释放状态。因为中断处理程序里只负责投递事件按键的语义解析全放在了Agent里所以中断服务程序可以做到几个微秒内就返回而整个系统的按键响应能力完全由AgentLoop的调度延迟决定。键盘代理这个例子还有一层意义它展示了Agent状态机的价值。如果没有ctx按键的组合逻辑比如CtrlC就必须依赖全局变量而全局变量是无法支撑多实例并存的。4. 踩坑实录AgentLoop最容易出错的地方AgentLoop作为一个框架模块它的坑往往不在自身逻辑而在与其他模块交互的边缘地带。这一节我整理了实际开发中遇到并解决过的四类典型问题每一条都是从调试器里捞出来的真实教训。4.1 事件“莫名其妙”地丢失我遇到的第一个严重问题是串口中断明明触发了但AgentLoop里对应的事件就是不存在。查了两天最终发现是环形队列的容量太小。串口中断在115200波特率下接收一个字节的时间大约86.8微秒中断处理程序每收到一个字节就投递一个事件。如果Agent处理速度稍慢或者高优先级的定时器事件占据了太多循环时间队列很容易满。队列满的时候我在代码里选择了dropped并返回但没有做任何补救。解决思路分两步第一步把串口事件队列容量从64扩大到256第二步更关键——给AgentLoop加了一条背压规则当某个优先级的队列满时投递方要主动暂停该设备的中断响应能力直到队列水位降下来。说到底内核里没有多少场景能容忍事件无限堆积与其在这个地方拼命扩容不如让源头节流。4.2 Agent阻塞导致整个Loop卡死这是第二个深坑。有一个调试用的Agent处理函数里顺手调用了一个print函数而那个输出函数本身是轮询串口的。串口忙碌时print函数会自己忙等串口空闲。问题在于这个Agent处理的事件本身就来自串口串口还没处理完又被Agent输出占用形成了内部的自锁死循环。AgentLoop对这样的场景没有任何保护机制因为它在设计上就假设了“Agent是快速且非阻塞的”。这是所有事件循环类框架的通用隐含前置条件。我的实际修复办法是在Agent处理函数的文档里加了一级硬性约束Agent不允许调用任何可能阻塞的接口需要做重活的Agent应该把任务切到独立的任务列表里或者投递一个“延迟事件”来延后处理。4.3 高优先级事件的“意外饥饿”优先级用得越久越容易发现一种不太直观的现象只要高优先级队列有事件低优先级队列就永远得不到执行。乍一看这似乎没问题——高优先级优先嘛——但后来我发现定时器中断被配置成1毫秒触发一次它是高优先级队列的常客结果就是低优先级的磁盘刷新Agent几乎从未运行过。我对这个问题的处理是给每个优先级添加一个简单的“配额计数”机制Loop在处理完N个高优先级事件后强制切换去处理一个低优先级事件。N根据系统负载可以动态调整DSH内核的默认值是8。这个机制不是完美的公平调度但足以保证低优先级Agent在极端负载下仍然能获得基础处理时间。4.4 事件数据的生命周期混乱最后一个问题纯粹是代码管理层面的。DSH内核的约定是“谁投递谁拥有”但代码写得久了这个约定就容易被忘记。有一版代码里某个Agent在处理完事件后顺手free了evt-data而投递它的模块在事件处理完后又会对同一指针做一次free——结果就是双重释放。这类问题在QEMU环境里比较隐蔽真实硬件上往往表现为随机崩溃。我最后的解决方式是给事件结构体增加了一个owner_id字段记录投递者的ID。投递者有义务在Agent处理完后归还数据而整个归还动作就是检查owner_id是否与当前处理循环所关联的来源模块一致。这种方式足够简单也足够直观。4.5 常见问题速查表为了让你排查问题时顺一点我把上面这些问题的现象、根因和解决方向整理成了一个表格现象可能根因排查方法解决方案事件丢失、计数骤增队列容量太小检查agloop_stats.dropped扩容队列加背压机制系统“死机”无响应Agent阻塞或自锁看CPU是否仍执行Loop主循环Agent内禁止阻塞调用低优先级Agent长期不执行高优先级抢占过重统计各优先级处理次数加配额强制轮转随机崩溃或内存异常事件数据所有权混乱查谁在释放evt-data用owner_id做审计注册多个同ID Agent成功注册接口未做去重检查打印Agent表注册时检查唯一性5. AgentLoop的进一步扩展把框架用起来AgentLoop的价值不只是“一个循环”它其实是DSH内核里事件驱动能力的地基。地基打完之后我很快在它上面长出了几个新东西这里挑两个比较有代表性的聊聊。5.1 用AgentLoop承载软中断DSH内核里的软中断机制最初是独立写的后来我干脆把它重构成AgentLoop的一个用途软中断本身就是一个高优先级的Agent系统调用通过一个特殊事件类型来触发它。这样一来软中断需要的优先级管理、排队能力、超时统计全都白捡到了——因为它们本来就在AgentLoop里实现了。这个重构让我对“框架”有了更深的理解真正值得抽象的不是某个具体功能而是那一层稳定的调度语义。软中断和外部中断在语义上都是“来了一个事件要在合适的时间处理”差异只在来源和优先级而这两点AgentLoop都已经参数化了。5.2 代理之间的消息总线另一个扩展方向是Agent之间的通信。最开始Agent是彼此孤立的键盘Agent处理完扫描码之后如果想通知任务调度器“有按键事件”只能通过全局变量——又回到了我第1节就批判的“标志位地狱”。后来我做了一条轻量的消息总线本质上就是一个广播Agent任何Agent都可以注册自己为“感兴趣的接收方”总线把事件复制或者引用转发给所有订阅者。在断言的严格性上总线比AgentLoop主路径更宽松一点因为广播不处理背压问题订阅者的处理速度如果跟不上消息会积压。但在DSH内核的教学场景中这层直观的“观察者模式”价值远大于性能损失。5.3 AgentLoop与你自己的项目怎么结合如果你不是在内核项目里而是在做嵌入式裸机开发、RTOS应用甚至只是写一个网络服务器的事件循环AgentLoop的这套思想同样可以直接移植。核心要迁移的不是代码而是三句话把“事件源”和“事件处理者”解耦来源只管投递Agent只管处理用统一的优先级队列替代散落的标志位轮询对事件数据的生命周期定下明确约定并且通过结构体字段把它固化下来。这三句话听起来平淡无奇但真正写代码的时候每一条都需要靠踩坑来真正理解。AgentLoop的这五千多行代码包括测试可以说就是用这些朴素原则锤炼出来的。最后再分享一个小经验。每次往AgentLoop里加新功能之前我会先在纸面上画出事件流谁投递、谁消费、中间经过几个队列、谁释放数据。这套流程坚持了八个系列文章如今已经成了我写DSH内核任何模块的固定习惯。你如果刚开始做类似的事件循环建议也试试——画出那条链路比你想象中更能找出隐藏的并发冲突。