feat(cca): BBR

quinn-proto/src/congestion.rs

@@ -1,27 +1,62 @@
 //! Logic for controlling the rate at which data is sent
 
-use std::time::{Duration, Instant};
+use crate::connection::paths::RttEstimator;
+use std::any::Any;
+use std::time::Instant;
 
+mod bbr;
 mod cubic;
 mod new_reno;
 
+pub use bbr::{Bbr, BbrConfig};
 pub use cubic::{Cubic, CubicConfig};
 pub use new_reno::{NewReno, NewRenoConfig};
 
 /// Common interface for different congestion controllers
 pub trait Controller: Send {
+    /// One or more packets were just sent
+    #[allow(unused_variables)]
+    fn on_sent(&mut self, now: Instant, bytes: u64, last_packet_number: u64) {}
+
     /// Packet deliveries were confirmed
     ///
     /// `app_limited` indicates whether the connection was blocked on outgoing
     /// application data prior to receiving these acknowledgements.
-    fn on_ack(&mut self, now: Instant, sent: Instant, bytes: u64, app_limited: bool, rtt: Duration);
+    #[allow(unused_variables)]
+    fn on_ack(
+        &mut self,
+        now: Instant,
+        sent: Instant,
+        bytes: u64,
+        app_limited: bool,
+        rtt: &RttEstimator,
+    ) {
+    }
+
+    /// Packets are acked in batches, all with the same `now` argument. This indicates one of those batches has completed.
+    #[allow(unused_variables)]
+    fn on_end_acks(
+        &mut self,
+        now: Instant,
+        in_flight: u64,
+        app_limited: bool,
+        largest_packet_num_acked: Option<u64>,
+    ) {
+    }
 
     /// Packets were deemed lost or marked congested
     ///
     /// `in_persistent_congestion` indicates whether all packets sent within the persistent
     /// congestion threshold period ending when the most recent packet in this batch was sent were
     /// lost.
-    fn on_congestion_event(&mut self, now: Instant, sent: Instant, is_persistent_congestion: bool);
+    /// `lost_bytes` indicates how many bytes were lost. This value will be 0 for ECN triggers.
+    fn on_congestion_event(
+        &mut self,
+        now: Instant,
+        sent: Instant,
+        is_persistent_congestion: bool,
+        lost_bytes: u64,
+    );
 
     /// Number of ack-eliciting bytes that may be in flight
     fn window(&self) -> u64;
@@ -31,6 +66,9 @@ pub trait Controller: Send {
 
     /// Initial congestion window
     fn initial_window(&self) -> u64;
+
+    /// Returns Self for use in down-casting to extract implementation details
+    fn into_any(self: Box<Self>) -> Box<dyn Any>;
 }
 
 /// Constructs controllers on demand

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

@@ -0,0 +1,121 @@
+use std::fmt::{Debug, Display, Formatter};
+use std::time::{Duration, Instant};
+
+use super::min_max::MinMax;
+
+#[derive(Clone, Debug)]
+pub(crate) struct BandwidthEstimation {
+    total_acked: u64,
+    prev_total_acked: u64,
+    acked_time: Option<Instant>,
+    prev_acked_time: Option<Instant>,
+    total_sent: u64,
+    prev_total_sent: u64,
+    sent_time: Option<Instant>,
+    prev_sent_time: Option<Instant>,
+    max_filter: MinMax,
+    acked_at_last_window: u64,
+}
+
+impl Default for BandwidthEstimation {
+    fn default() -> Self {
+        BandwidthEstimation {
+            total_acked: 0,
+            prev_total_acked: 0,
+            acked_time: None,
+            prev_acked_time: None,
+            total_sent: 0,
+            prev_total_sent: 0,
+            sent_time: None,
+            prev_sent_time: None,
+            max_filter: MinMax::new(10),
+            acked_at_last_window: 0,
+        }
+    }
+}
+
+impl Display for BandwidthEstimation {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{:.3} MB/s",
+            self.get_estimate() as f32 / (1024 * 1024) as f32
+        )
+    }
+}
+
+impl BandwidthEstimation {
+    pub fn on_sent(&mut self, now: Instant, bytes: u64) {
+        self.prev_total_sent = self.total_sent;
+        self.total_sent += bytes;
+        self.prev_sent_time = self.sent_time;
+        self.sent_time = Some(now);
+    }
+
+    pub fn on_ack(
+        &mut self,
+        now: Instant,
+        _sent: Instant,
+        bytes: u64,
+        round: u64,
+        app_limited: bool,
+    ) {
+        self.prev_total_acked = self.total_acked;
+        self.total_acked += bytes;
+        self.prev_acked_time = self.acked_time;
+        self.acked_time = Some(now);
+
+        if self.prev_sent_time.is_none() {
+            return;
+        }
+
+        let send_rate;
+        if self.sent_time.unwrap() > self.prev_sent_time.unwrap() {
+            send_rate = BandwidthEstimation::bw_from_delta(
+                self.total_sent - self.prev_total_sent,
+                self.sent_time.unwrap() - self.prev_sent_time.unwrap(),
+            )
+            .unwrap_or(0);
+        } else {
+            send_rate = u64::MAX; // will take the min of send and ack, so this is just a skip
+        }
+
+        let ack_rate;
+        if self.prev_acked_time.is_none() {
+            ack_rate = 0;
+        } else {
+            ack_rate = BandwidthEstimation::bw_from_delta(
+                self.total_acked - self.prev_total_acked,
+                self.acked_time.unwrap() - self.prev_acked_time.unwrap(),
+            )
+            .unwrap_or(0);
+        }
+
+        let bandwidth = send_rate.min(ack_rate);
+        if !app_limited && self.max_filter.get() < bandwidth {
+            self.max_filter.update_max(round, bandwidth);
+        }
+    }
+
+    pub fn bytes_acked_this_window(&self) -> u64 {
+        self.total_acked - self.acked_at_last_window
+    }
+
+    pub fn end_acks(&mut self, _current_round: u64, _app_limited: bool) {
+        self.acked_at_last_window = self.total_acked;
+    }
+
+    pub fn get_estimate(&self) -> u64 {
+        self.max_filter.get()
+    }
+
+    pub const fn bw_from_delta(bytes: u64, delta: Duration) -> Option<u64> {
+        let window_duration_ns = delta.as_nanos();
+        if window_duration_ns == 0 {
+            return None;
+        }
+        let b_ns = bytes * 1_000_000_000;
+        let bytes_per_second = b_ns / (window_duration_ns as u64);
+        Some(bytes_per_second)
+    }
+}

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

