modified Pacer such that congestion controller metrics can be used to adjust the burst size and the overall pacing rate of the connection

Author: Tipuch

quinn-proto/src/congestion.rs

@@ -85,11 +85,14 @@ pub trait Controller: Send + Sync {
     fn window(&self) -> u64;
 
     /// Retrieve implementation-specific metrics used to populate `qlog` traces when they are enabled
+    /// This is also used to alter the pacing of the connection with
+    /// `pacing_rate` and `send_quantum`
     fn metrics(&self) -> ControllerMetrics {
         ControllerMetrics {
             congestion_window: self.window(),
             ssthresh: None,
             pacing_rate: None,
+            send_quantum: None,
         }
     }
 
@@ -103,16 +106,20 @@ pub trait Controller: Send + Sync {
     fn into_any(self: Box<Self>) -> Box<dyn Any>;
 }
 
-/// Common congestion controller metrics
+/// Common congestion controller metrics used both for logging purposes
+/// but also to alter the pacing of the connection with
+/// `pacing_rate` and `send_quantum`
 #[derive(Default)]
 #[non_exhaustive]
 pub struct ControllerMetrics {
     /// Congestion window (bytes)
     pub congestion_window: u64,
     /// Slow start threshold (bytes)
     pub ssthresh: Option<u64>,
-    /// Pacing rate (bits/s)
+    /// Pacing rate (bytes/s)
     pub pacing_rate: Option<u64>,
+    /// Send Quantum (bytes) used to control the size of packet bursts
+    pub send_quantum: Option<u64>,
 }
 
 /// Constructs controllers on demand

quinn-proto/src/congestion/bbr/mod.rs

@@ -497,7 +497,8 @@ impl Controller for Bbr {
         ControllerMetrics {
             congestion_window: self.window(),
             ssthresh: None,
-            pacing_rate: Some(self.pacing_rate * 8),
+            pacing_rate: Some(self.pacing_rate),
+            send_quantum: None,
         }
     }
 

quinn-proto/src/congestion/cubic.rs

@@ -254,6 +254,7 @@ impl Controller for Cubic {
             congestion_window: self.window(),
             ssthresh: Some(self.state.ssthresh),
             pacing_rate: None,
+            send_quantum: None,
         }
     }
 

quinn-proto/src/congestion/new_reno.rs

@@ -120,6 +120,7 @@ impl Controller for NewReno {
             congestion_window: self.window(),
             ssthresh: Some(self.ssthresh),
             pacing_rate: None,
+            send_quantum: None,
         }
     }
 

quinn-proto/src/connection/mod.rs

