Pass proto-layer endpoint events internally

Groundwork for replacing generic message-passing with lighter
event-representation-aware communication strategies.

Author: Ralith

quinn-proto/src/connection/mod.rs

@@ -4,7 +4,7 @@ use std::{
     convert::TryFrom,
     fmt, io, mem,
     net::{IpAddr, SocketAddr},
-    sync::Arc,
+    sync::{mpsc, Arc},
     time::{Duration, Instant},
 };
 
@@ -24,14 +24,11 @@ use crate::{
     frame::{Close, Datagram, FrameStruct},
     packet::{Header, LongType, Packet, PartialDecode, SpaceId},
     range_set::ArrayRangeSet,
-    shared::{
-        ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent,
-        EndpointEventInner,
-    },
+    shared::{ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent},
     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, Side, StreamId, Transmit, TransportError,
+    TransportErrorCode, VarInt, MAX_STREAM_COUNT, MIN_INITIAL_SIZE, TIMER_GRANULARITY,
 };
 
 mod ack_frequency;
@@ -162,7 +159,7 @@ pub struct Connection {
     /// Total number of outgoing packets that have been deemed lost
     lost_packets: u64,
     events: VecDeque<Event>,
-    endpoint_events: VecDeque<EndpointEventInner>,
+    endpoint_events: EndpointEvents,
     /// Whether the spin bit is in use for this connection
     spin_enabled: bool,
     /// Outgoing spin bit state
@@ -253,6 +250,7 @@ impl Connection {
         version: u32,
         allow_mtud: bool,
         rng_seed: [u8; 32],
+        endpoint_events: EndpointEvents,
     ) -> Self {
         let side = if server_config.is_some() {
             Side::Server
@@ -314,7 +312,7 @@ impl Connection {
             retry_src_cid: None,
             lost_packets: 0,
             events: VecDeque::new(),
-            endpoint_events: VecDeque::new(),
+            endpoint_events,
             spin_enabled: config.allow_spin && rng.gen_ratio(7, 8),
             spin: false,
             spaces: [initial_space, PacketSpace::new(now), PacketSpace::new(now)],
@@ -407,10 +405,10 @@ impl Connection {
         None
     }
 
-    /// Return endpoint-facing events
+    /// Whether [`Endpoint::handle_events`] must be called in the immediate future
     #[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::take(&mut self.endpoint_events.dirty)
     }
 
     /// Provide control over streams
@@ -954,8 +952,8 @@ impl Connection {
     /// Process `ConnectionEvent`s generated by the associated `Endpoint`
     ///
     /// 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.
+    /// (including application `Event`s, endpoint events, and outgoing datagrams) that should be
+    /// checked through the relevant methods.
     pub fn handle_event(&mut self, event: ConnectionEvent, now: Instant) {
         use self::ConnectionEventInner::*;
         match event.0 {
@@ -1037,7 +1035,7 @@ impl Connection {
             match timer {
                 Timer::Close => {
                     self.state = State::Drained;
-                    self.endpoint_events.push_back(EndpointEventInner::Drained);
+                    self.endpoint_events.push(EndpointEvent::Drained);
                 }
                 Timer::Idle => {
                     self.kill(ConnectionError::TimedOut);
@@ -1071,7 +1069,7 @@ impl Connection {
                             self.local_cid_state.retire_prior_to()
                         );
                         self.endpoint_events
-                            .push_back(EndpointEventInner::NeedIdentifiers(num_new_cid));
+                            .push(EndpointEvent::NeedIdentifiers(num_new_cid));
                     }
                 }
                 Timer::MaxAckDelay => {
@@ -2168,7 +2166,7 @@ impl Connection {
             }
         }
         if !was_drained && self.state.is_drained() {
-            self.endpoint_events.push_back(EndpointEventInner::Drained);
+            self.endpoint_events.push(EndpointEvent::Drained);
             // 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);
@@ -2351,7 +2349,7 @@ impl Connection {
                     }
                     if let Some(token) = params.stateless_reset_token {
                         self.endpoint_events
-                            .push_back(EndpointEventInner::ResetToken(self.path.remote, token));
+                            .push(EndpointEvent::ResetToken(self.path.remote, token));
                     }
                     self.handle_peer_params(params)?;
                     self.issue_first_cids();
@@ -2661,10 +2659,7 @@ impl Connection {
                         .local_cid_state
                         .on_cid_retirement(sequence, self.peer_params.issue_cids_limit())?;
                     self.endpoint_events
-                        .push_back(EndpointEventInner::RetireConnectionId(
-                            sequence,
-                            allow_more_cids,
-                        ));
+                        .push(EndpointEvent::RetireConnectionId(sequence, allow_more_cids));
                 }
                 Frame::NewConnectionId(frame) => {
                     trace!(
@@ -2880,10 +2875,7 @@ impl Connection {
 
     fn set_reset_token(&mut self, reset_token: ResetToken) {
         self.endpoint_events
-            .push_back(EndpointEventInner::ResetToken(
-                self.path.remote,
-                reset_token,
-            ));
+            .push(EndpointEvent::ResetToken(self.path.remote, reset_token));
         self.peer_params.stateless_reset_token = Some(reset_token);
     }
 
@@ -2895,8 +2887,7 @@ impl Connection {
 
         // 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(n));
+        self.endpoint_events.push(EndpointEvent::NeedIdentifiers(n));
     }
 
     fn populate_packet(
@@ -3373,8 +3364,7 @@ impl Connection {
     #[cfg(test)]
     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(n));
+        self.endpoint_events.push(EndpointEvent::NeedIdentifiers(n));
     }
 
     /// Check the current active remote CID sequence
@@ -3415,7 +3405,7 @@ impl Connection {
         self.close_common();
         self.error = Some(reason);
         self.state = State::Drained;
-        self.endpoint_events.push_back(EndpointEventInner::Drained);
+        self.endpoint_events.push(EndpointEvent::Drained);
     }
 
     /// Storage size required for the largest packet known to be supported by the current path
@@ -3650,3 +3640,28 @@ impl SentFrames {
             && self.retransmits.is_empty(streams)
     }
 }
+
+pub(crate) struct EndpointEvents {
+    ch: ConnectionHandle,
+    sender: mpsc::Sender<(ConnectionHandle, EndpointEvent)>,
+    dirty: bool,
+}
+
+impl EndpointEvents {
+    pub(crate) fn new(
+        ch: ConnectionHandle,
+        sender: mpsc::Sender<(ConnectionHandle, EndpointEvent)>,
+    ) -> Self {
+        Self {
+            ch,
+            sender,
+            dirty: false,
+        }
+    }
+
+    fn push(&mut self, event: EndpointEvent) {
+        // If the endpoint has gone away, assume the caller is winding down regardless.
+        _ = self.sender.send((self.ch, event));
+        self.dirty = true;
+    }
+}

quinn-proto/src/endpoint.rs

@@ -4,7 +4,7 @@ use std::{
     fmt, iter,
     net::{IpAddr, SocketAddr},
     ops::{Index, IndexMut},
-    sync::Arc,
+    sync::{mpsc, Arc},
     time::{Instant, SystemTime},
 };
 
@@ -19,13 +19,12 @@ use crate::{
     cid_generator::{ConnectionIdGenerator, RandomConnectionIdGenerator},
     coding::BufMutExt,
     config::{ClientConfig, EndpointConfig, ServerConfig},
-    connection::{Connection, ConnectionError},
+    connection::{Connection, ConnectionError, EndpointEvents},
     crypto::{self, Keys, UnsupportedVersion},
     frame,
     packet::{Header, Packet, PacketDecodeError, PacketNumber, PartialDecode},
     shared::{
-        ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent,
-        EndpointEventInner, IssuedCid,
+        ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent, IssuedCid,
     },
     transport_parameters::TransportParameters,
     ResetToken, RetryToken, Side, Transmit, TransportConfig, TransportError, INITIAL_MTU,
@@ -45,6 +44,8 @@ pub struct Endpoint {
     server_config: Option<Arc<ServerConfig>>,
     /// Whether the underlying UDP socket promises not to fragment packets
     allow_mtud: bool,
+    event_send: mpsc::Sender<(ConnectionHandle, EndpointEvent)>,
+    event_recv: mpsc::Receiver<(ConnectionHandle, EndpointEvent)>,
 }
 
 impl Endpoint {
@@ -59,6 +60,7 @@ impl Endpoint {
         allow_mtud: bool,
         rng_seed: Option<[u8; 32]>,
     ) -> Self {
+        let (event_send, event_recv) = mpsc::channel();
         Self {
             rng: rng_seed.map_or(StdRng::from_entropy(), StdRng::from_seed),
             index: ConnectionIndex::default(),
@@ -67,6 +69,8 @@ impl Endpoint {
             config,
             server_config,
             allow_mtud,
+            event_send,
+            event_recv,
         }
     }
 
@@ -75,16 +79,25 @@ impl Endpoint {
         self.server_config = server_config;
     }
 
-    /// Process `EndpointEvent`s emitted from related `Connection`s
+    /// Process events from [`Connection`]s that have returned `true` from [`Connection::poll_endpoint_events`]
     ///
-    /// In turn, processing this event may return a `ConnectionEvent` for the same `Connection`.
-    pub fn handle_event(
+    /// May return a `ConnectionEvent` for any `Connection`. Call until `None` is returned.
+    pub fn handle_events(&mut self) -> Option<(ConnectionHandle, ConnectionEvent)> {
+        while let Ok((ch, event)) = self.event_recv.try_recv() {
+            if let Some(response) = self.handle_event(ch, event) {
+                return Some((ch, response));
+            }
+        }
+        None
+    }
+
+    fn handle_event(
         &mut self,
         ch: ConnectionHandle,
         event: EndpointEvent,
     ) -> Option<ConnectionEvent> {
-        use EndpointEventInner::*;
-        match event.0 {
+        use EndpointEvent::*;
+        match event {
             NeedIdentifiers(n) => {
                 return Some(self.send_new_identifiers(ch, n));
             }
@@ -564,7 +577,7 @@ impl Endpoint {
             }
             Err(e) => {
                 debug!("handshake failed: {}", e);
-                self.handle_event(ch, EndpointEvent(EndpointEventInner::Drained));
+                self.handle_event(ch, EndpointEvent::Drained);
                 match e {
                     ConnectionError::TransportError(e) => Some(DatagramEvent::Response(
                         self.initial_close(version, addresses, crypto, &src_cid, e, buf),
@@ -605,6 +618,7 @@ impl Endpoint {
             version,
             self.allow_mtud,
             rng_seed,
+            EndpointEvents::new(ch, self.event_send.clone()),
         );
 
         let id = self.connections.insert(ConnectionMeta {

quinn-proto/src/lib.rs

@@ -64,7 +64,7 @@ mod endpoint;
 pub use crate::endpoint::{ConnectError, ConnectionHandle, DatagramEvent, Endpoint};
 
 mod shared;
-pub use crate::shared::{ConnectionEvent, ConnectionId, EcnCodepoint, EndpointEvent};
+pub use crate::shared::{ConnectionEvent, ConnectionId, EcnCodepoint};
 
 mod transport_error;
 pub use crate::transport_error::{Code as TransportErrorCode, Error as TransportError};

quinn-proto/src/shared.rs

@@ -22,29 +22,8 @@ pub(crate) enum ConnectionEventInner {
     NewIdentifiers(Vec<IssuedCid>),
 }
 
-/// Events sent from a Connection to an Endpoint
-#[derive(Debug)]
-pub struct EndpointEvent(pub(crate) EndpointEventInner);
-
-impl EndpointEvent {
-    /// Construct an event that indicating that a `Connection` will no longer emit events
-    ///
-    /// Useful for notifying an `Endpoint` that a `Connection` has been destroyed outside of the
-    /// usual state machine flow, e.g. when being dropped by the user.
-    pub fn drained() -> Self {
-        Self(EndpointEventInner::Drained)
-    }
-
-    /// Determine whether this is the last event a `Connection` will emit
-    ///
-    /// Useful for determining when connection-related event loop state can be freed.
-    pub fn is_drained(&self) -> bool {
-        self.0 == EndpointEventInner::Drained
-    }
-}
-
 #[derive(Clone, Debug, Eq, PartialEq)]
-pub(crate) enum EndpointEventInner {
+pub(crate) enum EndpointEvent {
     /// The connection has been drained
     Drained,
     /// The reset token and/or address eligible for generating resets has been updated

quinn-proto/src/tests/util.rs

@@ -362,8 +362,8 @@ impl TestEndpoint {
         }
 
         loop {
-            let mut endpoint_events: Vec<(ConnectionHandle, EndpointEvent)> = vec![];
-            for (ch, conn) in self.connections.iter_mut() {
+            let mut endpoint_events = false;
+            for (_, conn) in self.connections.iter_mut() {
                 if self.timeout.map_or(false, |x| x <= now) {
                     self.timeout = None;
                     conn.handle_timeout(now);
@@ -375,9 +375,7 @@ impl TestEndpoint {
                     }
                 }
 
-                while let Some(event) = conn.poll_endpoint_events() {
-                    endpoint_events.push((*ch, event));
-                }
+                endpoint_events |= conn.poll_endpoint_events();
                 while let Some(transmit) = conn.poll_transmit(now, MAX_DATAGRAMS, &mut buf) {
                     let size = transmit.size;
                     self.outbound
@@ -386,15 +384,13 @@ impl TestEndpoint {
                 self.timeout = conn.poll_timeout();
             }
 
-            if endpoint_events.is_empty() {
+            if !endpoint_events {
                 break;
             }
 
-            for (ch, event) in endpoint_events {
-                if let Some(event) = self.handle_event(ch, event) {
-                    if let Some(conn) = self.connections.get_mut(&ch) {
-                        conn.handle_event(event, now);
-                    }
+            while let Some((ch, event)) = self.handle_events() {
+                if let Some(conn) = self.connections.get_mut(&ch) {
+                    conn.handle_event(event, now);
                 }
             }
         }