feat(s2n-quic-core): add congestion control events (#1521)

* feat(s2n-quic-core): add congestion controller events

* Wire up on_slow_start_exited

* Remove is_slow_start from congestion controller trait

* Add delivery rate and pacing rate events

* New snapshot results

* Adjust ranges in fuzz test

* add pacing fuzz test corpus

* Make congestion_controller::Publisher a trait and other PR feedback

* Fix formatting

* Change delivery_rate_updated to delivery_rate_sampled and add all rate sample fields

Author: WesleyRosenblum

quic/s2n-quic-core/src/event.rs

@@ -53,6 +53,7 @@ ident_into_event!(
     i64,
     usize,
     isize,
+    f32,
     Duration,
     bool,
     connection::Error,

quic/s2n-quic-core/src/event/generated.rs

@@ -1,7 +1,7 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::time::Timestamp;
+use crate::{event, event::IntoEvent, recovery::congestion_controller::Publisher, time::Timestamp};
 use core::{
     cmp::{max, Ordering},
     time::Duration,
@@ -62,6 +62,17 @@ impl Bandwidth {
             }
         }
     }
+
+    /// Represents the bandwidth as bytes per second
+    pub fn as_bytes_per_second(&self) -> u64 {
+        const ONE_SECOND_IN_NANOS: u64 = Duration::from_secs(1).as_nanos() as u64;
+
+        if *self == Bandwidth::INFINITY {
+            return u64::MAX;
+        }
+
+        ONE_SECOND_IN_NANOS / self.nanos_per_byte
+    }
 }
 
 impl Default for Bandwidth {
@@ -170,6 +181,23 @@ impl RateSample {
     }
 }
 
+impl IntoEvent<event::builder::RateSample> for RateSample {
+    fn into_event(self) -> event::builder::RateSample {
+        event::builder::RateSample {
+            interval: self.interval.into_event(),
+            delivered_bytes: self.delivered_bytes.into_event(),
+            lost_bytes: self.lost_bytes.into_event(),
+            ecn_ce_count: self.ecn_ce_count.into_event(),
+            is_app_limited: self.is_app_limited.into_event(),
+            prior_delivered_bytes: self.prior_delivered_bytes.into_event(),
+            bytes_in_flight: self.bytes_in_flight.into_event(),
+            prior_lost_bytes: self.prior_lost_bytes.into_event(),
+            prior_ecn_ce_count: self.prior_ecn_ce_count.into_event(),
+            delivery_rate_bytes_per_second: self.delivery_rate().as_bytes_per_second().into_event(),
+        }
+    }
+}
+
 /// Bandwidth estimator as defined in [Delivery Rate Estimation](https://datatracker.ietf.org/doc/draft-cheng-iccrg-delivery-rate-estimation/)
 /// and [BBR Congestion Control](https://datatracker.ietf.org/doc/draft-cardwell-iccrg-bbr-congestion-control/).
 #[derive(Clone, Debug, Default)]
