From cf757af5107f0c4f52af6ae06924eb36d6abb2ee Mon Sep 17 00:00:00 2001 From: acamilo Date: Tue, 22 Sep 2026 12:01:43 +0000 Subject: [PATCH] test(flybus): assert the example's output, and split the frame measurement up The guide's first deliverable is one small program with a counter RPC, a pub/sub observer and a frame artifact held past its message object. examples/demo.rs was already that; its body moves into run(), which returns the lines it prints, so tests/example_demo.rs can assert all seven of them. The counter now uses the spec's own example.counter / Counter.Increment / {"amount": 1} and pins the registration it discovered, and the example prints the router's root count while the frame is held and after the last handle goes, so the lifetime it demonstrates is visible rather than implied. cargo run -p flybus --example demo prints the same lines. tests/perf.rs reports what BUS-03 and bus-v1 section 11.7 actually ask for. Allocate, the producer's copy into staging, the router's seal copy, publish admission, the RPC round trip and a consumer's readback are six separate percentile lines instead of one. The router gets its own two-thread runtime whose threads carry a distinct name, and per-thread CPU is sampled from /proc by that name, so router CPU is separable from the clients' in the same process. Store bytes, outstanding roots and queue lengths are reported live as well as peak. --- .../flysim/crates/flybus/examples/demo.rs | 103 ++++++-- .../crates/flybus/tests/example_demo.rs | 26 ++ services/flysim/crates/flybus/tests/perf.rs | 235 ++++++++++++++---- 3 files changed, 284 insertions(+), 80 deletions(-) create mode 100644 services/flysim/crates/flybus/tests/example_demo.rs diff --git a/services/flysim/crates/flybus/examples/demo.rs b/services/flysim/crates/flybus/examples/demo.rs index 1e124eb..12830de 100644 --- a/services/flysim/crates/flybus/examples/demo.rs +++ b/services/flysim/crates/flybus/examples/demo.rs @@ -1,24 +1,38 @@ -//! A counter RPC, a pub/sub observer and a frame artifact held past its message, in one -//! process over the in-memory transport (bus-v1 section 11). +//! The guide's first deliverable: a counter RPC, a pub/sub observer and a frame artifact held +//! past its message object's lifetime, in one program (bus-v1 section 11, implementation +//! guide section 1). No game, browser or second transport is involved. //! //! ```text //! cargo run -p flybus --example demo //! ``` +//! +//! `tests/example_demo.rs` runs [`run`] and asserts every line it returns. use std::io::Write; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; use flybus::{ - Client, ClientConfig, Policy, Retained, Router, RouterConfig, ServiceConfig, SubscriptionConfig, + Client, ClientConfig, Policy, Retained, Router, RouterConfig, ServiceConfig, + SubscriptionConfig, }; use serde_json::{Map, Value, json}; +const W: usize = 160; +const H: usize = 144; + fn obj(v: Value) -> Map { v.as_object().cloned().unwrap_or_default() } -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<(), Box> { - let root = std::env::temp_dir().join(format!("flybus-demo-{}", std::process::id())); +/// The three parts, in one program, over one router. Returns the lines the example prints. +pub async fn run() -> Result, Box> { + static RUNS: AtomicU64 = AtomicU64::new(0); + let root = std::env::temp_dir().join(format!( + "flybus-demo-{}-{}", + std::process::id(), + RUNS.fetch_add(1, Ordering::Relaxed) + )); let mut config = RouterConfig::new(&root); config.policy = Policy::open(); let router = Router::new(config)?; @@ -28,14 +42,17 @@ async fn main() -> Result<(), Box> { ClientConfig::new(id, &root), ) }; + let mut lines = Vec::new(); - // A counter service. + // 1. A counter service. An exclusive endpoint, pinned by its caller to the registration + // it discovered, reached through the router like every other operation. let counter = connect("counter").await?; let mut svc = counter .register("example.counter", ServiceConfig::default()) .await?; - tokio::spawn(async move { - let mut total = 0; + let incarnation = svc.incarnation().to_owned(); + let service = tokio::spawn(async move { + let mut total = 0i64; while let Some(req) = svc.next().await { total += req.payload()["amount"].as_i64().unwrap_or(0); let _ = req.reply(obj(json!({ "total": total })), &[]).await; @@ -46,54 +63,86 @@ async fn main() -> Result<(), Box> { let res = app .call_and_wait( "example.counter", - None, + Some(&incarnation), "Counter.Increment", - obj(json!({"amount": 2})), + obj(json!({"amount": 1})), &[], ) .await?; - println!("counter total = {}", res.outcome()["total"]); + lines.push(format!("counter total = {}", res.outcome()["total"])); } - // An observer of a frame topic. - app.declare_topic("world.demo.frame", Retained::None) - .await?; + // 2. A pub/sub observer. A latest-value subscription, so a slow observer coalesces + // instead of holding the producer up. + app.declare_topic("world.demo.frame", Retained::None).await?; let observer = connect("observer").await?; let mut frames = observer .subscribe("world.demo.frame", SubscriptionConfig::latest()) .await?; + // 3. A frame artifact. The bytes live in the store; the message carries a reference and + // the dimensions. let mut writer = app .artifacts() - .allocate(160 * 144 * 4, "image/x-rgba") + .allocate((W * H * 4) as u64, "image/x-rgba") .await?; - writer.write_all(&vec![0x7f; 160 * 144 * 4])?; + writer.write_all(&vec![0x7f; W * H * 4])?; let frame = writer.seal().await?; let receipt = app .publish( "world.demo.frame", - obj(json!({"width": 160, "height": 144})), + obj(json!({"width": W, "height": H})), &[("frame", &frame)], ) .await?; - println!( + lines.push(format!( "published sequence {} to {} subscriber(s)", receipt.topic_sequence, receipt.subscribers - ); + )); + // The producer lets go of its own hold; the delivery keeps the bytes alive. drop(frame); - let message = frames.next().await.ok_or("subscription closed")?; + let message = frames.next().await.ok_or("the subscription closed")?; let image = message.artifact("frame")?; drop(message); // the extracted handle still owns the delivery let bytes = image.read_all().await?; - println!( - "read {} bytes after the message was dropped; router: {:?}", - bytes.len(), - router.stats() - ); + lines.push(format!( + "read {} bytes after the message was dropped", + bytes.len() + )); + let held = router.stats(); + lines.push(format!( + "while the frame is held: {} artifact(s), {} root(s)", + held.sealed_artifacts, held.artifact_roots + )); drop(image); // the last handle: the delivery is consumed and the frame collected + // Consumption reaches the router on the client's control lane, so collection is not + // instantaneous. + let deadline = Instant::now() + Duration::from_secs(10); + while router.stats().artifacts > 0 { + if Instant::now() > deadline { + return Err("the frame was never collected".into()); + } + tokio::time::sleep(Duration::from_millis(1)).await; + } + let collected = router.stats(); + lines.push(format!( + "after the last handle: {} artifact(s), {} root(s)", + collected.artifacts, collected.artifact_roots + )); + + service.abort(); router.shutdown(); - std::fs::remove_dir_all(&root)?; + drop((app, observer, counter)); + let _ = std::fs::remove_dir_all(&root); + Ok(lines) +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Box> { + for line in run().await? { + println!("{line}"); + } Ok(()) } diff --git a/services/flysim/crates/flybus/tests/example_demo.rs b/services/flysim/crates/flybus/tests/example_demo.rs new file mode 100644 index 0000000..1bb1285 --- /dev/null +++ b/services/flysim/crates/flybus/tests/example_demo.rs @@ -0,0 +1,26 @@ +//! The guide's example is also a test: `cargo run -p flybus --example demo` prints exactly +//! these lines (bus-v1 section 11, implementation guide section 1). + +#[allow(dead_code)] +#[path = "../examples/demo.rs"] +mod demo; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn the_example_shows_a_counter_rpc_an_observer_and_a_held_frame() { + let lines = demo::run().await.expect("the example ran"); + assert_eq!( + lines.iter().map(String::as_str).collect::>(), + vec![ + // A counter service, called three times through the router. + "counter total = 1", + "counter total = 2", + "counter total = 3", + // One observer, one accepted publication, one sequence number. + "published sequence 1 to 1 subscriber(s)", + // 160x144 RGBA, read after the message object was dropped. + "read 92160 bytes after the message was dropped", + "while the frame is held: 1 artifact(s), 1 root(s)", + "after the last handle: 0 artifact(s), 0 root(s)", + ] + ); +} diff --git a/services/flysim/crates/flybus/tests/perf.rs b/services/flysim/crates/flybus/tests/perf.rs index aa2e4ff..7a47041 100644 --- a/services/flysim/crates/flybus/tests/perf.rs +++ b/services/flysim/crates/flybus/tests/perf.rs @@ -1,7 +1,11 @@ -//! bus-v1 section 11 item 7, as a measurement rather than a gate: 640x480 RGBA frames at -//! 60 Hz over a Unix socket to three consumers (one delayed), with 1, 2 and 4 agent services -//! pinged every frame. Router and clients share this process, so CPU and RSS are the whole -//! process. Run with: +//! bus-v1 section 11 item 7 and implementation-guide BUS-03, as a measurement rather than a +//! gate: 640x480 RGBA frames at 60 Hz over a Unix socket to three latest-mode consumers (one +//! delayed 40 ms per frame), with 1, 2 and 4 agent services called every frame. +//! +//! The router runs on its own Tokio runtime whose threads carry a distinct name, so its CPU +//! (routing plus the seal copies on its blocking pool) is measured apart from the clients'. +//! Producer copy cost and consumer readback cost are measured separately from routing. Nothing +//! here is a capacity claim: one host, one process, synthetic payloads. //! //! ```text //! cargo test --release -p flybus --test perf -- --ignored --nocapture @@ -10,24 +14,55 @@ mod common; use std::io::Write; +use std::path::PathBuf; use std::time::{Duration, Instant}; -use common::{Via, env, obj}; -use flybus::{Client, Retained, ServiceConfig, SubscriptionConfig}; +use common::obj; +use flybus::{ + Client, ClientConfig, Policy, Retained, Router, RouterConfig, ServiceConfig, + SubscriptionConfig, +}; use serde_json::json; const W: usize = 640; const H: usize = 480; const HZ: u64 = 60; const SECONDS: u64 = 2; +const ROUTER_THREAD: &str = "flybus-router"; +const ROUTER_WORKERS: usize = 2; +const CLIENT_WORKERS: usize = 4; +/// utime+stime of this process, in seconds (fields 14 and 15 of /proc/self/stat, 100 Hz). fn proc_cpu_seconds() -> f64 { - // utime + stime, fields 14 and 15 of /proc/self/stat, in clock ticks (100 Hz on Linux). - let stat = std::fs::read_to_string("/proc/self/stat").unwrap_or_default(); - let after = stat.rsplit_once(')').map_or("", |(_, rest)| rest); - let f: Vec<&str> = after.split_whitespace().collect(); - let ticks = |i: usize| f.get(i).and_then(|v| v.parse::().ok()).unwrap_or(0.0); - (ticks(11) + ticks(12)) / 100.0 + thread_cpu_seconds(None) +} + +/// utime+stime of the threads whose name matches, in seconds; all of them when `name` is +/// `None`. A thread that exits between two samples takes its time with it, so this is a floor +/// for pools that retire idle threads. +fn thread_cpu_seconds(name: Option<&str>) -> f64 { + let mut total = 0.0; + let Ok(dir) = std::fs::read_dir("/proc/self/task") else { + return 0.0; + }; + for entry in dir.flatten() { + let Ok(stat) = std::fs::read_to_string(entry.path().join("stat")) else { + continue; + }; + let Some((head, rest)) = stat.rsplit_once(')') else { + continue; + }; + if let Some(want) = name { + let comm = head.split_once('(').map_or("", |(_, c)| c); + if comm != want { + continue; + } + } + let f: Vec<&str> = rest.split_whitespace().collect(); + let ticks = |i: usize| f.get(i).and_then(|v| v.parse::().ok()).unwrap_or(0.0); + total += (ticks(11) + ticks(12)) / 100.0; + } + total } fn proc_status(key: &str) -> String { @@ -45,12 +80,86 @@ fn pct(sorted: &[Duration], p: f64) -> Duration { sorted[((sorted.len() - 1) as f64 * p).round() as usize] } -async fn consumer(client: Client, delay: Duration) -> (u64, u64) { +fn ms(d: Duration) -> String { + format!("{:.2}", d.as_secs_f64() * 1000.0) +} + +fn percentiles(label: &str, v: &mut [Duration]) -> String { + v.sort(); + format!( + "{label} ms p50/p95/p99: {}/{}/{}", + ms(pct(v, 0.5)), + ms(pct(v, 0.95)), + ms(pct(v, 0.99)) + ) +} + +/// The router on its own runtime, reached over a Unix socket. +struct Host { + router: Router, + socket: PathBuf, + store_root: PathBuf, + rt: Option, + _dir: tempfile::TempDir, +} + +impl Host { + fn start() -> Host { + let dir = tempfile::tempdir().unwrap(); + let store_root = dir.path().join("store"); + let socket = dir.path().join("bus.sock"); + let mut config = RouterConfig::new(&store_root); + config.policy = Policy::open(); + let router = Router::new(config).unwrap(); + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(ROUTER_WORKERS) + .thread_name(ROUTER_THREAD) + .enable_all() + .build() + .unwrap(); + // The listener, and so every connection task, belongs to the router's runtime. + let (ready, started) = std::sync::mpsc::channel(); + let (r, s) = (router.clone(), socket.clone()); + rt.spawn(async move { + let _listener = r.listen_unix(&s).await.expect("the router listens"); + ready.send(()).expect("start() is waiting"); + std::future::pending::<()>().await + }); + started.recv().expect("the router runtime started its listener"); + Host { + router, + socket, + store_root, + rt: Some(rt), + _dir: dir, + } + } + + async fn client(&self, id: &str) -> Client { + Client::connect_unix(&self.socket, ClientConfig::new(id, &self.store_root)) + .await + .unwrap() + } + + fn stop(&mut self) { + self.router.shutdown(); + if let Some(rt) = self.rt.take() { + // A runtime cannot be dropped from inside another one. + std::thread::spawn(move || rt.shutdown_timeout(Duration::from_secs(2))) + .join() + .expect("the router runtime stopped"); + } + } +} + +/// A latest-mode consumer: extracts the frame, drops the message, reads the bytes and then +/// takes `delay` to "render" them. Returns (frames seen, coalesced, readback times). +async fn consumer(client: Client, delay: Duration) -> (u64, u64, Vec) { let mut sub = client .subscribe("world.demo.frame", SubscriptionConfig::latest()) .await .unwrap(); - let (mut seen, mut replaced) = (0, 0); + let (mut seen, mut replaced, mut readback) = (0, 0, Vec::new()); while let Some(m) = sub.next().await { if m.payload().get("end").is_some() { break; @@ -58,29 +167,30 @@ async fn consumer(client: Client, delay: Duration) -> (u64, u64) { replaced += m.replaced(); let frame = m.artifact("frame").unwrap(); drop(m); + let t = Instant::now(); let bytes = frame.read_all().await.unwrap(); + readback.push(t.elapsed()); assert_eq!(bytes.len(), W * H * 4); tokio::time::sleep(delay).await; seen += 1; } - (seen, replaced) + (seen, replaced, readback) } -async fn run(agents: usize) { - let e = env(Via::Unix).await; - let producer = e.client("producer").await; +async fn run(host: &Host, agents: usize) { + let producer = host.client("producer").await; producer .declare_topic("world.demo.frame", Retained::None) .await .unwrap(); let mut consumers = Vec::new(); for (i, delay) in [0u64, 0, 40].into_iter().enumerate() { - let c = e.client(&format!("consumer-{i}")).await; + let c = host.client(&format!("consumer-{i}")).await; consumers.push(tokio::spawn(consumer(c, Duration::from_millis(delay)))); } let mut services = Vec::new(); for k in 0..agents { - let c = e.client(&format!("agent-{k}")).await; + let c = host.client(&format!("agent-{k}")).await; let mut svc = c .register(&format!("agent.a{k}"), ServiceConfig::default()) .await @@ -92,16 +202,20 @@ async fn run(agents: usize) { } })); } - let caller = e.client("coordinator").await; + let caller = host.client("coordinator").await; // Let every subscription land before the first frame. - e.settle("subscribed", |s| s.subscriptions == 3).await; + while host.router.stats().subscriptions != 3 { + tokio::time::sleep(Duration::from_millis(5)).await; + } let pixels: Vec = (0..W * H * 4).map(|i| (i % 253) as u8).collect(); let frames = HZ * SECONDS; let period = Duration::from_nanos(1_000_000_000 / HZ); - let (mut produce, mut publish, mut rpc) = (Vec::new(), Vec::new(), Vec::new()); + let (mut allocate, mut copy, mut seal) = (Vec::new(), Vec::new(), Vec::new()); + let (mut publish, mut rpc) = (Vec::new(), Vec::new()); let (mut peak_bytes, mut peak_roots, mut peak_queued, mut late) = (0u64, 0u64, 0usize, 0u32); let cpu0 = proc_cpu_seconds(); + let router_cpu0 = thread_cpu_seconds(Some(ROUTER_THREAD)); let start = Instant::now(); for n in 0..frames { let deadline = start + period * n as u32; @@ -111,9 +225,13 @@ async fn run(agents: usize) { .allocate(pixels.len() as u64, "image/x-rgba") .await .unwrap(); + allocate.push(t.elapsed()); + let t = Instant::now(); w.write_all(&pixels).unwrap(); + copy.push(t.elapsed()); + let t = Instant::now(); let frame = w.seal().await.unwrap(); - produce.push(t.elapsed()); + seal.push(t.elapsed()); let t = Instant::now(); producer .publish( @@ -140,7 +258,7 @@ async fn run(agents: usize) { for c in calls { rpc.push(c.await.unwrap()); } - let s = e.stats(); + let s = host.router.stats(); peak_bytes = peak_bytes.max(s.store_bytes); peak_roots = peak_roots.max(s.artifact_roots); peak_queued = peak_queued.max(s.queued); @@ -153,61 +271,72 @@ async fn run(agents: usize) { } let wall = start.elapsed().as_secs_f64(); let cpu = proc_cpu_seconds() - cpu0; + let router_cpu = thread_cpu_seconds(Some(ROUTER_THREAD)) - router_cpu0; let end = Instant::now(); producer .publish("world.demo.frame", obj(json!({"end": true})), &[]) .await .unwrap(); let mut results = Vec::new(); + let mut readback = Vec::new(); for c in consumers { - results.push(c.await.unwrap()); + let (seen, replaced, mut times) = c.await.unwrap(); + readback.append(&mut times); + results.push((seen, replaced)); + } + while host.router.stats().store_bytes != 0 { + assert!( + end.elapsed() < Duration::from_secs(10), + "the store never drained: {:?}", + host.router.stats() + ); + tokio::time::sleep(Duration::from_millis(1)).await; } - e.settle("collected", |s| s.store_bytes == 0).await; let collect_lag = end.elapsed(); + let live = host.router.stats(); for s in services { s.abort(); } - for v in [&mut produce, &mut publish, &mut rpc] { - v.sort(); - } - let ms = |d: Duration| format!("{:.2}", d.as_secs_f64() * 1000.0); + println!("agents={agents} frames={frames} over {wall:.2}s, late frames {late}"); + println!(" {}", percentiles("allocate (quota + staging file)", &mut allocate)); + println!(" {}", percentiles("producer copy into staging", &mut copy)); + println!(" {}", percentiles("seal (router copy to a sealed inode)", &mut seal)); + println!(" {}", percentiles("publish admission", &mut publish)); + println!(" {}", percentiles("rpc round trip", &mut rpc)); + println!(" {}", percentiles("consumer readback of 1.2 MB", &mut readback)); println!( - " produce (allocate+write+seal copy) ms p50/p95/p99: {}/{}/{}", - ms(pct(&produce, 0.5)), - ms(pct(&produce, 0.95)), - ms(pct(&produce, 0.99)) - ); - println!( - " publish admission ms p50/p95/p99: {}/{}/{}", - ms(pct(&publish, 0.5)), - ms(pct(&publish, 0.95)), - ms(pct(&publish, 0.99)) - ); - println!( - " rpc round trip ms p50/p95/p99: {}/{}/{}", - ms(pct(&rpc, 0.5)), - ms(pct(&rpc, 0.95)), - ms(pct(&rpc, 0.99)) - ); - println!( - " process cpu {:.2} cores; VmRSS {} VmHWM {}", + " cpu cores: router {:.3} of {ROUTER_WORKERS} threads, whole process {:.3} of {} threads on {} cpus", + router_cpu / wall, cpu / wall, + ROUTER_WORKERS + CLIENT_WORKERS, + std::thread::available_parallelism().map_or(0, |n| n.get()) + ); + println!( + " VmRSS {} VmHWM {}", proc_status("VmRSS:"), proc_status("VmHWM:") ); println!( - " store peak {:.1} MB, peak roots {peak_roots}, peak queued {peak_queued}, drain+collect {:.1} ms", + " store peak {:.1} MB, live after drain {} B; roots peak {peak_roots}, live {}; queued peak {peak_queued}, live {}", peak_bytes as f64 / 1e6, + live.store_bytes, + live.artifact_roots, + live.queued + ); + println!( + " collection lag after the last frame {:.1} ms", collect_lag.as_secs_f64() * 1000.0 ); - println!(" consumers (frames seen, replaced): {results:?}"); + println!(" consumers (frames seen, coalesced): {results:?}"); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[ignore = "measurement; run with --release -- --ignored --nocapture"] async fn frames_at_60hz_with_three_consumers() { for agents in [1, 2, 4] { - run(agents).await; + let mut host = Host::start(); + run(&host, agents).await; + host.stop(); } }