feat: switch to noq

Author: Sherlock-Holo

compio-quic/Cargo.toml

@@ -21,7 +21,7 @@ compio-log = { workspace = true }
 compio-net = { workspace = true }
 compio-runtime = { workspace = true, features = ["time"] }
 
-quinn-proto = { version = "0.11.13", default-features = false }
+noq-proto = { version = "0.16.0", default-features = false }
 rustls = { workspace = true }
 rustls-platform-verifier = { workspace = true, optional = true }
 rustls-native-certs = { workspace = true, optional = true }
@@ -55,7 +55,7 @@ compio-runtime = { workspace = true, features = ["criterion"] }
 
 criterion = { workspace = true, features = ["async_tokio"] }
 http = "1.1.0"
-quinn = "0.11.9"
+noq = "0.17.0"
 rand = { workspace = true }
 rcgen = "0.14.1"
 socket2 = { workspace = true, features = ["all"] }
@@ -68,9 +68,9 @@ io-compat = ["futures-util/io"]
 platform-verifier = ["dep:rustls-platform-verifier"]
 native-certs = ["dep:rustls-native-certs"]
 webpki-roots = ["dep:webpki-roots"]
-qlog = ["quinn-proto/qlog"]
+qlog = ["noq-proto/qlog"]
 h3 = ["dep:h3", "dep:h3-datagram"]
-ring = ["quinn-proto/rustls-ring"]
+ring = ["noq-proto/rustls-ring"]
 windows-gro = []
 sync = []
 

compio-quic/src/builder.rs

@@ -1,7 +1,7 @@
 use std::{io, sync::Arc};
 
 use compio_net::ToSocketAddrsAsync;
