perf(http3): eliminate allocations with zero-copy Bytes

Cargo.toml

@@ -32,6 +32,7 @@ license = "MIT OR Apache-2.0"
 rust-version = "1.81.0"
 
 [workspace.dependencies]
+bytes = { version = "1.0", default-features = false }
 enum-map = { version = "2.7", default-features = false }
 enumset = { version = "1.1", default-features = false }
 hex = { version = "0.4", default-features = false }

neqo-http3/Cargo.toml

@@ -19,6 +19,7 @@ autobenches = false
 workspace = true
 
 [dependencies]
+bytes = { workspace = true }
 enumset = { workspace = true }
 log = { workspace = true }
 neqo-common = { path = "./../neqo-common" }

neqo-http3/src/client_events.rs

@@ -14,8 +14,8 @@ use crate::{
     connection::Http3State,
     features::extended_connect::{self, ExtendedConnectEvents, ExtendedConnectType},
     settings::HSettingType,
-    CloseType, Error, Http3StreamInfo, HttpRecvStreamEvents, PushId, RecvStreamEvents, Res,
-    SendStreamEvents,
+    CloseType, DatagramPayload, Error, Http3StreamInfo, HttpRecvStreamEvents, PushId,
+    RecvStreamEvents, Res, SendStreamEvents,
 };
 
 #[derive(Debug, PartialEq, Eq, Clone)]
@@ -40,7 +40,7 @@ pub enum WebTransportEvent {
     },
     Datagram {
         session_id: StreamId,
-        datagram: Vec<u8>,
+        datagram: DatagramPayload,
     },
 }
 
@@ -62,7 +62,7 @@ pub enum ConnectUdpEvent {
     },
     Datagram {
         session_id: StreamId,
-        datagram: Vec<u8>,
+        datagram: DatagramPayload,
     },
 }
 
