gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-09-25 04:35:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop sendmmsg support

Browse Source 
As we no longer buffer multiple transmits in memory, this complexity
is unused. GSO is expected to account for most, if not all, of the
performance benefit.
This commit is contained in:
Benjamin Saunders
2023-12-20 18:52:10 -08:00
parent 07222809c0
commit ee0882657a
9 changed files with 175 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quinn-udp/src/fallback.rs
+18 -30
View File
@@ -24,37 +24,25 @@ impl UdpSocketState {
}) })
} }
pub fn send(&self, socket: UdpSockRef<'_>, transmits: &[Transmit]) -> io::Result<usize> { pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit) -> io::Result<()> {
let mut sent = 0; let Err(e) = socket.0.send_to(
for transmit in transmits { &transmit.contents,
match socket.0.send_to( &socket2::SockAddr::from(transmit.destination),
&transmit.contents, ) else {
&socket2::SockAddr::from(transmit.destination), return Ok(());
) { };
Ok(_) => { if e.kind() == io::ErrorKind::WouldBlock {
sent += 1; return Err(e);
}
// We need to report that some packets were sent in this case, so we rely on
// errors being either harmlessly transient (in the case of WouldBlock) or
// recurring on the next call.
Err(_) if sent != 0 => return Ok(sent),
Err(e) => {
if e.kind() == io::ErrorKind::WouldBlock {
return Err(e);
}
// Other errors are ignored, since they will usually be handled
// by higher level retransmits and timeouts.
// - PermissionDenied errors have been observed due to iptable rules.
// Those are not fatal errors, since the
// configuration can be dynamically changed.
// - Destination unreachable errors have been observed for other
log_sendmsg_error(&self.last_send_error, e, transmit);
sent += 1;
}
}
} }
Ok(sent)
// Other errors are ignored, since they will usually be handled
// by higher level retransmits and timeouts.
// - PermissionDenied errors have been observed due to iptable rules.
// Those are not fatal errors, since the
// configuration can be dynamically changed.
// - Destination unreachable errors have been observed for other
log_sendmsg_error(&self.last_send_error, e, transmit);
Ok(())
} }
pub fn recv( pub fn recv(
quinn-udp/src/unix.rs
+53 -150
View File
@@ -156,8 +156,8 @@ impl UdpSocketState {
}) })
} }
pub fn send(&self, socket: UdpSockRef<'_>, transmits: &[Transmit]) -> io::Result<usize> { pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit) -> io::Result<()> {
send(self, socket.0, transmits) send(self, socket.0, transmit)
} }
pub fn recv( pub fn recv(
@@ -213,8 +213,8 @@ fn send(
#[allow(unused_variables)] // only used on Linux #[allow(unused_variables)] // only used on Linux
state: &UdpSocketState, state: &UdpSocketState,
io: SockRef<'_>, io: SockRef<'_>,
transmits: &[Transmit], transmit: &Transmit,
) -> io::Result<usize> { ) -> io::Result<()> {
#[allow(unused_mut)] // only mutable on FreeBSD #[allow(unused_mut)] // only mutable on FreeBSD
let mut encode_src_ip = true; let mut encode_src_ip = true;
#[cfg(target_os = "freebsd")] #[cfg(target_os = "freebsd")]
@@ -227,41 +227,22 @@ fn send(
} }
} }
} }
let mut msgs: [libc::mmsghdr; BATCH_SIZE] = unsafe { mem::zeroed() }; let mut msg_hdr: libc::msghdr = unsafe { mem::zeroed() };
let mut iovecs: [libc::iovec; BATCH_SIZE] = unsafe { mem::zeroed() }; let mut iovec: libc::iovec = unsafe { mem::zeroed() };
let mut cmsgs = [cmsg::Aligned([0u8; CMSG_LEN]); BATCH_SIZE]; let mut cmsgs = cmsg::Aligned([0u8; CMSG_LEN]);
// This assume_init looks a bit weird because one might think it let dst_addr = socket2::SockAddr::from(transmit.destination);
// assumes the SockAddr data to be initialized, but that call prepare_msg(
// refers to the whole array, which itself is made up of MaybeUninit transmit,
// containers. Their presence protects the SockAddr inside from &dst_addr,
// being assumed as initialized by the assume_init call. &mut msg_hdr,
// TODO: Replace this with uninit_array once it becomes MSRV-stable &mut iovec,
let mut addrs: [MaybeUninit<socket2::SockAddr>; BATCH_SIZE] = &mut cmsgs,
unsafe { MaybeUninit::uninit().assume_init() }; encode_src_ip,
for (i, transmit) in transmits.iter().enumerate().take(BATCH_SIZE) { state.sendmsg_einval(),
let dst_addr = unsafe { );
ptr::write(
addrs[i].as_mut_ptr(),
socket2::SockAddr::from(transmit.destination),
);
&*addrs[i].as_ptr()
};
prepare_msg(
transmit,
dst_addr,
&mut msgs[i].msg_hdr,
&mut iovecs[i],
&mut cmsgs[i],
encode_src_ip,
state.sendmsg_einval(),
);
}
let num_transmits = transmits.len().min(BATCH_SIZE);
loop { loop {
let n = unsafe { let n = unsafe { libc::sendmsg(io.as_raw_fd(), &msg_hdr, 0) };
sendmmsg_with_fallback(io.as_raw_fd(), msgs.as_mut_ptr(), num_transmits as _)
};
if n == -1 { if n == -1 {
let e = io::Error::last_os_error(); let e = io::Error::last_os_error();
match e.kind() { match e.kind() {
@@ -301,135 +282,57 @@ fn send(
// - EMSGSIZE is expected for MTU probes. Future work might be able to avoid // - EMSGSIZE is expected for MTU probes. Future work might be able to avoid
// these by automatically clamping the MTUD upper bound to the interface MTU. // these by automatically clamping the MTUD upper bound to the interface MTU.
if e.raw_os_error() != Some(libc::EMSGSIZE) { if e.raw_os_error() != Some(libc::EMSGSIZE) {
log_sendmsg_error(&state.last_send_error, e, &transmits[0]); log_sendmsg_error(&state.last_send_error, e, transmit);
} }
// The ERRORS section in https://man7.org/linux/man-pages/man2/sendmmsg.2.html return Ok(());
// describes that errors will only be returned if no message could be transmitted
// at all. Therefore drop the first (problematic) message,
// and retry the remaining ones.
return Ok(num_transmits.min(1));
} }
} }
} }
return Ok(n as usize); return Ok(());
} }
} }
#[cfg(any(target_os = "macos", target_os = "ios"))] #[cfg(any(target_os = "macos", target_os = "ios"))]
fn send(state: &UdpSocketState, io: SockRef<'_>, transmits: &[Transmit]) -> io::Result<usize> { fn send(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit) -> io::Result<()> {
let mut hdr: libc::msghdr = unsafe { mem::zeroed() }; let mut hdr: libc::msghdr = unsafe { mem::zeroed() };
let mut iov: libc::iovec = unsafe { mem::zeroed() }; let mut iov: libc::iovec = unsafe { mem::zeroed() };
let mut ctrl = cmsg::Aligned([0u8; CMSG_LEN]); let mut ctrl = cmsg::Aligned([0u8; CMSG_LEN]);
let mut sent = 0; let addr = socket2::SockAddr::from(transmit.destination);
prepare_msg(
while sent < transmits.len() { transmit,
let addr = socket2::SockAddr::from(transmits[sent].destination); &addr,
prepare_msg( &mut hdr,
&transmits[sent], &mut iov,
&addr, &mut ctrl,
&mut hdr, // Only tested on macOS and iOS
&mut iov, cfg!(target_os = "macos") || cfg!(target_os = "ios"),
&mut ctrl, state.sendmsg_einval(),
// Only tested on macOS and iOS );
cfg!(target_os = "macos") || cfg!(target_os = "ios"), let n = unsafe { libc::sendmsg(io.as_raw_fd(), &hdr, 0) };
state.sendmsg_einval(), if n == -1 {
); let e = io::Error::last_os_error();
let n = unsafe { libc::sendmsg(io.as_raw_fd(), &hdr, 0) }; match e.kind() {
if n == -1 { io::ErrorKind::Interrupted => {
let e = io::Error::last_os_error(); // Retry the transmission
match e.kind() { }
io::ErrorKind::Interrupted => { io::ErrorKind::WouldBlock => return Err(e),
// Retry the transmission _ => {
} // Other errors are ignored, since they will usually be handled
io::ErrorKind::WouldBlock if sent != 0 => return Ok(sent), // by higher level retransmits and timeouts.
io::ErrorKind::WouldBlock => return Err(e), // - PermissionDenied errors have been observed due to iptable rules.
_ => { // Those are not fatal errors, since the
// Other errors are ignored, since they will usually be handled // configuration can be dynamically changed.
// by higher level retransmits and timeouts. // - Destination unreachable errors have been observed for other
// - PermissionDenied errors have been observed due to iptable rules. // - EMSGSIZE is expected for MTU probes. Future work might be able to avoid
// Those are not fatal errors, since the // these by automatically clamping the MTUD upper bound to the interface MTU.
// configuration can be dynamically changed. if e.raw_os_error() != Some(libc::EMSGSIZE) {
// - Destination unreachable errors have been observed for other log_sendmsg_error(&state.last_send_error, e, transmit);
// - EMSGSIZE is expected for MTU probes. Future work might be able to avoid
// these by automatically clamping the MTUD upper bound to the interface MTU.
if e.raw_os_error() != Some(libc::EMSGSIZE) {
log_sendmsg_error(&state.last_send_error, e, &transmits[sent]);
}
sent += 1;
} }
} }
} else {
sent += 1;
} }
} }
Ok(sent) Ok(())
}
/// Implementation of `sendmmsg` with a fallback
/// to `sendmsg` if syscall is not available.
///
/// It uses [`libc::syscall`] instead of [`libc::sendmmsg`]
/// to avoid linking error on systems where libc does not contain `sendmmsg`.
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
unsafe fn sendmmsg_with_fallback(
sockfd: libc::c_int,
msgvec: *mut libc::mmsghdr,
vlen: libc::c_uint,
) -> libc::c_int {
let flags = 0;
#[cfg(not(target_os = "freebsd"))]
{
let ret = libc::syscall(libc::SYS_sendmmsg, sockfd, msgvec, vlen, flags) as libc::c_int;
if ret != -1 {
return ret;
}
}
// libc on FreeBSD implements `sendmmsg` as a high-level abstraction over `sendmsg`,
// thus `SYS_sendmmsg` constant and direct system call do not exist
#[cfg(target_os = "freebsd")]
{
let ret = libc::sendmmsg(sockfd, msgvec, vlen as usize, flags) as libc::c_int;
if ret != -1 {
return ret;
}
}
let e = io::Error::last_os_error();
match e.raw_os_error() {
Some(libc::ENOSYS) => {
// Fallback to `sendmsg`.
sendmmsg_fallback(sockfd, msgvec, vlen)
}
_ => -1,
}
}
/// Fallback implementation of `sendmmsg` using `sendmsg`
/// for systems which do not support `sendmmsg`
/// such as Linux <3.0.
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
unsafe fn sendmmsg_fallback(
sockfd: libc::c_int,
msgvec: *mut libc::mmsghdr,
vlen: libc::c_uint,
) -> libc::c_int {
let flags = 0;
if vlen == 0 {
return 0;
}
let n = libc::sendmsg(sockfd, &(*msgvec).msg_hdr, flags);
if n == -1 {
-1
} else {
// type of `msg_len` field differs on Linux and FreeBSD,
// it is up to the compiler to infer and cast `n` to correct type
(*msgvec).msg_len = n as _;
1
}
} }
#[cfg(not(any(target_os = "macos", target_os = "ios")))] #[cfg(not(any(target_os = "macos", target_os = "ios")))]
quinn-udp/src/windows.rs
+94 -107
View File
@@ -122,122 +122,109 @@ impl UdpSocketState {
}) })
} }
pub fn send(&self, socket: UdpSockRef<'_>, transmits: &[Transmit]) -> io::Result<usize> { pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit) -> io::Result<()> {
let mut sent = 0; // we cannot use [`socket2::sendmsg()`] and [`socket2::MsgHdr`] as we do not have access
for transmit in transmits { // to the inner field which holds the WSAMSG
// we cannot use [`socket2::sendmsg()`] and [`socket2::MsgHdr`] as we do not have access let mut ctrl_buf = cmsg::Aligned([0; CMSG_LEN]);
// to the inner field which holds the WSAMSG let daddr = socket2::SockAddr::from(transmit.destination);
let mut ctrl_buf = cmsg::Aligned([0; CMSG_LEN]);
let daddr = socket2::SockAddr::from(transmit.destination);
let mut data = WinSock::WSABUF { let mut data = WinSock::WSABUF {
buf: transmit.contents.as_ptr() as *mut _, buf: transmit.contents.as_ptr() as *mut _,
len: transmit.contents.len() as _, len: transmit.contents.len() as _,
}; };
let ctrl = WinSock::WSABUF { let ctrl = WinSock::WSABUF {
buf: ctrl_buf.0.as_mut_ptr(), buf: ctrl_buf.0.as_mut_ptr(),
len: ctrl_buf.0.len() as _, len: ctrl_buf.0.len() as _,
}; };
let mut wsa_msg = WinSock::WSAMSG { let mut wsa_msg = WinSock::WSAMSG {
name: daddr.as_ptr() as *mut _, name: daddr.as_ptr() as *mut _,
namelen: daddr.len(), namelen: daddr.len(),
lpBuffers: &mut data, lpBuffers: &mut data,
Control: ctrl, Control: ctrl,
dwBufferCount: 1, dwBufferCount: 1,
dwFlags: 0, dwFlags: 0,
}; };
// Add control messages (ECN and PKTINFO) // Add control messages (ECN and PKTINFO)
let mut encoder = unsafe { cmsg::Encoder::new(&mut wsa_msg) }; let mut encoder = unsafe { cmsg::Encoder::new(&mut wsa_msg) };
if let Some(ip) = transmit.src_ip { if let Some(ip) = transmit.src_ip {
let ip = std::net::SocketAddr::new(ip, 0); let ip = std::net::SocketAddr::new(ip, 0);
let ip = socket2::SockAddr::from(ip); let ip = socket2::SockAddr::from(ip);
match ip.family() { match ip.family() {
WinSock::AF_INET => { WinSock::AF_INET => {
let src_ip = let src_ip = unsafe { ptr::read(ip.as_ptr() as *const WinSock::SOCKADDR_IN) };
unsafe { ptr::read(ip.as_ptr() as *const WinSock::SOCKADDR_IN) }; let pktinfo = WinSock::IN_PKTINFO {
let pktinfo = WinSock::IN_PKTINFO { ipi_addr: src_ip.sin_addr,
ipi_addr: src_ip.sin_addr, ipi_ifindex: 0,
ipi_ifindex: 0, };
}; encoder.push(WinSock::IPPROTO_IP, WinSock::IP_PKTINFO, pktinfo);
encoder.push(WinSock::IPPROTO_IP, WinSock::IP_PKTINFO, pktinfo);
}
WinSock::AF_INET6 => {
let src_ip =
unsafe { ptr::read(ip.as_ptr() as *const WinSock::SOCKADDR_IN6) };
let pktinfo = WinSock::IN6_PKTINFO {
ipi6_addr: src_ip.sin6_addr,
ipi6_ifindex: unsafe { src_ip.Anonymous.sin6_scope_id },
};
encoder.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_PKTINFO, pktinfo);
}
_ => {
return Err(io::Error::from(io::ErrorKind::InvalidInput));
}
} }
} WinSock::AF_INET6 => {
let src_ip = unsafe { ptr::read(ip.as_ptr() as *const WinSock::SOCKADDR_IN6) };
// ECN is a C integer https://learn.microsoft.com/en-us/windows/win32/winsock/winsock-ecn let pktinfo = WinSock::IN6_PKTINFO {
let ecn = transmit.ecn.map_or(0, |x| x as c_int); ipi6_addr: src_ip.sin6_addr,
// True for IPv4 or IPv4-Mapped IPv6 ipi6_ifindex: unsafe { src_ip.Anonymous.sin6_scope_id },
let is_ipv4 = transmit.destination.is_ipv4() };
|| matches!(transmit.destination.ip(), IpAddr::V6(addr) if addr.to_ipv4_mapped().is_some()); encoder.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_PKTINFO, pktinfo);
if is_ipv4 { }
encoder.push(WinSock::IPPROTO_IP, WinSock::IP_ECN, ecn); _ => {
} else { return Err(io::Error::from(io::ErrorKind::InvalidInput));
encoder.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_ECN, ecn);
}
// Segment size is a u32 https://learn.microsoft.com/en-us/windows/win32/api/ws2tcpip/nf-ws2tcpip-wsasetudpsendmessagesize
if let Some(segment_size) = transmit.segment_size {
encoder.push(
WinSock::IPPROTO_UDP,
WinSock::UDP_SEND_MSG_SIZE,
segment_size as u32,
);
}
encoder.finish();
let mut len = 0;
let rc = unsafe {
WinSock::WSASendMsg(
socket.0.as_raw_socket() as usize,
&wsa_msg,
0,
&mut len,
ptr::null_mut(),
None,
)
};
if rc == 0 {
sent += 1;
} else if sent != 0 {
// We need to report that some packets were sent in this case, so we rely on
// errors being either harmlessly transient (in the case of WouldBlock) or
// recurring on the next call.
return Ok(sent);
} else {
let e = io::Error::last_os_error();
if e.kind() == io::ErrorKind::WouldBlock {
return Err(e);
} }
// Other errors are ignored, since they will usually be handled
// by higher level retransmits and timeouts.
// - PermissionDenied errors have been observed due to iptable rules.
// Those are not fatal errors, since the
// configuration can be dynamically changed.
// - Destination unreachable errors have been observed for other
log_sendmsg_error(&self.last_send_error, e, transmit);
sent += 1;
} }
} }
Ok(sent)
// ECN is a C integer https://learn.microsoft.com/en-us/windows/win32/winsock/winsock-ecn
let ecn = transmit.ecn.map_or(0, |x| x as c_int);
// True for IPv4 or IPv4-Mapped IPv6
let is_ipv4 = transmit.destination.is_ipv4()
|| matches!(transmit.destination.ip(), IpAddr::V6(addr) if addr.to_ipv4_mapped().is_some());
if is_ipv4 {
encoder.push(WinSock::IPPROTO_IP, WinSock::IP_ECN, ecn);
} else {
encoder.push(WinSock::IPPROTO_IPV6, WinSock::IPV6_ECN, ecn);
}
// Segment size is a u32 https://learn.microsoft.com/en-us/windows/win32/api/ws2tcpip/nf-ws2tcpip-wsasetudpsendmessagesize
if let Some(segment_size) = transmit.segment_size {
encoder.push(
WinSock::IPPROTO_UDP,
WinSock::UDP_SEND_MSG_SIZE,
segment_size as u32,
);
}
encoder.finish();
let mut len = 0;
let rc = unsafe {
WinSock::WSASendMsg(
socket.0.as_raw_socket() as usize,
&wsa_msg,
0,
&mut len,
ptr::null_mut(),
None,
)
};
if rc != 0 {
let e = io::Error::last_os_error();
if e.kind() == io::ErrorKind::WouldBlock {
return Err(e);
}
// Other errors are ignored, since they will usually be handled
// by higher level retransmits and timeouts.
// - PermissionDenied errors have been observed due to iptable rules.
// Those are not fatal errors, since the
// configuration can be dynamically changed.
// - Destination unreachable errors have been observed for other
log_sendmsg_error(&self.last_send_error, e, transmit);
}
Ok(())
} }
pub fn recv( pub fn recv(
quinn-udp/tests/tests.rs
+1 -3
View File
@@ -164,9 +164,7 @@ fn test_send_recv(send: &Socket, recv: &Socket, transmit: Transmit) {
// Reverse non-blocking flag set by `UdpSocketState` to make the test non-racy // Reverse non-blocking flag set by `UdpSocketState` to make the test non-racy
recv.set_nonblocking(false).unwrap(); recv.set_nonblocking(false).unwrap();
send_state send_state.send(send.into(), &transmit).unwrap();
.send((&send).into(), slice::from_ref(&transmit))
.unwrap();
let mut buf = [0; u16::MAX as usize]; let mut buf = [0; u16::MAX as usize];
let mut meta = RecvMeta::default(); let mut meta = RecvMeta::default();
quinn/src/connection.rs
+2 -2
View File
@@ -985,8 +985,8 @@ impl State {
return Ok(false); return Ok(false);
} }
let retry = match self.socket.try_send(std::slice::from_ref(&t)) { let retry = match self.socket.try_send(&t) {
Ok(n) => n == 0, Ok(()) => false,
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => true, Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => true,
Err(e) => return Err(e), Err(e) => return Err(e),
}; };
quinn/src/endpoint.rs
+2 -2
View File
@@ -515,10 +515,10 @@ fn respond(transmit: proto::Transmit, response_buffer: &mut BytesMut, socket: &d
// lost due to congestion further along the link, which // lost due to congestion further along the link, which
// similarly relies on peer retries for recovery. // similarly relies on peer retries for recovery.
let contents_len = transmit.size; let contents_len = transmit.size;
_ = socket.try_send(&[udp_transmit( _ = socket.try_send(&udp_transmit(
transmit, transmit,
response_buffer.split_to(contents_len).freeze(), response_buffer.split_to(contents_len).freeze(),
)]); ));
} }
#[inline] #[inline]
quinn/src/runtime.rs
+1 -1
View File
@@ -47,7 +47,7 @@ pub trait AsyncUdpSocket: Send + Sync + Debug + 'static {
/// ///
/// If this returns [`io::ErrorKind::WouldBlock`], [`UdpPoller::poll_writable`] must be called /// If this returns [`io::ErrorKind::WouldBlock`], [`UdpPoller::poll_writable`] must be called
/// to register the calling task to be woken when a send should be attempted again. /// to register the calling task to be woken when a send should be attempted again.
fn try_send(&self, transmits: &[Transmit]) -> Result<usize, io::Error>; fn try_send(&self, transmit: &Transmit) -> io::Result<()>;
/// Receive UDP datagrams, or register to be woken if receiving may succeed in the future /// Receive UDP datagrams, or register to be woken if receiving may succeed in the future
fn poll_recv( fn poll_recv(
quinn/src/runtime/async_io.rs
+2 -2
View File
@@ -102,8 +102,8 @@ impl AsyncUdpSocket for UdpSocket {
})) }))
} }
fn try_send(&self, transmits: &[udp::Transmit]) -> io::Result<usize> { fn try_send(&self, transmit: &udp::Transmit) -> io::Result<()> {
self.inner.send((&self.io).into(), transmits) self.inner.send((&self.io).into(), transmit)
} }
fn poll_recv( fn poll_recv(
quinn/src/runtime/tokio.rs
+2 -2
View File
@@ -58,9 +58,9 @@ impl AsyncUdpSocket for UdpSocket {
})) }))
} }
fn try_send(&self, transmits: &[udp::Transmit]) -> io::Result<usize> { fn try_send(&self, transmit: &udp::Transmit) -> io::Result<()> {
self.io.try_io(Interest::WRITABLE, || { self.io.try_io(Interest::WRITABLE, || {
self.inner.send((&self.io).into(), transmits) self.inner.send((&self.io).into(), transmit)
}) })
} }