生产化 Rust:FIX、内核旁路与 PGO

3.5.3 L4 你交付了一个测得过 tick-to-trade 延迟的撮合引擎,内部跑 SPSC 环、SIMD、core_affinity CPU 绑核,hdrhistogram 报分位。现在中信建投自营 IT 跟你坐下来,屏幕上是上线检查清单:​​"它会和我们的会话网关讲 FIX 4.4 吗?TCP 断了能续上吗?对接的同事问你要 NewOrderSingle 35=D 的回单 35=8。"​ —— 这就是 3.5.4 L4 要补的最后一个差距:从"延迟跑过基准"到"能投到 CFFEX 张江 COLO / SSE 浦东 / SZSE 福田机房、上线第一天不出事"的所有工程。本课五条工作流并立:(1) FIX 4.4 会话状态机 + 持久化序列号;(2) tokio-uring 替换 epoll 走内核旁路方向;(3) cargo pgo 两遍编译做反馈优化;(4) tracing + tokio-console 可观测;(5) cargo deny / cargo audit / tikv-jemallocator / reproducible builds 部署硬化。架构就是本课的论点。

工作流 1:FIX 4.4 会话状态机

FIX (Financial Information eXchange) 4.4 是国内券商 (中信建投、国泰君安、华泰证券) 与海外 broker 之间最广泛使用的机构对接协议。会话层管理登录、心跳、序列号、断线续传;应用层负责报单、撤单、改单、回单。本课只覆盖 4.4——FIX 5.0 SP2 / FIXT 1.1 的拆分、FIX/FAST 编码、FIX-over-UDP 都点到为止,留给上线对接文档与交易所认证套件。

use std::result::Result;

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionState {
    LoggedOut,
    LoggingIn,
    LoggedIn,
    ResendRequested,
    GapFilling,
    LoggingOut,
}

#[derive(Debug)]
pub struct FixMessage {
    pub msg_type: u8,         // ASCII: 'A','0','1','2','4','5','D','F','G','8'
    pub seq_num: u64,
    pub fields: Vec<(u32, String)>,
}

#[derive(Debug)]
pub enum SessionError {
    SequenceGap { expected: u64, got: u64 },
    UnexpectedMsgType { state: SessionState, msg_type: u8 },
    LogoutByCounterparty,
    Io(std::io::Error),
}

pub async fn handle_message(
    state: &mut SessionState,
    msg: FixMessage,
) -> Result<Vec<FixMessage>, SessionError> {
    use SessionState::*;
    match (state.clone(), msg.msg_type) {
        (LoggedOut, b'A') => { *state = LoggedIn; Ok(vec![/* echo Logon */]) }
        (LoggedIn, b'0') => Ok(vec![/* heartbeat reply */]),
        (LoggedIn, b'1') => Ok(vec![/* test-request reply: heartbeat with TestReqID */]),
        (LoggedIn, b'2') => { *state = ResendRequested; Ok(vec![/* iterate from store-and-forward */]) }
        (LoggedIn, b'4') => { *state = GapFilling; Ok(vec![/* SequenceReset with GapFillFlag=Y */]) }
        (LoggedIn, b'5') => { *state = LoggedOut; Err(SessionError::LogoutByCounterparty) }
        (LoggedIn, b'D' | b'F' | b'G') => Ok(vec![/* ExecutionReport 35=8 */]),
        (s, t) => Err(SessionError::UnexpectedMsgType { state: s, msg_type: t }),
    }
}

外层把会话嵌进 tokio 异步循环里:Framed<TcpStream, FixCodec>::next() 给你下一条 FixMessage,你 match 出 msg_type,转移状态,产生回应 (Vec<FixMessage>)。序列号断了——比如对端 seq=5 而你期望 seq=3——发 ResendRequest (35=2),对方回 SequenceReset (35=4) 带 GapFillFlag=Y 表示中间报文已经被持久化但不必重发应用消息。

​序列号持久化用 rusqlite 把状态存到本地 SQLite​​:每条入站 / 出站消息更新 (symbol_id, in_seq, out_seq),程序崩溃重启后从 SQLite 读回上次序列号继续。这是 store-and-forward 的最朴实实现,够用、可移植、CI 上免容器、fsync 行为可控。

