diff --git a/Cargo.lock b/Cargo.lock index 6f35d93..6ef01f2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1432,7 +1432,7 @@ dependencies = [ [[package]] name = "mqp2p" -version = "0.2.0" +version = "0.3.0" dependencies = [ "clap", "flume", diff --git a/crates/mqp2p/Cargo.toml b/crates/mqp2p/Cargo.toml index 1a3900a..c350822 100644 --- a/crates/mqp2p/Cargo.toml +++ b/crates/mqp2p/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqp2p" -version = "0.2.0" +version = "0.3.0" edition = "2024" rust-version = "1.88" license = "Apache-2.0" diff --git a/crates/mqp2p/src/peer.rs b/crates/mqp2p/src/peer.rs index e042ed8..591872e 100644 --- a/crates/mqp2p/src/peer.rs +++ b/crates/mqp2p/src/peer.rs @@ -1,5 +1,5 @@ use crate::error::{Error, Result}; -use crate::quic::{self, CertIdentity, QuicEndpoint}; +use crate::quic::{self, CertIdentity, QuicEndpoint, SegmentationOffload}; use crate::signaling::{ Candidate, CandidateKind, ConnectionOffer, ConnectionSync, MIN_LEASE_SECS, PEER_LEASE_SECS, PeerInfo, SignalingClient, heartbeat_interval_secs, @@ -20,6 +20,7 @@ pub struct PeerConfig { pub stun_server: Option, pub credentials: Option<(String, String)>, pub lease_secs: u64, + pub segmentation_offload: SegmentationOffload, } impl PeerConfig { @@ -31,6 +32,7 @@ impl PeerConfig { stun_server: Some(stun::DEFAULT_STUN_SERVER.into()), credentials: None, lease_secs: PEER_LEASE_SECS, + segmentation_offload: SegmentationOffload::Enabled, } } @@ -54,6 +56,11 @@ impl PeerConfig { self } + pub fn without_segmentation_offload(mut self) -> Self { + self.segmentation_offload = SegmentationOffload::Disabled; + self + } + pub fn with_credentials( mut self, username: impl Into, @@ -118,13 +125,14 @@ impl Peer { }; let quic_socket = std_socket.try_clone()?; - let quic = QuicEndpoint::bind( + let quic = QuicEndpoint::bind_with_offload( quic_socket, CertIdentity { cert_der: identity.cert_der.clone(), key_der: identity.key_der.clone(), fingerprint: identity.fingerprint.clone(), }, + config.segmentation_offload, )?; let client_id = format!("mqp2p-{}", config.name); diff --git a/crates/mqp2p/src/quic.rs b/crates/mqp2p/src/quic.rs index 411293e..50e40be 100644 --- a/crates/mqp2p/src/quic.rs +++ b/crates/mqp2p/src/quic.rs @@ -52,7 +52,22 @@ pub fn compute_fingerprint(cert_der: &[u8]) -> String { format!("sha256:{hex}") } -fn build_server_config(identity: &CertIdentity) -> Result { +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SegmentationOffload { + Enabled, + Disabled, +} + +fn transport_config(offload: SegmentationOffload) -> Arc { + let mut transport = quinn::TransportConfig::default(); + transport.enable_segmentation_offload(offload == SegmentationOffload::Enabled); + Arc::new(transport) +} + +fn build_server_config( + identity: &CertIdentity, + offload: SegmentationOffload, +) -> Result { let cert = CertificateDer::from(identity.cert_der.clone()); let key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(identity.key_der.clone())); @@ -66,12 +81,15 @@ fn build_server_config(identity: &CertIdentity) -> Result { let quic_server_config = quinn::crypto::rustls::QuicServerConfig::try_from(server_crypto) .map_err(|e| Error::Quic(format!("QUIC server config failed: {e}")))?; - Ok(ServerConfig::with_crypto(Arc::new(quic_server_config))) + let mut server_config = ServerConfig::with_crypto(Arc::new(quic_server_config)); + server_config.transport_config(transport_config(offload)); + Ok(server_config) } fn build_client_config( identity: &CertIdentity, expected_fingerprint: &str, + offload: SegmentationOffload, ) -> Result { let cert = CertificateDer::from(identity.cert_der.clone()); let key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(identity.key_der.clone())); @@ -89,7 +107,9 @@ fn build_client_config( let quic_client_config = quinn::crypto::rustls::QuicClientConfig::try_from(client_crypto) .map_err(|e| Error::Quic(format!("QUIC client config failed: {e}")))?; - Ok(quinn::ClientConfig::new(Arc::new(quic_client_config))) + let mut client_config = quinn::ClientConfig::new(Arc::new(quic_client_config)); + client_config.transport_config(transport_config(offload)); + Ok(client_config) } #[derive(Debug)] @@ -215,11 +235,20 @@ impl rustls::server::danger::ClientCertVerifier for CollectClientCert { pub struct QuicEndpoint { endpoint: Endpoint, identity: CertIdentity, + offload: SegmentationOffload, } impl QuicEndpoint { pub fn bind(socket: std::net::UdpSocket, identity: CertIdentity) -> Result { - let server_config = build_server_config(&identity)?; + Self::bind_with_offload(socket, identity, SegmentationOffload::Enabled) + } + + pub fn bind_with_offload( + socket: std::net::UdpSocket, + identity: CertIdentity, + offload: SegmentationOffload, + ) -> Result { + let server_config = build_server_config(&identity, offload)?; let runtime = quinn::default_runtime() .ok_or_else(|| Error::Quic("no async runtime available".into()))?; @@ -234,9 +263,13 @@ impl QuicEndpoint { let local = endpoint .local_addr() .map_err(|e| Error::Quic(format!("{e}")))?; - info!(addr = %local, "QUIC endpoint bound"); + info!(addr = %local, ?offload, "QUIC endpoint bound"); - Ok(Self { endpoint, identity }) + Ok(Self { + endpoint, + identity, + offload, + }) } pub fn local_addr(&self) -> Result { @@ -250,7 +283,7 @@ impl QuicEndpoint { } pub async fn connect(&self, addr: SocketAddr, remote_fingerprint: &str) -> Result { - let client_config = build_client_config(&self.identity, remote_fingerprint)?; + let client_config = build_client_config(&self.identity, remote_fingerprint, self.offload)?; let connection = self .endpoint diff --git a/crates/mqp2p/tests/quic_offload.rs b/crates/mqp2p/tests/quic_offload.rs new file mode 100644 index 0000000..15e8114 --- /dev/null +++ b/crates/mqp2p/tests/quic_offload.rs @@ -0,0 +1,33 @@ +use mqp2p::quic::{QuicEndpoint, SegmentationOffload, generate_self_signed_cert}; +use std::net::UdpSocket; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn endpoints_without_segmentation_offload_exchange_data() { + let id_a = generate_self_signed_cert().expect("cert a"); + let id_b = generate_self_signed_cert().expect("cert b"); + let fp_a = id_a.fingerprint.clone(); + let fp_b = id_b.fingerprint.clone(); + let socket = || UdpSocket::bind("127.0.0.1:0").expect("socket"); + let ep_a = QuicEndpoint::bind_with_offload(socket(), id_a, SegmentationOffload::Disabled) + .expect("bind a"); + let ep_b = QuicEndpoint::bind_with_offload(socket(), id_b, SegmentationOffload::Disabled) + .expect("bind b"); + let addr_b = ep_b.local_addr().expect("addr b"); + let payload = vec![0x5a_u8; 64 * 1024]; + let expected = payload.clone(); + + let receiver = tokio::spawn(async move { + let conn = ep_b.accept_with_fingerprint(&fp_a).await.expect("accept"); + let mut recv = conn.accept_uni().await.expect("stream"); + let received = recv.read_to_end(1024 * 1024).await.expect("read"); + conn.close(0u32.into(), b"done"); + received + }); + + let conn = ep_a.connect(addr_b, &fp_b).await.expect("connect"); + let mut send = conn.open_uni().await.expect("open"); + send.write_all(&payload).await.expect("write"); + send.finish().expect("finish"); + + assert_eq!(receiver.await.expect("receiver"), expected); +} diff --git a/crates/stitch-p2p/Cargo.toml b/crates/stitch-p2p/Cargo.toml index 906802d..08cf74a 100644 --- a/crates/stitch-p2p/Cargo.toml +++ b/crates/stitch-p2p/Cargo.toml @@ -13,7 +13,7 @@ readme = "README.md" [dependencies] ed25519-dalek = { version = "3", default-features = false, features = ["fast"], optional = true } fjall = { version = "3.0", optional = true } -mqp2p = { version = "0.2.0", path = "../mqp2p", optional = true } +mqp2p = { version = "0.3.0", path = "../mqp2p", optional = true } serde_json = { version = "1.0.150", optional = true } thiserror = "2.0.18" tokio = { version = "1.52.3", features = ["sync", "io-util", "macros", "rt", "rt-multi-thread", "time", "net"] } diff --git a/crates/stitch-tasks/Cargo.toml b/crates/stitch-tasks/Cargo.toml index df6b914..b758b8d 100644 --- a/crates/stitch-tasks/Cargo.toml +++ b/crates/stitch-tasks/Cargo.toml @@ -6,7 +6,7 @@ license = "Apache-2.0" publish = false [dependencies] -mqp2p = { version = "0.2.0", path = "../mqp2p" } +mqp2p = { version = "0.3.0", path = "../mqp2p" } rand = "0.10.1" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150"