@@ -283,7 +283,7 @@ impl ExtendedConnectEvents for Http3ClientEvents {
     fn new_datagram(
         &self,
         session_id: StreamId,
-        datagram: Vec<u8>,
+        datagram: DatagramPayload,
         connect_type: ExtendedConnectType,
     ) {
         let event = match connect_type {

neqo-http3/src/connection.rs

@@ -660,18 +660,24 @@ impl Http3Connection {
         }
     }
 
-    pub(crate) fn handle_datagram(&mut self, datagram: &[u8]) {
-        let mut decoder = Decoder::new(datagram);
-        let Some(stream) = decoder
-            .decode_varint()
-            .and_then(|id| self.recv_streams.get_mut(&StreamId::from(id * 4)))
+    pub(crate) fn handle_datagram(&mut self, datagram: Vec<u8>) {
+        let mut decoder = Decoder::new(&datagram);
+        let Some(id) = decoder.decode_varint() else {
+            qdebug!("[{self}] handle_datagram: failed to decode session ID");
+            return;
+        };
+        let varint_len = decoder.offset();
+
+        let Some(stream) = self
+            .recv_streams
+            .get_mut(&StreamId::from(id * 4))
             .and_then(|s| s.extended_connect_session())
         else {
             qdebug!("[{self}] handle_datagram for unknown extended connect session");
             return;
         };
 
-        stream.borrow_mut().datagram(decoder.as_ref());
+        stream.borrow_mut().datagram(datagram, varint_len);
     }
 
     fn check_stream_exists(&self, stream_type: Http3StreamType) -> Res<()> {

neqo-http3/src/connection_client.rs

@@ -1157,7 +1157,7 @@ impl Http3Client {
                     }
                 }
                 ConnectionEvent::Datagram(dgram) => {
-                    self.base_handler.handle_datagram(&dgram);
+                    self.base_handler.handle_datagram(dgram);
                 }
                 ConnectionEvent::SendStreamComplete { .. }
                 | ConnectionEvent::OutgoingDatagramOutcome { .. }

neqo-http3/src/connection_server.rs

@@ -366,7 +366,7 @@ impl Http3ServerHandler {
                         s.stream_writable();
                     }
                 }
-                ConnectionEvent::Datagram(dgram) => self.base_handler.handle_datagram(&dgram),
+                ConnectionEvent::Datagram(dgram) => self.base_handler.handle_datagram(dgram),
                 ConnectionEvent::AuthenticationNeeded
                 | ConnectionEvent::EchFallbackAuthenticationNeeded { .. }
                 | ConnectionEvent::ZeroRttRejected

neqo-http3/src/features/extended_connect/mod.rs

@@ -51,7 +51,7 @@ pub(crate) trait ExtendedConnectEvents: Debug {
     fn new_datagram(
         &self,
         session_id: StreamId,
-        datagram: Vec<u8>,
+        datagram: crate::DatagramPayload,
         connect_type: ExtendedConnectType,
     );
 }

neqo-http3/src/features/extended_connect/session.rs

@@ -398,20 +398,28 @@ impl Session {
         Ok(())
     }
 
-    pub(crate) fn datagram(&self, datagram: &[u8]) {
+    pub(crate) fn datagram(&self, datagram: Vec<u8>, payload_offset: usize) {
         if self.state != State::Active {
             qdebug!("[{self}]: received datagram on {:?} session.", self.state);
             return;
         }
-        let datagram = match self.protocol.dgram_context_id(datagram) {
-            Ok(datagram) => datagram,
+
+        // Validate the payload portion based on protocol requirements
+        match self.protocol.dgram_context_id(&datagram[payload_offset..]) {
+            Ok(slice) => {
+                // Calculate total offset: session_id varint + context_id (if any)
+                // If the returned slice is shorter than the input, a context identifier is present.
+                let context_offset = usize::from(slice.len() != datagram[payload_offset..].len());
+                let total_offset = payload_offset + context_offset;
+                let payload = crate::DatagramPayload::new(datagram, total_offset);
+
+                self.events
+                    .new_datagram(self.id, payload, self.protocol.connect_type());
+            }
             Err(e) => {
                 qdebug!("[{self}]: received datagram with invalid context identifier: {e}");
-                return;
             }
-        };
-        self.events
-            .new_datagram(self.id, datagram.to_vec(), self.protocol.connect_type());
+        }
     }
 
     fn has_data_to_send(&self) -> bool {

neqo-http3/src/features/extended_connect/tests/webtransport/mod.rs

@@ -614,7 +614,7 @@ impl WtTest {
                 Http3ClientEvent::WebTransport(WebTransportEvent::Datagram {
                     session_id,
                     datagram
-                }) if session_id == expected_stream_id && datagram == expected_dgram
+                }) if session_id == expected_stream_id && datagram.as_ref() == expected_dgram
             )
         };
         assert!(self.client.events().any(wt_datagram_event));
@@ -631,7 +631,7 @@ impl WtTest {
                 Http3ServerEvent::WebTransport(WebTransportServerEvent::Datagram {
                     session,
                     datagram
-                }) if session.stream_id() == expected_session.stream_id() && datagram == expected_dgram
+                }) if session.stream_id() == expected_session.stream_id() && datagram.as_ref() == expected_dgram
             )
         };
         assert!(self.server.events().any(wt_datagram_event));

neqo-http3/src/lib.rs

@@ -162,6 +162,7 @@ mod stream_type_reader;
 use std::{cell::RefCell, fmt::Debug, rc::Rc, time::Instant};
 
 use buffered_send_stream::BufferedStream;
+use bytes::Bytes;
 pub use client_events::{ConnectUdpEvent, Http3ClientEvent, WebTransportEvent};
 pub use conn_params::Http3Parameters;
 pub use connection::{Http3State, SessionAcceptAction};
@@ -184,6 +185,49 @@ use thiserror::Error;
 
 use crate::{features::extended_connect, priority::PriorityHandler};
 
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub struct DatagramPayload(Bytes);
+
+impl DatagramPayload {
+    /// # Panics
+    ///
+    /// If the payload offset lies outside the data.
+    #[must_use]
+    pub fn new(data: Vec<u8>, offset: usize) -> Self {
+        assert!(offset <= data.len());
+        Self(Bytes::from(data).slice(offset..))
+    }
+
+    #[must_use]
+    pub const fn payload(&self) -> &Bytes {
+        &self.0
+    }
+}
+
+impl AsRef<[u8]> for DatagramPayload {
+    fn as_ref(&self) -> &[u8] {
+        &self.0
+    }
+}
+
+impl From<DatagramPayload> for Bytes {
+    fn from(payload: DatagramPayload) -> Self {
+        payload.0
+    }
+}
+
+impl<const N: usize> PartialEq<[u8; N]> for DatagramPayload {
+    fn eq(&self, other: &[u8; N]) -> bool {
+        self.0 == other.as_slice()
+    }
+}
+
+impl PartialEq<&[u8]> for DatagramPayload {
+    fn eq(&self, other: &&[u8]) -> bool {
+        self.0 == *other
+    }
+}
+
 type Res<T> = Result<T, Error>;
 
 #[derive(Clone, Debug, PartialEq, Eq, Error)]