@@ -270,12 +298,13 @@ impl Estimator {
     //# rate sample based on a snapshot of connection delivery information from the time
     //# at which the packet was last transmitted.
     /// Called for each acknowledgement of one or more packets
-    pub fn on_ack(
+    pub fn on_ack<Pub: Publisher>(
         &mut self,
         bytes_acknowledged: usize,
         newest_acked_time_sent: Timestamp,
         newest_acked_packet_info: PacketInfo,
         now: Timestamp,
+        publisher: &mut Pub,
     ) {
         self.delivered_bytes += bytes_acknowledged as u64;
         self.delivered_time = Some(now);
@@ -318,6 +347,8 @@ impl Estimator {
         // so the values are up to date even when no loss or ECN CE markings are received.
         self.rate_sample.lost_bytes = self.lost_bytes - self.rate_sample.prior_lost_bytes;
         self.rate_sample.ecn_ce_count = self.ecn_ce_count - self.rate_sample.prior_ecn_ce_count;
+
+        publisher.on_delivery_rate_sampled(self.rate_sample);
     }
 
     /// Called when packets are declared lost

quic/s2n-quic-core/src/recovery/bandwidth/estimator.rs

@@ -0,0 +1,6 @@
+---
+source: quic/s2n-quic-core/src/recovery/bandwidth/estimator/tests.rs
+expression: ""
+---
+DeliveryRateSampled { path_id: 0, rate_sample: RateSample { interval: 1s, delivered_bytes: 1500, lost_bytes: 100, ecn_ce_count: 5, is_app_limited: false, prior_delivered_bytes: 15000, bytes_in_flight: 1500, prior_lost_bytes: 0, prior_ecn_ce_count: 0, delivery_rate_bytes_per_second: 1500 } }
+DeliveryRateSampled { path_id: 0, rate_sample: RateSample { interval: 1s, delivered_bytes: 1501, lost_bytes: 100, ecn_ce_count: 5, is_app_limited: false, prior_delivered_bytes: 15000, bytes_in_flight: 1500, prior_lost_bytes: 0, prior_ecn_ce_count: 0, delivery_rate_bytes_per_second: 1501 } }

quic/s2n-quic-core/src/recovery/bandwidth/estimator/snapshots/quic__s2n-quic-core__src__recovery__bandwidth__estimator__tests__events__app_limited.snap

@@ -0,0 +1,6 @@
+---
+source: quic/s2n-quic-core/src/recovery/bandwidth/estimator/tests.rs
+expression: ""
+---
+DeliveryRateSampled { path_id: 0, rate_sample: RateSample { interval: 4s, delivered_bytes: 1500, lost_bytes: 0, ecn_ce_count: 0, is_app_limited: false, prior_delivered_bytes: 0, bytes_in_flight: 0, prior_lost_bytes: 0, prior_ecn_ce_count: 0, delivery_rate_bytes_per_second: 375 } }
+DeliveryRateSampled { path_id: 0, rate_sample: RateSample { interval: 5s, delivered_bytes: 1500, lost_bytes: 0, ecn_ce_count: 0, is_app_limited: false, prior_delivered_bytes: 1500, bytes_in_flight: 1500, prior_lost_bytes: 0, prior_ecn_ce_count: 0, delivery_rate_bytes_per_second: 300 } }

quic/s2n-quic-core/src/recovery/bandwidth/estimator/snapshots/quic__s2n-quic-core__src__recovery__bandwidth__estimator__tests__events__on_packet_ack_implausible_ack_rate.snap

@@ -0,0 +1,7 @@
+---
+source: quic/s2n-quic-core/src/recovery/bandwidth/estimator/tests.rs
+expression: ""
+---
+DeliveryRateSampled { path_id: 0, rate_sample: RateSample { interval: 10s, delivered_bytes: 201500, lost_bytes: 150, ecn_ce_count: 15, is_app_limited: false, prior_delivered_bytes: 0, bytes_in_flight: 0, prior_lost_bytes: 0, prior_ecn_ce_count: 0, delivery_rate_bytes_per_second: 20150 } }
+DeliveryRateSampled { path_id: 0, rate_sample: RateSample { interval: 11s, delivered_bytes: 3000, lost_bytes: 0, ecn_ce_count: 0, is_app_limited: true, prior_delivered_bytes: 200000, bytes_in_flight: 3000, prior_lost_bytes: 150, prior_ecn_ce_count: 15, delivery_rate_bytes_per_second: 272 } }
+DeliveryRateSampled { path_id: 0, rate_sample: RateSample { interval: 11s, delivered_bytes: 4500, lost_bytes: 0, ecn_ce_count: 0, is_app_limited: true, prior_delivered_bytes: 200000, bytes_in_flight: 3000, prior_lost_bytes: 150, prior_ecn_ce_count: 15, delivery_rate_bytes_per_second: 409 } }

quic/s2n-quic-core/src/recovery/bandwidth/estimator/snapshots/quic__s2n-quic-core__src__recovery__bandwidth__estimator__tests__events__on_packet_ack_rate_sample.snap

@@ -2,7 +2,11 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use super::*;
-use crate::time::{Clock, NoopClock};
+use crate::{
+    event, path,
+    recovery::congestion_controller::PathPublisher,
+    time::{Clock, NoopClock},
+};
 
 #[test]
 fn bandwidth() {
@@ -138,6 +142,8 @@ fn on_packet_sent() {
 
 #[test]
 fn app_limited() {
+    let mut publisher = event::testing::Publisher::snapshot();
+    let mut publisher = PathPublisher::new(&mut publisher, path::Id::test_id());
     let first_sent_time = NoopClock.get_time();
     let delivered_time = first_sent_time + Duration::from_secs(1);
     let mut bw_estimator = Estimator {
@@ -176,18 +182,32 @@ fn app_limited() {
     };
 
     // Acknowledge all the bytes that were inflight when the app-limited period began
-    bw_estimator.on_ack(1500, delivered_time, packet_info, delivered_time);
+    bw_estimator.on_ack(
+        1500,
+        delivered_time,
+        packet_info,
+        delivered_time,
+        &mut publisher,
+    );
     // Still app_limited, since we need bytes to be acknowledged after the app limited period
     assert_eq!(Some(1500 + 15000), bw_estimator.app_limited_delivered_bytes);
 
     // Acknowledge one more byte
-    bw_estimator.on_ack(1, delivered_time, packet_info, delivered_time);
+    bw_estimator.on_ack(
+        1,
+        delivered_time,
+        packet_info,
+        delivered_time,
+        &mut publisher,
+    );
     // Now the app limited period is over
     assert_eq!(None, bw_estimator.app_limited_delivered_bytes);
 }
 
 #[test]
 fn on_packet_ack_rate_sample() {
+    let mut publisher = event::testing::Publisher::snapshot();
+    let mut publisher = PathPublisher::new(&mut publisher, path::Id::test_id());
     let t0 = NoopClock.get_time() + Duration::from_secs(60);
     let t1 = t0 + Duration::from_secs(1);
     let t2 = t0 + Duration::from_secs(2);
@@ -207,7 +227,7 @@ fn on_packet_ack_rate_sample() {
 
     let now = t0 + Duration::from_secs(10);
     let delivered_bytes = bw_estimator.delivered_bytes;
-    bw_estimator.on_ack(1500, t0, packet_1, now);
+    bw_estimator.on_ack(1500, t0, packet_1, now, &mut publisher);
 
     assert_eq!(bw_estimator.delivered_bytes, delivered_bytes + 1500);
     assert_eq!(bw_estimator.delivered_time, Some(now));
@@ -247,7 +267,7 @@ fn on_packet_ack_rate_sample() {
     // Ack a newer packet
     let now = now + Duration::from_secs(1);
     let delivered_bytes = bw_estimator.delivered_bytes;
-    bw_estimator.on_ack(1500, t2, packet_3, now);
+    bw_estimator.on_ack(1500, t2, packet_3, now, &mut publisher);
 
     assert_eq!(bw_estimator.delivered_bytes, delivered_bytes + 1500);
     assert_eq!(bw_estimator.delivered_time, Some(now));
@@ -283,7 +303,7 @@ fn on_packet_ack_rate_sample() {
     // Ack an older packet
     let now = now + Duration::from_secs(1);
     let delivered_bytes = bw_estimator.delivered_bytes;
-    bw_estimator.on_ack(1500, t1, packet_2, now);
+    bw_estimator.on_ack(1500, t1, packet_2, now, &mut publisher);
 
     assert_eq!(bw_estimator.delivered_bytes, delivered_bytes + 1500);
     assert_eq!(bw_estimator.delivered_time, Some(now));
@@ -327,19 +347,27 @@ fn on_packet_ack_rate_sample() {
 //# by capping the delivery rate sample to be no higher than the send rate.
 #[test]
 fn on_packet_ack_implausible_ack_rate() {
+    let mut publisher = event::testing::Publisher::snapshot();
+    let mut publisher = PathPublisher::new(&mut publisher, path::Id::test_id());
     let t0 = NoopClock.get_time();
     let mut bw_estimator = Estimator::default();
 
     // A packet is sent and acknowledged 4 seconds later
     let packet_info = bw_estimator.on_packet_sent(0, Some(false), t0);
     let t4 = t0 + Duration::from_secs(4);
-    bw_estimator.on_ack(1500, t0, packet_info, t4);
+    bw_estimator.on_ack(1500, t0, packet_info, t4, &mut publisher);
 
     // A packet is sent and acknowledged 1 second later
     let t5 = t0 + Duration::from_secs(5);
     let packet_info = bw_estimator.on_packet_sent(1500, Some(false), t5);
     let now = t0 + Duration::from_secs(6);
-    bw_estimator.on_ack(1500, t0 + Duration::from_secs(5), packet_info, now);
+    bw_estimator.on_ack(
+        1500,
+        t0 + Duration::from_secs(5),
+        packet_info,
+        now,
+        &mut publisher,
+    );
 
     let send_elapsed = t5 - packet_info.first_sent_time;
     let ack_elapsed = now - packet_info.delivered_time;
@@ -388,3 +416,12 @@ fn on_explicit_congestion() {
     assert_eq!(8, bw_estimator.ecn_ce_count);
     assert_eq!(8, bw_estimator.rate_sample.ecn_ce_count);
 }
+
+#[test]
+fn as_bytes_per_second() {
+    let bandwidth = Bandwidth::new(10_000, Duration::from_secs(1));
+
+    assert_eq!(10_000, bandwidth.as_bytes_per_second());
+    assert_eq!(0, Bandwidth::ZERO.as_bytes_per_second());
+    assert_eq!(u64::MAX, Bandwidth::INFINITY.as_bytes_per_second());
+}