nd_300/diagnostics/
bufferbloat.rs

1//! Real bufferbloat measurement.
2//!
3//! Idle baseline → sustained loaded-download phase → loaded-upload phase,
4//! with ping bursts running CONCURRENTLY with the saturating traffic (the
5//! old implementation slept 500ms into a single 25MB download and called
6//! three pings "loaded latency" — an estimate, not a measurement). Graded
7//! A+–F per direction by the loaded-vs-idle latency delta; overall grade is
8//! the worse of the two.
9
10use serde::Serialize;
11use std::time::Duration;
12
13use super::ping;
14use crate::config::Config;
15
16const TARGET: &str = "1.1.1.1";
17const IDLE_PROBES: u32 = 5;
18const DOWNLOAD_PROBES: u32 = 12;
19const UPLOAD_PROBES: u32 = 8;
20/// Concurrent saturating connections per phase.
21const DOWNLOAD_STREAMS: usize = 4;
22const UPLOAD_STREAMS: usize = 2;
23const DOWNLOAD_URL: &str = "https://speed.cloudflare.com/__down?bytes=100000000";
24const UPLOAD_URL: &str = "https://speed.cloudflare.com/__up";
25const UPLOAD_CHUNK_BYTES: usize = 10_000_000;
26/// Whole-module budget — the phases below are individually bounded, this is
27/// the backstop.
28const MODULE_BUDGET: Duration = Duration::from_secs(35);
29
30#[derive(Debug, Clone, Serialize)]
31pub struct BufferbloatResult {
32    /// Idle median latency (kept field name for JSON compatibility).
33    pub unloaded_latency_ms: f64,
34    /// Download-phase median loaded latency (kept).
35    pub loaded_latency_ms: Option<f64>,
36    /// Overall grade — the worse of the per-direction grades (kept).
37    pub grade: String,
38    /// Download-direction latency delta (kept; same value as
39    /// `download_bloat_ms`).
40    pub bloat_ms: Option<f64>,
41    pub description: String,
42    // ── Additive fields (v3.4.0+) ──
43    #[serde(skip_serializing_if = "Option::is_none")]
44    pub download_bloat_ms: Option<f64>,
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub upload_loaded_latency_ms: Option<f64>,
47    #[serde(skip_serializing_if = "Option::is_none")]
48    pub upload_bloat_ms: Option<f64>,
49    #[serde(skip_serializing_if = "Option::is_none")]
50    pub download_grade: Option<String>,
51    #[serde(skip_serializing_if = "Option::is_none")]
52    pub upload_grade: Option<String>,
53    pub samples_idle: u32,
54    pub samples_download: u32,
55    pub samples_upload: u32,
56}
57
58pub async fn collect(config: &Config) -> Option<BufferbloatResult> {
59    tokio::time::timeout(MODULE_BUDGET, collect_inner(config))
60        .await
61        .unwrap_or_default()
62}
63
64async fn collect_inner(config: &Config) -> Option<BufferbloatResult> {
65    // Idle baseline.
66    let idle_stats = run_burst(IDLE_PROBES).await?;
67    let idle = median(&idle_stats.times_ms)?;
68
69    // Honor --fast: the loaded phases deliberately saturate the link with
70    // ~hundreds of MB of traffic — exactly what --fast asks to avoid. (The
71    // old implementation silently downloaded 25 MB even under --fast.)
72    if config.skip_speed {
73        let grade = if idle < 20.0 {
74            "A"
75        } else if idle < 50.0 {
76            "B"
77        } else {
78            "C"
79        };
80        return Some(BufferbloatResult {
81            unloaded_latency_ms: idle,
82            loaded_latency_ms: None,
83            grade: grade.to_string(),
84            bloat_ms: None,
85            description: "Loaded latency not measured (run without --fast for full test)"
86                .to_string(),
87            download_bloat_ms: None,
88            upload_loaded_latency_ms: None,
89            upload_bloat_ms: None,
90            download_grade: None,
91            upload_grade: None,
92            samples_idle: idle_stats.received(),
93            samples_download: 0,
94            samples_upload: 0,
95        });
96    }
97
98    // Loaded download phase: saturate with concurrent streaming GETs while a
99    // ping burst spans the window.
100    let dl_stats = loaded_phase(Load::Download, DOWNLOAD_PROBES).await;
101    // Loaded upload phase: concurrent looping POSTs.
102    let ul_stats = loaded_phase(Load::Upload, UPLOAD_PROBES).await;
103
104    let dl_median = dl_stats.as_ref().and_then(|s| median(&s.times_ms));
105    let ul_median = ul_stats.as_ref().and_then(|s| median(&s.times_ms));
106
107    let dl_bloat = dl_median.map(|m| m - idle);
108    let ul_bloat = ul_median.map(|m| m - idle);
109
110    let dl_grade = dl_bloat.map(grade_delta);
111    let ul_grade = ul_bloat.map(grade_delta);
112    let overall = worse_grade(dl_grade.unwrap_or("A+"), ul_grade.unwrap_or("A+"));
113
114    let description = if dl_median.is_none() && ul_median.is_none() {
115        "Loaded latency could not be measured".to_string()
116    } else {
117        describe(overall).to_string()
118    };
119
120    Some(BufferbloatResult {
121        unloaded_latency_ms: idle,
122        loaded_latency_ms: dl_median,
123        grade: overall.to_string(),
124        bloat_ms: dl_bloat,
125        description,
126        download_bloat_ms: dl_bloat,
127        upload_loaded_latency_ms: ul_median,
128        upload_bloat_ms: ul_bloat,
129        download_grade: dl_grade.map(|g| g.to_string()),
130        upload_grade: ul_grade.map(|g| g.to_string()),
131        samples_idle: idle_stats.received(),
132        samples_download: dl_stats.map(|s| s.received()).unwrap_or(0),
133        samples_upload: ul_stats.map(|s| s.received()).unwrap_or(0),
134    })
135}
136
137enum Load {
138    Download,
139    Upload,
140}
141
142/// Run a ping burst while concurrent tasks saturate the link in the given
143/// direction. The saturating tasks are aborted as soon as the burst returns.
144async fn loaded_phase(load: Load, probes: u32) -> Option<ping::PingStats> {
145    let client = reqwest::Client::builder()
146        .timeout(Duration::from_secs(30))
147        .build()
148        .ok()?;
149
150    let mut tasks = Vec::new();
151    let streams = match load {
152        Load::Download => DOWNLOAD_STREAMS,
153        Load::Upload => UPLOAD_STREAMS,
154    };
155    for _ in 0..streams {
156        let client = client.clone();
157        let handle = match load {
158            Load::Download => tokio::spawn(async move {
159                loop {
160                    let Ok(mut resp) = client.get(DOWNLOAD_URL).send().await else {
161                        continue;
162                    };
163                    while let Ok(Some(_chunk)) = resp.chunk().await {}
164                }
165            }),
166            Load::Upload => tokio::spawn(async move {
167                let payload = vec![0u8; UPLOAD_CHUNK_BYTES];
168                loop {
169                    let _ = client.post(UPLOAD_URL).body(payload.clone()).send().await;
170                }
171            }),
172        };
173        tasks.push(handle);
174    }
175
176    // Give the streams a moment to ramp, then ping through the load.
177    tokio::time::sleep(Duration::from_millis(750)).await;
178    let stats = run_burst(probes).await;
179
180    for task in tasks {
181        task.abort();
182    }
183
184    stats
185}
186
187async fn run_burst(count: u32) -> Option<ping::PingStats> {
188    let stdout = ping::run_ping(TARGET, count).await?;
189    let stats = ping::parse_ping(&stdout, count);
190    if stats.received() == 0 {
191        None
192    } else {
193        Some(stats)
194    }
195}
196
197fn median(times: &[f64]) -> Option<f64> {
198    if times.is_empty() {
199        return None;
200    }
201    let mut sorted = times.to_vec();
202    sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
203    Some(sorted[sorted.len() / 2])
204}
205
206/// Grade a loaded-vs-idle latency delta (ms). Thresholds unchanged from the
207/// original implementation.
208fn grade_delta(delta_ms: f64) -> &'static str {
209    if delta_ms < 5.0 {
210        "A+"
211    } else if delta_ms < 30.0 {
212        "A"
213    } else if delta_ms < 60.0 {
214        "B"
215    } else if delta_ms < 200.0 {
216        "C"
217    } else if delta_ms < 400.0 {
218        "D"
219    } else {
220        "F"
221    }
222}
223
224fn grade_rank(grade: &str) -> u8 {
225    match grade {
226        "A+" => 0,
227        "A" => 1,
228        "B" => 2,
229        "C" => 3,
230        "D" => 4,
231        _ => 5,
232    }
233}
234
235fn worse_grade<'a>(a: &'a str, b: &'a str) -> &'a str {
236    if grade_rank(a) >= grade_rank(b) {
237        a
238    } else {
239        b
240    }
241}
242
243fn describe(grade: &str) -> &'static str {
244    match grade {
245        "A+" | "A" => "Excellent - minimal bufferbloat",
246        "B" => "Good - minor bufferbloat",
247        "C" => "Fair - moderate bufferbloat",
248        "D" => "Poor - significant bufferbloat",
249        _ => "Severe bufferbloat detected",
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256
257    #[test]
258    fn grade_delta_boundaries() {
259        assert_eq!(grade_delta(4.9), "A+");
260        assert_eq!(grade_delta(5.0), "A");
261        assert_eq!(grade_delta(29.9), "A");
262        assert_eq!(grade_delta(30.0), "B");
263        assert_eq!(grade_delta(59.9), "B");
264        assert_eq!(grade_delta(60.0), "C");
265        assert_eq!(grade_delta(199.9), "C");
266        assert_eq!(grade_delta(200.0), "D");
267        assert_eq!(grade_delta(399.9), "D");
268        assert_eq!(grade_delta(400.0), "F");
269    }
270
271    #[test]
272    fn worse_grade_picks_worse() {
273        assert_eq!(worse_grade("A+", "C"), "C");
274        assert_eq!(worse_grade("F", "A"), "F");
275        assert_eq!(worse_grade("B", "B"), "B");
276    }
277
278    #[test]
279    fn median_is_middle_element() {
280        assert_eq!(median(&[3.0, 1.0, 2.0]), Some(2.0));
281        assert_eq!(median(&[]), None);
282        assert_eq!(median(&[5.0]), Some(5.0));
283    }
284}
nd_300/diagnostics/bufferbloat.rs

nd_300/diagnostics/
bufferbloat.rs
