Compare commits
10
Commits
86c8d31b93
...
34730930e7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
34730930e7 | ||
|
|
72064c7a79 | ||
|
|
14be21db15 | ||
|
|
182f4fe602 | ||
|
|
17cd4a5ceb | ||
|
|
288b52d89c | ||
|
|
ecaddc579a | ||
|
|
072ee126f9 | ||
|
|
0f824635d1 | ||
|
|
078072b527 |
@@ -18,6 +18,7 @@ serde_json = { version = "1", optional = true }
|
|||||||
|
|
||||||
[features]
|
[features]
|
||||||
smarm-trace = ["smarm/smarm-trace"]
|
smarm-trace = ["smarm/smarm-trace"]
|
||||||
|
smarm-causal = ["smarm/smarm-causal"]
|
||||||
channels = []
|
channels = []
|
||||||
phoenix = ["channels", "dep:serde", "dep:serde_json"]
|
phoenix = ["channels", "dep:serde", "dep:serde_json"]
|
||||||
|
|
||||||
@@ -25,6 +26,10 @@ phoenix = ["channels", "dep:serde", "dep:serde_json"]
|
|||||||
serde = { version = "1", features = ["derive"] }
|
serde = { version = "1", features = ["derive"] }
|
||||||
serde_json = "1"
|
serde_json = "1"
|
||||||
|
|
||||||
|
[[example]]
|
||||||
|
name = "causal_bench"
|
||||||
|
required-features = ["smarm-causal"]
|
||||||
|
|
||||||
[profile.dev]
|
[profile.dev]
|
||||||
panic = "unwind"
|
panic = "unwind"
|
||||||
|
|
||||||
|
|||||||
@@ -378,3 +378,9 @@ reconnect cycles.
|
|||||||
- **Bench suite** — `urus-bench-spec.md` exists in the artefact store;
|
- **Bench suite** — `urus-bench-spec.md` exists in the artefact store;
|
||||||
wire it up once v0.2 lands (supervision changes the hot path not at all,
|
wire it up once v0.2 lands (supervision changes the hot path not at all,
|
||||||
but prove it).
|
but prove it).
|
||||||
|
- **Operator introspection via typed names** — the v0.2-era
|
||||||
|
`urus.server` / `urus.listener.{i}` pid tags were dropped in the
|
||||||
|
RFC 014 port (names are messageable endpoints now, self-registered
|
||||||
|
with a real `Sender<M>`). If wanted back, do it properly: register the
|
||||||
|
serve loop's shutdown/control channel under a typed `urus.server`
|
||||||
|
name instead of faking pid tags with unit channels.
|
||||||
|
|||||||
@@ -0,0 +1,322 @@
|
|||||||
|
//! Causal-profiling bench (RFC 007): a real urus webserver as the workload,
|
||||||
|
//! with a planted, known-answer bottleneck.
|
||||||
|
//!
|
||||||
|
//! Where smarm's `causal_pipeline` demo is synthetic, this exercises the
|
||||||
|
//! exact machinery the causal commits touched, for real:
|
||||||
|
//!
|
||||||
|
//! - epoll parks (`wait_readable_timeout` / `wait_writable_timeout`)
|
||||||
|
//! → park-gated resume credit
|
||||||
|
//! - keep-alive / request / write deadlines → timer-heap virtual time
|
||||||
|
//! - the experiment controller's fixed windows → wall-anchored timers
|
||||||
|
//! - mixed CPU (parse/serialize) and IO (socket) → non-trivial ranking
|
||||||
|
//!
|
||||||
|
//! Shape: `LOAD_CONNS` plain OS threads hammer `GET /order/:id` over
|
||||||
|
//! blocking loopback TCP (keep-alive). OS threads on the client side are
|
||||||
|
//! deliberate: they can't absorb injected virtual delay, so the only
|
||||||
|
//! delayable code is the server's — the same trick the smarm-side timer
|
||||||
|
//! tests use. The handler calls a single store actor (crud's pattern:
|
||||||
|
//! one actor owns the data, serialization is structural) which burns
|
||||||
|
//! `STORE_US` of calibrated *work* — work-shaped, not timed, because a
|
||||||
|
//! timed wait absorbs injected delay and flattens every experiment.
|
||||||
|
//!
|
||||||
|
//! Known answer: the store is serialized and saturated, so it is the
|
||||||
|
//! throughput ceiling (1e6/STORE_US rps). A virtual speedup of `store`
|
||||||
|
//! by p% predicts throughput ×1/(1-p): +33% @25%, +100% @50%. The other
|
||||||
|
//! five sites (parse, router, pipeline, serialize, socket-write — the
|
||||||
|
//! urus lib instrumentation) are tens of µs and parallel across
|
||||||
|
//! connections: predicted impact ≈0. Per the smarm-side validation runs,
|
||||||
|
//! absolute magnitudes are trustworthy to roughly ±15% while the
|
||||||
|
//! *ranking* is solid — the verdict thresholds reflect that.
|
||||||
|
//!
|
||||||
|
//! Run (needs cores; the verdict is SKIPPED below 4):
|
||||||
|
//! cargo run --release --example causal_bench --features smarm-causal
|
||||||
|
//!
|
||||||
|
//! Prints the summary, writes `profile.coz` next to the CWD, and exits
|
||||||
|
//! nonzero if the expected separation doesn't hold.
|
||||||
|
|
||||||
|
use std::io::{Read, Write};
|
||||||
|
use std::net::{SocketAddr, TcpListener, TcpStream};
|
||||||
|
use std::sync::atomic::{AtomicBool, Ordering};
|
||||||
|
use std::sync::{Arc, OnceLock};
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
use smarm::{channel, Sender};
|
||||||
|
use urus::{serve_with_shutdown, shutdown_handle, Config, Conn, Next, Pipeline, Router};
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Tunables
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
/// Planted bottleneck cost: calibrated work per store request, µs.
|
||||||
|
const STORE_US: u64 = 400;
|
||||||
|
/// Handler-side render work per request, µs. Runs under the `pipeline`
|
||||||
|
/// site (no inner guard), parallel across connection actors.
|
||||||
|
const RENDER_US: u64 = 50;
|
||||||
|
/// Client connections = OS load threads. Plenty to keep a 400µs store
|
||||||
|
/// saturated even at a 50% virtual speedup (see saturation note in main).
|
||||||
|
const LOAD_CONNS: usize = 16;
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Calibrated work (lifted from smarm's causal_pipeline demo)
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
/// LCG-mix `iters` times in dependent sequence (unvectorizable, un-elidable),
|
||||||
|
/// staying preemptible — and causal-sampleable/delayable — via `check!()`.
|
||||||
|
fn work_iters(iters: u64) {
|
||||||
|
let mut acc = 0x2545_f491_4f6c_dd1du64;
|
||||||
|
let mut i = 0u64;
|
||||||
|
while i < iters {
|
||||||
|
let chunk_end = (i + 256).min(iters);
|
||||||
|
while i < chunk_end {
|
||||||
|
acc = acc.wrapping_mul(6364136223846793005).wrapping_add(i);
|
||||||
|
i += 1;
|
||||||
|
}
|
||||||
|
std::hint::black_box(acc);
|
||||||
|
smarm::check!();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Iterations per microsecond on this machine, so stage costs are
|
||||||
|
/// meaningful in time while staying work-shaped.
|
||||||
|
fn calibrate_iters_per_us() -> u64 {
|
||||||
|
let n = 8_000_000u64;
|
||||||
|
let t = Instant::now();
|
||||||
|
work_iters(n);
|
||||||
|
(n / (t.elapsed().as_micros().max(1) as u64)).max(1)
|
||||||
|
}
|
||||||
|
|
||||||
|
static PER_US: OnceLock<u64> = OnceLock::new();
|
||||||
|
|
||||||
|
fn work_us(us: u64) {
|
||||||
|
work_iters(us * PER_US.get().copied().unwrap_or(1));
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Store actor — the planted bottleneck (crud's once-cell pattern)
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
enum Req {
|
||||||
|
Get { id: u64, reply: Sender<u64> },
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Set at teardown; the store polls it (a cross-thread `Sender::send`
|
||||||
|
/// can't wake a parked actor, so a bare `recv` would hang AllDone —
|
||||||
|
/// same limitation crud's store works around).
|
||||||
|
static STORE_STOP: AtomicBool = AtomicBool::new(false);
|
||||||
|
|
||||||
|
static STORE_TX: OnceLock<Sender<Req>> = OnceLock::new();
|
||||||
|
|
||||||
|
fn store_loop(rx: smarm::Receiver<Req>) {
|
||||||
|
loop {
|
||||||
|
let req = match rx.recv_timeout(Duration::from_millis(250)) {
|
||||||
|
Ok(r) => r,
|
||||||
|
Err(smarm::channel::RecvTimeoutError::Timeout) => {
|
||||||
|
if STORE_STOP.load(Ordering::Relaxed) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
Err(smarm::channel::RecvTimeoutError::Disconnected) => return,
|
||||||
|
};
|
||||||
|
let Req::Get { id, reply } = req;
|
||||||
|
{
|
||||||
|
// The known-answer site: serialized by construction (one
|
||||||
|
// actor, one request at a time), saturated by the load.
|
||||||
|
let _g = smarm::causal_site!("store");
|
||||||
|
work_us(STORE_US);
|
||||||
|
}
|
||||||
|
let _ = reply.send(id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Spawned from the first handler that runs — connection actors are
|
||||||
|
/// inside the runtime, which `spawn` requires; the static then shares
|
||||||
|
/// the Sender with every later handler.
|
||||||
|
fn store() -> &'static Sender<Req> {
|
||||||
|
STORE_TX.get_or_init(|| {
|
||||||
|
let (tx, rx) = channel::<Req>();
|
||||||
|
smarm::spawn(move || store_loop(rx));
|
||||||
|
tx
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Handler
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
fn order(conn: Conn, _next: Next) -> Conn {
|
||||||
|
let id: u64 = conn.params.get("id").and_then(|s| s.parse().ok()).unwrap_or(0);
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
store().send(Req::Get { id, reply: tx }).ok();
|
||||||
|
let got = rx.recv().expect("store dropped");
|
||||||
|
// Render work: attributed to the enclosing `pipeline` site.
|
||||||
|
work_us(RENDER_US);
|
||||||
|
conn.put_status(200).put_body(got.to_string())
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Load generation — plain OS threads, blocking loopback TCP, keep-alive
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
/// Read one HTTP/1.1 response (headers + content-length body) into `buf`,
|
||||||
|
/// draining exactly what was consumed. Errors mean: reconnect.
|
||||||
|
fn read_response(s: &mut TcpStream, buf: &mut Vec<u8>) -> std::io::Result<()> {
|
||||||
|
let mut tmp = [0u8; 4096];
|
||||||
|
let header_end = loop {
|
||||||
|
if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
|
||||||
|
break i + 4;
|
||||||
|
}
|
||||||
|
let n = s.read(&mut tmp)?;
|
||||||
|
if n == 0 {
|
||||||
|
return Err(std::io::ErrorKind::UnexpectedEof.into());
|
||||||
|
}
|
||||||
|
buf.extend_from_slice(&tmp[..n]);
|
||||||
|
};
|
||||||
|
let head = String::from_utf8_lossy(&buf[..header_end]).to_ascii_lowercase();
|
||||||
|
let len: usize = head
|
||||||
|
.lines()
|
||||||
|
.find_map(|l| l.strip_prefix("content-length:"))
|
||||||
|
.and_then(|v| v.trim().parse().ok())
|
||||||
|
.unwrap_or(0);
|
||||||
|
while buf.len() < header_end + len {
|
||||||
|
let n = s.read(&mut tmp)?;
|
||||||
|
if n == 0 {
|
||||||
|
return Err(std::io::ErrorKind::UnexpectedEof.into());
|
||||||
|
}
|
||||||
|
buf.extend_from_slice(&tmp[..n]);
|
||||||
|
}
|
||||||
|
buf.drain(..header_end + len);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn load_loop(port: u16, stop: Arc<AtomicBool>) {
|
||||||
|
let mut n: u64 = 1;
|
||||||
|
'outer: while !stop.load(Ordering::Relaxed) {
|
||||||
|
let mut s = match TcpStream::connect(("127.0.0.1", port)) {
|
||||||
|
Ok(s) => s,
|
||||||
|
Err(_) => {
|
||||||
|
std::thread::sleep(Duration::from_millis(10));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
s.set_read_timeout(Some(Duration::from_secs(5))).ok();
|
||||||
|
s.set_nodelay(true).ok();
|
||||||
|
let mut buf = Vec::with_capacity(4096);
|
||||||
|
while !stop.load(Ordering::Relaxed) {
|
||||||
|
let req = format!("GET /order/{n} HTTP/1.1\r\nhost: bench\r\n\r\n");
|
||||||
|
if s.write_all(req.as_bytes()).is_err() {
|
||||||
|
continue 'outer;
|
||||||
|
}
|
||||||
|
if read_response(&mut s, &mut buf).is_err() {
|
||||||
|
continue 'outer;
|
||||||
|
}
|
||||||
|
n += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// main — boot, load, sweep, verdict
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
let per_us = calibrate_iters_per_us();
|
||||||
|
PER_US.set(per_us).expect("PER_US set twice");
|
||||||
|
println!("calibration: {per_us} work iters/µs");
|
||||||
|
println!(
|
||||||
|
"planted: store {STORE_US}µs serialized (ceiling ≈{} rps), render {RENDER_US}µs, {LOAD_CONNS} conns",
|
||||||
|
1_000_000 / STORE_US
|
||||||
|
);
|
||||||
|
println!("theory: store +33% @25%, +100% @50%; every other site ≈0");
|
||||||
|
|
||||||
|
// Ephemeral port via bind-and-release (integration-test pattern; the
|
||||||
|
// race window before the server rebinds is far below flake threshold).
|
||||||
|
let port = {
|
||||||
|
let l = TcpListener::bind("127.0.0.1:0").expect("bind probe");
|
||||||
|
l.local_addr().expect("local_addr").port()
|
||||||
|
};
|
||||||
|
let addr: SocketAddr = format!("127.0.0.1:{port}").parse().expect("addr");
|
||||||
|
|
||||||
|
let (handle, signal) = shutdown_handle();
|
||||||
|
let server = std::thread::spawn(move || {
|
||||||
|
let pipe = Pipeline::new().plug(Router::new().get("/order/:id", order));
|
||||||
|
serve_with_shutdown(Config::new(addr), pipe, signal).expect("serve");
|
||||||
|
});
|
||||||
|
|
||||||
|
// Wait until it's accepting.
|
||||||
|
let up = (0..100).any(|_| {
|
||||||
|
std::thread::sleep(Duration::from_millis(50));
|
||||||
|
TcpStream::connect(addr).is_ok()
|
||||||
|
});
|
||||||
|
assert!(up, "server didn't come up on {addr}");
|
||||||
|
|
||||||
|
let stop = Arc::new(AtomicBool::new(false));
|
||||||
|
let loaders: Vec<_> = (0..LOAD_CONNS)
|
||||||
|
.map(|_| {
|
||||||
|
let stop = stop.clone();
|
||||||
|
std::thread::spawn(move || load_loop(port, stop))
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
|
||||||
|
// Warm up: queues to steady state, every site + the progress point
|
||||||
|
// registered by real traffic before the sweep enumerates them.
|
||||||
|
std::thread::sleep(Duration::from_millis(500));
|
||||||
|
|
||||||
|
// The controller runs on this plain OS thread: its windows are wall
|
||||||
|
// time by construction here; inside a runtime they'd be wall-anchored
|
||||||
|
// via the efbc254 timer path (the thing the jobrunner sweep confirms).
|
||||||
|
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),
|
||||||
|
});
|
||||||
|
|
||||||
|
stop.store(true, Ordering::Relaxed);
|
||||||
|
for l in loaders {
|
||||||
|
l.join().expect("loader panicked");
|
||||||
|
}
|
||||||
|
STORE_STOP.store(true, Ordering::Relaxed);
|
||||||
|
handle.shutdown();
|
||||||
|
server.join().expect("server panicked");
|
||||||
|
|
||||||
|
print!("{}", smarm::causal::render_summary(&results));
|
||||||
|
match std::fs::write("profile.coz", smarm::causal::render_coz(&results)) {
|
||||||
|
Ok(()) => println!("\nwrote profile.coz"),
|
||||||
|
Err(e) => eprintln!("\nfailed to write profile.coz: {e}"),
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verdict — same discipline as the smarm demo: skip where the
|
||||||
|
// separation physically can't exist.
|
||||||
|
let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
|
||||||
|
if cores < 4 {
|
||||||
|
println!("verdict: SKIPPED ({cores} cores; separation needs real parallelism)");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let mut failures: Vec<String> = Vec::new();
|
||||||
|
let impact =
|
||||||
|
|site: &str| smarm::causal::impact_pct(&results, site, 25, "responses");
|
||||||
|
let mut expect = |site: &str, ok: &dyn Fn(f64) -> bool, want: &str| match impact(site) {
|
||||||
|
Some(p) => {
|
||||||
|
let verdict = if ok(p) { "ok" } else { "FAIL" };
|
||||||
|
println!("verdict: {site} @25% -> {p:+.1}% (want {want}) {verdict}");
|
||||||
|
if !ok(p) {
|
||||||
|
failures.push(format!("{site}: {p:+.1}% (want {want})"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
println!("verdict: {site} @25% -> missing cell FAIL");
|
||||||
|
failures.push(format!("{site}: missing cell"));
|
||||||
|
}
|
||||||
|
};
|
||||||
|
expect("store", &|p| p > 15.0, "> +15%");
|
||||||
|
for site in ["parse", "router", "pipeline", "serialize", "socket-write"] {
|
||||||
|
expect(site, &|p| p < 10.0, "< +10%");
|
||||||
|
}
|
||||||
|
|
||||||
|
if failures.is_empty() {
|
||||||
|
println!("verdict: PASS — planted bottleneck found, cold sites quiet");
|
||||||
|
} else {
|
||||||
|
println!("verdict: FAIL — {}", failures.join("; "));
|
||||||
|
std::process::exit(1);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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");
|
||||||
|
}
|
||||||
@@ -0,0 +1,48 @@
|
|||||||
|
//! Plain benchmark server: the `load_profile` request path with zero causal
|
||||||
|
//! machinery — the baseline half of the ka/close A/B matrix (the
|
||||||
|
//! throughput-inversion chase).
|
||||||
|
//!
|
||||||
|
//! Identical route, handler, and `Config` construction to `load_profile`;
|
||||||
|
//! differs only in having no `smarm-causal` feature, no sweep, and no
|
||||||
|
//! shutdown path — it serves until killed. Throughput is measured
|
||||||
|
//! externally (wrk).
|
||||||
|
//!
|
||||||
|
//! Env:
|
||||||
|
//! URUS_PORT listen port (default 8080; binds 0.0.0.0)
|
||||||
|
//! URUS_SCHED_THREADS smarm scheduler OS threads (default: smarm's own
|
||||||
|
//! default). Malformed values panic rather than
|
||||||
|
//! silently falling back — a benchmark knob that
|
||||||
|
//! quietly reverts to default poisons the cell.
|
||||||
|
//!
|
||||||
|
//! cargo run --release --example plain_serve
|
||||||
|
|
||||||
|
use std::net::SocketAddr;
|
||||||
|
|
||||||
|
use urus::{serve_with, Config, Conn, Next, Pipeline, Router};
|
||||||
|
|
||||||
|
/// Mirrors `load_profile`'s handler byte for byte: param parse + JSON render.
|
||||||
|
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: u16 = std::env::var("URUS_PORT")
|
||||||
|
.ok()
|
||||||
|
.and_then(|v| v.parse().ok())
|
||||||
|
.unwrap_or(8080);
|
||||||
|
let addr: SocketAddr = format!("0.0.0.0:{port}").parse().expect("addr");
|
||||||
|
let sched_threads: Option<usize> = std::env::var("URUS_SCHED_THREADS").ok().map(|v| {
|
||||||
|
v.parse()
|
||||||
|
.unwrap_or_else(|_| panic!("URUS_SCHED_THREADS not a usize: {v:?}"))
|
||||||
|
});
|
||||||
|
// Audit line: lands in each bench cell's server.log so the effective
|
||||||
|
// scheduler count is recorded per cell, same discipline as mode-verify.
|
||||||
|
eprintln!("plain_serve: scheduler_threads={sched_threads:?}");
|
||||||
|
let mut cfg = Config::new(addr);
|
||||||
|
cfg.scheduler_threads = sched_threads;
|
||||||
|
let pipe = Pipeline::new().plug(Router::new().get("/json/:id", json_id));
|
||||||
|
serve_with(cfg, pipe).expect("serve");
|
||||||
|
}
|
||||||
+1
-1
@@ -92,7 +92,7 @@ impl WsHandler for ChatHandler {
|
|||||||
// subscribed as.
|
// subscribed as.
|
||||||
let _ = self
|
let _ = self
|
||||||
.bus()
|
.bus()
|
||||||
.broadcast_from(smarm::self_pid(), &self.topic(), text);
|
.broadcast_from(smarm::self_pid(), self.topic(), text);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn on_close(&mut self, _code: Option<u16>, _reason: &str) {
|
fn on_close(&mut self, _code: Option<u16>, _reason: &str) {
|
||||||
|
|||||||
Executable
+89
@@ -0,0 +1,89 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# E1: does the ka low-concurrency latency floor track idle-scheduler count?
|
||||||
|
#
|
||||||
|
# Mechanism under test (smarm): one shared level-triggered wake pipe; every
|
||||||
|
# completion byte wakes ALL idle schedulers; one drain-lock winner; losers
|
||||||
|
# stampede the timers/io/queue mutexes and re-sleep; `enqueue` is silent.
|
||||||
|
# Prediction if right: at c <= 8, ka latency tail shrinks and throughput
|
||||||
|
# rises as scheduler count drops (fewer idle pollers -> smaller herd);
|
||||||
|
# close mode (control) stays flat or worsens as threads drop.
|
||||||
|
#
|
||||||
|
# Sweeps URUS_SCHED_THREADS x CONNS over the plain ka/close matrix,
|
||||||
|
# PLAIN_ONLY — causal cells are irrelevant to E1. wrk side is byte-identical
|
||||||
|
# to the 96b40ad5 conns sweep (THREADS=4, --latency) for comparability.
|
||||||
|
#
|
||||||
|
# Knobs: THREADS_SET CONNS_SET DUR REPS OUT (+ everything the matrix takes)
|
||||||
|
set -euo pipefail
|
||||||
|
cd "$(dirname "$0")/.."
|
||||||
|
|
||||||
|
THREADS_SET="${THREADS_SET:-1 2 4 8}"
|
||||||
|
CONNS_SET="${CONNS_SET:-4 8}"
|
||||||
|
DUR="${DUR:-15}"
|
||||||
|
REPS="${REPS:-2}"
|
||||||
|
OUT="${OUT:-/workspace/results}"
|
||||||
|
|
||||||
|
mkdir -p "$OUT"
|
||||||
|
say() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$OUT/e1.log"; }
|
||||||
|
|
||||||
|
for t in $THREADS_SET; do
|
||||||
|
for c in $CONNS_SET; do
|
||||||
|
cell="$OUT/t${t}-c${c}"
|
||||||
|
say "=== E1 cell: sched_threads=$t conns=$c -> $cell ==="
|
||||||
|
URUS_SCHED_THREADS="$t" PLAIN_ONLY=1 \
|
||||||
|
DUR="$DUR" REPS="$REPS" CONNS="$c" OUT="$cell" \
|
||||||
|
bash scripts/ka-close-matrix.sh
|
||||||
|
# Cell audit: the knob must have actually reached the server. A cell
|
||||||
|
# whose server silently ran at default threads poisons the sweep — the
|
||||||
|
# exact failure class per-cell verification exists to catch.
|
||||||
|
for m in ka close; do
|
||||||
|
if ! grep -q "scheduler_threads=Some($t)" "$cell/plain-$m/server.log"; then
|
||||||
|
say "FATAL: t=$t c=$c mode=$m server.log lacks scheduler_threads=Some($t)"
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
done
|
||||||
|
say "cell t=$t c=$c audit OK"
|
||||||
|
done
|
||||||
|
done
|
||||||
|
|
||||||
|
say "=== E1 SUMMARY ==="
|
||||||
|
python3 - "$OUT" <<'PY' | tee -a "$OUT/e1.log"
|
||||||
|
import re, sys, pathlib, statistics
|
||||||
|
|
||||||
|
out = pathlib.Path(sys.argv[1])
|
||||||
|
|
||||||
|
def ms(tok):
|
||||||
|
m = re.match(r"([\d.]+)(us|ms|s)$", tok)
|
||||||
|
if not m:
|
||||||
|
return None
|
||||||
|
v = float(m.group(1))
|
||||||
|
return {"us": v / 1000, "ms": v, "s": v * 1000}[m.group(2)]
|
||||||
|
|
||||||
|
def cell_stats(d):
|
||||||
|
reps, pcts = [], {}
|
||||||
|
for rep in sorted(d.glob("rep*.txt")):
|
||||||
|
t = rep.read_text()
|
||||||
|
m = re.search(r"Requests/sec:\s+([\d.]+)", t)
|
||||||
|
if m:
|
||||||
|
reps.append(float(m.group(1)))
|
||||||
|
for p, tok in re.findall(r"^\s+(50|75|90|99)%\s+(\S+)$", t, re.M):
|
||||||
|
pcts.setdefault(p, []).append(ms(tok))
|
||||||
|
if not reps:
|
||||||
|
return None
|
||||||
|
return (statistics.mean(reps),
|
||||||
|
{p: statistics.mean([v for v in vs if v is not None])
|
||||||
|
for p, vs in pcts.items()})
|
||||||
|
|
||||||
|
cells = sorted(out.glob("t*-c*"))
|
||||||
|
hdr = f"{'cell':>10} {'mode':>6} {'req/s':>9} {'p50ms':>7} {'p75ms':>7} {'p90ms':>7} {'p99ms':>7}"
|
||||||
|
print(hdr); print("-" * len(hdr))
|
||||||
|
for cell in cells:
|
||||||
|
for mode in ("ka", "close"):
|
||||||
|
s = cell_stats(cell / f"plain-{mode}")
|
||||||
|
if s is None:
|
||||||
|
continue
|
||||||
|
rps, p = s
|
||||||
|
print(f"{cell.name:>10} {mode:>6} {rps:>9.0f} "
|
||||||
|
f"{p.get('50', float('nan')):>7.3f} {p.get('75', float('nan')):>7.3f} "
|
||||||
|
f"{p.get('90', float('nan')):>7.3f} {p.get('99', float('nan')):>7.3f}")
|
||||||
|
PY
|
||||||
|
say "=== E1 DONE ==="
|
||||||
Executable
+181
@@ -0,0 +1,181 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# ka-close-matrix.sh — controlled ka/close A/B for the throughput-inversion
|
||||||
|
# chase (handoff jar item), with the forgiveness-fix box validation folded
|
||||||
|
# into the causal cells (handoff v13 PENDING, pull-forward agreed 2026-07-20).
|
||||||
|
#
|
||||||
|
# Cells, run in order on fresh ports:
|
||||||
|
# plain-ka, plain-close examples/plain_serve (no causal feature).
|
||||||
|
# Metric: wrk Requests/sec, REPS reps after a
|
||||||
|
# discarded warmup.
|
||||||
|
# causal-ka, causal-close examples/load_profile (smarm-causal). Metric:
|
||||||
|
# the sweep summary + ledger audit (forgiveness
|
||||||
|
# column, books balance); wrk is backdrop load
|
||||||
|
# whose own numbers are injection-contaminated.
|
||||||
|
#
|
||||||
|
# Controls: byte-identical wrk invocation per mode pair except the
|
||||||
|
# `Connection: close` header; server and wrk pinned to disjoint core sets
|
||||||
|
# (SMT siblings left idle by default on the 5900X); the negotiated
|
||||||
|
# connection behavior is verified via curl per cell and logged, so a
|
||||||
|
# loadgen-config asymmetry can never silently explain a result again.
|
||||||
|
#
|
||||||
|
# Knobs (env, defaults for the 5900X box):
|
||||||
|
# DUR=30 REPS=2 CONNS=64 THREADS=4 PIN=1
|
||||||
|
# SERVER_CPUS=0-7 WRK_CPUS=8-11
|
||||||
|
# CAUSAL_WRK_DUR=300 BASE_PORT=8080
|
||||||
|
# OUT=/workspace/results
|
||||||
|
# PLAIN_BIN, CAUSAL_BIN binary paths (default: target/release/examples/*)
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
cd "$(dirname "$0")/.."
|
||||||
|
|
||||||
|
DUR="${DUR:-30}"
|
||||||
|
REPS="${REPS:-2}"
|
||||||
|
CONNS="${CONNS:-64}"
|
||||||
|
THREADS="${THREADS:-4}"
|
||||||
|
PIN="${PIN:-1}"
|
||||||
|
SERVER_CPUS="${SERVER_CPUS:-0-7}"
|
||||||
|
WRK_CPUS="${WRK_CPUS:-8-11}"
|
||||||
|
CAUSAL_WRK_DUR="${CAUSAL_WRK_DUR:-300}"
|
||||||
|
BASE_PORT="${BASE_PORT:-8080}"
|
||||||
|
OUT="${OUT:-/workspace/results}"
|
||||||
|
PLAIN_BIN="${PLAIN_BIN:-target/release/examples/plain_serve}"
|
||||||
|
CAUSAL_BIN="${CAUSAL_BIN:-target/release/examples/load_profile}"
|
||||||
|
|
||||||
|
mkdir -p "$OUT"
|
||||||
|
say() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$OUT/matrix.log"; }
|
||||||
|
|
||||||
|
pin_server=(); pin_wrk=()
|
||||||
|
if [ "$PIN" = 1 ]; then
|
||||||
|
pin_server=(taskset -c "$SERVER_CPUS")
|
||||||
|
pin_wrk=(taskset -c "$WRK_CPUS")
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ---- environment record --------------------------------------------------
|
||||||
|
{
|
||||||
|
date -u
|
||||||
|
uname -a
|
||||||
|
echo "nproc: $(nproc)"
|
||||||
|
lscpu -e 2>/dev/null || true
|
||||||
|
echo "port range: $(cat /proc/sys/net/ipv4/ip_local_port_range 2>/dev/null)"
|
||||||
|
echo "somaxconn: $(cat /proc/sys/net/core/somaxconn 2>/dev/null)"
|
||||||
|
echo "wrk: $(wrk --version 2>&1 | head -1 || true)"
|
||||||
|
echo "PIN=$PIN SERVER_CPUS=$SERVER_CPUS WRK_CPUS=$WRK_CPUS"
|
||||||
|
echo "DUR=$DUR REPS=$REPS CONNS=$CONNS THREADS=$THREADS CAUSAL_WRK_DUR=$CAUSAL_WRK_DUR"
|
||||||
|
} > "$OUT/env.txt"
|
||||||
|
say "env recorded -> env.txt"
|
||||||
|
|
||||||
|
# ---- helpers -------------------------------------------------------------
|
||||||
|
run_wrk() { # $1=mode $2=duration_s $3=port
|
||||||
|
if [ "$1" = close ]; then
|
||||||
|
"${pin_wrk[@]}" wrk -t "$THREADS" -c "$CONNS" -d "${2}s" --latency \
|
||||||
|
-H "Connection: close" "http://127.0.0.1:$3/json/7"
|
||||||
|
else
|
||||||
|
"${pin_wrk[@]}" wrk -t "$THREADS" -c "$CONNS" -d "${2}s" --latency \
|
||||||
|
"http://127.0.0.1:$3/json/7"
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
|
||||||
|
wait_port() { # $1=port
|
||||||
|
for _ in $(seq 1 150); do
|
||||||
|
curl -s -o /dev/null "http://127.0.0.1:$1/json/1" && return 0
|
||||||
|
sleep 0.2
|
||||||
|
done
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
|
||||||
|
verify_mode() { # $1=mode $2=port — record what actually goes over the wire
|
||||||
|
echo "--- single request, mode=$1 ---"
|
||||||
|
if [ "$1" = close ]; then
|
||||||
|
curl -sv --http1.1 -H "Connection: close" -o /dev/null \
|
||||||
|
"http://127.0.0.1:$2/json/7" 2>&1 | grep -iE "^(> |< )(GET|HTTP|connection)" || true
|
||||||
|
else
|
||||||
|
curl -sv --http1.1 -o /dev/null "http://127.0.0.1:$2/json/7" 2>&1 \
|
||||||
|
| grep -iE "^(> |< )(GET|HTTP|connection)" || true
|
||||||
|
fi
|
||||||
|
echo "--- reuse probe (two requests, one curl) ---"
|
||||||
|
curl -sv --http1.1 -o /dev/null -o /dev/null \
|
||||||
|
"http://127.0.0.1:$2/json/1" "http://127.0.0.1:$2/json/2" 2>&1 \
|
||||||
|
| grep -icE "re-us(ed|ing)" || true
|
||||||
|
}
|
||||||
|
|
||||||
|
# ---- plain cells ---------------------------------------------------------
|
||||||
|
run_plain() { # $1=mode $2=port
|
||||||
|
local mode="$1" port="$2" d="$OUT/plain-$1"
|
||||||
|
mkdir -p "$d"
|
||||||
|
say "=== plain / $mode (port $port) ==="
|
||||||
|
URUS_PORT="$port" "${pin_server[@]}" "$PLAIN_BIN" > "$d/server.log" 2>&1 &
|
||||||
|
local spid=$!
|
||||||
|
wait_port "$port" || { say "FATAL: plain server never came up"; cat "$d/server.log"; exit 1; }
|
||||||
|
verify_mode "$mode" "$port" > "$d/mode-verify.txt" 2>&1
|
||||||
|
ss -s > "$d/ss-before.txt" 2>/dev/null || true
|
||||||
|
run_wrk "$mode" 5 "$port" > "$d/warmup.txt" 2>&1
|
||||||
|
for r in $(seq 1 "$REPS"); do
|
||||||
|
run_wrk "$mode" "$DUR" "$port" > "$d/rep$r.txt" 2>&1
|
||||||
|
grep -E "Requests/sec|Latency |requests in|Socket errors|Non-2xx" "$d/rep$r.txt" \
|
||||||
|
| sed "s/^/ [plain-$mode r$r] /" | tee -a "$OUT/matrix.log" || true
|
||||||
|
done
|
||||||
|
ss -s > "$d/ss-after.txt" 2>/dev/null || true
|
||||||
|
awk '{printf "server cpu jiffies (utime+stime): %d\n", $14+$15}' \
|
||||||
|
"/proc/$spid/stat" > "$d/server-cpu.txt" 2>/dev/null || true
|
||||||
|
kill "$spid" 2>/dev/null || true
|
||||||
|
wait "$spid" 2>/dev/null || true
|
||||||
|
}
|
||||||
|
|
||||||
|
# ---- causal cells --------------------------------------------------------
|
||||||
|
run_causal() { # $1=mode $2=port
|
||||||
|
local mode="$1" port="$2" d="$OUT/causal-$1"
|
||||||
|
mkdir -p "$d"
|
||||||
|
say "=== causal / $mode (port $port) ==="
|
||||||
|
URUS_PORT="$port" COZ_OUT="$d/profile.coz" WARMUP_MS=2000 \
|
||||||
|
"${pin_server[@]}" "$CAUSAL_BIN" > "$d/sweep.log" 2>&1 &
|
||||||
|
local spid=$!
|
||||||
|
wait_port "$port" || { say "FATAL: causal server never came up"; cat "$d/sweep.log"; exit 1; }
|
||||||
|
verify_mode "$mode" "$port" > "$d/mode-verify.txt" 2>&1
|
||||||
|
run_wrk "$mode" "$CAUSAL_WRK_DUR" "$port" > "$d/wrk-backdrop.txt" 2>&1 &
|
||||||
|
local wpid=$!
|
||||||
|
local rc=0
|
||||||
|
wait "$spid" || rc=$?
|
||||||
|
kill "$wpid" 2>/dev/null || true
|
||||||
|
wait "$wpid" 2>/dev/null || true
|
||||||
|
say "causal/$mode server exit=$rc (sweep + audit in sweep.log)"
|
||||||
|
if [ "$rc" -ne 0 ]; then say "WARNING: causal/$mode exited nonzero"; fi
|
||||||
|
}
|
||||||
|
|
||||||
|
# ---- matrix --------------------------------------------------------------
|
||||||
|
run_plain ka "$BASE_PORT"
|
||||||
|
run_plain close "$((BASE_PORT + 1))"
|
||||||
|
if [ "${PLAIN_ONLY:-0}" != 1 ]; then
|
||||||
|
run_causal ka "$((BASE_PORT + 2))"
|
||||||
|
run_causal close "$((BASE_PORT + 3))"
|
||||||
|
fi
|
||||||
|
|
||||||
|
# ---- summary -------------------------------------------------------------
|
||||||
|
say "=== SUMMARY ==="
|
||||||
|
python3 - "$OUT" <<'PY' | tee -a "$OUT/matrix.log"
|
||||||
|
import re, sys, pathlib, statistics
|
||||||
|
out = pathlib.Path(sys.argv[1])
|
||||||
|
means = {}
|
||||||
|
for mode in ("ka", "close"):
|
||||||
|
vals = []
|
||||||
|
for rep in sorted((out / f"plain-{mode}").glob("rep*.txt")):
|
||||||
|
m = re.search(r"Requests/sec:\s+([\d.]+)", rep.read_text())
|
||||||
|
if m:
|
||||||
|
vals.append(float(m.group(1)))
|
||||||
|
if vals:
|
||||||
|
means[mode] = statistics.mean(vals)
|
||||||
|
print(f"plain-{mode}: reps={[f'{v:.0f}' for v in vals]} mean={means[mode]:.0f} req/s")
|
||||||
|
if len(means) == 2:
|
||||||
|
r = means["close"] / means["ka"]
|
||||||
|
print(f"close/ka ratio: {r:.3f}", "-> INVERSION PRESENT (close beats ka)" if r > 1.05
|
||||||
|
else "-> no inversion (ka >= close)" if r < 0.95 else "-> parity")
|
||||||
|
for mode in ("ka", "close"):
|
||||||
|
log = out / f"causal-{mode}" / "sweep.log"
|
||||||
|
if log.exists():
|
||||||
|
t = log.read_text()
|
||||||
|
keep = [l for l in t.splitlines()
|
||||||
|
if re.search(r"forgiv|books|audit|balance|total responses", l, re.I)]
|
||||||
|
print(f"-- causal-{mode} audit lines --")
|
||||||
|
for l in keep[:14]:
|
||||||
|
print(" " + l)
|
||||||
|
PY
|
||||||
|
say "=== MATRIX DONE ==="
|
||||||
@@ -48,8 +48,8 @@
|
|||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
use smarm::gen_server::{self, GenServer, ServerCtx};
|
use smarm::gen_server::{self, GenServer, GenServerCtx};
|
||||||
use smarm::{Down, ServerRef, Watcher};
|
use smarm::{Down, GenServerRef, Watcher};
|
||||||
|
|
||||||
use std::collections::VecDeque;
|
use std::collections::VecDeque;
|
||||||
use std::hash::Hash;
|
use std::hash::Hash;
|
||||||
@@ -87,7 +87,7 @@ pub trait ChannelSession<P>: Send + 'static {
|
|||||||
pub(super) struct SessionFactory<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> {
|
pub(super) struct SessionFactory<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> {
|
||||||
inner: Arc<dyn ChannelFactory<P>>,
|
inner: Arc<dyn ChannelFactory<P>>,
|
||||||
keyfn: fn(&str, &P) -> K,
|
keyfn: fn(&str, &P) -> K,
|
||||||
registry: ServerRef<Registry<P, K>>,
|
registry: GenServerRef<Registry<P, K>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The registry's working bounds for a session key.
|
/// The registry's working bounds for a session key.
|
||||||
@@ -140,14 +140,14 @@ pub(super) struct Join<P: Send + Sync + 'static, K> {
|
|||||||
inbound: Receiver<In<P>>,
|
inbound: Receiver<In<P>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
struct Registry<P: Send + Sync + 'static, K> {
|
struct Registry<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> {
|
||||||
factory: Arc<dyn ChannelFactory<P>>,
|
factory: Arc<dyn ChannelFactory<P>>,
|
||||||
cap: usize,
|
cap: usize,
|
||||||
ttl: Duration,
|
ttl: Duration,
|
||||||
sessions: HashMap<K, (Pid, Sender<Ctl<P>>)>,
|
sessions: HashMap<K, (Pid, Sender<Ctl<P>>)>,
|
||||||
/// Reverse index for `handle_down` — the reason `Key: Clone`.
|
/// Reverse index for `handle_down` — the reason `Key: Clone`.
|
||||||
pid_key: HashMap<Pid, K>,
|
pid_key: HashMap<Pid, K>,
|
||||||
watcher: Option<Watcher>,
|
watcher: Option<Watcher<Registry<P, K>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> GenServer for Registry<P, K> {
|
impl<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> GenServer for Registry<P, K> {
|
||||||
@@ -155,8 +155,9 @@ impl<P: Encode + Decode + Send + Sync + 'static, K: SessionKey> GenServer for Re
|
|||||||
type Reply = ();
|
type Reply = ();
|
||||||
type Cast = Join<P, K>;
|
type Cast = Join<P, K>;
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn init(&mut self, ctx: &ServerCtx) {
|
fn init(&mut self, ctx: &GenServerCtx<Self>) {
|
||||||
self.watcher = Some(ctx.watcher());
|
self.watcher = Some(ctx.watcher());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+2
-4
@@ -158,7 +158,9 @@ impl Body {
|
|||||||
// Empty `Vec`s are skipped by the pump (a zero-length chunk would
|
// Empty `Vec`s are skipped by the pump (a zero-length chunk would
|
||||||
// terminate chunked framing early), so they are safe to send but useless.
|
// terminate chunked framing early), so they are safe to send but useless.
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
pub enum RespBody {
|
pub enum RespBody {
|
||||||
|
#[default]
|
||||||
Empty,
|
Empty,
|
||||||
Bytes(Vec<u8>),
|
Bytes(Vec<u8>),
|
||||||
Stream(StreamBody),
|
Stream(StreamBody),
|
||||||
@@ -217,10 +219,6 @@ impl RespBody {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for RespBody {
|
|
||||||
fn default() -> Self { RespBody::Empty }
|
|
||||||
}
|
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// Params — path parameters extracted by the router.
|
// Params — path parameters extracted by the router.
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
+28
-7
@@ -19,7 +19,7 @@ use crate::net::OwnedFd;
|
|||||||
use crate::parser::{self, ParseError};
|
use crate::parser::{self, ParseError};
|
||||||
use crate::plug::Pipeline;
|
use crate::plug::Pipeline;
|
||||||
|
|
||||||
use smarm::ServerRef;
|
use smarm::GenServerRef;
|
||||||
|
|
||||||
use std::io::{self, ErrorKind};
|
use std::io::{self, ErrorKind};
|
||||||
use std::os::fd::RawFd;
|
use std::os::fd::RawFd;
|
||||||
@@ -92,7 +92,7 @@ pub fn run_connection(
|
|||||||
fd: OwnedFd,
|
fd: OwnedFd,
|
||||||
pipeline: Pipeline,
|
pipeline: Pipeline,
|
||||||
limits: ConnLimits,
|
limits: ConnLimits,
|
||||||
registry: ServerRef<ConnRegistry>,
|
registry: GenServerRef<ConnRegistry>,
|
||||||
) {
|
) {
|
||||||
// The OwnedFd cleans up via Drop on any exit path (panic, error, or
|
// The OwnedFd cleans up via Drop on any exit path (panic, error, or
|
||||||
// normal close). No explicit close calls below.
|
// normal close). No explicit close calls below.
|
||||||
@@ -157,11 +157,11 @@ pub fn run_connection(
|
|||||||
// If client sent `Expect: 100-continue`, emit it before reading the
|
// If client sent `Expect: 100-continue`, emit it before reading the
|
||||||
// body. RFC 7231 §5.1.1. We don't gate on app logic here; v1 always
|
// body. RFC 7231 §5.1.1. We don't gate on app logic here; v1 always
|
||||||
// accepts.
|
// accepts.
|
||||||
if parsed.expect_100 {
|
if parsed.expect_100
|
||||||
if write_all(raw, b"HTTP/1.1 100 Continue\r\n\r\n", Instant::now() + limits.write_timeout).is_err() {
|
&& write_all(raw, b"HTTP/1.1 100 Continue\r\n\r\n", Instant::now() + limits.write_timeout).is_err()
|
||||||
|
{
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
// `consumed_past_head`: how many RAW bytes of `buf` past the head
|
// `consumed_past_head`: how many RAW bytes of `buf` past the head
|
||||||
// this request's body occupied — for chunked bodies that is framing
|
// this request's body occupied — for chunked bodies that is framing
|
||||||
@@ -209,6 +209,8 @@ pub fn run_connection(
|
|||||||
// Catch panics at the actor boundary — a panicking handler should
|
// Catch panics at the actor boundary — a panicking handler should
|
||||||
// not take down the whole connection silently with no response.
|
// not take down the whole connection silently with no response.
|
||||||
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("pipeline");
|
||||||
pipeline.run(conn)
|
pipeline.run(conn)
|
||||||
}));
|
}));
|
||||||
|
|
||||||
@@ -273,7 +275,11 @@ pub fn run_connection(
|
|||||||
let keep_alive = keep_alive
|
let keep_alive = keep_alive
|
||||||
&& !(version == HttpVersion::Http10 && (is_stream || is_error));
|
&& !(version == HttpVersion::Http10 && (is_stream || is_error));
|
||||||
|
|
||||||
let head_bytes = parser::serialise_response(&response_conn, keep_alive);
|
let head_bytes = {
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("serialize");
|
||||||
|
parser::serialise_response(&response_conn, keep_alive)
|
||||||
|
};
|
||||||
if write_all(raw, &head_bytes, Instant::now() + limits.write_timeout).is_err() {
|
if write_all(raw, &head_bytes, Instant::now() + limits.write_timeout).is_err() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -285,10 +291,15 @@ pub fn run_connection(
|
|||||||
}
|
}
|
||||||
if !chunked {
|
if !chunked {
|
||||||
// EOF delimits the HTTP/1.0 stream body.
|
// EOF delimits the HTTP/1.0 stream body.
|
||||||
|
smarm::progress!("responses");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A full response (head + body, streamed or not) is on the wire:
|
||||||
|
// the unit of useful work for causal profiling (RFC 007).
|
||||||
|
smarm::progress!("responses");
|
||||||
|
|
||||||
// ----- 5. Loop or close. -----
|
// ----- 5. Loop or close. -----
|
||||||
if !keep_alive {
|
if !keep_alive {
|
||||||
return;
|
return;
|
||||||
@@ -358,7 +369,12 @@ fn read_head(
|
|||||||
// Try to parse what we already have. On the first iteration of a
|
// Try to parse what we already have. On the first iteration of a
|
||||||
// fresh keep-alive cycle, `buf` may already hold the next request.
|
// fresh keep-alive cycle, `buf` may already hold the next request.
|
||||||
if !buf.is_empty() {
|
if !buf.is_empty() {
|
||||||
match parser::parse_head(buf, limits.max_headers) {
|
let head = {
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("parse");
|
||||||
|
parser::parse_head(buf, limits.max_headers)
|
||||||
|
};
|
||||||
|
match head {
|
||||||
Ok(h) => {
|
Ok(h) => {
|
||||||
let deadline = request_deadline
|
let deadline = request_deadline
|
||||||
.unwrap_or_else(|| Instant::now() + limits.request_timeout);
|
.unwrap_or_else(|| Instant::now() + limits.request_timeout);
|
||||||
@@ -634,6 +650,11 @@ fn try_write_once(fd: RawFd, buf: &[u8]) {
|
|||||||
// wait_writable forever (the write-side twin of slowloris).
|
// wait_writable forever (the write-side twin of slowloris).
|
||||||
|
|
||||||
pub(crate) fn write_all(fd: RawFd, mut buf: &[u8], deadline: Instant) -> io::Result<()> {
|
pub(crate) fn write_all(fd: RawFd, mut buf: &[u8], deadline: Instant) -> io::Result<()> {
|
||||||
|
// The whole loop — writability parks included — runs under the
|
||||||
|
// `socket-write` causal site: a park inside a site is exactly what
|
||||||
|
// RFC 007's park-gated resume credit exists to attribute.
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("socket-write");
|
||||||
while !buf.is_empty() {
|
while !buf.is_empty() {
|
||||||
// Park on writability before each syscall, bounded by the budget.
|
// Park on writability before each syscall, bounded by the budget.
|
||||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||||
|
|||||||
@@ -33,7 +33,7 @@
|
|||||||
//! (roadmap v0.4+), which is why it exists as its own module rather than
|
//! (roadmap v0.4+), which is why it exists as its own module rather than
|
||||||
//! being inlined into `serve`.
|
//! being inlined into `serve`.
|
||||||
|
|
||||||
use smarm::{GenServer, Pid, ServerBuilder, ServerRef};
|
use smarm::{GenServer, Pid, GenServerBuilder, GenServerRef};
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
|
|
||||||
@@ -86,6 +86,7 @@ impl GenServer for ConnRegistry {
|
|||||||
type Reply = Reply;
|
type Reply = Reply;
|
||||||
type Cast = Cast;
|
type Cast = Cast;
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn handle_call(&mut self, request: Call) -> Reply {
|
fn handle_call(&mut self, request: Call) -> Reply {
|
||||||
match request {
|
match request {
|
||||||
@@ -137,8 +138,8 @@ impl GenServer for ConnRegistry {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn start() -> ServerRef<ConnRegistry> {
|
pub fn start() -> GenServerRef<ConnRegistry> {
|
||||||
ServerBuilder::new(ConnRegistry::default()).start()
|
GenServerBuilder::new(ConnRegistry::default()).start()
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@@ -149,13 +150,13 @@ pub fn start() -> ServerRef<ConnRegistry> {
|
|||||||
/// unwind, and panic unwind alike; the cast is infallible from the
|
/// unwind, and panic unwind alike; the cast is infallible from the
|
||||||
/// guard's perspective (a dead registry just returns an ignored Err).
|
/// guard's perspective (a dead registry just returns an ignored Err).
|
||||||
pub struct DeregisterGuard {
|
pub struct DeregisterGuard {
|
||||||
registry: ServerRef<ConnRegistry>,
|
registry: GenServerRef<ConnRegistry>,
|
||||||
pid: Pid,
|
pid: Pid,
|
||||||
make: fn(Pid) -> Cast,
|
make: fn(Pid) -> Cast,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl DeregisterGuard {
|
impl DeregisterGuard {
|
||||||
pub fn new(registry: ServerRef<ConnRegistry>, pid: Pid, make: fn(Pid) -> Cast) -> Self {
|
pub fn new(registry: GenServerRef<ConnRegistry>, pid: Pid, make: fn(Pid) -> Cast) -> Self {
|
||||||
Self { registry, pid, make }
|
Self { registry, pid, make }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-3
@@ -123,11 +123,9 @@ pub fn parse_head(buf: &[u8], max_headers: usize) -> Result<ParsedHead, ParseErr
|
|||||||
"connection" => {
|
"connection" => {
|
||||||
connection_hdr = Some(value.to_ascii_lowercase());
|
connection_hdr = Some(value.to_ascii_lowercase());
|
||||||
}
|
}
|
||||||
"expect" => {
|
"expect" if value.eq_ignore_ascii_case("100-continue") => {
|
||||||
if value.eq_ignore_ascii_case("100-continue") {
|
|
||||||
expect_100 = true;
|
expect_100 = true;
|
||||||
}
|
}
|
||||||
}
|
|
||||||
_ => {}
|
_ => {}
|
||||||
}
|
}
|
||||||
headers.append(&name_lower, value.to_string());
|
headers.append(&name_lower, value.to_string());
|
||||||
|
|||||||
+7
-6
@@ -58,7 +58,7 @@
|
|||||||
//! # Why there is no `register(name)` helper (yet)
|
//! # Why there is no `register(name)` helper (yet)
|
||||||
//!
|
//!
|
||||||
//! smarm's registry maps `name → Pid`, but a `Pid` cannot be turned back
|
//! smarm's registry maps `name → Pid`, but a `Pid` cannot be turned back
|
||||||
//! into a `ServerRef` (the ref *is* the inbox sender). A useful named
|
//! into a `GenServerRef` (the ref *is* the inbox sender). A useful named
|
||||||
//! lookup therefore needs either smarm support (registry-held senders)
|
//! lookup therefore needs either smarm support (registry-held senders)
|
||||||
//! or a process-global type-erased map here — both against the grain of
|
//! or a process-global type-erased map here — both against the grain of
|
||||||
//! the ratified design. Deferred; pass the handle.
|
//! the ratified design. Deferred; pass the handle.
|
||||||
@@ -67,8 +67,8 @@ use std::collections::{HashMap, HashSet};
|
|||||||
use std::fmt;
|
use std::fmt;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use smarm::gen_server::{self, GenServer, ServerCtx};
|
use smarm::gen_server::{self, GenServer, GenServerCtx};
|
||||||
use smarm::{channel, Down, Pid, Receiver, Sender, ServerRef, Watcher};
|
use smarm::{channel, Down, Pid, Receiver, Sender, GenServerRef, Watcher};
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// Public handle
|
// Public handle
|
||||||
@@ -92,7 +92,7 @@ impl std::error::Error for PubSubDown {}
|
|||||||
/// Payloads are broadcast as `Arc<M>`: one allocation per broadcast, not
|
/// Payloads are broadcast as `Arc<M>`: one allocation per broadcast, not
|
||||||
/// per subscriber.
|
/// per subscriber.
|
||||||
pub struct PubSub<M: Send + Sync + 'static> {
|
pub struct PubSub<M: Send + Sync + 'static> {
|
||||||
server: ServerRef<Table<M>>,
|
server: GenServerRef<Table<M>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<M: Send + Sync + 'static> Clone for PubSub<M> {
|
impl<M: Send + Sync + 'static> Clone for PubSub<M> {
|
||||||
@@ -226,7 +226,7 @@ struct Table<M: Send + Sync + 'static> {
|
|||||||
/// retires the entry so a reused-slot pid (fresh generation) gets a
|
/// retires the entry so a reused-slot pid (fresh generation) gets a
|
||||||
/// fresh monitor.
|
/// fresh monitor.
|
||||||
monitored: HashSet<Pid>,
|
monitored: HashSet<Pid>,
|
||||||
watcher: Option<Watcher>,
|
watcher: Option<Watcher<Table<M>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<M: Send + Sync + 'static> Table<M> {
|
impl<M: Send + Sync + 'static> Table<M> {
|
||||||
@@ -240,8 +240,9 @@ impl<M: Send + Sync + 'static> GenServer for Table<M> {
|
|||||||
type Reply = Reply;
|
type Reply = Reply;
|
||||||
type Cast = Cast<M>;
|
type Cast = Cast<M>;
|
||||||
type Info = ();
|
type Info = ();
|
||||||
|
type Timer = ();
|
||||||
|
|
||||||
fn init(&mut self, ctx: &ServerCtx) {
|
fn init(&mut self, ctx: &GenServerCtx<Self>) {
|
||||||
self.watcher = Some(ctx.watcher());
|
self.watcher = Some(ctx.watcher());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+16
-3
@@ -135,16 +135,29 @@ impl Plug for Router {
|
|||||||
// path matches but a same-path-different-method does, return 405.
|
// path matches but a same-path-different-method does, return 405.
|
||||||
// Otherwise pass through to `next` so outer pipelines can layer a
|
// Otherwise pass through to `next` so outer pipelines can layer a
|
||||||
// 404 handler (or skip and let the connection actor emit a default).
|
// 404 handler (or skip and let the connection actor emit a default).
|
||||||
|
//
|
||||||
|
// Matching runs under the `router` causal site (RFC 007); the
|
||||||
|
// winning handler and the `next` fall-through run outside it, so
|
||||||
|
// the site measures dispatch, not what it dispatches to.
|
||||||
let mut path_seen = false;
|
let mut path_seen = false;
|
||||||
for route in &self.routes {
|
let mut hit = None;
|
||||||
|
{
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("router");
|
||||||
|
for (i, route) in self.routes.iter().enumerate() {
|
||||||
if let Some(params) = route.pattern.match_path(&conn.path) {
|
if let Some(params) = route.pattern.match_path(&conn.path) {
|
||||||
if route.method == conn.method {
|
if route.method == conn.method {
|
||||||
let conn = conn.put_params(params);
|
hit = Some((i, params));
|
||||||
return route.handler.call(conn, next);
|
break;
|
||||||
}
|
}
|
||||||
path_seen = true;
|
path_seen = true;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
if let Some((i, params)) = hit {
|
||||||
|
let conn = conn.put_params(params);
|
||||||
|
return self.routes[i].handler.call(conn, next);
|
||||||
|
}
|
||||||
if path_seen {
|
if path_seen {
|
||||||
// Path is known, method isn't — RFC 7231 §6.5.5.
|
// Path is known, method isn't — RFC 7231 §6.5.5.
|
||||||
conn.put_status(405)
|
conn.put_status(405)
|
||||||
|
|||||||
+10
-17
@@ -17,7 +17,7 @@ use crate::conn_registry::{self, Call, Cast, ConnRegistry, Reply};
|
|||||||
use crate::net::{accept_nonblocking, bind_and_listen, OwnedFd};
|
use crate::net::{accept_nonblocking, bind_and_listen, OwnedFd};
|
||||||
use crate::plug::Pipeline;
|
use crate::plug::Pipeline;
|
||||||
|
|
||||||
use smarm::{ChildSpec, OneForOne, Restart, ServerRef, Strategy};
|
use smarm::{ChildSpec, OneForOne, Restart, GenServerRef, Strategy};
|
||||||
|
|
||||||
use std::io::{self, ErrorKind};
|
use std::io::{self, ErrorKind};
|
||||||
use std::net::{SocketAddr, ToSocketAddrs};
|
use std::net::{SocketAddr, ToSocketAddrs};
|
||||||
@@ -125,7 +125,7 @@ fn listener_loop(
|
|||||||
listener: Arc<OwnedFd>,
|
listener: Arc<OwnedFd>,
|
||||||
pipeline: Pipeline,
|
pipeline: Pipeline,
|
||||||
limits: ConnLimits,
|
limits: ConnLimits,
|
||||||
registry: ServerRef<ConnRegistry>,
|
registry: GenServerRef<ConnRegistry>,
|
||||||
shutdown: Arc<AtomicBool>,
|
shutdown: Arc<AtomicBool>,
|
||||||
) {
|
) {
|
||||||
let fd = listener.as_raw();
|
let fd = listener.as_raw();
|
||||||
@@ -327,15 +327,6 @@ pub fn serve_with_shutdown(
|
|||||||
let sf = shutdown_flag.clone();
|
let sf = shutdown_flag.clone();
|
||||||
sup = sup.child(ChildSpec::new(Restart::Transient, move || {
|
sup = sup.child(ChildSpec::new(Restart::Transient, move || {
|
||||||
println!("urus: listener {} starting", i);
|
println!("urus: listener {} starting", i);
|
||||||
// Named for whereis-style introspection. On a restart the
|
|
||||||
// old binding points at a dead pid; smarm's registry
|
|
||||||
// evicts stale bindings lazily, so re-registering the
|
|
||||||
// same name is fine. Ignore the result — a registry
|
|
||||||
// hiccup must not take the listener down.
|
|
||||||
let _ = smarm::register(
|
|
||||||
format!("urus.listener.{i}"),
|
|
||||||
smarm::self_pid(),
|
|
||||||
);
|
|
||||||
listener_loop(lfd.clone(), p.clone(), limits, r.clone(), sf.clone());
|
listener_loop(lfd.clone(), p.clone(), limits, r.clone(), sf.clone());
|
||||||
}));
|
}));
|
||||||
}
|
}
|
||||||
@@ -343,11 +334,13 @@ pub fn serve_with_shutdown(
|
|||||||
// faster than that trips the cap and tears the pool down — loud
|
// faster than that trips the cap and tears the pool down — loud
|
||||||
// failure over a zombie server.
|
// failure over a zombie server.
|
||||||
let sup_h = smarm::spawn(move || sup.run());
|
let sup_h = smarm::spawn(move || sup.run());
|
||||||
// Register via the JoinHandle's pid rather than inside the
|
// The old `urus.server` / `urus.listener.{i}` name registrations
|
||||||
// closure: the binding exists before the supervisor body runs a
|
// are gone with smarm's RFC 014 registry rework: `register` is now
|
||||||
// single instruction, so an early `whereis("urus.server")` can't
|
// `(Name<M>, Sender<M>)`, self-only — a name is a typed messaging
|
||||||
// race a None. Result ignored for the same reason as listeners.
|
// endpoint, not a pid tag. urus's bindings were introspection-only
|
||||||
let _ = smarm::register("urus.server", sup_h.pid());
|
// with no channel behind them, so they were dropped rather than
|
||||||
|
// faked with a unit channel. A real messageable `urus.server`
|
||||||
|
// name is in the icebox (ROADMAP.md).
|
||||||
|
|
||||||
// Block until told to shut down. We poll `try_recv` + `sleep`
|
// Block until told to shut down. We poll `try_recv` + `sleep`
|
||||||
// rather than parking in `recv`: a smarm `Sender::send` from a
|
// rather than parking in `recv`: a smarm `Sender::send` from a
|
||||||
@@ -399,7 +392,7 @@ pub fn serve_with_shutdown(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 5. Our ServerRef drops here. The registry's inbox closes once
|
// 5. Our GenServerRef drops here. The registry's inbox closes once
|
||||||
// the last conn's clone drops with it, and the runtime winds
|
// the last conn's clone drops with it, and the runtime winds
|
||||||
// down when the last actor exits.
|
// down when the last actor exits.
|
||||||
});
|
});
|
||||||
|
|||||||
+3
-18
@@ -89,24 +89,9 @@ fn hello_world() {
|
|||||||
assert_eq!(http_body(&resp), b"hello urus");
|
assert_eq!(http_body(&resp), b"hello urus");
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
// `server_and_listeners_are_registered` was deleted with the RFC 014 port:
|
||||||
fn server_and_listeners_are_registered() {
|
// urus no longer binds `urus.server` / `urus.listener.{i}` names (see the
|
||||||
// `whereis` must run inside the runtime (the test itself is a foreign
|
// note in serve.rs and the icebox entry in ROADMAP.md).
|
||||||
// OS thread with no runtime in its TLS), so probe from a handler —
|
|
||||||
// connection actors live in the runtime by construction.
|
|
||||||
let pipe = Pipeline::new().plug(
|
|
||||||
Router::new().get("/whereis", |c: Conn, _n: Next| {
|
|
||||||
let ok = smarm::whereis("urus.server").is_some()
|
|
||||||
&& smarm::whereis("urus.listener.0").is_some()
|
|
||||||
&& smarm::whereis("urus.listener.1").is_some(); // pool of 2
|
|
||||||
c.put_status(200).put_body(if ok { "registered" } else { "missing" })
|
|
||||||
})
|
|
||||||
);
|
|
||||||
let port = spawn_server(pipe);
|
|
||||||
let resp = send_request(port, b"GET /whereis HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
|
|
||||||
assert_eq!(http_status(&resp), 200);
|
|
||||||
assert_eq!(http_body(&resp), b"registered");
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn echo_body() {
|
fn echo_body() {
|
||||||
|
|||||||
Reference in New Issue
Block a user