diff --git a/quinn-udp/src/fallback.rs b/quinn-udp/src/fallback.rs index 87a3d7780..4442abc52 100644 --- a/quinn-udp/src/fallback.rs +++ b/quinn-udp/src/fallback.rs @@ -5,10 +5,7 @@ use std::{ time::Instant, }; -use super::{ - IO_ERROR_LOG_INTERVAL, RecvMeta, ReceivedDatagram, ReceivedDatagrams, Transmit, UdpSockRef, - log_sendmsg_error, -}; +use super::{IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, log_sendmsg_error}; /// Fallback UDP socket interface that stubs out all special functionality /// @@ -81,35 +78,6 @@ impl UdpSocketState { Ok(1) } - /// Receives datagrams from the socket, returning owned data. - /// - /// This is a higher-level API that handles buffer management internally. - /// Each datagram in the returned collection contains its own buffer. - pub fn recv_datagrams( - &self, - socket: UdpSockRef<'_>, - max_payload_size: usize, - ) -> io::Result { - let mut recv_buf = vec![0u8; max_payload_size]; - let mut bufs = [IoSliceMut::new(&mut recv_buf)]; - let mut metas = [RecvMeta::default()]; - - let msg_count = self.recv(socket, &mut bufs, &mut metas)?; - - let mut result = ReceivedDatagrams::new(); - for meta in metas.iter().take(msg_count) { - let data = recv_buf[..meta.len].to_vec(); - result.push(ReceivedDatagram { - data, - remote: meta.addr, - local_ip: meta.dst_ip, - ecn: meta.ecn, - }); - } - - Ok(result) - } - #[inline] pub fn max_gso_segments(&self) -> usize { 1 diff --git a/quinn-udp/src/unix.rs b/quinn-udp/src/unix.rs index 3562a5412..825feac12 100644 --- a/quinn-udp/src/unix.rs +++ b/quinn-udp/src/unix.rs @@ -16,8 +16,7 @@ use std::{ use socket2::SockRef; use super::{ - EcnCodepoint, IO_ERROR_LOG_INTERVAL, RecvMeta, ReceivedDatagram, ReceivedDatagrams, Transmit, - UdpSockRef, cmsg, log_sendmsg_error, + EcnCodepoint, IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, cmsg, log_sendmsg_error, }; // Adapted from https://github.com/apple-oss-distributions/xnu/blob/8d741a5de7ff4191bf97d57b9f54c2f6d4a15585/bsd/sys/socket_private.h @@ -235,61 +234,6 @@ impl UdpSocketState { recv(socket.0, bufs, meta) } - /// Receives datagrams from the socket, returning owned data. - /// - /// This is a higher-level API that handles buffer management and GRO splitting - /// internally. Each datagram in the returned collection contains its own buffer. - /// - /// # Arguments - /// - /// * `socket` - The UDP socket to receive from - /// * `max_payload_size` - Maximum expected UDP payload size (typically 65535 or less) - /// - /// # Returns - /// - /// A collection of received datagrams, or an error if the receive failed. - pub fn recv_datagrams( - &self, - socket: UdpSockRef<'_>, - max_payload_size: usize, - ) -> io::Result { - // Allocate buffer sized for GRO coalescing - let gro_segments = self.gro_segments.get(); - let buf_size = max_payload_size * gro_segments; - let mut recv_buf = vec![0u8; buf_size * BATCH_SIZE]; - - // Prepare IoSliceMut array for recv - let mut bufs: [IoSliceMut<'_>; BATCH_SIZE] = - std::array::from_fn(|_| IoSliceMut::new(&mut [])); - for (i, chunk) in recv_buf.chunks_mut(buf_size).enumerate().take(BATCH_SIZE) { - bufs[i] = IoSliceMut::new(chunk); - } - - let mut metas = [RecvMeta::default(); BATCH_SIZE]; - - // Call the underlying recv - let msg_count = recv(socket.0, &mut bufs, &mut metas)?; - - // Convert to ReceivedDatagrams, splitting by stride - let mut result = ReceivedDatagrams::new(); - for (meta, buf) in metas.iter().zip(bufs.iter()).take(msg_count) { - let mut offset = 0; - while offset < meta.len { - let stride = meta.stride.min(meta.len - offset); - let data = buf[offset..offset + stride].to_vec(); - result.push(ReceivedDatagram { - data, - remote: meta.addr, - local_ip: meta.dst_ip, - ecn: meta.ecn, - }); - offset += stride; - } - } - - Ok(result) - } - /// The maximum amount of segments which can be transmitted if a platform /// supports Generic Send Offload (GSO). /// diff --git a/quinn-udp/src/windows.rs b/quinn-udp/src/windows.rs index bb0a980b8..e3519a29d 100644 --- a/quinn-udp/src/windows.rs +++ b/quinn-udp/src/windows.rs @@ -16,8 +16,7 @@ use libc::{c_int, c_uint}; use windows_sys::Win32::Networking::WinSock; use crate::{ - EcnCodepoint, IO_ERROR_LOG_INTERVAL, RecvMeta, ReceivedDatagram, ReceivedDatagrams, Transmit, - UdpSockRef, + EcnCodepoint, IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, cmsg::{self, CMsgHdr}, log::debug, log_sendmsg_error, @@ -282,46 +281,6 @@ impl UdpSocketState { Ok(1) } - /// Receives datagrams from the socket, returning owned data. - /// - /// This is a higher-level API that handles buffer management and GRO splitting - /// internally. Each datagram in the returned collection contains its own buffer. - pub fn recv_datagrams( - &self, - socket: UdpSockRef<'_>, - max_payload_size: usize, - ) -> io::Result { - // Allocate buffer sized for URO coalescing - let gro_segments = self.gro_segments().get(); - let buf_size = max_payload_size * gro_segments; - let mut recv_buf = vec![0u8; buf_size]; - - let mut bufs = [IoSliceMut::new(&mut recv_buf)]; - let mut metas = [RecvMeta::default()]; - - // Call the underlying recv - let msg_count = self.recv(socket, &mut bufs, &mut metas)?; - - // Convert to ReceivedDatagrams, splitting by stride - let mut result = ReceivedDatagrams::new(); - for meta in metas.iter().take(msg_count) { - let mut offset = 0; - while offset < meta.len { - let stride = meta.stride.min(meta.len - offset); - let data = recv_buf[offset..offset + stride].to_vec(); - result.push(ReceivedDatagram { - data, - remote: meta.addr, - local_ip: meta.dst_ip, - ecn: meta.ecn, - }); - offset += stride; - } - } - - Ok(result) - } - /// The maximum amount of segments which can be transmitted if a platform /// supports Generic Send Offload (GSO). /// diff --git a/quinn/src/endpoint.rs b/quinn/src/endpoint.rs index cf7b1258b..f4202344c 100644 --- a/quinn/src/endpoint.rs +++ b/quinn/src/endpoint.rs @@ -2,8 +2,7 @@ use std::{ collections::VecDeque, fmt, future::Future, - io, - mem, + io, mem, net::{SocketAddr, SocketAddrV6}, pin::Pin, str, @@ -793,9 +792,7 @@ struct RecvState { } impl RecvState { - fn new( - sender: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>, - ) -> Self { + fn new(sender: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>) -> Self { Self { connections: ConnectionSet { senders: FxHashMap::default(), @@ -831,15 +828,16 @@ impl RecvState { now, addresses, datagram.ecn.map(proto_ecn), - Bytes::from(datagram.data).try_into_mut().expect("freshly created bytes is unique"), + Bytes::from(datagram.data) + .try_into_mut() + .expect("freshly created bytes is unique"), &mut response_buffer, ) { Some(DatagramEvent::NewConnection(incoming)) => { if self.connections.close.is_none() { self.incoming.push_back(incoming); } else { - let transmit = - endpoint.refuse(incoming, &mut response_buffer); + let transmit = endpoint.refuse(incoming, &mut response_buffer); respond(transmit, &response_buffer, sender); } } diff --git a/quinn/src/runtime/mod.rs b/quinn/src/runtime/mod.rs index 65fdac7d5..a4a1b2b68 100644 --- a/quinn/src/runtime/mod.rs +++ b/quinn/src/runtime/mod.rs @@ -66,10 +66,7 @@ pub trait AsyncUdpSocket: Send + Sync + Debug + 'static { /// /// This is a higher-level API that handles buffer management and GRO splitting internally. /// Each datagram in the returned collection contains its own buffer. - fn poll_recv_datagrams( - &mut self, - cx: &mut Context<'_>, - ) -> Poll>; + fn poll_recv_datagrams(&mut self, cx: &mut Context<'_>) -> Poll>; /// Look up the local IP address and port used by this socket fn local_addr(&self) -> io::Result; diff --git a/quinn/src/runtime/smol.rs b/quinn/src/runtime/smol.rs index 81e657a46..5309c36f2 100644 --- a/quinn/src/runtime/smol.rs +++ b/quinn/src/runtime/smol.rs @@ -123,12 +123,21 @@ impl AsyncUdpSocket for UdpSocket { // Prepare IoSliceMut array let mut bufs: [IoSliceMut<'_>; udp::BATCH_SIZE] = std::array::from_fn(|_| IoSliceMut::new(&mut [])); - for (i, chunk) in self.recv_buf.chunks_mut(buf_size).enumerate().take(udp::BATCH_SIZE) { + for (i, chunk) in self + .recv_buf + .chunks_mut(buf_size) + .enumerate() + .take(udp::BATCH_SIZE) + { bufs[i] = IoSliceMut::new(chunk); } let mut metas = [udp::RecvMeta::default(); udp::BATCH_SIZE]; - if let Ok(msg_count) = self.send.inner.recv((&self.send.io).into(), &mut bufs, &mut metas) { + if let Ok(msg_count) = + self.send + .inner + .recv((&self.send.io).into(), &mut bufs, &mut metas) + { // Convert to ReceivedDatagrams, splitting by stride let mut result = udp::ReceivedDatagrams::new(); for (meta, buf) in metas.iter().zip(bufs.iter()).take(msg_count) { diff --git a/quinn/src/runtime/tokio.rs b/quinn/src/runtime/tokio.rs index e1cb96172..2bf41436e 100644 --- a/quinn/src/runtime/tokio.rs +++ b/quinn/src/runtime/tokio.rs @@ -126,13 +126,20 @@ impl AsyncUdpSocket for UdpSocket { // Prepare IoSliceMut array let mut bufs: [IoSliceMut<'_>; udp::BATCH_SIZE] = std::array::from_fn(|_| IoSliceMut::new(&mut [])); - for (i, chunk) in self.recv_buf.chunks_mut(buf_size).enumerate().take(udp::BATCH_SIZE) { + for (i, chunk) in self + .recv_buf + .chunks_mut(buf_size) + .enumerate() + .take(udp::BATCH_SIZE) + { bufs[i] = IoSliceMut::new(chunk); } let mut metas = [udp::RecvMeta::default(); udp::BATCH_SIZE]; if let Ok(msg_count) = self.send.io.try_io(Interest::READABLE, || { - self.send.inner.recv((&self.send.io).into(), &mut bufs, &mut metas) + self.send + .inner + .recv((&self.send.io).into(), &mut bufs, &mut metas) }) { // Convert to ReceivedDatagrams, splitting by stride let mut result = udp::ReceivedDatagrams::new();