ntex_rt/
metrics.rs

1use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
2use std::sync::{Arc, OnceLock};
3use std::time::{Duration, Instant};
4
5use futures_timer::Delay;
6
7/// Number of [`Histogram`] buckets.
8pub const HISTOGRAM_BUCKETS: usize = 16;
9
10/// upper bound of the first bucket, in microseconds
11const FIRST_BUCKET_US: u64 = 16;
12
13/// Snapshot of a duration histogram.
14///
15/// Bucket `0` counts durations below 16µs, bucket `i` counts durations in
16/// `[16µs * 2^(i-1), 16µs * 2^i)`, and the last bucket counts durations of
17/// 262.144ms and longer.
18///
19/// Counts are totals since the histogram was created, the distribution over an
20/// interval is the [`delta`](Self::delta) of two snapshots.
21#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
22pub struct Histogram {
23    buckets: [u64; HISTOGRAM_BUCKETS],
24}
25
26impl Histogram {
27    /// Returns the bucket counts.
28    pub fn buckets(&self) -> &[u64; HISTOGRAM_BUCKETS] {
29        &self.buckets
30    }
31
32    /// Returns the total number of recorded durations.
33    pub fn count(&self) -> u64 {
34        self.buckets.iter().fold(0, |acc, c| acc.wrapping_add(*c))
35    }
36
37    /// Returns the duration range of a bucket.
38    ///
39    /// The lower bound is inclusive, the upper bound is exclusive and `None`
40    /// for the last bucket.
41    ///
42    /// # Panics
43    ///
44    /// Panics if `idx` is not less than [`HISTOGRAM_BUCKETS`].
45    pub fn bucket_range(idx: usize) -> (Duration, Option<Duration>) {
46        assert!(idx < HISTOGRAM_BUCKETS, "bucket index out of range");
47        let upper = |i: usize| Duration::from_micros(FIRST_BUCKET_US << i);
48        let lower = if idx == 0 { Duration::ZERO } else { upper(idx - 1) };
49        if idx == HISTOGRAM_BUCKETS - 1 {
50            (lower, None)
51        } else {
52            (lower, Some(upper(idx)))
53        }
54    }
55
56    /// Returns the counts recorded since the `prev` snapshot.
57    #[must_use]
58    pub fn delta(&self, prev: &Histogram) -> Histogram {
59        let mut buckets = self.buckets;
60        for (b, p) in buckets.iter_mut().zip(prev.buckets.iter()) {
61            *b = b.wrapping_sub(*p);
62        }
63        Histogram { buckets }
64    }
65
66    /// Returns the sum of two histograms, for example to combine the
67    /// histograms of several runtimes.
68    #[must_use]
69    pub fn merge(&self, other: &Histogram) -> Histogram {
70        let mut buckets = self.buckets;
71        for (b, o) in buckets.iter_mut().zip(other.buckets.iter()) {
72            *b = b.wrapping_add(*o);
73        }
74        Histogram { buckets }
75    }
76}
77
78/// Duration histogram updated by a single thread.
79#[derive(Debug, Default)]
80pub(crate) struct AtomicHistogram {
81    buckets: [AtomicU64; HISTOGRAM_BUCKETS],
82}
83
84impl AtomicHistogram {
85    fn bucket(d: Duration) -> usize {
86        let us = u64::try_from(d.as_micros()).unwrap_or(u64::MAX);
87        if us < FIRST_BUCKET_US {
88            0
89        } else {
90            // 16µs..32µs is bucket 1
91            let idx = (63 - us.leading_zeros() as usize) - 3;
92            idx.min(HISTOGRAM_BUCKETS - 1)
93        }
94    }
95
96    /// Records a duration, must be called by a single thread.
97    pub(crate) fn record(&self, d: Duration) {
98        let b = &self.buckets[Self::bucket(d)];
99        b.store(b.load(Ordering::Relaxed).wrapping_add(1), Ordering::Relaxed);
100    }
101
102    pub(crate) fn snapshot(&self) -> Histogram {
103        let mut buckets = [0; HISTOGRAM_BUCKETS];
104        for (b, a) in buckets.iter_mut().zip(self.buckets.iter()) {
105            *b = a.load(Ordering::Relaxed);
106        }
107        Histogram { buckets }
108    }
109}
110
111/// Snapshot of an arbiter's scheduling latency.
112///
113/// The arbiter runs a task that sleeps for `interval` and records how late it
114/// wakes up. Lateness includes the time the runtime spent running other tasks
115/// and handling driver events, and the timer's own wake-up latency.
116#[derive(Copy, Clone, Debug, PartialEq, Eq)]
117#[non_exhaustive]
118pub struct LatencyMetrics {
119    /// Sampling interval.
120    pub interval: Duration,
121    /// Lateness of the most recent sample.
122    pub last: Duration,
123    /// Lateness of all samples.
124    pub histogram: Histogram,
125    /// Time of the most recent sample, `None` if no sample has been recorded.
126    ///
127    /// A time much older than `interval` means the arbiter is blocked.
128    pub last_sample: Option<Instant>,
129    /// Number of stalls detected by the watchdog.
130    ///
131    /// An arbiter is stalled when it has not recorded a sample for longer
132    /// than `interval` plus the
133    /// [`ping_threshold`](crate::Builder::ping_threshold).
134    pub stalls: u64,
135    /// Total duration of finished stalls.
136    pub stall_duration: Duration,
137    /// Whether the arbiter is currently stalled.
138    pub stalled: bool,
139}
140
141/// Change of an arbiter stall state.
142#[derive(Debug, PartialEq, Eq)]
143pub(crate) enum Stall {
144    None,
145    /// Arbiter stalled, time since the last sample
146    Started(Duration),
147    /// Arbiter recovered, duration of the stall
148    Recovered(Duration),
149}
150
151#[derive(Debug, Default)]
152pub(crate) struct LatencyStats {
153    // in nanoseconds, zero if sampling is disabled
154    interval: AtomicU64,
155    last: AtomicU64,
156    // nanoseconds since `base`, plus one, zero before the first sample
157    last_sample: AtomicU64,
158    base: OnceLock<Instant>,
159    histogram: AtomicHistogram,
160    // written by the watchdog thread only
161    stalls: AtomicU64,
162    stall_duration: AtomicU64,
163    stalled: AtomicBool,
164    stalled_at: AtomicU64,
165}
166
167fn as_nanos(d: Duration) -> u64 {
168    u64::try_from(d.as_nanos()).unwrap_or(u64::MAX)
169}
170
171impl LatencyStats {
172    fn record(&self, lateness: Duration, now: Instant) {
173        let base = *self.base.get_or_init(|| now);
174        self.histogram.record(lateness);
175        self.last.store(as_nanos(lateness), Ordering::Relaxed);
176        self.last_sample
177            .store(as_nanos(now - base).saturating_add(1), Ordering::Relaxed);
178    }
179
180    pub(crate) fn metrics(&self) -> Option<LatencyMetrics> {
181        let interval = self.interval.load(Ordering::Relaxed);
182        if interval == 0 {
183            return None;
184        }
185        let last_sample = match self.last_sample.load(Ordering::Relaxed) {
186            0 => None,
187            n => self
188                .base
189                .get()
190                .map(|base| *base + Duration::from_nanos(n - 1)),
191        };
192        Some(LatencyMetrics {
193            interval: Duration::from_nanos(interval),
194            last: Duration::from_nanos(self.last.load(Ordering::Relaxed)),
195            histogram: self.histogram.snapshot(),
196            last_sample,
197            stalls: self.stalls.load(Ordering::Relaxed),
198            stall_duration: Duration::from_nanos(self.stall_duration.load(Ordering::Relaxed)),
199            stalled: self.stalled.load(Ordering::Relaxed),
200        })
201    }
202
203    /// Checks whether the arbiter stopped recording samples.
204    ///
205    /// Must be called from a single thread.
206    pub(crate) fn check(&self, now: Instant, threshold: Duration) -> Stall {
207        let interval = Duration::from_nanos(self.interval.load(Ordering::Relaxed));
208        let Some(base) = self.base.get() else {
209            return Stall::None;
210        };
211        if interval.is_zero() {
212            return Stall::None;
213        }
214
215        // sampler start time until the first sample
216        let last = self.last_sample.load(Ordering::Relaxed).saturating_sub(1);
217        if self.stalled.load(Ordering::Relaxed) {
218            let prev = self.stalled_at.load(Ordering::Relaxed);
219            if last == prev {
220                return Stall::None;
221            }
222            let gap = Duration::from_nanos(last.saturating_sub(prev));
223            let stall = gap.saturating_sub(interval);
224            self.stalled.store(false, Ordering::Relaxed);
225            self.stall_duration.store(
226                self.stall_duration
227                    .load(Ordering::Relaxed)
228                    .saturating_add(as_nanos(stall)),
229                Ordering::Relaxed,
230            );
231            Stall::Recovered(stall)
232        } else {
233            let age = now.saturating_duration_since(*base + Duration::from_nanos(last));
234            if age <= interval + threshold {
235                return Stall::None;
236            }
237            self.stalled_at.store(last, Ordering::Relaxed);
238            self.stalled.store(true, Ordering::Relaxed);
239            self.stalls.store(
240                self.stalls.load(Ordering::Relaxed).wrapping_add(1),
241                Ordering::Relaxed,
242            );
243            Stall::Started(age)
244        }
245    }
246}
247
248/// Samples the scheduling latency of the current arbiter.
249pub(crate) fn start_latency_sampler(stats: &Arc<LatencyStats>, interval_ms: usize) {
250    if interval_ms == 0 {
251        return;
252    }
253    let interval = Duration::from_millis(interval_ms as u64);
254    stats.base.get_or_init(Instant::now);
255    stats.interval.store(as_nanos(interval), Ordering::Relaxed);
256
257    let stats = stats.clone();
258    crate::spawn_detached(async move {
259        loop {
260            let deadline = Instant::now() + interval;
261            Delay::new(interval).await;
262            let now = Instant::now();
263            stats.record(now.saturating_duration_since(deadline), now);
264        }
265    });
266}
267
268#[cfg(test)]
269mod tests {
270    use super::*;
271
272    #[test]
273    fn histogram_buckets() {
274        let bucket = |us| AtomicHistogram::bucket(Duration::from_micros(us));
275        assert_eq!(bucket(0), 0);
276        assert_eq!(bucket(15), 0);
277        assert_eq!(bucket(16), 1);
278        assert_eq!(bucket(31), 1);
279        assert_eq!(bucket(32), 2);
280        assert_eq!(bucket(262_143), 14);
281        assert_eq!(bucket(262_144), 15);
282        assert_eq!(AtomicHistogram::bucket(Duration::MAX), 15);
283
284        for i in 0..HISTOGRAM_BUCKETS {
285            let (lower, upper) = Histogram::bucket_range(i);
286            assert_eq!(AtomicHistogram::bucket(lower), i);
287            if let Some(upper) = upper {
288                assert_eq!(AtomicHistogram::bucket(upper), i + 1);
289                assert_eq!(
290                    AtomicHistogram::bucket(upper.checked_sub(Duration::from_nanos(1)).unwrap()),
291                    i
292                );
293            }
294        }
295        assert_eq!(Histogram::bucket_range(15).1, None);
296    }
297
298    #[test]
299    fn histogram_delta() {
300        let h = AtomicHistogram::default();
301        h.record(Duration::from_micros(1));
302        let prev = h.snapshot();
303        h.record(Duration::from_micros(1));
304        h.record(Duration::from_millis(1));
305        let cur = h.snapshot();
306        assert_eq!(cur.count(), 3);
307
308        let delta = cur.delta(&prev);
309        assert_eq!(delta.count(), 2);
310        assert_eq!(delta.buckets()[0], 1);
311        assert_eq!(
312            delta.buckets()[AtomicHistogram::bucket(Duration::from_millis(1))],
313            1
314        );
315        assert_eq!(prev.merge(&delta), cur);
316    }
317
318    #[test]
319    fn latency_stats() {
320        let stats = LatencyStats::default();
321        assert_eq!(stats.metrics(), None);
322
323        stats.interval.store(100, Ordering::Relaxed);
324        let m = stats.metrics().unwrap();
325        assert_eq!((m.last_sample, m.histogram.count()), (None, 0));
326
327        let now = Instant::now();
328        stats.record(Duration::from_millis(2), now);
329        let m = stats.metrics().unwrap();
330        assert_eq!(m.interval, Duration::from_nanos(100));
331        assert_eq!(m.last, Duration::from_millis(2));
332        assert_eq!(m.last_sample, Some(now));
333        assert_eq!(m.histogram.count(), 1);
334    }
335
336    #[test]
337    fn stall_check() {
338        let ms = Duration::from_millis;
339        let threshold = ms(20);
340        let stats = LatencyStats::default();
341        let t0 = Instant::now();
342        assert_eq!(stats.check(t0 + ms(100), threshold), Stall::None);
343
344        stats.base.set(t0).unwrap();
345        stats.interval.store(as_nanos(ms(10)), Ordering::Relaxed);
346        assert_eq!(stats.check(t0 + ms(30), threshold), Stall::None);
347        assert_eq!(stats.check(t0 + ms(31), threshold), Stall::Started(ms(31)));
348        assert_eq!(stats.check(t0 + ms(50), threshold), Stall::None);
349        let m = stats.metrics().unwrap();
350        assert_eq!((m.stalls, m.stall_duration, m.stalled), (1, ms(0), true));
351
352        stats.record(ms(90), t0 + ms(100));
353        assert_eq!(
354            stats.check(t0 + ms(101), threshold),
355            Stall::Recovered(ms(90))
356        );
357        assert_eq!(stats.check(t0 + ms(120), threshold), Stall::None);
358        let m = stats.metrics().unwrap();
359        assert_eq!((m.stalls, m.stall_duration, m.stalled), (1, ms(90), false));
360
361        assert_eq!(stats.check(t0 + ms(131), threshold), Stall::Started(ms(31)));
362        assert_eq!(stats.metrics().unwrap().stalls, 2);
363    }
364}
ntex_rt/metrics.rs

ntex_rt/
metrics.rs