九个核心 MsgType:A (Logon)、0 (Heartbeat)、1 (TestRequest)、2 (ResendRequest)、4 (SequenceReset)、5 (Logout)、D (NewOrderSingle)、F (OrderCancelRequest)、G (OrderCancelReplaceRequest)、8 (ExecutionReport)。FIX 字段是 SOH (\x01) 分隔的 tag=value,例如 8=FIX.4.4|9=...|35=D|49=SENDER|56=TARGET|34=42|...|10=128|,tag=10 是 CheckSum (前面所有字节模 256)。

工作流 2:tokio-uring 替换 epoll

3.5.3 L4 的行情接收用 tokio::net::UdpSocket,底层走 epoll;每次 recv_from 唤醒一次内核态。tokio-uring (David Tolnay 等维护) 改写成基于 io_uring 的提交队列模型——​​每个 task 维护自己的 SQ/CQ,与内核共享内存,无需 epoll 唤醒系统调用​​。Linux 5.10+ 通用,Ice Lake-SP 上典型 P99 减少 1-3 µs(取决于 workload)。

fn main() {
    tokio_uring::start(async {
        let socket = tokio_uring::net::UdpSocket::bind("0.0.0.0:36001".parse().unwrap())
            .await
            .expect("bind");

        let mut buf = vec![0u8; 65536];
        loop {
            let (res, b) = socket.recv_from(buf).await;
            buf = b;
            let (n, peer) = res.expect("recv_from");
            // 内层处理与 3.5.3 L4 字节完全一致: decode -> book update -> compute signal -> route order
            process_md_datagram(&buf[..n], peer);
        }
    });
}

替换面非常小:#[tokio::main] async fn main() → tokio_uring::start(async { ... });tokio::net::UdpSocket::bind → tokio_uring::net::UdpSocket::bind;recv_from 现在按"接收完归还 buffer"风格返回 (Result<(usize, SocketAddr), Error>, Vec<u8>)——这是 io_uring SQ 借出 buffer 的 API 形状。内层 decode→撮合→路由的代码完全不动。两个不变量:​​每个 task 独占其 SQ​​(避免锁)、​​无 epoll 唤醒 syscall​​(每次 IO 不进内核态)。

详细的内核旁路实现 (DPDK 用户态驱动设置、AF_XDP socket 编程加 XDP_USE_NEED_WAKEUP、Mellanox VMA、Solarflare Onload、RoCE) 属于厂商文档地带——架构图层级触及即止。

工作流 3:cargo pgo 反馈优化

PGO (Profile-Guided Optimisation) 让编译器在第二遍编译时根据第一遍的运行 profile 重排基本块、内联热路径、消除冷分支。Rust 工具链通过 cargo pgo 两步法暴露这条能力:

$ cargo install cargo-pgo

# Pass 1: instrumented build
$ cargo pgo instrument build --release

# Pass 2: representative workload (e.g. 1 hour of UAT replay)
$ ./target/release/trading-engine --replay=fixtures/uat-day.pcap

# Pass 3: optimised build using the profile
$ cargo pgo optimize build --release

典型 5-15% 延迟改进,分支密集的代码(比如 FIX 解析、状态机、book 更新)受益最大。​​生产纪律:对发布的常驻二进制值得做 PGO,对一次性工具不必​​。要点:第二遍跑的 workload 必须代表真实负载——你拿 unit test 跑 PGO,优化的是你不在乎的路径。

工作流 4:可观测——tracing + tracing-subscriber + tokio-console

tracing 是 Rust 生态的结构化日志事实标准 (替代历史上 log 的非结构化路径)。在热路径函数上加 #[tracing::instrument],每次调用产生一个 span,可以发到 stdout 的 JSON、本地文件、或者通过 tracing-opentelemetry 桥到 Tempo / Jaeger。

use tracing::{info, instrument};
use tracing_subscriber::EnvFilter;

#[instrument(skip(book))]
fn book_update(book: &mut OrderBook, evt: &MdEvent) -> SignalDelta {
    /* ... */
    SignalDelta::default()
}

#[instrument]
fn compute_signal(delta: SignalDelta) -> Decision { /* ... */ Decision::Hold }

