feat(causal): load_profile example — external-loadgen causal target
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.
This commit is contained in:
@@ -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");
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user