gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-09-18 01:05:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove recv_datagrams from platform layer

Browse Source
This commit is contained in:
Ruediger Klaehn
2026-01-29 20:29:58 +02:00
parent 97530f37ef
commit dc613f2e79
7 changed files with 30 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quinn-udp/src/fallback.rs
+1 -33
View File
@@ -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<ReceivedDatagrams> {
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
quinn-udp/src/unix.rs
+1 -57
View File
@@ -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<ReceivedDatagrams> {
// 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).
///
quinn-udp/src/windows.rs
+1 -42
View File
@@ -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<ReceivedDatagrams> {
// 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).
///
quinn/src/endpoint.rs
+6 -8
View File
@@ -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);
}
}
quinn/src/runtime/mod.rs
+1 -4
View File
@@ -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<io::Result<ReceivedDatagrams>>;
fn poll_recv_datagrams(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<ReceivedDatagrams>>;
/// Look up the local IP address and port used by this socket
fn local_addr(&self) -> io::Result<SocketAddr>;
quinn/src/runtime/smol.rs
+11 -2
View File
@@ -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) {
quinn/src/runtime/tokio.rs
+9 -2
View File
@@ -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();