#[instrument]
fn route_order(d: Decision) { /* ... */ }

fn init_tracing() {
    tracing_subscriber::fmt()
        .json()
        .with_env_filter(EnvFilter::from_default_env())
        .init();
}

tokio-console 是 tokio 运行时的活跃监视器:RUSTFLAGS="--cfg tokio_unstable" cargo run --features console-subscriber 启动你的 binary,另一个终端 tokio-console http://127.0.0.1:6669 看实时 task 调度延迟、队列深度、busy/idle 占比。生产中​​只在 staging 或非热路径上启用​​——tokio-console 自身有一定的事件采集 overhead,不进 colo 二进制。

工作流 5:部署硬化清单

把延迟跑过基准的 binary 真的搬进 colo,还要过七道关:

# rust-toolchain.toml
[toolchain]
channel = "1.78.0"
components = ["rustfmt", "clippy"]

# Cargo.toml [profile.release]
[profile.release]
opt-level = 3
lto = "fat"
codegen-units = 1
panic = "abort"          # 不要在 colo 上 unwind
strip = "symbols"

# 锁定依赖版本
$ cat Cargo.lock | wc -l     # 必须 check in 到 git (binary crate)

# 许可证 + CVE 审计
$ cargo install cargo-deny
$ cargo deny check           # 失败 = CI fail

# RustSec 安全公告
$ cargo install cargo-audit
$ cargo audit                # 同上

# 可重复构建
$ cargo build --release --locked

# jemalloc 全局分配器
# 在 src/main.rs 顶部:
# #[global_allocator]
# static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;

# 跨平台发布
$ cargo install cargo-dist
$ cargo dist init

七件事并立:Cargo.lock check-in (binary crate 必须,library 可选);cargo deny check 许可证+CVE+重复依赖;cargo audit 对 RustSec advisory;--locked + rust-toolchain.toml 固定工具链做可重复构建;panic = "abort" 取消 unwind (释放的 binary 没必要 unwind,且 unwind 表占空间);tikv-jemallocator 替换系统 malloc——jemalloc 在多线程高并发的分配场景下比 glibc malloc 慢路径短(典型 5-10% 全局改善);cargo dist 跨平台发布。

国内一线 quant 与 Rust 团队 (TiKV、字节火山引擎、蚂蚁链) 把这清单写成 CI gate:任一项不过,合入不被允许。这就是从"跑过 benchmark 的代码"到"能放进 NY4 / 张江 COLO 上线第一天不出事的二进制"的工程纪律差。

容量练习:三个工作流叠加,P99 减少 30-60%

经验数字:PGO 5-15% + tokio-uring 1-3 µs(在 5-10 µs 的 baseline 上约 20-40%) + tikv-jemallocator 5-10% 全局——三者叠加在 3.5.3 L4 的 baseline 上通常给出 tick-to-trade P99 30-60% 的减少。架构就是论点:本课不教你新算法,本课教你把已知的工具链组合成生产二进制。

行业实践

FIX 4.4 在国内是中信建投 / 国泰君安 / 华泰证券 / 国信等头部券商 broker 端的机构对接默认协议;CFFEX / SSE / SZSE 与交易所自身的直连用专有协议(SSE 的 STEP 协议、SZSE 的 OES 协议、CFFEX 的 CTP API),但 FIX 4.4 / 4.2 是 broker 面的标准。硬件锚定 Intel Xeon Gold 6342 / 6354 (Ice Lake-SP) 或 AMD EPYC 9354 (Genoa),Rocky Linux 9 / Ubuntu 22.04 LTS,在 CFFEX 张江 / SSE 浦东 / SZSE 福田 COLO 机房与 3.5.3 / 3.4.x 一致。

Exercise