@@ -615,12 +615,15 @@ impl Connection {
 
                     // Check whether the next datagram is blocked by pacing
                     let smoothed_rtt = self.path.rtt.get();
+                    let controller_metrics = self.path.congestion.metrics();
                     if let Some(delay) = self.path.pacing.delay(
                         smoothed_rtt,
                         bytes_to_send,
                         self.path.current_mtu(),
                         self.path.congestion.window(),
                         now,
+                        controller_metrics.send_quantum,
+                        controller_metrics.pacing_rate,
                     ) {
                         self.timers.set(Timer::Pacing, delay);
                         congestion_blocked = true;

quinn-proto/src/connection/pacing.rs

@@ -45,13 +45,22 @@ impl Pacer {
     ///
     /// The 5/4 ratio used here comes from the suggestion that N = 1.25 in the draft IETF RFC for
     /// QUIC.
+    /// `capacity` in bytes is used to cap the number of bytes sent at once.
+    /// It can be leveraged in congestion control to regulate the maximum size of transmission aggregates.
+    /// e.g: <https://www.ietf.org/archive/id/draft-ietf-ccwg-bbr-04.html#name-send-quantum-csend_quantum>
+    ///
+    /// `pacing_rate` in bytes/sec is used to pace the connection and control the upper limit of how fast
+    /// we're sending data it can be leveraged in congestion control
+    /// 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>,
     ) -> Option<Instant> {
         debug_assert_ne!(
             window, 0,
@@ -61,12 +70,29 @@ impl Pacer {
         if window != self.last_window || mtu != self.last_mtu {
             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;
+            // here we cap the number of bytes sent at once during a burst
+            self.tokens = self.capacity.min(self.tokens);
+        }
+
+        if let Some(pacing_rate) = pacing_rate {
+            // if the bytes to send are below or equal to our maximum burst size there is no need for delay
+            if bytes_to_send <= self.capacity {
+                return None;
+            }
+            let capped_bytes_to_send = bytes_to_send.max(self.capacity);
+            // otherwise we calculate the delay such that we send at pacing_rate
+            let delay = Duration::from_secs_f64(capped_bytes_to_send as f64 / pacing_rate as f64);
+            return Some(now + delay);
+        }
+
         // if we can already send a packet, there is no need for delay
         if self.tokens >= bytes_to_send {
             return None;
@@ -174,17 +200,41 @@ mod tests {
 
         assert!(
             Pacer::new(rtt, 30000, 1500, new_instant)
-                .delay(Duration::from_micros(0), 0, 1500, 1, old_instant)
+                .delay(
+                    Duration::from_micros(0),
+                    0,
+                    1500,
+                    1,
+                    old_instant,
+                    None,
+                    None
+                )
                 .is_none()
         );
         assert!(
             Pacer::new(rtt, 30000, 1500, new_instant)
-                .delay(Duration::from_micros(0), 1600, 1500, 1, old_instant)
+                .delay(
+                    Duration::from_micros(0),
+                    1600,
+                    1500,
+                    1,
+                    old_instant,
+                    None,
+                    None
+                )
                 .is_none()
         );
         assert!(
             Pacer::new(rtt, 30000, 1500, new_instant)
-                .delay(Duration::from_micros(0), 1500, 1500, 3000, old_instant)
+                .delay(
+                    Duration::from_micros(0),
+                    1500,
+                    1500,
+                    3000,
+                    old_instant,
+                    None,
+                    None
+                )
                 .is_none()
         );
     }
@@ -227,27 +277,27 @@ mod tests {
         assert_eq!(pacer.tokens, pacer.capacity);
         let initial_tokens = pacer.tokens;
 
-        pacer.delay(rtt, mtu as u64, mtu, window * 2, now);
+        pacer.delay(rtt, mtu as u64, mtu, window * 2, now, None, None);
         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);
+        pacer.delay(rtt, mtu as u64, mtu, window / 2, now, None, None);
         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);
+        pacer.delay(rtt, mtu as u64, mtu * 2, window, now, None, None);
         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);
+        pacer.delay(rtt, mtu as u64, 20_000, window, now, None, None);
         assert_eq!(pacer.capacity, 20_000_u64 * MIN_BURST_SIZE);
     }
 
@@ -263,7 +313,7 @@ mod tests {
 
         for _ in 0..packet_capacity {
             assert_eq!(
-                pacer.delay(rtt, mtu as u64, mtu, window, old_instant),
+                pacer.delay(rtt, mtu as u64, mtu, window, old_instant, None, None),
                 None,
                 "When capacity is available packets should be sent immediately"
             );
@@ -274,7 +324,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)
+            .delay(rtt, mtu as u64, mtu, window, old_instant, None, None)
             .expect("Send must be delayed")
             .duration_since(old_instant);
 
@@ -292,15 +342,17 @@ mod tests {
                 mtu as u64,
                 mtu,
                 window,
-                old_instant + pace_duration / 2
+                old_instant + pace_duration / 2,
+                None,
+                None
             ),
             None
         );
         assert_eq!(pacer.tokens, pacer.capacity / 2);
 
         for _ in 0..packet_capacity / 2 {
             assert_eq!(
-                pacer.delay(rtt, mtu as u64, mtu, window, old_instant),
+                pacer.delay(rtt, mtu as u64, mtu, window, old_instant, None, None),
                 None,
                 "When capacity is available packets should be sent immediately"
             );
@@ -315,10 +367,40 @@ mod tests {
                 mtu as u64,
                 mtu,
                 window,
-                old_instant + pace_duration * 3 / 2
+                old_instant + pace_duration * 3 / 2,
+                None,
+                None
             ),
             None
         );
         assert_eq!(pacer.tokens, pacer.capacity);
     }
+
+    #[test]
+    fn adjusts_capacity_from_congestion() {
+        let window = 2_000_000;
+        let mtu = 1500;
+        let rtt = Duration::from_millis(50);
+        let now = Instant::now();
+        let send_quantum = Some(20000);
+        let mut pacer = Pacer::new(rtt, window, mtu, now);
+        pacer.delay(rtt, mtu as u64, mtu, window * 2, now, send_quantum, None);
+        assert_eq!(pacer.capacity, 20000);
+    }
+
+    #[test]
+    fn adjusts_pacing_from_congestion_metrics() {
+        let window = 2_000_000;
+        let mtu = 1500;
+        let rtt = Duration::from_millis(50);
+        let now = Instant::now();
+        let send_quantum = Some(20000);
+        let pacing_rate = Some(200_000_000); //200 MB/s in bytes/s
+        let mut pacer = Pacer::new(rtt, window, mtu, now);
+        // 40_000 B / 200_000_000 MB/s = 0.0002 sec = 200 microseconds
+        assert_eq!(
+            pacer.delay(rtt, 40000, mtu, window, now, send_quantum, pacing_rate),
+            Some(now + Duration::from_micros(200))
+        )
+    }
 }