From 6c54823364a34e95ed5c0b98f751ef47498c21ed Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 00:05:56 +0700 Subject: [PATCH 1/2] Add UDP sockets and IPv6 scope IDs --- src/net.rs | 367 +++++++++++++++++++++++++++++++++++++++++- src/reactor.rs | 18 +-- src/reactor/driver.rs | 62 +++++-- src/reactor/socket.rs | 272 ++++++++++++++++++++++++++++++- 4 files changed, 691 insertions(+), 28 deletions(-) diff --git a/src/net.rs b/src/net.rs index 002958e..616f0f9 100644 --- a/src/net.rs +++ b/src/net.rs @@ -1,4 +1,4 @@ -//! Non-blocking TCP over the reactor. +//! Non-blocking TCP and UDP over the reactor. //! //! # No `std::net` //! @@ -595,3 +595,368 @@ impl crate::io::Stream for TcpStream { Self::write_all(self, buffer).await } } + +pub use crate::reactor::socket::Datagram; + +/// An owned UDP socket driven by a nagoya reactor. +/// +/// Each receive consumes one datagram. Successful operations retain readiness +/// until a subsequent syscall returns `WouldBlock`, including receives of +/// empty datagrams. UDP must not use TCP's FIONREAD optimisation: zero can +/// mean that an empty datagram is queued. +/// +/// Operation futures borrow the socket exclusively, so there is only one +/// waiter per direction. A transport task can also use the poll methods to +/// drive sending and receiving together. It must call [`Self::cancel_recv`] +/// or [`Self::cancel_send`] when abandoning a pending poll operation. +/// +/// Dropping a pending receive or send future releases its waiter. A pending +/// receive has consumed nothing, and a pending send has sent nothing. +/// Dropping the socket deregisters it before closing the descriptor. +#[derive(Debug)] +pub struct UdpSocket { + // Fields drop in declaration order. The registration must go first. + registration: Registration, + inner: crate::reactor::socket::UdpSocket, +} + +impl UdpSocket { + /// Bind and register an IPv4 or IPv6 UDP socket. + /// + /// The bind is performed when this future is polled. It never waits for + /// network traffic. `SO_REUSEADDR` is disabled by default. + pub async fn bind(addr: Addr, handle: &Handle) -> Result { + Self::bind_with_reuse(addr, handle, false).await + } + + /// Bind with an explicit `SO_REUSEADDR` setting applied before binding. + /// + /// IPv6 sockets are IPv6-only. Own a separate IPv4 socket when both + /// families are needed. A port of zero selects an ephemeral port. + pub async fn bind_with_reuse(addr: Addr, handle: &Handle, reuse_addr: bool) -> Result { + let socket = crate::reactor::socket::UdpSocket::bind(addr, reuse_addr)?; + Self::from_socket(socket, handle) + } + + /// Register an owned non-blocking UDP socket. + pub fn from_socket(socket: crate::reactor::socket::UdpSocket, handle: &Handle) -> Result { + let registration = handle.register(socket.raw(), Interest::BOTH)?; + Ok(Self { + registration, + inner: socket, + }) + } + + /// Select a peer, returning the owned socket once the operation completes. + /// + /// UDP has no handshake. The kernel filters incoming datagrams by the + /// selected peer. The peer must have the same address family as the socket. + /// + /// This future owns the socket: cancelling an in-flight connect closes + /// it, rather than exposing a socket whose peer may have changed. + pub async fn connect(self, addr: Addr) -> Result { + match self.inner.connect(addr) { + Ok(()) => { + self.registration.mark_writable(); + return Ok(self); + } + Err(error) + if error.0 == libc::EINPROGRESS + || error.0 == libc::EALREADY + || error.interrupted() => {} + Err(error) if error.0 == libc::EISCONN => { + if self.inner.peer_addr()? == addr { + return Ok(self); + } + return Err(error); + } + Err(error) => return Err(error), + } + + core::future::poll_fn(|cx| { + if !self.registration.take_writable_or_park(cx.waker()) { + return Poll::Pending; + } + loop { + match self.inner.connect_error() { + Ok(()) => break, + Err(error) if error.interrupted() => continue, + Err(error) => return Poll::Ready(Err(error)), + } + } + match self.inner.peer_addr() { + Ok(peer) if peer == addr => { + self.registration.mark_writable(); + Poll::Ready(Ok(())) + } + Ok(_) => Poll::Ready(Err(crate::reactor::error::Errno(libc::EINVAL))), + Err(error) if error.0 == libc::ENOTCONN || error.interrupted() => { + self.registration.poll_writable(cx.waker()); + Poll::Pending + } + Err(error) => Poll::Ready(Err(error)), + } + }) + .await?; + Ok(self) + } + + /// This socket's bound address and actual local port. + pub fn local_addr(&self) -> Result { + self.inner.local_addr() + } + + /// The configured peer, or `ENOTCONN` when unconnected. + pub fn peer_addr(&self) -> Result { + self.inner.peer_addr() + } + + /// Receive one datagram into an initialised slice. + /// + /// An empty buffer still consumes one datagram and reports truncation + /// when that datagram is nonempty. Zero bytes never mean end of stream. + pub fn recv_from<'a>(&'a mut self, buffer: &'a mut [u8]) -> RecvFrom<'a> { + RecvFrom { + socket: self, + buffer, + } + } + + /// Append one datagram directly into a `BytesMut`'s spare capacity. + /// + /// No staging copy or initialisation of the spare capacity is needed. + /// The returned length counts only the appended bytes. The caller can + /// freeze the buffer to share immutable packet ownership without copying. + /// + /// An empty spare region returns `ENOBUFS` without consuming a datagram. + pub fn recv_from_buf<'a>(&'a mut self, buffer: &'a mut bytes::BytesMut) -> RecvFromBuf<'a> { + RecvFromBuf { + socket: self, + buffer, + } + } + + /// Send one datagram without copying or retaining the supplied bytes. + /// + /// This also sends empty datagrams. An oversized packet is an error; + /// no suffix is retried as a separate datagram. + pub fn send_to<'a>(&'a mut self, buffer: &'a [u8], destination: Addr) -> SendTo<'a> { + SendTo { + socket: self, + buffer, + destination: Some(destination), + } + } + + /// Send one datagram to the configured peer. + /// + /// Returns the kernel's error if no peer has been configured. + pub fn send<'a>(&'a mut self, buffer: &'a [u8]) -> SendTo<'a> { + SendTo { + socket: self, + buffer, + destination: None, + } + } + + /// Poll for one datagram, parking the reader only after `WouldBlock`. + pub fn poll_recv_from( + &mut self, + cx: &mut Context<'_>, + buffer: &mut [u8], + ) -> Poll> { + // SAFETY: the slice is live and writable for its whole length. + unsafe { self.poll_recv_into(cx, buffer.as_mut_ptr(), buffer.len()) } + } + + /// Poll a receive directly into uninitialised spare buffer capacity. + pub fn poll_recv_from_buf( + &mut self, + cx: &mut Context<'_>, + buffer: &mut bytes::BytesMut, + ) -> Poll> { + let filled = buffer.len(); + let spare = buffer.spare_capacity_mut(); + if spare.is_empty() { + self.cancel_recv(); + return Poll::Ready(Err(crate::reactor::error::Errno(libc::ENOBUFS))); + } + let pointer = spare.as_mut_ptr().cast::(); + let capacity = spare.len(); + // SAFETY: the spare region is allocated, writable and disjoint from + // the buffer's initialised prefix. The syscall reports copied bytes. + let result = unsafe { self.poll_recv_into(cx, pointer, capacity) }; + if let Poll::Ready(Ok(datagram)) = &result { + // SAFETY: precisely datagram.len bytes in the spare region were + // initialised, and the low-level socket clamps that to capacity. + unsafe { buffer.set_len(filled + datagram.len) }; + } + result + } + + unsafe fn poll_recv_into( + &mut self, + cx: &mut Context<'_>, + pointer: *mut u8, + len: usize, + ) -> Poll> { + if !self.registration.take_readable_or_park(cx.waker()) { + return Poll::Pending; + } + loop { + match self.inner.recv_from(pointer, len) { + Ok(datagram) => { + // A single datagram never proves the receive queue empty, + // regardless of its size or whether it was truncated. + self.registration.mark_readable(); + self.cancel_recv(); + return Poll::Ready(Ok(datagram)); + } + Err(error) if error.would_block() => { + self.registration.poll_readable(cx.waker()); + return Poll::Pending; + } + Err(error) if error.interrupted() => continue, + Err(error) => { + // An asynchronous UDP error need not drain queued data. + self.registration.mark_readable(); + self.cancel_recv(); + return Poll::Ready(Err(error)); + } + } + } + } + + /// Poll sending one datagram to an explicit destination. + pub fn poll_send_to( + &mut self, + cx: &mut Context<'_>, + buffer: &[u8], + destination: Addr, + ) -> Poll> { + self.poll_send_datagram(cx, buffer, Some(destination)) + } + + /// Poll sending one datagram to the configured peer. + pub fn poll_send(&mut self, cx: &mut Context<'_>, buffer: &[u8]) -> Poll> { + self.poll_send_datagram(cx, buffer, None) + } + + fn poll_send_datagram( + &mut self, + cx: &mut Context<'_>, + buffer: &[u8], + destination: Option, + ) -> Poll> { + if !self.registration.take_writable_or_park(cx.waker()) { + return Poll::Pending; + } + loop { + let result = match destination { + Some(destination) => self.inner.send_to(buffer, destination), + None => self.inner.send(buffer), + }; + match result { + Ok(sent) => { + self.registration.mark_writable(); + self.cancel_send(); + return Poll::Ready(Ok(sent)); + } + Err(error) if error.would_block() => { + self.registration.poll_writable(cx.waker()); + return Poll::Pending; + } + Err(error) if error.interrupted() => continue, + Err(error) => { + // An invalid destination or oversized packet does not + // establish that the socket's send buffer is full. + self.registration.mark_writable(); + self.cancel_send(); + return Poll::Ready(Err(error)); + } + } + } + } + + /// Release the waiter for an abandoned direct receive poll. + /// + /// Cached readiness is preserved. Receive futures call this on drop. + pub fn cancel_recv(&mut self) { + self.registration.cancel_read_wait(); + } + + /// Release the waiter for an abandoned direct send poll. + /// + /// Cached readiness is preserved. Send futures call this on drop. + pub fn cancel_send(&mut self) { + self.registration.cancel_write_wait(); + } +} + +/// The cancellation-safe future returned by [`UdpSocket::recv_from`]. +#[derive(Debug)] +pub struct RecvFrom<'a> { + socket: &'a mut UdpSocket, + buffer: &'a mut [u8], +} + +impl core::future::Future for RecvFrom<'_> { + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + this.socket.poll_recv_from(cx, this.buffer) + } +} + +impl Drop for RecvFrom<'_> { + fn drop(&mut self) { + self.socket.cancel_recv(); + } +} + +/// The cancellation-safe future returned by [`UdpSocket::recv_from_buf`]. +#[derive(Debug)] +pub struct RecvFromBuf<'a> { + socket: &'a mut UdpSocket, + buffer: &'a mut bytes::BytesMut, +} + +impl core::future::Future for RecvFromBuf<'_> { + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + this.socket.poll_recv_from_buf(cx, this.buffer) + } +} + +impl Drop for RecvFromBuf<'_> { + fn drop(&mut self) { + self.socket.cancel_recv(); + } +} + +/// The cancellation-safe future returned by UDP send operations. +#[derive(Debug)] +pub struct SendTo<'a> { + socket: &'a mut UdpSocket, + buffer: &'a [u8], + destination: Option, +} + +impl core::future::Future for SendTo<'_> { + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + this.socket + .poll_send_datagram(cx, this.buffer, this.destination) + } +} + +impl Drop for SendTo<'_> { + fn drop(&mut self) { + self.socket.cancel_send(); + } +} diff --git a/src/reactor.rs b/src/reactor.rs index a1a07f2..97915bb 100644 --- a/src/reactor.rs +++ b/src/reactor.rs @@ -28,18 +28,18 @@ //! //! The timer integration is the clearest case. The next deadline is cached, so //! a socket delivering ten thousand events a second never touches the timer -//! heap's lock to be told that nothing is due. Socket traffic and timer work -//! are independent: neither makes the other do any work. `select!` cannot -//! express that, because it folds every branch into one poll set. +//! heap's lock to be told, almost always, that nothing is due. Socket traffic +//! and timer work are independent: neither makes the other do any work. +//! `select!` cannot express that, because it folds every branch into one poll set. //! //! # What a caller gets //! //! [`Reactor`] owns the poller and a thread; [`Handle`] registers descriptors //! with it; [`Registration`] parks a waker for one descriptor's readability or -//! writability. [`TcpStream`] and [`TcpListener`] are the ordinary sockets -//! built on that, and [`TcpStream`] implements [`crate::io::Stream`], which -//! this crate declared without providing an implementation. [`resolve`] turns -//! a hostname into those addresses, and [`connect_any`] tries them in order. +//! writability. [`TcpStream`], [`TcpListener`] and [`UdpSocket`] are the ordinary +//! sockets built on that, and [`TcpStream`] implements [`crate::io::Stream`], +//! which this crate declared without providing an implementation. [`resolve`] +//! turns a hostname into those addresses, and [`connect_any`] tries them in order. //! [`Signal`] waits for a unix signal. The descriptor it registers is a //! `signalfd` on Linux and the read end of a pipe on the BSDs: [`Registration`] //! names a file descriptor, and `EVFILT_SIGNAL` names a signal number. @@ -71,7 +71,7 @@ pub use crate::net; pub use crate::resolve; pub use crate::signal; -pub use crate::net::{TcpListener, TcpStream}; +pub use crate::net::{TcpListener, TcpStream, UdpSocket}; // The module above and the function here share a name and do not collide: // a module is in the type namespace and a function in the value namespace. // Keeping the function called `resolve` is what makes `nagoya::reactor:: @@ -82,4 +82,4 @@ pub use driver::{Handle, Reactor, Registration, Sharded}; pub use error::{Errno, Result}; pub use local::{block_on, block_on_with, TaskSet}; pub use poller::{Event, Interest, Poller}; -pub use socket::{Addr, UnixPath, UNIX_PATH_CAPACITY}; +pub use socket::{Addr, Datagram, UnixPath, UNIX_PATH_CAPACITY}; diff --git a/src/reactor/driver.rs b/src/reactor/driver.rs index 14b1779..681a6dc 100644 --- a/src/reactor/driver.rs +++ b/src/reactor/driver.rs @@ -296,12 +296,26 @@ impl Registration { /// poller is edge triggered, so registering interest without first draining /// means waiting for an edge that has already passed. pub fn poll_readable(&self, waker: &Waker) { - let mut wakers = self.slot.wakers.lock().expect("reactor slot poisoned"); - // `will_wake` avoids the atomic refcount bump when the same task parks - // again, which on a busy connection is every single read. - match &wakers.reader { - Some(existing) if existing.will_wake(waker) => {} - _ => wakers.reader = Some(waker.clone()), + let ready = { + let mut wakers = self.slot.wakers.lock().expect("reactor slot poisoned"); + // An edge can arrive after EWOULDBLOCK and before this lock. + // In that case schedule another poll instead of parking after + // the poller has already dispatched the only available edge. + if wakers.readable || wakers.hangup { + true + } else { + // `will_wake` avoids the atomic refcount bump when the same + // task parks again. + match &wakers.reader { + Some(existing) if existing.will_wake(waker) => {} + _ => wakers.reader = Some(waker.clone()), + } + false + } + }; + // A waker can execute arbitrary code; invoke it outside the lock. + if ready { + waker.wake_by_ref(); } } @@ -331,13 +345,41 @@ impl Registration { /// Park `waker` until the descriptor is writable. See [`Self::poll_readable`]. pub fn poll_writable(&self, waker: &Waker) { - let mut wakers = self.slot.wakers.lock().expect("reactor slot poisoned"); - match &wakers.writer { - Some(existing) if existing.will_wake(waker) => {} - _ => wakers.writer = Some(waker.clone()), + let ready = { + let mut wakers = self.slot.wakers.lock().expect("reactor slot poisoned"); + if wakers.writable { + true + } else { + match &wakers.writer { + Some(existing) if existing.will_wake(waker) => {} + _ => wakers.writer = Some(waker.clone()), + } + false + } + }; + if ready { + waker.wake_by_ref(); } } + /// Release a cancelled UDP reader without consuming cached readiness. + pub(crate) fn cancel_read_wait(&self) { + let waiting = { + let mut wakers = self.slot.wakers.lock().expect("reactor slot poisoned"); + wakers.reader.take() + }; + drop(waiting); + } + + /// Release a cancelled UDP writer without consuming cached readiness. + pub(crate) fn cancel_write_wait(&self) { + let waiting = { + let mut wakers = self.slot.wakers.lock().expect("reactor slot poisoned"); + wakers.writer.take() + }; + drop(waiting); + } + /// Change what this descriptor is watched for. pub fn modify(&self, interest: Interest) -> Result<()> { self.shared.poller.modify(self.fd, self.token, interest) diff --git a/src/reactor/socket.rs b/src/reactor/socket.rs index e94eea5..58ae6a4 100644 --- a/src/reactor/socket.rs +++ b/src/reactor/socket.rs @@ -26,7 +26,8 @@ //! //! # What it is not //! -//! Not a general sockets library. There is no UDP and no dual-stack fallback. +//! Not a general sockets library. TCP and UDP use explicit address families; +//! there is no automatic dual-stack fallback. //! Name resolution is [`resolve`](super::resolve); an address arrives at this //! module already resolved. What the fleet does not use is not here. @@ -292,6 +293,11 @@ pub enum Addr { V4([u8; 4], u16), /// An IPv6 address and port. V6([u8; 16], u16), + /// An IPv6 address, port and interface scope ID. + /// + /// Use this for addresses such as link-local IPv6 addresses that require + /// an interface. A scope ID of zero is reported back as [`Self::V6`]. + V6Scoped([u8; 16], u16, u32), /// A Unix-domain socket's name, or its lack of one. Path(UnixPath), } @@ -319,7 +325,7 @@ impl Addr { /// transport is a file. pub const fn port(&self) -> u16 { match self { - Self::V4(_, port) | Self::V6(_, port) => *port, + Self::V4(_, port) | Self::V6(_, port) | Self::V6Scoped(_, port, _) => *port, Self::Path(_) => 0, } } @@ -328,7 +334,7 @@ impl Addr { const fn family(&self) -> libc::c_int { match self { Self::V4(..) => libc::AF_INET, - Self::V6(..) => libc::AF_INET6, + Self::V6(..) | Self::V6Scoped(..) => libc::AF_INET6, Self::Path(_) => libc::AF_UNIX, } } @@ -347,13 +353,17 @@ impl Addr { } mem::size_of::() as libc::socklen_t } - Self::V6(octets, port) => { + Self::V6(octets, port) | Self::V6Scoped(octets, port, _) => { let addr = storage as *mut libc::sockaddr_storage as *mut libc::sockaddr_in6; // SAFETY: as above, for the v6 layout. unsafe { (*addr).sin6_family = libc::AF_INET6 as libc::sa_family_t; (*addr).sin6_port = port.to_be(); (*addr).sin6_addr.s6_addr = *octets; + (*addr).sin6_scope_id = match self { + Self::V6Scoped(_, _, scope) => *scope, + _ => 0, + }; } mem::size_of::() as libc::socklen_t } @@ -399,10 +409,14 @@ impl Addr { let addr = storage as *const libc::sockaddr_storage as *const libc::sockaddr_in6; // SAFETY: the family field says this is a `sockaddr_in6`. unsafe { - Some(Self::V6( - (*addr).sin6_addr.s6_addr, - u16::from_be((*addr).sin6_port), - )) + let octets = (*addr).sin6_addr.s6_addr; + let port = u16::from_be((*addr).sin6_port); + let scope = (*addr).sin6_scope_id; + Some(if scope == 0 { + Self::V6(octets, port) + } else { + Self::V6Scoped(octets, port, scope) + }) } } libc::AF_UNIX => { @@ -570,6 +584,248 @@ fn addr_of(fd: i32, peer: bool) -> Result { Addr::read_from(&storage, len).ok_or(Errno(libc::EAFNOSUPPORT)) } +/// The result of consuming one UDP datagram. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Datagram { + /// Bytes copied into the supplied buffer. + /// + /// This can be zero for an empty datagram or an empty receive buffer. + /// Zero does not indicate end of stream. + pub len: usize, + /// The sender's address, including an IPv6 scope ID when present. + pub source: Addr, + /// Whether the buffer was too small for the datagram. + /// + /// The kernel discards the remainder of that datagram. The next receive + /// starts with the next datagram. The original length is not reported + /// because platforms differ in whether they make it available. + pub truncated: bool, +} + +/// Create a UDP descriptor without exposing a blocking socket. +fn datagram_socket(addr: &Addr) -> Result { + if matches!(addr, Addr::Path(_)) { + return Err(Errno(libc::EAFNOSUPPORT)); + } + + #[cfg(target_os = "linux")] + let fd = { + let raw = check(unsafe { + libc::socket( + addr.family(), + libc::SOCK_DGRAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + libc::IPPROTO_UDP, + ) + })?; + // SAFETY: this is a fresh descriptor owned exclusively here. + unsafe { Fd::from_raw(raw) } + }; + + #[cfg(not(target_os = "linux"))] + let fd = { + let raw = + check(unsafe { libc::socket(addr.family(), libc::SOCK_DGRAM, libc::IPPROTO_UDP) })?; + // Take ownership before setting flags so every failure closes it. + // SAFETY: this is a fresh descriptor owned exclusively here. + let fd = unsafe { Fd::from_raw(raw) }; + let flags = check(unsafe { libc::fcntl(fd.raw(), libc::F_GETFL) })?; + check(unsafe { libc::fcntl(fd.raw(), libc::F_SETFL, flags | libc::O_NONBLOCK) })?; + let flags = check(unsafe { libc::fcntl(fd.raw(), libc::F_GETFD) })?; + check(unsafe { libc::fcntl(fd.raw(), libc::F_SETFD, flags | libc::FD_CLOEXEC) })?; + set_flag(fd.raw(), libc::SOL_SOCKET, libc::SO_NOSIGPIPE, true)?; + fd + }; + + // An explicit IPv6 socket stays IPv6. Callers that need both families + // own one socket for each rather than depending on platform defaults. + if addr.family() == libc::AF_INET6 { + set_flag(fd.raw(), libc::IPPROTO_IPV6, libc::IPV6_V6ONLY, true)?; + } + Ok(fd) +} + +/// An owned, non-blocking IPv4 or IPv6 UDP socket. +/// +/// Operations here are single syscalls. The reactor-facing futures live in +/// [`crate::net::UdpSocket`]. Dropping this socket closes its descriptor. +#[derive(Debug)] +pub struct UdpSocket { + fd: Fd, + family: libc::c_int, +} + +impl UdpSocket { + /// Bind, optionally enabling `SO_REUSEADDR` before the bind. + /// + /// A port of zero requests an ephemeral port; [`Self::local_addr`] + /// reports the chosen port. Unix-domain addresses are refused. + pub fn bind(addr: Addr, reuse_addr: bool) -> Result { + let fd = datagram_socket(&addr)?; + set_flag(fd.raw(), libc::SOL_SOCKET, libc::SO_REUSEADDR, reuse_addr)?; + // SAFETY: zeroed storage is filled for the selected address family. + let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() }; + let len = addr.write_to(&mut storage); + // SAFETY: storage holds a valid address of len bytes. + check(unsafe { libc::bind(fd.raw(), core::ptr::addr_of!(storage).cast(), len) })?; + Ok(Self { + fd, + family: addr.family(), + }) + } + + /// The descriptor, still owned by this socket. + pub fn raw(&self) -> i32 { + self.fd.raw() + } + + /// This socket's bound address. + pub fn local_addr(&self) -> Result { + addr_of(self.raw(), false) + } + + /// The configured peer, or `ENOTCONN` for an unconnected socket. + pub fn peer_addr(&self) -> Result { + addr_of(self.raw(), true) + } + + /// Select a peer and let the kernel filter incoming datagrams by it. + /// + /// UDP has no handshake. A platform that returns `EINPROGRESS` requires + /// writable readiness and [`Self::connect_error`] before use. + pub fn connect(&self, addr: Addr) -> Result<()> { + if addr.family() != self.family { + return Err(Errno(libc::EAFNOSUPPORT)); + } + // SAFETY: zeroed storage is filled for the selected address family. + let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() }; + let len = addr.write_to(&mut storage); + // SAFETY: storage holds a valid address of len bytes. + check(unsafe { libc::connect(self.raw(), core::ptr::addr_of!(storage).cast(), len) })?; + Ok(()) + } + + /// Check the result of an in-flight connect. + pub fn connect_error(&self) -> Result<()> { + let mut value: libc::c_int = 0; + let mut len = mem::size_of::() as libc::socklen_t; + // SAFETY: SO_ERROR writes one int into the live value. + check(unsafe { + libc::getsockopt( + self.raw(), + libc::SOL_SOCKET, + libc::SO_ERROR, + core::ptr::addr_of_mut!(value).cast(), + core::ptr::addr_of_mut!(len), + ) + })?; + if value == 0 { + Ok(()) + } else { + Err(Errno(value)) + } + } + + /// Receive exactly one datagram into potentially uninitialised memory. + /// + /// Empty datagrams and empty receive buffers are valid. `MSG_TRUNC` + /// reports discarded bytes without consuming any following datagram. + /// + /// # Safety + /// + /// `pointer` must be valid for writes of `len` bytes. + pub unsafe fn recv_from(&self, pointer: *mut u8, len: usize) -> Result { + let mut storage: libc::sockaddr_storage = mem::zeroed(); + let mut iovec = libc::iovec { + iov_base: pointer.cast(), + iov_len: len, + }; + let mut message: libc::msghdr = mem::zeroed(); + message.msg_name = core::ptr::addr_of_mut!(storage).cast(); + message.msg_namelen = mem::size_of::() as libc::socklen_t; + message.msg_iov = core::ptr::addr_of_mut!(iovec); + message.msg_iovlen = 1; + + #[cfg(feature = "syscall-counters")] + crate::reactor::counters::RECV.fetch_add(1, core::sync::atomic::Ordering::Relaxed); + // Do not pass MSG_TRUNC as an input flag: Linux would then return the + // original length instead of the number of initialised bytes. + let received = libc::recvmsg(self.raw(), core::ptr::addr_of_mut!(message), 0); + if received < 0 { + let error = crate::reactor::error::last(); + #[cfg(feature = "syscall-counters")] + if error.would_block() { + crate::reactor::counters::RECV_WOULD_BLOCK + .fetch_add(1, core::sync::atomic::Ordering::Relaxed); + } + return Err(error); + } + let source = + Addr::read_from(&storage, message.msg_namelen).ok_or(Errno(libc::EAFNOSUPPORT))?; + Ok(Datagram { + len: (received as usize).min(len), + source, + truncated: message.msg_flags & libc::MSG_TRUNC != 0 || received as usize > len, + }) + } + + /// Send one complete datagram to an explicit destination. + pub fn send_to(&self, buffer: &[u8], destination: Addr) -> Result { + self.send_datagram(buffer, Some(destination)) + } + + /// Send one complete datagram to the configured peer. + pub fn send(&self, buffer: &[u8]) -> Result { + self.send_datagram(buffer, None) + } + + fn send_datagram(&self, buffer: &[u8], destination: Option) -> Result { + // SAFETY: zeroed storage is filled before its address is used. + let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() }; + let (address, len) = if let Some(destination) = destination { + if destination.family() != self.family { + return Err(Errno(libc::EAFNOSUPPORT)); + } + let len = destination.write_to(&mut storage); + (core::ptr::addr_of!(storage).cast::(), len) + } else { + (core::ptr::null(), 0) + }; + #[cfg(target_os = "linux")] + let flags = libc::MSG_NOSIGNAL; + #[cfg(not(target_os = "linux"))] + let flags = 0; + + #[cfg(feature = "syscall-counters")] + crate::reactor::counters::SEND.fetch_add(1, core::sync::atomic::Ordering::Relaxed); + // SAFETY: buffer and the optional address remain live for the call. + let sent = unsafe { + libc::sendto( + self.raw(), + buffer.as_ptr().cast(), + buffer.len(), + flags, + address, + len, + ) + }; + if sent < 0 { + let error = crate::reactor::error::last(); + #[cfg(feature = "syscall-counters")] + if error.would_block() { + crate::reactor::counters::SEND_WOULD_BLOCK + .fetch_add(1, core::sync::atomic::Ordering::Relaxed); + } + Err(error) + } else if sent as usize != buffer.len() { + // UDP writes are atomic. Never turn a short result into another + // datagram by retrying a suffix. + Err(Errno(libc::EMSGSIZE)) + } else { + Ok(sent as usize) + } + } +} + /// A connected stream socket: TCP, or Unix-domain. /// /// Still named for TCP because that is what almost every one of these is, and From e21cebe2496acf5460de2e0cda618a6e7f98c417 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 14 Oct 2026 01:21:19 +0700 Subject: [PATCH 2/2] index on master: 6c54823 Add UDP sockets and IPv6 scope IDs