krabka_bench_driver/soak/client/
windows.rs

1//! Steady and disturbed time, and the latency and throughput split across it.
2//!
3//! The contract's latency budgets and throughput floors apply to the steady
4//! windows only. A window is disturbed from a fault's injection until its
5//! cleanup plus the longest deadline of the recoveries it is measured for:
6//! that is how long the contract lets the cluster take to recover, so nothing
7//! inside it is held to a steady budget. Everything else, the settle tail
8//! included, is steady.
9//!
10//! A sample is classified when it is taken, against two sets of spans: the
11//! spans the schedule plans, and the spans the faults actually ran, which a
12//! slow cleanup can stretch past the plan. While a fault is injected and not
13//! yet cleared, every sample is disturbed.
14
15use {
16    crate::{
17        hist,
18        soak::{
19            contract::Contract,
20            evidence::{LatencySummary, PathMeasurement},
21            schedule::ScheduledFault,
22        },
23    },
24    hdrhistogram::{CreationError, Histogram},
25    krabka_units::prelude::*,
26    std::sync::Mutex,
27};
28
29/// A half-open interval of wall-clock Unix milliseconds.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct Span {
32    pub start: i64,
33    pub end: i64,
34}
35
36impl Span {
37    fn contains(self, at: i64) -> bool { self.start <= at && at < self.end }
38}
39
40/// The disturbed span the schedule plans for `fault`, for a load that
41/// started at `load_start_ms`.
42#[must_use]
43pub fn planned_span(contract: &Contract, fault: &ScheduledFault, load_start_ms: i64) -> Span {
44    Span {
45        start: load_start_ms + fault.start.millis_i64(),
46        end: load_start_ms
47            + (fault.end() + fault.recovery_budget(&contract.deadlines)).millis_i64(),
48    }
49}
50
51/// Milliseconds of `load` that no span of `spans` covers.
52#[must_use]
53pub fn steady_millis(load: Span, spans: &[Span]) -> i64 {
54    let mut clipped: Vec<Span> = spans
55        .iter()
56        .map(|span| Span {
57            start: span.start.max(load.start),
58            end: span.end.min(load.end),
59        })
60        .filter(|span| span.start < span.end)
61        .collect();
62    clipped.sort_by_key(|span| span.start);
63    let mut covered = 0;
64    let mut reached = load.start;
65    for span in clipped {
66        let start = span.start.max(reached);
67        if span.end > start {
68            covered += span.end - start;
69            reached = span.end;
70        }
71    }
72    (load.end - load.start).max(0) - covered
73}
74
75/// The spans that make time disturbed, shared by every task that classifies
76/// a sample.
77pub struct Windows {
78    planned: Vec<Span>,
79    actual: Mutex<Vec<Span>>,
80    /// The injection instant of the fault now in progress.
81    active_since: Mutex<Option<i64>>,
82}
83
84impl Windows {
85    #[must_use]
86    pub fn new(planned: Vec<Span>) -> Self {
87        Self {
88            planned,
89            actual: Mutex::new(Vec::new()),
90            active_since: Mutex::new(None),
91        }
92    }
93
94    /// Whether a sample taken at `at` is disturbed.
95    pub fn disturbed(&self, at: i64) -> bool {
96        let active = self
97            .active_since
98            .lock()
99            .map_or(true, |since| since.is_some_and(|since| at >= since));
100        let actual = self
101            .actual
102            .lock()
103            .map_or(true, |spans| spans.iter().any(|span| span.contains(at)));
104        active || actual || self.planned.iter().any(|span| span.contains(at))
105    }
106
107    /// A fault was injected at `at`.
108    pub fn injected(&self, at: i64) {
109        if let Ok(mut since) = self.active_since.lock() {
110            *since = Some(at);
111        }
112    }
113
114    /// The fault in progress was cleared; `span` is how long it disturbs.
115    pub fn cleared(&self, span: Span) {
116        if let Ok(mut spans) = self.actual.lock() {
117            spans.push(span);
118        }
119        if let Ok(mut since) = self.active_since.lock() {
120            *since = None;
121        }
122    }
123
124    /// Every span, planned and actual, for the steady-time total.
125    pub fn spans(&self) -> Vec<Span> {
126        let mut spans = self.planned.clone();
127        if let Ok(actual) = self.actual.lock() {
128            spans.extend(actual.iter().copied());
129        }
130        if let Ok(since) = self.active_since.lock()
131            && let Some(since) = *since
132        {
133            spans.push(Span {
134                start: since,
135                end: i64::MAX,
136            });
137        }
138        spans
139    }
140}
141
142/// Latency samples split by window, and the acknowledgements that count
143/// toward the steady rate.
144pub struct Latencies {
145    steady: Histogram<u64>,
146    disturbed: Histogram<u64>,
147    /// Samples taken in steady windows.
148    pub steady_count: u64,
149}
150
151impl Latencies {
152    /// No samples yet.
153    ///
154    /// # Errors
155    ///
156    /// The [`CreationError`] building a latency histogram returned.
157    pub fn new() -> Result<Self, CreationError> {
158        Ok(Self {
159            steady: hist::new()?,
160            disturbed: hist::new()?,
161            steady_count: 0,
162        })
163    }
164
165    pub fn record(&mut self, disturbed: bool, latency: Time) {
166        if disturbed {
167            hist::record(&mut self.disturbed, latency);
168        } else {
169            hist::record(&mut self.steady, latency);
170            self.steady_count += 1;
171        }
172    }
173
174    pub fn merge(&mut self, other: &Self) {
175        // Both sides were built by `hist::new`, so their bounds agree and the
176        // addition cannot fail.
177        let _ = self.steady.add(&other.steady);
178        let _ = self.disturbed.add(&other.disturbed);
179        self.steady_count += other.steady_count;
180    }
181
182    fn steady(&self) -> LatencySummary { summary(&self.steady) }
183
184    fn disturbed(&self) -> LatencySummary { summary(&self.disturbed) }
185}
186
187fn summary(histogram: &Histogram<u64>) -> LatencySummary {
188    let percentiles = hist::percentiles(histogram);
189    LatencySummary {
190        p50: percentiles.p50,
191        p99: percentiles.p99,
192        p999: percentiles.p999,
193        count: percentiles.count,
194    }
195}
196
197/// A path's measurement from its producer's and consumers' samples.
198///
199/// The steady rate is the producer's steady acknowledgements over the steady
200/// time of the load.
201#[must_use]
202pub fn measurement(
203    path: &str,
204    produce: &Latencies,
205    consume: &Latencies,
206    steady: Time,
207) -> PathMeasurement {
208    let acks = f64::from(u32::try_from(produce.steady_count).unwrap_or(u32::MAX));
209    let seconds = steady.secs_f64();
210    PathMeasurement {
211        path: path.to_owned(),
212        produce_steady: produce.steady(),
213        produce_disturbed: produce.disturbed(),
214        consume_steady: consume.steady(),
215        consume_disturbed: consume.disturbed(),
216        steady_rate: Frequency::from_per_sec(if seconds > 0.0 { acks / seconds } else { 0.0 }),
217    }
218}
219
220#[cfg(test)]
221mod tests {
222    use {
223        super::{Latencies, Span, Windows, measurement, planned_span, steady_millis},
224        crate::soak::{
225            contract::{Contract, FaultKind, Target, fixture::SELF_CHECK},
226            schedule::ScheduledFault,
227        },
228        assert2::check,
229        krabka_units::prelude::*,
230    };
231
232    const fn span(start: i64, end: i64) -> Span { Span { start, end } }
233
234    #[yare::parameterized(
235        nothing_disturbed = { &[], 100 },
236        one_inside = { &[span(10, 30)], 80 },
237        overlapping = { &[span(10, 30), span(20, 40)], 70 },
238        nested = { &[span(10, 50), span(20, 30)], 60 },
239        unsorted_disjoint = { &[span(60, 70), span(10, 20)], 80 },
240        clipped_at_both_ends = { &[span(-50, 10), span(90, 500)], 80 },
241        outside = { &[span(200, 300)], 100 },
242        everything = { &[span(0, 100)], 0 },
243    )]
244    fn steady_time_is_the_load_minus_every_span(spans: &[Span], steady: i64) {
245        check!(steady_millis(span(0, 100), spans) == steady);
246    }
247
248    #[test]
249    fn a_planned_span_runs_to_the_longest_deadline_after_cleanup() {
250        let mut contract = Contract::from_yaml(SELF_CHECK).unwrap();
251        contract.deadlines.leader_failover = secs(20);
252        contract.deadlines.isr_rejoin = secs(30);
253        let kind = FaultKind::Restart { graceful: false };
254        let restart = ScheduledFault {
255            id: "restart#0".into(),
256            plan: "restart".into(),
257            checks: kind.checks(Target::Leader),
258            kind,
259            target: Target::Leader,
260            start: secs(140),
261            duration: secs(5),
262        };
263        // Measured for leader_failover (20s) and isr_rejoin (30s): disturbed
264        // from 140s until 30s after its cleanup at 145s.
265        check!(planned_span(&contract, &restart, 1_000) == span(141_000, 176_000));
266    }
267
268    #[yare::parameterized(
269        before = { 5, false },
270        planned = { 15, true },
271        between = { 25, false },
272        actual = { 45, true },
273        active = { 70, true },
274    )]
275    fn a_sample_is_disturbed_inside_any_span_or_a_live_fault(at: i64, disturbed: bool) {
276        let windows = Windows::new(vec![span(10, 20)]);
277        windows.injected(40);
278        windows.cleared(span(40, 50));
279        windows.injected(60);
280        check!(windows.disturbed(at) == disturbed);
281    }
282
283    #[test]
284    fn a_cleared_fault_stops_disturbing_after_its_span() {
285        let windows = Windows::new(Vec::new());
286        windows.injected(40);
287        check!(windows.disturbed(1_000));
288        windows.cleared(span(40, 50));
289        check!(!windows.disturbed(1_000));
290        check!(windows.spans() == [span(40, 50)]);
291    }
292
293    #[test]
294    fn the_steady_rate_counts_only_steady_acknowledgements() {
295        let mut produce = Latencies::new().unwrap();
296        for _ in 0..20 {
297            produce.record(false, millis(4));
298        }
299        for _ in 0..7 {
300            produce.record(true, secs(3));
301        }
302        let mut consume = Latencies::new().unwrap();
303        let mut other = Latencies::new().unwrap();
304        other.record(false, millis(9));
305        consume.merge(&other);
306        let measured = measurement("p", &produce, &consume, secs(10));
307        check!(measured.produce_steady.count == 20);
308        check!(measured.produce_disturbed.count == 7);
309        check!(measured.produce_steady.p999 < millis(5));
310        check!(measured.produce_disturbed.p50 > secs(2));
311        check!(measured.consume_steady.count == 1);
312        check!(measured.consume_disturbed.count == 0);
313        check!((measured.steady_rate.per_sec_f64() - 2.0).abs() < 1.0e-9);
314    }
315}
krabka_bench_driver/soak/client/windows.rs

krabka_bench_driver/soak/client/
windows.rs
