decdn_node/load_shed/
egress.rs

1//! A lock-free rolling egress meter. Delivery loops call [`EgressMeter::record`]
2//! with each delivered chunk's byte length (the bao-stream content plus
3//! interleaved proof — the same quantity billed as `delivered`, not the full
4//! QUIC frame); a periodic task calls [`EgressMeter::sample`] once per fixed
5//! interval to fold the bytes delivered since the last sample into an
6//! exponentially-weighted bytes/sec rate that the load-shed policy and the
7//! metrics gauge read.
8
9use std::sync::atomic::{AtomicU64, Ordering};
10
11/// Cumulative delivered bytes plus the last computed EWMA rate. All state
12/// is atomic, so `record` stays off any lock on the hot delivery path.
13#[derive(Debug, Default)]
14pub struct EgressMeter {
15    /// Monotonic total delivered bytes.
16    cumulative: AtomicU64,
17    /// `cumulative` as of the last `sample` — the baseline for the next delta.
18    last_sample_cumulative: AtomicU64,
19    /// Whether `sample` has run at least once (seeds the EWMA on the first run).
20    seeded: AtomicU64,
21    /// Last computed bytes/sec EWMA.
22    bps: AtomicU64,
23}
24
25impl EgressMeter {
26    /// An unseeded meter. The first [`Self::sample`] seeds the EWMA rather
27    /// than reporting a rate against an unknown baseline.
28    #[must_use]
29    pub fn new() -> Self {
30        Self::default()
31    }
32
33    /// Add `bytes` of delivered content to the running total. Called per chunk
34    /// with the same length billed as `delivered` (bao content plus proof).
35    pub fn record(&self, bytes: u64) {
36        self.cumulative.fetch_add(bytes, Ordering::Relaxed);
37    }
38
39    /// Fold the bytes delivered since the last call into the EWMA and return the
40    /// new bytes/sec. `interval_secs` is the fixed tick period; `0` is a no-op
41    /// that returns the prior rate.
42    pub fn sample(&self, interval_secs: u64) -> u64 {
43        if interval_secs == 0 {
44            return self.bps.load(Ordering::Relaxed);
45        }
46        let now = self.cumulative.load(Ordering::Relaxed);
47        let prev = self.last_sample_cumulative.swap(now, Ordering::Relaxed);
48        let delta = now.saturating_sub(prev);
49        let instant = delta / interval_secs;
50        // Seed on the first sample; afterwards halve toward the instant rate
51        // (EWMA, alpha = 0.5) using integer math to stay off floats.
52        let next = if self.seeded.swap(1, Ordering::Relaxed) == 0 {
53            instant
54        } else {
55            let last = self.bps.load(Ordering::Relaxed);
56            (instant.saturating_add(last)) / 2
57        };
58        self.bps.store(next, Ordering::Relaxed);
59        next
60    }
61
62    /// The most recently computed bytes/sec, read on the admission hot path.
63    #[must_use]
64    pub fn current_bps(&self) -> u64 {
65        self.bps.load(Ordering::Relaxed)
66    }
67}
68
69#[cfg(test)]
70mod tests;
decdn_node/load_shed/egress.rs

decdn_node/load_shed/
egress.rs
