prns_runtime/manifold/
throughput.rs

1//! The interface data-rate meter: the *active* transfer rate (while bytes are actually moving, how fast is the pipe?) averaged over the last few data events, not over wall-clock time. For example, a 15 kB burst that finishes in half a second moved at roughly 30 kB/s; no activity for another half-second still means 30 kB/s, not 15.
2
3use crate::engine::InstantMillis;
4use crate::interfaces::TransferRates;
5
6const CONSIDERED_IDLE_AFTER_MS: u64 = 2_000;
7const SAMPLE_WINDOW: usize = 8;
8
9struct ActiveRate {
10    last_ms: u64,
11    seen: bool,
12    samples: [u32; SAMPLE_WINDOW],
13    /// Number of samples taken, capped at `SAMPLE_WINDOW`; the mean is over this many slots.
14    filled: usize,
15    /// Next slot to overwrite once the ring is full.
16    head: usize,
17}
18
19impl ActiveRate {
20    const fn new() -> Self {
21        Self {
22            last_ms: 0,
23            seen: false,
24            samples: [0; SAMPLE_WINDOW],
25            filled: 0,
26            head: 0,
27        }
28    }
29
30    fn record(&mut self, now: InstantMillis, bytes: u64) {
31        let now = now.0;
32        let dt = if self.seen {
33            now.saturating_sub(self.last_ms)
34                .clamp(1, CONSIDERED_IDLE_AFTER_MS)
35        } else {
36            CONSIDERED_IDLE_AFTER_MS
37        };
38        let sample = u32::try_from(bytes.saturating_mul(8_000) / dt).unwrap_or(u32::MAX);
39        self.samples[self.head] = sample;
40        self.head = (self.head + 1) % SAMPLE_WINDOW;
41        self.filled = (self.filled + 1).min(SAMPLE_WINDOW);
42        self.seen = true;
43        self.last_ms = now;
44    }
45
46    /// The mean of the recent timing samples. Held between bursts; `0` only before any data has moved.
47    fn rate(&self) -> u32 {
48        if self.filled == 0 {
49            return 0;
50        }
51        let sum: u64 = self.samples[..self.filled]
52            .iter()
53            .map(|&sample| u64::from(sample))
54            .sum();
55        u32::try_from(sum / self.filled as u64).unwrap_or(u32::MAX)
56    }
57}
58
59pub struct ThroughputLedger {
60    rx: ActiveRate,
61    tx: ActiveRate,
62}
63
64impl Default for ThroughputLedger {
65    fn default() -> Self {
66        Self::new()
67    }
68}
69
70impl ThroughputLedger {
71    #[must_use]
72    pub const fn new() -> Self {
73        Self {
74            rx: ActiveRate::new(),
75            tx: ActiveRate::new(),
76        }
77    }
78
79    pub fn record_rx(&mut self, now: InstantMillis, bytes: u64) {
80        self.rx.record(now, bytes);
81    }
82
83    pub fn record_tx(&mut self, now: InstantMillis, bytes: u64) {
84        self.tx.record(now, bytes);
85    }
86
87    pub fn rates(&self) -> TransferRates {
88        TransferRates {
89            rx_bps: self.rx.rate(),
90            tx_bps: self.tx.rate(),
91        }
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98
99    #[test]
100    fn rate_is_the_active_transfer_rate_not_a_wall_clock_average() {
101        let mut ledger = ThroughputLedger::new();
102        ledger.record_rx(InstantMillis(0), 1_000);
103        ledger.record_rx(InstantMillis(100), 1_000);
104        assert_eq!(ledger.rates().rx_bps, 42_000);
105    }
106
107    #[test]
108    fn a_quick_burst_reads_its_real_rate_not_diluted_by_surrounding_idle() {
109        let mut ledger = ThroughputLedger::new();
110        let mut t = 1_000;
111        ledger.record_tx(InstantMillis(t), 1_500);
112        for _ in 0..9 {
113            t += 10;
114            ledger.record_tx(InstantMillis(t), 1_500);
115        }
116        assert_eq!(ledger.rates().tx_bps, 1_200_000);
117    }
118
119    #[test]
120    fn the_rate_is_the_mean_of_the_recent_samples() {
121        let mut ledger = ThroughputLedger::new();
122        ledger.record_rx(InstantMillis(0), 1_000);
123        ledger.record_rx(InstantMillis(100), 1_000);
124        ledger.record_rx(InstantMillis(300), 1_000);
125        assert_eq!(ledger.rates().rx_bps, 41_333);
126    }
127
128    #[test]
129    fn the_rate_is_held_between_bursts_not_dropped_to_zero() {
130        let mut ledger = ThroughputLedger::new();
131        ledger.record_rx(InstantMillis(0), 1_000);
132        ledger.record_rx(InstantMillis(100), 1_000);
133        assert_eq!(ledger.rates().rx_bps, 42_000);
134    }
135
136    #[test]
137    fn an_event_after_an_idle_gap_samples_the_floor_not_the_smear() {
138        let mut ledger = ThroughputLedger::new();
139        ledger.record_tx(InstantMillis(0), 1_000);
140        ledger.record_tx(InstantMillis(100), 1_000);
141        ledger.record_tx(InstantMillis(5_000), 1_000);
142        assert_eq!(ledger.rates().tx_bps, 29_333);
143        ledger.record_tx(InstantMillis(5_100), 1_000);
144        assert_eq!(ledger.rates().tx_bps, 42_000);
145    }
146
147    #[test]
148    fn sparse_lone_frames_read_a_nonzero_floor() {
149        let mut ledger = ThroughputLedger::new();
150        ledger.record_tx(InstantMillis(1_000), 167);
151        ledger.record_tx(InstantMillis(181_000), 167);
152        ledger.record_tx(InstantMillis(361_000), 167);
153        assert_eq!(ledger.rates().tx_bps, 668);
154        assert_eq!(ledger.rates().rx_bps, 0, "nothing was ever received");
155    }
156}
prns_runtime/manifold/throughput.rs

prns_runtime/manifold/
throughput.rs
