From b605c27fc0ba64d80b04660169f2404f2e3d22de Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Mon, 12 Oct 2026 22:33:44 +0900 Subject: [PATCH] bus-bench: decoder stand-in and mixed-load runs Raw frames to four subscribers plus envelopes to five, at the car's rate and at a full bus. Raw never suffered; envelope deliveries are the load, with the knee near 40,000 deliveries/s. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz --- README.md | 2 +- docs/impl/reports/BUS-BENCH.md | 51 +++++++++++++- scripts/bus-bench.sh | 22 +++--- scripts/bus-mix.sh | 76 ++++++++++++++++++++ src/bus-bench/main.rs | 124 +++++++++++++++++++++++++++++++-- 5 files changed, 256 insertions(+), 19 deletions(-) create mode 100755 scripts/bus-mix.sh diff --git a/README.md b/README.md index c9c859d..dbbb191 100644 --- a/README.md +++ b/README.md @@ -16,7 +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)) | +| `bus-bench` | not installed: a test tool | Raw CAN and decoded-signal load over MQTT, measured: loss, latency, broker CPU ([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 index cad110f..66b466e 100644 --- a/docs/impl/reports/BUS-BENCH.md +++ b/docs/impl/reports/BUS-BENCH.md @@ -12,6 +12,7 @@ The car's usual services (recorder, orin-status, svc-gw, svc-vision) kept runnin ```bash scripts/bus-bench.sh ./bus-bench z27ag-20261009-163736-ign.log 20 # on the Orin +scripts/bus-mix.sh ./bus-bench z27ag-20261009-163736-ign.log 20 # after bus-bench.sh ``` Loads: @@ -67,10 +68,56 @@ The car's broker now has `set_tcp_nodelay true` (orin-nano-env-setup `MOSQUITTO. Clients that publish or consume raw CAN set `set_tcp_nodelay(true)` too. At once a second (orin-status) it does not matter. +## The bus as the car will use it + +`scripts/bus-mix.sh`: the raw stream (10 ms batches) to four subscribers (the decoder +plus three: svc-gw, telemetry-gw, a debug tool), and a stand-in `decoder` publishing +envelopes on one retained topic per signal, to five subscribers that each take every +signal (svc-gw, recorder, svc-nav, svc-vision, telemetry-gw). Loads keep the car's +27 CAN IDs and their rates (2–100 Hz): the capture as recorded, and played 5× faster +for a full bus. Signals per frame and a per-signal rate cap (latest value wins) +stand in for the decoder design nobody has made yet. 20 s runs. + +| Load | Signals/frame | Cap | Envelopes/s | Deliveries/s (×5) | Broker CPU | Envelope latency p50 / p99 (ms) | +|---|---|---|---|---|---|---| +| car | 3 | 50 Hz | 1,600 | 8,000 | 22 % | 13 / 21 | +| car | 3 | none | 2,550 | 12,800 | 35 % | 13 / 20 | +| car | 8 | 20 Hz | 2,550 | 12,800 | 32 % | 14 / 27 | +| car | 8 | 50 Hz | 4,300 | 21,600 | 47 % | 16 / 32 | +| car | 8 | none | 6,800 | 34,000 | 51 % | 14 / 47 | +| full bus | 3 | 50 Hz | 3,100 | 15,500 | 39 % | 12 / 18 | +| full bus | 8 | 20 Hz | 3,500 | 17,300 | 44 % | 22 / 41 | +| full bus | 8 | 50 Hz | 8,100 | 40,600 | 57 % | 15 / **391** | +| full bus | 3 | none | 12,800 | 63,800 | 74 % | **636 / 1,472** | +| full bus | 8 | none | 16,300 offered | — | 75 % | **5,380 / 5,994**; decoder queue overflowing | + +- **The raw stream never suffered.** All four raw subscribers got every frame in every + run, p99 ≤ 16 ms, including the runs where the envelopes collapsed. +- **Envelopes are the load.** The broker's cost follows deliveries (messages × + subscribers). Up to ~22,000 deliveries/s it stays under half a core with p99 near + 30 ms; around 40,000 the tail goes to hundreds of ms; past ~60,000 queues grow by + seconds. Mosquitto is single-threaded, so this is a ceiling, not a share. +- **Each subscriber pays too.** Parsing JSON envelopes cost each sink 14–38 % of a + core at these rates; five of them took two cores. A real subscriber that takes + everything (svc-gw) pays the same. +- Envelope loss is not measured exactly (no sequence numbers in the envelope); + the decoder's own `refused` counts what it could not even queue. + +What keeps it in the comfortable range, any of which is enough at the car's rate: + +1. **Cap each signal's rate in the decoder**, latest value wins. 20–50 Hz is still + 10–50× OBD-II polling, and the raw stream keeps the full rate for anything that + needs it. +2. **Subscribe narrowly.** Only a consumer that shows or stores everything (svc-gw) + needs `yukart/vehicle/+`; svc-nav and svc-vision need a handful of topics. +3. **Publish slow signals on change**, with a heartbeat; temperatures and states do + not need 50 Hz. +4. **The recorder need not record decoded signals at full rate.** It records raw + frames, and decoding them again offline gives the same signals. + ## 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. +- The real decoder: signals per frame, rates and topic layout are stand-ins. diff --git a/scripts/bus-bench.sh b/scripts/bus-bench.sh index 2c6a703..7a02a07 100755 --- a/scripts/bus-bench.sh +++ b/scripts/bus-bench.sh @@ -6,12 +6,13 @@ # 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 +# SINKS subscribers per run; each reports separately (1) # 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:-}" +BATCHES="${BATCHES:-0 10}"; FLAGS="${FLAGS:-}"; SINKS="${SINKS:-1}" if ! ip link show vcan0 >/dev/null 2>&1; then sudo modprobe vcan @@ -26,9 +27,12 @@ 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=$! + local sinks=() i + for i in $(seq "$SINKS"); do + # shellcheck disable=SC2086 + "$BIN" sink --secs "$SECS" $FLAGS > "$OUT/sink$i" 2>/dev/null & + sinks+=($!) + done sleep 0.5 # shellcheck disable=SC2086 "$BIN" bridge --batch-ms "$batch" --secs $((SECS + 3)) $FLAGS > "$OUT/bridge" 2>/dev/null & @@ -38,15 +42,15 @@ run() { 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" + wait "${sinks[@]}" "$bridge" python3 - "$name" "$batch" "$FLAGS" "$OUT" "$((m1 - m0))" "$HZ" "$t0" "$t1" <<'PY' -import json, sys +import glob, 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")), + "sinks": [json.load(open(p)) for p in sorted(glob.glob(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) +for d in r["sinks"] + [r["bridge"]]: + d.pop("role") print(json.dumps(r)) PY } diff --git a/scripts/bus-mix.sh b/scripts/bus-mix.sh new file mode 100755 index 0000000..686dc92 --- /dev/null +++ b/scripts/bus-mix.sh @@ -0,0 +1,76 @@ +#!/usr/bin/env bash +# The bus as the car will use it: raw frames to several subscribers, a decoder turning +# them into envelopes, several envelope subscribers. Runs on the car Orin against a +# vcan0, beside bus-bench.sh (which makes vcan0). Prints one JSON line per run. +# +# usage: bus-mix.sh BUS_BENCH_BINARY CAPTURE.log [secs] +# env: CASES "batch_ms signals max_hz" per run, ';'-separated (default: the matrix below) +# +# Loads keep the car's own CAN IDs and their relative rates: the capture as recorded +# (~890 frames/s), and the same capture played 5× faster (~4,450 frames/s, a full bus). +set -euo pipefail + +BIN="$1"; CAPTURE="$2"; SECS="${3:-20}" +RAW_SINKS=3 # besides the decoder: svc-gw, telemetry-gw, a debug tool +ENV_SINKS=5 # svc-gw, recorder, svc-nav, svc-vision, telemetry-gw + +ip link show vcan0 >/dev/null 2>&1 || { echo "vcan0 missing: run bus-bench.sh once"; exit 1; } +OUT=$(mktemp -d) +FAST="$OUT/fast.log" +awk 'NR == 1 { t0 = substr($1, 2, length($1) - 2) } + { t = substr($1, 2, length($1) - 2); printf "(%.6f) %s %s\n", t0 + (t - t0) / 5, $2, $3 }' "$CAPTURE" > "$FAST" + +ticks() { awk '{print $14 + $15}' "/proc/$1/stat"; } +HZ=$(getconf CLK_TCK) + +# run NAME LOG BATCH_MS SIGNALS MAX_HZ +run() { + local name="$1" log="$2" batch="$3" signals="$4" maxhz="$5" pids=() i + rm -f "$OUT"/*.json + for i in $(seq "$RAW_SINKS"); do + "$BIN" sink --secs "$SECS" > "$OUT/raw$i.json" 2>/dev/null & pids+=($!) + done + for i in $(seq "$ENV_SINKS"); do + "$BIN" sink --topic 'yukart/bench/vehicle/+' --secs "$SECS" > "$OUT/env$i.json" 2>/dev/null & pids+=($!) + done + "$BIN" decoder --signals "$signals" --max-hz "$maxhz" --secs $((SECS + 2)) > "$OUT/decoder.json" 2>/dev/null & pids+=($!) + sleep 0.5 + "$BIN" bridge --batch-ms "$batch" --secs $((SECS + 3)) > "$OUT/bridge.json" 2>/dev/null & pids+=($!) + sleep 0.5 + local broker m0 m1 t0 t1 loops + loops=$(( SECS / 2 + 2 )) # the fast log lasts 2 s; enough loops either way + broker=$(pgrep -x mosquitto); m0=$(ticks "$broker"); t0=$(date +%s.%N) + timeout "$((SECS + 1))" canplayer -l "$loops" -I "$log" vcan0=can_pcan >/dev/null 2>&1 || true + m1=$(ticks "$broker"); t1=$(date +%s.%N) + wait "${pids[@]}" + python3 - "$name" "$OUT" "$((m1 - m0))" "$HZ" "$t0" "$t1" <<'PY' +import glob, json, sys +name, out, ticks, hz, t0, t1 = sys.argv[1:] +load = lambda p: json.load(open(p)) +raw = [load(p) for p in sorted(glob.glob(f"{out}/raw*.json"))] +env = [load(p) for p in sorted(glob.glob(f"{out}/env*.json"))] +dec, br = load(f"{out}/decoder.json"), load(f"{out}/bridge.json") +worst = lambda xs, k: max(x["latency_ms"][k] for x in xs) +print(json.dumps({ + "run": name, + "bridge": {k: br[k] for k in ("batch_ms", "frames", "messages", "refused_frames", "kernel_drops", "cpu_pct")}, + "raw_sinks": {"n": len(raw), "frames_per_s": min(x["frames_per_s"] for x in raw), + "lost": sum(x["lost_frames"] for x in raw), + "p99_ms": worst(raw, "p99"), "max_ms": worst(raw, "max")}, + "decoder": {k: dec[k] for k in ("signals_per_frame", "max_hz", "topics", "published_per_s", "refused", "cpu_pct")}, + "envelope_sinks": {"n": len(env), "per_s": min(x["messages_per_s"] for x in env), + "p50_ms": worst(env, "p50"), "p99_ms": worst(env, "p99"), "max_ms": worst(env, "max"), + "cpu_pct_each": max(x["cpu_pct"] for x in env)}, + "mosquitto_cpu_pct": round(100 * int(ticks) / int(hz) / (float(t1) - float(t0)), 1)})) +PY +} + +CASES="${CASES:-10 3 0;10 8 0;10 8 20;0 8 0}" +for log in "$CAPTURE" "$FAST"; do + rate=$([ "$log" = "$FAST" ] && echo "full bus" || echo "car") + IFS=';' read -ra cases <<< "$CASES" + for c in "${cases[@]}"; do + read -r batch signals maxhz <<< "$c" + run "$rate, raw batch ${batch} ms, ${signals} signals/frame, cap ${maxhz} Hz" "$log" "$batch" "$signals" "$maxhz" + done +done diff --git a/src/bus-bench/main.rs b/src/bus-bench/main.rs index 9bd073f..cd37574 100644 --- a/src/bus-bench/main.rs +++ b/src/bus-bench/main.rs @@ -5,6 +5,13 @@ //! 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. //! +//! `decoder` stands in for the real decoder's load: it subscribes to the raw frames +//! and publishes `--signals` envelopes per frame, one topic per (CAN ID, signal), +//! retained, as the decoder will on `yukart/vehicle/`. `--max-hz` caps each +//! signal's rate, keeping the latest value. Its topics live under `yukart/bench/` +//! so svc-gw and the app never see them, and are cleared when it exits. A `sink` +//! on those topics measures envelopes instead of frames. +//! //! Test tool, not a service: the wire format below is provisional and lives only //! here until contracts freezes `yukart/can/raw`. //! @@ -36,7 +43,11 @@ usage: --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: + bus-bench decoder [options] yukart/can/raw → envelopes on yukart/bench/vehicle/ + --signals N signals per frame (3) + --max-hz H cap each signal's rate; 0 = every frame (0) + --secs N stop after N seconds (30) + all: --broker HOST:PORT (127.0.0.1:1883) --topic TOPIC (yukart/can/raw) --nagle leave Nagle's algorithm on (rumqttc's default)"; @@ -63,6 +74,8 @@ struct Opts { broker: (String, u16), topic: String, nagle: bool, + signals: u32, + max_hz: f64, } fn parse(mut args: impl Iterator) -> Result { @@ -73,6 +86,8 @@ fn parse(mut args: impl Iterator) -> Result { broker: ("127.0.0.1".into(), 1883), topic: "yukart/can/raw".into(), nagle: false, + signals: 3, + max_hz: 0.0, }; while let Some(a) = args.next() { let mut val = || args.next().ok_or(format!("{a} needs a value")); @@ -82,6 +97,8 @@ fn parse(mut args: impl Iterator) -> Result { "--secs" => o.secs = val()?.parse().map_err(|_| "--secs needs a number")?, "--topic" => o.topic = val()?, "--nagle" => o.nagle = true, + "--signals" => o.signals = val()?.parse().map_err(|_| "--signals needs a number")?, + "--max-hz" => o.max_hz = val()?.parse().map_err(|_| "--max-hz needs a number")?, "--broker" => { let v = val()?; let (h, p) = v.rsplit_once(':').ok_or("--broker must be HOST:PORT")?; @@ -231,6 +248,83 @@ fn decode(p: &[u8], now: i64) -> Option<(u64, u32, Vec)> { Some((seq, count, lat)) } +/// Every frame of one raw message: (ts_us, can_id, data bytes). +fn frames(p: &[u8]) -> Option)>> { + decode(p, 0)?; + Some( + p[HEADER..] + .chunks_exact(FRAME) + .map(|f| { + let len = (f[12] as usize).min(8); + (i64::from_le_bytes(f[0..8].try_into().unwrap()), u32::from_le_bytes(f[8..12].try_into().unwrap()), f[16..16 + len].to_vec()) + }) + .collect(), + ) +} + +fn mono_ns() -> i64 { + (clock::monotonic() * 1e9) as i64 +} + +fn decoder(o: Opts) -> ExitCode { + let (tx, rx) = sync_channel::>(1 << 14); + let on_publish = move |p: &[u8]| { + let _ = tx.try_send(p.to_vec()); + }; + let id = format!("bus-bench-decoder-{}", std::process::id()); + let Some(client) = connect(&id, &o, 1 << 14, Some(o.topic.clone()), on_publish) else { + eprintln!("bus-bench: no broker at {}:{}", o.broker.0, o.broker.1); + return ExitCode::FAILURE; + }; + let min_gap = if o.max_hz > 0.0 { Duration::from_secs_f64(1.0 / o.max_hz) } else { Duration::ZERO }; + let mut last: std::collections::HashMap = Default::default(); + let (mut in_frames, mut published, mut held, mut refused) = (0u64, 0u64, 0u64, 0u64); + let start = Instant::now(); + let cpu0 = cpu_secs(); + while start.elapsed() < Duration::from_secs(o.secs) { + let p = match rx.recv_timeout(Duration::from_millis(100)) { + Ok(p) => p, + Err(RecvTimeoutError::Timeout) => continue, + Err(RecvTimeoutError::Disconnected) => break, + }; + for (ts, can_id, data) in frames(&p).unwrap_or_default() { + in_frames += 1; + for k in 0..o.signals { + let name = format!("sig_{:03x}_{k}", can_id & libc::CAN_EFF_MASK); + let now = Instant::now(); + if let Some(t) = last.get(&name) { + if now.duration_since(*t) < min_gap { + held += 1; // the latest value wins; this one is never sent + continue; + } + } + last.insert(name.clone(), now); + let value = data.get(k as usize % data.len().max(1)).copied().unwrap_or(0) as f64 * 0.5; + let env = json!({ "timestamp": ts, "monotonic": mono_ns(), "value": value, "unit": "" }); + match client.try_publish(format!("yukart/bench/vehicle/{name}"), QoS::AtMostOnce, true, env.to_string()) { + Ok(()) => published += 1, + Err(_) => refused += 1, + } + } + } + } + let secs = start.elapsed().as_secs_f64(); + let cpu = (cpu_secs() - cpu0) / secs * 100.0; + // Leave no retained bench values behind. + for name in last.keys() { + let _ = client.try_publish(format!("yukart/bench/vehicle/{name}"), QoS::AtMostOnce, true, Vec::new()); + } + thread::sleep(Duration::from_millis(500)); + println!( + "{}", + json!({ "role": "decoder", "signals_per_frame": o.signals, "max_hz": o.max_hz, "secs": (secs * 10.0).round() / 10.0, + "frames_in": in_frames, "topics": last.len(), "published": published, + "published_per_s": (published as f64 / secs).round(), "held_by_cap": held, "refused": refused, + "cpu_pct": (cpu * 10.0).round() / 10.0 }) + ); + ExitCode::SUCCESS +} + fn pct(sorted: &[i64], q: f64) -> f64 { if sorted.is_empty() { return 0.0; @@ -243,13 +337,27 @@ 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())); + let raw = !topic.contains("/vehicle/"); + let on_publish = move |p: &[u8]| { + let got = if raw { + decode(p, now_us()) + } else { + // An envelope: no sequence numbers, so loss is counted against the decoder's report. + serde_json::from_slice::(p) + .ok() + .and_then(|v| v["timestamp"].as_i64()) + .map(|t| (0, 1, vec![now_us() - t])) + }; + match got { + Some((s, c, l)) => { + let _ = tx.try_send((s, c, l, p.len())); + } + None if p.is_empty() => {} // a retained value being cleared + None => eprintln!("bus-bench: malformed message ({} bytes)", 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 { + let id = format!("bus-bench-sink-{}", std::process::id()); // several sinks may run at once + let Some(_client) = connect(&id, &o, 16, Some(topic), on_publish) else { eprintln!("bus-bench: no broker at {}:{}", o.broker.0, o.broker.1); return ExitCode::FAILURE; }; @@ -272,6 +380,7 @@ fn sink(o: Opts) -> ExitCode { cpu0 = cpu_secs(); } match next { + _ if !raw => {} Some(want) if s > want => lost += s - want, Some(want) if s < want => reordered += 1, _ => {} @@ -292,7 +401,7 @@ fn sink(o: Opts) -> ExitCode { let r = |v: f64| (v * 100.0).round() / 100.0; println!( "{}", - json!({ "role": "sink", "secs": r(secs), "frames": frames, "messages": msgs, + json!({ "role": if raw { "sink" } else { "envelope 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)), @@ -317,6 +426,7 @@ fn main() -> ExitCode { match role.as_str() { "bridge" => bridge(o), "sink" => sink(o), + "decoder" => decoder(o), _ => { eprintln!("{USAGE}"); ExitCode::from(2)