Communicate endpoint events in -proto w/ an intrusive queue

.github/workflows/rust.yml

@@ -58,7 +58,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
-      - uses: dtolnay/rust-toolchain@1.63.0
+      - uses: dtolnay/rust-toolchain@1.67.0
       - uses: Swatinem/rust-cache@v2
       - run: cargo check --lib --all-features -p quinn-udp -p quinn-proto -p quinn
 

bench/Cargo.toml

@@ -2,7 +2,7 @@
 name = "bench"
 version = "0.1.0"
 edition = "2021"
-rust-version = "1.63"
+rust-version = "1.67"
 license = "MIT OR Apache-2.0"
 publish = false
 

perf/Cargo.toml

@@ -2,7 +2,7 @@
 name = "perf"
 version = "0.1.0"
 edition = "2021"
-rust-version = "1.63"
+rust-version = "1.67"
 license = "MIT OR Apache-2.0"
 publish = false
 

quinn-proto/Cargo.toml

@@ -2,7 +2,7 @@
 name = "quinn-proto"
 version = "0.11.0"
 edition = "2021"
-rust-version = "1.63"
+rust-version = "1.67"
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/quinn-rs/quinn"
 description = "State machine for the QUIC transport protocol"

quinn-proto/src/connection/mod.rs

@@ -4,7 +4,7 @@ use std::{
     convert::TryFrom,
     fmt, io, mem,
     net::{IpAddr, SocketAddr},
-    sync::Arc,
+    sync::{atomic::Ordering, Arc},
     time::{Duration, Instant},
 };
 
@@ -25,13 +25,14 @@ use crate::{
     packet::{Header, LongType, Packet, PartialDecode, SpaceId},
     range_set::ArrayRangeSet,
     shared::{
-        ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent,
-        EndpointEventInner,
+        ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvents,
+        EndpointEventsQueue,
     },
     token::ResetToken,
     transport_parameters::TransportParameters,
-    Dir, EndpointConfig, Frame, Side, StreamId, Transmit, TransportError, TransportErrorCode,
-    VarInt, MAX_STREAM_COUNT, MIN_INITIAL_SIZE, TIMER_GRANULARITY,
+    ConnectionHandle, Dir, EndpointConfig, Frame, SharedList, Side, StreamId, Transmit,
+    TransportError, TransportErrorCode, VarInt, MAX_STREAM_COUNT, MIN_INITIAL_SIZE,
+    TIMER_GRANULARITY,
 };
 
 mod ack_frequency;
@@ -89,10 +90,10 @@ use timer::{Timer, TimerTable};
 
 /// Protocol state and logic for a single QUIC connection
 ///
-/// Objects of this type receive [`ConnectionEvent`]s and emit [`EndpointEvent`]s and application
+/// Objects of this type receive [`ConnectionEvent`]s and emit endpoint wakeups and application
 /// [`Event`]s to make progress. To handle timeouts, a `Connection` returns timer updates and
-/// expects timeouts through various methods. A number of simple getter methods are exposed
-/// to allow callers to inspect some of the connection state.
+/// expects timeouts through various methods. A number of simple getter methods are exposed to allow
+/// callers to inspect some of the connection state.
 ///
 /// `Connection` has roughly 4 types of methods:
 ///
