diff --git a/Cargo.toml b/Cargo.toml index 27b626c..5147fdf 100644 --- a/Cargo.toml +++ b/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 } diff --git a/README.md b/README.md index 8f081b6..c9c859d 100644 --- a/README.md +++ b/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 diff --git a/docs/impl/reports/BUS-BENCH.md b/docs/impl/reports/BUS-BENCH.md new file mode 100644 index 0000000..cad110f --- /dev/null +++ b/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. diff --git a/scripts/bus-bench.sh b/scripts/bus-bench.sh new file mode 100755 index 0000000..2c6a703 --- /dev/null +++ b/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 diff --git a/src/bus-bench/main.rs b/src/bus-bench/main.rs new file mode 100644 index 0000000..9bd073f --- /dev/null +++ b/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) -> Result { + 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, + mut on_publish: impl FnMut(&[u8]) + Send + 'static, +) -> Option { + 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) { + 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 = 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, 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)> { + 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, 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 = None; + let mut lat: Vec = Vec::new(); + let mut first: Option = 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()); + } +}