use std::collections::VecDeque;
use std::io::{self, Read, Write};
use std::net::{Shutdown, SocketAddr, TcpStream};
use std::pin::Pin;
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use std::task::{Context, Poll, Waker};
use std::time::{Duration, Instant};
use tokio::io::ReadBuf;
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, warn};
use crate::runtime::task::{StackSizeClass, ThreadPriority, block_on_sync, spawn_dedicated_thread};
use crate::runtime::worker_pool::{Job, SetLease, Worker, WorkerPool, WorkerRole};
pub const DEFAULT_READ_CHUNK: usize = 4096;
#[cfg(all(unix, not(target_os = "rtems")))]
const FIONREAD_REQUEST: libc::c_ulong = libc::FIONREAD as libc::c_ulong;
#[cfg(target_os = "rtems")]
const FIONREAD_REQUEST: libc::c_ulong = 0x4004_667F;
#[cfg(unix)]
pub fn pending_bytes<F: std::os::fd::AsRawFd>(sock: &F) -> io::Result<usize> {
let mut n: libc::c_int = 0;
let rc = unsafe {
libc::ioctl(
sock.as_raw_fd(),
FIONREAD_REQUEST as _,
&mut n as *mut libc::c_int,
)
};
if rc != 0 {
return Err(io::Error::last_os_error());
}
Ok(n.max(0) as usize)
}
#[cfg(not(unix))]
pub fn pending_bytes<F>(_sock: &F) -> io::Result<usize> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"FIONREAD unavailable on this platform",
))
}
pub fn is_socket_timeout(kind: io::ErrorKind) -> bool {
matches!(kind, io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut)
}
fn pump_thread_lost(role: &str, label: &str, what: &str) {
crate::runtime::log::errlog_sev_printf(
crate::runtime::log::ErrlogSevEnum::Major,
&format!(
"{label}: the {role} thread {what}; this connection is being torn \
down. Other connections are unaffected."
),
);
warn!(label, role, what, "blocking socket pump: a thread was lost");
}
pub const MAX_DIAL_WORKERS: usize = 4;
struct DialRequest {
target: SocketAddr,
reply: oneshot::Sender<io::Result<TcpStream>>,
}
struct DialQueue {
pending: VecDeque<DialRequest>,
busy: usize,
workers: usize,
}
pub struct DialPool {
name_prefix: &'static str,
priority: ThreadPriority,
queue: Mutex<DialQueue>,
work: Condvar,
}
impl DialPool {
pub const fn new(name_prefix: &'static str, priority: ThreadPriority) -> Self {
Self {
name_prefix,
priority,
queue: Mutex::new(DialQueue {
pending: VecDeque::new(),
busy: 0,
workers: 0,
}),
work: Condvar::new(),
}
}
pub fn worker_count(&self) -> usize {
self.lock().workers
}
pub fn queue_depth(&self) -> (usize, usize) {
let q = self.lock();
(q.pending.len(), q.busy)
}
pub fn dial(
&'static self,
target: SocketAddr,
) -> io::Result<oneshot::Receiver<io::Result<TcpStream>>> {
let (reply, rx) = oneshot::channel();
let req = DialRequest { target, reply };
let mut q = self.lock();
if q.pending.len() + q.busy < q.workers || q.workers >= MAX_DIAL_WORKERS {
q.pending.push_back(req);
drop(q);
self.work.notify_one();
return Ok(rx);
}
let index = q.workers;
q.workers += 1;
drop(q);
if let Err(e) = spawn_dedicated_thread(
format!("{} {index}", self.name_prefix),
self.priority,
StackSizeClass::Small,
move || self.worker_loop(),
) {
self.lock().workers -= 1;
return Err(e);
}
self.lock().pending.push_back(req);
self.work.notify_one();
Ok(rx)
}
fn worker_loop(&self) -> ! {
loop {
let req = {
let mut q = self.lock();
loop {
if let Some(req) = q.pending.pop_front() {
q.busy += 1;
break req;
}
q = self.work.wait(q).unwrap_or_else(|e| e.into_inner());
}
};
let dialed = (!req.reply.is_closed()).then(|| TcpStream::connect(req.target));
self.lock().busy -= 1;
if let Some(dialed) = dialed {
let _ = req.reply.send(dialed);
}
}
}
fn lock(&self) -> MutexGuard<'_, DialQueue> {
self.queue.lock().unwrap_or_else(|e| e.into_inner())
}
}
pub struct ChannelReader {
rx: mpsc::Receiver<Vec<u8>>,
cur: Vec<u8>,
pos: usize,
}
impl ChannelReader {
pub fn new(rx: mpsc::Receiver<Vec<u8>>) -> Self {
Self {
rx,
cur: Vec::new(),
pos: 0,
}
}
}
impl tokio::io::AsyncRead for ChannelReader {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
if buf.remaining() == 0 {
return Poll::Ready(Ok(()));
}
let me = &mut *self;
loop {
if me.pos < me.cur.len() {
let n = (me.cur.len() - me.pos).min(buf.remaining());
buf.put_slice(&me.cur[me.pos..me.pos + n]);
me.pos += n;
if me.pos == me.cur.len() {
me.cur.clear();
me.pos = 0;
}
return Poll::Ready(Ok(()));
}
match me.rx.poll_recv(cx) {
Poll::Ready(Some(chunk)) => {
if chunk.is_empty() {
continue;
}
me.cur = chunk;
me.pos = 0;
}
Poll::Ready(None) => return Poll::Ready(Ok(())),
Poll::Pending => return Poll::Pending,
}
}
}
}
fn reader_pump(
sock: Arc<TcpStream>,
tx: mpsc::Sender<Vec<u8>>,
chunk_size: usize,
label: String,
read_timeout: Option<Duration>,
) {
let mut sock = &*sock;
let mut chunk = vec![0u8; chunk_size];
loop {
match wait_readable(sock, read_timeout.map(|t| Instant::now() + t)) {
Ok(true) => {}
Ok(false) => {
debug!(label, "blocking reader: receive timeout, ending connection");
break;
}
Err(e) => {
debug!(label, error = %e, "blocking reader: wait failed");
break;
}
}
let n = match sock.read(&mut chunk) {
Ok(0) => break,
Ok(n) => n,
Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) if e.kind() == io::ErrorKind::WouldBlock => continue,
Err(e) if is_socket_timeout(e.kind()) => {
debug!(label, "blocking reader: receive timeout, ending connection");
break;
}
Err(e) => {
debug!(label, error = %e, "blocking reader: read failed");
break;
}
};
if !matches!(block_on_sync(tx.send(chunk[..n].to_vec())), Ok(Ok(()))) {
break;
}
}
}
pub struct ReaderPumpGuard {
sock: Arc<TcpStream>,
label: String,
job: Option<Job>,
}
impl Drop for ReaderPumpGuard {
fn drop(&mut self) {
if let Some(job) = self.job.take() {
let _ = self.sock.shutdown(Shutdown::Both);
if job.join().is_err() {
pump_thread_lost("reader", &self.label, "panicked");
}
}
}
}
pub fn spawn_reader_pump(
worker: Worker,
sock: Arc<TcpStream>,
label: &str,
chunk_size: usize,
queue_depth: usize,
) -> (ChannelReader, ReaderPumpGuard) {
let read_timeout = sock_read_timeout(&sock);
spawn_reader_pump_with_timeout(worker, sock, label, chunk_size, queue_depth, read_timeout)
}
fn sock_read_timeout(sock: &TcpStream) -> Option<Duration> {
sock.read_timeout().ok().flatten()
}
fn spawn_reader_pump_with_timeout(
worker: Worker,
sock: Arc<TcpStream>,
label: &str,
chunk_size: usize,
queue_depth: usize,
read_timeout: Option<Duration>,
) -> (ChannelReader, ReaderPumpGuard) {
let (tx, rx) = mpsc::channel::<Vec<u8>>(queue_depth);
let pump_sock = sock.clone();
let pump_label = label.to_string();
let job = worker.run(move || reader_pump(pump_sock, tx, chunk_size, pump_label, read_timeout));
(
ChannelReader::new(rx),
ReaderPumpGuard {
sock,
label: label.to_string(),
job: Some(job),
},
)
}
#[derive(Default)]
struct WriteRoom {
waker: Mutex<Option<Waker>>,
}
impl WriteRoom {
fn park(&self, cx: &Context<'_>) {
*self.waker.lock().expect("write-room waker poisoned") = Some(cx.waker().clone());
}
fn wake(&self) {
let waker = self.waker.lock().expect("write-room waker poisoned").take();
if let Some(w) = waker {
w.wake();
}
}
}
pub struct ChannelWriter {
tx: mpsc::WeakSender<Vec<u8>>,
room: Arc<WriteRoom>,
}
fn write_closed() -> io::Error {
io::Error::new(
io::ErrorKind::BrokenPipe,
"the writer pump thread has ended",
)
}
impl tokio::io::AsyncWrite for ChannelWriter {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
if buf.is_empty() {
return Poll::Ready(Ok(0));
}
let Some(tx) = self.tx.upgrade() else {
return Poll::Ready(Err(write_closed()));
};
self.room.park(cx);
match tx.try_send(buf.to_vec()) {
Ok(()) => Poll::Ready(Ok(buf.len())),
Err(mpsc::error::TrySendError::Full(_)) => Poll::Pending,
Err(mpsc::error::TrySendError::Closed(_)) => Poll::Ready(Err(write_closed())),
}
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
#[cfg(target_os = "rtems")]
const POLLOUT_EVENT: libc::c_short = 0x0004;
#[cfg(target_os = "rtems")]
const POLLIN_EVENT: libc::c_short = 0x0001;
#[cfg(target_os = "rtems")]
const SEND_DONTWAIT: libc::c_int = 0x0080;
#[cfg(all(unix, not(target_os = "rtems")))]
const POLLOUT_EVENT: libc::c_short = libc::POLLOUT;
#[cfg(all(unix, not(target_os = "rtems")))]
const POLLIN_EVENT: libc::c_short = libc::POLLIN;
#[cfg(all(unix, not(target_os = "rtems")))]
const SEND_DONTWAIT: libc::c_int = libc::MSG_DONTWAIT;
#[cfg(unix)]
fn own_blocking_mode(sock: &TcpStream) -> io::Result<()> {
sock.set_nonblocking(true)
}
#[cfg(not(unix))]
fn own_blocking_mode(_sock: &TcpStream) -> io::Result<()> {
Ok(())
}
#[cfg(unix)]
fn wait_readable(sock: &TcpStream, deadline: Option<Instant>) -> io::Result<bool> {
use std::os::fd::AsRawFd;
loop {
let ms = match deadline {
Some(d) => {
let remaining = d.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Ok(false);
}
remaining.as_millis().max(1).min(libc::c_int::MAX as u128) as libc::c_int
}
None => -1,
};
let mut fds = libc::pollfd {
fd: sock.as_raw_fd(),
events: POLLIN_EVENT,
revents: 0,
};
let rc = unsafe { libc::poll(&mut fds, 1, ms) };
if rc > 0 {
return Ok(true);
}
if rc == 0 {
return Ok(false);
}
let e = io::Error::last_os_error();
if e.kind() != io::ErrorKind::Interrupted {
return Err(e);
}
}
}
#[cfg(not(unix))]
fn wait_readable(sock: &TcpStream, deadline: Option<Instant>) -> io::Result<bool> {
let Some(d) = deadline else {
sock.set_read_timeout(None)?;
return Ok(true);
};
let remaining = d.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Ok(false);
}
sock.set_read_timeout(Some(remaining.max(Duration::from_millis(1))))?;
Ok(true)
}
#[cfg(unix)]
fn wait_writable(sock: &TcpStream, deadline: Instant) -> io::Result<bool> {
use std::os::fd::AsRawFd;
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Ok(false);
}
let ms = remaining.as_millis().max(1).min(libc::c_int::MAX as u128) as libc::c_int;
let mut fds = libc::pollfd {
fd: sock.as_raw_fd(),
events: POLLOUT_EVENT,
revents: 0,
};
let rc = unsafe { libc::poll(&mut fds, 1, ms) };
if rc > 0 {
return Ok(true);
}
if rc == 0 {
return Ok(false);
}
let e = io::Error::last_os_error();
if e.kind() != io::ErrorKind::Interrupted {
return Err(e);
}
}
}
#[cfg(unix)]
fn write_some(sock: &TcpStream, buf: &[u8]) -> io::Result<usize> {
use std::os::fd::AsRawFd;
let n = unsafe {
libc::send(
sock.as_raw_fd(),
buf.as_ptr().cast(),
buf.len(),
SEND_DONTWAIT,
)
};
if n < 0 {
return Err(io::Error::last_os_error());
}
Ok(n as usize)
}
#[cfg(not(unix))]
fn wait_writable(sock: &TcpStream, deadline: Instant) -> io::Result<bool> {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Ok(false);
}
sock.set_write_timeout(Some(remaining.max(Duration::from_millis(1))))?;
Ok(true)
}
#[cfg(not(unix))]
fn write_some(sock: &TcpStream, buf: &[u8]) -> io::Result<usize> {
let mut sock = sock;
sock.write(buf)
}
pub fn write_frame_deadline(
sock: &TcpStream,
frame: &[u8],
send_timeout: Duration,
) -> io::Result<()> {
own_blocking_mode(sock)?;
let mut sock = sock;
let deadline = Instant::now() + send_timeout;
let mut off = 0;
while off < frame.len() {
if !wait_writable(sock, deadline)? {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"send deadline expired with the frame incomplete",
));
}
match write_some(sock, &frame[off..]) {
Ok(0) => {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"peer accepted no bytes",
));
}
Ok(n) => off += n,
Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
Err(e) if is_socket_timeout(e.kind()) => {}
Err(e) => return Err(e),
}
}
sock.flush()
}
fn writer_pump(
sock: Arc<TcpStream>,
mut rx: mpsc::Receiver<Vec<u8>>,
room: Arc<WriteRoom>,
send_timeout: Duration,
label: String,
) {
while let Ok(Some(frame)) = block_on_sync(rx.recv()) {
room.wake();
if let Err(e) = write_frame_deadline(&sock, &frame, send_timeout) {
debug!(label, error = %e, "blocking writer: send failed, ending connection");
break;
}
}
room.wake();
let _ = sock.shutdown(Shutdown::Both);
}
pub struct WriterPumpGuard {
frames: Option<mpsc::Sender<Vec<u8>>>,
label: String,
job: Option<Job>,
}
impl Drop for WriterPumpGuard {
fn drop(&mut self) {
drop(self.frames.take());
if let Some(job) = self.job.take() {
if job.join().is_err() {
pump_thread_lost("writer", &self.label, "panicked");
}
}
}
}
pub fn spawn_writer_pump(
worker: Worker,
sock: Arc<TcpStream>,
label: &str,
send_timeout: Duration,
queue_depth: usize,
) -> (ChannelWriter, WriterPumpGuard) {
let (tx, rx) = mpsc::channel::<Vec<u8>>(queue_depth);
let room = Arc::new(WriteRoom::default());
let adapter = ChannelWriter {
tx: tx.downgrade(),
room: room.clone(),
};
let pump_label = label.to_string();
let job = worker.run(move || writer_pump(sock, rx, room, send_timeout, pump_label));
(
adapter,
WriterPumpGuard {
frames: Some(tx),
label: label.to_string(),
job: Some(job),
},
)
}
pub struct GuardedReader {
inner: ChannelReader,
_guard: ReaderPumpGuard,
_lease: Arc<SetLease>,
}
impl tokio::io::AsyncRead for GuardedReader {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().inner).poll_read(cx, buf)
}
}
pub struct GuardedWriter {
inner: ChannelWriter,
_guard: WriterPumpGuard,
_lease: Arc<SetLease>,
}
impl tokio::io::AsyncWrite for GuardedWriter {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
Pin::new(&mut self.get_mut().inner).poll_write(cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().inner).poll_flush(cx)
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
}
}
#[derive(Clone, Debug)]
pub struct PumpConfig {
pub read_timeout: Duration,
pub send_timeout: Duration,
pub chunk_size: usize,
pub queue_depth: usize,
}
impl Default for PumpConfig {
fn default() -> Self {
Self {
read_timeout: Duration::from_secs(64_000),
send_timeout: Duration::from_secs(30),
chunk_size: DEFAULT_READ_CHUNK,
queue_depth: 1,
}
}
}
pub fn drive_socket_blocking(
pool: &WorkerPool<2>,
stream: TcpStream,
label: &str,
config: &PumpConfig,
) -> io::Result<(GuardedReader, GuardedWriter)> {
let _ = stream.set_nodelay(true);
own_blocking_mode(&stream)?;
stream.set_read_timeout(Some(config.read_timeout))?;
let (lease, [reader_worker, writer_worker]) = pool.acquire()?;
let lease = Arc::new(lease);
let stream = Arc::new(stream);
let (reader, reader_guard) = spawn_reader_pump_with_timeout(
reader_worker,
stream.clone(),
label,
config.chunk_size,
config.queue_depth,
Some(config.read_timeout),
);
let (writer, writer_guard) = spawn_writer_pump(
writer_worker,
stream,
label,
config.send_timeout,
config.queue_depth,
);
Ok((
GuardedReader {
inner: reader,
_guard: reader_guard,
_lease: lease.clone(),
},
GuardedWriter {
inner: writer,
_guard: writer_guard,
_lease: lease,
},
))
}
pub fn circuit_roster(
reader_priority: ThreadPriority,
writer_priority: ThreadPriority,
) -> [WorkerRole; 2] {
[
WorkerRole {
suffix: "reader",
stack: StackSizeClass::Small,
priority: reader_priority,
},
WorkerRole {
suffix: "writer",
stack: StackSizeClass::Small,
priority: writer_priority,
},
]
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::{TcpListener, TcpStream as StdTcpStream};
use std::thread;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
fn production_scope(src: &str) -> &str {
match src.find("\n#[cfg(test)]") {
Some(i) => &src[..i],
None => src,
}
}
fn code_only(src: &str) -> String {
src.lines()
.map(|line| match line.find("//") {
Some(i) => &line[..i],
None => line,
})
.collect::<Vec<_>>()
.join("\n")
}
#[test]
fn the_blocking_io_seam_has_no_async_runtime_symbols() {
let prod = code_only(production_scope(include_str!("blocking_io.rs")));
assert!(
prod.contains("fn drive_socket_blocking"),
"production slice no longer covers the seam"
);
let forbidden = [
concat!("tokio", "::net"),
concat!("tokio", "::time"),
concat!("tokio", "::", "spawn"),
concat!("block", "_in_place"),
concat!(".", "await"),
];
for token in forbidden {
assert_eq!(
prod.matches(token).count(),
0,
"the blocking I/O seam must not reference `{token}`: it has no async \
net/timer/spawn on RTEMS, and every await goes through `block_on_sync`"
);
}
}
#[test]
fn the_seam_never_duplicates_a_descriptor() {
let prod = code_only(production_scope(include_str!("blocking_io.rs")));
assert!(
prod.contains("fn drive_socket_blocking"),
"production slice no longer covers the seam"
);
for token in [concat!("try", "_clone"), concat!("F_DUP", "FD")] {
assert_eq!(
prod.matches(token).count(),
0,
"`{token}` is back in the blocking I/O seam: on RTEMS 6 every fd \
duplication of a socket fails ENXIO. The read and write roles come \
from one descriptor shared through an `Arc`."
);
}
}
#[epics_macros_rs::epics_test]
async fn channel_reader_loses_no_bytes_when_a_select_race_is_lost() {
let (tx, rx) = mpsc::channel::<Vec<u8>>(1);
let mut reader = ChannelReader::new(rx);
tx.send(b"ABCDEFGH".to_vec()).await.expect("chunk queued");
let mut small = [0u8; 3];
let n = reader.read(&mut small).await.expect("first read");
assert_eq!(&small[..n], b"ABC");
for _ in 0..4 {
let mut buf = [0u8; 8];
tokio::select! {
biased;
_ = std::future::ready(()) => {}
_ = reader.read(&mut buf) => unreachable!("the ready arm wins under `biased`"),
}
}
let mut rest = [0u8; 8];
let n = reader.read(&mut rest).await.expect("read after lost races");
assert_eq!(
&rest[..n],
b"DEFGH",
"a lost race must not eat the parked tail of the chunk"
);
for _ in 0..4 {
let mut buf = [0u8; 8];
tokio::select! {
biased;
_ = std::future::ready(()) => {}
_ = reader.read(&mut buf) => unreachable!("the ready arm wins under `biased`"),
}
}
tx.send(b"IJKL".to_vec())
.await
.expect("second chunk queued");
let mut after = [0u8; 8];
let n = reader
.read(&mut after)
.await
.expect("read after pending races");
assert_eq!(
&after[..n],
b"IJKL",
"a chunk must not be taken out of the channel by a poll that returned Pending"
);
drop(tx);
let mut eof = [0u8; 8];
assert_eq!(
reader.read(&mut eof).await.expect("eof read"),
0,
"all senders dropped must surface as a zero-length read"
);
}
#[epics_macros_rs::epics_test]
async fn a_zero_length_read_consumes_nothing() {
let (tx, rx) = mpsc::channel::<Vec<u8>>(1);
let mut reader = ChannelReader::new(rx);
tx.send(b"XY".to_vec()).await.expect("chunk queued");
let mut none = [0u8; 0];
assert_eq!(reader.read(&mut none).await.expect("empty read"), 0);
let mut buf = [0u8; 8];
let n = reader.read(&mut buf).await.expect("real read");
assert_eq!(&buf[..n], b"XY", "the chunk survived a zero-length read");
}
#[epics_macros_rs::epics_test]
async fn channel_writer_does_not_keep_the_frame_channel_open() {
let (tx, mut rx) = mpsc::channel::<Vec<u8>>(1);
let room = Arc::new(WriteRoom::default());
let mut writer = ChannelWriter {
tx: tx.downgrade(),
room,
};
writer.write_all(b"frame").await.expect("queued");
assert_eq!(rx.recv().await.as_deref(), Some(&b"frame"[..]));
drop(tx);
assert!(
rx.recv().await.is_none(),
"a live ChannelWriter must not keep the channel open once the only \
strong sender is gone"
);
assert!(
writer.write_all(b"after").await.is_err(),
"writing to a closed channel must be an error, not a silent drop"
);
}
fn socket_pair() -> (StdTcpStream, StdTcpStream) {
let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind");
let addr = listener.local_addr().expect("addr");
let client = StdTcpStream::connect(addr).expect("connect");
let (server, _) = listener.accept().expect("accept");
(client, server)
}
#[cfg(unix)]
#[test]
fn the_deadline_loop_ends_a_trickling_peer() {
let (client, server) = socket_pair();
let send_timeout = Duration::from_millis(200);
client
.set_write_timeout(Some(send_timeout / 4))
.expect("sndtimeo");
let big = vec![0u8; 8 * 1024 * 1024];
let started = Instant::now();
let err = write_frame_deadline(&client, &big, send_timeout)
.expect_err("a peer that never reads must trip the deadline");
assert_eq!(err.kind(), io::ErrorKind::TimedOut);
assert!(
started.elapsed() < send_timeout * 20,
"the deadline bounded the whole frame, not each syscall: {:?}",
started.elapsed()
);
drop(server);
}
#[cfg(unix)]
#[test]
fn the_deadline_holds_with_no_socket_send_timeout() {
let (client, server) = socket_pair();
let send_timeout = Duration::from_millis(200);
let big = vec![0u8; 8 * 1024 * 1024];
let (tx, rx) = std::sync::mpsc::channel();
let started = Instant::now();
thread::spawn(move || {
let outcome = write_frame_deadline(&client, &big, send_timeout).map_err(|e| e.kind());
let _ = tx.send(outcome);
});
let outcome = rx
.recv_timeout(send_timeout * 100)
.expect("the frame's deadline must end the write without a socket timeout to lean on");
assert_eq!(
outcome.expect_err("a peer that never reads must trip the deadline"),
io::ErrorKind::TimedOut
);
assert!(
started.elapsed() < send_timeout * 20,
"the deadline bounded the whole frame: {:?}",
started.elapsed()
);
drop(server);
}
#[test]
fn the_deadline_loop_delivers_a_frame_to_a_reading_peer() {
let (client, mut server) = socket_pair();
let send_timeout = Duration::from_secs(5);
client
.set_write_timeout(Some(send_timeout / 4))
.expect("sndtimeo");
let reader = thread::spawn(move || {
let mut got = vec![0u8; 5];
server.read_exact(&mut got).expect("read");
got
});
write_frame_deadline(&client, b"hello", send_timeout).expect("delivered");
assert_eq!(reader.join().expect("reader"), b"hello");
}
static TEST_POOL: std::sync::LazyLock<WorkerPool<2>> = std::sync::LazyLock::new(|| {
WorkerPool::new(
"test",
circuit_roster(ThreadPriority::Low, ThreadPriority::Low),
64,
)
});
fn lease_pair() -> (SetLease, Worker, Worker) {
let (lease, [reader, writer]) = TEST_POOL.acquire().expect("acquire a test set");
(lease, reader, writer)
}
#[cfg(unix)]
#[test]
fn the_reader_guard_returns_a_pump_parked_in_read() {
let (client, server) = socket_pair();
client
.set_read_timeout(Some(Duration::from_secs(64_000)))
.expect("rcvtimeo");
let (lease, reader_worker, _writer_worker) = lease_pair();
let (reader, guard) = spawn_reader_pump(reader_worker, Arc::new(client), "parked", 4096, 1);
let started = Instant::now();
drop(reader);
drop(guard);
assert!(
started.elapsed() < Duration::from_secs(10),
"the guard's shutdown must return a parked read, not wait out SO_RCVTIMEO"
);
drop(lease);
drop(server);
}
#[test]
fn the_writer_guard_drops_its_sender_before_joining() {
let (client, server) = socket_pair();
let (lease, _reader_worker, writer_worker) = lease_pair();
let (writer, guard) = spawn_writer_pump(
writer_worker,
Arc::new(client),
"sender-order",
Duration::from_secs(5),
1,
);
let started = Instant::now();
drop(writer);
drop(guard);
assert!(
started.elapsed() < Duration::from_secs(10),
"dropping the guard must end a pump parked on recv(), not hang"
);
drop(lease);
drop(server);
}
#[derive(Clone, Default)]
struct CapturedLines(Arc<Mutex<Vec<String>>>);
impl tracing::Subscriber for CapturedLines {
fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
true
}
fn event(&self, event: &tracing::Event<'_>) {
struct Fields<'a>(&'a mut String);
impl tracing::field::Visit for Fields<'_> {
fn record_debug(
&mut self,
field: &tracing::field::Field,
value: &dyn std::fmt::Debug,
) {
use std::fmt::Write;
let _ = write!(self.0, " {}={value:?}", field.name());
}
}
let mut line = event.metadata().target().to_string();
event.record(&mut Fields(&mut line));
self.0.lock().expect("captured lines").push(line);
}
fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
tracing::span::Id::from_u64(1)
}
fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
fn enter(&self, _: &tracing::span::Id) {}
fn exit(&self, _: &tracing::span::Id) {}
}
fn lines_while(f: impl FnOnce()) -> Vec<String> {
let captured = CapturedLines::default();
tracing::subscriber::with_default(captured.clone(), f);
let lines = captured.0.lock().expect("captured lines");
lines.clone()
}
#[test]
fn a_panicked_reader_pump_is_reported_and_not_discarded() {
let (client, server) = socket_pair();
let (lease, reader_worker, _writer_worker) = lease_pair();
let lines = lines_while(|| {
let _guard = ReaderPumpGuard {
sock: Arc::new(client),
label: "PVA connection 127.0.0.1:0".to_string(),
job: Some(reader_worker.run(|| panic!("reader blew up"))),
};
});
assert!(
lines
.iter()
.any(|l| l.starts_with("epics_base_rs::errlog")
&& l.contains("reader thread panicked")),
"a panicked reader must reach errlog, which prints whatever the log \
configuration is — including an RTEMS console. Captured: {lines:?}"
);
drop(lease);
drop(server);
}
#[test]
fn a_panicked_writer_pump_is_reported_and_not_discarded() {
let (frames, _rx) = mpsc::channel::<Vec<u8>>(1);
let (lease, _reader_worker, writer_worker) = lease_pair();
let lines = lines_while(|| {
let _guard = WriterPumpGuard {
frames: Some(frames),
label: "PVA connection 127.0.0.1:0".to_string(),
job: Some(writer_worker.run(|| panic!("writer blew up"))),
};
});
assert!(
lines
.iter()
.any(|l| l.starts_with("epics_base_rs::errlog")
&& l.contains("writer thread panicked")),
"a panicked writer dropped whatever frames were still queued; that \
must not be silent. Captured: {lines:?}"
);
drop(lease);
}
#[test]
fn a_pump_that_ends_cleanly_is_not_announced() {
let (client, server) = socket_pair();
let (lease, reader_worker, _writer_worker) = lease_pair();
let lines = lines_while(|| {
let _guard = ReaderPumpGuard {
sock: Arc::new(client),
label: "PVA connection 127.0.0.1:0".to_string(),
job: Some(reader_worker.run(|| {})),
};
});
assert!(
!lines
.iter()
.any(|l| l.contains("was lost") || l.contains("panicked")),
"an ordinary connection teardown must print nothing: {lines:?}"
);
drop(lease);
drop(server);
}
#[test]
fn every_pump_loss_goes_through_the_announcement() {
let prod = production_scope(include_str!("blocking_io.rs"));
assert_eq!(
code_only(prod)
.matches(concat!("let _ = jo", "b.join()"))
.count(),
0,
"a discarded join result is a panicked pump nobody hears about"
);
for owner in [
"impl Drop for ReaderPumpGuard",
"impl Drop for WriterPumpGuard",
] {
let at = prod
.find(owner)
.unwrap_or_else(|| panic!("`{owner}` is gone from this module"));
let body = &prod[at..(at + 900).min(prod.len())];
assert!(
body.contains(concat!("pump_thread_", "lost(")),
"`{owner}` can lose a pump thread without saying so"
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn one_descriptor_serves_both_pumps() {
let (client, mut server) = socket_pair();
let (mut reader, mut writer) = drive_socket_blocking(
&TEST_POOL,
client,
"127.0.0.1:0",
&PumpConfig {
read_timeout: Duration::from_secs(5),
send_timeout: Duration::from_secs(5),
..PumpConfig::default()
},
)
.expect("pumps started");
let peer = thread::spawn(move || {
let mut got = vec![0u8; 4];
server.read_exact(&mut got).expect("peer read");
server.write_all(b"pong").expect("peer write");
got
});
writer.write_all(b"ping").await.expect("wrote");
let mut got = [0u8; 4];
reader.read_exact(&mut got).await.expect("read back");
assert_eq!(&got, b"pong");
assert_eq!(peer.join().expect("peer"), b"ping");
}
#[epics_macros_rs::epics_test]
async fn sequential_dials_reuse_one_worker() {
static POOL: DialPool = DialPool::new("test-dial", ThreadPriority::Low);
const DIALS: usize = 8;
let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
let addr = listener.local_addr().expect("addr");
let acceptor = thread::spawn(move || {
(0..DIALS)
.map(|_| listener.accept().expect("accept").0)
.collect::<Vec<_>>()
});
for i in 0..DIALS {
let dialed = POOL.dial(addr).expect("dial submitted");
let stream = dialed
.await
.expect("the worker must reply")
.expect("connect to a live listener");
assert_eq!(
POOL.worker_count(),
1,
"dial {i} created a new thread instead of reusing the idle \
worker: sequential dials must borrow one thread, not one each"
);
drop(stream);
}
assert_eq!(
POOL.worker_count(),
1,
"{DIALS} sequential dials must have created exactly one thread"
);
drop(acceptor.join().expect("acceptor"));
}
}