Derive pacing capacity from window and time

The pacer's behavior currently makes the library extremely inefficient.
Each `Pacer::delay` call would only allow a single
datagram to be sent, and then instruct the connection to wait a tiny
time-slice (which is far smaller than the timer granularity). When the timer
really elapses (e.g. a tokio timer after 1ms) only 1 packet can be sent
again.

This change improves on this by deriving the pacer capacity based on
how big bursts should be, and how much delay we want to have between
those. A pacer delay bigger than timer granularity is desirable for
efficiency and performance reasons. Here 2ms had been chosen, which
had proven effective in tests with an injected RTT.

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);
     }
 }

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,