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.
This commit is contained in:
3 files changed
+119
-49
No files matched your search
@@ -4,7 +4,7 @@
|
|||||||
mod in_tree_server;
|
mod in_tree_server;
|
||||||
|
|
||||||
use std::{
|
use std::{
|
||||||
io::{ErrorKind, Read},
|
io::{self, ErrorKind, Read},
|
||||||
net::{SocketAddr, UdpSocket},
|
net::{SocketAddr, UdpSocket},
|
||||||
process::{Command, Output, Stdio},
|
process::{Command, Output, Stdio},
|
||||||
sync::mpsc,
|
sync::mpsc,
|
||||||
@@ -1177,35 +1177,68 @@ fn start_open_failure_server_inner(
|
|||||||
FakeServer { addr, done }
|
FakeServer { addr, done }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A fake server's `recv_from`, retrying an interrupted syscall
|
/// A `recv_from` that retries an interrupted syscall
|
||||||
/// (`EINTR`/`ErrorKind::Interrupted`) instead of panicking on it: real
|
/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright: real
|
||||||
/// signal delivery (e.g. process-group signals during CI) can interrupt a
|
/// signal delivery (e.g. process-group signals during CI) can interrupt a
|
||||||
/// blocking receive here without there being anything actually wrong.
|
/// 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<u8>, SocketAddr) {
|
fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
|
||||||
let mut buf = [0_u8; 2048];
|
let mut buf = [0_u8; 2048];
|
||||||
loop {
|
match recv_from_retrying_interrupted(socket, &mut buf) {
|
||||||
match socket.recv_from(&mut buf) {
|
Ok((size, peer)) => (buf[..size].to_vec(), peer),
|
||||||
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}"),
|
Err(err) => panic!("fake server recv_from failed: {err}"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
/// Like [`recv_request`], but treats any non-interrupting error (including a
|
/// Like [`recv_request`], but treats any error, including a configured read
|
||||||
/// configured read timeout) as "nothing more to receive" rather than a
|
/// timeout, as "nothing more to receive" rather than a failure, since
|
||||||
/// failure, since callers use this to detect the peer going quiet. An
|
/// callers use this to detect the peer going quiet.
|
||||||
/// interrupted syscall carries no such information and is retried instead of
|
|
||||||
/// being misclassified as quiet.
|
|
||||||
fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
|
fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
|
||||||
let mut buf = [0_u8; 2048];
|
let mut buf = [0_u8; 2048];
|
||||||
loop {
|
recv_from_retrying_interrupted(socket, &mut buf)
|
||||||
match socket.recv_from(&mut buf) {
|
.ok()
|
||||||
Ok((size, peer)) => return Some((buf[..size].to_vec(), peer)),
|
.map(|(size, peer)| (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<u8> {
|
fn open_reply(flags: u8, token: u64, params: &Params) -> Vec<u8> {
|
||||||
|
|||||||
@@ -153,14 +153,18 @@ impl ErrorQueueControlBuffer {
|
|||||||
|
|
||||||
/// Nonblocking drain of a single record from `fd`'s `MSG_ERRQUEUE`.
|
/// Nonblocking drain of a single record from `fd`'s `MSG_ERRQUEUE`.
|
||||||
///
|
///
|
||||||
/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) or
|
/// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) — the
|
||||||
/// when this one read was interrupted (`EINTR`). TX timestamp capture is
|
/// caller in [`super::drain_tx_timestamps`] stops its drain there, exactly
|
||||||
/// optional best-effort metadata, not a network operation: an interrupted
|
/// as it always has. An interrupted read (`EINTR`) is different: it says
|
||||||
/// read here is not a socket/network error and must not become one, so it is
|
/// nothing about whether the queue is empty, so treating it the same as
|
||||||
/// treated as no record obtained on this drain rather than retried in a
|
/// `EWOULDBLOCK` could end a drain early and strand a still-queued record
|
||||||
/// loop. The bounded caller in [`super::drain_tx_timestamps`] naturally
|
/// past the point its matching probe is looked up and removed (see the
|
||||||
/// revisits the queue on its next call, which is how every other record is
|
/// crate's `AGENTS.md`). TX timestamp capture is optional best-effort
|
||||||
/// picked up too.
|
/// 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,
|
/// `SOF_TIMESTAMPING_OPT_TSONLY` notifications carry no meaningful payload,
|
||||||
/// so a zero-length iovec is enough: correlation is by `id`, never by
|
/// so a zero-length iovec is enough: correlation is by `id`, never by
|
||||||
/// payload content.
|
/// payload content.
|
||||||
@@ -181,7 +185,8 @@ pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result<Option<ErrorQ
|
|||||||
.map_err(|_| io::Error::from(io::ErrorKind::InvalidData))?;
|
.map_err(|_| io::Error::from(io::ErrorKind::InvalidData))?;
|
||||||
Ok(Some(classify(cmsgs)))
|
Ok(Some(classify(cmsgs)))
|
||||||
}
|
}
|
||||||
Err(Errno::EWOULDBLOCK | Errno::EINTR) => Ok(None),
|
Err(Errno::EWOULDBLOCK) => Ok(None),
|
||||||
|
Err(Errno::EINTR) => Ok(Some(ErrorQueueRecord::Ignored)),
|
||||||
Err(errno) => Err(io::Error::from(errno)),
|
Err(errno) => Err(io::Error::from(errno)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,12 +4,12 @@ mod in_tree_server;
|
|||||||
mod real_irtt;
|
mod real_irtt;
|
||||||
|
|
||||||
use std::{
|
use std::{
|
||||||
io::ErrorKind,
|
io::{self, ErrorKind},
|
||||||
net::{SocketAddr, UdpSocket},
|
net::{SocketAddr, UdpSocket},
|
||||||
process::Command,
|
process::Command,
|
||||||
sync::mpsc,
|
sync::mpsc,
|
||||||
thread::{self, JoinHandle},
|
thread::{self, JoinHandle},
|
||||||
time::{Duration, SystemTime},
|
time::{Duration, Instant, SystemTime},
|
||||||
};
|
};
|
||||||
|
|
||||||
use irtt_client::{Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, SocketConfig};
|
use irtt_client::{Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, SocketConfig};
|
||||||
@@ -448,32 +448,64 @@ where
|
|||||||
FakeServer { addr, rx, done }
|
FakeServer { addr, rx, done }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A fake server's `recv_from`, retrying an interrupted syscall
|
/// A `recv_from` that retries an interrupted syscall
|
||||||
/// (`EINTR`/`ErrorKind::Interrupted`) instead of panicking on it.
|
/// (`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<u8>, SocketAddr) {
|
fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
|
||||||
let mut buf = [0_u8; 8192];
|
let mut buf = [0_u8; 8192];
|
||||||
loop {
|
match recv_from_retrying_interrupted(socket, &mut buf) {
|
||||||
match socket.recv_from(&mut buf) {
|
Ok((size, peer)) => (buf[..size].to_vec(), peer),
|
||||||
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}"),
|
Err(err) => panic!("fake server recv_from failed: {err}"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
/// Like [`recv_request`], but any non-interrupting error (including a
|
/// Like [`recv_request`], but any error, including a configured read
|
||||||
/// configured read timeout) means "nothing more to receive" rather than a
|
/// timeout, means "nothing more to receive" rather than a failure.
|
||||||
/// failure. An interrupted syscall carries no such information and is
|
|
||||||
/// retried instead of being misclassified as quiet.
|
|
||||||
fn recv_request_timeout(socket: &UdpSocket) -> Option<Vec<u8>> {
|
fn recv_request_timeout(socket: &UdpSocket) -> Option<Vec<u8>> {
|
||||||
let mut buf = [0_u8; 8192];
|
let mut buf = [0_u8; 8192];
|
||||||
loop {
|
recv_from_retrying_interrupted(socket, &mut buf)
|
||||||
match socket.recv_from(&mut buf) {
|
.ok()
|
||||||
Ok((size, _)) => return Some(buf[..size].to_vec()),
|
.map(|(size, _)| buf[..size].to_vec())
|
||||||
Err(err) if err.kind() == ErrorKind::Interrupted => continue,
|
|
||||||
Err(_) => return None,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn assert_started(outcome: OpenOutcome) -> NegotiatedParams {
|
fn assert_started(outcome: OpenOutcome) -> NegotiatedParams {
|
||||||
|
|||||||
Reference in new issue
Block a user