crates/gunnar-loadbench/src/engine.rs
//! **The load engine.** `gatling_for_each(n, workers, f)` and nothing else:
//! all `n` requests are known before the run starts, one output slot per
//! request is pre-allocated, and `workers` threads drain a shared atomic cursor
//! with no barrier between them. rayon is forbidden by the law; this is the
//! engine the constellation already owns.
//!
//! ## The rate cap is an OPEN model
//!
//! Request `i` has a send time fixed before the run: `t0 + i / rate`. A worker
//! that reaches unit `i` early parks until that instant; a worker that reaches
//! it late sends immediately and records how late it was ([`Sample::late_ns`]).
//!
//! That slip is the number that keeps the bench honest. A closed-loop generator
//! (N threads in a tight loop) silently slows down when the server does, and
//! then reports a flattering p99 about a request rate it never actually
//! offered. Here the schedule does not bend: if the estate cannot keep up, the
//! lateness grows and is reported, and the throughput line says what was
//! ACHIEVED rather than what was asked for.
//!
//! ## The deadline
//!
//! `--deadline` is a wall clock over the whole run. A unit whose turn comes
//! after the deadline is not sent at all and comes back as `None`; the count of
//! those is reported as `not_issued`. Stopping is not a failure of the estate
//! and is not counted as one.
use std::time::{Duration, Instant};
use crate::stats::Sample;
/// One probe's plan. Every field here is a declared number: nothing in this
/// struct has a default that the command line can forget to set.
#[derive(Debug, Clone, Copy)]
pub struct Plan {
/// How many requests to issue in total.
pub n: usize,
/// Max requests in flight — the worker count.
pub workers: usize,
/// Requests per second offered, across all workers.
pub rate_per_s: f64,
/// Per-request budget. Exceeding it is an `Outcome::Timeout`, not a panic.
/// Enforced by the probe's own client (ureq's `timeout_global`, the socket
/// timeouts on the SSH probe), so it is carried here for the record and for
/// the summary rather than read by the engine.
#[allow(dead_code)]
pub timeout: Duration,
}
/// What a run produced: the samples that were issued, and how many were not.
pub struct Run {
pub samples: Vec<Sample>,
/// Units the deadline cut off before they were sent.
pub not_issued: usize,
/// Observed wall-clock span of the issued requests.
pub wall_s: f64,
}
/// Park until `at`, in bounded steps so a long park still notices the deadline.
/// Returns false if the deadline passed first.
fn wait_until(at: Instant, deadline: Instant) -> bool {
loop {
let now = Instant::now();
if now >= deadline {
return false;
}
if now >= at {
return true;
}
let step = (at - now).min(Duration::from_millis(20));
std::thread::sleep(step);
}
}
/// Run one probe. `f(i)` issues request `i` and returns its sample; it is
/// called from `workers` threads at once and must be `Sync`.
///
/// `f` receives the instant the request was actually released, so it can record
/// its own latency from the same clock the schedule uses.
pub fn run<F>(plan: Plan, deadline: Instant, f: F) -> Run
where
F: Fn(usize, Instant) -> Sample + Sync,
{
if plan.n == 0 {
return Run {
samples: Vec::new(),
not_issued: 0,
wall_s: 0.0,
};
}
let t0 = Instant::now();
let gap = if plan.rate_per_s > 0.0 {
Duration::from_secs_f64(1.0 / plan.rate_per_s)
} else {
Duration::ZERO
};
let slots: Vec<Option<Sample>> =
gatling::gatling_forkjoin::gatling_for_each(plan.n, plan.workers, |i| -> Option<Sample> {
let scheduled = t0 + gap.checked_mul(i as u32).unwrap_or(Duration::ZERO);
if !wait_until(scheduled, deadline) {
return None;
}
let released = Instant::now();
let mut s = f(i, released);
// Lateness is measured against the schedule, not against the
// previous request: slip does not get forgiven by accumulating.
s.late_ns = released.saturating_duration_since(scheduled).as_nanos() as u64;
Some(s)
});
let wall_s = t0.elapsed().as_secs_f64();
let mut samples = Vec::with_capacity(plan.n);
let mut not_issued = 0usize;
for slot in slots {
match slot {
Some(s) => samples.push(s),
None => not_issued += 1,
}
}
Run {
samples,
not_issued,
wall_s,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::outcome::Outcome;
fn plan(n: usize, workers: usize, rate: f64) -> Plan {
Plan {
n,
workers,
rate_per_s: rate,
timeout: Duration::from_secs(1),
}
}
#[test]
fn every_unit_is_issued_exactly_once() {
let seen = std::sync::Mutex::new(Vec::new());
let r = run(
plan(64, 8, 10_000.0),
Instant::now() + Duration::from_secs(30),
|i, _| {
seen.lock().unwrap().push(i);
Sample::refused(Outcome::Ok, 0, 0)
},
);
assert_eq!(r.samples.len(), 64);
assert_eq!(r.not_issued, 0);
let mut v = seen.into_inner().unwrap();
v.sort_unstable();
assert_eq!(v, (0..64).collect::<Vec<_>>());
}
#[test]
fn the_rate_cap_actually_paces() {
// 20 units at 100/s cannot finish sooner than ~0.19 s however many
// workers there are.
let r = run(
plan(20, 16, 100.0),
Instant::now() + Duration::from_secs(30),
|_, _| Sample::refused(Outcome::Ok, 0, 0),
);
assert_eq!(r.samples.len(), 20);
assert!(
r.wall_s > 0.15,
"finished in {}s, the cap did not hold",
r.wall_s
);
}
#[test]
fn the_deadline_stops_the_run_and_the_rest_is_not_issued_rather_than_failed() {
let r = run(
plan(10_000, 4, 50.0),
Instant::now() + Duration::from_millis(300),
|_, _| Sample::refused(Outcome::Ok, 0, 0),
);
assert!(r.not_issued > 0, "the deadline issued everything");
assert!(r.samples.len() < 10_000);
assert_eq!(r.samples.len() + r.not_issued, 10_000);
assert!(
r.wall_s < 3.0,
"the deadline did not stop it: {}s",
r.wall_s
);
}
#[test]
fn lateness_is_recorded_when_the_work_cannot_keep_up() {
// one worker, 1000/s asked, 5 ms of work each: it must fall behind.
let r = run(
plan(20, 1, 1000.0),
Instant::now() + Duration::from_secs(30),
|_, _| {
std::thread::sleep(Duration::from_millis(5));
Sample::refused(Outcome::Ok, 0, 0)
},
);
let worst = r.samples.iter().map(|s| s.late_ns).max().unwrap();
assert!(worst > 10_000_000, "no slip recorded: {worst} ns");
}
#[test]
fn zero_units_is_not_a_panic() {
let r = run(plan(0, 4, 10.0), Instant::now(), |_, _| {
Sample::refused(Outcome::Ok, 0, 0)
});
assert_eq!(r.samples.len(), 0);
}
}