openpilot cereal 消息系统：Cap'n Proto 消息规范、msgq 发布/订阅与 fork 兼容性设计

cereal 是 openpilot 的消息系统，它基于 msgq 作为发布/订阅（pub/sub）后端，使用 Cap'n Proto 完成结构体序列化，是 openpilot 各守护进程（controlsd、modeld、locationd、pandad 等）之间通信以及日志录制、回放（process replay）的底层基石。本文围绕 openpilot/cereal/README.md 展开，结合 log.capnp、custom.capnp、services.py 与 messaging 模块 的实际源码，讲清消息规范的定义方式、发布/订阅的实战用法、向后兼容性的维护规则，以及 fork 作者如何安全地扩展自定义事件。读完本文，你将能够：独立读懂 Event 消息结构、编写符合规范的发布/订阅代码、理解服务注册表（queue size、频率、抽稀）的机制，并为自己的 fork 正确添加自定义消息而不破坏日志兼容性。

一、cereal 的架构：msgq + Cap'n Proto

根据 README 的定义：

cereal is the messaging system for openpilot. It uses msgq as a pub/sub backend, and Cap'n Proto for serialization of the structs.

也就是说 cereal 由两层组成：

1. 传输层 —— msgq：一个独立维护的消息队列实现，负责进程间的发布/订阅传输（Unix socket 之上的共享内存风格队列）。cereal 的 Python 侧入口 messaging/init.py 第 1～5 行直接依赖它，导入 Context、Poller、SubSocket、PubSocket、drain_sock_raw 等底层原语；
2. 序列化层 —— Cap'n Proto：所有消息结构体用 Cap'n Proto 的 .capnp schema 定义，编译后在 C++ 与 Python 中分别有对应绑定（仓库通过 scons 构建，文件头部注释 "must be built with scons" 即指这一点）。

这种分层带来一个关键好处：消息在进程间传递的是原始字节，收发两端只需要 schema 就能解析，不需要运行时携带完整 schema——这正是日志回放（replay）能够跨版本工作的基础之一。

二、消息规范：log.capnp 中的 Event 结构体

README 指出，所有消息类型都定义在 log.capnp 中，核心是一个名为 Event 的结构体（log.capnp#L2523-L2527）：

struct Event {
  logMonoTime @0 :UInt64;  # nanoseconds
  valid @67 :Bool = true;

