bus-bench: raw CAN over MQTT, measured on the car Orin

SocketCAN to yukart/can/raw and back, per frame or batched, with loss and
latency from the kernel timestamp. No loss at a saturated 500 kbit/s bus;
batched costs the broker ~1 % of a core. Report in docs/impl/reports.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz

Cargo.toml

@@ -13,6 +13,10 @@ path = "src/recorder/main.rs"
 name = "orin-status"
 path = "src/orin-status/main.rs"
 
+[[bin]]
+name = "bus-bench"
+path = "src/bus-bench/main.rs"
+
 [dependencies]
 libc = "0.2"
 rumqttc = { version = "0.25", default-features = false }

README.md

@@ -16,6 +16,7 @@ needed.
 | Binary | Installed as | Does |
 |---|---|---|
 | `recorder` | `yukart-recorder.service` | Raw CAN + GNSS to disk, in sessions, under a storage bound |
+| `bus-bench` | not installed: a test tool | Raw CAN over MQTT, measured: SocketCAN → `yukart/can/raw` → loss and latency ([report](docs/impl/reports/BUS-BENCH.md)) |
 | `orin-status` | `yukart-orin-status.service` | The Orin's health on the bus, once a second: temperatures, CPU/GPU load, memory, power, fan, WiFi/LTE/tablet links, UPS battery (`yukart/host/*`) |
 
 ```bash

docs/impl/reports/BUS-BENCH.md

@@ -0,0 +1,76 @@
+---
+type: report
+date: 2026-10-12
+---
+# Raw CAN over MQTT: bus bench
+
+Can mosquitto carry `yukart/can/raw` at real CAN rates (project-yukart decision 4)?
+Measured on the car Orin (mosquitto 2.0.11, localhost) with `bus-bench`: `bridge`
+reads a `vcan0` and publishes frames, `sink` subscribes, checks sequence numbers
+and measures latency from the kernel's reception timestamp to the subscriber.
+The car's usual services (recorder, orin-status, svc-gw, svc-vision) kept running.
+
+```bash
+scripts/bus-bench.sh ./bus-bench z27ag-20261009-163736-ign.log 20     # on the Orin
+```
+
+Loads:
+
+- **car replay**: the in-car capture (`z27ag-20261009-163736-ign.log`, engine at
+  idle) at its own timing, about 890 frames/s. This is the real bus.
+- **full bus**: `cangen -L 8` at about 4,400 frames/s. 500 kbit/s carries roughly
+  3,700–4,500 8-byte frames/s depending on bit stuffing, so this is a saturated bus.
+- **flood**: `cangen -g 0`, as fast as vcan goes, to find where it breaks.
+
+Two encodings: one MQTT message per frame, or frames batched over a 10 ms window.
+Both use the same provisional binary format (24 bytes a frame, 12-byte header;
+see `src/bus-bench/main.rs`). QoS 0, not retained.
+
+## Results (20 s runs; full bus 15 s)
+
+With `TCP_NODELAY` on both the clients and the broker (see below):
+
+| Load | Encoding | Frames/s delivered | Lost | Latency p50 / p99 / max (ms) | CPU: broker · bridge · sink (% of a core) |
+|---|---|---|---|---|---|
+| car replay | per frame | 893 | 0 | 0.31 / 0.92 / 1.3 | 4.7 · 3.5 · 3.8 |
+| car replay | 10 ms batch | 893 | 0 | 8.7 / 15.3 / 17.6 | 0.9 · 1.2 · 0.6 |
+| full bus | per frame | 4,440 | 0 | 0.16 / 0.30 / 3.3 | 27.5 · 21.7 · 17.0 |
+| full bus | 10 ms batch | 4,562 | 0 | 5.3 / 10.4 / 10.8 | 1.1 · 3.5 · 0.8 |
+| flood | per frame | 45,700 | 84 % | ~1.8 s, queues full | 85 · 86 · 81 |
+| flood | 10 ms batch | 268,800 | 0 | 5.3 / 10.2 / 11.1 | 1.3 · 63 · 0.9 |
+
+- **Real rates hold, with margin.** Per frame breaks at about 45,000 messages/s:
+  ten times a saturated bus and fifty times the real one.
+- **Batching is the clear choice.** At a full bus it costs the broker about 1 % of a
+  core instead of 28 %, at a fixed ~100 messages/s whatever the bus load. Frames are
+  never lost: the batched flood (60 times a full bus) still lost none, and the
+  bridge's own CAN reads became the limit. The price is up to one window of
+  latency (the 10 ms here). That is no problem for a dashboard, a recorder or a
+  decoder.
+- Per frame stays the option for a consumer that needs sub-millisecond latency. Nothing
+  planned does.
+
+## Nagle: the 40 ms tail
+
+The first runs had a p99.9 near 40 ms at every load: Nagle's algorithm holding small
+writes back until the previous one is acknowledged, against delayed ACKs. Both
+sides do it by default: rumqttc (`NetworkOptions::set_tcp_nodelay`, off by default)
+and mosquitto (`set_tcp_nodelay`, off by default).
+
+| Full bus (~3,000 frames/s) | per frame p99 / p99.9 / max | 10 ms batch p99 / max |
+|---|---|---|
+| Nagle on both sides | 23.5 / 41.7 / 45 | 65.6 / 80.9 |
+| client no-delay, broker Nagle | 0.61 / 38.0 / 47 | 43.3 / 55.3 |
+| no-delay on both | 0.33 / 0.67 / 2.8 | 10.5 / 10.8 |
+
+The car's broker now has `set_tcp_nodelay true` (orin-nano-env-setup `MOSQUITTO.md`).
+Clients that publish or consume raw CAN set `set_tcp_nodelay(true)` too. At once a
+second (orin-status) it does not matter.
+
+## Not tested
+
+- A real CAN adapter: vcan has no bus timing, and the PCAN-USB's own reception and
+  timestamps are already covered by the recorder.
+- Several subscribers on `yukart/can/raw` at once (recorder reads SocketCAN directly,
+  so expected consumers are the decoder and perhaps svc-gw).
+- The broker under memory pressure, or with the GPU/vision load of later phases.

scripts/bus-bench.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+# Raw CAN over MQTT, measured on the car Orin against a vcan0 (project-yukart decision 4).
+# Runs on the Orin; sudo once, to create vcan0. Prints one JSON line per run: the
+# sink's and the bridge's reports, and mosquitto's CPU over the run.
+#
+#   usage: bus-bench.sh BUS_BENCH_BINARY CAPTURE.log [secs]
+#   env:   BATCHES  batch windows in ms, 0 = one message per frame   ("0 10")
+#          FLAGS    extra options for both roles, e.g. --nagle
+#          GAP_MS   cangen gap for the full-bus run (0.15: ~4500 frames/s here, past what
+#                   500 kbit/s can carry with 8-byte frames)
+set -euo pipefail
+
+BIN="$1"; CAPTURE="$2"; SECS="${3:-20}"
+BATCHES="${BATCHES:-0 10}"; FLAGS="${FLAGS:-}"
+
+if ! ip link show vcan0 >/dev/null 2>&1; then
+  sudo modprobe vcan
+  sudo ip link add vcan0 type vcan
+fi
+sudo ip link set vcan0 up txqueuelen 10000
+
+ticks() { awk '{print $14 + $15}' "/proc/$1/stat"; }
+HZ=$(getconf CLK_TCK)
+OUT=$(mktemp -d)
+
+# run NAME BATCH_MS GENERATOR...: the generator feeds vcan0 for a little over SECS seconds.
+run() {
+  local name="$1" batch="$2"; shift 2
+  # shellcheck disable=SC2086
+  "$BIN" sink --secs "$SECS" $FLAGS > "$OUT/sink" 2>/dev/null &
+  local sink=$!
+  sleep 0.5
+  # shellcheck disable=SC2086
+  "$BIN" bridge --batch-ms "$batch" --secs $((SECS + 3)) $FLAGS > "$OUT/bridge" 2>/dev/null &
+  local bridge=$!
+  sleep 0.5
+  local broker m0 m1 t0 t1
+  broker=$(pgrep -x mosquitto); m0=$(ticks "$broker"); t0=$(date +%s.%N)
+  timeout "$((SECS + 1))" "$@" >/dev/null 2>&1 || true
+  m1=$(ticks "$broker"); t1=$(date +%s.%N)
+  wait "$sink" "$bridge"
+  python3 - "$name" "$batch" "$FLAGS" "$OUT" "$((m1 - m0))" "$HZ" "$t0" "$t1" <<'PY'
+import json, sys
+name, batch, flags, out, ticks, hz, t0, t1 = sys.argv[1:]
+r = {"run": name, "batch_ms": int(batch), "flags": flags,
+     "sink": json.load(open(f"{out}/sink")), "bridge": json.load(open(f"{out}/bridge")),
+     "mosquitto_cpu_pct": round(100 * int(ticks) / int(hz) / (float(t1) - float(t0)), 1)}
+for k in ("role",):
+    r["sink"].pop(k); r["bridge"].pop(k)
+print(json.dumps(r))
+PY
+}
+
+LOOPS=$(( (SECS + 9) / 10 + 1 ))    # the capture is 10 s; loop past the run's end
+for batch in $BATCHES; do
+  run "car replay"  "$batch" canplayer -l "$LOOPS" -I "$CAPTURE" vcan0=can_pcan
+  run "full bus"    "$batch" cangen vcan0 -g "${GAP_MS:-0.15}" -L 8
+  run "flood"       "$batch" cangen vcan0 -g 0 -L 8
+done

src/bus-bench/main.rs

@@ -0,0 +1,342 @@
+//! bus-bench — can MQTT carry raw CAN at real rates? (project-yukart decision 4)
+//!
+//! `bridge` reads a SocketCAN interface and publishes frames to `yukart/can/raw`,
+//! one message per frame or batched by time. `sink` subscribes, checks every frame
+//! arrived in order, and measures latency from the kernel's reception timestamp to
+//! arrival at the subscriber. Both sides run on the same host, so one clock serves.
+//!
+//! Test tool, not a service: the wire format below is provisional and lives only
+//! here until contracts freezes `yukart/can/raw`.
+//!
+//!   message = seq u64 (first frame's) · count u32 · count × frame        (little-endian)
+//!   frame   = ts_us i64 · can_id u32 (with EFF/RTR/ERR flags) · len u8 · 3 pad · data 8   (24 bytes)
+
+#[path = "../recorder/can.rs"]
+#[allow(dead_code)]
+mod can;
+#[path = "../recorder/clock.rs"]
+#[allow(dead_code)]
+mod clock;
+
+use std::process::ExitCode;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc::{sync_channel, RecvTimeoutError};
+use std::sync::Arc;
+use std::thread;
+use std::time::{Duration, Instant};
+
+use rumqttc::{Client, Event, MqttOptions, NetworkOptions, Packet, QoS};
+use serde_json::json;
+
+const USAGE: &str = "\
+usage:
+  bus-bench bridge [options]     SocketCAN → yukart/can/raw
+      --iface NAME        CAN interface                       (vcan0)
+      --batch-ms N        batch window; 0 = one message per frame (10)
+      --secs N            stop after N seconds                (30)
+  bus-bench sink [options]       yukart/can/raw → counts and latency
+      --secs N            stop N seconds after the first frame (30)
+  both:
+      --broker HOST:PORT                                      (127.0.0.1:1883)
+      --topic TOPIC                                           (yukart/can/raw)
+      --nagle             leave Nagle's algorithm on (rumqttc's default)";
+
+const HEADER: usize = 12;
+const FRAME: usize = 24;
+
+/// CPU time this process has used, user + system, in seconds.
+fn cpu_secs() -> f64 {
+    let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
+    unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut ru) };
+    let tv = |t: libc::timeval| t.tv_sec as f64 + t.tv_usec as f64 * 1e-6;
+    tv(ru.ru_utime) + tv(ru.ru_stime)
+}
+
+fn now_us() -> i64 {
+    (clock::realtime() * 1e6) as i64
+}
+
+struct Opts {
+    iface: String,
+    batch_ms: u64,
+    secs: u64,
+    broker: (String, u16),
+    topic: String,
+    nagle: bool,
+}
+
+fn parse(mut args: impl Iterator<Item = String>) -> Result<Opts, String> {
+    let mut o = Opts {
+        iface: "vcan0".into(),
+        batch_ms: 10,
+        secs: 30,
+        broker: ("127.0.0.1".into(), 1883),
+        topic: "yukart/can/raw".into(),
+        nagle: false,
+    };
+    while let Some(a) = args.next() {
+        let mut val = || args.next().ok_or(format!("{a} needs a value"));
+        match a.as_str() {
+            "--iface" => o.iface = val()?,
+            "--batch-ms" => o.batch_ms = val()?.parse().map_err(|_| "--batch-ms needs a number")?,
+            "--secs" => o.secs = val()?.parse().map_err(|_| "--secs needs a number")?,
+            "--topic" => o.topic = val()?,
+            "--nagle" => o.nagle = true,
+            "--broker" => {
+                let v = val()?;
+                let (h, p) = v.rsplit_once(':').ok_or("--broker must be HOST:PORT")?;
+                o.broker = (h.into(), p.parse().map_err(|_| "--broker must be HOST:PORT")?);
+            }
+            _ => return Err(format!("unknown option {a}")),
+        }
+    }
+    Ok(o)
+}
+
+/// A client whose event loop runs on its own thread. `connected` follows ConnAck;
+/// incoming publishes go to `on_publish`. Returns once connected, or None after 5 s.
+fn connect(
+    id: &str,
+    o: &Opts,
+    cap: usize,
+    subscribe: Option<String>,
+    mut on_publish: impl FnMut(&[u8]) + Send + 'static,
+) -> Option<Client> {
+    let mut opts = MqttOptions::new(id, o.broker.0.clone(), o.broker.1);
+    opts.set_keep_alive(Duration::from_secs(5));
+    opts.set_max_packet_size(1 << 20, 1 << 20);
+    let (client, mut connection) = Client::new(opts, cap);
+    let mut net = NetworkOptions::new();
+    net.set_tcp_nodelay(!o.nagle);
+    connection.eventloop.set_network_options(net);
+    let connected = Arc::new(AtomicBool::new(false));
+    let (c, up) = (client.clone(), connected.clone());
+    thread::spawn(move || {
+        for ev in connection.iter() {
+            match ev {
+                Ok(Event::Incoming(Packet::ConnAck(_))) => {
+                    if let Some(t) = &subscribe {
+                        let _ = c.try_subscribe(t.clone(), QoS::AtMostOnce);
+                    } else {
+                        up.store(true, Ordering::Relaxed);
+                    }
+                }
+                Ok(Event::Incoming(Packet::SubAck(_))) => up.store(true, Ordering::Relaxed),
+                Ok(Event::Incoming(Packet::Publish(p))) => on_publish(&p.payload),
+                Ok(_) => {}
+                Err(e) => {
+                    eprintln!("{e}");
+                    up.store(false, Ordering::Relaxed);
+                    thread::sleep(Duration::from_millis(200));
+                }
+            }
+        }
+    });
+    let t = Instant::now();
+    while !connected.load(Ordering::Relaxed) {
+        if t.elapsed() > Duration::from_secs(5) {
+            return None;
+        }
+        thread::sleep(Duration::from_millis(10));
+    }
+    Some(client)
+}
+
+fn encode(f: &can::Frame, out: &mut Vec<u8>) {
+    out.extend_from_slice(&(f.sec * 1_000_000 + f.usec).to_le_bytes());
+    out.extend_from_slice(&f.id.to_le_bytes());
+    out.extend_from_slice(&[f.len, 0, 0, 0]);
+    out.extend_from_slice(&f.data);
+}
+
+fn bridge(o: Opts) -> ExitCode {
+    let Some(client) = connect("bus-bench-bridge", &o, 4096, None, |_| {}) else {
+        eprintln!("bus-bench: no broker at {}:{}", o.broker.0, o.broker.1);
+        return ExitCode::FAILURE;
+    };
+    let mut sock = match can::CanSocket::open(&o.iface, Duration::from_millis(2)) {
+        Ok(s) => s,
+        Err(e) => {
+            eprintln!("bus-bench: {}: {e}", o.iface);
+            return ExitCode::FAILURE;
+        }
+    };
+    let window = Duration::from_millis(o.batch_ms);
+    let (mut seq, mut frames, mut msgs, mut refused, mut kdrops) = (0u64, 0u64, 0u64, 0u64, 0u64);
+    let mut buf: Vec<u8> = Vec::with_capacity(HEADER + FRAME * 512);
+    let mut count = 0u32;
+    let mut opened = Instant::now();
+    let start = Instant::now();
+    let cpu0 = cpu_secs();
+    let mut flush = |buf: &mut Vec<u8>, count: &mut u32, seq: &mut u64| {
+        if *count == 0 {
+            return;
+        }
+        buf[8..12].copy_from_slice(&count.to_le_bytes());
+        match client.try_publish(o.topic.clone(), QoS::AtMostOnce, false, buf.clone()) {
+            Ok(()) => msgs += 1,
+            Err(_) => refused += *count as u64, // client queue full: these frames are lost
+        }
+        *seq += *count as u64;
+        *count = 0;
+        buf.clear();
+    };
+    while start.elapsed() < Duration::from_secs(o.secs) {
+        match sock.recv() {
+            Ok(can::Recv::Frame(f, d)) => {
+                kdrops += d as u64;
+                if count == 0 {
+                    buf.extend_from_slice(&seq.to_le_bytes());
+                    buf.extend_from_slice(&[0; 4]);
+                    opened = Instant::now();
+                }
+                encode(&f, &mut buf);
+                count += 1;
+                frames += 1;
+            }
+            Ok(can::Recv::Timeout) => {}
+            Err(e) => {
+                eprintln!("bus-bench: {}: {e}", o.iface);
+                return ExitCode::FAILURE;
+            }
+        }
+        if count > 0 && opened.elapsed() >= window {
+            flush(&mut buf, &mut count, &mut seq);
+        }
+    }
+    flush(&mut buf, &mut count, &mut seq);
+    thread::sleep(Duration::from_millis(500)); // let the queue drain before the process exits
+    let secs = start.elapsed().as_secs_f64();
+    let cpu = (cpu_secs() - cpu0) / secs * 100.0;
+    println!(
+        "{}",
+        json!({ "role": "bridge", "batch_ms": o.batch_ms, "secs": (secs * 10.0).round() / 10.0,
+                "frames": frames, "messages": msgs, "refused_frames": refused, "kernel_drops": kdrops,
+                "cpu_pct": (cpu * 10.0).round() / 10.0 })
+    );
+    ExitCode::SUCCESS
+}
+
+/// What the sink learns from one message: (first seq, frames, per-frame latency µs).
+fn decode(p: &[u8], now: i64) -> Option<(u64, u32, Vec<i64>)> {
+    let seq = u64::from_le_bytes(p.get(0..8)?.try_into().ok()?);
+    let count = u32::from_le_bytes(p.get(8..12)?.try_into().ok()?);
+    if p.len() != HEADER + FRAME * count as usize {
+        return None;
+    }
+    let lat = p[HEADER..]
+        .chunks_exact(FRAME)
+        .map(|f| now - i64::from_le_bytes(f[0..8].try_into().unwrap()))
+        .collect();
+    Some((seq, count, lat))
+}
+
+fn pct(sorted: &[i64], q: f64) -> f64 {
+    if sorted.is_empty() {
+        return 0.0;
+    }
+    let i = ((sorted.len() - 1) as f64 * q).round() as usize;
+    sorted[i] as f64 / 1000.0
+}
+
+fn sink(o: Opts) -> ExitCode {
+    // Decoding happens on the event-loop thread; the main thread only collects.
+    let (tx, rx) = sync_channel::<(u64, u32, Vec<i64>, usize)>(1 << 16);
+    let topic = o.topic.clone();
+    let on_publish = move |p: &[u8]| match decode(p, now_us()) {
+        Some((s, c, l)) => {
+            let _ = tx.try_send((s, c, l, p.len()));
+        }
+        None => eprintln!("bus-bench: malformed message ({} bytes)", p.len()),
+    };
+    let Some(_client) = connect("bus-bench-sink", &o, 16, Some(topic), on_publish) else {
+        eprintln!("bus-bench: no broker at {}:{}", o.broker.0, o.broker.1);
+        return ExitCode::FAILURE;
+    };
+    eprintln!("bus-bench: subscribed to {}; waiting for frames", o.topic);
+    let (mut frames, mut msgs, mut bytes, mut lost, mut reordered) = (0u64, 0u64, 0u64, 0u64, 0u64);
+    let mut next: Option<u64> = None;
+    let mut lat: Vec<i64> = Vec::new();
+    let mut first: Option<Instant> = None;
+    let mut cpu0 = 0.0;
+    loop {
+        if let Some(t) = first {
+            if t.elapsed() >= Duration::from_secs(o.secs) {
+                break;
+            }
+        }
+        match rx.recv_timeout(Duration::from_millis(100)) {
+            Ok((s, c, l, n)) => {
+                if first.is_none() {
+                    first = Some(Instant::now());
+                    cpu0 = cpu_secs();
+                }
+                match next {
+                    Some(want) if s > want => lost += s - want,
+                    Some(want) if s < want => reordered += 1,
+                    _ => {}
+                }
+                next = Some(s + c as u64);
+                frames += c as u64;
+                msgs += 1;
+                bytes += n as u64;
+                lat.extend(l);
+            }
+            Err(RecvTimeoutError::Timeout) => {}
+            Err(RecvTimeoutError::Disconnected) => break,
+        }
+    }
+    let secs = first.map(|t| t.elapsed().as_secs_f64()).unwrap_or(0.0);
+    let cpu = (cpu_secs() - cpu0) / secs.max(1e-9) * 100.0;
+    lat.sort_unstable();
+    let r = |v: f64| (v * 100.0).round() / 100.0;
+    println!(
+        "{}",
+        json!({ "role": "sink", "secs": r(secs), "frames": frames, "messages": msgs,
+                "frames_per_s": (frames as f64 / secs.max(1e-9)).round(),
+                "messages_per_s": (msgs as f64 / secs.max(1e-9)).round(),
+                "kbytes_per_s": r(bytes as f64 / 1000.0 / secs.max(1e-9)),
+                "lost_frames": lost, "reordered_messages": reordered,
+                "latency_ms": { "p50": r(pct(&lat, 0.5)), "p99": r(pct(&lat, 0.99)),
+                                "p999": r(pct(&lat, 0.999)), "max": r(pct(&lat, 1.0)) },
+                "cpu_pct": (cpu * 10.0).round() / 10.0 })
+    );
+    ExitCode::SUCCESS
+}
+
+fn main() -> ExitCode {
+    let mut args = std::env::args().skip(1);
+    let role = args.next().unwrap_or_default();
+    let o = match parse(args) {
+        Ok(o) => o,
+        Err(e) => {
+            eprintln!("bus-bench: {e}\n{USAGE}");
+            return ExitCode::from(2);
+        }
+    };
+    match role.as_str() {
+        "bridge" => bridge(o),
+        "sink" => sink(o),
+        _ => {
+            eprintln!("{USAGE}");
+            ExitCode::from(2)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn frame_round_trips_through_the_wire_format() {
+        let f = can::Frame { sec: 10, usec: 5, id: 0x449, len: 8, data: [1, 2, 3, 4, 5, 6, 7, 8] };
+        let mut m = 7u64.to_le_bytes().to_vec();
+        m.extend_from_slice(&1u32.to_le_bytes());
+        encode(&f, &mut m);
+        assert_eq!(m.len(), HEADER + FRAME);
+        let (seq, count, lat) = decode(&m, 10_000_105).unwrap();
+        assert_eq!((seq, count, lat), (7, 1, vec![100]));
+        assert!(decode(&m[..m.len() - 1], 0).is_none());
+    }
+}