1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
//! Sub-frame progress observation for the paid-pull receive path.
//!
//! `ProgressReader` wraps the QUIC receive stream as an `AsyncRead` and counts bytes as
//! they arrive, including mid-frame. `read_frame` fills a whole message with one
//! `read_exact`, and that call drives many `poll_read`s, each of which advances the
//! counter. The counter is the primitive the throughput floor reads: it is
//! frame-size-independent because it moves on bytes off the wire, not on decoded frames.
use std::{
collections::VecDeque,
pin::Pin,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
task::{Context, Poll},
time::Duration,
};
use tokio::io::{AsyncRead, ReadBuf};
use tokio::time::Instant;
/// An `AsyncRead` that tallies every byte it yields into a shared counter.
///
/// Wrapping the receive stream at the byte layer — beneath `read_frame`'s message
/// decode — is what makes progress observable inside a single large frame. Each
/// `poll_read` adds the bytes it filled, so a frame that arrives in many transport
/// reads advances the counter continuously rather than all at once on decode.
pub(crate) struct ProgressReader<'a, R> {
inner: &'a mut R,
counter: Arc<AtomicU64>,
}
impl<'a, R> ProgressReader<'a, R> {
/// Wrap `inner`, adding each read's byte count to `counter`.
pub(crate) const fn new(inner: &'a mut R, counter: Arc<AtomicU64>) -> Self {
Self { inner, counter }
}
}
impl<R: AsyncRead + Unpin> AsyncRead for ProgressReader<'_, R> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
let before = buf.filled().len();
let res = Pin::new(&mut *self.inner).poll_read(cx, buf);
if let Poll::Ready(Ok(())) = &res {
let delta = buf.filled().len().saturating_sub(before) as u64;
if delta > 0 {
self.counter.fetch_add(delta, Ordering::Relaxed);
}
}
res
}
}
/// The streaming-stage stall policy: a minimum throughput `floor_bps` sustained over a
/// trailing `window`.
#[derive(Clone, Copy, Debug)]
pub(crate) struct FloorConfig {
/// The trailing window over which throughput is measured.
pub(crate) window: Duration,
/// The minimum bytes-per-second the transfer must sustain over the window. `0`
/// disables the throughput test and leaves pure idle detection: at least one byte
/// per window.
pub floor_bps: u64,
}
/// The outcome of one [`ThroughputFloor::evaluate`].
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum FloorVerdict {
/// Throughput is at or above the floor, or the window is not yet judgeable.
Ok,
/// Throughput has stayed below the floor across a full window.
Stalled,
}
/// Watches a byte counter and reports a stall when throughput falls below the floor.
///
/// It reads the same [`ProgressReader`] counter the receive loop is filling, so the
/// signal is frame-size-independent. The window must warm — a full `window` of streaming
/// must elapse — before the floor can fire, which gives the first-byte stage its own time
/// budget: for the first window no byte is required to have arrived. [`Self::pause`] /
/// [`Self::resume`] exclude a self-inflicted credit-window payment pause, so the payer's
/// own unpaid balance is never read as a sender stall.
pub(crate) struct ThroughputFloor {
cfg: FloorConfig,
counter: Arc<AtomicU64>,
samples: VecDeque<(Instant, u64)>,
started: Instant,
paused_since: Option<Instant>,
}
impl ThroughputFloor {
/// Start watching `counter` from `started` (the moment streaming begins).
pub(crate) const fn new(cfg: FloorConfig, counter: Arc<AtomicU64>, started: Instant) -> Self {
Self {
cfg,
counter,
samples: VecDeque::new(),
started,
paused_since: None,
}
}
/// Stop counting the interval that starts now: the receiver owes the covering proof,
/// so a delivery pause here is self-inflicted, not the sender's stall.
pub(crate) const fn pause(&mut self, now: Instant) {
if self.paused_since.is_none() {
self.paused_since = Some(now);
}
}
/// Resume counting. The paused span is folded out of the timeline: retained samples
/// and the warmup origin shift forward by the span, so the pause reads as no elapsed
/// time rather than as a stall.
pub(crate) fn resume(&mut self, now: Instant) {
if let Some(since) = self.paused_since.take() {
let span = now.saturating_duration_since(since);
for sample in &mut self.samples {
if let Some(shifted) = sample.0.checked_add(span) {
sample.0 = shifted;
}
}
if let Some(shifted) = self.started.checked_add(span) {
self.started = shifted;
}
}
}
/// When the newest sample was taken, or `None` before the first. A gap to
/// `now` much longer than the sample period means the reader was not polled.
pub(crate) fn last_sampled_at(&self) -> Option<Instant> {
self.samples.back().map(|&(t, _)| t)
}
/// Bytes counted across the retained window: the newest sample less the
/// baseline.
pub(crate) fn window_bytes(&self) -> u64 {
match (self.samples.front(), self.samples.back()) {
(Some(&(_, first)), Some(&(_, last))) => last.saturating_sub(first),
_ => 0,
}
}
/// Sample the counter and judge the trailing window. Returns [`FloorVerdict::Stalled`]
/// only once the window is warm and the bytes across it fall below `floor_bps · window`
/// (or below one byte when `floor_bps == 0`).
pub(crate) fn evaluate(&mut self, now: Instant) -> FloorVerdict {
if self.paused_since.is_some() {
return FloorVerdict::Ok;
}
let current = self.counter.load(Ordering::Relaxed);
self.samples.push_back((now, current));
let window = self.cfg.window;
// Drop samples older than the window, but keep the newest one at or before the
// window boundary as the baseline so the delta spans the whole window.
if let Some(boundary) = now.checked_sub(window) {
while self.samples.len() >= 2 {
match self.samples.get(1) {
Some(&(t1, _)) if t1 <= boundary => {
self.samples.pop_front();
}
_ => break,
}
}
}
// The window is not judgeable until a full window of streaming has elapsed; this
// grace is the first-byte time budget.
if now.saturating_duration_since(self.started) < window {
return FloorVerdict::Ok;
}
let baseline = self.samples.front().map_or(current, |&(_, v)| v);
let delta = current.saturating_sub(baseline);
if delta < required_bytes(window, self.cfg.floor_bps) {
FloorVerdict::Stalled
} else {
FloorVerdict::Ok
}
}
}
/// Bytes that must cross the window to clear the floor: `floor_bps · window`, never below
/// one byte so `floor_bps == 0` still catches a full wedge.
fn required_bytes(window: Duration, floor_bps: u64) -> u64 {
let req = u128::from(floor_bps).saturating_mul(window.as_millis()) / 1000;
u64::try_from(req).unwrap_or(u64::MAX).max(1)
}
#[cfg(test)]
mod floor_tests;
#[cfg(test)]
mod tests;