  union {
    # *********** log metadata ***********
    initData @1 :InitData;
    sentinel @73 :Sentinel;
    ...

每个 Event 都有两个公共字段，随后由一个大型 union 定义具体的包类型：

• logMonoTime：单调时钟时间戳，单位纳秒。Python 侧 new_message() 在 messaging/init.py#L73-L85 中自动填充：'logMonoTime': int(time.monotonic() * 1e9)，因此发布者通常无需手工设置；
• valid：标记本条消息是否有效。new_message() 默认将其置为 False，发布者需显式置真后该消息才被认为有效——测试 test_messaging.py#L52-L59 就断言了 assert not msg.valid（新建消息默认无效）；
• union：每个 union 分支对应一个"服务"（service），如 carState、modelV2、can、controlsState、deviceState 等。从源码结构看，union 字段名与 services.py 注册表中的服务名一一对应，测试中用 log.Event.schema.union_fields 与 SERVICE_LIST.keys() 取交集来枚举全部可发布事件（test_messaging.py#L15），这实际上构成了 schema 与注册表的一致性校验。

消息设计最佳实践（README 原文规则）

README 的 "Best Practices" 一节给出了三条硬性设计规范，新增或修改任何消息字段时必须遵守：

• 所有字段必须使用 SI 单位，除非字段名中另有说明；
• 字段名在所在消息的上下文中必须完全无歧义；
• 所有取值都应便于绘图、便于以最小解析成本被人读取。

这三条规则解释了 openpilot 日志为何可以直接被 plotjuggler 等工具可视化：量纲统一、命名自解释、结构扁平化。

服务注册表：频率、抽稀与队列大小

除了 schema，services.py 为每个服务登记了运行期元数据，这是理解 cereal 消息流控的关键（services.py#L6-L24）：

class QueueSize(IntEnum):
  BIG = 10 * 1024 * 1024      # 10MB - video frames, large AI outputs
  MEDIUM = 2 * 1024 * 1024    # 2MB - high freq (CAN), livestream
  SMALL = 250 * 1024          # 250KB - most services

每个服务是一个 Service(should_log, frequency, decimation, queue_size) 四元组，例如：

"gyroscope": (True, 104., 104),
"can": (True, 100., 2053, QueueSize.BIG),  # decimation gives ~3 msgs in a full segment
"controlsState": (True, 100., 10, QueueSize.MEDIUM),
"modelV2": (True, 20., None, QueueSize.BIG),

• should_log：该服务是否被写入 qlog；
• frequency：预期发布频率（Hz），如 IMU 服务为 104 Hz；
• decimation：录制 qlog 时的抽稀倍数，高频服务（如 can 抽稀 2053 倍）大幅压缩日志体积；
• queue_size：msgq 队列段大小。大消息（视频帧、模型输出）用 BIG（10MB），高频 CAN 与直播流用 MEDIUM（2MB），其余默认 SMALL（250KB）。

messaging 层创建套接字时会据此自动选择队列大小（messaging/init.py#L50-L61 中 pub_sock/sub_sock 都从 SERVICE_LIST 查表取 queue_size）。

注册表还承担跨语言生成职责：build_header()（services.py#L100-L118）把整张表渲染成 C++ 头文件，供 C++ 侧进程复用同一份服务定义。测试 test_services.py#L14-L24 用参数化方式遍历全部服务断言 frequency <= 104 且 decimation != 0，并实际调用 clang++ 编译生成的头文件以验证其合法性——这保证了服务参数不会被随意破坏。

三、发布与订阅：README 中的完整示例

README 给出了最小可运行的发布/订阅对，完整继承如下（注意 sensorEvents 在当前版本 schema 中已更名为 gyroscope/accelerometer 等，新代码请以 services.py 中实际注册的服务名为准）：

订阅端：

import openpilot.cereal.messaging as messaging

# in subscriber
sm = messaging.SubMaster(['sensorEvents'])
while 1:
  sm.update()
  print(sm['sensorEvents'])

发布端：

# in publisher
pm = messaging.PubMaster(['sensorEvents'])
dat = messaging.new_message('sensorEvents', size=1)
dat.sensorEvents[0] = {"gyro": {"v": [0.1, -0.1, 0.1]}}
pm.send('sensorEvents', dat)

其中 size 参数用于 union 字段是列表（List(...)）类型时初始化列表长度；对列表类型的服务，new_message 需要显式传入 size（对照 test_messaging.py#L52-L59，对 KjException 即列表类型重试并传入 size）。

SubMaster：不只是收消息，还做健康检查

SubMaster（messaging/init.py#L181-L278）是 openpilot 各守护进程订阅的标准封装，其内部机制比 README 示例丰富得多：

• update() 基于 Poller 做非阻塞轮询，对 poll 指定的服务阻塞等待、对其余服务非阻塞收取（sub_sock(..., conflate=True) 只保留最新一帧）；
• alive 判定：静态频率服务要求"距上次接收时间 < 10 倍预期周期"，否则判为不存活（L263-L265）；频率 ≤ 1e-5 Hz 的按需服务（如 userBookmark）则永远视为存活；
• freq_ok 判定：由 FrequencyTracker 用滑动平均比较实际频率与 SERVICE_LIST 中的预期频率（L138-L178）；
• 对外暴露 all_alive() / all_freq_ok() / all_valid() / all_checks() 组合判定。valid 直接取自每条消息的 valid 字段（L261），因此发布方不置 valid=True 会被订阅方健康检查拒绝——这是 Event.valid 字段的实际意义所在。

PubMaster 则封装 pub_sock，send() 会把 Cap'n Proto builder 序列化为字节（to_bytes()）后入队；它还提供 wait_for_readers_to_update() / all_readers_updated()（L292-L300），用于需要确认所有订阅者已消费的场景。

更底层的原语（drain_sock、recv_one、recv_one_retry 等）也在测试 test_messaging.py 中被逐一覆盖，覆盖了空队列超时、乱序到达、延迟发送等边界情形，可作为自定义收发逻辑的参考实现。

四、维护向后兼容性

README 的 "Maintaining backwards-compatibility" 一节确立了修改消息规范的第一原则：新版本的 cereal 必须能解析旧日志。具体规则：

• 安全的改动：新增结构体、向已有结构体新增成员字段；
• 不安全的改动：其余大多数修改（如改动/删除字段、改动字段类型、移动字段 ID）都会破坏旧日志的解析。

这与 Cap'n Proto 的演进模型一致：序列化以字段 ID（@n）寻址，新增字段对旧二进制透明，而改动或复用已占用的 ID 则会让旧数据被错误解释。log.capnp 中的实际做法印证了这一点：Event union 的字段 ID 只增不复用（从 @1 一路排到 @152），例如新加入的 driverMonitoringState @151、lateralDelay @146 都使用新的编号，旧服务（narrowRoadCameraState @2 等）的 ID 从未变动。

另外值得注意 log.capnp#L5-L6：Event 通过 using Car = import "/car.capnp" 与 using Custom = import "custom.capnp" 引入外部 schema 的结构体（如 carState @22 :Car.CarState），车辆相关结构与 fork 预留结构都以此方式挂载进同一个 union。

五、Custom forks：custom.capnp 中的预留事件

这是 README 中信息量最大的部分。fork 作者往往也想给自己的定制逻辑加消息，但直接在 log.capnp 里插入自定义结构体会与主线后续改动冲突——rebase 回主线 openpilot 后，fork 旧日志将全部无法解析（字段 ID 被主线占用或含义改变）。

openpilot 的解决方案是：在 custom.capnp 中预留一批空结构体，主线保证它们永远为空、永远存在：

# custom.capnp: a home for empty structs reserved for custom forks
# These structs are guaranteed to remain reserved and empty in mainline
# cereal, so use these if you want custom events in your fork.

# DO rename the structs
# DON'T change the identifier (e.g. @0x81c2f05a394cf4af)

struct CustomReserved0 @0x81c2f05a394cf4af {
}

当前主线保留了 CustomReserved0 到 CustomReserved19 共 20 个空结构体（custom.capnp#L13-L71）。在 Event union 中同样预留了对应位置：3 个裸数据槽 customReservedRawData0/1/2 :Data（@124–@126，可塞任意字节）和 20 个结构化槽 customReserved0..19（@107–@119，指向 Custom.CustomReservedN），见 log.capnp#L2624-L2654。

README 给出了一份完整的兼容改动 diff 示例，规则可概括为四条：

1. 可以重命名结构体（CustomReserved0 → SteeringInfo）；
2. 绝不能改结构体标识符（@0x81c2f05a394cf4af）；
3. 可以重命名 union 字段（customReservedRawData0 → rawCanData），但 **@ 号之后的一切（ID、类型）**不能动；
4. 不能改 union 字段的 ID（如 @107），也不能改它指向的结构体。

完整 diff 示例：

diff --git a/openpilot/cereal/custom.capnp b/openpilot/cereal/custom.capnp
index 3348e859e..3365c7b98 100644
--- a/openpilot/cereal/custom.capnp
+++ b/openpilot/cereal/custom.capnp
@@ -10,7 +10,11 @@ $Cxx.namespace("cereal");
 # DO rename the structs
 # DON'T change the identifier (e.g. @0x81c2f05a394cf4af)

-struct CustomReserved0 @0x81c2f05a394cf4af {
+struct SteeringInfo @0x81c2f05a394cf4af {
+  active @0 :Bool;
+  steeringAngleDeg @1 :Float32;
+  steeringRateDeg @2 :Float32;
+  steeringAccelDeg @3 :Float32;
 }

 struct CustomReserved1 @0xaedffd8f31e7b55d {
diff --git a/openpilot/cereal/log.capnp b/openpilot/cereal/log.capnp
index 1209f3fd9..b189f58b6 100644
--- a/openpilot/cereal/log.capnp
+++ b/openpilot/cereal/log.capnp
@@ -2558,14 +2558,14 @@ struct Event {

     # DO change the name of the field
     # DON'T change anything after the "@"
-    customReservedRawData0 @124 :Data;
+    rawCanData @124 :Data;
     customReservedRawData1 @125 :Data;
     customReservedRawData2 @126 :Data;

     # DO change the name of the field and struct
     # DON'T change the ID (e.g. @107)
     # DON'T change which struct it points to
-    customReserved0 @107 :Custom.CustomReserved0;
+    steeringInfo @107 :Custom.SteeringInfo;
     customReserved1 @108 :Custom.CustomReserved1;
     customReserved2 @109 :Custom.CustomReserved2;
     customReserved3 @110 :Custom.CustomReserved3;

README 对此给出的保证是：如果 fork 只修改这些预留槽位，就能同时保持与所有主线 openpilot 版本以及 fork 自身全部旧日志的向后兼容。原因从 Cap'n Proto 序列化角度可以推断：结构体 ID 与 union 分支 ID 均不变，主线永远不会向这些槽位写入任何数据（它们在主线上恒为空），因此主线日志解析 fork 结构体时读到的永远是默认值，反之亦然，互不污染。

配套的注册流程也已在源码中留好口子：services.py#L94 中 customReservedRawData0 已经注册为 (True, 0.)（零频率、按需发布、写日志），fork 者把字段重命名后同步更新注册表即可。

六、验证体系：如何确认你的 cereal 改动是安全的

仓库自带两层自动化校验，可用于验证消息规范的完整性：

1. openpilot/cereal/messaging/tests/test_services.py：参数化遍历 SERVICE_LIST 全部服务，断言频率不超过 104 Hz、抽稀倍数非零；另用 clang++ -std=c++11 实际编译 build_header() 生成的 C++ 头文件，确保 Python 注册表与 C++ 侧生成的定义一致且合法；
2. openpilot/cereal/messaging/tests/test_messaging.py：以 log.Event.schema.union_fields ∩ SERVICE_LIST 动态枚举全部事件，对每个事件分别验证 new_message 的时间戳新鲜度与 valid=False 默认值、union 分支名（msg.which()）与服务名匹配，并对 drain_sock / recv_one / recv_one_retry 等收发原语做端到端测试。

此外 messaging/tests/test_pub_sub_master.py 覆盖 PubMaster/SubMaster 的高层语义。任何对 schema 的改动（尤其是新增字段、占用预留槽位）都应在这组测试通过后提交。

七、小结

cereal 的设计可以归纳为四点：

• 单一 Event 信封：所有服务共用 Event { logMonoTime, valid, union } 结构，union 分支名即服务名，使发布/订阅、日志录制、回放三者共用同一套 schema（log.capnp#L2523）；
• SI 单位 + 无歧义命名 + 可绘图三条字段规范，保证日志数据的人机双可读性；
• 服务注册表集中管理频率、抽稀、队列大小，并通过代码生成同步到 C++，测试保证约束不被破坏（services.py）；
• 预留槽位机制（custom.capnp + union 中 @107–@119、@124–@126）让 fork 可以在不改主线 schema 演进方向的前提下安全扩展消息，同时保持双向日志兼容。

对需要扩展 openpilot 消息层的开发者而言，操作顺序建议为：先在 custom.capnp 中选定一个未使用的 CustomReservedN 填充字段 → 在 log.capnp 的 Event union 中重命名对应字段（保持 ID 与指向不变）→ 在 services.py 注册服务名与频率 → 运行 openpilot/cereal/messaging/tests/ 下的测试验证一致性。