@@ -130,6 +131,7 @@ pub struct Connection {
     endpoint_config: Arc<EndpointConfig>,
     server_config: Option<Arc<ServerConfig>>,
     config: Arc<TransportConfig>,
+    endpoint_events: EndpointEventsTracker,
     rng: StdRng,
     crypto: Box<dyn crypto::Session>,
     /// The CID we initially chose, for use during the handshake
@@ -162,7 +164,6 @@ pub struct Connection {
     /// Total number of outgoing packets that have been deemed lost
     lost_packets: u64,
     events: VecDeque<Event>,
-    endpoint_events: VecDeque<EndpointEventInner>,
     /// Whether the spin bit is in use for this connection
     spin_enabled: bool,
     /// Outgoing spin bit state
@@ -242,6 +243,7 @@ impl Connection {
         endpoint_config: Arc<EndpointConfig>,
         server_config: Option<Arc<ServerConfig>>,
         config: Arc<TransportConfig>,
+        endpoint_events: EndpointEventsTracker,
         init_cid: ConnectionId,
         loc_cid: ConnectionId,
         rem_cid: ConnectionId,
@@ -272,6 +274,7 @@ impl Connection {
         let mut this = Self {
             endpoint_config,
             server_config,
+            endpoint_events,
             crypto,
             handshake_cid: loc_cid,
             rem_handshake_cid: rem_cid,
@@ -313,7 +316,6 @@ impl Connection {
             retry_src_cid: None,
             lost_packets: 0,
             events: VecDeque::new(),
-            endpoint_events: VecDeque::new(),
             spin_enabled: config.allow_spin && rng.gen_ratio(7, 8),
             spin: false,
             spaces: [initial_space, PacketSpace::new(now), PacketSpace::new(now)],
@@ -406,10 +408,10 @@ impl Connection {
         None
     }
 
-    /// Return endpoint-facing events
+    /// Whether the endpoint must be woken to process events
     #[must_use]
-    pub fn poll_endpoint_events(&mut self) -> Option<EndpointEvent> {
-        self.endpoint_events.pop_front().map(EndpointEvent)
+    pub fn poll_endpoint_events(&mut self) -> bool {
+        mem::replace(&mut self.endpoint_events.notify_needed, false)
     }
 
     /// Provide control over streams
@@ -966,7 +968,7 @@ impl Connection {
     /// Will execute protocol logic upon receipt of a connection event, in turn preparing signals
     /// (including application `Event`s, `EndpointEvent`s and outgoing datagrams) that should be
     /// extracted through the relevant methods.
-    pub fn handle_event(&mut self, event: ConnectionEvent) {
+    pub fn handle_event(&mut self, event: ConnectionEvent, now: Instant) {
         use self::ConnectionEventInner::*;
         match event.0 {
             Datagram {
@@ -1011,19 +1013,13 @@ impl Connection {
                     self.set_loss_detection_timer(now);
                 }
             }
-            NewIdentifiers(ids, now) => {
+            NewIdentifiers(ids) => {
                 self.local_cid_state.new_cids(&ids, now);
                 ids.into_iter().rev().for_each(|frame| {
                     self.spaces[SpaceId::Data].pending.new_cids.push(frame);
                 });
                 // Update Timer::PushNewCid
-                if self
-                    .timers
-                    .get(Timer::PushNewCid)
-                    .map_or(true, |x| x <= now)
-                {
-                    self.reset_cid_retirement();
-                }
+                self.reset_cid_retirement();
             }
         }
     }
@@ -1047,7 +1043,10 @@ impl Connection {
             match timer {
                 Timer::Close => {
                     self.state = State::Drained;
-                    self.endpoint_events.push_back(EndpointEventInner::Drained);
+                    self.endpoint_events
+                        .get()
+                        .drained
+                        .store(true, Ordering::Relaxed);
                 }
                 Timer::Idle => {
                     self.kill(ConnectionError::TimedOut);
@@ -1081,7 +1080,9 @@ impl Connection {
                             self.local_cid_state.retire_prior_to()
                         );
                         self.endpoint_events
-                            .push_back(EndpointEventInner::NeedIdentifiers(now, num_new_cid));
+                            .get()
+                            .need_identifiers
+                            .fetch_add(num_new_cid, Ordering::Relaxed);
                     }
                 }
                 Timer::MaxAckDelay => {
@@ -2199,7 +2200,10 @@ impl Connection {
             }
         }
         if !was_drained && self.state.is_drained() {
-            self.endpoint_events.push_back(EndpointEventInner::Drained);
+            self.endpoint_events
+                .get()
+                .drained
+                .store(true, Ordering::Relaxed);
             // Close timer may have been started previously, e.g. if we sent a close and got a
             // stateless reset in response
             self.timers.stop(Timer::Close);
@@ -2375,11 +2379,11 @@ impl Connection {
                         }
                     }
                     if let Some(token) = params.stateless_reset_token {
-                        self.endpoint_events
-                            .push_back(EndpointEventInner::ResetToken(self.path.remote, token));
+                        *self.endpoint_events.get().reset_token.lock().unwrap() =
+                            Some((self.path.remote, token));
                     }
                     self.handle_peer_params(params)?;
-                    self.issue_first_cids(now);
+                    self.issue_first_cids();
                 } else {
                     // Server-only
                     self.spaces[SpaceId::Data].pending.handshake_done = true;
@@ -2426,7 +2430,7 @@ impl Connection {
                                 reason: "transport parameters missing".into(),
                             })?;
                     self.handle_peer_params(params)?;
-                    self.issue_first_cids(now);
+                    self.issue_first_cids();
                     self.init_0rtt();
                 }
                 Ok(())
@@ -2693,12 +2697,18 @@ impl Connection {
                     let allow_more_cids = self
                         .local_cid_state
                         .on_cid_retirement(sequence, self.peer_params.issue_cids_limit())?;
+                    if allow_more_cids {
+                        self.endpoint_events
+                            .get()
+                            .need_identifiers
+                            .fetch_add(1, Ordering::Relaxed);
+                    }
                     self.endpoint_events
-                        .push_back(EndpointEventInner::RetireConnectionId(
-                            now,
-                            sequence,
-                            allow_more_cids,
-                        ));
+                        .get()
+                        .retire_cids
+                        .lock()
+                        .unwrap()
+                        .push(sequence);
                 }
                 Frame::NewConnectionId(frame) => {
                     trace!(
@@ -2913,24 +2923,23 @@ impl Connection {
     }
 
     fn set_reset_token(&mut self, reset_token: ResetToken) {
-        self.endpoint_events
-            .push_back(EndpointEventInner::ResetToken(
-                self.path.remote,
-                reset_token,
-            ));
+        *self.endpoint_events.get().reset_token.lock().unwrap() =
+            Some((self.path.remote, reset_token));
         self.peer_params.stateless_reset_token = Some(reset_token);
     }
 
     /// Issue an initial set of connection IDs to the peer upon connection
-    fn issue_first_cids(&mut self, now: Instant) {
+    fn issue_first_cids(&mut self) {
         if self.local_cid_state.cid_len() == 0 {
             return;
         }
 
         // Subtract 1 to account for the CID we supplied while handshaking
         let n = self.peer_params.issue_cids_limit() - 1;
         self.endpoint_events
-            .push_back(EndpointEventInner::NeedIdentifiers(now, n));
+            .get()
+            .need_identifiers
+            .fetch_add(n, Ordering::Relaxed);
     }
 
     fn populate_packet(
@@ -3407,10 +3416,12 @@ impl Connection {
     /// Instruct the peer to replace previously issued CIDs by sending a NEW_CONNECTION_ID frame
     /// with updated `retire_prior_to` field set to `v`
     #[cfg(test)]
-    pub(crate) fn rotate_local_cid(&mut self, v: u64, now: Instant) {
+    pub(crate) fn rotate_local_cid(&mut self, v: u64) {
         let n = self.local_cid_state.assign_retire_seq(v);
         self.endpoint_events
-            .push_back(EndpointEventInner::NeedIdentifiers(now, n));
+            .get()
+            .need_identifiers
+            .fetch_add(n, Ordering::Relaxed);
     }
 
     /// Check the current active remote CID sequence
@@ -3451,7 +3462,10 @@ impl Connection {
         self.close_common();
         self.error = Some(reason);
         self.state = State::Drained;
-        self.endpoint_events.push_back(EndpointEventInner::Drained);
+        self.endpoint_events
+            .get()
+            .drained
+            .store(true, Ordering::Relaxed);
     }
 }
 
@@ -3679,3 +3693,43 @@ impl SentFrames {
             && self.retransmits.is_empty(streams)
     }
 }
+
+/// Helper struct ensuring endpoint events get queued and generate notifications
+pub(super) struct EndpointEventsTracker {
+    events: Arc<EndpointEvents>,
+    queue: Arc<SharedList<EndpointEvents, EndpointEventsQueue>>,
+    notify_needed: bool,
+}
+
+impl EndpointEventsTracker {
+    pub(super) fn new(
+        ch: ConnectionHandle,
+        queue: Arc<SharedList<EndpointEvents, EndpointEventsQueue>>,
+    ) -> Self {
+        Self {
+            events: Arc::new(EndpointEvents::new(ch)),
+            queue,
+            notify_needed: false,
+        }
+    }
+
+    fn get(&mut self) -> EndpointEventsGuard<'_> {
+        EndpointEventsGuard(self)
+    }
+}
+
+struct EndpointEventsGuard<'a>(&'a mut EndpointEventsTracker);
+
+impl std::ops::Deref for EndpointEventsGuard<'_> {
+    type Target = EndpointEvents;
+
+    fn deref(&self) -> &EndpointEvents {
+        &self.0.events
+    }
+}
+
+impl Drop for EndpointEventsGuard<'_> {
+    fn drop(&mut self) {
+        self.0.notify_needed |= self.0.queue.push(self.0.events.clone());
+    }
+}