Check for app-limited at the beginning of the packet recv / send loop

The new logic matches the app-limited detection algorithm described in
the BBR RFC draft, which involves checking for the previous state of the
connection before timeout and ack processing.

Author: antoniovicente

quiche/src/lib.rs

@@ -575,6 +575,7 @@ pub struct Config {
     grease: bool,
 
     cc_algorithm: CongestionControlAlgorithm,
+    enable_bbr_app_limited_fix: bool,
     custom_bbr_params: Option<BbrParams>,
     initial_congestion_window_packets: usize,
     enable_relaxed_loss_threshold: bool,
@@ -653,6 +654,7 @@ impl Config {
             application_protos: Vec::new(),
             grease: true,
             cc_algorithm: CongestionControlAlgorithm::CUBIC,
+            enable_bbr_app_limited_fix: false,
             custom_bbr_params: None,
             initial_congestion_window_packets:
                 DEFAULT_INITIAL_CONGESTION_WINDOW_PACKETS,
@@ -1067,6 +1069,15 @@ impl Config {
         self.cc_algorithm = algo;
     }
 
+    /// Enable a rework of how the BBR congestion control implementation
+    /// computes app-limited.  This config parameter will be removed after we
+    /// build confidence on the new implementation or decide to roll it back.
+    ///
+    /// Defaults to `false`.
+    pub fn set_enable_bbr_app_limited_fix(&mut self, value: bool) {
+        self.enable_bbr_app_limited_fix = value;
+    }
+
     /// Sets custom BBR settings.
     ///
     /// This API is experimental and will be removed in the future.
@@ -2233,6 +2244,32 @@ impl<F: BufFactory> Connection<F> {
         Ok(())
     }
 
+    /// Enable a rework of how the BBR congestion control implementation
+    /// computes app-limited.  This config parameter will be removed after we
+    /// build confidence on the new implementation or decide to roll it back.
+    ///
+    /// Currently this only applies if cc_algorithm is
+    /// `CongestionControlAlgorithm::Bbr2Gcongestion`.
+    ///
+    /// This function can only be called inside one of BoringSSL's handshake
+    /// callbacks, before any packet has been sent. Calling this function any
+    /// other time will have no effect.
+    ///
+    /// See [`Config::set_enable_bbr_app_limited_fix()`].
+    ///
+    /// [`Config::set_enable_bbr_app_limited_fix()`]: struct.Config.html#method.set_enable_bbr_app_limited_fix
+    #[cfg(feature = "boringssl-boring-crate")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "boringssl-boring-crate")))]
+    pub fn set_enable_bbr_app_limited_fix_in_handshake(
+        ssl: &mut boring::ssl::SslRef, value: bool,
+    ) -> Result<()> {
+        let ex_data = tls::ExData::from_ssl_ref(ssl).ok_or(Error::TlsFail)?;
+
+        ex_data.recovery_config.enable_bbr_app_limited_fix = value;
+
+        Ok(())
+    }
+
     /// Sets the congestion control algorithm used by string.
     ///
     /// This function can only be called inside one of BoringSSL's handshake
@@ -2494,6 +2531,51 @@ impl<F: BufFactory> Connection<F> {
         Ok(())
     }
 
+    /// Returns true if at least 1 stream has headers or body data to write.
+    pub fn has_flushable_stream(&self) -> bool {
+        self.streams.has_flushable()
+    }
+
+    /// Signal the start of an iteration of the event loop that will
+    /// receive packets and then attempt to send packets.
+    ///
+    /// This hook is used to implement the "Detecting application-limited
+    /// phases" logic described in the BBR RFC draft which is described at:
+    /// https://www.ietf.org/archive/id/draft-ietf-ccwg-bbr-04.html#name-detecting-application-limit
+    ///
+    /// This function must be invoked at the beginning of each iteration of the
+    /// connection work loop, before receiving packets from the network and
+    /// processing ACKs/timeouts.
+    ///
+    /// The expected work loop order is:
+    /// 1. Call work_loop_round_start with the value of has_flushable_stream
+    ///    from the end of the previous loop, before polling for additional
+    ///    application data.
+    /// 2. Process timeouts by calling conn.on_timeout().
+    /// 3. Call conn.recv() to process received packets which contain ACKs and
+    ///    stream data.
+    /// 4. Call conn.send_on_path() to generate new packets and send them.
+    /// 5. Save the value of conn.has_flushable_stream() for the next iteration.
+    /// 6. Poll for additional data from upstream or wait for the next send
+    ///    timeout.
+    ///
+    /// In cases where the new packet generate + send loop yields early due to
+    /// an interation limit or the sendmsg on the socket returning EAGAIN, we
+    /// should expect had_flushable_stream_before_poll to remain true since the
+    /// connection wasn't able to send all the available data despite not
+    /// consuming the full congestion window.  The call to
+    /// bbr_check_if_app_limited() will not mark the connection app-limited
+    /// at the last-sent-packet-number because had_flushable_stream_before_poll
+    /// is true; this is correct and expected behavior.
+    pub fn work_loop_round_start(
+        &mut self, had_flushable_stream_before_poll: bool, now: &Instant,
+    ) {
+        for (_, path) in self.paths.iter_mut() {
+            path.recovery
+                .bbr_check_if_app_limited(had_flushable_stream_before_poll, now);
+        }
+    }
+
     /// Processes QUIC packets received from the peer.
     ///
     /// On success the number of bytes processed from the input buffer is

quiche/src/recovery/congestion/recovery.rs

@@ -956,7 +956,6 @@ impl RecoveryOps for LegacyRecovery {
         self.detect_lost_packets(epoch, now, "")
     }
 
-    // FIXME only used by gcongestion
     fn on_app_limited(&mut self) {
         // Not implemented for legacy recovery, update_app_limited and
         // delivery_rate_update_app_limited used instead.
@@ -1035,6 +1034,13 @@ impl RecoveryOps for LegacyRecovery {
         false
     }
 
+    fn bbr_check_if_app_limited(
+        &mut self, _had_flushable_stream_before_poll: bool, _now: &Instant,
+    ) {
+        // Not implemented -- only used by the BBR implementation
+        // which lives in the gcongestion directory.
+    }
+
     fn lost_count(&self) -> usize {
         self.congestion.lost_count
     }

quiche/src/recovery/gcongestion/bbr/bandwidth_sampler.rs

@@ -501,7 +501,7 @@ impl BandwidthSampler {
         }
     }
 
-    #[allow(dead_code)]
+    #[cfg(test)]
     pub(crate) fn is_app_limited(&self) -> bool {
         self.is_app_limited
     }

quiche/src/recovery/gcongestion/bbr2.rs

@@ -739,8 +739,9 @@ impl CongestionControl for BBRv2 {
         self.last_quiescence_start.is_none()
     }
 
-    fn is_cwnd_limited(&self, bytes_in_flight: usize) -> bool {
-        bytes_in_flight >= self.get_congestion_window()
+    #[cfg(test)]
+    fn is_app_limited(&self) -> bool {
+        self.mode.network_model().is_app_limited()
     }
 
     fn pacing_rate(

quiche/src/recovery/gcongestion/bbr2/network_model.rs

@@ -753,6 +753,11 @@ impl BBRv2NetworkModel {
         self.bandwidth_sampler.on_app_limited()
     }
 
+    #[cfg(test)]
+    pub(super) fn is_app_limited(&self) -> bool {
+        self.bandwidth_sampler.is_app_limited()
+    }
+
     pub(super) fn loss_events_in_round(&self) -> usize {
         self.loss_events_in_round
     }

quiche/src/recovery/gcongestion/mod.rs

@@ -110,8 +110,8 @@ pub(super) trait CongestionControl: Debug {
 
     fn is_in_recovery(&self) -> bool;
 
-    #[allow(dead_code)]
-    fn is_cwnd_limited(&self, bytes_in_flight: usize) -> bool;
+    #[cfg(test)]
+    fn is_app_limited(&self) -> bool;
 
     fn pacing_rate(
         &self, bytes_in_flight: usize, rtt_stats: &RttStats,

quiche/src/recovery/gcongestion/pacer.rs

@@ -273,12 +273,7 @@ impl Pacer {
     }
 
     #[cfg(test)]
-    pub fn is_app_limited(&self, bytes_in_flight: usize) -> bool {
-        !self.is_cwnd_limited(bytes_in_flight)
-    }
-
-    #[cfg(test)]
-    fn is_cwnd_limited(&self, bytes_in_flight: usize) -> bool {
-        !self.pacing_limited && self.sender.is_cwnd_limited(bytes_in_flight)
+    pub fn is_app_limited(&self) -> bool {
+        self.sender.is_app_limited()
     }
 }

quiche/src/recovery/gcongestion/recovery.rs

@@ -31,6 +31,7 @@ use crate::recovery::RecoveryConfig;
 use crate::recovery::RecoveryOps;
 use crate::recovery::RecoveryStats;
 use crate::recovery::ReleaseDecision;
+use crate::recovery::ReleaseTime;
 use crate::recovery::Sent;
 use crate::recovery::StartupExit;
 use crate::recovery::GRANULARITY;
@@ -466,6 +467,8 @@ pub struct GRecovery {
     lost_reuse: Vec<Lost>,
 
     pacer: Pacer,
+
+    enable_bbr_app_limited_fix: bool,
 }
 
 impl GRecovery {
@@ -520,6 +523,8 @@ impl GRecovery {
 
             newly_acked: Vec::new(),
             lost_reuse: Vec::new(),
+            enable_bbr_app_limited_fix: recovery_config
+                .enable_bbr_app_limited_fix,
         })
     }
 
@@ -639,6 +644,14 @@ impl GRecovery {
             self.loss_timer.update(timeout);
         }
     }
+
+    fn next_release_time_in_past(&self, now: &Instant) -> bool {
+        let next_release_time = self.pacer.get_next_release_time();
+        match next_release_time.time {
+            ReleaseTime::Immediate => true,
+            ReleaseTime::At(time) => time.lt(now),
+        }
+    }
 }
 
 impl RecoveryOps for GRecovery {
@@ -1035,7 +1048,9 @@ impl RecoveryOps for GRecovery {
 
     // FIXME only used by gcongestion
     fn on_app_limited(&mut self) {
-        self.pacer.on_app_limited(self.bytes_in_flight.get())
+        if !self.enable_bbr_app_limited_fix {
+            self.pacer.on_app_limited(self.bytes_in_flight.get())
+        }
     }
 
     #[cfg(test)]
@@ -1099,7 +1114,7 @@ impl RecoveryOps for GRecovery {
 
     #[cfg(test)]
     fn app_limited(&self) -> bool {
-        self.pacer.is_app_limited(self.bytes_in_flight.get())
+        self.pacer.is_app_limited()
     }
 
     // FIXME only used by congestion
@@ -1124,6 +1139,47 @@ impl RecoveryOps for GRecovery {
         true
     }
 
+    fn bbr_check_if_app_limited(
+        &mut self, had_flushable_stream_before_poll: bool, now: &Instant,
+    ) {
+        if !self.enable_bbr_app_limited_fix {
+            return;
+        }
+
+        // Implements CheckIfApplicationLimited from the BBR RFC.
+        //
+        // had_flushable_stream_before_poll is used to check for `NoUnsentData()`
+        // and skips this check there is data on the connection's tx
+        // buffer. Retransmisions are done with the help of stream send
+        // buffers so the `NoUnsentData()` check also covers the
+        // `C.lost_out <= C.retrans_out` check.
+        //
+        // The cwnd vs inflight check needs to take frame overheads
+        // into account since due to these overheads it may not be
+        // possible for the connection to use every last byte of the
+        // available window.  The check could be relaxed further by
+        // asserting that cwnd - inflight is less than 1 packet's
+        // worth of bytes.
+        //
+        // The check for C.pending_transmissions == 0 is done by the
+        // call to next_release_time_in_past; this method returns true
+        // when lower layers and the pacer would send a packet on the
+        // network immediately if one were provided.
+        //
+        // The case of Immediate next_release_time results in the work_loop
+        // iterating until min(cwnd, tx buffer) is exhausted.  If we decide to
+        // exit the send loop earlier we should mark the connection in some way so
+        // the next call to bbr_check_if_app_limited does not mark the
+        // connection app-limited after yielding when C.pending_transmissions > 0.
+        if !had_flushable_stream_before_poll &&
+            self.cwnd() >
+                self.bytes_in_flight.get() + frame::MAX_STREAM_OVERHEAD &&
+            self.next_release_time_in_past(now)
+        {
+            self.pacer.on_app_limited(self.bytes_in_flight.get())
+        }
+    }
+
     #[cfg(feature = "qlog")]
     fn state_str(&self, _now: Instant) -> &'static str {
         self.pacer.state_str()

quiche/src/recovery/mod.rs

@@ -130,6 +130,7 @@ pub struct RecoveryConfig {
     pub max_send_udp_payload_size: usize,
     pub max_ack_delay: Duration,
     pub cc_algorithm: CongestionControlAlgorithm,
+    pub enable_bbr_app_limited_fix: bool,
     pub custom_bbr_params: Option<BbrParams>,
     pub hystart: bool,
     pub pacing: bool,
@@ -145,6 +146,7 @@ impl RecoveryConfig {
             max_send_udp_payload_size: config.max_send_udp_payload_size,
             max_ack_delay: Duration::ZERO,
             cc_algorithm: config.cc_algorithm,
+            enable_bbr_app_limited_fix: config.enable_bbr_app_limited_fix,
             custom_bbr_params: config.custom_bbr_params,
             hystart: config.hystart,
             pacing: config.pacing,
@@ -315,6 +317,10 @@ pub trait RecoveryOps {
     fn get_next_release_time(&self) -> ReleaseDecision;
 
     fn gcongestion_enabled(&self) -> bool;
+
+    fn bbr_check_if_app_limited(
+        &mut self, had_flushable_stream_before_poll: bool, now: &Instant,
+    );
 }
 
 impl Recovery {

quiche/src/test_utils.rs

@@ -340,9 +340,16 @@ pub fn recv_send<F: BufFactory>(
     Ok(off)
 }
 
+pub fn run_work_loop_round_start_hook(conn: &mut Connection) {
+    let has_flushable_stream = conn.has_flushable_stream();
+    conn.work_loop_round_start(has_flushable_stream, &Instant::now());
+}
+
 pub fn process_flight(
     conn: &mut Connection, flight: Vec<(Vec<u8>, SendInfo)>,
 ) -> Result<()> {
+    run_work_loop_round_start_hook(conn);
+
     for (mut pkt, si) in flight {
         let info = RecvInfo {
             to: si.to,