From ca2389174fd0e103fe645fd3a9f6ee45b3c69429 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 17 Aug 2026 00:23:24 +0700 Subject: [PATCH 1/2] fix(io): retry interrupted receives Real signal delivery on Linux (observed via x86_64-musl workspace testing) can interrupt blocking and nonblocking recv/recvmsg calls with EINTR. Several production and test receive paths treated that as a fatal error rather than a transient interruption: - Client::open's blocking receive loop and Client::recv_once/ recv_available's blocking receive returned ClientError::Socket on Interrupted instead of retrying within their existing timeout budget. - AsyncClient::poll_open and poll_recv's nonblocking recvmsg path (Linux ancillary) failed the same way instead of retrying through the existing readiness/work-budget machinery. - The server's ancillary receive loop could terminate the listener on an interrupted nonblocking recvmsg. - The client's MSG_ERRQUEUE TX-timestamp drain turned an interrupted read of optional metadata into a fatal ClientError::Socket. - Fake-server test helpers panicked or misclassified EINTR as a timeout instead of retrying it. Interrupted is now retried transparently everywhere a receive can observe it, while preserving each path's existing timing contract: - Blocking Open reuses its already-absolute per-attempt deadline, so interruptions resume within the same attempt without retransmitting or extending the deadline. - recv_once/recv_available compute one absolute deadline per logical receive (when a timeout is configured) so repeated interruptions cannot extend it past the configured timeout, and restore the socket's configured read timeout afterward so a later call in the same recv_available budget is unaffected. - AsyncClient's poll_open already bounds its retry loop with OPEN_POLL_WORK_BUDGET; poll_recv gains an equivalent RECV_POLL_WORK_BUDGET, since unlike WouldBlock, Interrupted does not clear Tokio's readiness state, so an unbounded retry loop was otherwise possible under sustained signal delivery. - The server's receive loop retries Interrupted the same way it already retries WouldBlock, yielding to the runtime after a bounded number of consecutive interruptions for the same reason. - The error-queue drain treats an interrupted read as no record obtained this drain (deferring to the next natural drain point) rather than a socket error, matching its existing EWOULDBLOCK handling. Deterministic regression tests use a small test-only injection hook (mirroring existing send-injection test hooks) rather than real signal timing, covering: Open completing despite interruptions with no retransmit, Open's deadline not extending under repeated interruptions, recv_once classifying a reply normally despite interruptions, recv_once's timeout not extending under repeated interruptions, and the AsyncClient equivalents for open and recv. No protocol, session, or timeout-classification behavior changes; no new dependencies; send-side EINTR handling is unchanged and out of scope. --- crates/irtt-cli/tests/client_multi_target.rs | 31 ++++- crates/irtt-client/src/async_client.rs | 73 +++++++++- crates/irtt-client/src/async_client/tests.rs | 47 +++++++ crates/irtt-client/src/client.rs | 129 ++++++++++++++++-- crates/irtt-client/src/client/tests/open.rs | 57 ++++++++ .../probes_replies/interrupted_receive.rs | 80 +++++++++++ .../src/client/tests/probes_replies/mod.rs | 1 + .../src/receive/linux/error_queue.rs | 11 +- crates/irtt-client/tests/support/mod.rs | 25 +++- crates/irtt-server/src/socket_io/ancillary.rs | 26 +++- 10 files changed, 448 insertions(+), 32 deletions(-) create mode 100644 crates/irtt-client/src/client/tests/probes_replies/interrupted_receive.rs diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index c78691c..5b823f1 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::Read, + io::{ErrorKind, Read}, net::{SocketAddr, UdpSocket}, process::{Command, Output, Stdio}, sync::mpsc, @@ -1177,18 +1177,35 @@ 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 +/// signal delivery (e.g. process-group signals during CI) can interrupt a +/// blocking receive here without there being anything actually wrong. fn recv_request(socket: &UdpSocket) -> (Vec, SocketAddr) { let mut buf = [0_u8; 2048]; - let (size, peer) = socket.recv_from(&mut buf).unwrap(); - (buf[..size].to_vec(), peer) + 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}"), + } + } } +/// 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. fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec, SocketAddr)> { let mut buf = [0_u8; 2048]; - socket - .recv_from(&mut buf) - .ok() - .map(|(size, peer)| (buf[..size].to_vec(), peer)) + 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, + } + } } fn open_reply(flags: u8, token: u64, params: &Params) -> Vec { diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index ec9289e..2c2bfa6 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -41,6 +41,19 @@ use crate::{ /// so one managed target cannot retain the current-thread runtime. const OPEN_POLL_WORK_BUDGET: usize = 64; +/// Maximum immediately available receive work performed by one +/// [`AsyncClient::poll_recv`] poll. +/// +/// A `WouldBlock` readiness false positive is naturally bounded: `try_io` +/// clears the reactor's readiness on `WouldBlock`, so the next +/// `poll_recv_ready` genuinely awaits a fresh edge. An `Interrupted` +/// (`EINTR`) nonblocking `recvmsg` does not clear that readiness — the +/// syscall was interrupted, not told there was nothing to read — so without +/// an explicit bound a sustained signal delivery rate could otherwise retry +/// in a tight loop on this executor thread. Bounding both cases the same way +/// this crate already bounds open-attempt work keeps that impossible. +const RECV_POLL_WORK_BUDGET: usize = 64; + #[cfg(test)] use crate::client::ProbeSendTimestamps; @@ -174,6 +187,11 @@ struct AsyncClientTestHooks { sends: RefCell>, send_attempts: Cell, receive_would_block: Cell, + /// Number of subsequent receive attempts (`poll_open` or `poll_recv`) + /// that should report `io::ErrorKind::Interrupted` instead of performing + /// a real nonblocking receive, deterministically exercising interrupted- + /// syscall retry behavior. + inject_recv_interrupted: Cell, fail_open_dscp: Cell, fail_cleanup_send: Cell, fail_peer_close_dscp: Cell, @@ -205,6 +223,18 @@ impl AsyncClientTestHooks { fn take_probe_timestamps(&self) -> Option { self.probe_timestamps.borrow_mut().pop_front() } + + /// Consumes one pending injected interruption, if any, reporting whether + /// the caller should report `io::ErrorKind::Interrupted` instead of + /// attempting a real receive. + fn take_injected_interrupted(&self) -> bool { + let remaining = self.inject_recv_interrupted.get(); + if remaining == 0 { + return false; + } + self.inject_recv_interrupted.set(remaining - 1); + true + } } /// Low-level Tokio IRTT client for one connected UDP target. @@ -408,9 +438,26 @@ impl AsyncClient { Poll::Ready(Ok(true)) => {} } + #[cfg(test)] + if self.test_hooks.take_injected_interrupted() { + continue; + } let datagram = match try_recv_tokio_datagram(&self.socket, &mut state.buffer) { Ok(datagram) => datagram, - Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue, + // WouldBlock is a readiness false positive; Interrupted + // (EINTR) is a transient nonblocking-recvmsg interruption. + // Both retry through the surrounding `OPEN_POLL_WORK_BUDGET`- + // bounded loop above, which re-checks readiness before the + // next attempt and yields back to the executor once that + // budget is exhausted. + Err(error) + if matches!( + error.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted + ) => + { + continue + } Err(error) => return Poll::Ready(Err(ClientError::Socket(error))), }; if datagram.received_at.mono > state.deadline() { @@ -613,7 +660,7 @@ impl AsyncClient { return Poll::Ready(Err(error)); } - loop { + for _ in 0..RECV_POLL_WORK_BUDGET { match self.socket.poll_recv_ready(cx) { Poll::Pending => return Poll::Pending, Poll::Ready(Ok(())) => {} @@ -628,9 +675,23 @@ impl AsyncClient { .set(self.test_hooks.receive_would_block.get() - 1); continue; } + #[cfg(test)] + if self.test_hooks.take_injected_interrupted() { + continue; + } let datagram = match try_recv_tokio_datagram(&self.socket, &mut self.recv_buffer) { Ok(datagram) => datagram, - Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue, + // See `RECV_POLL_WORK_BUDGET`: both a readiness false + // positive and an interrupted nonblocking recvmsg retry + // within this bounded loop rather than failing the receive. + Err(error) + if matches!( + error.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted + ) => + { + continue + } Err(error) => return Poll::Ready(Err(ClientError::Socket(error))), }; @@ -659,6 +720,12 @@ impl AsyncClient { } return Poll::Ready(Ok(events)); } + + // Truncated known immediate receive work, exactly as `poll_open` + // does: arrange for another poll even if no new socket-readiness + // edge occurs, rather than spinning this executor thread further. + cx.waker().wake_by_ref(); + Poll::Pending } pub(crate) fn poll_close( diff --git a/crates/irtt-client/src/async_client/tests.rs b/crates/irtt-client/src/async_client/tests.rs index 35d3515..9e1278b 100644 --- a/crates/irtt-client/src/async_client/tests.rs +++ b/crates/irtt-client/src/async_client/tests.rs @@ -883,6 +883,53 @@ fn receive_retries_false_readiness_and_shares_packet_classification() { assert_eq!(server.finish().len(), 2); } +#[test] +fn receive_retries_interrupted_syscall_and_shares_packet_classification() { + let server = start_server(move |socket, tx| { + let (params, peer) = open_session(&socket, &tx, None); + let (probe_packet, _) = recv_packet(&socket, &tx); + let sequence = echo_request_sequence(&probe_packet, None); + socket + .send_to( + &echo_reply(¶ms, sequence, TOKEN, flags::FLAG_REPLY, None), + peer, + ) + .unwrap(); + }); + + runtime().block_on(async { + let mut client = opened_client(server.addr, 0).await; + client.send_probe().await.unwrap(); + client.test_hooks.inject_recv_interrupted.set(2); + + assert!(matches!( + client.recv().await.unwrap().as_slice(), + [ClientEvent::EchoReply { seq: 0, .. }] + )); + }); + assert_eq!(server.finish().len(), 2); +} + +#[test] +fn open_retries_interrupted_syscall_within_the_same_attempt() { + let server = start_open_server(None, 0); + runtime().block_on(async { + let mut client = AsyncClient::connect(config(server.addr, None, 0)) + .await + .unwrap(); + client.test_hooks.inject_recv_interrupted.set(3); + + assert!(matches!( + client.open().await, + Ok(OpenOutcome::Started { .. }) + )); + // Only one open request reached the server: interruptions were + // absorbed inside the single open attempt, with no retransmit. + assert_eq!(client.test_hooks.send_attempts.get(), 1); + }); + server.finish(); +} + #[test] fn peer_close_dscp_failure_preserves_events_and_logical_ownership() { let server = start_peer_close_server(); diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index e8cc940..3ada7f5 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -14,7 +14,7 @@ use crate::{ config::{ClientConfig, RecvBudget}, error::ClientError, event::{ClientEvent, OpenOutcome}, - receive::{drain_tx_timestamps, recv_datagram, try_enable_tx_timestamping}, + receive::{drain_tx_timestamps, recv_datagram, try_enable_tx_timestamping, ReceivedDatagram}, session::machine::{ recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent, SessionMachine, MAX_OPEN_PACKET_SIZE, @@ -69,6 +69,10 @@ struct ClientTestHooks { fail_dscp_restore: Cell, last_restored_read_timeout: Cell>, prepared_active_session_before_dscp: Cell, + /// Number of subsequent receive attempts that should report + /// `io::ErrorKind::Interrupted` before a real receive is attempted. + /// Decremented on every injected receive; `0` means inject nothing. + inject_recv_interrupted: Cell, } #[cfg(test)] @@ -264,18 +268,40 @@ impl Client { self.recv_once_inner() } - fn recv_once_inner(&mut self) -> Result, ClientError> { - let datagram = match recv_datagram(&self.socket, &mut self.recv_buffer) { - Ok(datagram) => datagram, - Err(err) - if matches!( - err.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ) => - { - return Ok(vec![]); + /// Receive one datagram into `self.recv_buffer`. In test builds, a + /// pending `inject_recv_interrupted` count reports + /// `io::ErrorKind::Interrupted` instead of performing a real receive, + /// deterministically exercising interrupted-syscall retry behavior. + fn recv_datagram_once(&mut self) -> Result { + #[cfg(test)] + { + let remaining = self.test_hooks.inject_recv_interrupted.get(); + if remaining > 0 { + self.test_hooks.inject_recv_interrupted.set(remaining - 1); + return Err(io::Error::from(io::ErrorKind::Interrupted)); } - Err(err) => return Err(ClientError::Socket(err)), + } + recv_datagram(&self.socket, &mut self.recv_buffer) + } + + /// Same as [`Self::recv_datagram_once`], receiving into a caller-supplied + /// buffer instead of `self.recv_buffer` (used before the negotiated + /// receive buffer size is known, i.e. during Open). + fn recv_datagram_once_into(&self, buf: &mut [u8]) -> Result { + #[cfg(test)] + { + let remaining = self.test_hooks.inject_recv_interrupted.get(); + if remaining > 0 { + self.test_hooks.inject_recv_interrupted.set(remaining - 1); + return Err(io::Error::from(io::ErrorKind::Interrupted)); + } + } + recv_datagram(&self.socket, buf) + } + + fn recv_once_inner(&mut self) -> Result, ClientError> { + let Some(datagram) = self.recv_datagram_retrying_interrupted()? else { + return Ok(vec![]); }; // One final bounded drain before processing the reply, so a TX @@ -297,6 +323,71 @@ impl Client { Ok(events) } + /// Receive one datagram, transparently retrying an interrupted syscall + /// (`io::ErrorKind::Interrupted`, e.g. `EINTR`) without treating it as a + /// genuine socket error, timeout, or empty receive. + /// + /// One logical receive keeps one timeout budget: when a receive timeout + /// is configured, an absolute deadline is computed once here and each + /// retry after an interruption restores the socket's remaining time + /// against that same deadline, so repeated interruptions cannot extend + /// this receive past its configured timeout. An unconfigured timeout + /// (`None`) keeps blocking indefinitely after an interruption, matching + /// its no-timeout contract. The socket's read timeout is restored to the + /// caller-configured value before returning, so a later call in the same + /// `recv_available` budget is unaffected by an earlier interruption here. + fn recv_datagram_retrying_interrupted( + &mut self, + ) -> Result, ClientError> { + let configured_timeout = self.runtime.config().socket_config.recv_timeout; + let deadline = match configured_timeout { + Some(timeout) => Some( + Instant::now() + .checked_add(timeout) + .ok_or(ClientError::DurationOverflow)?, + ), + None => None, + }; + let mut timeout_narrowed = false; + + let result = loop { + match self.recv_datagram_once() { + Ok(datagram) => break Ok(Some(datagram)), + Err(err) + if matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + break Ok(None); + } + Err(err) if err.kind() == io::ErrorKind::Interrupted => { + if let Some(deadline) = deadline { + let Some(remaining) = deadline.checked_duration_since(Instant::now()) + else { + break Ok(None); + }; + if remaining.is_zero() { + break Ok(None); + } + if let Err(err) = self.socket.set_read_timeout(Some(remaining)) { + break Err(ClientError::Socket(err)); + } + timeout_narrowed = true; + } + } + Err(err) => break Err(ClientError::Socket(err)), + } + }; + + if timeout_narrowed { + if let Err(err) = self.socket.set_read_timeout(configured_timeout) { + return Err(ClientError::Socket(err)); + } + } + result + } + /// Receive and classify datagrams until a receive produces no events or the /// receive budget is exhausted. pub fn recv_available(&mut self, budget: RecvBudget) -> Result, ClientError> { @@ -496,12 +587,22 @@ impl Client { } self.socket.set_read_timeout(Some(remaining))?; - let datagram = match recv_datagram(&self.socket, &mut buf) { + let datagram = match self.recv_datagram_once_into(&mut buf) { Ok(datagram) => datagram, + // WouldBlock/TimedOut mean the attempt's own deadline + // elapsed; Interrupted (EINTR) means the syscall was + // merely interrupted and carries no timing information of + // its own. Both retry through the same loop, which + // recomputes `remaining` and re-applies it as the read + // timeout on every iteration — so an interruption resumes + // within this same absolute deadline rather than + // resetting or extending it. Err(err) if matches!( err.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + io::ErrorKind::WouldBlock + | io::ErrorKind::TimedOut + | io::ErrorKind::Interrupted ) => { if Instant::now() >= deadline { diff --git a/crates/irtt-client/src/client/tests/open.rs b/crates/irtt-client/src/client/tests/open.rs index 0c5515c..142af3b 100644 --- a/crates/irtt-client/src/client/tests/open.rs +++ b/crates/irtt-client/src/client/tests/open.rs @@ -64,6 +64,63 @@ fn open_retries_after_first_timeout() { server.join(); } +#[test] +fn open_survives_interrupted_receive_within_same_attempt() { + let server = start_fake_server(|socket, tx| { + let (first, peer) = recv_request(&socket, &tx); + let params = Params::decode(&first[4..]).unwrap(); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + }); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(500)], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + // Several interrupted receive attempts before the real reply arrives + // must not be treated as a timeout, must not retransmit the open + // request, and must not advance to a next configured attempt. + client.test_hooks.inject_recv_interrupted.set(5); + + let outcome = client.open().unwrap(); + assert_open_started(outcome); + // Exactly one request reached the server: interruptions were absorbed + // inside the single configured attempt. + assert_eq!(server.rx.iter().take(1).count(), 1); + server.join(); +} + +#[test] +fn open_interrupted_receive_does_not_extend_the_attempt_deadline() { + // A fake server that never replies: with a short attempt timeout, the + // real question is whether interruptions can push the effective wait + // past the configured deadline. They must not: the deadline stays + // absolute across any number of interruptions. + let server = start_fake_server(|socket, tx| { + let _ = recv_request(&socket, &tx); + }); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(200)], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + // Each interruption carries no timing information of its own, so the + // loop must keep re-checking the real deadline rather than treating an + // interruption as "more time available" or a reason to fail outright. + client.test_hooks.inject_recv_interrupted.set(20); + + let started = Instant::now(); + let result = client.open(); + let elapsed = started.elapsed(); + + assert!(matches!(result, Err(ClientError::OpenTimeout))); + assert!( + elapsed < Duration::from_millis(800), + "interrupted receives must not extend the attempt deadline, took {elapsed:?}" + ); + server.join(); +} + #[test] fn open_timeout_after_all_timeouts() { let server = timeout_server(Duration::from_millis(700)); diff --git a/crates/irtt-client/src/client/tests/probes_replies/interrupted_receive.rs b/crates/irtt-client/src/client/tests/probes_replies/interrupted_receive.rs new file mode 100644 index 0000000..64ce90d --- /dev/null +++ b/crates/irtt-client/src/client/tests/probes_replies/interrupted_receive.rs @@ -0,0 +1,80 @@ +use super::*; + +/// An interrupted receive (`io::ErrorKind::Interrupted`, e.g. `EINTR`) must be +/// invisible to the caller: retried transparently, with the eventual reply +/// classified exactly as if no interruption had occurred. +#[test] +fn interrupted_receive_is_retried_and_the_reply_is_still_classified() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + let mut buf = [0_u8; 2048]; + let (size, _) = socket.recv_from(&mut buf).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + let _ = size; + let ts = TimestampFields::default(); + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); + socket.send_to(&reply_packet, peer).unwrap(); + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_secs(2)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + client.test_hooks.inject_recv_interrupted.set(10); + let events = client.recv_once().unwrap(); + + assert_eq!(events.len(), 1); + assert!( + matches!(&events[0], ClientEvent::EchoReply { seq: 0, .. }), + "expected a normal EchoReply despite interrupted receive attempts, got {:?}", + events[0] + ); + + client.close().unwrap(); + server.join(); +} + +/// One logical receive keeps one timeout budget: repeated interruptions must +/// not extend a `recv_once` call past its configured receive timeout. +#[test] +fn interrupted_receive_does_not_extend_the_configured_timeout() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(100)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + + // No reply is ever coming; only interrupted attempts occur before the + // real socket takes over and legitimately times out. + client.test_hooks.inject_recv_interrupted.set(20); + let started = Instant::now(); + let events = client.recv_once().unwrap(); + let elapsed = started.elapsed(); + + assert!(events.is_empty()); + assert!( + elapsed < Duration::from_millis(500), + "interrupted receives must not extend the configured receive timeout, took {elapsed:?}" + ); + + client.close().unwrap(); + server.join(); +} diff --git a/crates/irtt-client/src/client/tests/probes_replies/mod.rs b/crates/irtt-client/src/client/tests/probes_replies/mod.rs index f02efd3..1f779ba 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/mod.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/mod.rs @@ -20,6 +20,7 @@ fn kernel_rx_timestamp_unavailable_skip(test_name: &str) { } mod burst_recv; +mod interrupted_receive; mod packet_length; mod probe_schedule; mod reply_classification; diff --git a/crates/irtt-client/src/receive/linux/error_queue.rs b/crates/irtt-client/src/receive/linux/error_queue.rs index 58e280d..5657f7f 100644 --- a/crates/irtt-client/src/receive/linux/error_queue.rs +++ b/crates/irtt-client/src/receive/linux/error_queue.rs @@ -153,7 +153,14 @@ impl ErrorQueueControlBuffer { /// Nonblocking drain of a single record from `fd`'s `MSG_ERRQUEUE`. /// -/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`). +/// 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. /// `SOF_TIMESTAMPING_OPT_TSONLY` notifications carry no meaningful payload, /// so a zero-length iovec is enough: correlation is by `id`, never by /// payload content. @@ -174,7 +181,7 @@ pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result Ok(None), + Err(Errno::EWOULDBLOCK | Errno::EINTR) => Ok(None), 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 e96dfb0..cd75037 100644 --- a/crates/irtt-client/tests/support/mod.rs +++ b/crates/irtt-client/tests/support/mod.rs @@ -4,6 +4,7 @@ mod in_tree_server; mod real_irtt; use std::{ + io::ErrorKind, net::{SocketAddr, UdpSocket}, process::Command, sync::mpsc, @@ -447,17 +448,31 @@ where FakeServer { addr, rx, done } } +/// A fake server's `recv_from`, retrying an interrupted syscall +/// (`EINTR`/`ErrorKind::Interrupted`) instead of panicking on it. fn recv_request(socket: &UdpSocket) -> (Vec, SocketAddr) { let mut buf = [0_u8; 8192]; - let (size, peer) = socket.recv_from(&mut buf).unwrap(); - (buf[..size].to_vec(), peer) + 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}"), + } + } } +/// 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. fn recv_request_timeout(socket: &UdpSocket) -> Option> { let mut buf = [0_u8; 8192]; - match socket.recv_from(&mut buf) { - Ok((size, _)) => Some(buf[..size].to_vec()), - Err(_) => None, + 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, + } } } diff --git a/crates/irtt-server/src/socket_io/ancillary.rs b/crates/irtt-server/src/socket_io/ancillary.rs index bfe277b..f3953c7 100644 --- a/crates/irtt-server/src/socket_io/ancillary.rs +++ b/crates/irtt-server/src/socket_io/ancillary.rs @@ -152,6 +152,18 @@ pub(super) async fn receive_bound( receive_ancillary(socket, buffer, false).await } +/// Consecutive interrupted (`EINTR`) nonblocking receives tolerated before +/// this receive loop explicitly yields to the runtime. +/// +/// `try_io` only clears the reactor's readiness state on `WouldBlock`; an +/// `Interrupted` result leaves it untouched, so `socket.readable().await` +/// can resolve immediately again without a genuine new readiness edge. Under +/// a sustained signal delivery rate that could otherwise retry in a tight +/// loop without ever suspending this task. Yielding periodically bounds that +/// without treating the interruption as any kind of failure or terminating +/// the listener. +const INTERRUPTED_YIELD_THRESHOLD: u32 = 64; + /// `require_destination` is the listener's own wildcard-ness. It decides both /// what this receive must recover and how severe a truncated control buffer is. async fn receive_ancillary( @@ -159,12 +171,24 @@ async fn receive_ancillary( buffer: &mut [u8], require_destination: bool, ) -> io::Result> { + let mut consecutive_interrupted = 0_u32; loop { socket.readable().await?; match socket.try_io(Interest::READABLE, || { receive_once(socket, buffer, require_destination) }) { - Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + consecutive_interrupted = 0; + continue; + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => { + consecutive_interrupted += 1; + if consecutive_interrupted >= INTERRUPTED_YIELD_THRESHOLD { + consecutive_interrupted = 0; + tokio::task::yield_now().await; + } + continue; + } result => return result, } } -- 2.54.0 From a598dbdc5c66a7452523e2b6bcb5eaf6b21295f1 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 17 Aug 2026 00:32:37 +0700 Subject: [PATCH 2/2] 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 { -- 2.54.0