feat(s2n-quic-platform): prioritized socket for s2n-quic's IO (#3014)

Author: boquan-fang

quic/s2n-quic-core/events/platform.rs

@@ -123,6 +123,18 @@ enum PlatformFeatureConfiguration {
     MaxMtu { mtu: u16 },
 }
 
+#[event("platform:rx_socket")]
+#[subject(endpoint)]
+/// Emitted for each receive socket with per-socket packet counts
+struct PlatformRxSocketStats {
+    /// Whether this socket is the prioritized socket
+    is_prioritized: bool,
+
+    /// The number of packets received on this socket since the last event
+    #[counter("packets.total")]
+    count: usize,
+}
+
 #[event("platform:event_loop_wakeup")]
 #[subject(endpoint)]
 struct PlatformEventLoopWakeup {

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

@@ -3107,6 +3107,27 @@ pub mod api {
     }
     #[derive(Clone, Debug)]
     #[non_exhaustive]
+    #[doc = " Emitted for each receive socket with per-socket packet counts"]
+    pub struct PlatformRxSocketStats {
+        #[doc = " Whether this socket is the prioritized socket"]
+        pub is_prioritized: bool,
+        #[doc = " The number of packets received on this socket since the last event"]
+        pub count: usize,
+    }
+    #[cfg(any(test, feature = "testing"))]
+    impl crate::event::snapshot::Fmt for PlatformRxSocketStats {
+        fn fmt(&self, fmt: &mut core::fmt::Formatter) -> core::fmt::Result {
+            let mut fmt = fmt.debug_struct("PlatformRxSocketStats");
+            fmt.field("is_prioritized", &self.is_prioritized);
+            fmt.field("count", &self.count);
+            fmt.finish()
+        }
+    }
+    impl Event for PlatformRxSocketStats {
+        const NAME: &'static str = "platform:rx_socket";
+    }
+    #[derive(Clone, Debug)]
+    #[non_exhaustive]
     pub struct PlatformEventLoopWakeup {
         pub timeout_expired: bool,
         pub rx_ready: bool,
@@ -4576,6 +4597,19 @@ pub mod tracing {
             tracing :: event ! (target : "platform_feature_configured" , parent : parent , tracing :: Level :: DEBUG , { configuration = tracing :: field :: debug (configuration) });
         }
         #[inline]
+        fn on_platform_rx_socket_stats(
+            &mut self,
+            meta: &api::EndpointMeta,
+            event: &api::PlatformRxSocketStats,
+        ) {
+            let parent = self.parent(meta);
+            let api::PlatformRxSocketStats {
+                is_prioritized,
+                count,
+            } = event;
+            tracing :: event ! (target : "platform_rx_socket_stats" , parent : parent , tracing :: Level :: DEBUG , { is_prioritized = tracing :: field :: debug (is_prioritized) , count = tracing :: field :: debug (count) });
+        }
+        #[inline]
         fn on_platform_event_loop_wakeup(
             &mut self,
             meta: &api::EndpointMeta,
@@ -7009,6 +7043,27 @@ pub mod builder {
         }
     }
     #[derive(Clone, Debug)]
+    #[doc = " Emitted for each receive socket with per-socket packet counts"]
+    pub struct PlatformRxSocketStats {
+        #[doc = " Whether this socket is the prioritized socket"]
+        pub is_prioritized: bool,
+        #[doc = " The number of packets received on this socket since the last event"]
+        pub count: usize,
+    }
+    impl IntoEvent<api::PlatformRxSocketStats> for PlatformRxSocketStats {
+        #[inline]
+        fn into_event(self) -> api::PlatformRxSocketStats {
+            let PlatformRxSocketStats {
+                is_prioritized,
+                count,
+            } = self;
+            api::PlatformRxSocketStats {
+                is_prioritized: is_prioritized.into_event(),
+                count: count.into_event(),
+            }
+        }
+    }
+    #[derive(Clone, Debug)]
     pub struct PlatformEventLoopWakeup {
         pub timeout_expired: bool,
         pub rx_ready: bool,
@@ -7955,6 +8010,16 @@ mod traits {
             let _ = meta;
             let _ = event;
         }
+        #[doc = "Called when the `PlatformRxSocketStats` event is triggered"]
+        #[inline]
+        fn on_platform_rx_socket_stats(
+            &mut self,
+            meta: &api::EndpointMeta,
+            event: &api::PlatformRxSocketStats,
+        ) {
+            let _ = meta;
+            let _ = event;
+        }
         #[doc = "Called when the `PlatformEventLoopWakeup` event is triggered"]
         #[inline]
         fn on_platform_event_loop_wakeup(
@@ -8676,6 +8741,15 @@ mod traits {
             (self.1).on_platform_feature_configured(meta, event);
         }
         #[inline]
+        fn on_platform_rx_socket_stats(
+            &mut self,
+            meta: &api::EndpointMeta,
+            event: &api::PlatformRxSocketStats,
+        ) {
+            (self.0).on_platform_rx_socket_stats(meta, event);
+            (self.1).on_platform_rx_socket_stats(meta, event);
+        }
+        #[inline]
         fn on_platform_event_loop_wakeup(
             &mut self,
             meta: &api::EndpointMeta,
@@ -8771,6 +8845,8 @@ mod traits {
         fn on_platform_rx_error(&mut self, event: builder::PlatformRxError);
         #[doc = "Publishes a `PlatformFeatureConfigured` event to the publisher's subscriber"]
         fn on_platform_feature_configured(&mut self, event: builder::PlatformFeatureConfigured);
+        #[doc = "Publishes a `PlatformRxSocketStats` event to the publisher's subscriber"]
+        fn on_platform_rx_socket_stats(&mut self, event: builder::PlatformRxSocketStats);
         #[doc = "Publishes a `PlatformEventLoopWakeup` event to the publisher's subscriber"]
         fn on_platform_event_loop_wakeup(&mut self, event: builder::PlatformEventLoopWakeup);
         #[doc = "Publishes a `PlatformEventLoopSleep` event to the publisher's subscriber"]
@@ -8900,6 +8976,13 @@ mod traits {
             self.subscriber.on_event(&self.meta, &event);
         }
         #[inline]
+        fn on_platform_rx_socket_stats(&mut self, event: builder::PlatformRxSocketStats) {
+            let event = event.into_event();
+            self.subscriber
+                .on_platform_rx_socket_stats(&self.meta, &event);
+            self.subscriber.on_event(&self.meta, &event);
+        }
+        #[inline]
         fn on_platform_event_loop_wakeup(&mut self, event: builder::PlatformEventLoopWakeup) {
             let event = event.into_event();
             self.subscriber
@@ -9558,6 +9641,7 @@ pub mod testing {
             pub platform_rx: u64,
             pub platform_rx_error: u64,
             pub platform_feature_configured: u64,
+            pub platform_rx_socket_stats: u64,
             pub platform_event_loop_wakeup: u64,
             pub platform_event_loop_sleep: u64,
             pub platform_event_loop_started: u64,
@@ -9605,6 +9689,7 @@ pub mod testing {
                     platform_rx: 0,
                     platform_rx_error: 0,
                     platform_feature_configured: 0,
+                    platform_rx_socket_stats: 0,
                     platform_event_loop_wakeup: 0,
                     platform_event_loop_sleep: 0,
                     platform_event_loop_started: 0,
@@ -9754,6 +9839,17 @@ pub mod testing {
                 let out = format!("{meta:?} {event:?}");
                 self.output.push(out);
             }
+            fn on_platform_rx_socket_stats(
+                &mut self,
+                meta: &api::EndpointMeta,
+                event: &api::PlatformRxSocketStats,
+            ) {
+                self.platform_rx_socket_stats += 1;
+                let meta = crate::event::snapshot::Fmt::to_snapshot(meta);
+                let event = crate::event::snapshot::Fmt::to_snapshot(event);
+                let out = format!("{meta:?} {event:?}");
+                self.output.push(out);
+            }
             fn on_platform_event_loop_wakeup(
                 &mut self,
                 meta: &api::EndpointMeta,
@@ -9855,6 +9951,7 @@ pub mod testing {
         pub platform_rx: u64,
         pub platform_rx_error: u64,
         pub platform_feature_configured: u64,
+        pub platform_rx_socket_stats: u64,
         pub platform_event_loop_wakeup: u64,
         pub platform_event_loop_sleep: u64,
         pub platform_event_loop_started: u64,
@@ -9951,6 +10048,7 @@ pub mod testing {
                 platform_rx: 0,
                 platform_rx_error: 0,
                 platform_feature_configured: 0,
+                platform_rx_socket_stats: 0,
                 platform_event_loop_wakeup: 0,
                 platform_event_loop_sleep: 0,
                 platform_event_loop_started: 0,
@@ -10779,6 +10877,17 @@ pub mod testing {
             let out = format!("{meta:?} {event:?}");
             self.output.push(out);
         }
+        fn on_platform_rx_socket_stats(
+            &mut self,
+            meta: &api::EndpointMeta,
+            event: &api::PlatformRxSocketStats,
+        ) {
+            self.platform_rx_socket_stats += 1;
+            let meta = crate::event::snapshot::Fmt::to_snapshot(meta);
+            let event = crate::event::snapshot::Fmt::to_snapshot(event);
+            let out = format!("{meta:?} {event:?}");
+            self.output.push(out);
+        }
         fn on_platform_event_loop_wakeup(
             &mut self,
             meta: &api::EndpointMeta,
@@ -10879,6 +10988,7 @@ pub mod testing {
         pub platform_rx: u64,
         pub platform_rx_error: u64,
         pub platform_feature_configured: u64,
+        pub platform_rx_socket_stats: u64,
         pub platform_event_loop_wakeup: u64,
         pub platform_event_loop_sleep: u64,
         pub platform_event_loop_started: u64,
@@ -10965,6 +11075,7 @@ pub mod testing {
                 platform_rx: 0,
                 platform_rx_error: 0,
                 platform_feature_configured: 0,
+                platform_rx_socket_stats: 0,
                 platform_event_loop_wakeup: 0,
                 platform_event_loop_sleep: 0,
                 platform_event_loop_started: 0,
@@ -11069,6 +11180,13 @@ pub mod testing {
             let out = format!("{event:?}");
             self.output.push(out);
         }
+        fn on_platform_rx_socket_stats(&mut self, event: builder::PlatformRxSocketStats) {
+            self.platform_rx_socket_stats += 1;
+            let event = event.into_event();
+            let event = crate::event::snapshot::Fmt::to_snapshot(&event);
+            let out = format!("{event:?}");
+            self.output.push(out);
+        }
         fn on_platform_event_loop_wakeup(&mut self, event: builder::PlatformEventLoopWakeup) {
             self.platform_event_loop_wakeup += 1;
             let event = event.into_event();

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

@@ -185,6 +185,8 @@ mod id {
         PLATFORM_RX__ERRORS__DROPPED,
         PLATFORM_RX_ERROR,
         PLATFORM_FEATURE_CONFIGURED,
+        PLATFORM_RX_SOCKET_STATS,
+        PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL,
         PLATFORM_EVENT_LOOP_WAKEUP,
         PLATFORM_EVENT_LOOP_SLEEP,
         PLATFORM_EVENT_LOOP_SLEEP__PROCESSING_DURATION,
@@ -409,6 +411,9 @@ mod id {
     pub const PLATFORM_RX__ERRORS__DROPPED: usize = InfoId::PLATFORM_RX__ERRORS__DROPPED as usize;
     pub const PLATFORM_RX_ERROR: usize = InfoId::PLATFORM_RX_ERROR as usize;
     pub const PLATFORM_FEATURE_CONFIGURED: usize = InfoId::PLATFORM_FEATURE_CONFIGURED as usize;
+    pub const PLATFORM_RX_SOCKET_STATS: usize = InfoId::PLATFORM_RX_SOCKET_STATS as usize;
+    pub const PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL: usize =
+        InfoId::PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL as usize;
     pub const PLATFORM_EVENT_LOOP_WAKEUP: usize = InfoId::PLATFORM_EVENT_LOOP_WAKEUP as usize;
     pub const PLATFORM_EVENT_LOOP_SLEEP: usize = InfoId::PLATFORM_EVENT_LOOP_SLEEP as usize;
     pub const PLATFORM_EVENT_LOOP_SLEEP__PROCESSING_DURATION: usize =
@@ -496,6 +501,8 @@ mod id {
         COUNTERS_PLATFORM_RX__ERRORS__DROPPED__TOTAL,
         COUNTERS_PLATFORM_RX_ERROR,
         COUNTERS_PLATFORM_FEATURE_CONFIGURED,
+        COUNTERS_PLATFORM_RX_SOCKET_STATS,
+        COUNTERS_PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL,
         COUNTERS_PLATFORM_EVENT_LOOP_WAKEUP,
         COUNTERS_PLATFORM_EVENT_LOOP_SLEEP,
         COUNTERS_PLATFORM_EVENT_LOOP_STARTED,
@@ -619,6 +626,10 @@ mod id {
     pub const COUNTERS_PLATFORM_RX_ERROR: usize = Counters::COUNTERS_PLATFORM_RX_ERROR as usize;
     pub const COUNTERS_PLATFORM_FEATURE_CONFIGURED: usize =
         Counters::COUNTERS_PLATFORM_FEATURE_CONFIGURED as usize;
+    pub const COUNTERS_PLATFORM_RX_SOCKET_STATS: usize =
+        Counters::COUNTERS_PLATFORM_RX_SOCKET_STATS as usize;
+    pub const COUNTERS_PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL: usize =
+        Counters::COUNTERS_PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL as usize;
     pub const COUNTERS_PLATFORM_EVENT_LOOP_WAKEUP: usize =
         Counters::COUNTERS_PLATFORM_EVENT_LOOP_WAKEUP as usize;
     pub const COUNTERS_PLATFORM_EVENT_LOOP_SLEEP: usize =
@@ -911,7 +922,7 @@ mod id {
     pub const NOMINAL_TIMERS_SLOW_START_EXITED__LATENCY: usize =
         NominalTimers::NOMINAL_TIMERS_SLOW_START_EXITED__LATENCY as usize;
 }
-static INFO: &[Info; 172usize] = &[
+static INFO: &[Info; 174usize] = &[
     info::Builder {
         id: id::APPLICATION_PROTOCOL_INFORMATION,
         name: Str::new("application_protocol_information\0"),
@@ -1920,6 +1931,18 @@ static INFO: &[Info; 172usize] = &[
         units: Units::None,
     }
     .build(),
+    info::Builder {
+        id: id::PLATFORM_RX_SOCKET_STATS,
+        name: Str::new("platform_rx_socket_stats\0"),
+        units: Units::None,
+    }
+    .build(),
+    info::Builder {
+        id: id::PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL,
+        name: Str::new("platform_rx_socket_stats.packets.total\0"),
+        units: Units::None,
+    }
+    .build(),
     info::Builder {
         id: id::PLATFORM_EVENT_LOOP_WAKEUP,
         name: Str::new("platform_event_loop_wakeup\0"),
@@ -1952,7 +1975,7 @@ pub struct ConnectionContext {
 }
 pub struct Subscriber<R: Registry> {
     #[allow(dead_code)]
-    counters: Box<[R::Counter; 82usize]>,
+    counters: Box<[R::Counter; 84usize]>,
     #[allow(dead_code)]
     bool_counters: Box<[R::BoolCounter; 3usize]>,
     #[allow(dead_code)]
@@ -1987,7 +2010,7 @@ impl<R: Registry> Subscriber<R> {
     #[allow(unused_mut)]
     #[inline]
     pub fn new(registry: R) -> Self {
-        let mut counters = Vec::with_capacity(82usize);
+        let mut counters = Vec::with_capacity(84usize);
         let mut bool_counters = Vec::with_capacity(3usize);
         let mut nominal_counters = Vec::with_capacity(31usize);
         let mut nominal_counter_offsets = Vec::with_capacity(31usize);
@@ -2077,6 +2100,9 @@ impl<R: Registry> Subscriber<R> {
         counters.push(registry.register_counter(&INFO[id::PLATFORM_RX__ERRORS__DROPPED__TOTAL]));
         counters.push(registry.register_counter(&INFO[id::PLATFORM_RX_ERROR]));
         counters.push(registry.register_counter(&INFO[id::PLATFORM_FEATURE_CONFIGURED]));
+        counters.push(registry.register_counter(&INFO[id::PLATFORM_RX_SOCKET_STATS]));
+        counters
+            .push(registry.register_counter(&INFO[id::PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL]));
         counters.push(registry.register_counter(&INFO[id::PLATFORM_EVENT_LOOP_WAKEUP]));
         counters.push(registry.register_counter(&INFO[id::PLATFORM_EVENT_LOOP_SLEEP]));
         counters.push(registry.register_counter(&INFO[id::PLATFORM_EVENT_LOOP_STARTED]));
@@ -2746,6 +2772,12 @@ impl<R: Registry> Subscriber<R> {
                 id::COUNTERS_PLATFORM_FEATURE_CONFIGURED => {
                     (&INFO[id::PLATFORM_FEATURE_CONFIGURED], entry)
                 }
+                id::COUNTERS_PLATFORM_RX_SOCKET_STATS => {
+                    (&INFO[id::PLATFORM_RX_SOCKET_STATS], entry)
+                }
+                id::COUNTERS_PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL => {
+                    (&INFO[id::PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL], entry)
+                }
                 id::COUNTERS_PLATFORM_EVENT_LOOP_WAKEUP => {
                     (&INFO[id::PLATFORM_EVENT_LOOP_WAKEUP], entry)
                 }
@@ -4817,6 +4849,27 @@ impl<R: Registry> event::Subscriber for Subscriber<R> {
         let _ = meta;
     }
     #[inline]
+    fn on_platform_rx_socket_stats(
+        &mut self,
+        meta: &api::EndpointMeta,
+        event: &api::PlatformRxSocketStats,
+    ) {
+        #[allow(unused_imports)]
+        use api::*;
+        self.count(
+            id::PLATFORM_RX_SOCKET_STATS,
+            id::COUNTERS_PLATFORM_RX_SOCKET_STATS,
+            1usize,
+        );
+        self.count(
+            id::PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL,
+            id::COUNTERS_PLATFORM_RX_SOCKET_STATS__PACKETS__TOTAL,
+            event.count,
+        );
+        let _ = event;
+        let _ = meta;
+    }
+    #[inline]
     fn on_platform_event_loop_wakeup(
         &mut self,
         meta: &api::EndpointMeta,