ntex_rt/
metrics.rs

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

ntex_rt/
metrics.rs
