1 · SansIO#

https://sans-io.readthedocs.io/ https://www.firezone.dev/blog/sans-io

S 实现网络协议/流式解析器/消息编解码器时，同一套逻辑要跑在 sync socket、asyncio、线程池、测试桩等不同 I/O 环境上 C 协议逻辑和 I/O 写死在一起 → 难测试（要起真实连接）、难复用（sync→async≈重写）、难定位 bug（语义+时序+环境混在一起） Q 能否把”协议/业务本身”和”怎么收发数据”彻底剥离，让核心逻辑可测试、可复用、可适配不同运行时？ A SansIO：把核心写成不碰 I/O 的纯状态机，I/O 由外层 event loop 驱动

---

SansIO = Sans I/O，字面意思是”没有 I/O 的协议实现”。

核心思想：把一个系统拆成两层：

1. 纯状态机 / 纯协议层：只消费字节、事件、命令，产出状态变化或待发送的数据。不拥有 socket、不阻塞、不等待、不调度。
2. I/O 适配层（Event Loop）：负责 socket、文件、线程、async、timer、TLS 等外部交互，驱动状态机运转。

这样核心逻辑不依赖具体运行时，只依赖”输入是什么、输出是什么”。

1.1 · 与相关模式的关系

SansIO 不是孤立的发明，它是几种更广泛的设计原则在网络协议领域的具体体现：

模式	核心思想	与 SansIO 的关系
Clean Architecture (Bob Martin)	业务规则在中心，I/O 在外圈	SansIO 的协议层 = Use Case 层
Functional Core, Imperative Shell (Gary Bernhardt)	纯函数做决策，副作用在边界	SansIO core = functional core
依赖反转原则 (DIP)	策略不依赖实现细节，二者通过抽象通信	协议层通过 Transmit/Event 抽象与 I/O 层解耦
Model-View-Controller	模型不关心展示和输入细节	协议状态机 = Model

1.2 · 典型结构

socket / asyncio / test harness
        |
        v
  Event Loop (I/O adapter)
        |
        v
  SansIO Core (pure state machine)
    - parser
    - state machine
    - protocol rules
        |
        v
  events / outgoing bytes / commands

1.3 · 状态机的 API 模式#

SansIO 核心对外暴露的 API 通常可以归纳为四类：

输入:
  handle_input(data)      -- 收到网络数据
  handle_timeout(now)     -- 时间推进到 now

输出（轮询式）:
  poll_transmit()         -- 取出待发送的数据
  poll_timeout()          -- 询问下次超时时间

这组 API 不绑定任何特定协议——STUN、QUIC、WireGuard、ICE 都可以用同样的签名来实现。这正是 SansIO 组件可以任意组合的基础。

1.3.1 · 字节流 vs 数据报#

• TCP / 字节流协议：使用单一输入缓冲区 + 单一输出缓冲区，receive_bytes(data) / data_to_send() 即可。
• UDP / 数据报协议：需要保留报文边界，输入输出改为 [Transmit(dst, bytes)] 的列表。

1.3.2 · Event 抽象#

协议本质上是事件的序列化机制。SansIO 把收到的字节翻译成语义化事件：

core.receive_data(raw_bytes)
events = core.pop_events()
# events → [RequestReceived(headers), DataReceived(body), ...]

发送侧有两种风格：

1. 命名方法：core.send_headers(...) → 内部缓冲待发字节（hyper-h2 风格）
2. 对称事件：core.send(Event) → 输入和输出使用同一套事件类型（h11 风格）

1.4 · 抽象时间

SansIO 状态机不调用 time.now() / Instant::now()，而是把时间作为参数注入：

fn handle_timeout(&mut self, now: Instant) { ... }
fn poll_timeout(&self) -> Option<Instant> { ... }

好处：

• 测试中可以瞬间”快进”5 分钟，验证超时、重传、keep-alive 等行为
• 没有 wall-clock 依赖，测试 100% 确定性
• Event loop 只需在 poll_timeout() 返回的时间点唤醒状态机即可

1.5 · Event Loop 驱动#