-use quinn_proto::{
+use noq_proto::{
     ClientConfig, ServerConfig,
     crypto::rustls::{QuicClientConfig, QuicServerConfig},
 };

compio-quic/src/connection.rs

@@ -17,10 +17,10 @@ use futures_util::{
     select, stream,
 };
 #[cfg(rustls)]
-use quinn_proto::crypto::rustls::HandshakeData;
-use quinn_proto::{
-    ConnectionHandle, ConnectionStats, Dir, EndpointEvent, Side, StreamEvent, StreamId, VarInt,
-    congestion::Controller,
+use noq_proto::crypto::rustls::HandshakeData;
+use noq_proto::{
+    ConnectionHandle, ConnectionStats, Dir, EndpointEvent, PathId, Side, StreamEvent, StreamId,
+    VarInt, congestion::Controller,
 };
 use rustc_hash::FxHashMap as HashMap;
 use thiserror::Error;
@@ -36,12 +36,12 @@ use crate::{
 #[derive(Debug)]
 pub(crate) enum ConnectionEvent {
     Close(VarInt, Bytes),
-    Proto(quinn_proto::ConnectionEvent),
+    Proto(noq_proto::ConnectionEvent),
 }
 
 #[derive(Debug)]
 pub(crate) struct ConnectionState {
-    pub(crate) conn: quinn_proto::Connection,
+    pub(crate) conn: noq_proto::Connection,
     pub(crate) error: Option<ConnectionError>,
     connected: bool,
     worker: Option<JoinHandle<()>>,
@@ -134,7 +134,7 @@ fn implicit_close(this: &Shared<ConnectionInner>) {
 impl ConnectionInner {
     fn new(
         handle: ConnectionHandle,
-        conn: quinn_proto::Connection,
+        conn: noq_proto::Connection,
         socket: Socket,
         events_tx: Sender<(ConnectionHandle, EndpointEvent)>,
         events_rx: Receiver<ConnectionEvent>,
@@ -246,7 +246,7 @@ impl ConnectionInner {
             }
 
             while let Some(event) = state.conn.poll() {
-                use quinn_proto::Event::*;
+                use noq_proto::Event::*;
                 match event {
                     HandshakeDataReady => {
                         if let Some(waker) = state.on_handshake_data.take() {
@@ -282,6 +282,16 @@ impl ConnectionInner {
                         .for_each(Waker::wake),
                     DatagramReceived => state.datagram_received.drain(..).for_each(Waker::wake),
                     DatagramsUnblocked => state.datagrams_unblocked.drain(..).for_each(Waker::wake),
+
+                    HandshakeConfirmed => {
+                        todo!()
+                    }
+                    Path(_) => {
+                        todo!()
+                    }
+                    NatTraversal(_) => {
+                        todo!()
+                    }
                 }
             }
 
@@ -313,19 +323,37 @@ macro_rules! conn_fn {
         /// This will return `None` for clients, or when the platform does not
         /// expose this information.
         pub fn local_ip(&self) -> Option<IpAddr> {
-            self.0.state().conn.local_ip()
+            let state = self.0.state();
+
+            state
+                .conn
+                .paths()
+                .iter()
+                .filter_map(|id| state.conn.network_path(*id).ok())
+                .next()
+                .unwrap()
+                .local_ip
         }
 
         /// The peer's UDP address.
         ///
         /// Will panic if called after `poll` has returned `Ready`.
         pub fn remote_address(&self) -> SocketAddr {
-            self.0.state().conn.remote_address()
+            let state = self.0.state();
+
+            state
+                .conn
+                .paths()
+                .iter()
+                .filter_map(|id| state.conn.network_path(*id).ok())
+                .next()
+                .unwrap()
+                .remote
         }
 
         /// Current best estimate of this connection's latency (round-trip-time).
-        pub fn rtt(&self) -> Duration {
-            self.0.state().conn.rtt()
+        pub fn rtt(&self, path_id: PathId) -> Option<Duration> {
+            self.0.state().conn.rtt(path_id)
         }
 
         /// Connection statistics.
@@ -335,8 +363,12 @@ macro_rules! conn_fn {
 
         /// Current state of the congestion control algorithm. (For debugging
         /// purposes)
-        pub fn congestion_state(&self) -> Box<dyn Controller> {
-            self.0.state().conn.congestion_state().clone_box()
+        pub fn congestion_state(&self, path_id: PathId) -> Option<Box<dyn Controller>> {
+            self.0
+                .state()
+                .conn
+                .congestion_state(path_id)
+                .map(|state| state.clone_box())
         }
 
         /// Cryptographic identity of the peer.
@@ -375,7 +407,7 @@ macro_rules! conn_fn {
             output: &mut [u8],
             label: &[u8],
             context: &[u8],
-        ) -> Result<(), quinn_proto::crypto::ExportKeyingMaterialError> {
+        ) -> Result<(), noq_proto::crypto::ExportKeyingMaterialError> {
             self.0
                 .state()
                 .conn
@@ -395,7 +427,7 @@ impl Connecting {
 
     pub(crate) fn new(
         handle: ConnectionHandle,
-        conn: quinn_proto::Connection,
+        conn: noq_proto::Connection,
         socket: Socket,
         events_tx: Sender<(ConnectionHandle, EndpointEvent)>,
         events_rx: Receiver<ConnectionEvent>,
@@ -573,14 +605,14 @@ impl Connection {
         state.wake();
     }
 
-    /// See [`quinn_proto::TransportConfig::send_window()`]
+    /// See [`noq_proto::TransportConfig::send_window()`]
     pub fn set_send_window(&self, send_window: u64) {
         let mut state = self.0.state();
         state.conn.set_send_window(send_window);
         state.wake();
     }
 
-    /// See [`quinn_proto::TransportConfig::receive_window()`]
+    /// See [`noq_proto::TransportConfig::receive_window()`]
     pub fn set_receive_window(&self, receive_window: VarInt) {
         let mut state = self.0.state();
         state.conn.set_receive_window(receive_window);
@@ -685,7 +717,7 @@ impl Connection {
         cx: Option<&mut Context>,
         data: Bytes,
     ) -> Result<(), Result<SendDatagramError, Bytes>> {
-        use quinn_proto::SendDatagramError::*;
+        use noq_proto::SendDatagramError::*;
         let mut state = self.0.try_state().map_err(|e| Ok(e.into()))?;
         state
             .conn
@@ -945,13 +977,13 @@ pub enum ConnectionError {
     /// The peer violated the QUIC specification as understood by this
     /// implementation
     #[error(transparent)]
-    TransportError(#[from] quinn_proto::TransportError),
+    TransportError(#[from] noq_proto::TransportError),
     /// The peer's QUIC stack aborted the connection automatically
     #[error("aborted by peer: {0}")]
-    ConnectionClosed(quinn_proto::ConnectionClose),
+    ConnectionClosed(noq_proto::ConnectionClose),
     /// The peer closed the connection
     #[error("closed by peer: {0}")]
-    ApplicationClosed(quinn_proto::ApplicationClose),
+    ApplicationClosed(noq_proto::ApplicationClose),
     /// The peer is unable to continue processing this connection, usually due
     /// to having restarted
     #[error("reset by peer")]
@@ -961,8 +993,8 @@ pub enum ConnectionError {
     ///
     /// If neither side is sending keep-alives, a connection will time out after
     /// a long enough idle period even if the peer is still reachable. See
-    /// also [`TransportConfig::max_idle_timeout()`](quinn_proto::TransportConfig::max_idle_timeout())
-    /// and [`TransportConfig::keep_alive_interval()`](quinn_proto::TransportConfig::keep_alive_interval()).
+    /// also [`TransportConfig::max_idle_timeout()`](noq_proto::TransportConfig::max_idle_timeout())
+    /// and [`TransportConfig::keep_alive_interval()`](noq_proto::TransportConfig::keep_alive_interval()).
     #[error("timed out")]
     TimedOut,
     /// The local application closed the connection
@@ -976,9 +1008,9 @@ pub enum ConnectionError {
     CidsExhausted,
 }
 
-impl From<quinn_proto::ConnectionError> for ConnectionError {
-    fn from(value: quinn_proto::ConnectionError) -> Self {
-        use quinn_proto::ConnectionError::*;
+impl From<noq_proto::ConnectionError> for ConnectionError {
+    fn from(value: noq_proto::ConnectionError) -> Self {
+        use noq_proto::ConnectionError::*;
 
         match value {
             VersionMismatch => ConnectionError::VersionMismatch,

compio-quic/src/endpoint.rs

@@ -21,9 +21,9 @@ use compio_net::UdpSocket;
 use compio_runtime::JoinHandle;
 use flume::{Receiver, Sender, unbounded};
 use futures_util::{FutureExt, StreamExt, future, select, task::AtomicWaker};
-use quinn_proto::{
+use noq_proto::{
     ClientConfig, ConnectError, ConnectionError, ConnectionHandle, DatagramEvent, EndpointConfig,
-    EndpointEvent, ServerConfig, Transmit, VarInt,
+    EndpointEvent, FourTuple, ServerConfig, Transmit, VarInt,
 };
 use rustc_hash::FxHashMap as HashMap;
 
@@ -34,12 +34,12 @@ use crate::{
 
 #[derive(Debug)]
 struct EndpointState {
-    endpoint: quinn_proto::Endpoint,
+    endpoint: noq_proto::Endpoint,
     worker: Option<JoinHandle<()>>,
     connections: HashMap<ConnectionHandle, Sender<ConnectionEvent>>,
     close: Option<(VarInt, Bytes)>,
     exit_on_idle: bool,
-    incoming: VecDeque<quinn_proto::Incoming>,
+    incoming: VecDeque<noq_proto::Incoming>,
     incoming_wakers: VecDeque<Waker>,
     stats: EndpointStats,
 }
@@ -68,8 +68,10 @@ impl EndpointState {
             let mut resp_buf = Vec::new();
             match self.endpoint.handle(
                 now,
-                meta.remote,
-                meta.local_ip,
+                FourTuple {
+                    remote: meta.remote,
+                    local_ip: meta.local_ip,
+                },
                 meta.ecn,
                 data,
                 &mut resp_buf,
@@ -112,7 +114,7 @@ impl EndpointState {
         self.connections.is_empty()
     }
 
-    fn poll_incoming(&mut self, cx: &mut Context) -> Poll<Option<quinn_proto::Incoming>> {
+    fn poll_incoming(&mut self, cx: &mut Context) -> Poll<Option<noq_proto::Incoming>> {
         if self.close.is_none() {
             if let Some(incoming) = self.incoming.pop_front() {
                 Poll::Ready(Some(incoming))
@@ -128,7 +130,7 @@ impl EndpointState {
     fn new_connection(
         &mut self,
         handle: ConnectionHandle,
-        conn: quinn_proto::Connection,
+        conn: noq_proto::Connection,
         socket: Socket,
         events_tx: Sender<(ConnectionHandle, EndpointEvent)>,
     ) -> Connecting {
@@ -173,11 +175,10 @@ impl EndpointInner {
 
         Ok(Self {
             state: Mutex::new(EndpointState {
-                endpoint: quinn_proto::Endpoint::new(
+                endpoint: noq_proto::Endpoint::new(
                     Arc::new(config),
                     server_config.map(Arc::new),
                     allow_mtud,
-                    None,
                 ),
                 worker: None,
                 connections: HashMap::default(),
@@ -237,7 +238,7 @@ impl EndpointInner {
 
     pub(crate) fn accept(
         &self,
-        incoming: quinn_proto::Incoming,
+        incoming: noq_proto::Incoming,
         server_config: Option<ServerConfig>,
     ) -> Result<Connecting, ConnectionError> {
         let mut state = self.state.lock();
@@ -260,7 +261,7 @@ impl EndpointInner {
         }
     }
 
-    pub(crate) fn refuse(&self, incoming: quinn_proto::Incoming) {
+    pub(crate) fn refuse(&self, incoming: noq_proto::Incoming) {
         let mut state = self.state.lock();
         state.stats.refused_handshakes += 1;
         let mut resp_buf = Vec::new();
@@ -269,18 +270,15 @@ impl EndpointInner {
     }
 
     #[allow(clippy::result_large_err)]
-    pub(crate) fn retry(
-        &self,
-        incoming: quinn_proto::Incoming,
-    ) -> Result<(), quinn_proto::RetryError> {
+    pub(crate) fn retry(&self, incoming: noq_proto::Incoming) -> Result<(), noq_proto::RetryError> {
         let mut state = self.state.lock();
         let mut resp_buf = Vec::new();
         let transmit = state.endpoint.retry(incoming, &mut resp_buf)?;
         self.respond(resp_buf, transmit);
         Ok(())
     }
 
-    pub(crate) fn ignore(&self, incoming: quinn_proto::Incoming) {
+    pub(crate) fn ignore(&self, incoming: noq_proto::Incoming) {
         let mut state = self.state.lock();
         state.stats.ignored_handshakes += 1;
         state.endpoint.ignore(incoming);

compio-quic/src/incoming.rs

@@ -6,14 +6,14 @@ use std::{
 };
 
 use futures_util::FutureExt;
-use quinn_proto::{ConnectionId, ServerConfig};
+use noq_proto::{ConnectionId, ServerConfig};
 use thiserror::Error;
 
 use crate::{Connecting, Connection, ConnectionError, EndpointRef};
 
 #[derive(Debug)]
 pub(crate) struct IncomingInner {
-    pub(crate) incoming: quinn_proto::Incoming,
+    pub(crate) incoming: noq_proto::Incoming,
     pub(crate) endpoint: EndpointRef,
 }
 
@@ -23,7 +23,7 @@ pub(crate) struct IncomingInner {
 pub struct Incoming(Option<IncomingInner>);
 
 impl Incoming {
-    pub(crate) fn new(incoming: quinn_proto::Incoming, endpoint: EndpointRef) -> Self {
+    pub(crate) fn new(incoming: noq_proto::Incoming, endpoint: EndpointRef) -> Self {
         Self(Some(IncomingInner { incoming, endpoint }))
     }
 
@@ -103,7 +103,7 @@ impl Incoming {
 
     /// The original destination CID when initiating the connection
     pub fn orig_dst_cid(&self) -> ConnectionId {
-        *self.0.as_ref().unwrap().incoming.orig_dst_cid()
+        self.0.as_ref().unwrap().incoming.orig_dst_cid()
     }
 }