diff --git a/quinn-udp/src/fallback.rs b/quinn-udp/src/fallback.rs index 4e8d608c4..b5663cdea 100644 --- a/quinn-udp/src/fallback.rs +++ b/quinn-udp/src/fallback.rs @@ -24,37 +24,25 @@ impl UdpSocketState { }) } - pub fn send(&self, socket: UdpSockRef<'_>, transmits: &[Transmit]) -> io::Result { - 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( diff --git a/quinn-udp/src/unix.rs b/quinn-udp/src/unix.rs index b21e58f71..bebd1b388 100644 --- a/quinn-udp/src/unix.rs +++ b/quinn-udp/src/unix.rs @@ -156,8 +156,8 @@ impl UdpSocketState { }) } - pub fn send(&self, socket: UdpSockRef<'_>, transmits: &[Transmit]) -> io::Result { - 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 { + 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; 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 { +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")))] diff --git a/quinn-udp/src/windows.rs b/quinn-udp/src/windows.rs index d448c827b..29743b95d 100644 --- a/quinn-udp/src/windows.rs +++ b/quinn-udp/src/windows.rs @@ -122,122 +122,109 @@ impl UdpSocketState { }) } - pub fn send(&self, socket: UdpSockRef<'_>, transmits: &[Transmit]) -> io::Result { - 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( diff --git a/quinn-udp/tests/tests.rs b/quinn-udp/tests/tests.rs index 817f6c20c..a6c6c9668 100644 --- a/quinn-udp/tests/tests.rs +++ b/quinn-udp/tests/tests.rs @@ -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(); diff --git a/quinn/src/connection.rs b/quinn/src/connection.rs index 430671f4e..4b8c3a454 100644 --- a/quinn/src/connection.rs +++ b/quinn/src/connection.rs @@ -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), }; diff --git a/quinn/src/endpoint.rs b/quinn/src/endpoint.rs index 589eef2ca..27a76cb58 100644 --- a/quinn/src/endpoint.rs +++ b/quinn/src/endpoint.rs @@ -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] diff --git a/quinn/src/runtime.rs b/quinn/src/runtime.rs index c6fe6de3f..823c617f7 100644 --- a/quinn/src/runtime.rs +++ b/quinn/src/runtime.rs @@ -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; + 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( diff --git a/quinn/src/runtime/async_io.rs b/quinn/src/runtime/async_io.rs index e42fe3a26..3eb2c4f21 100644 --- a/quinn/src/runtime/async_io.rs +++ b/quinn/src/runtime/async_io.rs @@ -102,8 +102,8 @@ impl AsyncUdpSocket for UdpSocket { })) } - fn try_send(&self, transmits: &[udp::Transmit]) -> io::Result { - 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( diff --git a/quinn/src/runtime/tokio.rs b/quinn/src/runtime/tokio.rs index 5eb4d26a5..42f63b214 100644 --- a/quinn/src/runtime/tokio.rs +++ b/quinn/src/runtime/tokio.rs @@ -58,9 +58,9 @@ impl AsyncUdpSocket for UdpSocket { })) } - fn try_send(&self, transmits: &[udp::Transmit]) -> io::Result { + 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) }) }