From a598dbdc5c66a7452523e2b6bcb5eaf6b21295f1 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 17 Aug 2026 00:32:37 +0700 Subject: [PATCH] fix(io): address review feedback on interrupted-receive retries - Error-queue drain: an interrupted MSG_ERRQUEUE read no longer ends the bounded drain early as if the queue were empty (which could strand a still-queued TX timestamp past the point its probe is looked up and removed). It now reports ErrorQueueRecord::Ignored, consuming one attempt of the drain's existing per-call work budget and retrying the same slot instead. - Fake-server recv_from test helpers (client_multi_target.rs and tests/support/mod.rs) now keep one absolute deadline across interruptions, re-applying only the remaining time on each retry and restoring the configured timeout afterward, mirroring the production blocking-receive fix these tests exist to guard rather than restarting the full timeout on every interruption. --- crates/irtt-cli/tests/client_multi_target.rs | 75 +++++++++++++------ .../src/receive/linux/error_queue.rs | 23 +++--- crates/irtt-client/tests/support/mod.rs | 74 ++++++++++++------ 3 files changed, 121 insertions(+), 51 deletions(-) diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index 5b823f1..a8d9ef3 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -4,7 +4,7 @@ mod in_tree_server; use std::{ - io::{ErrorKind, Read}, + io::{self, ErrorKind, Read}, net::{SocketAddr, UdpSocket}, process::{Command, Output, Stdio}, sync::mpsc, @@ -1177,35 +1177,68 @@ fn start_open_failure_server_inner( FakeServer { addr, done } } -/// A fake server's `recv_from`, retrying an interrupted syscall -/// (`EINTR`/`ErrorKind::Interrupted`) instead of panicking on it: real +/// A `recv_from` that retries an interrupted syscall +/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright: real /// signal delivery (e.g. process-group signals during CI) can interrupt a /// blocking receive here without there being anything actually wrong. +/// +/// Keeps one absolute deadline across any number of interruptions, exactly +/// like the production blocking receive fix this test file exists to guard: +/// each retry re-applies only the *remaining* time against the socket's +/// currently configured read timeout, so a signal storm cannot turn this +/// into an unbounded wait. The configured timeout is restored before +/// returning so a later call is unaffected by an earlier interruption here. +fn recv_from_retrying_interrupted( + socket: &UdpSocket, + buf: &mut [u8], +) -> io::Result<(usize, SocketAddr)> { + let configured_timeout = socket.read_timeout()?; + let deadline = configured_timeout.and_then(|timeout| Instant::now().checked_add(timeout)); + let mut timeout_narrowed = false; + + let result = loop { + match socket.recv_from(buf) { + Ok(received) => break Ok(received), + Err(err) if err.kind() == ErrorKind::Interrupted => { + if let Some(deadline) = deadline { + let Some(remaining) = deadline.checked_duration_since(Instant::now()) else { + break Err(io::Error::from(ErrorKind::TimedOut)); + }; + if remaining.is_zero() { + break Err(io::Error::from(ErrorKind::TimedOut)); + } + if let Err(err) = socket.set_read_timeout(Some(remaining)) { + break Err(err); + } + timeout_narrowed = true; + } + } + Err(err) => break Err(err), + } + }; + + if timeout_narrowed { + socket.set_read_timeout(configured_timeout)?; + } + result +} + fn recv_request(socket: &UdpSocket) -> (Vec, SocketAddr) { let mut buf = [0_u8; 2048]; - loop { - match socket.recv_from(&mut buf) { - Ok((size, peer)) => return (buf[..size].to_vec(), peer), - Err(err) if err.kind() == ErrorKind::Interrupted => continue, - Err(err) => panic!("fake server recv_from failed: {err}"), - } + match recv_from_retrying_interrupted(socket, &mut buf) { + Ok((size, peer)) => (buf[..size].to_vec(), peer), + Err(err) => panic!("fake server recv_from failed: {err}"), } } -/// Like [`recv_request`], but treats any non-interrupting error (including a -/// configured read timeout) as "nothing more to receive" rather than a -/// failure, since callers use this to detect the peer going quiet. An -/// interrupted syscall carries no such information and is retried instead of -/// being misclassified as quiet. +/// Like [`recv_request`], but treats any error, including a configured read +/// timeout, as "nothing more to receive" rather than a failure, since +/// callers use this to detect the peer going quiet. fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec, SocketAddr)> { let mut buf = [0_u8; 2048]; - loop { - match socket.recv_from(&mut buf) { - Ok((size, peer)) => return Some((buf[..size].to_vec(), peer)), - Err(err) if err.kind() == ErrorKind::Interrupted => continue, - Err(_) => return None, - } - } + recv_from_retrying_interrupted(socket, &mut buf) + .ok() + .map(|(size, peer)| (buf[..size].to_vec(), peer)) } fn open_reply(flags: u8, token: u64, params: &Params) -> Vec { diff --git a/crates/irtt-client/src/receive/linux/error_queue.rs b/crates/irtt-client/src/receive/linux/error_queue.rs index 5657f7f..c224919 100644 --- a/crates/irtt-client/src/receive/linux/error_queue.rs +++ b/crates/irtt-client/src/receive/linux/error_queue.rs @@ -153,14 +153,18 @@ impl ErrorQueueControlBuffer { /// Nonblocking drain of a single record from `fd`'s `MSG_ERRQUEUE`. /// -/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) or -/// when this one read was interrupted (`EINTR`). TX timestamp capture is -/// optional best-effort metadata, not a network operation: an interrupted -/// read here is not a socket/network error and must not become one, so it is -/// treated as no record obtained on this drain rather than retried in a -/// loop. The bounded caller in [`super::drain_tx_timestamps`] naturally -/// revisits the queue on its next call, which is how every other record is -/// picked up too. +/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) — the +/// caller in [`super::drain_tx_timestamps`] stops its drain there, exactly +/// as it always has. An interrupted read (`EINTR`) is different: it says +/// nothing about whether the queue is empty, so treating it the same as +/// `EWOULDBLOCK` could end a drain early and strand a still-queued record +/// past the point its matching probe is looked up and removed (see the +/// crate's `AGENTS.md`). TX timestamp capture is optional best-effort +/// metadata, not a network operation, so an interruption here must also +/// never become a socket/network error; instead it is reported as +/// [`ErrorQueueRecord::Ignored`], which consumes one attempt of the bounded +/// caller's existing per-drain work budget and lets the loop immediately +/// retry the same slot rather than stopping or spinning unboundedly. /// `SOF_TIMESTAMPING_OPT_TSONLY` notifications carry no meaningful payload, /// so a zero-length iovec is enough: correlation is by `id`, never by /// payload content. @@ -181,7 +185,8 @@ pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result Ok(None), + Err(Errno::EWOULDBLOCK) => Ok(None), + Err(Errno::EINTR) => Ok(Some(ErrorQueueRecord::Ignored)), Err(errno) => Err(io::Error::from(errno)), } } diff --git a/crates/irtt-client/tests/support/mod.rs b/crates/irtt-client/tests/support/mod.rs index cd75037..24274e0 100644 --- a/crates/irtt-client/tests/support/mod.rs +++ b/crates/irtt-client/tests/support/mod.rs @@ -4,12 +4,12 @@ mod in_tree_server; mod real_irtt; use std::{ - io::ErrorKind, + io::{self, ErrorKind}, net::{SocketAddr, UdpSocket}, process::Command, sync::mpsc, thread::{self, JoinHandle}, - time::{Duration, SystemTime}, + time::{Duration, Instant, SystemTime}, }; use irtt_client::{Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, SocketConfig}; @@ -448,32 +448,64 @@ where FakeServer { addr, rx, done } } -/// A fake server's `recv_from`, retrying an interrupted syscall -/// (`EINTR`/`ErrorKind::Interrupted`) instead of panicking on it. +/// A `recv_from` that retries an interrupted syscall +/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright, keeping +/// one absolute deadline across any number of interruptions exactly like the +/// production blocking receive fix this test file exists to guard: each +/// retry re-applies only the *remaining* time against the socket's +/// currently configured read timeout, so a signal storm cannot turn this +/// into an unbounded wait. The configured timeout is restored before +/// returning so a later call is unaffected by an earlier interruption here. +fn recv_from_retrying_interrupted( + socket: &UdpSocket, + buf: &mut [u8], +) -> io::Result<(usize, SocketAddr)> { + let configured_timeout = socket.read_timeout()?; + let deadline = configured_timeout.and_then(|timeout| Instant::now().checked_add(timeout)); + let mut timeout_narrowed = false; + + let result = loop { + match socket.recv_from(buf) { + Ok(received) => break Ok(received), + Err(err) if err.kind() == ErrorKind::Interrupted => { + if let Some(deadline) = deadline { + let Some(remaining) = deadline.checked_duration_since(Instant::now()) else { + break Err(io::Error::from(ErrorKind::TimedOut)); + }; + if remaining.is_zero() { + break Err(io::Error::from(ErrorKind::TimedOut)); + } + if let Err(err) = socket.set_read_timeout(Some(remaining)) { + break Err(err); + } + timeout_narrowed = true; + } + } + Err(err) => break Err(err), + } + }; + + if timeout_narrowed { + socket.set_read_timeout(configured_timeout)?; + } + result +} + fn recv_request(socket: &UdpSocket) -> (Vec, SocketAddr) { let mut buf = [0_u8; 8192]; - loop { - match socket.recv_from(&mut buf) { - Ok((size, peer)) => return (buf[..size].to_vec(), peer), - Err(err) if err.kind() == ErrorKind::Interrupted => continue, - Err(err) => panic!("fake server recv_from failed: {err}"), - } + match recv_from_retrying_interrupted(socket, &mut buf) { + Ok((size, peer)) => (buf[..size].to_vec(), peer), + Err(err) => panic!("fake server recv_from failed: {err}"), } } -/// Like [`recv_request`], but any non-interrupting error (including a -/// configured read timeout) means "nothing more to receive" rather than a -/// failure. An interrupted syscall carries no such information and is -/// retried instead of being misclassified as quiet. +/// Like [`recv_request`], but any error, including a configured read +/// timeout, means "nothing more to receive" rather than a failure. fn recv_request_timeout(socket: &UdpSocket) -> Option> { let mut buf = [0_u8; 8192]; - loop { - match socket.recv_from(&mut buf) { - Ok((size, _)) => return Some(buf[..size].to_vec()), - Err(err) if err.kind() == ErrorKind::Interrupted => continue, - Err(_) => return None, - } - } + recv_from_retrying_interrupted(socket, &mut buf) + .ok() + .map(|(size, _)| buf[..size].to_vec()) } fn assert_started(outcome: OpenOutcome) -> NegotiatedParams {