diff --git a/quinn-proto/src/connection/pacing.rs b/quinn-proto/src/connection/pacing.rs index 97e4cc753..6802530cf 100644 --- a/quinn-proto/src/connection/pacing.rs +++ b/quinn-proto/src/connection/pacing.rs @@ -4,20 +4,28 @@ use std::time::{Duration, Instant}; use tracing::warn; -/// A simple token-bucket pacer. The bucket starts full and has an adjustable capacity. Once the -/// bucket is empty, further transmission is blocked. The bucket refills at a rate slightly faster -/// than one congestion window per RTT. +/// A simple token-bucket pacer +/// +/// The pacer's capacity is derived on a fraction of the congestion window +/// which can be sent in regular intervals +/// Once the bucket is empty, further transmission is blocked. +/// The bucket refills at a rate slightly faster +/// than one congestion window per RTT, as recommended in +/// https://tools.ietf.org/html/draft-ietf-quic-recovery-34#section-7.7 pub struct Pacer { capacity: u64, + last_window: u64, tokens: u64, prev: Instant, } impl Pacer { /// Obtains a new [`Pacer`]. - pub fn new(capacity: u64, now: Instant) -> Self { + pub fn new(smoothed_rtt: Duration, window: u64, mtu: u16, now: Instant) -> Self { + let capacity = optimal_capacity(smoothed_rtt, window, mtu); Self { capacity, + last_window: window, tokens: capacity, prev: now, } @@ -47,6 +55,14 @@ impl Pacer { "zero-sized congestion control window is nonsense" ); + if window != self.last_window { + self.capacity = optimal_capacity(smoothed_rtt, window, mtu); + + // Clamp the tokens + self.tokens = self.capacity.min(self.tokens); + self.last_window = window; + } + // if we can already send a packet, there is no need for delay if self.tokens >= mtu.into() { return None; @@ -93,6 +109,45 @@ impl Pacer { } } +/// Calculates a pacer capacity for a certain window and RTT +/// +/// 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); + + let capacity = ((window as u128 * BURST_INTERVAL_NANOS) / rtt) as u64; + + // Small bursts are less efficient (no GSO), could increase latency and don't effectively + // use the channel's buffer capacity. Large bursts might block the connection on sending. + capacity + .max(MIN_BURST_SIZE * mtu as u64) + .min(MAX_BURST_SIZE * mtu as u64) +} + +/// The burst interval +/// +/// The capacity will we refilled in 4/5 of that time. +/// 2ms is chosen here since framework timers might have 1ms precision. +/// If kernel-level pacing is supported later a higher time here might be +/// more applicable. +const BURST_INTERVAL_NANOS: u128 = 2_000_000; // 2ms + +/// Allows some usage of GSO, and doesn't slow down the handshake. +const MIN_BURST_SIZE: u64 = 10; + +/// Creating 256 packets took 1ms in a benchmark, so larger bursts don't make sense. +const MAX_BURST_SIZE: u64 = 256; + #[cfg(test)] mod tests { use super::*; @@ -101,32 +156,124 @@ mod tests { fn does_not_panic_on_bad_instant() { let old_instant = Instant::now(); let new_instant = old_instant + Duration::from_micros(15); - assert!(Pacer::new(1500, new_instant) + let rtt = Duration::from_micros(400); + + assert!(Pacer::new(rtt, 30000, 1500, new_instant) .delay(Duration::from_micros(0), 0, 1, old_instant) .is_none()); - assert!(Pacer::new(1500, new_instant) + assert!(Pacer::new(rtt, 30000, 1500, new_instant) .delay(Duration::from_micros(0), 1600, 1, old_instant) .is_none()); - assert!(Pacer::new(1500, new_instant) + assert!(Pacer::new(rtt, 30000, 1500, new_instant) .delay(Duration::from_micros(0), 1500, 3000, old_instant) .is_none()); } + #[test] + fn derives_initial_capacity() { + let window = 2_000_000; + let mtu = 1500; + let rtt = Duration::from_millis(50); + let now = Instant::now(); + + let pacer = Pacer::new(rtt, window, mtu, now); + assert_eq!( + pacer.capacity, + (window as u128 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + ); + assert_eq!(pacer.tokens, pacer.capacity); + + let pacer = Pacer::new(Duration::from_millis(0), window, mtu, now); + assert_eq!(pacer.capacity, MAX_BURST_SIZE * mtu as u64); + assert_eq!(pacer.tokens, pacer.capacity); + + let pacer = Pacer::new(rtt, 1, mtu, now); + assert_eq!(pacer.capacity, MIN_BURST_SIZE * mtu as u64); + assert_eq!(pacer.tokens, pacer.capacity); + } + + #[test] + fn adjusts_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, now); + assert_eq!( + pacer.capacity, + (window as u128 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + ); + assert_eq!(pacer.tokens, pacer.capacity); + let initial_tokens = pacer.tokens; + + pacer.delay(rtt, mtu, window * 2, now); + assert_eq!( + pacer.capacity, + (2 * window as u128 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + ); + assert_eq!(pacer.tokens, initial_tokens); + + pacer.delay(rtt, mtu, window / 2, now); + assert_eq!( + pacer.capacity, + (window as u128 / 2 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + ); + assert_eq!(pacer.tokens, initial_tokens / 2); + } + #[test] fn computes_pause_correctly() { + let window = 2_000_000u64; + let mtu = 1000; + let rtt = Duration::from_millis(50); let old_instant = Instant::now(); - let mut pacer = Pacer::new(1500, old_instant + Duration::from_micros(15)); + + let mut pacer = Pacer::new(rtt, window, mtu, old_instant); + let packet_capacity = pacer.capacity / mtu as u64; + + for _ in 0..packet_capacity { + assert_eq!( + pacer.delay(rtt, mtu, window, old_instant), + None, + "When capacity is available packets should be sent immediately" + ); + + pacer.on_transmit(mtu); + } + + let pace_duration = Duration::from_nanos((BURST_INTERVAL_NANOS * 4 / 5) as u64); + assert_eq!( - pacer.delay(Duration::from_micros(0), 1600, 1, old_instant), - None, - "Zero RTT means that we should send immediately" + pacer + .delay(rtt, mtu, window, old_instant) + .expect("Send must be delayed") + .duration_since(old_instant), + pace_duration ); - let computed_delay = pacer.delay(Duration::from_micros(5), 1600, 1, old_instant); + + // Refill half of the tokens assert_eq!( - computed_delay, - Some(pacer.prev + Duration::from_micros(400)), - "Difference between expected and computed delays is {}ns", - (computed_delay.unwrap() - pacer.prev).as_nanos() + pacer.delay(rtt, mtu, window, old_instant + pace_duration / 2), + None ); + assert_eq!(pacer.tokens, pacer.capacity / 2); + + for _ in 0..packet_capacity / 2 { + assert_eq!( + pacer.delay(rtt, mtu, window, old_instant), + None, + "When capacity is available packets should be sent immediately" + ); + + pacer.on_transmit(mtu); + } + + // Refill all capacity by waiting more than the expected duration + assert_eq!( + pacer.delay(rtt, mtu, window, old_instant + pace_duration * 3 / 2), + None + ); + assert_eq!(pacer.tokens, pacer.capacity); } } diff --git a/quinn-proto/src/connection/paths.rs b/quinn-proto/src/connection/paths.rs index 6bd2bb107..212cf6083 100644 --- a/quinn-proto/src/connection/paths.rs +++ b/quinn-proto/src/connection/paths.rs @@ -39,7 +39,7 @@ impl PathData { remote, rtt: RttEstimator::new(initial_rtt), sending_ecn: true, - pacing: Pacer::new(congestion.initial_window(), now), + pacing: Pacer::new(initial_rtt, congestion.initial_window(), MIN_MTU, now), congestion, challenge: None, challenge_pending: false, @@ -52,10 +52,11 @@ impl PathData { pub fn from_previous(remote: SocketAddr, prev: &PathData, now: Instant) -> Self { let congestion = prev.congestion.clone_box(); + let smoothed_rtt = prev.rtt.get(); PathData { remote, rtt: prev.rtt, - pacing: Pacer::new(congestion.initial_window(), now), + pacing: Pacer::new(smoothed_rtt, congestion.window(), prev.mtu, now), sending_ecn: true, congestion, challenge: None,