From 17cd4a5ceb9d2fd7df0aee5f14e4a6535ffafc7c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 17 Jul 2026 19:36:04 +0000 Subject: [PATCH] =?UTF-8?q?feat(causal):=20load=5Fprofile=20example=20?= =?UTF-8?q?=E2=80=94=20external-loadgen=20causal=20target?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Server-only discovery tool: real hot path, no planted bottleneck, no in-process loadgen. wrk (or any external generator, pinned to other cores) drives GET /json/:id; the sweep runs the causal_bench plan over the five lib sites. Trust gates only (traffic flowed, ledger printed) — no known-answer verdict, this one asks the question instead. --- examples/load_profile.rs | 106 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 examples/load_profile.rs diff --git a/examples/load_profile.rs b/examples/load_profile.rs new file mode 100644 index 0000000..095f2e1 --- /dev/null +++ b/examples/load_profile.rs @@ -0,0 +1,106 @@ +//! Causal load-profiling target (RFC 007): the real urus hot path under +//! EXTERNAL load — no planted bottleneck, no in-process load generation. +//! +//! Where `causal_bench` validates the machinery against a planted, +//! known-answer store, this is the discovery tool: boot a bare server, +//! let an external generator (wrk, pinned to different cores) drive it, +//! sweep virtual speedups over the five lib sites (parse, router, +//! pipeline, serialize, socket-write), and report which one causally +//! limits throughput. External load is deliberate: the generator lives +//! outside the smarm runtime, so it cannot absorb injected virtual +//! delay — the same property `causal_bench` gets from plain OS threads. +//! +//! No pass/fail verdict — there is no known answer here. Trust gates +//! only: traffic actually flowed through the sweep, and the ledger +//! audit (printed) balances. +//! +//! Env: +//! URUS_PORT listen port (default 8080; binds 0.0.0.0) +//! COZ_OUT coz profile output path (default profile.coz) +//! WARMUP_MS settle time after first traffic, ms (default 2000) +//! +//! Orchestration contract (the jobrunner run.sh): start this, wait for +//! it to accept, point wrk at `GET /json/:id` with a duration that +//! outlives the sweep, and stop wrk when this process exits. +//! +//! cargo run --release --example load_profile --features smarm-causal + +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +use urus::{serve_with_shutdown, shutdown_handle, Config, Conn, Next, Pipeline, Router}; + +fn env_u64(name: &str, default: u64) -> u64 { + std::env::var(name).ok().and_then(|v| v.parse().ok()).unwrap_or(default) +} + +/// The whole handler: param parse + JSON render. Deliberately thin — the +/// subject is the lib path around it, not application work. +fn json_id(conn: Conn, _next: Next) -> Conn { + let id: u64 = conn.params.get("id").and_then(|s| s.parse().ok()).unwrap_or(0); + conn.put_status(200) + .put_header("content-type", "application/json") + .put_body(format!("{{\"id\":{id}}}")) +} + +fn main() { + let port = env_u64("URUS_PORT", 8080) as u16; + let warmup = Duration::from_millis(env_u64("WARMUP_MS", 2000)); + let coz_out = std::env::var("COZ_OUT").unwrap_or_else(|_| "profile.coz".into()); + + let addr: SocketAddr = format!("0.0.0.0:{port}").parse().expect("addr"); + let (handle, signal) = shutdown_handle(); + let server = std::thread::spawn(move || { + let pipe = Pipeline::new().plug(Router::new().get("/json/:id", json_id)); + serve_with_shutdown(Config::new(addr), pipe, signal).expect("serve"); + }); + + // Readiness is the orchestrator's job (TCP probe); ours is to not + // sweep before real traffic has registered every site and the + // progress point — `run_experiments` enumerates *registered* sites. + // 100 completed responses guarantees full end-to-end coverage. + let responses = |snap: &[(String, u64)]| { + snap.iter().find(|(n, _)| n == "responses").map(|(_, c)| *c).unwrap_or(0) + }; + let t0 = Instant::now(); + let seen = loop { + let n = responses(&smarm::causal::progress_snapshot()); + if n >= 100 { + break n; + } + assert!( + t0.elapsed() < Duration::from_secs(120), + "no load after 120s ({n} responses) — is the generator running?" + ); + std::thread::sleep(Duration::from_millis(100)); + }; + println!("traffic up: {seen} responses; settling {warmup:?}"); + std::thread::sleep(warmup); + + // Same plan as causal_bench, for comparability across workloads. + let results = smarm::causal::run_experiments(&smarm::causal::ExperimentPlan { + speedups_pct: vec![0, 25, 50], + experiment: Duration::from_millis(700), + cooldown: Duration::from_millis(150), + }); + + handle.shutdown(); + server.join().expect("server panicked"); + + print!("{}", smarm::causal::render_summary(&results)); + print!("{}", smarm::causal::render_ledger_audit(&results)); + match std::fs::write(&coz_out, smarm::causal::render_coz(&results)) { + Ok(()) => println!("\nwrote {coz_out}"), + Err(e) => eprintln!("\nfailed to write {coz_out}: {e}"), + } + + // Trust gate: the sweep is meaningless if load didn't flow through it. + let total: u64 = results + .iter() + .flat_map(|r| r.deltas.iter()) + .filter(|(n, _)| n == "responses") + .map(|(_, c)| c) + .sum(); + println!("total responses across windows: {total}"); + assert!(total > 0, "sweep saw zero progress"); +}