gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-09-24 12:13:05 +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> {
let mut sent = 0;
for transmit in transmits {
match socket.0.send_to(
&transmit.contents,
&socket2::SockAddr::from(transmit.destination),
) {
Ok(_) => {
sent += 1;
}
// 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;
}
}
pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit) -> io::Result<()> {
let Err(e) = socket.0.send_to(
&transmit.contents,
&socket2::SockAddr::from(transmit.destination),
) else {
return Ok(());
};
if e.kind() == io::ErrorKind::WouldBlock {
return Err(e);
}
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(
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> {
send(self, socket.0, transmits)
pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit) -> io::Result<()> {
send(self, socket.0, transmit)
}
pub fn recv(
@@ -213,8 +213,8 @@ fn send(
#[allow(unused_variables)] // only used on Linux
state: &UdpSocketState,
io: SockRef<'_>,
transmits: &[Transmit],
) -> io::Result<usize> {
transmit: &Transmit,
) -> io::Result<()> {
#[allow(unused_mut)] // only mutable on FreeBSD
let mut encode_src_ip = true;
#[cfg(target_os = "freebsd")]
@@ -227,41 +227,22 @@ fn send(
}
}
}
let mut msgs: [libc::mmsghdr; BATCH_SIZE] = unsafe { mem::zeroed() };
let mut iovecs: [libc::iovec; BATCH_SIZE] = unsafe { mem::zeroed() };
let mut cmsgs = [cmsg::Aligned([0u8; CMSG_LEN]); BATCH_SIZE];
// This assume_init looks a bit weird because one might think it
// assumes the SockAddr data to be initialized, but that call
// refers to the whole array, which itself is made up of MaybeUninit
// containers. Their presence protects the SockAddr inside from
// being assumed as initialized by the assume_init call.
// TODO: Replace this with uninit_array once it becomes MSRV-stable
let mut addrs: [MaybeUninit<socket2::SockAddr>; BATCH_SIZE] =
unsafe { MaybeUninit::uninit().assume_init() };
for (i, transmit) in transmits.iter().enumerate().take(BATCH_SIZE) {
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);
let mut msg_hdr: libc::msghdr = unsafe { mem::zeroed() };
let mut iovec: libc::iovec = unsafe { mem::zeroed() };
let mut cmsgs = cmsg::Aligned([0u8; CMSG_LEN]);
let dst_addr = socket2::SockAddr::from(transmit.destination);
prepare_msg(
transmit,
&dst_addr,
&mut msg_hdr,
&mut iovec,
&mut cmsgs,
encode_src_ip,
state.sendmsg_einval(),
);
loop {
let n = unsafe {
sendmmsg_with_fallback(io.as_raw_fd(), msgs.as_mut_ptr(), num_transmits as _)
};
let n = unsafe { libc::sendmsg(io.as_raw_fd(), &msg_hdr, 0) };
if n == -1 {
let e = io::Error::last_os_error();
match e.kind() {
@@ -301,135 +282,57 @@ fn send(
// - 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[0]);
log_sendmsg_error(&state.last_send_error, e, transmit);
}
// The ERRORS section in https://man7.org/linux/man-pages/man2/sendmmsg.2.html
// 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(());
}
}
}
return Ok(n as usize);
return Ok(());
}
}
#[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 iov: libc::iovec = unsafe { mem::zeroed() };
let mut ctrl = cmsg::Aligned([0u8; CMSG_LEN]);
let mut sent = 0;
while sent < transmits.len() {
let addr = socket2::SockAddr::from(transmits[sent].destination);
prepare_msg(
&transmits[sent],
&addr,
&mut hdr,
&mut iov,
&mut ctrl,
// Only tested on macOS and iOS
cfg!(target_os = "macos") || cfg!(target_os = "ios"),
state.sendmsg_einval(),
);
let n = unsafe { libc::sendmsg(io.as_raw_fd(), &hdr, 0) };
if n == -1 {
let e = io::Error::last_os_error();
match e.kind() {
io::ErrorKind::Interrupted => {
// Retry the transmission
}
io::ErrorKind::WouldBlock if sent != 0 => return Ok(sent),
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
// - 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;
let addr = socket2::SockAddr::from(transmit.destination);
prepare_msg(
transmit,
&addr,
&mut hdr,
&mut iov,
&mut ctrl,
// Only tested on macOS and iOS
cfg!(target_os = "macos") || cfg!(target_os = "ios"),
state.sendmsg_einval(),
);
let n = unsafe { libc::sendmsg(io.as_raw_fd(), &hdr, 0) };
if n == -1 {
let e = io::Error::last_os_error();
match e.kind() {
io::ErrorKind::Interrupted => {
// Retry the transmission
}
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
// - 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, transmit);
}
}
} else {
sent += 1;
}
}
Ok(sent)
}
/// 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
}
Ok(())
}
#[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> {
let mut sent = 0;
for transmit in transmits {
// we cannot use [`socket2::sendmsg()`] and [`socket2::MsgHdr`] as we do not have access
// to the inner field which holds the WSAMSG
let mut ctrl_buf = cmsg::Aligned([0; CMSG_LEN]);
let daddr = socket2::SockAddr::from(transmit.destination);
pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit) -> io::Result<()> {
// we cannot use [`socket2::sendmsg()`] and [`socket2::MsgHdr`] as we do not have access
// to the inner field which holds the WSAMSG
let mut ctrl_buf = cmsg::Aligned([0; CMSG_LEN]);
let daddr = socket2::SockAddr::from(transmit.destination);
let mut data = WinSock::WSABUF {
buf: transmit.contents.as_ptr() as *mut _,
len: transmit.contents.len() as _,
};
let mut data = WinSock::WSABUF {
buf: transmit.contents.as_ptr() as *mut _,
len: transmit.contents.len() as _,
};
let ctrl = WinSock::WSABUF {
buf: ctrl_buf.0.as_mut_ptr(),
len: ctrl_buf.0.len() as _,
};
let ctrl = WinSock::WSABUF {
buf: ctrl_buf.0.as_mut_ptr(),
len: ctrl_buf.0.len() as _,
};
let mut wsa_msg = WinSock::WSAMSG {
name: daddr.as_ptr() as *mut _,
namelen: daddr.len(),
lpBuffers: &mut data,
Control: ctrl,
dwBufferCount: 1,
dwFlags: 0,
};
let mut wsa_msg = WinSock::WSAMSG {
name: daddr.as_ptr() as *mut _,
namelen: daddr.len(),
lpBuffers: &mut data,
Control: ctrl,
dwBufferCount: 1,
dwFlags: 0,
};
// Add control messages (ECN and PKTINFO)
let mut encoder = unsafe { cmsg::Encoder::new(&mut wsa_msg) };
// Add control messages (ECN and PKTINFO)
let mut encoder = unsafe { cmsg::Encoder::new(&mut wsa_msg) };
if let Some(ip) = transmit.src_ip {
let ip = std::net::SocketAddr::new(ip, 0);
let ip = socket2::SockAddr::from(ip);
match ip.family() {
WinSock::AF_INET => {
let src_ip =
unsafe { ptr::read(ip.as_ptr() as *const WinSock::SOCKADDR_IN) };
let pktinfo = WinSock::IN_PKTINFO {
ipi_addr: src_ip.sin_addr,
ipi_ifindex: 0,
};
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));
}
if let Some(ip) = transmit.src_ip {
let ip = std::net::SocketAddr::new(ip, 0);
let ip = socket2::SockAddr::from(ip);
match ip.family() {
WinSock::AF_INET => {
let src_ip = unsafe { ptr::read(ip.as_ptr() as *const WinSock::SOCKADDR_IN) };
let pktinfo = WinSock::IN_PKTINFO {
ipi_addr: src_ip.sin_addr,
ipi_ifindex: 0,
};
encoder.push(WinSock::IPPROTO_IP, WinSock::IP_PKTINFO, pktinfo);
}
}
// 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 {
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);
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));
}
// 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(
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
recv.set_nonblocking(false).unwrap();
send_state
.send((&send).into(), slice::from_ref(&transmit))
.unwrap();
send_state.send(send.into(), &transmit).unwrap();
let mut buf = [0; u16::MAX as usize];
let mut meta = RecvMeta::default();
quinn/src/connection.rs
+2 -2
View File
@@ -985,8 +985,8 @@ impl State {
return Ok(false);
}
let retry = match self.socket.try_send(std::slice::from_ref(&t)) {
Ok(n) => n == 0,
let retry = match self.socket.try_send(&t) {
Ok(()) => false,
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => true,
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
// similarly relies on peer retries for recovery.
let contents_len = transmit.size;
_ = socket.try_send(&[udp_transmit(
_ = socket.try_send(&udp_transmit(
transmit,
response_buffer.split_to(contents_len).freeze(),
)]);
));
}
#[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
/// 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
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> {
self.inner.send((&self.io).into(), transmits)
fn try_send(&self, transmit: &udp::Transmit) -> io::Result<()> {
self.inner.send((&self.io).into(), transmit)
}
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.inner.send((&self.io).into(), transmits)
self.inner.send((&self.io).into(), transmit)
})
}