mirror of
https://github.com/bdbrown4/peltier.io.git
synced 2026-11-07 10:58:28 +00:00
phase4: service-mode latency bench + mechanical ROI report generator
- bench-runner service mode (SPEC §3.1 mode c): a coordinated-omission- correct, open-loop, fixed-rate load generator; latency measured from the intended send time so queueing is counted, not hidden. Interleaved A/B of two server binaries, exact percentiles (bounded sample, no HDR bucketing error), bootstrap p50/p99 CIs via the existing ratio-CI machine. service-calibrate: A/A false-positive + injected latency- regression detection (calibration in progress). - targets/cjson/service.c: minimal single-threaded HTTP server wrapping the patched cJSON.c (trust-layer, outside the workspace allowlist); env-toggled per-request busy-wait for the injection calibration only. - report generator (crates/report bin): reads a ledger row + optional service-latency JSON, renders throughput→cores→dollars and/or latency percentiles, every figure with its CI, pricing source, and the methodology printed inline (SPEC §9); flags any non-accepted row so a rejected change never reads as a committed saving. - ledger::report_row; justfile service/service-calibrate/report. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EcKPWQ3MQD4NdEHwar96u2
This commit is contained in:
Claude
@@ -0,0 +1,248 @@
|
||||
//! Service-mode latency bench (SPEC §3.1 mode (c)). A coordinated-
|
||||
//! omission-CORRECT, open-loop, fixed-rate load generator: requests are
|
||||
//! scheduled at `start + i/rate` and each latency is measured from its
|
||||
//! *intended* send time, not its actual send time. So when the server
|
||||
//! falls behind, the requests that pile up behind it each carry the full
|
||||
//! queueing delay a real client would see — the exact error (coordinated
|
||||
//! omission) that closed-loop tools like `ab` hide. We store every
|
||||
//! latency and take exact percentiles (no HdrHistogram bucketing error;
|
||||
//! the sample is bounded).
|
||||
//!
|
||||
//! Interleaved A/B, like the wall-time bench: each session yields one
|
||||
//! percentile (p50/p99) per side; the existing bootstrap ratio-CI machine
|
||||
//! then works unchanged, treating "p99 latency" the way it treats "wall
|
||||
//! time" — lower is better, speedup = baseline / candidate.
|
||||
|
||||
use anyhow::{anyhow, ensure, Result};
|
||||
use std::io::{BufRead, BufReader, Read, Write};
|
||||
use std::net::{Ipv4Addr, TcpStream};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::stats::{self, RatioCi};
|
||||
|
||||
pub struct ServiceCfg {
|
||||
pub server_bin: String,
|
||||
pub doc: String,
|
||||
pub iters: u64,
|
||||
pub rate: f64,
|
||||
pub count: usize,
|
||||
pub warmup: usize,
|
||||
pub workers: usize,
|
||||
/// Prefix that pins the SERVER to a core (e.g. `taskset -c 2`); the
|
||||
/// load generator runs on the other cores.
|
||||
pub pin_prefix: String,
|
||||
/// Calibration only: inject this many µs of busy-wait per request
|
||||
/// (via HOTPATH_INJECT_US) to validate the harness detects a known
|
||||
/// latency regression. None in every real run.
|
||||
pub inject_us: Option<u64>,
|
||||
}
|
||||
|
||||
/// Result of one interleaved A/B service compare.
|
||||
pub struct CompareResult {
|
||||
pub ci50: RatioCi,
|
||||
pub ci99: RatioCi,
|
||||
pub base_p50: Vec<f64>,
|
||||
pub cand_p50: Vec<f64>,
|
||||
pub base_p99: Vec<f64>,
|
||||
pub cand_p99: Vec<f64>,
|
||||
pub drop_rate: f64,
|
||||
}
|
||||
|
||||
/// Interleaved A/B: `rounds` alternating baseline/candidate sessions,
|
||||
/// each yielding one p50 and one p99; bootstrap ratio-CI over the rounds.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub fn run_compare(
|
||||
base: &ServiceCfg,
|
||||
cand: &ServiceCfg,
|
||||
rounds: usize,
|
||||
bootstrap_iters: usize,
|
||||
confidence: f64,
|
||||
seed: u64,
|
||||
mut on_round: impl FnMut(usize, f64, f64, f64, f64),
|
||||
) -> Result<CompareResult> {
|
||||
let (mut b50, mut b99, mut c50, mut c99) = (Vec::new(), Vec::new(), Vec::new(), Vec::new());
|
||||
let mut dropped = 0usize;
|
||||
for k in 0..rounds {
|
||||
let bs = run_session(base, free_port()?)?;
|
||||
let cs = run_session(cand, free_port()?)?;
|
||||
dropped += bs.dropped + cs.dropped;
|
||||
let (bp50, bp99) = (
|
||||
percentile(&bs.latencies, 0.50),
|
||||
percentile(&bs.latencies, 0.99),
|
||||
);
|
||||
let (cp50, cp99) = (
|
||||
percentile(&cs.latencies, 0.50),
|
||||
percentile(&cs.latencies, 0.99),
|
||||
);
|
||||
b50.push(bp50);
|
||||
b99.push(bp99);
|
||||
c50.push(cp50);
|
||||
c99.push(cp99);
|
||||
on_round(k, bp50, bp99, cp50, cp99);
|
||||
}
|
||||
let total = rounds * base.count * 2;
|
||||
let drop_rate = dropped as f64 / total as f64;
|
||||
let ci50 = stats::bootstrap_ratio_ci(&b50, &c50, bootstrap_iters, confidence, seed);
|
||||
let ci99 = stats::bootstrap_ratio_ci(&b99, &c99, bootstrap_iters, confidence, seed);
|
||||
Ok(CompareResult {
|
||||
ci50,
|
||||
ci99,
|
||||
base_p50: b50,
|
||||
cand_p50: c50,
|
||||
base_p99: b99,
|
||||
cand_p99: c99,
|
||||
drop_rate,
|
||||
})
|
||||
}
|
||||
|
||||
pub struct SessionResult {
|
||||
pub latencies: Vec<f64>,
|
||||
pub dropped: usize,
|
||||
}
|
||||
|
||||
/// Exact percentile (0.0–1.0) of a sorted-in-place copy.
|
||||
pub fn percentile(latencies: &[f64], q: f64) -> f64 {
|
||||
if latencies.is_empty() {
|
||||
return f64::NAN;
|
||||
}
|
||||
let mut xs = latencies.to_vec();
|
||||
xs.sort_by(|a, b| a.partial_cmp(b).unwrap());
|
||||
let idx = ((xs.len() as f64 - 1.0) * q).round() as usize;
|
||||
xs[idx.min(xs.len() - 1)]
|
||||
}
|
||||
|
||||
fn spawn_server(cfg: &ServiceCfg, port: u16) -> Result<Child> {
|
||||
let inject = match cfg.inject_us {
|
||||
Some(us) => format!("HOTPATH_INJECT_US={us} "),
|
||||
None => String::new(),
|
||||
};
|
||||
let cmd = format!(
|
||||
"{inject}{pin} {bin} {port} {doc} {iters}",
|
||||
pin = cfg.pin_prefix,
|
||||
bin = cfg.server_bin,
|
||||
doc = cfg.doc,
|
||||
iters = cfg.iters,
|
||||
);
|
||||
let mut child = Command::new("sh")
|
||||
.arg("-c")
|
||||
.arg(cmd.trim())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::null())
|
||||
.spawn()?;
|
||||
// Block until the server prints "READY <port>".
|
||||
let stdout = child
|
||||
.stdout
|
||||
.take()
|
||||
.ok_or_else(|| anyhow!("no server stdout"))?;
|
||||
let mut reader = BufReader::new(stdout);
|
||||
let mut line = String::new();
|
||||
reader.read_line(&mut line)?;
|
||||
ensure!(
|
||||
line.starts_with("READY"),
|
||||
"server did not signal READY (got {line:?})"
|
||||
);
|
||||
Ok(child)
|
||||
}
|
||||
|
||||
fn set_linger_zero(s: &TcpStream) {
|
||||
// SO_LINGER with a 0 timeout → RST on close instead of FIN, so the
|
||||
// tens of thousands of short client connections a load session opens
|
||||
// don't pile up in TIME_WAIT and exhaust the ephemeral port range.
|
||||
// std::net::TcpStream::set_linger is still unstable, so go through
|
||||
// setsockopt directly. The response is fully read before close, so the
|
||||
// RST is harmless.
|
||||
use std::os::unix::io::AsRawFd;
|
||||
let l = libc::linger {
|
||||
l_onoff: 1,
|
||||
l_linger: 0,
|
||||
};
|
||||
unsafe {
|
||||
libc::setsockopt(
|
||||
s.as_raw_fd(),
|
||||
libc::SOL_SOCKET,
|
||||
libc::SO_LINGER,
|
||||
&l as *const libc::linger as *const libc::c_void,
|
||||
std::mem::size_of::<libc::linger>() as libc::socklen_t,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn one_request(port: u16) -> std::io::Result<()> {
|
||||
let mut s = TcpStream::connect((Ipv4Addr::LOCALHOST, port))?;
|
||||
s.set_nodelay(true)?;
|
||||
set_linger_zero(&s);
|
||||
s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n")?;
|
||||
let mut buf = [0u8; 512];
|
||||
// Read the full response (server sends Content-Length + Connection: close).
|
||||
loop {
|
||||
match s.read(&mut buf) {
|
||||
Ok(0) => break,
|
||||
Ok(_) => {}
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// One open-loop load session against a freshly spawned server.
|
||||
pub fn run_session(cfg: &ServiceCfg, port: u16) -> Result<SessionResult> {
|
||||
let mut child = spawn_server(cfg, port)?;
|
||||
// Warm up closed-loop (untimed) so the first timed request doesn't eat
|
||||
// page faults / cold caches.
|
||||
for _ in 0..cfg.warmup {
|
||||
let _ = one_request(port);
|
||||
}
|
||||
|
||||
let period = Duration::from_secs_f64(1.0 / cfg.rate);
|
||||
let start = Instant::now();
|
||||
let idx = Arc::new(AtomicUsize::new(0));
|
||||
let out = Arc::new(Mutex::new((Vec::<f64>::with_capacity(cfg.count), 0usize)));
|
||||
let count = cfg.count;
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for _ in 0..cfg.workers {
|
||||
let idx = idx.clone();
|
||||
let out = out.clone();
|
||||
handles.push(std::thread::spawn(move || {
|
||||
let mut local = Vec::new();
|
||||
let mut dropped = 0usize;
|
||||
loop {
|
||||
let i = idx.fetch_add(1, Ordering::Relaxed);
|
||||
if i >= count {
|
||||
break;
|
||||
}
|
||||
let target = start + period * (i as u32);
|
||||
let now = Instant::now();
|
||||
if now < target {
|
||||
std::thread::sleep(target - now);
|
||||
}
|
||||
// Latency from the INTENDED send time — the CO correction.
|
||||
match one_request(port) {
|
||||
Ok(()) => local.push(Instant::now().duration_since(target).as_secs_f64()),
|
||||
Err(_) => dropped += 1,
|
||||
}
|
||||
}
|
||||
let mut g = out.lock().unwrap();
|
||||
g.0.extend(local);
|
||||
g.1 += dropped;
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
let _ = h.join();
|
||||
}
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
|
||||
let (latencies, dropped) = Arc::try_unwrap(out).unwrap().into_inner().unwrap();
|
||||
Ok(SessionResult { latencies, dropped })
|
||||
}
|
||||
|
||||
/// Pick a free loopback TCP port (small race between probe-close and the
|
||||
/// server bind; harmless on loopback with SO_REUSEADDR).
|
||||
pub fn free_port() -> Result<u16> {
|
||||
let l = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?;
|
||||
Ok(l.local_addr()?.port())
|
||||
}
|
||||
Reference in New Issue
Block a user