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(); } }