gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-09-11 22:10:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(proto): sync bbrv3 from quinn (#785)

Browse Source 
## Description

syncs BBR3 from quinn.

It's important to know that the latest version requires a breaking
change to the controller-related traits. To avoid this breaking change,
the BBR3 code itself is left untouched and the trait implementation
defers to current api. I found this approach the best middle ground to
keep it easy to sync once we are able to introduce breaking changes
again.

## Breaking Changes

None

## Notes & open questions

none

## Change checklist
<!-- Remove any that are not relevant. -->
- [x] Self-review.
- [x] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [x] Tests if relevant.
- [x] All breaking changes documented.
- [x] This PR was created by a human that thought critically about the
      proposed change and wrote an as clear and concise description as
      they could.
- [x] This PR isn't slop, and is carefully crafted to do have the
      intented effect.
- [ ] `cargo make` passes locally.
This commit is contained in:
Diva Martínez
2026-08-12 09:15:21 -05:00
committed by GitHub
parent 898cea7d03
commit 64a7e4cb9b
8 changed files with 6709 additions and 1044 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.config/nextest.toml
+6 -1
View File
@@ -21,4 +21,9 @@ default-filter = 'test(~proptests::)'
[profile.ci]
slow-timeout = { period = "5s", terminate-after = 3 }
fail-fast = false
default-filter = 'all()'
default-filter = 'all()'
[[profile.ci.overrides]]
filter = 'test(congestion::bbr3)'
platform = 'cfg(target_env = "musl")'
slow-timeout = { period = "5s", terminate-after = 5 }
noq-proto/src/congestion.rs
+6
View File
@@ -23,6 +23,12 @@ pub trait Controller: Send + Sync + std::fmt::Debug {
#[allow(unused_variables)]
fn on_packet_sent(&mut self, now: Instant, bytes: u16, pn: u64) {}
/// The connection had data to send but was blocked by the congestion window
///
/// Reports the spec's `C.is_cwnd_limited` signal to the controller: the sender fully utilized
/// the congestion window at this point in the current round trip.
fn on_cwnd_limited(&mut self) {}
/// Packet deliveries were confirmed
///
/// `app_limited` indicates whether the connection was blocked on outgoing
noq-proto/src/congestion/bbr3/max_filter.rs
+26 -22
View File
@@ -2,28 +2,7 @@ use std::fmt::Debug;
const MAX_FILTER_LEN: usize = 3;
/// Based on Linux kernel code released here:
/// <https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/?id=f672258391b42a5c7cc2732c9c063e56a85c8dbe>
///
/// Kathleen Nichols' algorithm for tracking the maximum
/// value of a data stream over some fixed time interval. (E.g.,
/// the maximum Bandwidth achieved over the past 3 rounds.) It uses constant
/// space and constant time per update yet almost always delivers
/// the same maximum as an implementation that has to keep all the
/// data in the window.
///
/// The algorithm keeps track of the best, 2nd best & 3rd highest max
/// values, maintaining an invariant that the measurement time of
/// the n'th best >= n-1'th best. It also makes sure that the three
/// values are widely separated in the time window since that bounds
/// the worst case error when that data is monotonically increasing
/// over the window.
///
/// Upon getting a new max, we can forget everything earlier because
/// it has no value - the new max is >= everything else in the window
/// by definition, and it samples the most recent one. So we restart fresh on
/// every new max and overwrites 2nd & 3rd choices. The same property
/// holds for 2nd & 3rd best.
/// Tracks the maximum value of a data stream over a fixed time window.
#[derive(Copy, Clone, Debug)]
pub(super) struct MaxFilter {
window: u64,
@@ -43,6 +22,8 @@ impl MaxFilter {
self.samples[0].value.unwrap_or(0)
}
/// Update the tracked maximum with a new `measurement` at `current_round`.
///
/// `current_round` represents a sequence number counting upwards from 0 monotonically
/// `measurement` is what is tracked as the max values over time
pub(super) fn update_max(&mut self, current_round: u64, measurement: u64) {
@@ -123,6 +104,29 @@ struct MaxSample {
value: Option<u64>,
}
// Based on Linux kernel code released here
// <https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/?id=f672258391b42a5c7cc2732c9c063e56a85c8dbe>
//
// Kathleen Nichols' algorithm for tracking the maximum
// value of a data stream over some fixed time interval. (E.g.,
// the maximum Bandwidth achieved over the past 3 rounds.) It uses constant
// space and constant time per update yet almost always delivers
// the same maximum as an implementation that has to keep all the
// data in the window.
//
// The algorithm keeps track of the best, 2nd best & 3rd highest max
// values, maintaining an invariant that the measurement time of
// the n'th best >= n-1'th best. It also makes sure that the three
// values are widely separated in the time window since that bounds
// the worst case error when that data is monotonically increasing
// over the window.
//
// Upon getting a new max, we can forget everything earlier because
// it has no value - the new max is >= everything else in the window
// by definition, and it samples the most recent one. So we restart fresh on
// every new max and overwrites 2nd & 3rd choices. The same property
// holds for 2nd & 3rd best.
#[cfg(test)]
mod test {
use super::*;
noq-proto/src/congestion/bbr3/mod.rs
+5955 -906
View File
File diff suppressed because it is too large Load Diff
noq-proto/src/connection/mod.rs
+60 -30
View File
@@ -1325,12 +1325,32 @@ impl Connection {
// Handshake and Data(PathId::ZERO) spaces.
let mut last_packet_number = None;
// If we end up not sending anything, we need to know if that was because there was
// nothing to send or because we were congestion blocked.
let mut congestion_blocked = false;
// Set when either the congestion window or the pacer held a send back; drives
// `app_limited`, since neither case means the application ran dry.
let mut send_blocked = false;
// Set only when the congestion window itself was full. This is the spec's
// `C.is_cwnd_limited`, which a pacing delay must not stand in for.
let mut cwnd_blocked = false;
let path = self.path_data(path_id);
// `C.send_quantum` bounds one aggregate scheduled and transmitted together as a unit,
// which for a GSO batch is its datagram count. Controllers that don't compute one leave
// the batch bounded only by what the caller offered.
// <https://www.ietf.org/archive/id/draft-ietf-ccwg-bbr-06.html#section-5.6.3>
// Nothing `poll_transmit_on_path` does alters these, so one snapshot serves the whole call.
let controller_metrics = path.congestion.metrics();
let max_datagrams = match controller_metrics.send_quantum {
Some(send_quantum) => {
let datagrams = send_quantum / u64::from(path.current_mtu());
let datagrams = usize::try_from(datagrams).unwrap_or(usize::MAX);
max_datagrams.min(NonZeroUsize::new(datagrams).unwrap_or(NonZeroUsize::MIN))
}
None => max_datagrams,
};
// Set the segment size to this path's MTU for on-path data.
let pmtu = self.path_data(path_id).current_mtu().into();
let pmtu = path.current_mtu().into();
let mut transmit = TransmitBuf::new(buf, max_datagrams, pmtu);
// Iterate over the available spaces.
@@ -1349,12 +1369,20 @@ impl Connection {
connection_close_pending,
pad_datagram,
) {
PollPathSpaceStatus::NothingToSend {
congestion_blocked: cb,
} => {
congestion_blocked |= cb;
PollPathSpaceStatus::NothingToSend { path_blocked } => {
// Continue checking other spaces, tail-loss probes may need to be sent
// in all spaces.
match path_blocked {
PathBlocked::No => {}
PathBlocked::AntiAmplification => {
send_blocked = true;
}
PathBlocked::Congestion => {
cwnd_blocked = true;
send_blocked = true;
}
PathBlocked::Pacing => send_blocked = true,
}
}
PollPathSpaceStatus::WrotePacket {
last_packet_number: pn,
@@ -1379,7 +1407,7 @@ impl Connection {
}
}
if last_packet_number.is_some() || congestion_blocked {
if last_packet_number.is_some() || send_blocked {
self.qlog.emit_recovery_metrics(
path_id,
&mut self
@@ -1391,8 +1419,13 @@ impl Connection {
);
}
self.path_data_mut(path_id).app_limited =
last_packet_number.is_none() && !congestion_blocked;
let path = self.path_data_mut(path_id);
path.app_limited = last_packet_number.is_none() && !send_blocked;
if cwnd_blocked {
path.congestion.on_cwnd_limited();
}
match last_packet_number {
Some(last_packet_number) => {
@@ -1497,7 +1530,7 @@ impl Connection {
trace!(?space_id, %path_id, "nothing to send in space");
}
PollPathSpaceStatus::NothingToSend {
congestion_blocked: false,
path_blocked: PathBlocked::No,
}
}
};
@@ -1507,20 +1540,16 @@ impl Connection {
// if we will need to start a new datagram. If we are coalescing into an already
// started datagram we do not need to check congestion control again.
if transmit.datagram_remaining_mut() == 0 {
let congestion_blocked =
let path_blocked =
self.path_congestion_check(space_id, path_id, transmit, &can_send, now);
if congestion_blocked != PathBlocked::No {
if path_blocked != PathBlocked::No {
// Previous iterations of this loop may have built packets already.
return match last_packet_number {
Some(pn) => PollPathSpaceStatus::WrotePacket {
last_packet_number: pn,
pad_datagram,
},
None => {
return PollPathSpaceStatus::NothingToSend {
congestion_blocked: true,
};
}
None => PollPathSpaceStatus::NothingToSend { path_blocked },
};
}
@@ -1534,11 +1563,7 @@ impl Connection {
last_packet_number: pn,
pad_datagram,
},
None => {
return PollPathSpaceStatus::NothingToSend {
congestion_blocked: false,
};
}
None => PollPathSpaceStatus::NothingToSend { path_blocked },
};
}
@@ -1602,7 +1627,7 @@ impl Connection {
// datagram and try and start another packet here. Then be stopped by the
// same confidentiality limit.
return PollPathSpaceStatus::NothingToSend {
congestion_blocked: false,
path_blocked: PathBlocked::No,
};
};
last_packet_number = Some(builder.packet_number);
@@ -4560,8 +4585,8 @@ impl Connection {
})
.unwrap_or_default();
if self.total_authed_packets > 1
|| packet.payload.len() <= 16 // token + 16 byte tag
|| !is_valid_retry
|| packet.payload.len() <= 16 // token + 16 byte tag
|| !is_valid_retry
{
trace!("discarding invalid Retry");
// - After the client has received and processed an Initial or Retry packet from
@@ -6149,6 +6174,10 @@ impl Connection {
self.ack_frequency
.ack_frequency_sent(path_id, builder.packet_number, max_ack_delay);
path.congestion.on_ack_frequency_update(
config.ack_eliciting_threshold.into_inner(),
max_ack_delay,
);
}
// PATH_CHALLENGE (on-path)
@@ -7232,10 +7261,11 @@ pub trait NetworkChangeHint: fmt::Debug + 'static {
/// Return value for [`Connection::poll_transmit_path_space`].
#[derive(Debug)]
enum PollPathSpaceStatus {
/// Nothing to send in the space, nothing was written into the [`TransmitBuf`].
/// Nothing was written into the [`TransmitBuf`].
NothingToSend {
/// If true there was data to send but congestion control did not allow so.
congestion_blocked: bool,
/// [`PathBlocked`] helps differentiate whether the path had something but was blocked by
/// the congestoin window/pacing vs the path having no data queued for sending.
path_blocked: PathBlocked,
},
/// One or more packets have been written into the [`TransmitBuf`].
WrotePacket {
noq-proto/src/connection/pacing.rs
+452 -81
View File
@@ -1,5 +1,6 @@
//! Pacing of packet transmissions.
use crate::congestion::ControllerMetrics;
use crate::{Duration, Instant};
use tracing::warn;
@@ -15,13 +16,22 @@ use tracing::warn;
#[derive(Debug)]
pub(super) struct Pacer {
capacity: u64,
last_window: u64,
last_mtu: u16,
/// Inputs [`Self::capacity`] was derived from, or `None` if it was derived from a pacing rate.
last_window_inputs: Option<WindowInputs>,
tokens: u64,
max_bytes_per_second: Option<u64>,
prev: Instant,
}
/// Inputs a window-derived [`Pacer::capacity`] was calculated from.
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
struct WindowInputs {
/// Congestion window in bytes, after [`rate_limited_window`] has clamped it.
window: u64,
/// MTU of the path in bytes.
mtu: u16,
}
impl Pacer {
/// Obtains a new [`Pacer`].
pub(super) fn new(
@@ -35,8 +45,7 @@ impl Pacer {
let capacity = optimal_capacity(smoothed_rtt, window, mtu);
Self {
capacity,
last_window: window,
last_mtu: mtu,
last_window_inputs: Some(WindowInputs { window, mtu }),
tokens: capacity,
max_bytes_per_second,
prev: now,
@@ -55,55 +64,48 @@ impl Pacer {
/// Return how long we need to wait before sending `bytes_to_send`.
///
/// If we can send a packet right away, this returns `None`. Otherwise, returns
/// `Some(d)`, where `d` is the duration after which this function should be called
/// again.
/// If we can send a packet right away, this returns `None`. Otherwise, returns `Some(d)`, where
/// `d` is the time before this function should be called again.
///
/// The 5/4 ratio used here comes from the suggestion that N = 1.25 in the draft IETF
/// RFC for QUIC.
/// The 5/4 ratio used here comes from the suggestion that N = 1.25 in the draft IETF RFC for
/// QUIC.
/// `controller_metrics` provides [`ControllerMetrics`] from the congestion controller used to
/// adjust pacing.
///
/// `capacity` (bytes) and `pacing_rate` (bytes/s) are optional overrides supplied by
/// the congestion controller (e.g. BBRv3's `send_quantum` / `pacing_rate`). They take
/// precedence over the window-derived defaults, but are still subject to the static
/// `max_bytes_per_second` cap configured at construction time.
/// Two of its fields are consumed here:
/// - `congestion_window` (bytes) sets the refill rate when the controller does not compute a
/// rate of its own: one window per `smoothed_rtt`, times the 5/4 ratio above.
/// - `pacing_rate` (bytes/sec) sets the upper limit of how fast we're sending data, and takes
/// precedence over `congestion_window` when present. e.g:
/// <https://www.ietf.org/archive/id/draft-ietf-ccwg-bbr-04.html#name-pacing-rate-cpacing_rate>
pub(super) fn delay(
&mut self,
smoothed_rtt: Duration,
bytes_to_send: u64,
mtu: u16,
window: u64,
now: Instant,
capacity: Option<u64>,
pacing_rate: Option<u64>,
controller_metrics: &ControllerMetrics,
) -> Option<Duration> {
let window = controller_metrics.congestion_window;
debug_assert_ne!(
window, 0,
"zero-sized congestion control window is nonsense"
);
// A controller that computes its own sending rate drives the bucket directly; the
// window- and RTT-derived refill below is used only when no rate is reported.
if let Some(pacing_rate) = controller_metrics.pacing_rate {
return self.delay_at_rate(pacing_rate, bytes_to_send, mtu, now);
}
let window = rate_limited_window(smoothed_rtt, window, self.max_bytes_per_second);
if window != self.last_window || mtu != self.last_mtu {
let inputs = WindowInputs { window, mtu };
if self.last_window_inputs != Some(inputs) {
self.capacity = optimal_capacity(smoothed_rtt, window, mtu);
// Clamp the tokens
// here we cap the number of bytes sent at once during a burst
self.tokens = self.capacity.min(self.tokens);
self.last_window = window;
self.last_mtu = mtu;
}
if let Some(capacity) = capacity {
self.capacity = capacity;
self.tokens = self.capacity.min(self.tokens);
}
if let Some(pacing_rate) = pacing_rate
&& bytes_to_send > self.capacity
{
// Pace at the controller-supplied rate; cap the static rate-limit through the
// `rate_limited_window` window above.
let capped_bytes_to_send = bytes_to_send.max(self.capacity);
let delay = Duration::from_secs_f64(capped_bytes_to_send as f64 / pacing_rate as f64);
return Some(delay);
self.last_window_inputs = Some(inputs);
}
// if we can already send a packet, there is no need for delay
@@ -151,41 +153,95 @@ impl Pacer {
// this is the time at which the pacing window becomes empty
Some((unscaled_delay / 5) * 4)
}
/// Return how long we need to wait before sending `bytes_to_send` when the congestion
/// controller dictates an explicit `pacing_rate` in bytes/sec.
///
/// Credit accumulates in `tokens` at `pacing_rate` for the time elapsed since the last
/// refill, bounded by a burst budget derived from that same rate. If the credit on hand
/// is short, the returned delay indicates when the shortfall will have been earned.
fn delay_at_rate(
&mut self,
pacing_rate: u64,
bytes_to_send: u64,
mtu: u16,
now: Instant,
) -> Option<Duration> {
// An explicit rate is clamped directly. The 1.25 correction in
// `rate_limited_window` exists only to cancel out the legacy refill speedup, which
// this path does not apply. A rate of zero would divide by zero below.
let rate = match self.max_bytes_per_second {
Some(max_bytes_per_second) => Ord::min(pacing_rate, max_bytes_per_second),
None => pacing_rate,
}
.max(1);
let capacity = rate_capacity(rate, mtu);
if capacity != self.capacity {
self.capacity = capacity;
// here we cap the number of bytes sent at once during a burst
self.tokens = self.capacity.min(self.tokens);
}
// Invalidate the window path's cache: its inputs no longer describe `capacity`.
self.last_window_inputs = None;
let time_elapsed = now.checked_duration_since(self.prev).unwrap_or_else(|| {
warn!("received a timestamp early than a previous recorded time, ignoring");
Default::default()
});
let new_tokens = (rate as f64 * time_elapsed.as_secs_f64()) as u64;
// Advance `prev` only once whole bytes have been earned, so elapsed time too short to
// pay for a single byte is carried over rather than discarded. Without this, a slow
// rate polled frequently would never accumulate anything.
if new_tokens > 0 {
self.tokens = self.tokens.saturating_add(new_tokens).min(self.capacity);
self.prev = now;
}
// Capped at the burst budget so that a `bytes_to_send` exceeding the whole bucket is
// still released eventually, rather than waiting for a level the bucket never reaches.
let target = Ord::min(bytes_to_send, self.capacity);
if self.tokens >= target {
return None;
}
// Wait for the shortfall only. Deriving the delay from `bytes_to_send` would re-arm
// the same interval on every poll and never retire, stalling the connection.
let deficit = target - self.tokens;
Some(Duration::from_secs_f64(deficit as f64 / rate as f64))
}
}
/// Calculates a pacer capacity for a certain window and RTT
/// Calculates a pacer capacity for a pacing rate
///
/// The goal is to emit a burst (of size `capacity`) in timer intervals
/// which compromise between
/// - ideally distributing datagrams over time
/// - constantly waking up the connection to produce additional datagrams
///
/// Too short burst intervals means we will never meet them since the timer
/// accuracy in user-space is not high enough. If we miss the interval by more
/// than 25%, we will lose that part of the congestion window since no additional
/// tokens for the extra-elapsed time can be stored.
///
/// Too long burst intervals make pacing less effective.
fn optimal_capacity(smoothed_rtt: Duration, window: u64, mtu: u16) -> u64 {
let rtt = smoothed_rtt.as_nanos().max(1);
/// Burst intervals trade distributing datagrams over time against waking the connection up more
/// often than user-space timer accuracy can service; overshooting one by more than 25% loses the
/// tokens for the extra elapsed time.
fn rate_capacity(pacing_rate: u64, mtu: u16) -> u64 {
let mtu = u64::from(mtu);
let bytes_in =
|interval: Duration| ((pacing_rate as u128 * interval.as_nanos()) / 1_000_000_000) as u64;
let target_capacity = ((window as u128 * TARGET_BURST_INTERVAL.as_nanos()) / rtt) as u64;
let target_capacity = bytes_in(TARGET_BURST_INTERVAL);
// Never restrict capacity below one MTU.
let max_capacity = Ord::max(
((window as u128 * MAX_BURST_INTERVAL.as_nanos()) / rtt) as u64,
mtu,
);
let max_capacity = Ord::max(bytes_in(MAX_BURST_INTERVAL), mtu);
// Batch the greater of `TARGET_BURST_INTERVAL` or `MIN_BURST_SIZE` worth of traffic at a
// time. To avoid inducing excessive latency, limit that result to at most `MAX_BURST_INTERVAL`
// worth of traffic.
// time, limited to at most `MAX_BURST_INTERVAL` worth to avoid inducing excessive latency.
Ord::min(
max_capacity,
target_capacity.clamp(MIN_BURST_SIZE * mtu, MAX_BURST_SIZE * mtu),
)
}
/// Calculates a pacer capacity for a certain window and RTT, which imply a rate
fn optimal_capacity(smoothed_rtt: Duration, window: u64, mtu: u16) -> u64 {
let rtt = smoothed_rtt.as_nanos().max(1);
let rate = u64::try_from(window as u128 * 1_000_000_000 / rtt).unwrap_or(u64::MAX);
rate_capacity(rate, mtu)
}
/// Clamps the window to limit the sending rate to `max_bytes_per_second`.
///
/// If `max_bytes_per_second` is `None`, the original window is returned.
@@ -226,6 +282,167 @@ const MAX_BURST_SIZE: u64 = 256;
mod tests {
use super::*;
/// 100 Mbit/s in bytes/sec, the rate used by the controller-paced tests.
const TEST_PACING_RATE: u64 = 12_500_000;
/// Metrics from a controller that does not compute a rate of its own, as Cubic and Reno
/// report them.
fn unpaced_metrics(congestion_window: u64) -> ControllerMetrics {
ControllerMetrics {
congestion_window,
..Default::default()
}
}
/// Metrics as a delay-based controller such as BBR3 reports them: both a pacing rate
/// and a send quantum are always present.
fn paced_metrics(
congestion_window: u64,
pacing_rate: u64,
send_quantum: u64,
) -> ControllerMetrics {
ControllerMetrics {
congestion_window,
pacing_rate: Some(pacing_rate),
send_quantum: Some(send_quantum),
..Default::default()
}
}
/// Polls `pacer` repeatedly at the single instant `now`, transmitting one `mtu`-sized
/// datagram each time it is allowed to, until it asks the caller to wait.
///
/// Returns when the pacer wants to be polled again and the number of bytes
/// emitted before it blocked, or `None` if it never blocked.
fn burst_until_blocked(
pacer: &mut Pacer,
rtt: Duration,
mtu: u16,
now: Instant,
metrics: &ControllerMetrics,
) -> Option<(Duration, u64)> {
let mut sent = 0;
for _ in 0..10_000 {
match pacer.delay(rtt, u64::from(mtu), mtu, now, metrics) {
Some(resume_after) => return Some((resume_after, sent)),
None => {
pacer.on_transmit(mtu);
sent += u64::from(mtu);
}
}
}
None
}
/// Drives an always-backlogged sender through `pacer` for `duration` of simulated time the
/// way `poll_transmit` does: send whenever the pacer allows it, otherwise jump to the instant
/// it asked to be polled again. Returns the bytes emitted.
fn bytes_sent_over(
pacer: &mut Pacer,
rtt: Duration,
mtu: u16,
start: Instant,
duration: Duration,
metrics: &ControllerMetrics,
) -> u64 {
/// Guards against a pacer that never advances time; far above the ~8k polls a correct
/// pacer needs for one second at [`TEST_PACING_RATE`].
const MAX_POLLS: u64 = 1_000_000;
let deadline = start + duration;
let mut at = start;
let mut sent = 0;
let mut polls = 0;
while at < deadline {
polls += 1;
assert!(
polls < MAX_POLLS,
"pacer made no progress: {sent} bytes emitted without reaching the deadline"
);
match pacer.delay(rtt, u64::from(mtu), mtu, at, metrics) {
None => {
pacer.on_transmit(mtu);
sent += u64::from(mtu);
}
Some(resume) => at += resume,
}
}
sent
}
#[test]
fn blocks_greedy_sender_at_controller_pacing_rate() {
let mtu = 1500;
let rtt = Duration::from_millis(50);
let window = 2_000_000;
let now = Instant::now();
// `send_quantum` at BBR3's `2 * SMSS` floor, i.e. what it reports at low rates.
let metrics = paced_metrics(window, TEST_PACING_RATE, 2 * u64::from(mtu));
let mut pacer = Pacer::new(rtt, window, mtu, None, now);
// Honouring a finite rate is only possible by delaying, so a sender polling at a
// single instant must eventually be told to wait.
assert!(
burst_until_blocked(&mut pacer, rtt, mtu, now, &metrics).is_some(),
"pacer never blocked while polled at a single instant, so the controller's \
pacing rate is not being enforced"
);
}
#[test]
fn pacing_delay_unblocks_once_it_expires() {
let mtu = 1500;
let rtt = Duration::from_millis(50);
let window = 2_000_000;
let now = Instant::now();
let metrics = paced_metrics(window, TEST_PACING_RATE, 2 * u64::from(mtu));
let mut pacer = Pacer::new(rtt, window, mtu, None, now);
let (resume_after, _) = burst_until_blocked(&mut pacer, rtt, mtu, now, &metrics)
.expect("pacer must block once the burst budget is spent");
// `poll_transmit` re-runs when the pacing timer fires. If the pacer re-derives the
// same delay from the new `now` it would re-arm forever and the connection stalls.
assert_eq!(
pacer.delay(rtt, u64::from(mtu), mtu, now + resume_after, &metrics),
None,
"the delay the pacer asked for must be long enough to unblock the send"
);
}
#[test]
fn aggregate_throughput_matches_controller_pacing_rate() {
const SECONDS: u64 = 1;
let mtu = 1500;
let rtt = Duration::from_millis(50);
let window = 2_000_000;
let metrics = paced_metrics(window, TEST_PACING_RATE, 2 * u64::from(mtu));
let start = Instant::now();
let mut pacer = Pacer::new(rtt, window, mtu, None, start);
let sent = bytes_sent_over(
&mut pacer,
rtt,
mtu,
start,
Duration::from_secs(SECONDS),
&metrics,
);
let expected = TEST_PACING_RATE * SECONDS;
// Slack covers the bucket the pacer starts full plus the trailing partial burst.
let slack = expected / 20;
assert!(
sent <= expected + slack,
"emitted {sent} bytes in {SECONDS}s, but pacing_rate allows only {expected}"
);
assert!(
sent + slack >= expected,
"emitted {sent} bytes in {SECONDS}s, underrunning pacing_rate {expected}"
);
}
#[test]
fn does_not_panic_on_bad_instant() {
let old_instant = Instant::now();
@@ -238,10 +455,8 @@ mod tests {
Duration::from_micros(0),
0,
1500,
1,
old_instant,
None,
None
&unpaced_metrics(1),
)
.is_none()
);
@@ -251,10 +466,8 @@ mod tests {
Duration::from_micros(0),
1600,
1500,
1,
old_instant,
None,
None
&unpaced_metrics(1),
)
.is_none()
);
@@ -264,10 +477,8 @@ mod tests {
Duration::from_micros(0),
1500,
1500,
3000,
old_instant,
None,
None
&unpaced_metrics(3000),
)
.is_none()
);
@@ -311,27 +522,27 @@ mod tests {
assert_eq!(pacer.tokens, pacer.capacity);
let initial_tokens = pacer.tokens;
pacer.delay(rtt, mtu as u64, mtu, window * 2, now, None, None);
pacer.delay(rtt, mtu as u64, mtu, now, &unpaced_metrics(window * 2));
assert_eq!(
pacer.capacity,
(2 * window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64
);
assert_eq!(pacer.tokens, initial_tokens);
pacer.delay(rtt, mtu as u64, mtu, window / 2, now, None, None);
pacer.delay(rtt, mtu as u64, mtu, now, &unpaced_metrics(window / 2));
assert_eq!(
pacer.capacity,
(window as u128 / 2 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64
);
assert_eq!(pacer.tokens, initial_tokens / 2);
pacer.delay(rtt, mtu as u64, mtu * 2, window, now, None, None);
pacer.delay(rtt, mtu as u64, mtu * 2, now, &unpaced_metrics(window));
assert_eq!(
pacer.capacity,
(window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64
);
pacer.delay(rtt, mtu as u64, 20_000, window, now, None, None);
pacer.delay(rtt, mtu as u64, 20_000, now, &unpaced_metrics(window));
assert_eq!(pacer.capacity, 20_000_u64 * MIN_BURST_SIZE);
}
@@ -347,7 +558,7 @@ mod tests {
for _ in 0..packet_capacity {
assert_eq!(
pacer.delay(rtt, mtu as u64, mtu, window, old_instant, None, None),
pacer.delay(rtt, mtu as u64, mtu, old_instant, &unpaced_metrics(window)),
None,
"When capacity is available packets should be sent immediately"
);
@@ -358,7 +569,7 @@ mod tests {
let pace_duration = Duration::from_nanos((TARGET_BURST_INTERVAL.as_nanos() * 4 / 5) as u64);
let actual_delay = pacer
.delay(rtt, mtu as u64, mtu, window, old_instant, None, None)
.delay(rtt, mtu as u64, mtu, old_instant, &unpaced_metrics(window))
.expect("Send must be delayed");
let diff = actual_delay.abs_diff(pace_duration);
@@ -374,10 +585,8 @@ mod tests {
rtt,
mtu as u64,
mtu,
window,
old_instant + pace_duration / 2,
None,
None,
&unpaced_metrics(window),
),
None
);
@@ -385,7 +594,7 @@ mod tests {
for _ in 0..packet_capacity / 2 {
assert_eq!(
pacer.delay(rtt, mtu as u64, mtu, window, old_instant, None, None),
pacer.delay(rtt, mtu as u64, mtu, old_instant, &unpaced_metrics(window)),
None,
"When capacity is available packets should be sent immediately"
);
@@ -399,10 +608,8 @@ mod tests {
rtt,
mtu as u64,
mtu,
window,
old_instant + pace_duration * 3 / 2,
None,
None,
&unpaced_metrics(window),
),
None
);
@@ -418,14 +625,14 @@ mod tests {
let mut pacer = Pacer::new(rtt, window, mtu, Some(2_000), old_instant);
assert_eq!(
pacer.delay(rtt, 1_000, mtu, window, old_instant, None, None),
pacer.delay(rtt, 1_000, mtu, old_instant, &unpaced_metrics(window)),
None,
"When capacity is available packets should be sent immediately"
);
pacer.on_transmit(mtu);
let actual_delay = pacer
.delay(rtt, 1_000, mtu, window, old_instant, None, None)
.delay(rtt, 1_000, mtu, old_instant, &unpaced_metrics(window))
.expect("Send must be delayed");
let expected_delay = Duration::from_millis(500);
@@ -439,6 +646,170 @@ mod tests {
// Should be able to send after a while
let now = old_instant + expected_delay / 2;
assert_eq!(pacer.delay(rtt, 500, mtu, window, now, None, None), None);
assert_eq!(
pacer.delay(rtt, 500, mtu, now, &unpaced_metrics(window)),
None
);
}
#[test]
fn derives_burst_budget_from_controller_pacing_rate() {
let window = 2_000_000;
let mtu = 1500;
let rtt = Duration::from_millis(50);
let now = Instant::now();
// A window twice the one the pacer was built with must not disturb a budget the
// controller's rate determines.
let metrics = paced_metrics(window * 2, TEST_PACING_RATE, 2 * u64::from(mtu));
let mut pacer = Pacer::new(rtt, window, mtu, None, now);
pacer.delay(rtt, u64::from(mtu), mtu, now, &metrics);
assert_eq!(pacer.capacity, rate_capacity(TEST_PACING_RATE, mtu));
// 2ms of traffic at 100 Mbit/s, i.e. `TARGET_BURST_INTERVAL` worth.
assert_eq!(pacer.capacity, 25_000);
}
#[test]
fn pacing_delay_covers_exactly_the_token_shortfall() {
// 200 MB/s in bytes/s
const RATE: u64 = 200_000_000;
let window = 2_000_000;
let mtu = 1500;
let rtt = Duration::from_millis(50);
let now = Instant::now();
let metrics = paced_metrics(window, RATE, 2 * u64::from(mtu));
let mut pacer = Pacer::new(rtt, window, mtu, None, now);
let (resume_after, _) = burst_until_blocked(&mut pacer, rtt, mtu, now, &metrics)
.expect("pacer must block once the burst budget is spent");
// The wait pays for the credit still missing, not for the whole datagram: charging
// for bytes already covered by tokens on hand would pace below `pacing_rate`.
let deficit = u64::from(mtu) - pacer.tokens;
assert_eq!(
resume_after,
Duration::from_secs_f64(deficit as f64 / RATE as f64)
);
}
#[test]
fn burst_is_bounded_by_the_target_interval() {
let window = 2_000_000;
let mtu = 1500;
let rtt = Duration::from_millis(50);
let now = Instant::now();
let metrics = paced_metrics(window, TEST_PACING_RATE, 2 * u64::from(mtu));
let mut pacer = Pacer::new(rtt, window, mtu, None, now);
let (_, burst) = burst_until_blocked(&mut pacer, rtt, mtu, now, &metrics)
.expect("pacer must block once the burst budget is spent");
// Bursting more than `TARGET_BURST_INTERVAL` worth of traffic is what fills bottleneck
// queues; falling far short of it wakes the connection up more often than the timer can
// service. One datagram of slack either way is inherent in releasing whole datagrams.
let budget = rate_capacity(TEST_PACING_RATE, mtu);
assert!(
burst <= budget + u64::from(mtu),
"burst of {burst} bytes overshoots the {budget} byte budget by over one datagram"
);
assert!(
burst + u64::from(mtu) >= budget,
"burst of {burst} bytes undershoots the {budget} byte budget by over one datagram"
);
}
#[test]
fn shrinks_burst_budget_when_pacing_rate_drops() {
let window = 2_000_000;
let mtu = 1500;
let rtt = Duration::from_millis(50);
let now = Instant::now();
let quantum = 2 * u64::from(mtu);
let fast = paced_metrics(window, TEST_PACING_RATE, quantum);
let slow = paced_metrics(window, TEST_PACING_RATE / 10, quantum);
let mut pacer = Pacer::new(rtt, window, mtu, None, now);
// Earn credit at the high rate, as during a ProbeBW_UP phase...
assert_eq!(pacer.delay(rtt, u64::from(mtu), mtu, now, &fast), None);
assert_eq!(pacer.capacity, rate_capacity(TEST_PACING_RATE, mtu));
// ...then a gain change lowers it. Credit earned at the old rate must not survive as a
// burst the new rate cannot pay for.
pacer.delay(rtt, u64::from(mtu), mtu, now, &slow);
let budget = rate_capacity(TEST_PACING_RATE / 10, mtu);
assert_eq!(pacer.capacity, budget);
assert!(
pacer.tokens <= budget,
"{} tokens outlive the {budget} byte budget of the lowered rate",
pacer.tokens
);
}
#[test]
fn rate_path_does_not_leave_stale_window_capacity() {
let window = 2_000_000;
let mtu = 1500;
let rtt = Duration::from_millis(50);
let now = Instant::now();
let mut pacer = Pacer::new(rtt, window, mtu, None, now);
// A controller free to report a rate on some calls and not on others is within what
// `ControllerMetrics` allows. The rate path leaves its own budget behind...
pacer.delay(
rtt,
u64::from(mtu),
mtu,
now,
&paced_metrics(window, TEST_PACING_RATE, 2 * u64::from(mtu)),
);
assert_eq!(pacer.capacity, rate_capacity(TEST_PACING_RATE, mtu));
// ...so the window path must not mistake it for a budget of its own, even though
// neither the window nor the MTU it keys on has changed.
pacer.delay(rtt, u64::from(mtu), mtu, now, &unpaced_metrics(window));
assert_eq!(
pacer.capacity,
optimal_capacity(rtt, window, mtu),
"the window path kept a burst budget the rate path derived"
);
}
#[test]
fn max_bytes_per_second_overrides_a_higher_controller_rate() {
const SECONDS: u64 = 1;
/// 1 Mbit/s in bytes/sec, two orders of magnitude under [`TEST_PACING_RATE`].
const LIMIT: u64 = 125_000;
let window = 2_000_000;
let mtu = 1500;
let rtt = Duration::from_millis(50);
let start = Instant::now();
let metrics = paced_metrics(window, TEST_PACING_RATE, 2 * u64::from(mtu));
let mut pacer = Pacer::new(rtt, window, mtu, Some(LIMIT), start);
let sent = bytes_sent_over(
&mut pacer,
rtt,
mtu,
start,
Duration::from_secs(SECONDS),
&metrics,
);
// The configured ceiling binds even though the controller asks for far more.
let expected = LIMIT * SECONDS;
let slack = expected / 20;
assert!(
sent <= expected + slack,
"emitted {sent} bytes in {SECONDS}s, over the {expected} byte ceiling"
);
assert!(
sent + slack >= expected,
"emitted {sent} bytes in {SECONDS}s, underrunning the {expected} byte ceiling"
);
}
}
noq-proto/src/connection/paths.rs
+1 -3
View File
@@ -608,10 +608,8 @@ impl PathData {
smoothed_rtt,
bytes_to_send,
self.current_mtu(),
metrics.congestion_window,
now,
metrics.send_quantum,
metrics.pacing_rate,
&metrics,
)
}
noq-proto/src/tests/mod.rs
+203 -1
View File
@@ -1,8 +1,13 @@
use std::{
any::Any,
convert::TryInto,
mem,
net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
sync::Arc,
num::NonZeroUsize,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
};
use assert_matches::assert_matches;
@@ -31,6 +36,7 @@ use crate::{
StreamEvent, Transmit, TransportConfig, TransportErrorCode, VarInt, WriteError,
cid_generator::{ConnectionIdGenerator, RandomConnectionIdGenerator},
coding::{Decodable, Encodable},
congestion::{Controller, ControllerFactory, ControllerMetrics},
crypto::rustls::{QuicServerConfig, configured_provider},
frame::{self, Frame, FrameStruct},
packet::{FixedLengthConnectionIdParser, PartialDecode},
@@ -4426,6 +4432,202 @@ fn handshake_confirmation_no_resumption_shortcut() {
assert_matches!(pair.client_conn_mut(ch).poll(), None);
}
/// A controller whose window is effectively unbounded but which always reports a low pacing
/// rate, so the only thing that can ever block a send is the pacer. Counts how many times the
/// connection reports the spec's `C.is_cwnd_limited` signal.
#[derive(Debug)]
struct PacingOnlyController {
cwnd_limited_reports: Arc<AtomicU64>,
}
impl Controller for PacingOnlyController {
fn on_cwnd_limited(&mut self) {
self.cwnd_limited_reports.fetch_add(1, Ordering::Relaxed);
}
fn on_congestion_event(
&mut self,
_now: Instant,
_sent: Instant,
_is_persistent_congestion: bool,
_is_ecn: bool,
_lost_bytes: u64,
_largest_lost: u64,
) {
}
fn on_mtu_update(&mut self, _new_mtu: u16) {}
fn window(&self) -> u64 {
u64::MAX / 2
}
fn metrics(&self) -> ControllerMetrics {
ControllerMetrics {
congestion_window: self.window(),
ssthresh: None,
// 1 Mbit/s in bytes/sec: low enough that a bulk transfer is pacing-blocked almost
// continuously.
pacing_rate: Some(125_000),
send_quantum: Some(2 * 1200),
}
}
fn clone_box(&self) -> Box<dyn Controller> {
Box::new(Self {
cwnd_limited_reports: self.cwnd_limited_reports.clone(),
})
}
fn initial_window(&self) -> u64 {
u64::MAX / 2
}
fn into_any(self: Box<Self>) -> Box<dyn Any> {
self
}
}
struct PacingOnlyConfig {
cwnd_limited_reports: Arc<AtomicU64>,
}
impl ControllerFactory for PacingOnlyConfig {
fn build(self: Arc<Self>, _now: Instant, _current_mtu: u16) -> Box<dyn Controller> {
Box::new(PacingOnlyController {
cwnd_limited_reports: self.cwnd_limited_reports.clone(),
})
}
}
/// `C.is_cwnd_limited` means the sender filled the congestion window. A flow held back by
/// pacing is not cwnd-limited and a paced controller such as BBR3 is pacing-limited by
/// design, so conflating the two pins the signal true and breaks the decisions built on it.
#[test]
fn cwnd_limited_is_not_reported_when_only_pacing_blocks() {
let _guard = subscribe();
let reports = Arc::new(AtomicU64::new(0));
let mut transport = TransportConfig::default();
transport.congestion_controller_factory(Arc::new(PacingOnlyConfig {
cwnd_limited_reports: reports.clone(),
}));
let mut client_cfg = client_config();
client_cfg.transport = Arc::new(transport);
let mut pair = Pair::default();
let (client_ch, _) = pair.connect_with(client_cfg);
let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap();
pair.client_send(client_ch, s)
.write(&[42; 64 * 1024])
.unwrap();
pair.drive();
assert_eq!(
reports.load(Ordering::Relaxed),
0,
"connection reported cwnd-limited while only pacing was holding sends back"
);
}
/// A controller that never limits the window or the rate, but reports a small send quantum, so
/// the quantum is the only thing that can bound an aggregate handed to the NIC.
#[derive(Debug, Clone)]
struct FixedQuantumController {
send_quantum: u64,
}
impl Controller for FixedQuantumController {
fn on_congestion_event(
&mut self,
_now: Instant,
_sent: Instant,
_is_persistent_congestion: bool,
_is_ecn: bool,
_lost_bytes: u64,
_largest_lost: u64,
) {
}
fn on_mtu_update(&mut self, _new_mtu: u16) {}
fn window(&self) -> u64 {
u64::MAX / 2
}
fn metrics(&self) -> ControllerMetrics {
ControllerMetrics {
congestion_window: self.window(),
ssthresh: None,
// 1 GB/s: high enough that the pacer never delays within a single batch.
pacing_rate: Some(1_000_000_000),
send_quantum: Some(self.send_quantum),
}
}
fn clone_box(&self) -> Box<dyn Controller> {
Box::new(self.clone())
}
fn initial_window(&self) -> u64 {
u64::MAX / 2
}
fn into_any(self: Box<Self>) -> Box<dyn Any> {
self
}
}
impl ControllerFactory for FixedQuantumController {
fn build(self: Arc<Self>, _now: Instant, _current_mtu: u16) -> Box<dyn Controller> {
Box::new((*self).clone())
}
}
/// `C.send_quantum` bounds the size of one aggregate scheduled and transmitted together, which
/// for a GSO batch means the number of datagrams written in a single call.
#[test]
fn send_quantum_bounds_the_gso_batch() {
/// Datagrams the reported quantum should permit per batch.
const QUANTUM_DATAGRAMS: usize = 3;
/// Datagrams the caller is willing to accept, well above the quantum.
const MAX_DATAGRAMS: NonZeroUsize = NonZeroUsize::new(10).expect("non zero");
let _guard = subscribe();
let mut transport = TransportConfig::default();
transport.congestion_controller_factory(Arc::new(FixedQuantumController {
send_quantum: QUANTUM_DATAGRAMS as u64 * DEFAULT_MTU as u64,
}));
let mut client_cfg = client_config();
client_cfg.transport = Arc::new(transport);
let mut pair = Pair::default();
let (client_ch, _) = pair.connect_with(client_cfg);
let s = pair.client_streams(client_ch).open(Dir::Uni).unwrap();
pair.client_send(client_ch, s)
.write(&[42; 64 * 1024])
.unwrap();
let now = pair.time;
let mut buf = Vec::new();
let transmit = pair
.client_conn_mut(client_ch)
.poll_transmit(now, MAX_DATAGRAMS, &mut buf)
.expect("a stream write must produce a transmit");
let datagrams = match transmit.segment_size {
Some(segment_size) => buf.len().div_ceil(segment_size),
None => 1,
};
assert!(
datagrams <= QUANTUM_DATAGRAMS,
"batched {datagrams} datagrams, over the {QUANTUM_DATAGRAMS} the send quantum allows"
);
}
/// This test used to fail due to incorrectly encoding frame::MaybeFrame::None
/// as 8 bytes of zeroes, instead of a single zero byte that's the correct
/// representation of a minimal zero as QUIC varint.