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.

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
+//! The guide's first deliverable: a counter RPC, a pub/sub observer and a frame artifact held
-//! process over the in-memory transport (bus-v1 section 11).
+//! 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<String, Value> {
     v.as_object().cloned().unwrap_or_default()
 }
 
-#[tokio::main(flavor = "current_thread")]
+/// The three parts, in one program, over one router. Returns the lines the example prints.
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
+pub async fn run() -> Result<Vec<String>, Box<dyn std::error::Error>> {
-    let root = std::env::temp_dir().join(format!("flybus-demo-{}", std::process::id()));
+    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<dyn std::error::Error>> {
             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 incarnation = svc.incarnation().to_owned();
-        let mut total = 0;
+    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<dyn std::error::Error>> {
         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.
+    // 2. A pub/sub observer. A latest-value subscription, so a slow observer coalesces
-    app.declare_topic("world.demo.frame", Retained::None)
+    //    instead of holding the producer up.
-        .await?;
+    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!(
+    lines.push(format!(
-        "read {} bytes after the message was dropped; router: {:?}",
+        "read {} bytes after the message was dropped",
-        bytes.len(),
+        bytes.len()
-        router.stats()
+    ));
-    );
+    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<dyn std::error::Error>> {
+    for line in run().await? {
+        println!("{line}");
+    }
     Ok(())
 }

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<_>>(),
+        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)",
+        ]
+    );
+}

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
+//! bus-v1 section 11 item 7 and implementation-guide BUS-03, as a measurement rather than a
-//! 60 Hz over a Unix socket to three consumers (one delayed), with 1, 2 and 4 agent services
+//! gate: 640x480 RGBA frames at 60 Hz over a Unix socket to three latest-mode consumers (one
-//! pinged every frame. Router and clients share this process, so CPU and RSS are the whole
+//! delayed 40 ms per frame), with 1, 2 and 4 agent services called every frame.
-//! process. Run with:
+//!
+//! 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 common::obj;
-use flybus::{Client, Retained, ServiceConfig, SubscriptionConfig};
+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).
+    thread_cpu_seconds(None)
-    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();
+/// utime+stime of the threads whose name matches, in seconds; all of them when `name` is
-    let ticks = |i: usize| f.get(i).and_then(|v| v.parse::<f64>().ok()).unwrap_or(0.0);
+/// `None`. A thread that exits between two samples takes its time with it, so this is a floor
-    (ticks(11) + ticks(12)) / 100.0
+/// 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::<f64>().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<tokio::runtime::Runtime>,
+    _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<Duration>) {
     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) {
+async fn run(host: &Host, agents: usize) {
-    let e = env(Via::Unix).await;
+    let producer = host.client("producer").await;
-    let producer = e.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<u8> = (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: {}/{}/{}",
+        "  cpu cores: router {:.3} of {ROUTER_WORKERS} threads, whole process {:.3} of {} threads on {} cpus",
-        ms(pct(&produce, 0.5)),
+        router_cpu / wall,
-        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 / 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();
     }
 }