From 35a3b998f77d8f11d238a8ce67ddaa299265db93 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 29 Sep 2026 02:04:57 +0700 Subject: [PATCH 1/3] Add UdpSocket to the reactor net::UdpSocket: bind, bind_reuse_addr, from_socket, local_addr, recv/send buffer sizes, and poll_recv_from/recv_from, poll_send_to/send_to. recvfrom writes straight into the caller's slice: no internal buffer, no allocation per datagram. Same readiness discipline as TcpStream: one waker per direction, so one task can receive while another sends. --- src/net.rs | 180 ++++++++++++++++++++++++++++++++++++++++-- src/reactor.rs | 9 ++- src/reactor/socket.rs | 167 ++++++++++++++++++++++++++++++++++++++- tests/reactor_net.rs | 124 +++++++++++++++++++++++++++-- 4 files changed, 463 insertions(+), 17 deletions(-) diff --git a/src/net.rs b/src/net.rs index e32212a..abe87fb 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` //! @@ -16,10 +16,12 @@ //! //! # The edge triggered contract //! -//! Every read and write loops until the kernel says `EWOULDBLOCK`, and only -//! then parks a waker. Registering interest without first draining would wait -//! for an edge that has already passed, and the task would hang with data -//! sitting in the socket buffer. +//! Readiness is parked only after the kernel says `EWOULDBLOCK`. A TCP read +//! can return after filling its slice, so it restores readiness when data may +//! remain. A UDP read consumes one datagram and restores readiness for the +//! next poll; only an attempted receive that would block proves the queue is +//! drained. Registering interest without either step would wait for an edge +//! that has already passed, and the task would hang with data in the socket. // Reading into uninitialised memory is the one thing this module does that the // safe subset cannot express; see `poll_read_buf`. @@ -31,7 +33,9 @@ use core::task::{Context, Poll}; use crate::reactor::driver::{Handle, Registration}; use crate::reactor::error::Result; use crate::reactor::poller::Interest; -use crate::reactor::socket::{Addr, TcpListener as Listener, TcpSocket}; +use crate::reactor::socket::{ + Addr, TcpListener as Listener, TcpSocket, UdpSocket as DatagramSocket, +}; /// Name resolution, which tokio also puts under `net`. /// @@ -581,6 +585,170 @@ impl core::future::Future for Accept<'_> { } } +/// A message-oriented socket that yields instead of blocking. +/// +/// The receiver and sender have separate readiness slots, so one task may +/// receive while another sends on the same socket. There is one waker slot per +/// direction: serialize receives with each other and sends with each other. +#[derive(Debug)] +pub struct UdpSocket { + inner: DatagramSocket, + registration: Registration, +} + +impl UdpSocket { + /// Bind a non-blocking socket and register it for reads and writes. + pub fn bind(addr: Addr, handle: &Handle) -> Result { + Self::from_socket(DatagramSocket::bind(addr)?, handle) + } + + /// Bind with `SO_REUSEADDR` enabled for IP addresses before they are + /// claimed. + pub fn bind_reuse_addr(addr: Addr, handle: &Handle) -> Result { + Self::from_socket(DatagramSocket::bind_reuse_addr(addr)?, handle) + } + + /// Adopt an existing non-blocking datagram socket. + pub fn from_socket(socket: DatagramSocket, handle: &Handle) -> Result { + let registration = handle.register(socket.raw(), Interest::BOTH)?; + Ok(Self { + inner: socket, + registration, + }) + } + + /// This socket's own address. + pub fn local_addr(&self) -> Result { + self.inner.local_addr() + } + + /// Set the kernel receive-buffer capacity in bytes. + pub fn set_recv_buffer(&self, size: usize) -> Result<()> { + self.inner.set_recv_buffer(size) + } + + /// Set the kernel send-buffer capacity in bytes. + pub fn set_send_buffer(&self, size: usize) -> Result<()> { + self.inner.set_send_buffer(size) + } + + /// Receive one datagram directly into `buffer`. + /// + /// The receive is zero-copy at the crate boundary: the kernel writes into + /// the caller's slice, with no intermediate buffer or per-datagram + /// allocation. A larger datagram is truncated to fit. + pub fn poll_recv_from( + &self, + cx: &mut Context<'_>, + buffer: &mut [u8], + ) -> Poll> { + if !self.registration.take_readable_or_park(cx.waker()) { + return Poll::Pending; + } + loop { + // SAFETY: `buffer` is live for this call, and the socket writes no + // more than its length into it. + match unsafe { self.inner.recv_from(buffer.as_mut_ptr(), buffer.len()) } { + Ok(received) => { + // One datagram does not drain the receive queue. Keep this + // registration ready so the next poll can try another; + // if there is none, it will see EWOULDBLOCK and park. + // This also handles queued zero-length datagrams, which + // FIONREAD cannot distinguish from an empty queue. + self.registration.mark_readable(); + return Poll::Ready(Ok(received)); + } + Err(error) if error.would_block() => { + self.registration.poll_readable(cx.waker()); + return Poll::Pending; + } + Err(error) if error.interrupted() => continue, + Err(error) => return Poll::Ready(Err(error)), + } + } + } + + /// Send one datagram to `target`, parking if the send buffer is full. + pub fn poll_send_to( + &self, + cx: &mut Context<'_>, + buffer: &[u8], + target: Addr, + ) -> Poll> { + if !self.registration.take_writable_or_park(cx.waker()) { + return Poll::Pending; + } + loop { + match self.inner.send_to(buffer, target) { + Ok(sent) => { + // Datagram sends are atomic: after this one fit, let the + // next send attempt the syscall. If concurrent senders + // have since filled the buffer it will park on + // EWOULDBLOCK below. + self.registration.mark_writable(); + 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) => return Poll::Ready(Err(error)), + } + } + } + + /// Receive one datagram, as a future. + pub fn recv_from<'a>(&'a self, buffer: &'a mut [u8]) -> RecvFrom<'a> { + RecvFrom { + socket: self, + buffer, + } + } + + /// Send one datagram to `target`, as a future. + pub fn send_to<'a>(&'a self, buffer: &'a [u8], target: Addr) -> SendTo<'a> { + SendTo { + socket: self, + buffer, + target, + } + } +} + +/// The future returned by [`UdpSocket::recv_from`]. +#[derive(Debug)] +pub struct RecvFrom<'a> { + socket: &'a UdpSocket, + buffer: &'a mut [u8], +} + +impl core::future::Future for RecvFrom<'_> { + type Output = Result<(usize, Addr)>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + this.socket.poll_recv_from(cx, this.buffer) + } +} + +/// The future returned by [`UdpSocket::send_to`]. +#[derive(Debug)] +pub struct SendTo<'a> { + socket: &'a UdpSocket, + buffer: &'a [u8], + target: Addr, +} + +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_to(cx, this.buffer, this.target) + } +} + /// The byte stream trait this crate declared, with something behind it at last. /// /// [`io::Stream`](crate::io::Stream) was written here because two crates always diff --git a/src/reactor.rs b/src/reactor.rs index a1a07f2..a9d03a0 100644 --- a/src/reactor.rs +++ b/src/reactor.rs @@ -36,9 +36,10 @@ //! //! [`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 +//! 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`] @@ -71,7 +72,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:: diff --git a/src/reactor/socket.rs b/src/reactor/socket.rs index 0dfc7c6..af3f840 100644 --- a/src/reactor/socket.rs +++ b/src/reactor/socket.rs @@ -26,8 +26,8 @@ //! //! # What it is not //! -//! Not a general sockets library. There is no UDP and no dual-stack fallback. -//! Name resolution is [`resolve`](super::resolve); an address arrives at this +//! Not a general sockets library. There is no 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. // Calling the kernel is this module's entire purpose. @@ -533,6 +533,35 @@ fn stream_socket(addr: &Addr) -> Result { Ok(fd) } +/// Create a non-blocking datagram socket for `addr`'s family. +fn datagram_socket(addr: &Addr) -> Result { + // SOCK_NONBLOCK and SOCK_CLOEXEC are Linux extensions to `socket`; the + // BSDs set both flags after creation, before the descriptor is published. + #[cfg(target_os = "linux")] + let raw = check(unsafe { + libc::socket( + addr.family(), + libc::SOCK_DGRAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + 0, + ) + })?; + + #[cfg(not(target_os = "linux"))] + let raw = { + let raw = check(unsafe { libc::socket(addr.family(), libc::SOCK_DGRAM, 0) })?; + // SAFETY: `raw` is a descriptor the call above just returned. + unsafe { + let flags = libc::fcntl(raw, libc::F_GETFL); + libc::fcntl(raw, libc::F_SETFL, flags | libc::O_NONBLOCK); + libc::fcntl(raw, libc::F_SETFD, libc::FD_CLOEXEC); + } + raw + }; + + // SAFETY: `raw` is a fresh descriptor owned exclusively here. + Ok(unsafe { Fd::from_raw(raw) }) +} + /// Set a boolean socket option. fn set_flag(fd: i32, level: libc::c_int, name: libc::c_int, on: bool) -> Result<()> { let value: libc::c_int = i32::from(on); @@ -549,6 +578,22 @@ fn set_flag(fd: i32, level: libc::c_int, name: libc::c_int, on: bool) -> Result< Ok(()) } +/// Set an integer socket option, used for send and receive buffer sizes. +fn set_socket_int(fd: i32, name: libc::c_int, value: usize) -> Result<()> { + let value = libc::c_int::try_from(value).map_err(|_| Errno(libc::EINVAL))?; + // SAFETY: SO_RCVBUF and SO_SNDBUF take an int, which is what is passed. + check(unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + name, + core::ptr::addr_of!(value).cast(), + mem::size_of::() as libc::socklen_t, + ) + })?; + Ok(()) +} + /// Ask the kernel for one of a socket's own addresses. fn addr_of(fd: i32, peer: bool) -> Result { // SAFETY: an all-zero `sockaddr_storage` is valid; the kernel fills it. @@ -816,6 +861,124 @@ impl TcpSocket { } } +/// A non-blocking datagram socket. +#[derive(Debug)] +pub struct UdpSocket(Fd); + +impl UdpSocket { + /// Bind a datagram socket to `addr`. + pub fn bind(addr: Addr) -> Result { + Self::bind_with(addr, false) + } + + /// Bind with `SO_REUSEADDR` enabled for IP addresses before they are + /// claimed. + pub fn bind_reuse_addr(addr: Addr) -> Result { + Self::bind_with(addr, true) + } + + fn bind_with(addr: Addr, reuse_addr: bool) -> Result { + let fd = datagram_socket(&addr)?; + if reuse_addr && addr.family() != libc::AF_UNIX { + set_flag(fd.raw(), libc::SOL_SOCKET, libc::SO_REUSEADDR, true)?; + } + + // SAFETY: zeroed storage, then written by `write_to` for this 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)) + } + + /// Adopt an existing non-blocking datagram descriptor. + /// + /// # Safety + /// + /// `fd` must be an open, non-blocking datagram socket that nothing else + /// will close. + pub unsafe fn from_raw(fd: i32) -> Self { + Self(Fd::from_raw(fd)) + } + + /// The descriptor, for registering with the poller. + #[inline] + pub fn raw(&self) -> i32 { + self.0.raw() + } + + /// This socket's own address. + pub fn local_addr(&self) -> Result { + addr_of(self.raw(), false) + } + + /// Set the kernel receive-buffer capacity in bytes. + pub fn set_recv_buffer(&self, size: usize) -> Result<()> { + set_socket_int(self.raw(), libc::SO_RCVBUF, size) + } + + /// Set the kernel send-buffer capacity in bytes. + pub fn set_send_buffer(&self, size: usize) -> Result<()> { + set_socket_int(self.raw(), libc::SO_SNDBUF, size) + } + + /// Receive one datagram directly into caller-owned memory. + /// + /// A datagram larger than `len` is truncated by the kernel. The returned + /// count is the number of bytes copied into the buffer. + /// + /// # Safety + /// + /// `pointer` must be valid for writes of `len` bytes. + pub unsafe fn recv_from(&self, pointer: *mut u8, len: usize) -> Result<(usize, Addr)> { + // SAFETY: zeroed storage is valid for the kernel to fill with a peer + // address, and `address_len` describes the full storage. + let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() }; + let mut address_len = mem::size_of::() as libc::socklen_t; + // SAFETY: the caller guarantees the receive buffer, while the address + // storage and its length are live locals of the right types. + let read = unsafe { + libc::recvfrom( + self.raw(), + pointer.cast(), + len, + 0, + core::ptr::addr_of_mut!(storage).cast(), + core::ptr::addr_of_mut!(address_len), + ) + }; + if read < 0 { + return Err(crate::reactor::error::last()); + } + let addr = Addr::read_from(&storage, address_len).ok_or(Errno(libc::EAFNOSUPPORT))?; + Ok((read as usize, addr)) + } + + /// Send one datagram from caller-owned memory to `target`. + pub fn send_to(&self, buffer: &[u8], target: Addr) -> Result { + // SAFETY: zeroed storage, then written by `write_to` for this family. + let mut storage: libc::sockaddr_storage = unsafe { mem::zeroed() }; + let address_len = target.write_to(&mut storage); + // SAFETY: the buffer and address storage are live for the call, and + // `address_len` describes the address written above. + let sent = unsafe { + libc::sendto( + self.raw(), + buffer.as_ptr().cast(), + buffer.len(), + 0, + core::ptr::addr_of!(storage).cast(), + address_len, + ) + }; + if sent < 0 { + Err(crate::reactor::error::last()) + } else { + Ok(sent as usize) + } + } +} + /// A listening stream socket: TCP, or Unix-domain. #[derive(Debug)] pub struct TcpListener { diff --git a/tests/reactor_net.rs b/tests/reactor_net.rs index d794e22..17b51db 100644 --- a/tests/reactor_net.rs +++ b/tests/reactor_net.rs @@ -1,12 +1,12 @@ //! Sockets over the reactor, through the public API. //! //! These live here rather than inside `src/reactor/net.rs` because they need -//! nothing the crate does not already export: a listener, a stream, a reactor -//! and `block_on`. A test that only uses the public surface is an integration -//! test, and keeping it beside the implementation only makes the -//! implementation longer to read. +//! nothing the crate does not already export: a listener, a stream or +//! datagram socket, a reactor and `block_on`. A test that only uses the public +//! surface is an integration test, and keeping it beside the implementation +//! only makes the implementation longer to read. -use nagoya::reactor::{Addr, Reactor, TcpListener, TcpStream}; +use nagoya::reactor::{Addr, Reactor, TcpListener, TcpStream, UdpSocket}; /// A socket is usable through the trait, not just beside it. /// @@ -61,6 +61,120 @@ fn local() -> Addr { Addr::localhost(0) } +#[test] +fn udp_round_trips_to_an_explicit_addr() { + let reactor = Reactor::local().expect("reactor"); + let handle = reactor.handle(); + let receiver = UdpSocket::bind(local(), &handle).expect("receiver bind"); + let sender = UdpSocket::bind(local(), &handle).expect("sender bind"); + let destination = receiver.local_addr().expect("receiver addr"); + let source = sender.local_addr().expect("sender addr"); + + let (sent, received, from, payload) = nagoya::reactor::block_on_with(&reactor, async { + let sent = sender.send_to(b"ping", destination).await.expect("send"); + let mut buffer = [0u8; 8]; + let (received, from) = receiver.recv_from(&mut buffer).await.expect("receive"); + (sent, received, from, buffer[..received].to_vec()) + }); + + assert_eq!(sent, 4); + assert_eq!(received, 4); + assert_eq!(payload, b"ping"); + assert_eq!(from, source); +} + +#[test] +fn udp_receives_many_queued_datagrams_without_loss() { + const COUNT: u32 = 64; + + let reactor = Reactor::local().expect("reactor"); + let handle = reactor.handle(); + let receiver = UdpSocket::bind(local(), &handle).expect("receiver bind"); + let sender = UdpSocket::bind(local(), &handle).expect("sender bind"); + let destination = receiver.local_addr().expect("receiver addr"); + let source = sender.local_addr().expect("sender addr"); + + let received = nagoya::reactor::block_on_with(&reactor, async { + let first = 0u32.to_be_bytes(); + sender.send_to(&first, destination).await.expect("send"); + sender.send_to(&[], destination).await.expect("empty send"); + for sequence in 1..COUNT { + let bytes = sequence.to_be_bytes(); + sender.send_to(&bytes, destination).await.expect("send"); + } + + let mut buffer = [0u8; 4]; + let mut seen = Vec::with_capacity(COUNT as usize); + let (length, from) = receiver.recv_from(&mut buffer).await.expect("receive"); + assert_eq!(length, buffer.len()); + assert_eq!(from, source); + seen.push(u32::from_be_bytes(buffer)); + + let (length, from) = receiver + .recv_from(&mut buffer) + .await + .expect("empty receive"); + assert_eq!(length, 0); + assert_eq!(from, source); + // Sequence zero and the empty datagram were read above. + for _ in 1..COUNT { + let (length, from) = receiver.recv_from(&mut buffer).await.expect("receive"); + assert_eq!(length, buffer.len()); + assert_eq!(from, source); + seen.push(u32::from_be_bytes(buffer)); + } + seen + }); + + assert_eq!(received, (0..COUNT).collect::>()); +} + +#[test] +fn udp_receive_parks_and_wakes_for_the_next_datagram() { + use core::task::Poll; + + let reactor = Reactor::local().expect("reactor"); + let handle = reactor.handle(); + let receiver = UdpSocket::bind(local(), &handle).expect("receiver bind"); + let destination = receiver.local_addr().expect("receiver addr"); + let (pending_tx, pending_rx) = std::sync::mpsc::channel(); + + // Send only after the receive poll has returned Pending, so this exercises + // the parked waker rather than a datagram that arrived before first poll. + let sender = std::thread::spawn(move || { + pending_rx.recv().expect("receive poll"); + let sender = std::net::UdpSocket::bind(("127.0.0.1", 0)).expect("peer bind"); + sender + .send_to( + b"wake", + std::net::SocketAddr::from(([127, 0, 0, 1], destination.port())), + ) + .expect("peer send"); + }); + + let mut buffer = [0u8; 8]; + let mut announced_pending = false; + let (length, from) = nagoya::reactor::block_on_with( + &reactor, + std::future::poll_fn(|cx| match receiver.poll_recv_from(cx, &mut buffer) { + Poll::Pending => { + if !announced_pending { + pending_tx.send(()).expect("announce pending receive"); + announced_pending = true; + } + Poll::Pending + } + ready => ready, + }), + ) + .expect("receive"); + + sender.join().expect("peer"); + assert_eq!(length, 4); + assert_eq!(&buffer[..length], b"wake"); + assert!(matches!(from, Addr::V4([127, 0, 0, 1], port) if port != 0)); +} + #[test] fn a_connection_round_trips_through_the_reactor() { let reactor = Reactor::start().expect("reactor"); From 1c27c52ffa2497e2430a41d7b064e568375f5530 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 29 Sep 2026 02:04:57 +0700 Subject: [PATCH 2/3] Release 0.1.15 --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 88f717d..c409c20 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nagoya" -version = "0.1.14" +version = "0.1.15" edition = "2021" rust-version = "1.85" license = "MIT OR Apache-2.0" From f3e9cb29f6f80338a8bcdcc92d8d7d6c460fe573 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 29 Sep 2026 02:08:29 +0700 Subject: [PATCH 3/3] Note 0.1.15 in the README --- README.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/README.md b/README.md index 50174a8..33fd462 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,11 @@ Typst is the canonical documentation source. Run `sh scripts/build-guide.sh` to create both PDFs in `docs/`; Rust examples are extracted and checked by `cargo test --doc`. +Version 0.1.15 adds `net::UdpSocket` to the reactor: `bind`, `recv_from` and +`send_to`, with the receive writing straight into the caller's slice, so a +datagram costs no buffer and no allocation. One task can receive while +another sends on the same socket. + Version 0.1.14 adds `Runtime::builder()`, which takes what `with_tuning` does plus the worker threads' stack size. Without it a worker's stack is whatever `RUST_MIN_STACK` says, or two MiB, so a caller whose tasks recurse deeply had