diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index c78691c..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::Read, + io::{self, ErrorKind, Read}, net::{SocketAddr, UdpSocket}, process::{Command, Output, Stdio}, sync::mpsc, @@ -1177,16 +1177,66 @@ fn start_open_failure_server_inner( FakeServer { addr, done } } -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) +/// 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]; + 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 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]; - socket - .recv_from(&mut buf) + recv_from_retrying_interrupted(socket, &mut buf) .ok() .map(|(size, peer)| (buf[..size].to_vec(), peer)) } 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..c224919 100644 --- a/crates/irtt-client/src/receive/linux/error_queue.rs +++ b/crates/irtt-client/src/receive/linux/error_queue.rs @@ -153,7 +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`). +/// 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. @@ -175,6 +186,7 @@ pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result 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 e96dfb0..24274e0 100644 --- a/crates/irtt-client/tests/support/mod.rs +++ b/crates/irtt-client/tests/support/mod.rs @@ -4,11 +4,12 @@ mod in_tree_server; mod real_irtt; use std::{ + 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}; @@ -447,18 +448,64 @@ where FakeServer { addr, rx, done } } -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) +/// 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]; + 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 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]; - match socket.recv_from(&mut buf) { - Ok((size, _)) => Some(buf[..size].to_vec()), - Err(_) => None, - } + recv_from_retrying_interrupted(socket, &mut buf) + .ok() + .map(|(size, _)| buf[..size].to_vec()) } fn assert_started(outcome: OpenOutcome) -> NegotiatedParams { 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, } }