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，按键的组合逻辑（比如Ctrl+C）就必须依赖全局变量，而全局变量是无法支撑多实例并存的。

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 常见问题速查表

为了让你排查问题时顺一点，我把上面这些问题的现象、根因和解决方向整理成了一个表格：

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内核任何模块的固定习惯。你如果刚开始做类似的事件循环，建议也试试——画出那条链路，比你想象中更能找出隐藏的并发冲突。