Introduce max_outgoing_bytes_per_second option

Author: aochagavia

quinn-proto/src/config/transport.rs

@@ -41,6 +41,7 @@ pub struct TransportConfig {
     pub(crate) mtu_discovery_config: Option<MtuDiscoveryConfig>,
     pub(crate) pad_to_mtu: bool,
     pub(crate) ack_frequency_config: Option<AckFrequencyConfig>,
+    pub(crate) max_outgoing_bytes_per_second: Option<u64>,
 
     pub(crate) persistent_congestion_threshold: u32,
     pub(crate) keep_alive_interval: Option<Duration>,
@@ -243,6 +244,14 @@ impl TransportConfig {
         self
     }
 
+    /// Configures an outbound rate limit (in bytes per second) for each connection.
+    ///
+    /// Defaults to `None`, which disables rate limiting.
+    pub fn max_outgoing_bytes_per_second(&mut self, value: Option<u64>) -> &mut Self {
+        self.max_outgoing_bytes_per_second = value;
+        self
+    }
+
     /// Number of consecutive PTOs after which network is considered to be experiencing persistent congestion.
     pub fn persistent_congestion_threshold(&mut self, value: u32) -> &mut Self {
         self.persistent_congestion_threshold = value;
@@ -376,6 +385,7 @@ impl Default for TransportConfig {
             mtu_discovery_config: Some(MtuDiscoveryConfig::default()),
             pad_to_mtu: false,
             ack_frequency_config: None,
+            max_outgoing_bytes_per_second: None,
 
             persistent_congestion_threshold: 3,
             keep_alive_interval: None,
@@ -413,6 +423,7 @@ impl fmt::Debug for TransportConfig {
             mtu_discovery_config,
             pad_to_mtu,
             ack_frequency_config,
+            max_outgoing_bytes_per_second,
             persistent_congestion_threshold,
             keep_alive_interval,
             crypto_buffer_size,
@@ -442,6 +453,10 @@ impl fmt::Debug for TransportConfig {
             .field("mtu_discovery_config", mtu_discovery_config)
             .field("pad_to_mtu", pad_to_mtu)
             .field("ack_frequency_config", ack_frequency_config)
+            .field(
+                "max_outgoing_bytes_per_second",
+                max_outgoing_bytes_per_second,
+            )
             .field(
                 "persistent_congestion_threshold",
                 persistent_congestion_threshold,

quinn-proto/src/connection/pacing.rs

@@ -17,22 +17,36 @@ pub(super) struct Pacer {
     last_window: u64,
     last_mtu: u16,
     tokens: u64,
+    max_bytes_per_second: Option<u64>,
     prev: Instant,
 }
 
 impl Pacer {
     /// Obtains a new [`Pacer`].
-    pub(super) fn new(smoothed_rtt: Duration, window: u64, mtu: u16, now: Instant) -> Self {
+    pub(super) fn new(
+        smoothed_rtt: Duration,
+        window: u64,
+        mtu: u16,
+        max_bytes_per_second: Option<u64>,
+        now: Instant,
+    ) -> Self {
+        let window = rate_limited_window(smoothed_rtt, window, max_bytes_per_second);
         let capacity = optimal_capacity(smoothed_rtt, window, mtu);
         Self {
             capacity,
             last_window: window,
             last_mtu: mtu,
             tokens: capacity,
+            max_bytes_per_second,
             prev: now,
         }
     }
 
+    /// Obtains the `max_bytes_per_second` used when this [`Pacer`] was constructed.
+    pub(crate) fn max_bytes_per_second(&self) -> Option<u64> {
+        self.max_bytes_per_second
+    }
+
     /// Record that a packet has been transmitted.
     pub(super) fn on_transmit(&mut self, packet_length: u16) {
         self.tokens = self.tokens.saturating_sub(packet_length.into())
@@ -58,6 +72,7 @@ impl Pacer {
             "zero-sized congestion control window is nonsense"
         );
 
+        let window = rate_limited_window(smoothed_rtt, window, self.max_bytes_per_second);
         if window != self.last_window || mtu != self.last_mtu {
             self.capacity = optimal_capacity(smoothed_rtt, window, mtu);
 
@@ -147,6 +162,27 @@ fn optimal_capacity(smoothed_rtt: Duration, window: u64, mtu: u16) -> u64 {
     )
 }
 
+/// 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.
+fn rate_limited_window(
+    smoothed_rtt: Duration,
+    window: u64,
+    max_bytes_per_second: Option<u64>,
+) -> u64 {
+    let Some(max_bytes_per_second) = max_bytes_per_second else {
+        return window;
+    };
+
+    let rate_window = max_bytes_per_second as f64 * smoothed_rtt.as_secs_f64();
+
+    // the pacer refills tokens at x1.25 speed, so we shrink the window to cancel out the speedup
+    // (otherwise the actual sending rate could be higher than `max_bytes_per_second`)
+    let adjusted_rate_window = (rate_window / 1.25).round();
+
+    Ord::min(window, Ord::max(adjusted_rate_window as u64, 1))
+}
+
 /// Period of traffic to batch together on a reasonably fast connection
 const TARGET_BURST_INTERVAL: Duration = Duration::from_millis(2);
 
@@ -173,17 +209,17 @@ mod tests {
         let rtt = Duration::from_micros(400);
 
         assert!(
-            Pacer::new(rtt, 30000, 1500, new_instant)
+            Pacer::new(rtt, 30000, 1500, None, new_instant)
                 .delay(Duration::from_micros(0), 0, 1500, 1, old_instant)
                 .is_none()
         );
         assert!(
-            Pacer::new(rtt, 30000, 1500, new_instant)
+            Pacer::new(rtt, 30000, 1500, None, new_instant)
                 .delay(Duration::from_micros(0), 1600, 1500, 1, old_instant)
                 .is_none()
         );
         assert!(
-            Pacer::new(rtt, 30000, 1500, new_instant)
+            Pacer::new(rtt, 30000, 1500, None, new_instant)
                 .delay(Duration::from_micros(0), 1500, 1500, 3000, old_instant)
                 .is_none()
         );
@@ -196,18 +232,18 @@ mod tests {
         let rtt = Duration::from_millis(50);
         let now = Instant::now();
 
-        let pacer = Pacer::new(rtt, window, mtu, now);
+        let pacer = Pacer::new(rtt, window, mtu, None, now);
         assert_eq!(
             pacer.capacity,
             (window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64
         );
         assert_eq!(pacer.tokens, pacer.capacity);
 
-        let pacer = Pacer::new(Duration::from_millis(0), window, mtu, now);
+        let pacer = Pacer::new(Duration::from_millis(0), window, mtu, None, 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);
+        let pacer = Pacer::new(rtt, 1, mtu, None, now);
         assert_eq!(pacer.capacity, mtu as u64);
         assert_eq!(pacer.tokens, pacer.capacity);
     }
@@ -219,7 +255,7 @@ mod tests {
         let rtt = Duration::from_millis(50);
         let now = Instant::now();
 
-        let mut pacer = Pacer::new(rtt, window, mtu, now);
+        let mut pacer = Pacer::new(rtt, window, mtu, None, now);
         assert_eq!(
             pacer.capacity,
             (window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64
@@ -258,7 +294,7 @@ mod tests {
         let rtt = Duration::from_millis(50);
         let old_instant = Instant::now();
 
-        let mut pacer = Pacer::new(rtt, window, mtu, old_instant);
+        let mut pacer = Pacer::new(rtt, window, mtu, None, old_instant);
         let packet_capacity = pacer.capacity / mtu as u64;
 
         for _ in 0..packet_capacity {
@@ -321,4 +357,38 @@ mod tests {
         );
         assert_eq!(pacer.tokens, pacer.capacity);
     }
+
+    #[test]
+    fn computes_pause_correctly_for_rate_limited() {
+        let window = 2_000_000u64;
+        let mtu = 1000;
+        let rtt = Duration::from_millis(50);
+        let old_instant = Instant::now();
+
+        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,
+            "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)
+            .expect("Send must be delayed")
+            .duration_since(old_instant);
+
+        let expected_delay = Duration::from_millis(500);
+        let diff = actual_delay.abs_diff(expected_delay);
+
+        // Allow up to 2ns difference due to rounding
+        assert!(
+            diff < Duration::from_nanos(2),
+            "expected ≈ {expected_delay:?}, got {actual_delay:?} (diff {diff:?})"
+        );
+
+        // 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);
+    }
 }

quinn-proto/src/connection/paths.rs

@@ -75,6 +75,7 @@ impl PathData {
                 config.initial_rtt,
                 congestion.initial_window(),
                 config.get_initial_mtu(),
+                config.max_outgoing_bytes_per_second,
                 now,
             ),
             congestion,
@@ -118,7 +119,13 @@ impl PathData {
         Self {
             remote,
             rtt: prev.rtt,
-            pacing: Pacer::new(smoothed_rtt, congestion.window(), prev.current_mtu(), now),
+            pacing: Pacer::new(
+                smoothed_rtt,
+                congestion.window(),
+                prev.current_mtu(),
+                prev.pacing.max_bytes_per_second(),
+                now,
+            ),
             sending_ecn: true,
             congestion,
             challenge: None,