@@ -0,0 +1,150 @@
+/*
+ * Based on Google code released under BSD license here:
+ * https://groups.google.com/forum/#!topic/bbr-dev/3RTgkzi5ZD8
+ */
+
+/*
+ * Kathleen Nichols' algorithm for tracking the minimum (or maximum)
+ * value of a data stream over some fixed time interval.  (E.g.,
+ * the minimum RTT over the past five minutes.) It uses constant
+ * space and constant time per update yet almost always delivers
+ * the same minimum as an implementation that has to keep all the
+ * data in the window.
+ *
+ * The algorithm keeps track of the best, 2nd best & 3rd best min
+ * values, maintaining an invariant that the measurement time of
+ * the n'th best >= n-1'th best. It also makes sure that the three
+ * values are widely separated in the time window since that bounds
+ * the worse case error when that data is monotonically increasing
+ * over the window.
+ *
+ * Upon getting a new min, we can forget everything earlier because
+ * it has no value - the new min is <= everything else in the window
+ * by definition and it samples the most recent. So we restart fresh on
+ * every new min and overwrites 2nd & 3rd choices. The same property
+ * holds for 2nd & 3rd best.
+ */
+
+use std::fmt::Debug;
+
+#[derive(Debug, Copy, Clone, Default)]
+struct MinMaxSample {
+    /// round number, not a timestamp
+    time: u64,
+    value: u64,
+}
+
+#[derive(Copy, Clone, Debug)]
+pub(crate) struct MinMax {
+    /// round count, not a timestamp
+    window: u64,
+    samples: [MinMaxSample; 3],
+}
+
+impl MinMax {
+    pub fn new(round_window: u64) -> Self {
+        MinMax {
+            window: round_window,
+            samples: [Default::default(); 3],
+        }
+    }
+
+    pub fn get(&self) -> u64 {
+        self.samples[0].value
+    }
+
+    fn fill(&mut self, sample: MinMaxSample) {
+        self.samples.fill(sample);
+    }
+
+    pub fn reset(&mut self) {
+        self.fill(Default::default())
+    }
+
+    /// update_min is also defined in the original source, but removed here since it is not used.
+    pub fn update_max(&mut self, current_round: u64, measurement: u64) {
+        let sample = MinMaxSample {
+            time: current_round,
+            value: measurement,
+        };
+
+        if self.samples[0].value == 0  /* uninitialised */
+            || /* found new max? */ sample.value >= self.samples[0].value
+            || /* nothing left in window? */ sample.time - self.samples[2].time > self.window
+        {
+            self.fill(sample); /* forget earlier samples */
+            return;
+        }
+
+        if sample.value >= self.samples[1].value {
+            self.samples[2] = sample;
+            self.samples[1] = sample;
+        } else if sample.value >= self.samples[2].value {
+            self.samples[2] = sample;
+        }
+
+        self.subwin_update(sample);
+    }
+
+    /* As time advances, update the 1st, 2nd, and 3rd choices. */
+    fn subwin_update(&mut self, sample: MinMaxSample) {
+        let dt = sample.time - self.samples[0].time;
+        if dt > self.window {
+            /*
+             * Passed entire window without a new sample so make 2nd
+             * choice the new sample & 3rd choice the new 2nd choice.
+             * we may have to iterate this since our 2nd choice
+             * may also be outside the window (we checked on entry
+             * that the third choice was in the window).
+             */
+            self.samples[0] = self.samples[1];
+            self.samples[1] = self.samples[2];
+            self.samples[2] = sample;
+            if sample.time - self.samples[0].time > self.window {
+                self.samples[0] = self.samples[1];
+                self.samples[1] = self.samples[2];
+                self.samples[2] = sample;
+            }
+        } else if self.samples[1].time == self.samples[0].time && dt > self.window / 4 {
+            /*
+             * We've passed a quarter of the window without a new sample
+             * so take a 2nd choice from the 2nd quarter of the window.
+             */
+            self.samples[2] = sample;
+            self.samples[1] = sample;
+        } else if self.samples[2].time == self.samples[1].time && dt > self.window / 2 {
+            /*
+             * We've passed half the window without finding a new sample
+             * so take a 3rd choice from the last half of the window
+             */
+            self.samples[2] = sample;
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test() {
+        let round = 25;
+        let mut min_max = MinMax::new(10);
+        min_max.update_max(round + 1, 100);
+        assert_eq!(100, min_max.get());
+        min_max.update_max(round + 3, 120);
+        assert_eq!(120, min_max.get());
+        min_max.update_max(round + 5, 160);
+        assert_eq!(160, min_max.get());
+        min_max.update_max(round + 7, 100);
+        assert_eq!(160, min_max.get());
+        min_max.update_max(round + 10, 100);
+        assert_eq!(160, min_max.get());
+        min_max.update_max(round + 14, 100);
+        assert_eq!(160, min_max.get());
+        min_max.update_max(round + 16, 100);
+        assert_eq!(100, min_max.get());
+        min_max.update_max(round + 18, 130);
+        assert_eq!(130, min_max.get());
+    }
+}