DSH内核写到第六篇调度器已经能在几个任务之间来回切换内存管理也把分配器跑稳了。接下来要打通的是 Agent 与 Agent 之间的数据通路。这一篇解决两个问题一个 Agent 想给另一个 Agent 发消息怎么安全地找到对方接收方阻塞等待时怎么被合适地唤醒发送方遇到队列满了怎么睡眠而不会把系统搞死。配套的这套东西就是 Agent 句柄与收发队列。这个阶段对整个内核来说非常关键。前几篇我们把每个执行体叫 Agent它有自己的栈、上下文和调度状态但还没有一条像样的通信管道。没有通信管道调度器调度得再好各个 Agent 也只是孤岛。而通信管道的设计里最容易被低估的就是句柄层。很多初学者觉得给个指针就行了反正内核里都是同一个地址空间。但实际上一旦你开始做资源授权、生命周期管理和异常退出裸指针会变成一串串悬垂指针和双重释放。我会把这一篇的设计思路、数据结构、并发细节和踩坑过程全部展开。代码是 C 风格适合任何正在从零写内核、写教学操作系统、或者做嵌入式 RTOS 扩展的人参考。我不会只给结论会把当时为什么纠结、最后为什么这么选也讲清楚。1. 为什么不能直接传指针Agent 句柄到底在防什么在没有句柄之前我的代码里到处都是struct dsh_agent *agent直接传来传去。看起来省事其实埋了三个雷。第一个雷是权限无法收缩。指针一旦暴露出去持有者可以访问这个结构体的任意字段甚至可以伪造一个dsh_agent对象把自己的数据填进去。内核里很多崩溃和提权漏洞本质都是“不该拿到指针的人拿到了指针”。而我们只是做一个教学内核但也不能在这个地方养成坏习惯。第二个雷是生命周期没法追踪。A 任务拿着指向某 Agent 的指针那个 Agent 退出了结构体被释放A 手里的指针就悬空了。你当然可以说“释放前先通知所有引用者”可是你怎么知道都有谁在引用在这个阶段没有自动追踪机制纯靠自觉迟早出事。第三个雷是对象迁移。后面如果做内核堆压缩、做地址空间切换Agent 控制块本身可能被搬走。指针是绝对地址一搬就全断了。但句柄是逻辑编号表项里的地址更新一下所有持有句柄的调用方都不受影响。打个比方直接把门锁的内部结构告诉访客和给访客一张房卡是两回事。房卡不知道门锁长什么样但前台一查就知道这个卡号对应哪个房间、几点到期、能不能进。句柄就是这个房卡。所以最终设计是每个 Agent 在创建时获得一个全局唯一的句柄类型是uint32_t用户态和内核其他模块只准持有这个整数不允许直接持有结构体指针。结构体的真实地址只存在于句柄表里并且由内核统一管理。1.1 句柄值里为什么要塞一个代际如果一个句柄只是“槽位索引”那还有个经典问题槽位复用。假设 Agent A 占着句柄槽位 3句柄值是 3。后来 A 退出槽位空出来。又新建一个 Agent B系统把槽位 3 分配给它。这时候之前持有“句柄 3”的调用方本来以为自己还在操作 A结果拿到的是 B。这是典型的 ABA 问题。我的解决办法是句柄值由“槽位索引”和“代际”两部分组成。#define DSH_HANDLE_INDEX_MASK 0xffffu #define DSH_HANDLE_GEN_SHIFT 16 static inline uint32_t dsh_handle_encode(uint16_t index, uint16_t generation) { return ((uint32_t)generation DSH_HANDLE_GEN_SHIFT) | index; }每次槽位被释放再分配时代际加一。旧句柄就算还带着旧的槽位编号代际校验也会失败。这样悬垂句柄最多返回一个-EINVAL或-ENOTCONN而不是把一个消息发到错误的 Agent 上。要注意代际不是无限增长的。我用的是 16 位所以会在复用前代际溢出。真溢出时系统直接放弃这个槽位不再分配同时打印一条错误。早期内核没有必要为了省槽位牺牲可靠性。1.2 引用计数句柄表只是入口对象生命周期靠 refcount句柄表解决“怎么找到对象”但“找到之后怎么保证对象不会被并发释放”是另一个问题。经典场景线程 A 拿着句柄准备发送消息已经通过句柄表拿到了struct dsh_agent *agent的地址。同一时刻线程 B 关闭这个句柄并且因为这是最后一个持有者直接释放了 agent 对象。A 还没来得及操作手里的指针就已经变成悬垂的了。解决这个问题不能靠“关闭时等别人用完”因为你没法知道别人什么时候用完。通用做法是引用计数句柄表项里保存一个指向 agent 的强引用也就是refcount。任何从句柄表“取出”agent 的路径都必须先对 agent 做一次原子引用计数增加然后再解锁句柄表。操作结束后调用统一的 put 操作释放这个引用。当用户关闭句柄时只是把句柄表项清除并对 agent 做一次 put。如果还有别的地方正拿着引用agent 不会被释放等所有引用都归零才真正销毁。这是内核里最常用的生命周期模型。不是“释放时小心翼翼地检查有没有人用”而是“释放必须等到最后一个用户主动放手”。我早期觉得引用计数啰嗦后来被悬垂指针折磨过两次再也不那么想了。2. 句柄表的具体实现从分配、校验到并发释放DSH 的句柄表是一个定长数组默认 512 个槽位。这个数字对教学内核足够用而且定长数组让分配逻辑非常简单遍历找空槽即可。表项结构体是这种风格struct dsh_handle_entry { struct dsh_agent_object *agent; uint16_t generation; uint16_t flags; struct dsh_handle_entry *next_free; }; struct dsh_handle_table { struct dsh_handle_entry entries[DSH_AGENT_HANDLE_MAX]; struct dsh_handle_entry *free_list; dsh_spinlock_t lock; };我没有用哈希表因为 Agent 数量少密集编号反而比哈希更简单。哈希表适合句柄值很分散、数量很大的场景但教学内核里 512 个槽位从头遍历一次也就几十纳秒完全不是瓶颈。全局有一把自旋锁保护整个句柄表。这个选择是刻意的句柄表的操作非常短无非就是加载、比较、赋值不会睡眠。自旋锁在这种临界区里是最合适的就算多核竞争等一小会儿也就拿到了。但要注意绝不能在持有这把锁的时候调用可能睡眠的函数。2.1 句柄分配与释放的完整流程创建一个新 Agent 时会执行dsh_agent_handle_allocuint32_t dsh_agent_handle_alloc(struct dsh_agent_object *agent) { uint32_t handle 0; uint64_t irqflags; dsh_spin_lock_irqsave(g_handle_table.lock, irqflags); struct dsh_handle_entry *e g_handle_table.free_list; if (!e) { dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return 0; /* 表满 */ } g_handle_table.free_list e-next_free; e-agent agent; e-generation; if (e-generation 0) { /* 代际溢出放弃这个槽位 */ dsh_handle_table_park_slot(e); dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return 0; } agent-refcount_inc(agent); handle dsh_handle_encode(e - g_handle_table.entries, e-generation); dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return handle; }分配时的refcount_inc是句柄表自己的那一份引用。因此每个 Agent 只要还有句柄在表里引用计数就至少是 1。关闭句柄时对应做一次refcount_dec。这样即使 Agent 自己已经决定退出只要用户还有句柄没关底层对象就不会凭空消失。释放句柄的函数逻辑正好反过来校验代际清空 agent 指针把槽位挂回空闲链表最后对 agent 做一次 put。这里有个细节关闭句柄时如果发现代理已经处于“正在退出”状态我们仍然要执行同样的表项清理和 put只是后续对被关闭对象的新操作都会立刻返回错误。2.2 句柄校验与查找先加引用再返回指针从句柄到对象的核心函数是dsh_agent_lookupstruct dsh_agent_object *dsh_agent_lookup(uint32_t handle) { uint32_t idx handle DSH_HANDLE_INDEX_MASK; uint32_t gen handle DSH_HANDLE_GEN_SHIFT; struct dsh_handle_entry *e; struct dsh_agent_object *agent NULL; uint64_t irqflags; if (idx DSH_AGENT_HANDLE_MAX) return NULL; dsh_spin_lock_irqsave(g_handle_table.lock, irqflags); e g_handle_table.entries[idx]; if (e-agent e-generation gen) { agent e-agent; agent-refcount_inc(agent); } dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return agent; }注意顺序拿到 agent 后还在锁内就完成了引用计数增加然后再解锁。因为一旦解锁另一个线程可能马上关掉句柄并释放 agent所以“取指针”和“加引用”必须是一个不可分割的动作。调用方使用完拿到的 agent 后必须调用dsh_agent_put_ref(agent)归还引用。我在每个发送路径的函数入口处都会写清楚提示dsh_agent_lookup返回的是带引用的指针。调用方不负责释放 agent但负责调用put_ref。漏掉这一步Agent 会永远无法销毁。这个提示是我在项目代码注释里反复强调的。内核里引用计数泄漏比普通内存泄漏难查得多因为它的表现不是立刻崩而是“某些对象回收不掉”直到某一天句柄表满了。3. 收发队列有界环形缓冲区与等待唤醒句柄解决了“找到对方”真正传数据的是收发队列。每个 Agent 创建时都会伴随一个接收邮箱。其它 Agent 通过句柄找到这个 Agent然后往邮箱里 push 消息Agent 自己从邮箱里 pop 消息处理。我坚持使用有界环形缓冲区没有做成无界链表。核心原因是无界队列会掩盖调度问题生产者不断生产消费者跟不上内存像滚雪球一样增长最后系统在内存耗尽中崩溃。有界队列是天然的背压机制。这跟水管一样有容量才有压力水压会让上游减速没有容量水只会把水池撑爆。队列结构大概长这样struct dsh_mailbox { dsh_spinlock_t lock; struct dsh_message **buf; uint32_t capacity; uint32_t mask; uint32_t head; /* 出队位置 */ uint32_t tail; /* 入队位置 */ uint32_t count; dsh_wait_queue_head_t recv_wq; dsh_wait_queue_head_t send_wq; bool closing; };capacity必须是 2 的幂这样头和尾的自增都能直接用 mask取模比% capacity快得多。3.1 消息内存的所有权转移谁分配谁释放队列里每个元素是一个struct dsh_message *指针不在入队时拷贝数据。谁分配消息谁就要负责它的生命周期但所有权会转移发送方kmalloc一块消息内存把数据填进去。调用 push把指针放进环形缓冲区。一旦进入缓冲区发送方不再持有这块内存所有权归属队列。接收方从队列里 pop 得到指针处理完消息后自行kfree。为什么用指针而不是直接在队列里存data[64]因为消息长度不确定。如果队列元素固定存 64 字节传大消息时既浪费空间又需要多次拷贝。存指针的话小消息还可以走栈上的临时结构体只是发送前需要拷到堆上这个开销可以接受。所有权转移的规则必须写死在接口文档里。我见过很多野路子代码发送方入队后还继续修改消息内容接收方看到的永远是改了一半的数据。这正是所有权定义不清楚造成的。3.2 锁与 waitqueue为什么不能抱着自旋锁睡觉这是整个队列设计里我花时间最多的地方。队列自身的入队和出队只需要一把自旋锁保护操作非常短。但问题在于队列会满会空。发送方遇到满队列要睡接收方遇到空队列也要睡。睡眠是不能抱着自旋锁睡的否则持有锁的线程睡着其它所有想进出队列的线程全部卡死甚至调度器本身都可能转不动最后整个内核挂掉。所以通用做法是用自旋锁保护环形缓冲区的数据用等待队列来管理阻塞线程。睡眠之前先把自己挂到等待队列上然后把自旋锁释放掉再调度。唤醒的一方在完成入队/出队操作后释放锁然后调用 wake_up。这里的核心难度在于“防止丢失唤醒”。想象这种时序接收方发现队列空。接收方准备睡觉。就在这个准备睡觉的间隙发送方来了把消息放进队列。发送方调用 wake_up但接收方此时还没进入睡眠状态wake_up 没找到它。接收方真的睡过去了永远没人唤醒它。丢失唤醒的本质是判断条件和加入等待队列没有形成一个原子操作。解决方法是把“加入等待队列”的动作提前到“检查条件”之前并且在同一个自旋锁的保护下完成static int mailbox_wait_or_push(struct dsh_mailbox *mb, ...) { unsigned long irqflags; dsh_wait_entry_t wait; int ret 0; dsh_wait_entry_init(wait, current_agent); dsh_spin_lock_irqsave(mb-lock, irqflags); while (mb-count mb-capacity !mb-closing) { if (timeout_expired(...)) { ret -EAGAIN; goto out; } /* 加入等待队列这个动作受 mb-lock 保护 */ dsh_wait_queue_add(mb-send_wq, wait); dsh_spin_unlock_irqrestore(mb-lock, irqflags); dsh_schedule(); /* 睡眠 */ dsh_spin_lock_irqsave(mb-lock, irqflags); dsh_wait_queue_remove(mb-send_wq, wait); } if (mb-closing) { ret -ESHUTDOWN; goto out; } /* 入队 */ mb-buf[mb-tail mb-mask] msg; mb-tail; mb-count; ret 0; out: dsh_spin_unlock_irqrestore(mb-lock, irqflags); return ret; }流程里的关键点先加等待队列再释放锁之后才睡。因为加入等待队列的动作发生在条件判断的同一个临界区里所以发送方不可能插入“入队-唤醒”和“接收方判断-等待”之间。唤醒者发现接收方在等待队列里就一定能把睡下去的人叫醒。醒过来之后会重新抢锁然后while循环再次检查条件。这里的循环很重要因为内核的唤醒可能是“虚假唤醒”也可能是多个接收者同时醒来其中一个把消息抢走了。无论哪种情况都必须重查条件不能凭一次被唤醒就假设一定可以进展。3.3 超时与中断打断的处理原则队列的 push/pop 都支持超时时间。超时判断依赖本地时钟比较单调时间戳不依赖墙钟避免对时导致的行为抖动。dsh_schedule()返回后并不代表条件一定满足了。我们先看时间戳超时了就返回-EAGAIN没超时就继续循环检查条件。这个等待函数还预留了一个状态位把当前 Agent 标记成“可被中断睡眠”。早期版本没有信号机制所以这个状态位暂时不会触发但数据结构里保留 TASK_INTERRUPTIBLE 类似的语义后面做信号系统时不用改接口。每个队列有两个等待队列头一个给发送方一个给接收方。入队完成后只需要唤醒接收方里的一个等待者出队完成后也只需要唤醒发送方里的一个等待者。多消费者同时被唤醒是巨大的浪费所以接收方等待项要设成独占等待。4. 句柄与队列联动发送接收 API 的事务性设计有了句柄表和邮箱队列发送接收接口就水到渠成了。int dsh_agent_send(uint32_t target_handle, const void *data, size_t len, uint32_t flags, int64_t timeout_ns); int dsh_agent_recv(uint32_t self_handle, void *buf, size_t cap, uint32_t flags, int64_t timeout_ns);recv也要穿句柄吗要。虽然接收方本质上只从自己的邮箱里取消息但统一接口要求用户先打开自己的 Agent 句柄然后才能接收。这样句柄既是权限凭证也是后续权限隔离的挂载点。现在看是有点啰嗦但对齐了未来的安全模型。发送路径上的操作序列是校验消息长度和参数。dsh_agent_lookup(target_handle)拿到目标 agent 的带引用指针。目标 agent 内部的引用计数已经帮我们保证了对象不会在发送中途被销毁。分配消息内存填充消息头来源句柄、长度、序号。调用目标 agent 的邮箱 push 操作。无论成败最后都要dsh_agent_put_ref(target)归还引用。接收路径则简单一些通过句柄找到自己这个 agent从它的邮箱里 pop。pop 到消息后把消息拷贝到用户缓冲区释放消息内存。4.1 优雅关闭目标 Agent 退出时阻塞发送者怎么处理Agent 退出不能直接把邮箱销毁了事。如果还有发送者正阻塞在满队列上或者接收者正阻塞在空队列上直接把邮箱释放会让这些沉睡线程醒来后访问一个不存在的对象。我的做法是引入closing状态。Agent 退出时先把邮件标志设置为closing。之后任何新的 push 都直接返回-ESHUTDOWN不再入队。唤醒所有阻塞在send_wq上的线程让它们重新检查条件并退出。对于 recv 等待者不是立即全部赶到门外而是先把已经入队的消息尽量交给接收者等队列空掉之后再让新的 recv 返回-ESHUTDOWN。这个设计遵循了一个原则发送是短路径应该立刻失败接收是消费路径尽量让消费者先把缓冲里已经存在的数据消费完。很多协议里 graceful shutdown 也是这样处理的先停止新数据进入再排空已有数据。实现时有个细节closing的判断必须在自旋锁保护下进行。否则关闭线程和发送线程可能同时看到不一致的状态导致一条消息在关闭之后入队成功。4.2 一个生产者两个消费者的正确性自测写完队列之后我设计了一个最基础的压力自测一个生产者往一个 Agent 的邮箱里投 1000 条消息两个消费者线程同时从这个邮箱里取消息。每取到一条就在自己本地的计数器上加一。这个测试看起来简单却能暴露很多问题。第一版跑出来的结果是两个消费者的计数加起来不稳定比如 999 或 998偶尔还出现同一序号的消息被两个消费者都收到的现象。排查后发现问题出在出队函数里两个消费者在while (mb-count 0)的判断和实际dequeue之间没有全程持有同一把锁。我在入队函数里老老实实加了锁出队函数为了“性能”把锁缩小到了最小结果条件检查和出队动作之间被打断了。修复后的出队逻辑是条件检查、取 head 位置的指针、递增 head、递减 count全部在一个自旋锁临界区内完成。消息内存的所有权转移也在锁内完成。之后锁外再调用唤醒发送方的操作。自测通过后我又加了一个超时场景消费者设置 50ms 的超时去等一个空邮箱正常情况下应该返回-EAGAIN在没有消费者唤醒的干扰下这个超时误差必须控制在 1ms 以内。这个测试是用来验证等待队列超时机制的免得后面系统调的时候超时时间变成摆设。5. 实测踩坑记录从死锁到泄漏的排查链路这一节写的是我在实现这套句柄和队列时真实撞上的三个坑。每个坑都折腾了不少时间回头来看都是很典型的内核并发问题。5.1 坑一持有自旋锁时调用 wake_up系统直接重启最开始我的入队函数长得很自然先加锁入队然后立刻在锁内调用唤醒函数。逻辑上感觉没毛病但跑多生产者压力测试时几千条消息之后直接重启。定位过程是先通过崩溃现场的任务栈回溯发现唤醒函数内部在尝试获取另一个任务的运行队列锁。而那个任务恰好也在等待邮箱锁。两条路径的锁顺序刚好相反变成了循环等待。这个现象的本质是wake_up 不是一个简单的标记操作它可能需要把目标任务从睡眠队列迁移到可运行队列这期间涉及调度器自己的锁。如果你在持有邮箱锁的同时去拿调度器锁而调度路径上又有人试图拿邮箱锁就形成死锁。修复方案很简单也很有效把唤醒动作挪到临界区外面。入队/出队和唤醒不再处于同一个锁保护之下。入队后记录“需要唤醒”释放邮箱锁之后再执行唤醒。代码变成这样dsh_spin_lock_irqsave(mb-lock, irqflags); /* 入队…… */ dsh_spin_unlock_irqrestore(mb-lock, irqflags); dsh_wait_queue_wake_one(mb-recv_wq); /* 锁外唤醒 */锁外唤醒有一个轻微代价可能多唤醒一次空跑。但正确性比那一点冗余开销重要得多。5.2 坑二句柄表槽位被“幽灵引用”占满系统跑了一段时间某个测试用例突然报告句柄分配失败。我检查了调度器Agent 数量远远没到 512 个但句柄表就是满的。用调试打印看了一眼表项发现大量槽位的agent指针不为空但它们的引用计数没有归零这些 Agent 其实早就应该退出了。真正的原因是某个接口的实现里我在lookup成功后就忘了调用put_ref。最讽刺的是那个接口只在内核内部被调用并没有暴露给用户态所以我完全没意识到它会造成引用泄漏。每次调用都偷偷给 Agent 加了一笔永远不还的引用计数Agent 退出时引用计数一直不为零销毁函数不敢动它句柄表项也就一直占着。排查这种问题不能靠肉眼。我在 Agent 销毁路径上加了一个统计打印它“当前存活引用数”和“句柄表项快照”。一跑就发现某个模块调用一次某个 Agent 的引用计数就多一。顺着调用栈找到漏掉put_ref的接口修复后槽位占用数立刻恢复正常。5.3 坑三代际没校验复用的槽位串消息这是最早一版句柄设计没有代际概念时出的问题。当时句柄值就是槽位索引本身。一个 Agent 退出后槽位被复用旧句柄和新 Agent 对上了号。结果就是测试里明明往 Agent B 发消息日志却显示 B 收到了一条标记为“来自旧 Agent A 的上一个生命周期”的消息。排查了很久最后发现发送方缓存的是一个旧句柄而接收方是新的 Agent消息就这么阴差阳错地投递了。后来我把代际加进句柄值并且让句柄表释放时强制代际自增。所有旧句柄在查找时都会被代际校验拦下来。这个教训让我明白句柄不只是“一个索引”它本质上是一条带版本号的安全凭证。每次我在文档里看到别人写“关闭、复用、旧 id 可能会短暂地指向新对象”时都会想起这次的串消息事故。生成代际是最好的防御哪怕偶尔会产生一个无法分配的槽位也比串消息强百倍。6. 性能粗测与后续规划实现了正确性之后我开始关心队列到底能跑多快。测法是固定两个 Agent一个专发一个专收发 10 万条 8 字节的短消息统计总耗时。早期版本带调试输出速度惨不忍睹大约跑 200ms把wake_up移到锁外、把接收方等待改成独占唤醒之后降到 120ms 左右。这个量级相当于每秒约 80 万条消息对教学内核来说够用了。这里给出的数字不是标准 benchmark只是自测参考。调试内核还开着大量日志和断言性能一定比专业 RTOS 差很远。但性能优化里有两个结论是可迁移的第一锁的临界区要短。队列的临界区里只有指针搬运和计数更新绝对不出现消息拷贝、日志输出、内存分配这类慢操作。消息内存的分配是放在入队之前完成的这让临界区的长度非常稳定。第二伪共享是真实存在的。队列的head和tail分别被消费者和生产者高频写。如果它们落在一个缓存行里两个核会互相拖累。把head和tail用缓存行大小对齐分开之后同一台机器上测试快了大约 10%。这个优化不复杂但需要知道硬件缓存行的长度我直接在结构体定义里留了__cacheline_aligned的占位符。后续计划有两件事。第一是把这套邮箱机制接到系统调用层让用户态任务能够通过句柄收发消息这是 DSH 从“玩具内核”变成“能用内核”的关键一步。第二是考虑零拷贝大消息超过一定大小的消息不走指针搬运进邮箱而是通过共享内存区传递邮箱里只放一个“区域描述符”。这样能避免大消息反复分配和释放带来的开销。句柄和队列这两块做完之后Agent 之间的协作终于不再是纸上谈兵了。下一步我会把注意力转向系统调用的中断入口和参数拷贝让内核态和用户态真正打通那时候的测试就不只是自测脚本而是可以跑真实的应用逻辑了。 
长河君 · 2026/10/11 3:39:00 · 阅读 1.2k 手写内核通信基石：Agent句柄与收发队列的设计与实践
DSH内核写到第六篇调度器已经能在几个任务之间来回切换内存管理也把分配器跑稳了。接下来要打通的是 Agent 与 Agent 之间的数据通路。这一篇解决两个问题一个 Agent 想给另一个 Agent 发消息怎么安全地找到对方接收方阻塞等待时怎么被合适地唤醒发送方遇到队列满了怎么睡眠而不会把系统搞死。配套的这套东西就是 Agent 句柄与收发队列。这个阶段对整个内核来说非常关键。前几篇我们把每个执行体叫 Agent它有自己的栈、上下文和调度状态但还没有一条像样的通信管道。没有通信管道调度器调度得再好各个 Agent 也只是孤岛。而通信管道的设计里最容易被低估的就是句柄层。很多初学者觉得给个指针就行了反正内核里都是同一个地址空间。但实际上一旦你开始做资源授权、生命周期管理和异常退出裸指针会变成一串串悬垂指针和双重释放。我会把这一篇的设计思路、数据结构、并发细节和踩坑过程全部展开。代码是 C 风格适合任何正在从零写内核、写教学操作系统、或者做嵌入式 RTOS 扩展的人参考。我不会只给结论会把当时为什么纠结、最后为什么这么选也讲清楚。1. 为什么不能直接传指针Agent 句柄到底在防什么在没有句柄之前我的代码里到处都是struct dsh_agent *agent直接传来传去。看起来省事其实埋了三个雷。第一个雷是权限无法收缩。指针一旦暴露出去持有者可以访问这个结构体的任意字段甚至可以伪造一个dsh_agent对象把自己的数据填进去。内核里很多崩溃和提权漏洞本质都是“不该拿到指针的人拿到了指针”。而我们只是做一个教学内核但也不能在这个地方养成坏习惯。第二个雷是生命周期没法追踪。A 任务拿着指向某 Agent 的指针那个 Agent 退出了结构体被释放A 手里的指针就悬空了。你当然可以说“释放前先通知所有引用者”可是你怎么知道都有谁在引用在这个阶段没有自动追踪机制纯靠自觉迟早出事。第三个雷是对象迁移。后面如果做内核堆压缩、做地址空间切换Agent 控制块本身可能被搬走。指针是绝对地址一搬就全断了。但句柄是逻辑编号表项里的地址更新一下所有持有句柄的调用方都不受影响。打个比方直接把门锁的内部结构告诉访客和给访客一张房卡是两回事。房卡不知道门锁长什么样但前台一查就知道这个卡号对应哪个房间、几点到期、能不能进。句柄就是这个房卡。所以最终设计是每个 Agent 在创建时获得一个全局唯一的句柄类型是uint32_t用户态和内核其他模块只准持有这个整数不允许直接持有结构体指针。结构体的真实地址只存在于句柄表里并且由内核统一管理。1.1 句柄值里为什么要塞一个代际如果一个句柄只是“槽位索引”那还有个经典问题槽位复用。假设 Agent A 占着句柄槽位 3句柄值是 3。后来 A 退出槽位空出来。又新建一个 Agent B系统把槽位 3 分配给它。这时候之前持有“句柄 3”的调用方本来以为自己还在操作 A结果拿到的是 B。这是典型的 ABA 问题。我的解决办法是句柄值由“槽位索引”和“代际”两部分组成。#define DSH_HANDLE_INDEX_MASK 0xffffu #define DSH_HANDLE_GEN_SHIFT 16 static inline uint32_t dsh_handle_encode(uint16_t index, uint16_t generation) { return ((uint32_t)generation DSH_HANDLE_GEN_SHIFT) | index; }每次槽位被释放再分配时代际加一。旧句柄就算还带着旧的槽位编号代际校验也会失败。这样悬垂句柄最多返回一个-EINVAL或-ENOTCONN而不是把一个消息发到错误的 Agent 上。要注意代际不是无限增长的。我用的是 16 位所以会在复用前代际溢出。真溢出时系统直接放弃这个槽位不再分配同时打印一条错误。早期内核没有必要为了省槽位牺牲可靠性。1.2 引用计数句柄表只是入口对象生命周期靠 refcount句柄表解决“怎么找到对象”但“找到之后怎么保证对象不会被并发释放”是另一个问题。经典场景线程 A 拿着句柄准备发送消息已经通过句柄表拿到了struct dsh_agent *agent的地址。同一时刻线程 B 关闭这个句柄并且因为这是最后一个持有者直接释放了 agent 对象。A 还没来得及操作手里的指针就已经变成悬垂的了。解决这个问题不能靠“关闭时等别人用完”因为你没法知道别人什么时候用完。通用做法是引用计数句柄表项里保存一个指向 agent 的强引用也就是refcount。任何从句柄表“取出”agent 的路径都必须先对 agent 做一次原子引用计数增加然后再解锁句柄表。操作结束后调用统一的 put 操作释放这个引用。当用户关闭句柄时只是把句柄表项清除并对 agent 做一次 put。如果还有别的地方正拿着引用agent 不会被释放等所有引用都归零才真正销毁。这是内核里最常用的生命周期模型。不是“释放时小心翼翼地检查有没有人用”而是“释放必须等到最后一个用户主动放手”。我早期觉得引用计数啰嗦后来被悬垂指针折磨过两次再也不那么想了。2. 句柄表的具体实现从分配、校验到并发释放DSH 的句柄表是一个定长数组默认 512 个槽位。这个数字对教学内核足够用而且定长数组让分配逻辑非常简单遍历找空槽即可。表项结构体是这种风格struct dsh_handle_entry { struct dsh_agent_object *agent; uint16_t generation; uint16_t flags; struct dsh_handle_entry *next_free; }; struct dsh_handle_table { struct dsh_handle_entry entries[DSH_AGENT_HANDLE_MAX]; struct dsh_handle_entry *free_list; dsh_spinlock_t lock; };我没有用哈希表因为 Agent 数量少密集编号反而比哈希更简单。哈希表适合句柄值很分散、数量很大的场景但教学内核里 512 个槽位从头遍历一次也就几十纳秒完全不是瓶颈。全局有一把自旋锁保护整个句柄表。这个选择是刻意的句柄表的操作非常短无非就是加载、比较、赋值不会睡眠。自旋锁在这种临界区里是最合适的就算多核竞争等一小会儿也就拿到了。但要注意绝不能在持有这把锁的时候调用可能睡眠的函数。2.1 句柄分配与释放的完整流程创建一个新 Agent 时会执行dsh_agent_handle_allocuint32_t dsh_agent_handle_alloc(struct dsh_agent_object *agent) { uint32_t handle 0; uint64_t irqflags; dsh_spin_lock_irqsave(g_handle_table.lock, irqflags); struct dsh_handle_entry *e g_handle_table.free_list; if (!e) { dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return 0; /* 表满 */ } g_handle_table.free_list e-next_free; e-agent agent; e-generation; if (e-generation 0) { /* 代际溢出放弃这个槽位 */ dsh_handle_table_park_slot(e); dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return 0; } agent-refcount_inc(agent); handle dsh_handle_encode(e - g_handle_table.entries, e-generation); dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return handle; }分配时的refcount_inc是句柄表自己的那一份引用。因此每个 Agent 只要还有句柄在表里引用计数就至少是 1。关闭句柄时对应做一次refcount_dec。这样即使 Agent 自己已经决定退出只要用户还有句柄没关底层对象就不会凭空消失。释放句柄的函数逻辑正好反过来校验代际清空 agent 指针把槽位挂回空闲链表最后对 agent 做一次 put。这里有个细节关闭句柄时如果发现代理已经处于“正在退出”状态我们仍然要执行同样的表项清理和 put只是后续对被关闭对象的新操作都会立刻返回错误。2.2 句柄校验与查找先加引用再返回指针从句柄到对象的核心函数是dsh_agent_lookupstruct dsh_agent_object *dsh_agent_lookup(uint32_t handle) { uint32_t idx handle DSH_HANDLE_INDEX_MASK; uint32_t gen handle DSH_HANDLE_GEN_SHIFT; struct dsh_handle_entry *e; struct dsh_agent_object *agent NULL; uint64_t irqflags; if (idx DSH_AGENT_HANDLE_MAX) return NULL; dsh_spin_lock_irqsave(g_handle_table.lock, irqflags); e g_handle_table.entries[idx]; if (e-agent e-generation gen) { agent e-agent; agent-refcount_inc(agent); } dsh_spin_unlock_irqrestore(g_handle_table.lock, irqflags); return agent; }注意顺序拿到 agent 后还在锁内就完成了引用计数增加然后再解锁。因为一旦解锁另一个线程可能马上关掉句柄并释放 agent所以“取指针”和“加引用”必须是一个不可分割的动作。调用方使用完拿到的 agent 后必须调用dsh_agent_put_ref(agent)归还引用。我在每个发送路径的函数入口处都会写清楚提示dsh_agent_lookup返回的是带引用的指针。调用方不负责释放 agent但负责调用put_ref。漏掉这一步Agent 会永远无法销毁。这个提示是我在项目代码注释里反复强调的。内核里引用计数泄漏比普通内存泄漏难查得多因为它的表现不是立刻崩而是“某些对象回收不掉”直到某一天句柄表满了。3. 收发队列有界环形缓冲区与等待唤醒句柄解决了“找到对方”真正传数据的是收发队列。每个 Agent 创建时都会伴随一个接收邮箱。其它 Agent 通过句柄找到这个 Agent然后往邮箱里 push 消息Agent 自己从邮箱里 pop 消息处理。我坚持使用有界环形缓冲区没有做成无界链表。核心原因是无界队列会掩盖调度问题生产者不断生产消费者跟不上内存像滚雪球一样增长最后系统在内存耗尽中崩溃。有界队列是天然的背压机制。这跟水管一样有容量才有压力水压会让上游减速没有容量水只会把水池撑爆。队列结构大概长这样struct dsh_mailbox { dsh_spinlock_t lock; struct dsh_message **buf; uint32_t capacity; uint32_t mask; uint32_t head; /* 出队位置 */ uint32_t tail; /* 入队位置 */ uint32_t count; dsh_wait_queue_head_t recv_wq; dsh_wait_queue_head_t send_wq; bool closing; };capacity必须是 2 的幂这样头和尾的自增都能直接用 mask取模比% capacity快得多。3.1 消息内存的所有权转移谁分配谁释放队列里每个元素是一个struct dsh_message *指针不在入队时拷贝数据。谁分配消息谁就要负责它的生命周期但所有权会转移发送方kmalloc一块消息内存把数据填进去。调用 push把指针放进环形缓冲区。一旦进入缓冲区发送方不再持有这块内存所有权归属队列。接收方从队列里 pop 得到指针处理完消息后自行kfree。为什么用指针而不是直接在队列里存data[64]因为消息长度不确定。如果队列元素固定存 64 字节传大消息时既浪费空间又需要多次拷贝。存指针的话小消息还可以走栈上的临时结构体只是发送前需要拷到堆上这个开销可以接受。所有权转移的规则必须写死在接口文档里。我见过很多野路子代码发送方入队后还继续修改消息内容接收方看到的永远是改了一半的数据。这正是所有权定义不清楚造成的。3.2 锁与 waitqueue为什么不能抱着自旋锁睡觉这是整个队列设计里我花时间最多的地方。队列自身的入队和出队只需要一把自旋锁保护操作非常短。但问题在于队列会满会空。发送方遇到满队列要睡接收方遇到空队列也要睡。睡眠是不能抱着自旋锁睡的否则持有锁的线程睡着其它所有想进出队列的线程全部卡死甚至调度器本身都可能转不动最后整个内核挂掉。所以通用做法是用自旋锁保护环形缓冲区的数据用等待队列来管理阻塞线程。睡眠之前先把自己挂到等待队列上然后把自旋锁释放掉再调度。唤醒的一方在完成入队/出队操作后释放锁然后调用 wake_up。这里的核心难度在于“防止丢失唤醒”。想象这种时序接收方发现队列空。接收方准备睡觉。就在这个准备睡觉的间隙发送方来了把消息放进队列。发送方调用 wake_up但接收方此时还没进入睡眠状态wake_up 没找到它。接收方真的睡过去了永远没人唤醒它。丢失唤醒的本质是判断条件和加入等待队列没有形成一个原子操作。解决方法是把“加入等待队列”的动作提前到“检查条件”之前并且在同一个自旋锁的保护下完成static int mailbox_wait_or_push(struct dsh_mailbox *mb, ...) { unsigned long irqflags; dsh_wait_entry_t wait; int ret 0; dsh_wait_entry_init(wait, current_agent); dsh_spin_lock_irqsave(mb-lock, irqflags); while (mb-count mb-capacity !mb-closing) { if (timeout_expired(...)) { ret -EAGAIN; goto out; } /* 加入等待队列这个动作受 mb-lock 保护 */ dsh_wait_queue_add(mb-send_wq, wait); dsh_spin_unlock_irqrestore(mb-lock, irqflags); dsh_schedule(); /* 睡眠 */ dsh_spin_lock_irqsave(mb-lock, irqflags); dsh_wait_queue_remove(mb-send_wq, wait); } if (mb-closing) { ret -ESHUTDOWN; goto out; } /* 入队 */ mb-buf[mb-tail mb-mask] msg; mb-tail; mb-count; ret 0; out: dsh_spin_unlock_irqrestore(mb-lock, irqflags); return ret; }流程里的关键点先加等待队列再释放锁之后才睡。因为加入等待队列的动作发生在条件判断的同一个临界区里所以发送方不可能插入“入队-唤醒”和“接收方判断-等待”之间。唤醒者发现接收方在等待队列里就一定能把睡下去的人叫醒。醒过来之后会重新抢锁然后while循环再次检查条件。这里的循环很重要因为内核的唤醒可能是“虚假唤醒”也可能是多个接收者同时醒来其中一个把消息抢走了。无论哪种情况都必须重查条件不能凭一次被唤醒就假设一定可以进展。3.3 超时与中断打断的处理原则队列的 push/pop 都支持超时时间。超时判断依赖本地时钟比较单调时间戳不依赖墙钟避免对时导致的行为抖动。dsh_schedule()返回后并不代表条件一定满足了。我们先看时间戳超时了就返回-EAGAIN没超时就继续循环检查条件。这个等待函数还预留了一个状态位把当前 Agent 标记成“可被中断睡眠”。早期版本没有信号机制所以这个状态位暂时不会触发但数据结构里保留 TASK_INTERRUPTIBLE 类似的语义后面做信号系统时不用改接口。每个队列有两个等待队列头一个给发送方一个给接收方。入队完成后只需要唤醒接收方里的一个等待者出队完成后也只需要唤醒发送方里的一个等待者。多消费者同时被唤醒是巨大的浪费所以接收方等待项要设成独占等待。4. 句柄与队列联动发送接收 API 的事务性设计有了句柄表和邮箱队列发送接收接口就水到渠成了。int dsh_agent_send(uint32_t target_handle, const void *data, size_t len, uint32_t flags, int64_t timeout_ns); int dsh_agent_recv(uint32_t self_handle, void *buf, size_t cap, uint32_t flags, int64_t timeout_ns);recv也要穿句柄吗要。虽然接收方本质上只从自己的邮箱里取消息但统一接口要求用户先打开自己的 Agent 句柄然后才能接收。这样句柄既是权限凭证也是后续权限隔离的挂载点。现在看是有点啰嗦但对齐了未来的安全模型。发送路径上的操作序列是校验消息长度和参数。dsh_agent_lookup(target_handle)拿到目标 agent 的带引用指针。目标 agent 内部的引用计数已经帮我们保证了对象不会在发送中途被销毁。分配消息内存填充消息头来源句柄、长度、序号。调用目标 agent 的邮箱 push 操作。无论成败最后都要dsh_agent_put_ref(target)归还引用。接收路径则简单一些通过句柄找到自己这个 agent从它的邮箱里 pop。pop 到消息后把消息拷贝到用户缓冲区释放消息内存。4.1 优雅关闭目标 Agent 退出时阻塞发送者怎么处理Agent 退出不能直接把邮箱销毁了事。如果还有发送者正阻塞在满队列上或者接收者正阻塞在空队列上直接把邮箱释放会让这些沉睡线程醒来后访问一个不存在的对象。我的做法是引入closing状态。Agent 退出时先把邮件标志设置为closing。之后任何新的 push 都直接返回-ESHUTDOWN不再入队。唤醒所有阻塞在send_wq上的线程让它们重新检查条件并退出。对于 recv 等待者不是立即全部赶到门外而是先把已经入队的消息尽量交给接收者等队列空掉之后再让新的 recv 返回-ESHUTDOWN。这个设计遵循了一个原则发送是短路径应该立刻失败接收是消费路径尽量让消费者先把缓冲里已经存在的数据消费完。很多协议里 graceful shutdown 也是这样处理的先停止新数据进入再排空已有数据。实现时有个细节closing的判断必须在自旋锁保护下进行。否则关闭线程和发送线程可能同时看到不一致的状态导致一条消息在关闭之后入队成功。4.2 一个生产者两个消费者的正确性自测写完队列之后我设计了一个最基础的压力自测一个生产者往一个 Agent 的邮箱里投 1000 条消息两个消费者线程同时从这个邮箱里取消息。每取到一条就在自己本地的计数器上加一。这个测试看起来简单却能暴露很多问题。第一版跑出来的结果是两个消费者的计数加起来不稳定比如 999 或 998偶尔还出现同一序号的消息被两个消费者都收到的现象。排查后发现问题出在出队函数里两个消费者在while (mb-count 0)的判断和实际dequeue之间没有全程持有同一把锁。我在入队函数里老老实实加了锁出队函数为了“性能”把锁缩小到了最小结果条件检查和出队动作之间被打断了。修复后的出队逻辑是条件检查、取 head 位置的指针、递增 head、递减 count全部在一个自旋锁临界区内完成。消息内存的所有权转移也在锁内完成。之后锁外再调用唤醒发送方的操作。自测通过后我又加了一个超时场景消费者设置 50ms 的超时去等一个空邮箱正常情况下应该返回-EAGAIN在没有消费者唤醒的干扰下这个超时误差必须控制在 1ms 以内。这个测试是用来验证等待队列超时机制的免得后面系统调的时候超时时间变成摆设。5. 实测踩坑记录从死锁到泄漏的排查链路这一节写的是我在实现这套句柄和队列时真实撞上的三个坑。每个坑都折腾了不少时间回头来看都是很典型的内核并发问题。5.1 坑一持有自旋锁时调用 wake_up系统直接重启最开始我的入队函数长得很自然先加锁入队然后立刻在锁内调用唤醒函数。逻辑上感觉没毛病但跑多生产者压力测试时几千条消息之后直接重启。定位过程是先通过崩溃现场的任务栈回溯发现唤醒函数内部在尝试获取另一个任务的运行队列锁。而那个任务恰好也在等待邮箱锁。两条路径的锁顺序刚好相反变成了循环等待。这个现象的本质是wake_up 不是一个简单的标记操作它可能需要把目标任务从睡眠队列迁移到可运行队列这期间涉及调度器自己的锁。如果你在持有邮箱锁的同时去拿调度器锁而调度路径上又有人试图拿邮箱锁就形成死锁。修复方案很简单也很有效把唤醒动作挪到临界区外面。入队/出队和唤醒不再处于同一个锁保护之下。入队后记录“需要唤醒”释放邮箱锁之后再执行唤醒。代码变成这样dsh_spin_lock_irqsave(mb-lock, irqflags); /* 入队…… */ dsh_spin_unlock_irqrestore(mb-lock, irqflags); dsh_wait_queue_wake_one(mb-recv_wq); /* 锁外唤醒 */锁外唤醒有一个轻微代价可能多唤醒一次空跑。但正确性比那一点冗余开销重要得多。5.2 坑二句柄表槽位被“幽灵引用”占满系统跑了一段时间某个测试用例突然报告句柄分配失败。我检查了调度器Agent 数量远远没到 512 个但句柄表就是满的。用调试打印看了一眼表项发现大量槽位的agent指针不为空但它们的引用计数没有归零这些 Agent 其实早就应该退出了。真正的原因是某个接口的实现里我在lookup成功后就忘了调用put_ref。最讽刺的是那个接口只在内核内部被调用并没有暴露给用户态所以我完全没意识到它会造成引用泄漏。每次调用都偷偷给 Agent 加了一笔永远不还的引用计数Agent 退出时引用计数一直不为零销毁函数不敢动它句柄表项也就一直占着。排查这种问题不能靠肉眼。我在 Agent 销毁路径上加了一个统计打印它“当前存活引用数”和“句柄表项快照”。一跑就发现某个模块调用一次某个 Agent 的引用计数就多一。顺着调用栈找到漏掉put_ref的接口修复后槽位占用数立刻恢复正常。5.3 坑三代际没校验复用的槽位串消息这是最早一版句柄设计没有代际概念时出的问题。当时句柄值就是槽位索引本身。一个 Agent 退出后槽位被复用旧句柄和新 Agent 对上了号。结果就是测试里明明往 Agent B 发消息日志却显示 B 收到了一条标记为“来自旧 Agent A 的上一个生命周期”的消息。排查了很久最后发现发送方缓存的是一个旧句柄而接收方是新的 Agent消息就这么阴差阳错地投递了。后来我把代际加进句柄值并且让句柄表释放时强制代际自增。所有旧句柄在查找时都会被代际校验拦下来。这个教训让我明白句柄不只是“一个索引”它本质上是一条带版本号的安全凭证。每次我在文档里看到别人写“关闭、复用、旧 id 可能会短暂地指向新对象”时都会想起这次的串消息事故。生成代际是最好的防御哪怕偶尔会产生一个无法分配的槽位也比串消息强百倍。6. 性能粗测与后续规划实现了正确性之后我开始关心队列到底能跑多快。测法是固定两个 Agent一个专发一个专收发 10 万条 8 字节的短消息统计总耗时。早期版本带调试输出速度惨不忍睹大约跑 200ms把wake_up移到锁外、把接收方等待改成独占唤醒之后降到 120ms 左右。这个量级相当于每秒约 80 万条消息对教学内核来说够用了。这里给出的数字不是标准 benchmark只是自测参考。调试内核还开着大量日志和断言性能一定比专业 RTOS 差很远。但性能优化里有两个结论是可迁移的第一锁的临界区要短。队列的临界区里只有指针搬运和计数更新绝对不出现消息拷贝、日志输出、内存分配这类慢操作。消息内存的分配是放在入队之前完成的这让临界区的长度非常稳定。第二伪共享是真实存在的。队列的head和tail分别被消费者和生产者高频写。如果它们落在一个缓存行里两个核会互相拖累。把head和tail用缓存行大小对齐分开之后同一台机器上测试快了大约 10%。这个优化不复杂但需要知道硬件缓存行的长度我直接在结构体定义里留了__cacheline_aligned的占位符。后续计划有两件事。第一是把这套邮箱机制接到系统调用层让用户态任务能够通过句柄收发消息这是 DSH 从“玩具内核”变成“能用内核”的关键一步。第二是考虑零拷贝大消息超过一定大小的消息不走指针搬运进邮箱而是通过共享内存区传递邮箱里只放一个“区域描述符”。这样能避免大消息反复分配和释放带来的开销。句柄和队列这两块做完之后Agent 之间的协作终于不再是纸上谈兵了。下一步我会把注意力转向系统调用的中断入口和参数拷贝让内核态和用户态真正打通那时候的测试就不只是自测脚本而是可以跑真实的应用逻辑了。