数据截至 (上游 commit aae05d825bb5)
主 机↔客机通信:二进制帧协议、virtio-console relay 与 agentd 会话循环
30 秒导读: Microsandbox 把不受信任的代码关进一台微型虚拟机(microVM)。 主机上的进程(CLI、SDK)要给里面下命令、收结果,靠的是一根虚拟串口。这一章讲这根 管子怎么工作:帧长什么样、字节怎么在主机和客机之间流动、以及最关键的——一根物理管子怎么 同时服务好几个客户端而不串线。
本章只讲"通信管道"本身。进程怎么被拉起、微VM 怎么启动 → 见 01-lifecycle;
tcp.* 帧背后真正的用户态网络栈 → 见 05-network-security。
1. 这根"管子"要解决什么(零基础也能懂)
问题:主机和客机是两个世界。 客机是一台真正的虚拟机,有自己的内核、自己的内存、自己的 进程表。主机进程不能直接调用客机里的函数,也读不到客机进程的 stdout。它们之间隔着虚拟化边界—— 这正是沙箱安全的来源,但也意味着任何交互都得显式地"搬字节"过去。
唯一的通道是一根虚拟串口。 虚拟机监视器(VMM,这里是 libkrun)给客机插了一个
virtio-console 设备,名字叫 "agent"。它两头是打通的:客机往里写的字节,主机能读到;主机
写进去的,客机能读到。就是一根 双向的"水管",一次只能顺序地流字节。
于是需要一套约定,把结构化的请求塞进这根管子。 光有字节流不够——"执行 ls -la"、
"这是它的 stdout"、"它退出码是 0" 这些语义,得编码成一个个帧(frame),一个帧一条消息。
这套编码 + 帧头 + 消息类型的约定,就是 crates/protocol 定义的线协议(wire protocol)。
难点在于:一根管子,多个说话人。 你可能同时开了三个 msb exec(三个独立的 SDK 客户端),
还在后台跑文件传输。它们全都挤在这同一根串口上。客机回一个 stdout 帧,主机侧怎么知道
该把它交给三个客户端里的哪一个?本章的高潮就是 relay 的答案:不改帧头,靠 correlation id
分区间。
用起来是这样的(SDK 侧,概念示意):
// 示意,非源码 —— 展示"一个客户端如何借这根管子跟客机对话"
let bridge = AgentBridge::connect_sandbox("my-box").await?; // 连上 relay 的 Unix socket
let (corr_id, stream) = bridge.stream_open(flags, exec_body).await?; // 开一个 exec 会话
bridge.send(corr_id, 0, stdin_body).await?; // 往同一会话喂 stdin
while let Some(frame) = bridge.stream_next(handle).await? { // 收 stdout/stderr/exited
// frame.id 一定等于 corr_id;relay 保证不会串进别人的帧
}
一句话直觉: 把这根管子想成一条共享的传送带。每个包裹(帧)贴了一个"寄件人编号" (correlation id);relay 是分拣中心,预先给每个客户端划了一段不重叠的编号段,客机回来的 包裹按编号一算就知道该退回哪个客户端——分拣时根本不用拆包裹(不解 CBOR)。
2. 顶层全景(它大概怎么转)
五段接力。 从 SDK 调一行代码,到客机里真的跑起一个进程,字节要经过五道关。先看全景图, 再逐段拆。
怎么读这张图:从上到下是一次请求的去程(host→guest),回程(guest→host)沿反方向走同一条路。 中间那根粗线就是唯一的物理管子 virtio-console。
┌──────────────────────────── 主机进程(sandbox process)────────────────────────────┐
│ │
│ SDK 客户端 A SDK 客户端 B SDK 客户端 C │
│ (msb exec) (文件传输) (另一个 exec) │
│ │ Unix socket / named pipe │ │ │
│ └──────────────┬─────────────┴──────────────────┘ │
│ ▼ │
│ ┌──────────────────┐ 每个客户端握手时分到一段不重叠的 id 区间 │
│ │ AgentRelay │ 回程按 id / STEP 算出 slot,退回正确的客户端 │
│ │ (relay.rs) │ │
│ └───────┬──────────┘ │
│ rx_ring ▲ │ tx_ring ← 两个无锁环形队列 + wake pipe(console.rs) │
│ │ ▼ │
│ ┌───────┴───────────────┐ │
│ │ AgentConsoleBackend │ libkrun 的 console 设备从这里读/写 │
│ └───────────────────────┘ │
└══════════════════════ virtio-console 串口 "agent" ═══════════════════════════════════╡
┌───────────────────────── ─── 客机(microVM guest)────────────────────────────────────┐
│ ┌───────────────────────┐ │
│ │ /dev/hvc? (serial.rs) │ agentd 打开这个串口 fd │
│ └───────────┬───────────┘ │
│ ▼ │
│ ┌──────────────────┐ 单线程 tokio select 循环: │
│ │ agentd 主循环 │ 读帧 → handle_message → 派发到会话 │
│ │ (agent.rs) │ 会话输出 → 编码成帧 → 写回串口 │
│ └───────┬──────────┘ │
│ ExecSession │ FsReadSession / FsWriteSession TcpSession │
│ (真进程) ▼ (文件流) (真 socket) │
│ ┌──────────────────┐ │
│ │ 被沙箱的子进程 │ ← 这里才真正 fork/exec 用户代码 │
│ └──────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
各部件一句话职责:
| 部件 | 干什么 | 在哪 |
|---|---|---|
| 线协议 codec | 把 Message 编解码成 [len][id][flags][CBOR] 帧 | crates/protocol/lib/codec.rs |
| 消息类型表 | 定义所有帧类型和它们的域(exec/fs/tcp/core) | crates/protocol/lib/message.rs |
AgentConsoleBackend | 主机侧 console 设备后端,读写两个环形队列 | crates/runtime/lib/console.rs:77 |
AgentRelay | 多路复用:socket ↔ 环形队列,按 id 路由 | crates/runtime/lib/relay.rs:97 |
| agentd 主循环 | 客机侧:串口 I/O + 会话管理 + 心跳 | crates/agentd/lib/agent.rs:149 |
ExecSession 等 | 客机侧一个 exec/fs/tcp 会话的句柄 | crates/agentd/lib/session.rs:122 |
AgentBridge / AgentClient | SDK 侧客户端桥,发帧/收帧 | sdk/rust/lib/agent/bridge.rs:50 |
主线走一遍(高层,不进代码): SDK 调 stream_open → AgentClient 从自己分到的 id 区间里
领一个 correlation id,把 ExecRequest 编成帧,写进 Unix socket → relay 的 client_reader_task
读到帧,原样塞进 rx_ring → console 后端把字节喂进串口 → agentd 读到帧,handle_message
派发,ExecSession::spawn fork 出真进程 → 进程的 stdout 被后台任务读出来,编成
ExecStdout 帧(帧的 id 沿用请求的 id)写回串口 → relay 的 ring_reader_task 读到,用
id / STEP 算出是哪个客户端,推给它 → SDK 的 stream_next 收到帧。去程回程共用一根管子,
全靠帧头里那个 id 分辨归属。
3. 线协议:帧长什么样
这一节讲最底层的约定——一个帧的字节布局,以及帧头为什么要把 id/flags 摆在 CBOR 外面。
3.1 帧格式:定长二进制头 + CBOR 身体
每个帧的线格式是固定的:
[len: u32 BE][id: u32 BE][flags: u8][CBOR(v, t, p)]
└─4 字节──┘ └───────── len 覆盖的范围 ─────────────┘
len:长度前缀,不含自己,覆盖后面id + flags + CBOR的总字节数。id:correlation id,关联请求与响应、标识会话。flags:一个字节的标志位(见 3.3)。- CBOR 身体:
v(协议版本)、t(消息类型)、p(载荷字节)。
关键设计:id 和 flags 复制在二进制帧头里,而不是埋在 CBOR 里。 这样 relay 这种
"中间人"不用解 CBOR 就能读到路由要用的 id 和 flags——分拣不用拆包裹。代码里这是靠给
Message 的这两个字段打 #[serde(skip)] 实现的:
真实实现见 crates/protocol/lib/message.rs:47 的 Message 结构体——id(:65)和
flags(:71)都标了 #[serde(skip)],所以 CBOR 里没有它们;它们由 codec 在帧头单独读写。
两组大小常量:
| 常量 | 值 | 含义 | 位置 |
|---|---|---|---|
MAX_FRAME_SIZE | 4 MiB | 单帧上限(len 之后的全部) | crates/protocol/lib/codec.rs:23 |
FRAME_HEADER_SIZE | 5 | id(4) + flags(1) | crates/protocol/lib/message.rs:32 |
3.2 一个帧怎么读出来
codec 提供两套 API:CBOR-盲的原始帧(给 relay 用)和 CBOR-aware 的类型帧(给收发端用)。 先看原始帧解码的骨架——这正是 relay 和 agentd 热路径上跑的东西:
// 示意,非源码 —— 提炼自 try_decode_raw_from_buf 的判断顺序
if buf.len() < 4 { return None; } // 连长度前缀都不够
let frame_len = u32::from_be(buf[0..4]); // 读长度
if frame_len > MAX_FRAME_SIZE { return Err(TooLarge); } // 防炸内存
if buf.len() < 4 + frame_len { return None; } // 帧还没到齐,等更多字节
if frame_len < FRAME_HEADER_SIZE { return Err(TooShort); } // 连 id+flags 都放不下
let id = u32::from_be(buf[4..8]); // 帧头里直接取 id
let flags = buf[8]; // 帧头里直接取 flags
let body = buf[9 .. 4+frame_len]; // 剩下是 CBOR,先不解
重点看:id 和 flags 是从固定偏移直接切出来的,CBOR 身体原样留着不动。
真实实现:crates/protocol/lib/codec.rs:82 try_decode_raw_from_buf(同步、增量解析,给基于
缓冲区的场景)、:121 read_raw_frame(异步、从任意 AsyncRead 读一整帧)、:259
decode_message_frame(从借用切片解一帧成 Message,不拷贝)。写侧是 :54
encode_raw_to_buf。RawFrame(:36)就是"帧头解了、身体没动"的中间态。
3.3 flags:三个标志位驱动"控制流"
flags 是一个字节,目前用三位。它们不是可有可无的元数据——relay 和两端靠它们判断会话边界和 控制意图,同样不用解 CBOR:
| 标志 | 位 | 含义 | 打在哪些类型上 |
|---|---|---|---|
FLAG_TERMINAL | 0b001 | 该 id 的最后一帧 | ExecExited/ExecFailed/FsResponse/TcpClosed/TcpFailed/CoreError |
FLAG_SESSION_START | 0b010 | 新会话的第一帧 | ExecRequest/FsRequest/TcpConnect |
FLAG_SHUTDOWN | 0b100 | 请求沙箱关机 | Shutdown |
flags 不是随手设的,而是由消息类型算出来的——MessageType::flags()
(crates/protocol/lib/message.rs:274)是一张 match 表,Message::new/with_payload
构造时自动填。这带来一个不变量:flags 必须和类型对得上。agentd 收帧时会校验
msg.flags != msg.t.flags(),不符就回一个 core.error(crates/agentd/lib/agent.rs:265)。
3.4 分域消息:一张类型全景表
MessageType(crates/protocol/lib/message.rs:96)是所有帧类型的单一来源。每个变体用
#[strum(serialize = "core.xxx")] 声明它的线字符串,as_str/from_wire_str 都由它派生。
按域分组:
core 域(会话无关的控制与握手,crates/protocol/lib/core.rs):
| 类型 | 线字符串 | 载荷 | 方向 |
|---|---|---|---|
Ready | core.ready | Ready(boot 计时 + agent 版本):15 | 客机→主机 |
InitResolved | core.init.resolved | InitResolved(默认用户):97 | 客机→主机 |
InitAck | core.init.ack | InitAck{}:117 | 主机→客机 |
Shutdown | core.shutdown | 无 | 主机→客机 |
RelayClientDisconnected | core.relay.client.disconnected | id 区间:125 | 主机relay→客机 |
ClockSync | core.clock.sync | ClockSync(主机纳秒时间):44 | 主机→客机 |
CoreError | core.error | CoreError(kind+message):56 | 双向 |
exec 域(crates/protocol/lib/exec.rs,一次命令执行的全生命周期):
| 类型 | 载荷 | 说明 |
|---|---|---|
ExecRequest | ExecRequest:13 | cmd/args/env/cwd/user/tty/rlimits——开一个 exec 会话 |
ExecStarted | ExecStarted{pid}:65 | 进程已起,回 PID |
ExecStdin | ExecStdin{data}:72 | 主机喂 stdin;空 data = EOF |
ExecStdout/ExecStderr | {data}:98/:106 | 客机流出输出 |
ExecExited | ExecExited{code}:114 | 进程退出(terminal) |
ExecFailed | ExecFailed{kind,errno,...}:123 | 没跑起来(区别于退出);带 ExecFailureKind 分类 |
ExecSignal | ExecSignal{signal}:205 | 主机发信号 |
ExecResize | ExecResize{rows,cols}:195 | PTY 改尺寸 |
注意 ExecExited 与 ExecFailed 的刻意区分:前者是"用户程序跑了、给了退出码",后者是
"用户程序压根没跑起来"(二进制找不到、权限不足、cwd 无效……),后者带结构化的
ExecFailureKind(:150)供 CLI/SDK 挑提示语。
fs 域(crates/protocol/lib/fs.rs,文件系统操作 + 大文件流式传输):
FsRequest(:229)包一个FsOp(:20)枚举——Stat/List/Read/Write/OpenFile/Mkdir… 一大票操作。FsResponse(:287)是终结响应(ok/error/data)。FsData(:304)是流式数据块,读写大文件时一块块传;空 data 表示 EOF。单块上限FS_CHUNK_SIZE = 3 MiB(:12),刻意压在 4 MiB 帧上限之下留 CBOR 信封余量。
tcp 域(crates/protocol/lib/tcp.rs,把客机内发起的 TCP 连接代理出去):
TcpConnect{host,port}(:11)开会话 →TcpConnected确认。TcpData{data}(:25)双向流字节;TcpEof(:33)半关;TcpClosed(:41)/TcpFailed(:45)终结。- 这些帧只是"管道语义";字节背后真正走的用户态网络栈见 05-network-security。
heartbeat(crates/protocol/lib/heartbeat.rs)不是走串口的帧,而是 agentd 写进客机
/.msb/heartbeat.json 的存活脉搏:Heartbeat(:12)带 heartbeat_seq/activity_seq/
各类活动会话计数 + ActivityCounters(:40)。主机据此判断沙箱是否还活着(见 §6.3)。
3.5 版本协商:一条轴,两处闸门
协议只有一条版本轴 PROTOCOL_VERSION = 7(crates/protocol/lib/message.rs:12)。每个消息
类型标注它"从第几代开始有":MessageType::min_protocol_version()(:287)——core/exec 是第 1 代
基线,fs 要第 2 代,tcp 要第 4 代,core.error 是第 5 代。
发送方在发之前就闸门:is_available_at(peer_generation)(:325)判断对端够不够新。新主机遇到
旧客机时,发不出去的高版本特性会被本地拒掉(返回 typed error),只有那一个特性失败,整个会话
不受影响。SDK 侧用 AgentClient::ensure_version_compat(packages/agent-client/rust/lib/client.rs:395)
做同一件事。协商结果是 min(自己的版本, 对端 ready 帧里报的 v)(client.rs:644)。
4. 传输载体:字节怎么真正过界
上一节讲了"帧长什么样"。这一节讲"帧怎么被搬过虚拟化边界"——没有 socket,没有系统调用在数据 路径上,只有环形队列 + memcpy。
4.1 客机侧:agentd 找到并打开那根串口
virtio-console 在客机内表现为一个字符设备。agentd 启动时扫 /sys/class/virtio-ports/,读每个
端口的 name 文件,找到名字 等于 "agent" 的那个,返回 /dev/<port_id>:
真实实现:crates/agentd/lib/serial.rs:25 find_serial_port;端口名常量
AGENT_PORT_NAME = "agent" 在 crates/protocol/lib/lib.rs:79,由 serial.rs:15 re-export。
打开用 open_serial_port(crates/agentd/lib/agent.rs:386)——只 open 一次,因为
virtio-console 多端口设备第二次 open 会 EBUSY;早期握手(blocking)和主循环(async)复用同一个 fd。
4.2 主机侧:环形队列取代 socketpair
主机侧不走 socket。AgentConsoleBackend(crates/runtime/lib/console.rs:77)是交给 libkrun 的
console 设备后端,它背后是 ConsoleSharedState(:56)——两个无锁环形队列加两个唤醒管道:
| 字段 | 方向 | 谁 push | 谁 pop |
|---|---|---|---|
tx_ring | 客机→主机 | console TX 线程 | relay 的 ring reader |
rx_ring | 主机→客机 | relay 的 ring writer | console RX 线程 |
tx_wake | 信号 | — | 唤醒 relay:"客机有数据了" |
rx_wake | 信号 | — | 唤醒 console RX 线程:"客机的数据备好了" |
命名以客机视角为准:tx = 客机发出的,rx = 客机收到的。
为什么这么设计:数据路径上零系统调用。 队列用 crossbeam_queue::ArrayQueue,搬数据靠
memcpy(切片拷贝);只有"通知对方有活干"才写一次 WakePipe(1 字节管道写)。这跟网络 crate 里
smoltcp 的 SharedState 是同一套路数。
后端的 read/write 都是 &self(console.rs:182/:222):RX 线程调 read(先服务上次没塞下的
pending 余料,再从 rx_ring pop),TX 线程调 write(往 tx_ring push 再 tx_wake.wake())。
队列满就返回 WouldBlock——背压天然形成。
一个易漏的细节:read 一进来先 rx_wake.drain() 再查队列(console.rs:185),这样即使 VMM 用
边沿触发轮询,后续通知也不会丢。
5. relay 的多客户端路由(本章高潮)
AgentRelay(crates/runtime/lib/relay.rs:97)是整个通信层最巧的部件。它要解决的核心难题
在 §1 说过:一根物理串口,多个 SDK 客户端,回程帧怎么不串线。
5.1 核心把戏:correlation id 分区间
答案是给帧头里那个 32 位 id 划段。relay 预设最多 128 个客户端槽:
AGENT_RELAY_MAX_CLIENTS = 128
AGENT_RELAY_ID_RANGE_STEP = u32::MAX / 128 ≈ 33_554_431
(常量在 crates/protocol/lib/lib.rs:104、:107。)
握手时,relay 给新客户端分一个 slot,并把这个 slot 的专属 id 区间告诉它:
slot 的 id 区间 = [ slot*STEP + 1 , slot*STEP + STEP )
slot 0 : [ 1 , 33_554_431 )
slot 1 : [ 33_554_432 , 67_108_862 )
slot 2 : [ 67_108_863 , ... )
...
去程:客户端只能用自己区间内的 id当 correlation id。回程:客机把响应帧的 id 沿用请求的 id 原样发回,relay 只需一个除法就知道该退给谁:
client_slot = frame.id / AGENT_RELAY_ID_RANGE_STEP
妙就妙在:relay 全程不改帧头、不解 CBOR。 区间是不重叠的,所以"id 属于哪个客户端"是一道
纯算术题。分配区间在 relay.rs:479(id_offset = slot * AGENT_RELAY_ID_RANGE_STEP),回程路由在
relay.rs:826(frame.id / AGENT_RELAY_ID_RANGE_STEP)。
5.2 握手:socket 连进来,拿到区间 + ready 帧
SDK 客户端经 Unix socket(Windows 上是 named pipe)连入。relay 的 run 循环
(relay.rs:398)accept 到连接后:
- 从 128 个槽里找一个空槽(
relay.rs:457起)。满了就拒。 - 算出该槽的
id_start/id_end_exclusive。 - 一次性写回握手包:
[id_start: u32 BE][id_end_exclusive: u32 BE][缓存的 ready 帧字节...](relay.rs:490)。ready 帧是启动时从客机收到并缓存的core.ready(见 §5.4)。
客户端拿到这 8 字节就知道了自己的地盘。SDK 侧的解析见 §7.2。
5.3 两个方向,各一条后台任务
relay 内部把两个方向拆成两条 tokio 任务,加每个客户端各一条读/写任务:
去程(client → guest):
client_reader_task ──帧──▶ agent_tx (mpsc, 有界) ──▶ ring_writer_task ──▶ rx_ring ──▶ 客机
回程(guest → client):
tx_ring ──▶ ring_reader_task ──按 id 算 slot──▶ 该 client 的 write_tx ──▶ 客户端 socket
ring_writer_task(relay.rs:728):把各客户端汇来的帧塞进rx_ring,满了带退避重试, 不丢帧。ring_reader_task(relay.rs:763):从tx_ring拼出完整帧(try_extract_frame,:619), 算出 slot,推给对应客户端的写通道。这里刻意先把帧全提取出来再路由,避免持锁跨 async 写。client_reader_task(relay.rs:928):读某客户端的帧,校验、登记会话,转发去agent_tx。
入口校验很重要: 客户端只能用自己区间内的 id,否则断连。is_client_frame_allowed
(relay.rs:1106)放行两种:id 落在 [id_start, id_end);或 core.shutdown 这种进程级控制帧
(id=0 + FLAG_SHUTDOWN)——后者不是相关请求,SDK 用 id 0 发。
断连清理: 客户端掉线时,client_reader_task 收尾(relay.rs:1023 起):给它每个还活着的
会话发 ExecSignal{signal:9}(SIGKILL)杀掉客机侧进程,再发一条 RelayClientDisconnected
(带该客户端的 id 区间)让客机清理该区间内的 fs/tcp 会话与文件句柄,最后释放 slot。
5.4 wait_ready:先等客机"我起来了"
run 之前必须先 wait_ready(relay.rs:309)。它阻塞地从 tx_ring 读帧,直到收到
core.ready,把这帧缓存下来给后续握手用。期间还处理早期 init 握手:收到 InitResolved
就装好 bind identity map(用户卷的属主映射),回一个 InitAck;若配了 map 却在
InitResolved 之前先收到 ready,判为错误——保证顺序。
拿到 ready 后,若挂了 exec.log 写手,会写一行 --- sandbox started --- 标记。刻意放在
确认客机真起来之后才写:启动前就失败(挂载错误等)会让 exec.log 保持 空,让
boot-error.json 独自讲故事。
5.5 SessionInfo 与 exec 会话捕获(exec.log)
relay 顺路做一件事:把主机侧观察到的 exec 会话 stdout/stderr 抄一份进 exec.log(JSON Lines)。
难点:协议 correlation id 会在 slot 回收后被复用(每个 msb exec 是独立客户端,slot 0 释放后
重分,同一个 id 可能一辈子出现两次)。所以 relay 另铸一个单调递增的 session_id 供人看。
SessionInfo(relay.rs:49)就记这个单调 id 加一个 is_pty 标志。
流程:
client_reader_task看到ExecRequest(FLAG_SESSION_START且解出类型是 exec)时,从next_session_id领一个单调 id,连同 pty 标志插进session_registry(relay.rs:983起)。ring_reader_task每转发一个客机帧,顺手tap_frame_into_log(relay.rs:666):按 id 查SessionInfo,把ExecStdout/ExecStderr的载荷写进日志;pty 模式合并成Output标签。- 收到 terminal 帧(
ExecExited/ExecFailed)就从 registry 删掉,避免泄漏。
日志写手 LogWriter(crates/runtime/lib/exec_log.rs:64)每块输出写一行 JSON
({"t":..,"s":"stdout","d":..,"id":..},见 ExecLogEntry:43),底层是带轮转的
RotatingLog(10 MiB × 3)。捕获是尽力而为:任何解码失败都记 debug 后丢弃,
绝不能干扰路由主路径。
6. 客机 agent loop:一根线程搞定一切
翻到管子的客机那头。agentd 的主循环(crates/agentd/lib/agent.rs:149 run)是单线程 tokio
select,一手管串口 I/O,一手管所有会话。
6.1 select 两条腿
'agent: loop {
select! {
// 腿 A:串口可读 → 读字节 → 拼帧 → 校验 flags → handle_message → 派发
async_port.readable() => { ... 增量解帧,逐条 handle_message ... }
// 腿 B:某个会话有输出 → 编成帧 → 写回串口
session_rx.recv() => { Stdout/Stderr/Exited/Raw → encode → flush }
}
}
(结构见 agent.rs:192 起的 'agent: loop。)
腿 A 读帧后先查 msg.flags == msg.t.flags(),不符回 core.error(agent.rs:239);再看要不要刷新
空闲计时器(ClockSync 这类维护流量不算用户活动,message_refreshes_idle_timer:666);然后
handle_message(agent.rs:397)按类型派发。
handle_message 是一张大 match:ExecRequest→ExecSession::spawn(成功回 ExecStarted,失败回
ExecFailed);ExecStdin/ExecResize/ExecSignal→找到会话操作;FsRequest/FsData→fs 模块;
TcpConnect/TcpData/TcpEof/TcpClose→tcp 模块;Shutdown→给所有会话发 SIGTERM、清空、请求
客机 poweroff(request_guest_poweroff,:1101)。
6.2 会话管理:三张表
AgentState(agent.rs:65)用三张 HashMap<u32, _> 按 correlation id 存活跃会话:
sessions(exec)、write_sessions/read_sessions(fs 流)、tcp_sessions 。
ExecSession(crates/agentd/lib/session.rs:122)是一个 exec 会话的句柄:PID、PTY master fd
(tty 模式)、child stdin(pipe 模式)。spawn(:272)按 req.tty 走 PTY 或管道两条路,并起一条
后台任务读输出,通过 mpsc 把 SessionOutput(:133,Stdout/Stderr/Exited/Raw)送回主循环——
主循环那条腿 B 就是在收这些。
默认用户解析:resolve_default_user(session.rs:838)把 "name" 或 "name:group" 或数字
uid/gid 解析成 (uid, gid);空则回 (0,0)(root)。这个值也是早期 InitResolved 握手报给主机的
(report_init_context,agent.rs:370)。
6.3 心跳:一根独立 OS 线程
心跳故意不放在 tokio 里,而是一根专门的 OS 线程(spawn_heartbeat_thread,agent.rs:679)。
原因很实在:单线程 async 运行时被 exec 输出洪水占满时,一个心跳任务会被饿死,脉搏停跳,主机
误判沙箱死了就杀掉它——而这沙箱其实活得好好的。一根内核调度的普通线程不受 async 运行时拖累,
无论多忙都按 HEARTBEAT_INTERVAL_SECS(1 秒)写一次 /.msb/heartbeat.json。
心跳内容靠 ActivityTracker(agent.rs:73)累计:activity_seq(每次有意义活动 +1)加
ActivityCounters(host/guest 消息数、exec 输出字节、fs/tcp 字节)。主循环每处理完一批就
publish_heartbeat_snapshot 把最新快照通过 watch 通道推给心跳 线程(无锁 borrow)。
7. SDK 侧客户端桥
管子的最外层是 SDK。两层封装:AgentClient(Rust 惯用)和 AgentBridge(FFI 形状,给
Node/Python/Go 绑定)。
7.1 两层:AgentClient 与 AgentBridge
AgentBridge(sdk/rust/lib/agent/bridge.rs:50)是"字节进、字节出"的薄壳:不用泛型、不用消费
self、不跨 FFI 回调;每个方法 &self。它把 correlation id 标识的流包成不透明的 u64
StreamHandle,外语绑定不用持有 tokio Receiver。CBOR 编解码留给调用方语言;桥只搬字节。
三个核心方法:
request(:115):一发一收,用于FsRequest→FsResponse这类恰好一个终结响应的 RPC。stream_open(:153):开流式会话,返回(correlation_id, StreamHandle)。stream_next(:188):从流里拉下一帧;收到FLAG_TERMINAL帧后自动摘掉该流;流结束返回None。send(:136):在已有 correlation id 上发后续帧(ExecStdin/FsData…)。
BridgeFrame(:36)就是 {id, flags, body} 的 FFI 友好视图。
7.2 握手与 id 领用
AgentClient::connect(sdk/rust/lib/agent/mod.rs:80 起,底层
packages/agent-client/rust/lib/client.rs)连上 socket 后跑 perform_handshake
(client.rs:568):读 8 字节 → 前 4 是 id_min、后 4 是 id_max(还兼容 pre-0.5 旧握手格式),
再读缓存的 ready 帧。于是客户端知道了自己的 id 区间。
之后每次发请求,reserve_id(client.rs:478)从 [id_min, id_max) 里领一个没在用的 id:原子
next_id 递增、到顶回绕、跳过 0 和已占用的。领到就把响应通道 tx 登进 pending 表。这样
每个在飞的请求/流各占一个 id,回来的帧按 id 找回自己的通道。
7.3 回程分发
客户端起一条 reader_loop(client.rs:773)不停读帧,交给 dispatch_frame(:795):按 frame.id
在 pending 表里找对应通道推进去;若帧带 FLAG_TERMINAL 就顺手把该 id 从表里删掉(会话结束)。
注意主机 relay 与 SDK 用的是同一招、两层生效:relay 按 id / STEP 把帧路由到正确的客户端
进程;客户端内部再按 id 把帧路由到正确的在飞请求。同一个 32 位 id,一次分区间定客户端,
一次查表定请求。
8. 巧妙之处(值得带走的技术)
-
帧头把路由字段拎出 CBOR。
id/flags复制在定长二进制头里(message.rs:47的#[serde(skip)]+ codec 单独读写),中间人零解码就能路由。这是整条链"relay 不拆包"的地基。 -
一根管子多路复用,靠 id 分区间而非改帧头。
slot*STEP分配、id/STEP回收 (relay.rs:479/:826),不重叠所以是纯算术。128 个客户端各自一段,共用一根 virtio-console。 -
数据路径零系统调用。 console 后端用无锁
ArrayQueue+memcpy,只在"通知有活"时写一次 1 字节 wake pipe(console.rs:56/:222)。 -
心跳跑在 OS 线程,不在 async 里。 躲开 exec 输出洪水饿死心跳任务导致沙箱被误杀的坑 (
agent.rs:679的整段注释就是这个教训)。 -
ExecExitedvsExecFailed语义分离。 "跑了给退出码" 和 "压根没起来" 是两类事件,后者带 结构化ExecFailureKind(exec.rs:150)供上层挑提示语——错误可诊断,而非一个笼统的非零码。 -
单调 session_id 抵御 id 复用。 correlation id 会随 slot 回收复用,relay 另铸单调 id (
SessionInfo,relay.rs:49)让 exec.log 里每个会话唯一可辨。 -
版本闸门在发送前。 高版本特性发不出去时只失败那一个操作,不拖垮会话 (
is_available_at,message.rs:325)。
9. 边界与局限(诚实)
-
单帧硬上限 4 MiB。
MAX_FRAME_SIZE(codec.rs:23)。大文件必须走FsData分块流 (每块 ≤ 3 MiB,fs.rs:12),不能一帧塞下。 -
最多 128 个并发 SDK 客户端。
AGENT_RELAY_MAX_CLIENTS(lib.rs:104)。槽满了 relay 直接 拒绝新连接(relay.rs:473)。这是固定常量,不是可调旋钮。 -
串口是纯字节流,靠长度前缀自定界。 一旦帧长字段被解成超大值(数据损坏),relay 会 清空整个缓冲而非只吞 4 字节——因为把身体字节当成新长度会级联出错(
relay.rs:630的注释)。 -
relay 是单进程内的中间人,不是网络代理。 客户端经本地 Unix socket/named pipe 连入;它不做 跨主机转发。
-
心跳文件在客机内,主机据文件判活。 若客机文件系统卡死,心跳也停;主机把"脉搏停"当"沙箱死", 这是设计取舍(宁可误杀也不放任僵死)。
-
agentd 主循环单线程。 一条 CPU 密集的
handle_message会占住解帧;真正的重活(读进程输出、 网络)都甩给后台任务经 mpsc 回流,主循环只做编解码与派发。
横向看:同货架其它 agent 运行时多用 gRPC/HTTP + 独立进程通道;Microsandbox 选了一根 virtio 串口 + 自定义二进制帧,把"多路复用"下放到一个 32 位 id 的区间算术里——换来的是零额外 内核 通道、数据路径零系统调用,代价是自己扛下分帧、路由、版本协商这些本该由成熟 RPC 框架代劳的事。
10. 代码地图(导航索引)
| 主题 | 文件 | 关键符号 |
|---|---|---|
| 帧编解码 | crates/protocol/lib/codec.rs | MAX_FRAME_SIZE、RawFrame、try_decode_raw_from_buf、read_raw_frame、decode_message_frame、encode_raw_to_buf |
| 消息信封与类型 | crates/protocol/lib/message.rs | Message、MessageType、FRAME_HEADER_SIZE、FLAG_TERMINAL、FLAG_SESSION_START、FLAG_SHUTDOWN、flags()、min_protocol_version()、is_available_at() |
| exec 消息 | crates/protocol/lib/exec.rs | ExecRequest、ExecStarted、ExecStdout、ExecExited、ExecFailed、ExecFailureKind、ExecSignal |
| fs 消息 | crates/protocol/lib/fs.rs | FsRequest、FsOp、FsResponse、FsData、FS_CHUNK_SIZE |
| tcp 消息 | crates/protocol/lib/tcp.rs | TcpConnect、TcpData、TcpEof、TcpClosed、TcpFailed |
| core 消息 | crates/protocol/lib/core.rs | Ready、InitResolved、InitAck、ClockSync、CoreError、RelayClientDisconnected |
| 心跳 | crates/protocol/lib/heartbeat.rs | Heartbeat、ActivityCounters |
| 协议常量 | crates/protocol/lib/lib.rs | AGENT_PORT_NAME、AGENT_RELAY_MAX_CLIENTS、AGENT_RELAY_ID_RANGE_STEP、RUNTIME_MOUNT_POINT |
| 客机串口发现 | crates/agentd/lib/serial.rs | find_serial_port、AGENT_PORT_NAME |
| 主机 console 后端 | crates/runtime/lib/console.rs | ConsoleSharedState(tx_ring/rx_ring/tx_wake/rx_wake)、AgentConsoleBackend、read、write |
| 主机 relay | crates/runtime/lib/relay.rs | AgentRelay、wait_ready、run、ring_reader_task、ring_writer_task、client_reader_task、try_extract_frame、is_client_frame_allowed、SessionInfo、tap_frame_into_log |
| exec.log 捕获 | crates/runtime/lib/exec_log.rs | LogWriter、write_chunk、ExecLogEntry、EXEC_LOG_FILENAME |
| 客机 agent 循环 | crates/agentd/lib/agent.rs | run、handle_message、open_serial_port、report_init_context、AgentState、ActivityTracker、spawn_heartbeat_thread |
| 客机会话 | crates/agentd/lib/session.rs | ExecSession、spawn、SessionOutput、resolve_default_user |
| 客机 fs/tcp 会话 | crates/agentd/lib/fs.rs、tcp.rs | FsReadSession、FsWriteSession、FsStreamSession、TcpSession、owner_id |
| SDK 客户端桥 | sdk/rust/lib/agent/bridge.rs、mod.rs | AgentBridge、BridgeFrame、StreamHandle、request、stream_open、stream_next、AgentClient |
| SDK 底层传输 | packages/agent-client/rust/lib/client.rs | perform_handshake、reserve_id、reader_loop、dispatch_frame、ensure_version_compat_for、negotiated_version |