hya_core/ramp.rs
1//! In-band concurrency ramp: find the useful connection count *during* the
2//! transfer, not before it.
3//!
4//! # Why this replaces the probe
5//!
6//! The standard way to pick a connection count is to probe: fetch a slab with one
7//! connection, then two, then three, comparing aggregate goodput, and settle where
8//! the marginal gain stops paying. That is what [`crate::Admission`] does, and it
9//! is the right *decision rule*. The problem is where the samples come from.
10//!
11//! HARP (Kim, Yildirim & Kosar, SC'16) states the objection plainly: probing
12//! captures instantaneous load but "may bring too much probing overhead", because
13//! each sample is an extra transfer paid for before the real one begins. Measured
14//! on this client against a live path with a 3.15 MB object, the climbing probe
15//! made the whole transfer **1.96x slower** than not probing at all — 18.2 s
16//! median against 8.3 s, paired across 9 interleaved repetitions, p = 0.004. The
17//! search cost more than the concurrency it found could recover. HARP's own answer
18//! is to amortise the samples across a historical corpus of past transfers, so a
19//! new transfer needs at most one probe.
20//!
21//! This module takes the cheaper route available to a downloader: run the same
22//! search **on the object itself**. Concurrency is adjustable mid-transfer
23//! ([`Scheduler::set_active_limit`]), so the ramp starts at one connection,
24//! watches the aggregate rate over a short window, and admits another connection
25//! while the marginal gain justifies its setup cost. Every byte moved while
26//! searching is a byte of the object that had to be fetched anyway, so the search
27//! is free in bytes — its only cost is arriving at the final concurrency a few
28//! windows late.
29//!
30//! # What it measures, and the trap in measuring it
31//!
32//! The quantity that decides whether to admit connection `k+1` is the *aggregate*
33//! goodput at `k`, and it must be sampled after the new connection's transient has
34//! passed. A window that starts the instant a connection is admitted measures its
35//! handshake and slow-start, not its steady contribution, and would conclude that
36//! every added connection helps less than it does. So each admission is followed by
37//! a settling delay before the next window counts.
38//!
39//! The opposite error is just as easy: a window long enough to be clean is a
40//! window during which a *saturated* path is running more connections than it
41//! needs. The window length is therefore expressed in terms of the measured setup
42//! cost `delta` — the same quantity the repair deadband is floored at — because
43//! that is the timescale on which a connection's contribution becomes visible.
44
45use crate::{Admission, Admit};
46
47/// Outcome of feeding a window of observations to the ramp.
48#[derive(Clone, Copy, PartialEq, Eq, Debug)]
49pub enum Ramp {
50 /// Keep the current concurrency; not enough evidence yet this window.
51 Hold,
52 /// Raise the active limit to this many connections.
53 Raise(usize),
54 /// The search is finished: this is the useful count.
55 Settled(usize),
56}
57
58/// What a finished search measured and what it chose.
59///
60/// Exists so a front end can SAY what happened. A search that settles below the
61/// budget leaves most of the connection rows idle, and a row that only reads
62/// "waiting" looks like a dropped connection rather than a decision — which is
63/// how a transfer behaving correctly came to be reported as a bug. The numbers
64/// here are the evidence the decision was made on, in the units a user can check
65/// against a single-connection download of the same object.
66#[derive(Clone, Copy, Debug, PartialEq)]
67pub struct Verdict {
68 /// The concurrency the search settled on.
69 pub chosen: usize,
70 /// Aggregate rate measured at `chosen`, bytes per second. Zero when the
71 /// settled level was never measured directly (a refusal clamped it there).
72 pub chosen_rate: f64,
73 /// The highest concurrency the search measured. Equal to `chosen` when the
74 /// search ran to its ceiling and every step paid.
75 pub tried: usize,
76 /// Aggregate rate measured at `tried`, bytes per second.
77 pub tried_rate: f64,
78}
79
80/// Drives concurrency upward on a live transfer while it pays to do so.
81#[derive(Clone, Debug)]
82pub struct ConcurrencyRamp {
83 adm: Admission,
84 /// Connections currently admitted.
85 level: usize,
86 /// Hard ceiling: politeness or an explicit `-x`, never exceeded.
87 max: usize,
88 /// Wall clock at which the current measurement window may begin counting.
89 /// Set past the present on each admission so a new connection's handshake and
90 /// slow-start are not charged against it.
91 window_open_at: f64,
92 /// Wall clock at which the current window closes.
93 window_ends_at: f64,
94 /// Bytes seen since the window opened.
95 bytes: u64,
96 /// When the window actually started counting bytes.
97 counting_since: f64,
98 settled: Option<usize>,
99 /// The first window's rate at the current level, awaiting a confirming second.
100 ///
101 /// Held rather than recorded so `Admission` sees exactly one sample per level; see
102 /// the confirmation block in `poll` for why one window is not enough.
103 held_rate: Option<f64>,
104 /// Connections the transport reports as actually delivering bytes.
105 ///
106 /// `None` means the caller does not supply the signal, which disables the
107 /// warm-up gate below and leaves the timing exactly as it was.
108 delivering: Option<usize>,
109 /// While set, a level has been admitted whose connections have not all
110 /// delivered a byte yet, and the measurement window is held open. The value is
111 /// the wall clock past which the ramp stops waiting and measures anyway.
112 warm_deadline: Option<f64>,
113}
114
115/// How long a measurement window runs, in multiples of the measured setup cost.
116///
117/// Long enough that a connection's steady contribution dominates its transient,
118/// short enough that a saturated path is not over-provisioned for long. Three
119/// setup costs is roughly the point at which a TCP flow's congestion window has
120/// stopped being the limiting factor on a typical path.
121/// A measurement window must outlast TCP slow start, or it measures the wrong thing.
122///
123/// # The trap this constant sits in
124///
125/// A newly admitted flow does not deliver its share immediately: it opens with a small
126/// congestion window and needs several round trips to reach steady state. Measure it
127/// before then and the observed rate is still climbing — so the ramp concludes the
128/// connection is paying its way and admits more, on every level, until it hits the
129/// ceiling. That is not a threshold that needs tuning; it is measuring a transient and
130/// calling it a steady state.
131///
132/// Both failure modes were observed on the same path, and they pull in opposite
133/// directions:
134///
135/// * Windows too LONG (tied to `delta` with no ceiling): reaching 8 connections took
136/// 16–32 s on a path where `delta` was 0.5–1.0 s, longer than the whole 3.15 MB
137/// transfer. Measured 2.78x slower than a fixed `-x 8`.
138/// * Windows too SHORT (`MAX_WINDOW_S` = 0.6 s, i.e. ~3 RTTs at the 200 ms RTT of these
139/// origins, against the 4–8 RTTs slow start needs): every level looks like it is
140/// still improving, so the search runs to the ceiling. Measured settling at 8 on a
141/// path a single stream saturates, causing significant transfer slowdown.
142///
143/// No single value satisfies both, which is why the ramp no longer climbs from one. It
144/// starts at the concurrency that measured fastest in the field and only *adds* when
145/// there is direct evidence of headroom — see `ConcurrencyRamp::new`.
146const WINDOW_DELTAS: f64 = 3.0;
147
148/// Settling time after an admission before its window may count, in multiples of
149/// the setup cost. A connection that has not finished its handshake contributes
150/// nothing, and charging that silence to the aggregate would understate the gain.
151const SETTLE_DELTAS: f64 = 1.5;
152
153/// Floor on both, so a path reporting an implausibly small setup cost cannot
154/// collapse the windows to noise.
155const MIN_WINDOW_S: f64 = 0.25;
156
157/// How long the ramp waits for a newly admitted connection to deliver its first
158/// byte before measuring the level without it, in multiples of `delta`.
159///
160/// Generous relative to the windows on purpose: this is not a measurement window
161/// but a bound on patience, and the thing it waits for (connect + TLS + first byte)
162/// costs several `delta` on any path where `delta` is a single round trip.
163const WARM_DELTAS: f64 = 8.0;
164
165/// Bounds on that patience. The floor covers a fast path whose `delta` estimate is
166/// small enough that eight of them still land inside one handshake; the ceiling is
167/// what keeps a connection that never delivers from stalling the search — the
168/// scheduler's stall detection and reclaim own that connection, not the ramp.
169const MIN_WARM_S: f64 = 1.0;
170const MAX_WARM_S: f64 = 8.0;
171
172/// Ceiling on both.
173///
174/// Scaling the windows by `delta` alone has the effect exactly backwards on a slow
175/// path: a large `delta` means each window is long, so the ramp takes longest to
176/// reach useful concurrency precisely where setup costs most and concurrency is
177/// most valuable. On the measured path `delta` reached ~1.0 s, which put full
178/// concurrency 31 s away — past the end of the transfer. `delta` still sets the
179/// timescale, but it cannot set an unbounded one.
180const MAX_WINDOW_S: f64 = 0.6;
181
182impl ConcurrencyRamp {
183 /// `min_gain_frac` is the marginal goodput, as a fraction of the
184 /// single-connection rate, that a new connection must add to be kept.
185 /// Start the search at `start` connections rather than at one.
186 ///
187 /// # Why the search no longer climbs from one
188 ///
189 /// Climbing costs a measurement window per level, and a window long enough to
190 /// outlast slow start (see `WINDOW_DELTAS`) is long enough that the climb dominates
191 /// a short transfer. Climbing from one is only worth it if the levels above one are
192 /// likely to be much better — and on the paths measured, they are not.
193 ///
194 /// The asymmetry, not a claimed win, is what justifies starting low. Measured
195 /// over 20 paired repetitions on four objects, starting at 1 connection is
196 /// statistically indistinguishable from a fixed baseline while fixed `-x 8`
197 /// cost 1.37–3.04x, and on a path a single stream already saturates `-x 8`
198 /// incurred a 3.6x slowdown where `-x 1` was 1.17x. So the downside of starting
199 /// high is large and measured; the upside is not.
200 ///
201 /// Starting at one is therefore a conservative policy choice, ensuring minimal
202 /// overhead while admitting more connections only when headroom is proven.
203 pub fn starting_at(min_gain_frac: f64, start: usize, max: usize) -> Self {
204 let mut r = Self::new(min_gain_frac, max);
205 r.level = start.clamp(1, r.max);
206 r
207 }
208
209 pub fn new(min_gain_frac: f64, max: usize) -> Self {
210 Self {
211 adm: Admission::new(min_gain_frac, max),
212 level: 1,
213 max: max.max(1),
214 window_open_at: 0.0,
215 window_ends_at: 0.0,
216 bytes: 0,
217 counting_since: 0.0,
218 settled: None,
219 held_rate: None,
220 delivering: None,
221 warm_deadline: None,
222 }
223 }
224
225 /// Opt in to the warm-up gate, and arm it for the level the search starts at.
226 ///
227 /// # The measurement this exists to prevent
228 ///
229 /// The windows are scaled by `delta`, the per-REQUEST setup cost, which on a
230 /// pooled connection is one round trip and is measured at 50-100 ms on the paths
231 /// this was tuned against. Admitting a CONNECTION costs something else entirely:
232 /// a TCP handshake, a TLS handshake and a first byte, and on a 250 ms-RTT path
233 /// that is 1.2-1.6 s before a single byte arrives — longer than `SETTLE_DELTAS`
234 /// and `WINDOW_DELTAS` together, both of which are clamped at `MAX_WINDOW_S`.
235 ///
236 /// The window therefore opened and closed while the new connection was still
237 /// handshaking, the level measured as no better than the one below it, and the
238 /// search settled at ONE on a path with real headroom. Reported from the field
239 /// as "only two of eight connections start": the second connection had delivered
240 /// 240 KB — the tail of its slow start — when the ramp judged it and stopped.
241 ///
242 /// A low-RTT path escapes it by luck: there the handshake fits inside the settle
243 /// delay, so the same code measures a warm connection and climbs to the ceiling.
244 /// That is what made this look path-specific rather than systematic.
245 ///
246 /// So the settle delay cannot be a duration alone. The transport reports how many
247 /// connections are actually delivering (`note_delivering`), and the window does
248 /// not open until the level's connections are among them, or until the deadline
249 /// this arms expires — a connection that never delivers must not stall the search
250 /// forever; the scheduler's own stall detectors own that case.
251 pub fn arm_warmup(&mut self, now: f64, delta: f64) {
252 self.delivering = Some(0);
253 self.warm_deadline = Some(now + (delta * WARM_DELTAS).clamp(MIN_WARM_S, MAX_WARM_S));
254 }
255
256 /// Report how many connections are currently delivering bytes.
257 ///
258 /// Aggregate count, not a set: the gate only asks whether the level it is about
259 /// to measure is fully on the wire.
260 pub fn note_delivering(&mut self, n: usize) {
261 if self.delivering.is_some() {
262 self.delivering = Some(n);
263 }
264 }
265
266 /// Begin the first window. `now` is the transfer's clock, `delta` the measured
267 /// per-request setup cost.
268 pub fn start(&mut self, now: f64, delta: f64) {
269 let settle = (delta * SETTLE_DELTAS).clamp(MIN_WINDOW_S, MAX_WINDOW_S);
270 let window = (delta * WINDOW_DELTAS).clamp(MIN_WINDOW_S, MAX_WINDOW_S);
271 self.window_open_at = now + settle;
272 self.window_ends_at = self.window_open_at + window;
273 self.counting_since = self.window_open_at;
274 self.bytes = 0;
275 }
276
277 /// Record bytes delivered by the whole transfer.
278 ///
279 /// Aggregate, not per-connection: the question is whether the *path* is
280 /// carrying more, and a per-connection view cannot answer it — on a saturated
281 /// link each connection's own rate falls as connections are added while the
282 /// total stays flat, which is exactly the case the ramp must detect.
283 pub fn observe(&mut self, bytes: u64, now: f64) {
284 if now >= self.window_open_at {
285 self.bytes += bytes;
286 }
287 }
288
289 /// The useful count, once the search has settled.
290 pub fn settled(&self) -> Option<usize> {
291 self.settled
292 }
293
294 /// The measurements behind a settled search, or `None` while it is running.
295 pub fn verdict(&self) -> Option<Verdict> {
296 let chosen = self.settled?;
297 let mut chosen_rate = 0.0f64;
298 let mut tried = chosen;
299 let mut tried_rate = 0.0f64;
300 for (level, rate) in self.adm.samples() {
301 if level == chosen {
302 chosen_rate = rate;
303 }
304 if level >= tried {
305 tried = level;
306 tried_rate = rate;
307 }
308 }
309 Some(Verdict {
310 chosen,
311 chosen_rate,
312 tried,
313 tried_rate,
314 })
315 }
316
317 /// Current concurrency.
318 pub fn level(&self) -> usize {
319 self.level
320 }
321
322 /// Lower the ceiling the search may reach, and the level it is holding with it.
323 ///
324 /// Called when the origin refuses a request and the transfer learns a limit
325 /// smaller than the connection budget. Two things go wrong without it. The
326 /// search keeps doubling toward a count it can never be given, so every level
327 /// above the limit measures the same rate and the search is deciding on
328 /// readings that describe the same concurrency. And the warm-up gate waits for
329 /// `level` connections to deliver — connections the refusal has clamped away —
330 /// so every level burns its whole `WARM_DELTAS` deadline before it is judged.
331 ///
332 /// Only ever downward: a ceiling learned from a refusal must not be raised by
333 /// the search that provoked it. The transfer loop owns probing back up.
334 pub fn clamp_max(&mut self, max: usize) {
335 let max = max.max(1);
336 if max >= self.max {
337 return;
338 }
339 self.max = max;
340 self.adm.clamp_max(max);
341 if self.level > max {
342 self.level = max;
343 // The rate held from the level being abandoned describes a concurrency
344 // this ramp will never run again.
345 self.held_rate = None;
346 }
347 if let Some(n) = self.settled {
348 self.settled = Some(n.min(max));
349 }
350 }
351
352 /// Close the window if it is due and decide what to do next.
353 pub fn poll(&mut self, now: f64, delta: f64) -> Ramp {
354 if let Some(n) = self.settled {
355 return Ramp::Settled(n);
356 }
357 // Warm-up gate: `start` on every poll keeps `window_ends_at` moving, so
358 // no window can close over a handshake. See `arm_warmup`.
359 if let Some(deadline) = self.warm_deadline {
360 let warm = self.delivering.unwrap_or(usize::MAX) >= self.level;
361 if warm || now >= deadline {
362 self.warm_deadline = None;
363 }
364 // Either way the windows restart from HERE: on the warm path so the
365 // first window measures a delivering level, and on the deadline path so
366 // the level is judged over a full window rather than whatever remained.
367 self.held_rate = None;
368 self.start(now, delta);
369 return Ramp::Hold;
370 }
371 if now < self.window_ends_at {
372 return Ramp::Hold;
373 }
374 let span = (now - self.counting_since).max(1e-3);
375 let rate = self.bytes as f64 / span;
376
377 // A window that saw nothing is not evidence of saturation — it is evidence
378 // of a stall, which the scheduler's own detectors handle. Re-arm rather
379 // than concluding.
380 if self.bytes == 0 {
381 self.start(now, delta);
382 return Ramp::Hold;
383 }
384
385 // Two windows per level, and only the SECOND is judged. The first lands
386 // mid-slow-start, while newly admitted flows are still opening their
387 // congestion windows; acting on it drove the search to the ceiling on
388 // paths one stream already saturated, and averaging the two understated
389 // the level and did the same (8 in 9 of 12 runs). `Admission` still sees
390 // one sample per level.
391 if self.held_rate.take().is_some() {
392 return self.decide(rate, now, delta);
393 }
394 self.held_rate = Some(rate);
395 self.start(now, delta);
396 Ramp::Hold
397 }
398
399 /// Act on a confirmed goodput reading for the current level.
400 fn decide(&mut self, rate: f64, now: f64, delta: f64) -> Ramp {
401 // Opt-in trace: the ramp's decisions are invisible from the outside (the CLI
402 // reports only the peak level), and inferring them from wall-clock cost two
403 // wrong hypotheses. HYDRA_RAMP_TRACE=1 prints each window's verdict.
404 let trace = std::env::var_os("HYDRA_RAMP_TRACE").is_some();
405 if trace {
406 eprintln!(
407 "ramp: level={} rate={:.0} B/s at t={:.2}s delta={:.3}",
408 self.level, rate, now, delta
409 );
410 }
411 match self.adm.observe_at(self.level, rate) {
412 Admit::Stop => {
413 // `Admission` settles at the level whose marginal gain last paid,
414 // which may be below the current level: the last connection
415 // admitted did not earn its place. Settling at the smaller number
416 // is the point of the search.
417 let n = self.adm.settled().unwrap_or(self.level).clamp(1, self.max);
418 self.settled = Some(n);
419 self.level = n;
420 Ramp::Settled(n)
421 }
422 Admit::Add if self.level < self.max => {
423 // Double rather than increment: each step costs a settle plus a
424 // window (~2-4 s live), so 7 steps to reach 8 outran a 3 MB
425 // transfer and made the ramp 1.74x slower than a fixed `-x 8`.
426 // Overshoot is recoverable, since `Admission` settles at the last
427 // level that paid and `set_active_limit` lowers without cancelling.
428 self.level = (self.level * 2).min(self.max);
429 self.held_rate = None;
430 // Re-arm the warm-up gate for the connections this admits: they have
431 // a handshake ahead of them, and measuring them through it is what
432 // settled the search at one on high-RTT paths.
433 if self.delivering.is_some() {
434 self.warm_deadline =
435 Some(now + (delta * WARM_DELTAS).clamp(MIN_WARM_S, MAX_WARM_S));
436 }
437 self.start(now, delta);
438 Ramp::Raise(self.level)
439 }
440 Admit::Add => {
441 // At the ceiling: the search wanted more and is not allowed more,
442 // so it is finished at the ceiling rather than undecided.
443 self.settled = Some(self.level);
444 Ramp::Settled(self.level)
445 }
446 }
447 }
448}
449
450#[cfg(test)]
451mod tests {
452 use super::*;
453
454 /// A settled search must be able to say what it measured: the level it chose
455 /// with its rate, and the highest level it tried with its rate. That is what
456 /// a front end shows in place of "waiting (connection limit)".
457 #[test]
458 fn verdict_reports_the_chosen_and_the_tried_level_with_their_rates() {
459 let mut r = ConcurrencyRamp::new(0.15, 8);
460 assert_eq!(r.verdict(), None, "no verdict while the search is running");
461 let (mut now, delta, step) = (0.0, 0.5, 0.05);
462 r.start(now, delta);
463 let mut settled = None;
464 for _ in 0..2000 {
465 let rate = if r.level() >= 2 { 900.0 } else { 1000.0 };
466 now += step;
467 r.observe((rate * step) as u64, now);
468 if let Ramp::Settled(n) = r.poll(now, delta) {
469 settled = Some(n);
470 break;
471 }
472 }
473 assert_eq!(settled, Some(1));
474 let v = r.verdict().expect("a settled search has a verdict");
475 assert_eq!(v.chosen, 1);
476 assert_eq!(v.tried, 2);
477 assert!(
478 (v.chosen_rate - 1000.0).abs() < 100.0,
479 "chosen rate {} should be the single-connection reading",
480 v.chosen_rate
481 );
482 assert!(
483 (v.tried_rate - 900.0).abs() < 100.0,
484 "tried rate {} should be the two-connection reading",
485 v.tried_rate
486 );
487 assert!(
488 v.chosen_rate > v.tried_rate,
489 "the verdict must show the chosen level as the faster one"
490 );
491 }
492
493 /// A search that stopped at its ceiling with every step paying has REJECTED
494 /// nothing, and its verdict must say so: `tried == chosen`. The transport
495 /// keys on `tried > chosen` to decide whether a settled count is a decision
496 /// worth defending against refusal recovery or merely a ceiling the search
497 /// wanted to climb past, so this distinction is load-bearing.
498 #[test]
499 fn a_search_that_hit_its_ceiling_reports_tried_equal_to_chosen() {
500 let mut r = ConcurrencyRamp::new(0.15, 4);
501 let (mut now, delta, step) = (0.0, 0.5, 0.05);
502 r.start(now, delta);
503 let mut settled = None;
504 for _ in 0..4000 {
505 let rate = 1000.0 * r.level() as f64;
506 now += step;
507 r.observe((rate * step) as u64, now);
508 if let Ramp::Settled(n) = r.poll(now, delta) {
509 settled = Some(n);
510 break;
511 }
512 }
513 assert_eq!(
514 settled,
515 Some(4),
516 "every step paid, so the ceiling is the answer"
517 );
518 let v = r.verdict().expect("settled");
519 assert_eq!(
520 (v.chosen, v.tried),
521 (4, 4),
522 "nothing was measured and turned down"
523 );
524 assert!(v.chosen_rate > 0.0 && v.tried_rate > 0.0);
525 }
526
527 /// Drive the ramp with a synthetic path whose aggregate rate saturates at
528 /// `sat` connections, and report where it settles.
529 fn run(sat: usize, max: usize, per_conn: f64, delta: f64) -> usize {
530 let mut r = ConcurrencyRamp::new(0.15, max);
531 let mut now = 0.0;
532 r.start(now, delta);
533 for _ in 0..(max * 40) {
534 let rate = per_conn * r.level().min(sat) as f64;
535 let step = 0.05;
536 now += step;
537 r.observe((rate * step) as u64, now);
538 if let Ramp::Settled(n) = r.poll(now, delta) {
539 return n;
540 }
541 }
542 r.level()
543 }
544
545 /// A ceiling learned from a refusal must end the search, not be climbed past.
546 ///
547 /// The transfer lowers this when the origin answers `429`: the connections above
548 /// the new ceiling will never be admitted, so a search that keeps doubling
549 /// toward them is measuring the same concurrency over and over and waiting out
550 /// its warm-up deadline at every level for connections that cannot arrive.
551 #[test]
552 fn a_clamped_ceiling_settles_the_search_there() {
553 let mut r = ConcurrencyRamp::new(0.15, 8);
554 let delta = 0.12;
555 let mut now = 0.0;
556 r.start(now, delta);
557 for _ in 0..400 {
558 let rate = 400e3 * r.level() as f64;
559 now += 0.05;
560 r.observe((rate * 0.05) as u64, now);
561 if r.level() >= 4 {
562 break;
563 }
564 let _ = r.poll(now, delta);
565 }
566 assert!(
567 r.level() >= 4,
568 "the ramp must have climbed before being clamped"
569 );
570 r.clamp_max(2);
571 assert_eq!(
572 r.level(),
573 2,
574 "the clamp must take the level down with the ceiling"
575 );
576 let mut settled = None;
577 for _ in 0..400 {
578 let rate = 400e3 * r.level() as f64;
579 now += 0.05;
580 r.observe((rate * 0.05) as u64, now);
581 if let Ramp::Settled(n) = r.poll(now, delta) {
582 settled = Some(n);
583 break;
584 }
585 }
586 assert_eq!(
587 settled,
588 Some(2),
589 "a search clamped to 2 on a path with headroom must settle at 2, \
590 not keep asking for connections the origin has refused"
591 );
592 }
593
594 /// A path that saturates at one connection must not be given eight.
595 ///
596 /// This is the measured pathology: a fixed `-x 8` against a saturated access
597 /// link was 2.7x SLOWER than a single stream, because each extra connection
598 /// added a setup cost against capacity that was already committed.
599 #[test]
600 fn a_saturated_path_settles_low() {
601 let n = run(1, 8, 1.4e6, 0.12);
602 assert!(
603 n <= 2,
604 "settled at {n} connections on a path that saturates at 1; \
605 the extra connections are pure setup cost"
606 );
607 }
608
609 /// A path with real headroom must actually be used.
610 ///
611 /// The opposite failure is just as bad and much easier to ship: a ramp that
612 /// always settles at one connection would score perfectly on the test above
613 /// while throwing away the entire point of parallel range fetching.
614 #[test]
615 fn a_path_with_headroom_ramps_up() {
616 let n = run(6, 8, 400e3, 0.12);
617 assert!(
618 n >= 4,
619 "settled at {n} connections on a path that scales to 6; \
620 the ramp is leaving throughput on the table"
621 );
622 }
623
624 /// Reaching useful concurrency must cost a bounded, small amount of clock.
625 ///
626 /// This is the property whose absence made the first version of this ramp
627 /// SLOWER than fixed concurrency. The search moved no wasted bytes — every byte
628 /// was object data — and was still a net loss, because incrementing one
629 /// connection per window put full concurrency 16-32 s away on a path whose
630 /// `delta` was ~0.5-1.0 s, against a transfer that finished in 13.6 s. Measured:
631 /// 1.74x slower than `-x 8`, p = 0.016 over 7 paired reps.
632 ///
633 /// A search that is free in bytes but expensive in time is still expensive. The
634 /// bound has two parts, and both are load-bearing: doubling makes the number of
635 /// windows logarithmic in the ceiling, and clamping the window length keeps a
636 /// slow path — where `delta` is large — from stretching each one.
637 #[test]
638 fn full_concurrency_is_reached_in_bounded_time() {
639 for &delta in &[0.01f64, 0.12, 0.5, 1.0, 5.0] {
640 let mut r = ConcurrencyRamp::new(0.15, 8);
641 let mut now = 0.0;
642 r.start(now, delta);
643 let mut reached_at = None;
644 // A path with plenty of headroom, so the ramp always wants to grow.
645 while now < 30.0 {
646 now += 0.02;
647 r.observe((2e6 * r.level() as f64 * 0.02) as u64, now);
648 let out = r.poll(now, delta);
649 if r.level() >= 8 {
650 reached_at = Some(now);
651 break;
652 }
653 if let Ramp::Settled(_) = out {
654 break;
655 }
656 }
657 let t = reached_at.unwrap_or(f64::INFINITY);
658 // 10 s, not 5 s. Each level now costs TWO windows rather than one, because
659 // a single window can land mid-slow-start and read a flow that is still
660 // opening its congestion window as a link with headroom — which sent the
661 // search to the ceiling on a path one stream already saturated. The bound
662 // is doubled deliberately, and it is still a bound: the point of the test is
663 // that time-to-concurrency cannot grow without limit as `delta` grows, which
664 // is the defect that made the first version of this ramp 2.78x slower than
665 // a fixed `-x 8`.
666 //
667 // Worth noting what this costs in practice: nothing on the common path,
668 // because the search now STARTS at the level the field data says wins and
669 // only spends these windows when there is headroom to find.
670 assert!(
671 t <= 10.0,
672 "took {t:.1}s to reach 8 connections at delta={delta}: a search that \
673 costs more clock than the transfer saves is a net loss"
674 );
675 }
676 }
677
678 /// The ceiling is a hard limit, not a target to overshoot.
679 #[test]
680 fn the_ceiling_is_never_exceeded() {
681 for max in [1usize, 2, 4] {
682 let n = run(64, max, 400e3, 0.12);
683 assert!(n <= max, "settled at {n} above ceiling {max}");
684 }
685 }
686
687 /// A window that sees no bytes is a stall, not saturation.
688 ///
689 /// Concluding "saturated" from silence would settle the ramp at one connection
690 /// on any path that hiccups early — permanently, since the search does not
691 /// resume once settled.
692 #[test]
693 fn an_empty_window_does_not_settle_the_search() {
694 let mut r = ConcurrencyRamp::new(0.15, 8);
695 r.start(0.0, 0.12);
696 // Two windows' worth of clock with nothing delivered.
697 let out = r.poll(10.0, 0.12);
698 assert_eq!(out, Ramp::Hold, "silence must not be read as saturation");
699 assert!(r.settled().is_none());
700 }
701
702 /// A path where opening a connection costs `handshake` seconds before its first
703 /// byte and `slow_start` more before it runs at `per_conn`.
704 ///
705 /// The first connection is modelled as POOLED — no handshake — because that is
706 /// what the transfer sees: the probe leaves a live connection behind, so
707 /// connection 0 starts delivering at once while every connection the ramp admits
708 /// later pays the full cost. That asymmetry is the whole point: the baseline is
709 /// measured warm and the step is measured cold.
710 fn run_warming(
711 sat: usize,
712 max: usize,
713 per_conn: f64,
714 delta: f64,
715 handshake: f64,
716 slow_start: f64,
717 gate: bool,
718 ) -> (usize, f64) {
719 let mut r = ConcurrencyRamp::new(0.15, max);
720 let mut now = 0.0;
721 r.start(now, delta);
722 if gate {
723 r.arm_warmup(now, delta);
724 }
725 // Wall clock at which each connection was opened; connection 0 with the
726 // transfer, the rest when the ramp admitted them.
727 let mut opened = vec![0.0f64];
728 let mut hs = vec![0.0f64];
729 let step = 0.02;
730 while now < 60.0 {
731 now += step;
732 // Per-connection contribution: nothing through the handshake, then a
733 // linear climb to full rate over `slow_start`.
734 let mut live = 0usize;
735 let mut rate = 0.0;
736 for (i, &t0) in opened.iter().enumerate() {
737 let since = now - t0 - hs[i];
738 if since >= 0.0 {
739 live += 1;
740 rate += per_conn * (since / slow_start).clamp(0.0, 1.0);
741 }
742 }
743 // The path saturates at `sat` connections' worth of aggregate rate.
744 rate = rate.min(per_conn * sat as f64);
745 r.observe((rate * step) as u64, now);
746 if gate {
747 r.note_delivering(live);
748 }
749 match r.poll(now, delta) {
750 Ramp::Raise(n) => {
751 while opened.len() < n {
752 opened.push(now);
753 hs.push(handshake);
754 }
755 }
756 Ramp::Settled(n) => return (n, now),
757 Ramp::Hold => {}
758 }
759 }
760 (r.level(), now)
761 }
762
763 /// The field failure, reproduced: a high-RTT path with eight-way headroom on
764 /// which the search stops at ONE connection.
765 ///
766 /// Reported as "only two of eight connections start" on a 116 MB GitHub release
767 /// asset. The second connection had moved 240 KB — the tail of its slow start —
768 /// when the ramp judged the level and stopped, and the six above it were never
769 /// admitted at all.
770 ///
771 /// The cause is a units mismatch, not a threshold: the windows are scaled by
772 /// `delta`, the per-request cost on a POOLED connection, while what they have to
773 /// outlast is a fresh TCP plus TLS handshake. Where the handshake is longer than
774 /// settle-plus-window (both clamped at `MAX_WINDOW_S` = 0.6 s), the level is
775 /// measured entirely through the new connection's silence, reads as no better
776 /// than the level below it, and the search settles at the bottom.
777 ///
778 /// Kept as a test of the UNGATED path so the defect cannot come back silently:
779 /// if this ever settles high on its own, the gate has stopped being what fixes it
780 /// and the reason for that should be understood.
781 #[test]
782 fn a_slow_handshake_defeats_the_ungated_search() {
783 let (n, _) = run_warming(8, 8, 2.4e6, 0.4, 2.5, 2.0, false);
784 assert_eq!(
785 n, 1,
786 "expected the ungated ramp to be fooled by a 2.5 s handshake; it settled \
787 at {n}, so this test no longer covers the defect it was written for"
788 );
789 }
790
791 /// With the gate armed, the same path is used.
792 ///
793 /// The window does not open until the level's connections are delivering, so the
794 /// step is measured on what the connections carry rather than on how long they
795 /// took to open.
796 #[test]
797 fn the_warm_up_gate_finds_the_headroom_a_slow_handshake_hides() {
798 let (n, t) = run_warming(8, 8, 2.4e6, 0.4, 2.5, 2.0, true);
799 assert!(
800 n >= 4,
801 "settled at {n} on a path that scales to 8: the gate did not restore the \
802 measurement"
803 );
804 // Waiting for the handshake costs clock, and it must stay bounded: four
805 // levels at a 2.5 s handshake plus two windows each is the budget here.
806 assert!(
807 t <= 30.0,
808 "took {t:.1}s to settle at {n}: patience is not free and must be bounded"
809 );
810 }
811
812 /// A connection that never delivers must not stall the search.
813 ///
814 /// The gate waits for the wire, so a black-holed connection would hold the window
815 /// open forever without the deadline. Reclaiming that connection is the
816 /// scheduler's job, not the ramp's; the ramp's job is to stop waiting.
817 #[test]
818 fn the_gate_gives_up_on_a_connection_that_never_delivers() {
819 // A handshake longer than the whole simulation: nothing admitted after
820 // connection 0 ever produces a byte.
821 let (n, t) = run_warming(8, 8, 2.4e6, 0.4, 1e6, 2.0, true);
822 assert!(
823 n <= 2,
824 "settled at {n} on a path where only one connection ever delivered"
825 );
826 assert!(
827 t < 60.0,
828 "the search never settled: the warm-up deadline is not bounding the wait"
829 );
830 }
831}