yukart/svc-core
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bus-bench: decoder stand-in and mixed-load runs

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz
This commit is contained in:
Mikkeli Matlock
2026-10-12 22:33:44 +09:00
parent bba48efd90
commit b605c27fc0
5 changed files with 256 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bus-bench/main.rs
+117 -7
View File
@@ -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/<signal>`. `--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/<signal>
--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<Item = String>) -> Result<Opts, String> {
@@ -73,6 +86,8 @@ fn parse(mut args: impl Iterator<Item = String>) -> Result<Opts, String> {
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<Item = String>) -> Result<Opts, String> {
"--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<i64>)> {
Some((seq, count, lat))
}
/// Every frame of one raw message: (ts_us, can_id, data bytes).
fn frames(p: &[u8]) -> Option<Vec<(i64, u32, Vec<u8>)>> {
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::<Vec<u8>>(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<String, Instant> = 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<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()));
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::<serde_json::Value>(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)