(a) 把 FIX 4.4 会话状态机接进 3.5.3 L4 的撮合引擎,通过 Framed<TcpStream, FixCodec> 与 rusqlite 持久化序列号;通过合成 UAT 套件(登录、心跳、序列跳跃后的 gap-fill、TCP 中断后的 resend-request、正常 logout)。 (b) 把 tokio::net::UdpSocket 行情接收换成 tokio_uring::net::UdpSocket,跑在 tokio_uring::start 下;重跑 3.5.3 L4 的 tick-to-trade 基准;报告 P50 / P99 / P99.9 延迟与基线对比。 (c) 三种方式构建撮合引擎——普通 release、PGO 插桩版、PGO 优化版(在跑过代表性 workload 后);各自的 tick-to-trade P99 延迟。 (d) 用 #[tracing::instrument] 给热路径加观测,接入 tokio-console;抓 30 秒运行快照,展示四个热路径 task (feed / book / signal / router) 的调度延迟、队列深度、busy/idle 比。 (e) 部署硬化清单(Cargo.lock check-in、cargo deny check、cargo audit、--locked 可重复构建、rust-toolchain.toml 固定、panic = "abort"、tikv-jemallocator)全部应用一遍;报告二进制大小、依赖审计摘要、构建可重复性测试结果。

提示

FIX 状态机调试时, 用一个 fixture 文件存入站消息字节流; tokio_test::block_on(handle_message(&mut state, fix_message)) 反复测试状态转移, 不必拉起真实 TCP。

提示

cargo pgo optimize 失败常因 profile 数据缺失: 先确认 instrument build 跑过 workload, profile 落在 target/pgo-profiles/。命令链: instrument → run → optimize, 缺一不可。

本课五个 Fenced 代码块——FIX 会话状态机骨架、tokio-uring 替换 diff、cargo pgo 三阶段流程、tracing + tokio-console 接线、部署硬化清单——是把 3.5.3 L4 撮合引擎搬进 COLO 机房前缺失的最后一块拼图。

上线前的最后一公里

工程意义上还有三件事本课未展开,但每一件都要在你真正上线那天落地。​​第一,延迟监控的回路必须是持续的,不是上线那天测一次​​:写一个 cron job 或者 systemd timer,每小时跑一次 tick-to-trade 基准,把 P50 / P99 / P99.9 落到 Prometheus,Grafana 上看趋势。Rust 工具链升级、内核 patch、JVM 邻居(同主机若有其他进程)的资源争抢都会让你的 P99 默默漂移。​​第二,FIX 会话的对账逻辑要有独立的离线脚本​​:每天收盘后跑一个 Python 脚本,拉 SQLite 持久化的序列号 + ExecutionReport 历史,与 broker 端的对账文件比对,差异落到告警。生产代码再正确,跨进程跨时区跨交易所的对账永远会出小差错,要做成日常工作流而不是事件响应。​​第三,降级路径要演练​​:io_uring 在某些内核版本上有 bug、PGO 的 profile 数据过期会让优化偏离、jemalloc 在某些工作负载下不如系统 malloc。生产系统永远有一个"退回稳定版"的开关,而退回演练每季度做一次——别让"上次降级是什么时候"变成生产事故现场的第一个问题。

与 3.5 主题的关系

3.5.1 教会你 Rust 语法、所有权、Result<T, E>,你能写一个单线程的 Monte Carlo 二进制。3.5.2 教会你 Arc<Mutex<T>>、tokio、原子序,你能写一个多线程的网络服务。3.5.3 教会你缓存布局、SIMD、criterion、ITCH 5.0 解码、CPU 绑核,你能写一台测得过基准的撮合引擎。3.5.4(本模块)教会你 unsafe 的健全性论证、FFI 互操作、Python 互操作、生产硬化清单,你能把那台撮合引擎真正交付给一个团队上线运营。四个模块累加起来,是把一名"会写 Rust"的工程师变成"能在 Rust 量化团队独立负责一个生产服务"的工程师的最短路径。

模块完成。从 L1 // SAFETY: 纪律出发,经 L2 C/C++ 互操作,L3 Python 互操作,到本课的生产硬化压轴——你现在有一份能投到 CFFEX 张江 / SSE 浦东 / SZSE 福田 COLO 机房的 Rust 撮合二进制,讲 FIX 4.4 给中信建投 / 国泰君安 / 华泰证券,跑在 io_uring + PGO + jemalloc 上,被 tracing + tokio-console 观测。下一个主题 (3.6 软件工程与 SRE) 把视野从单进程性能拉到整个团队的代码协作、CI/CD、监控告警——这台 Rust 引擎在那个上下文里只是一个被治理的产物。