neqo-http3/src/server_connection_events.rs

@@ -64,7 +64,7 @@ pub enum WebTransportEvent {
     NewStream(Http3StreamInfo),
     Datagram {
         session_id: StreamId,
-        datagram: Vec<u8>,
+        datagram: crate::DatagramPayload,
     },
 }
 
@@ -81,7 +81,7 @@ pub enum ConnectUdpEvent {
     },
     Datagram {
         session_id: StreamId,
-        datagram: Vec<u8>,
+        datagram: crate::DatagramPayload,
     },
 }
 
@@ -220,7 +220,7 @@ impl ExtendedConnectEvents for Http3ServerConnEvents {
     fn new_datagram(
         &self,
         session_id: StreamId,
-        datagram: Vec<u8>,
+        datagram: crate::DatagramPayload,
         connect_type: ExtendedConnectType,
     ) {
         let event = match connect_type {

neqo-http3/src/server_events.rs

@@ -508,7 +508,7 @@ pub enum WebTransportServerEvent {
     NewStream(Http3OrWebTransportStream),
     Datagram {
         session: WebTransportRequest,
-        datagram: Vec<u8>,
+        datagram: crate::DatagramPayload,
     },
 }
 
@@ -525,7 +525,7 @@ pub enum ConnectUdpServerEvent {
     },
     Datagram {
         session: ConnectUdpRequest,
-        datagram: Vec<u8>,
+        datagram: crate::DatagramPayload,
     },
 }
 
@@ -724,12 +724,20 @@ impl Http3ServerEvents {
         ));
     }
 
-    pub(crate) fn webtransport_datagram(&self, session: WebTransportRequest, datagram: Vec<u8>) {
+    pub(crate) fn webtransport_datagram(
+        &self,
+        session: WebTransportRequest,
+        datagram: crate::DatagramPayload,
+    ) {
         self.insert(Http3ServerEvent::WebTransport(
             WebTransportServerEvent::Datagram { session, datagram },
         ));
     }
-    pub(crate) fn connect_udp_datagram(&self, session: ConnectUdpRequest, datagram: Vec<u8>) {
+    pub(crate) fn connect_udp_datagram(
+        &self,
+        session: ConnectUdpRequest,
+        datagram: crate::DatagramPayload,
+    ) {
         self.insert(Http3ServerEvent::ConnectUdp(
             ConnectUdpServerEvent::Datagram { session, datagram },
         ));

neqo-http3/tests/connect_udp.rs

@@ -6,6 +6,7 @@
 
 #![cfg(test)]
 
+use bytes::Bytes;
 use neqo_common::{event::Provider as _, header::HeadersExt as _, qinfo, Datagram, Tos};
 use neqo_crypto::AuthenticationStatus;
 use neqo_http3::{
@@ -143,7 +144,7 @@ fn exchange_packets_through_proxy(
                 DEFAULT_ADDR,
                 DEFAULT_ADDR,
                 Tos::default(),
-                datagram,
+                Bytes::from(datagram),
             ))
         }
         _ => None,
@@ -184,7 +185,7 @@ fn exchange_packets_through_proxy(
                 DEFAULT_ADDR,
                 DEFAULT_ADDR,
                 Tos::default(),
-                datagram,
+                Bytes::from(datagram),
             ))
         } else {
             None

neqo-http3/tests/webtransport.rs

@@ -490,7 +490,7 @@ fn wt_session_ok_and_wt_datagram_in_same_udp_datagram() {
             client.events().next(),
             Some(Http3ClientEvent::WebTransport(
                 WebTransportEvent::Datagram{ session_id, datagram }
-            )) if session_id == wt_server_session.stream_id() && datagram == b"PING",
+            )) if session_id == wt_server_session.stream_id() && &datagram == b"PING",
         ),
         "Should receive datagram"
     );