cronoq_core/
stats.rs

1//! Streaming statistics: EWMA, quantile estimates, log histograms and windowed metrics.
2
3use core::time::Duration;
4
5use crate::time::Time;
6
7/// Exponentially weighted moving average.
8#[derive(Copy, Clone, Debug, Default)]
9pub struct Ewma {
10    value: f64,
11    init: bool,
12}
13
14impl Ewma {
15    /// Add a sample with weight `alpha` (the first sample initializes the average).
16    #[inline]
17    pub fn update(&mut self, x: f64, alpha: f64) {
18        if self.init {
19            self.value += alpha * (x - self.value);
20        } else {
21            self.value = x;
22            self.init = true;
23        }
24    }
25    /// Current average, or `None` before the first sample.
26    #[inline]
27    pub fn get(&self) -> Option<f64> {
28        self.init.then_some(self.value)
29    }
30}
31
32/// Streaming quantile estimate (stochastic approximation with adaptive step).
33#[derive(Copy, Clone, Debug)]
34pub struct QuantileEst {
35    q: f64,
36    p: f64,
37    n: u64,
38}
39
40impl QuantileEst {
41    /// Estimator for quantile `p` (e.g. `0.9`).
42    pub fn new(p: f64) -> Self {
43        QuantileEst { q: 0.0, p, n: 0 }
44    }
45    /// Add a sample; `scale` is the typical magnitude of the data (sets the step size).
46    #[inline]
47    pub fn update(&mut self, x: f64, scale: f64) {
48        self.n += 1;
49        if self.n == 1 {
50            self.q = x;
51            return;
52        }
53        // Step proportional to the scale of the data; decays to a floor so it tracks drift.
54        let decay = if self.n > 2500 {
55            0.02
56        } else {
57            (1.0 / (self.n as f64).sqrt()).max(0.02)
58        };
59        let step = scale.max(1e-9) * decay;
60        if x > self.q {
61            self.q += step * self.p;
62        } else if x < self.q {
63            self.q -= step * (1.0 - self.p);
64        }
65        if self.q < 0.0 {
66            self.q = 0.0;
67        }
68    }
69    /// Current estimate, or `None` before the first sample.
70    #[inline]
71    pub fn get(&self) -> Option<f64> {
72        (self.n > 0).then_some(self.q)
73    }
74}
75
76/// Observed runtimes for a `(pool, kind)` (seconds).
77#[derive(Copy, Clone, Debug)]
78pub struct DurStats {
79    /// Exponentially weighted mean runtime.
80    pub mean: Ewma,
81    /// Streaming median estimate.
82    pub p50: QuantileEst,
83    /// Streaming 90th-percentile estimate.
84    pub p90: QuantileEst,
85    /// Number of samples observed.
86    pub count: u64,
87}
88
89impl Default for DurStats {
90    fn default() -> Self {
91        DurStats {
92            mean: Ewma::default(),
93            p50: QuantileEst::new(0.5),
94            p90: QuantileEst::new(0.9),
95            count: 0,
96        }
97    }
98}
99
100impl DurStats {
101    /// Record one runtime, in seconds.
102    pub fn observe(&mut self, secs: f64) {
103        self.count += 1;
104        self.mean.update(secs, 0.1);
105        let scale = self.mean.get().unwrap_or(secs);
106        self.p50.update(secs, scale);
107        self.p90.update(secs, scale);
108    }
109    /// Typical runtime: the median once there are a few samples, the mean before that.
110    pub fn estimate(&self) -> Option<Duration> {
111        if self.count < 3 {
112            return self.mean.get().map(Duration::from_secs_f64);
113        }
114        self.p50.get().map(Duration::from_secs_f64)
115    }
116    /// 90th-percentile runtime (the mean while there are fewer than three samples).
117    pub fn p90(&self) -> Option<Duration> {
118        if self.count < 3 {
119            return self.mean.get().map(Duration::from_secs_f64);
120        }
121        self.p90.get().map(Duration::from_secs_f64)
122    }
123}
124
125/// Lightweight runtime tracker when only a p90 is needed (the hold lead for fixed starts).
126#[derive(Copy, Clone, Debug)]
127pub(crate) struct P90Stats {
128    mean: Ewma,
129    p90: QuantileEst,
130    count: u64,
131}
132
133impl Default for P90Stats {
134    fn default() -> Self {
135        P90Stats {
136            mean: Ewma::default(),
137            p90: QuantileEst::new(0.9),
138            count: 0,
139        }
140    }
141}
142
143impl P90Stats {
144    #[inline]
145    pub(crate) fn observe(&mut self, secs: f64) {
146        self.count += 1;
147        self.mean.update(secs, 0.1);
148        let scale = self.mean.get().unwrap_or(secs);
149        self.p90.update(secs, scale);
150    }
151    pub(crate) fn p90(&self) -> Option<Duration> {
152        let v = if self.count < 3 {
153            self.mean.get()
154        } else {
155            self.p90.get()
156        };
157        v.map(Duration::from_secs_f64)
158    }
159}
160
161/// Log-linear histogram over nanoseconds (8 sub-buckets per power of two, ~9% resolution).
162#[derive(Clone, Debug, Default)]
163pub struct Histogram {
164    counts: Vec<u64>,
165    total: u64,
166    sum_ns: u128,
167}
168
169const SUB: u32 = 8;
170const BUCKETS: usize = 64 * SUB as usize;
171
172impl Histogram {
173    #[inline]
174    fn index(ns: u64) -> usize {
175        if ns < SUB as u64 {
176            return ns as usize;
177        }
178        let msb = 63 - ns.leading_zeros(); // >= 3
179        let shift = msb - 3;
180        let sub = ((ns >> shift) & (SUB as u64 - 1)) as u32;
181        ((msb - 2) * SUB + sub) as usize
182    }
183    #[inline]
184    fn lower_bound(idx: usize) -> u64 {
185        let idx = idx as u32;
186        if idx < SUB {
187            return idx as u64;
188        }
189        let msb = idx / SUB + 2;
190        let sub = idx % SUB;
191        // Saturates past the last bucket, whose upper bound is beyond u64.
192        let v = ((SUB + sub) as u128) << (msb - 3);
193        v.min(u64::MAX as u128) as u64
194    }
195    /// Record one duration.
196    pub fn record(&mut self, d: Duration) {
197        let ns = d.as_nanos().min(u64::MAX as u128) as u64;
198        if self.counts.is_empty() {
199            self.counts = vec![0; BUCKETS];
200        }
201        self.counts[Self::index(ns)] += 1;
202        self.total += 1;
203        self.sum_ns += ns as u128;
204    }
205    /// Add all samples of `other`.
206    pub fn merge(&mut self, other: &Histogram) {
207        if other.total == 0 {
208            return;
209        }
210        if self.counts.is_empty() {
211            self.counts = vec![0; BUCKETS];
212        }
213        for (a, b) in self.counts.iter_mut().zip(&other.counts) {
214            *a += *b;
215        }
216        self.total += other.total;
217        self.sum_ns += other.sum_ns;
218    }
219    /// Number of samples.
220    pub fn count(&self) -> u64 {
221        self.total
222    }
223    /// Exact mean of the samples (zero when empty).
224    pub fn mean(&self) -> Duration {
225        if self.total == 0 {
226            return Duration::ZERO;
227        }
228        Duration::from_nanos((self.sum_ns / self.total as u128) as u64)
229    }
230    /// Approximate quantile `q` in `[0, 1]` (bucket midpoint; zero when empty).
231    pub fn quantile(&self, q: f64) -> Duration {
232        if self.total == 0 {
233            return Duration::ZERO;
234        }
235        let target = ((q.clamp(0.0, 1.0) * self.total as f64).ceil() as u64).max(1);
236        let mut acc = 0;
237        for (i, c) in self.counts.iter().enumerate() {
238            acc += c;
239            if acc >= target {
240                let lo = Self::lower_bound(i);
241                let hi = Self::lower_bound(i + 1).max(lo);
242                return Duration::from_nanos(lo + (hi - lo) / 2);
243            }
244        }
245        Duration::ZERO
246    }
247    /// Remove all samples (keeps the allocation).
248    pub fn clear(&mut self) {
249        if self.total > 0 {
250            self.counts.iter_mut().for_each(|c| *c = 0);
251        }
252        self.total = 0;
253        self.sum_ns = 0;
254    }
255}
256
257/// Counters accumulated per time bucket.
258#[derive(Clone, Debug, Default)]
259pub struct Bucket {
260    pub dispatched: u64,
261    pub inversions: u64,
262    pub fixed_total: u64,
263    pub fixed_hits: u64,
264    pub succeeded: u64,
265    pub failed: u64,
266    pub retried: u64,
267    pub busy_secs: f64,
268    pub capacity_secs: f64,
269    pub work_secs: f64,
270    pub wait: Histogram,
271    pub run: Histogram,
272}
273
274impl Bucket {
275    fn clear(&mut self) {
276        let wait = core::mem::take(&mut self.wait);
277        let run = core::mem::take(&mut self.run);
278        *self = Bucket {
279            wait,
280            run,
281            ..Default::default()
282        };
283        self.wait.clear();
284        self.run.clear();
285    }
286    fn merge(&mut self, o: &Bucket) {
287        self.dispatched += o.dispatched;
288        self.inversions += o.inversions;
289        self.fixed_total += o.fixed_total;
290        self.fixed_hits += o.fixed_hits;
291        self.succeeded += o.succeeded;
292        self.failed += o.failed;
293        self.retried += o.retried;
294        self.busy_secs += o.busy_secs;
295        self.capacity_secs += o.capacity_secs;
296        self.work_secs += o.work_secs;
297        self.wait.merge(&o.wait);
298        self.run.merge(&o.run);
299    }
300}
301
302/// Sliding window made of a ring of buckets.
303#[derive(Clone, Debug)]
304pub struct Window {
305    buckets: Vec<Bucket>,
306    bucket_ns: u64,
307    current: u64,
308}
309
310const WINDOW_BUCKETS: usize = 12;
311
312impl Window {
313    pub fn new(window: Duration) -> Self {
314        let bucket_ns = (window.as_nanos() as u64 / WINDOW_BUCKETS as u64).max(1);
315        Window {
316            buckets: vec![Bucket::default(); WINDOW_BUCKETS],
317            bucket_ns,
318            current: 0,
319        }
320    }
321    /// Bucket for `now`, rotating out expired buckets.
322    pub fn at(&mut self, now: Time) -> &mut Bucket {
323        let idx = now.0 / self.bucket_ns;
324        if idx > self.current {
325            let steps = (idx - self.current).min(WINDOW_BUCKETS as u64);
326            for s in 1..=steps {
327                let b = ((self.current + s) % WINDOW_BUCKETS as u64) as usize;
328                self.buckets[b].clear();
329            }
330            self.current = idx;
331        }
332        let b = (self.current % WINDOW_BUCKETS as u64) as usize;
333        &mut self.buckets[b]
334    }
335    pub fn total(&mut self, now: Time) -> Bucket {
336        self.at(now);
337        let mut acc = Bucket::default();
338        for b in &self.buckets {
339            acc.merge(b);
340        }
341        acc
342    }
343    pub fn span(&self) -> Duration {
344        Duration::from_nanos(self.bucket_ns * WINDOW_BUCKETS as u64)
345    }
346}
347
348#[cfg(test)]
349mod tests {
350    use super::*;
351
352    #[test]
353    fn histogram_quantiles_are_close() {
354        let mut h = Histogram::default();
355        for i in 1..=1000u64 {
356            h.record(Duration::from_micros(i));
357        }
358        let p50 = h.quantile(0.5).as_micros() as f64;
359        let p99 = h.quantile(0.99).as_micros() as f64;
360        assert!((p50 - 500.0).abs() / 500.0 < 0.1, "p50={p50}");
361        assert!((p99 - 990.0).abs() / 990.0 < 0.1, "p99={p99}");
362    }
363
364    #[test]
365    fn histogram_index_monotonic() {
366        let mut last = 0;
367        for ns in [0u64, 1, 7, 8, 9, 15, 16, 100, 1_000, 1 << 40, u64::MAX] {
368            let i = Histogram::index(ns);
369            assert!(i >= last);
370            assert!(i < BUCKETS);
371            assert!(Histogram::lower_bound(i) <= ns);
372            last = i;
373        }
374    }
375
376    #[test]
377    fn quantile_estimate_converges() {
378        let mut d = DurStats::default();
379        for i in 0..5000 {
380            d.observe(((i * 7919) % 100) as f64 / 10.0); // uniform over 0..10
381        }
382        let p50 = d.p50.get().unwrap();
383        let p90 = d.p90.get().unwrap();
384        assert!((p50 - 5.0).abs() < 1.0, "p50={p50}");
385        assert!((p90 - 9.0).abs() < 1.0, "p90={p90}");
386    }
387
388    #[test]
389    fn window_expires_old_buckets() {
390        let mut w = Window::new(Duration::from_secs(60));
391        w.at(Time::ZERO).dispatched += 5;
392        assert_eq!(w.total(Time::from_secs_f64(30.0)).dispatched, 5);
393        assert_eq!(w.total(Time::from_secs_f64(120.0)).dispatched, 0);
394    }
395}
cronoq_core/stats.rs

cronoq_core/
stats.rs
