Skip to main content

max / makenotwork

5.5 KB · 152 lines History Blame Raw
1 //! How long it takes to get onto the blocking pool, sampled through the run.
2 //!
3 //! The description layer's whole runtime cost is that quasi's router is sync, so
4 //! quasi-axum dispatches every described request on `spawn_blocking` and the
5 //! handler holds that thread across its database or upstream call. This server
6 //! shares that pool with argon2 hashing, content exports and the file scanner,
7 //! so the question a load run has to answer is whether described requests starve
8 //! them.
9 //!
10 //! Throughput does not answer it and neither does per-endpoint latency: a pool
11 //! at its limit shows up as everything getting slower together, which is
12 //! indistinguishable from the box being busy. What separates them is DISPATCH
13 //! DELAY — how long a fresh `spawn_blocking` waits before its closure starts
14 //! running. On an unsaturated pool that is microseconds whatever the load;
15 //! it only grows when every thread is occupied and tokio has to wait for one or
16 //! spawn another.
17 //!
18 //! `RuntimeMetrics::num_blocking_threads` would say it directly, but it is
19 //! behind `tokio_unstable` and this tree does not build with it. This measures
20 //! the same saturation from the outside, with no cfg and no build flag: the
21 //! probe is what a request arriving at that instant would have experienced.
22 //!
23 //! Read it alongside the argon2 signup mean (`POST /join/step/account`), which
24 //! is the same fact seen from the other end: signup is the heaviest genuine
25 //! blocking-pool user in the mix, so if described requests are crowding the pool
26 //! both numbers move together.
27
28 use std::sync::{Arc, Mutex};
29 use std::time::{Duration, Instant};
30
31 /// Samples dispatch delay until it is stopped.
32 pub(super) struct BlockingProbe {
33 samples: Arc<Mutex<Vec<Duration>>>,
34 sampler: tokio::task::JoinHandle<()>,
35 }
36
37 impl BlockingProbe {
38 /// Start sampling every `interval`.
39 ///
40 /// One in-flight sample at a time, on purpose: the probe measures the pool,
41 /// and a probe that queued its own work would be measuring itself.
42 pub(super) fn start(interval: Duration) -> Self {
43 let samples = Arc::new(Mutex::new(Vec::new()));
44 let into = Arc::clone(&samples);
45
46 let sampler = tokio::spawn(async move {
47 loop {
48 let queued = Instant::now();
49 // The closure records how long it waited to BEGIN, not how long
50 // it ran. It does nothing else, so the two are not confusable.
51 let waited = tokio::task::spawn_blocking(move || queued.elapsed()).await;
52 if let Ok(waited) = waited {
53 into.lock().unwrap().push(waited);
54 }
55 tokio::time::sleep(interval).await;
56 }
57 });
58
59 BlockingProbe { samples, sampler }
60 }
61
62 /// Stop sampling and summarise.
63 pub(super) fn finish(self) -> BlockingReport {
64 self.sampler.abort();
65 let mut samples = std::mem::take(&mut *self.samples.lock().unwrap());
66 samples.sort();
67 BlockingReport::of(&samples)
68 }
69 }
70
71 /// What the probe saw over one run.
72 pub(super) struct BlockingReport {
73 pub count: usize,
74 pub p50: Duration,
75 pub p95: Duration,
76 pub p99: Duration,
77 pub max: Duration,
78 /// Samples that waited longer than [`STALL`]. The number that matters: on a
79 /// pool with headroom it is zero, and it stops being zero before mean
80 /// latency moves at all.
81 pub stalls: usize,
82 }
83
84 /// The line above which a dispatch delay is a queue rather than scheduling
85 /// noise. Dispatch onto an idle pool is single-digit microseconds; a millisecond
86 /// means the sample waited for a thread.
87 pub(super) const STALL: Duration = Duration::from_millis(1);
88
89 impl BlockingReport {
90 /// Print the probe's summary under the endpoint table.
91 pub(super) fn print(&self) {
92 println!(" Blocking-pool dispatch delay ({} samples):", self.count);
93 if self.count == 0 {
94 println!(" no samples");
95 println!();
96 return;
97 }
98 println!(
99 " p50 {} p95 {} p99 {} max {}",
100 format_dur(self.p50),
101 format_dur(self.p95),
102 format_dur(self.p99),
103 format_dur(self.max),
104 );
105 println!(
106 " stalls over {}: {} ({:.1}%)",
107 format_dur(STALL),
108 self.stalls,
109 self.stalls as f64 / self.count as f64 * 100.0,
110 );
111 if self.stalls == 0 {
112 println!(" the pool had headroom throughout");
113 }
114 println!();
115 }
116
117 fn of(sorted: &[Duration]) -> Self {
118 if sorted.is_empty() {
119 return BlockingReport {
120 count: 0,
121 p50: Duration::ZERO,
122 p95: Duration::ZERO,
123 p99: Duration::ZERO,
124 max: Duration::ZERO,
125 stalls: 0,
126 };
127 }
128 let at = |pct: f64| {
129 let idx = ((sorted.len() as f64 * pct) as usize).min(sorted.len() - 1);
130 sorted[idx]
131 };
132 BlockingReport {
133 count: sorted.len(),
134 p50: at(0.50),
135 p95: at(0.95),
136 p99: at(0.99),
137 max: sorted[sorted.len() - 1],
138 stalls: sorted.iter().filter(|d| **d > STALL).count(),
139 }
140 }
141 }
142
143 /// Same rendering the endpoint table uses, so the two read against each other.
144 fn format_dur(d: Duration) -> String {
145 let us = d.as_micros();
146 if us >= 1000 {
147 format!("{:.1}ms", us as f64 / 1000.0)
148 } else {
149 format!("{us}us")
150 }
151 }
152