状态机本身是惰性的（就像 Rust 的 Future 需要 runtime poll 才能推进），需要 event loop 来驱动：

loop {
    // 1. 处理 I/O 输入
    data = socket.recv()
    core.handle_input(data)

    // 2. 处理超时
    if now >= core.poll_timeout():
        core.handle_timeout(now)

    // 3. 发送输出
    while transmit = core.poll_transmit():
        socket.send(transmit.dst, transmit.data)
}

Event loop 不理解协议细节（不知道是请求-响应还是多轮握手），它只做：

• 从 socket 读 → 喂给状态机
• 从状态机取 → 写到 socket
• 管理定时器

这意味着你可以自由选择：sendmmsg 减少系统调用、多协议复用单 socket、自定义 backpressure 等。

1.6 · 组合性

因为所有 SansIO 组件暴露相同的 handle_input / handle_timeout / poll_transmit / poll_timeout 接口，组合变得极其简单：

• 想同时查询 5 个 STUN 服务器？创建 5 个 StunBinding 实例，依次调用即可
• 想把 ICE + WireGuard 组合成”魔法隧道”？把两个状态机串联（Firezone 的 snownet 就是这样做的）
• 想只用 WebRTC 库的 ICE 部分？直接导入 IceAgent，不需要整个 WebRTC 栈

1.7 · 为什么这种拆分有效

I/O 和协议变化速度不同：

• 协议层关注语义正确性、状态迁移、边界条件
• I/O 层关注运行时模型、阻塞/非阻塞、并发、资源管理

把两者揉在一起，每个 bug 都变成”语义 + 时序 + 环境”混合问题；拆开后，协议 bug 可以像纯逻辑代码一样被单测覆盖。

1.8 · 优点

1. 测试极快且确定：直接喂输入、断言输出，不建连接、不起 event loop、不 mock socket。测试永远不会 flaky
2. 跨运行时复用：同一核心可配 sync、async、线程、mock、任何框架
3. 边界清晰：协议正确性和 I/O 正确性分别验证
4. 容易做 fuzz / property test / state machine test：核心是纯输入→输出模型
5. 容易移植：替换 transport 层不需要重写协议状态机
6. 组合性好：多个 SansIO 组件可以自由嵌套组合
7. 与 Rust 所有权模型天然契合：状态机用 &mut self 表达状态变更，避免 Arc<Mutex<T>> 和 channel 的复杂性

1.9 · 代价

1. 前期设计成本更高：需要先设计事件、命令、状态边界
2. 代码看起来更绕：比”直接 recv 后当场处理”多了一层 event loop
3. Event loop 的 bug 难调：比如 poll_timeout 返回值不推进 → event loop 忙循环
4. 生态还不够普及：（尤其在 Rust 中）多数库仍然直接做 I/O

1.10 · 适合的场景

• 协议复杂，状态多
• 需要同时支持 sync/async
• 对可测试性要求高
• 希望把 parser / codec / state machine 做成库
• 需要长期维护，预计 transport 会变化

1.11 · 不适合的场景

• 一次性脚本
• 非常薄的 I/O 封装
• 协议和运行时强绑定，抽离后收益很小

1.12 · 实践中的 SansIO 库#

Python（SansIO 概念的发源地）：

• h11 — HTTP/1.1，对称事件模型
• hyper-h2 — HTTP/2
• wsproto — WebSocket
• gidgethub — GitHub API

Rust（Rust 的所有权模型让 SansIO 尤其自然）：

• quinn-proto — QUIC
• quiche — Cloudflare 的 QUIC
• str0m — WebRTC
• connlib/snownet — Firezone 的 ICE + WireGuard

1.13 · 判断标准

如果核心逻辑能这样测试，它就是 SansIO：

core = ProtocolCore()
core.receive_data(input_bytes)
assert core.pop_events() == [...]
assert core.data_to_send() == [...]

如果必须创建 socket、启动 event loop、等待网络事件才能验证逻辑，那它还不是 SansIO。

1.14 · 一句话

SansIO 不是不要 I/O，而是让 I/O 成为外围，把协议和状态机沉到中心。