smarm
aeacaf6118
Improve reliability under high load: - tests/stress.rs: New comprehensive stress test suite (448 lines) - Fine-tune I/O & runtime scheduling edge cases - Pin versions & fix MSRV compatibility
522 lines
18 KiB
Rust
522 lines
18 KiB
Rust
//! Off-scheduler IO: blocking-work offload and epoll-based fd readiness.
|
|
//!
|
|
//! `block_on_io(closure)` runs `closure` on a dedicated worker OS thread,
|
|
//! parks the calling actor in the meantime, and returns the closure's
|
|
//! value when it completes. Lets actors call into blocking C libraries,
|
|
//! synchronous file IO, or anything else that doesn't fit the readiness
|
|
//! model.
|
|
//!
|
|
//! `wait_readable(fd)` / `wait_writable(fd)` register interest in an fd
|
|
//! with epoll and park the calling actor. When the fd becomes ready, the
|
|
//! epoll thread unparks the actor. The actual `read(2)`/`write(2)` syscall
|
|
//! runs back on the scheduler thread, *inside* the actor — buffer never
|
|
//! leaves the actor, no copying through an intermediary thread. Built on
|
|
//! these are the conveniences `read(fd, &mut buf)` and `write(fd, &buf)`.
|
|
//!
|
|
//! Architecture
|
|
//! ============
|
|
//! Per `run()`, two OS threads:
|
|
//! - **epoll thread**: owns the epollfd. Loops in `epoll_wait`. On a
|
|
//! ready fd, pushes `Completion::FdReady { pid, fd, events }` to the
|
|
//! shared completion queue and writes the scheduler-wake pipe. On the
|
|
//! shutdown pipe (also registered in epollfd), exits.
|
|
//! - **pool thread**: blocks on the request mpsc. Runs the closure
|
|
//! inside `catch_unwind`, pushes `Completion::Blocking { pid, result }`,
|
|
//! writes the scheduler-wake pipe.
|
|
//!
|
|
//! Both threads share a single `completions: Arc<Mutex<VecDeque<Completion>>>`
|
|
//! and the same scheduler-wake pipe.
|
|
//!
|
|
//! `epoll_ctl` (register/unregister fd interest) is called by the
|
|
//! scheduler thread *directly* on the epollfd. That's well-defined per
|
|
//! `epoll_ctl(2)`: a thread may be calling `epoll_wait` on the epollfd
|
|
//! while another thread calls `epoll_ctl`. Avoids needing a second mpsc
|
|
//! and a second wake mechanism.
|
|
//!
|
|
//! Epoll mode
|
|
//! ==========
|
|
//! Level-triggered with EPOLLONESHOT. After a wakeup the kernel
|
|
//! auto-disarms the fd, so we never get two wakeups for one
|
|
//! `wait_readable` call. The scheduler explicitly `EPOLL_CTL_DEL`s the fd
|
|
//! on completion to free the slot for re-registration. Net effect: each
|
|
//! `wait_readable(fd)` is one ADD, one wakeup, one DEL — symmetric and
|
|
//! stateless between calls.
|
|
//!
|
|
//! Fd hygiene
|
|
//! ==========
|
|
//! If an actor dies while waiting on an fd, the registration is leaked
|
|
//! (the fd stays in the epollfd, armed). EPOLLONESHOT bounds the damage:
|
|
//! at most one stale wakeup, after which the kernel disarms. The stale
|
|
//! wakeup hits a dead pid in `waiters` and is dropped. Acceptable for v0.2;
|
|
//! a future pass should DEL on actor death.
|
|
//!
|
|
//! Buffers used with `read`/`write` should be on fds opened with
|
|
//! `O_NONBLOCK`. If they aren't, the syscall may block the scheduler
|
|
//! thread despite the readiness notification (the fd reporting readable
|
|
//! doesn't guarantee the syscall completes without blocking — e.g. a
|
|
//! signal could be delivered). Documented; not enforced.
|
|
//!
|
|
//! Panic handling
|
|
//! ==============
|
|
//! The pool worker runs the closure inside `catch_unwind` and ships either
|
|
//! the return value or the panic payload back to the scheduler.
|
|
//! `block_on_io` resumes the panic on the calling actor's stack, so the
|
|
//! actor's supervisor sees a real `Signal::Panic` as if the work had run
|
|
//! inline. Fd-wait primitives don't run user code on the IO thread, so
|
|
//! they have no equivalent panic-propagation path.
|
|
|
|
use crate::pid::Pid;
|
|
use std::any::Any;
|
|
use std::collections::{HashMap, VecDeque};
|
|
use std::io;
|
|
use std::os::fd::RawFd;
|
|
use std::panic;
|
|
use std::sync::mpsc;
|
|
use std::sync::{Arc, Mutex};
|
|
use std::thread::JoinHandle as OsJoinHandle;
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Wire types
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// What the pool stores while computing a result. `Ok` is the closure's
|
|
/// return value (boxed as `Any`); `Err` is the panic payload.
|
|
pub type IoResult = Result<Box<dyn Any + Send>, Box<dyn Any + Send>>;
|
|
|
|
struct Request {
|
|
pid: Pid,
|
|
/// The work to perform. Returns the wire-form result directly.
|
|
work: Box<dyn FnOnce() -> IoResult + Send>,
|
|
}
|
|
|
|
/// Completion message from either IO thread back to the scheduler.
|
|
pub enum Completion {
|
|
/// A `block_on_io` closure has finished (Ok = return value, Err = panic
|
|
/// payload).
|
|
Blocking { pid: Pid, result: IoResult },
|
|
/// An fd registered via `wait_readable`/`wait_writable` is ready. The
|
|
/// scheduler looks up the parked pid in `waiters`, unparks it, and
|
|
/// removes the entry. `pid` isn't in this variant because the epoll
|
|
/// thread doesn't have access to the `waiters` map; the scheduler
|
|
/// thread owns that.
|
|
FdReady { fd: RawFd, events: u32 },
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// IoThread — created per `run()`, owned by `SchedulerState`.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
pub struct IoThread {
|
|
// ----- Channels & queues -----
|
|
|
|
/// Submission queue into the blocking-work pool.
|
|
tx: mpsc::Sender<Request>,
|
|
/// Shared completion queue, fed by both the pool and the epoll thread.
|
|
completions: Arc<Mutex<VecDeque<Completion>>>,
|
|
/// Pipe the scheduler polls in its idle path. Both IO threads write to
|
|
/// `wake_write` after pushing a completion.
|
|
wake_read: RawFd,
|
|
wake_write: RawFd,
|
|
|
|
// ----- Epoll machinery -----
|
|
|
|
/// The epollfd, owned by `IoThread`. Callable cross-thread via
|
|
/// `epoll_ctl` per the man page.
|
|
epollfd: RawFd,
|
|
/// Pipe used to signal the epoll thread to exit. Registered inside the
|
|
/// epollfd so a single `epoll_wait` covers both fd readiness and
|
|
/// shutdown.
|
|
shutdown_read: RawFd,
|
|
shutdown_write: RawFd,
|
|
/// One parked actor per registered fd. Populated by `wait_readable` /
|
|
/// `wait_writable` and drained by the scheduler when a `FdReady`
|
|
/// completion is processed.
|
|
pub waiters: HashMap<RawFd, Pid>,
|
|
|
|
// ----- Threads -----
|
|
|
|
pool_thread: Option<OsJoinHandle<()>>,
|
|
epoll_thread: Option<OsJoinHandle<()>>,
|
|
|
|
/// Number of `block_on_io` requests in-flight. Used by the scheduler's
|
|
/// idle path to decide whether to wait on the pipe or exit. Fd waits
|
|
/// are not counted here; they're counted by `waiters.len()`.
|
|
pub outstanding: u32,
|
|
}
|
|
|
|
impl IoThread {
|
|
pub fn start() -> io::Result<Self> {
|
|
// Scheduler-facing wake pipe.
|
|
let (wake_read, wake_write) = make_pipe()?;
|
|
// Pool submission channel + shared completion queue.
|
|
let (tx, rx) = mpsc::channel::<Request>();
|
|
let completions: Arc<Mutex<VecDeque<Completion>>> =
|
|
Arc::new(Mutex::new(VecDeque::new()));
|
|
|
|
// Epoll machinery.
|
|
let epollfd = unsafe { libc::epoll_create1(libc::EPOLL_CLOEXEC) };
|
|
if epollfd < 0 {
|
|
// Best-effort fd cleanup before bailing.
|
|
unsafe {
|
|
libc::close(wake_read);
|
|
libc::close(wake_write);
|
|
}
|
|
return Err(io::Error::last_os_error());
|
|
}
|
|
|
|
let (shutdown_read, shutdown_write) = match make_pipe() {
|
|
Ok(p) => p,
|
|
Err(e) => {
|
|
unsafe {
|
|
libc::close(epollfd);
|
|
libc::close(wake_read);
|
|
libc::close(wake_write);
|
|
}
|
|
return Err(e);
|
|
}
|
|
};
|
|
|
|
// Register the shutdown pipe in epollfd. We use a sentinel `data`
|
|
// value to recognise shutdown events. RawFd values are non-negative,
|
|
// so u64::MAX is unambiguously not a real fd-data encoding.
|
|
let mut shutdown_ev = libc::epoll_event {
|
|
events: libc::EPOLLIN as u32,
|
|
u64: SHUTDOWN_EPOLL_TOKEN,
|
|
};
|
|
if unsafe {
|
|
libc::epoll_ctl(
|
|
epollfd,
|
|
libc::EPOLL_CTL_ADD,
|
|
shutdown_read,
|
|
&mut shutdown_ev as *mut _,
|
|
)
|
|
} < 0
|
|
{
|
|
let e = io::Error::last_os_error();
|
|
unsafe {
|
|
libc::close(epollfd);
|
|
libc::close(shutdown_read);
|
|
libc::close(shutdown_write);
|
|
libc::close(wake_read);
|
|
libc::close(wake_write);
|
|
}
|
|
return Err(e);
|
|
}
|
|
|
|
// Spawn pool thread.
|
|
let pool_comps = completions.clone();
|
|
let pool_thread = std::thread::Builder::new()
|
|
.name("smarm-io-pool".into())
|
|
.spawn(move || pool_loop(rx, pool_comps, wake_write))?;
|
|
|
|
// Spawn epoll thread.
|
|
let epoll_comps = completions.clone();
|
|
let epoll_thread = std::thread::Builder::new()
|
|
.name("smarm-io-epoll".into())
|
|
.spawn(move || epoll_loop(epollfd, epoll_comps, wake_write))?;
|
|
|
|
Ok(Self {
|
|
tx,
|
|
completions,
|
|
wake_read,
|
|
wake_write,
|
|
epollfd,
|
|
shutdown_read,
|
|
shutdown_write,
|
|
waiters: HashMap::new(),
|
|
pool_thread: Some(pool_thread),
|
|
epoll_thread: Some(epoll_thread),
|
|
outstanding: 0,
|
|
})
|
|
}
|
|
|
|
/// Hand a request to the pool. Increments `outstanding`.
|
|
pub fn submit(&mut self, pid: Pid, work: Box<dyn FnOnce() -> IoResult + Send>) {
|
|
self.outstanding += 1;
|
|
// Send can only fail if the pool has hung up, which only happens
|
|
// on shutdown. submit during shutdown is a bug.
|
|
self.tx
|
|
.send(Request { pid, work })
|
|
.expect("io pool hung up unexpectedly");
|
|
}
|
|
|
|
/// Drain every available completion. Caller (the scheduler) routes the
|
|
/// results and updates `outstanding` / `waiters` accordingly.
|
|
pub fn drain_completions(&mut self) -> Vec<Completion> {
|
|
let mut q = self.completions.lock().unwrap();
|
|
let mut out = Vec::with_capacity(q.len());
|
|
while let Some(c) = q.pop_front() {
|
|
out.push(c);
|
|
}
|
|
out
|
|
}
|
|
|
|
pub fn wake_fd(&self) -> RawFd {
|
|
self.wake_read
|
|
}
|
|
|
|
/// Register interest in `fd` becoming readable/writable; record `pid`
|
|
/// as the parked waiter. The epoll thread will push a `FdReady`
|
|
/// completion when the kernel signals.
|
|
///
|
|
/// EPOLLONESHOT: one wakeup per registration. The scheduler must
|
|
/// `epoll_del` on completion to free the slot for re-registration.
|
|
pub fn epoll_register(
|
|
&mut self,
|
|
fd: RawFd,
|
|
pid: Pid,
|
|
readable: bool,
|
|
writable: bool,
|
|
) -> io::Result<()> {
|
|
// Two actors waiting on the same fd would be a misuse: the kernel
|
|
// delivers exactly one EPOLLONESHOT wakeup, so the second waiter
|
|
// would hang. Reject up front.
|
|
if self.waiters.contains_key(&fd) {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::AlreadyExists,
|
|
"fd already has a parked waiter",
|
|
));
|
|
}
|
|
|
|
// Defensive cleanup: if a previous actor died while waiting on this
|
|
// fd, the kernel-side registration was leaked (we don't walk all
|
|
// waiters on actor death). A bare DEL is harmless if the fd isn't
|
|
// registered (ENOENT), and removes any leak.
|
|
unsafe {
|
|
libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut());
|
|
}
|
|
|
|
let mut events: u32 = libc::EPOLLONESHOT as u32;
|
|
if readable {
|
|
events |= libc::EPOLLIN as u32;
|
|
}
|
|
if writable {
|
|
events |= libc::EPOLLOUT as u32;
|
|
}
|
|
let mut ev = libc::epoll_event {
|
|
events,
|
|
u64: fd as u64,
|
|
};
|
|
let r = unsafe {
|
|
libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_ADD, fd, &mut ev as *mut _)
|
|
};
|
|
if r < 0 {
|
|
return Err(io::Error::last_os_error());
|
|
}
|
|
self.waiters.insert(fd, pid);
|
|
Ok(())
|
|
}
|
|
|
|
/// Remove `fd` from the epollfd. Called by the scheduler after a
|
|
/// `FdReady` completion, so the next `wait_readable(fd)` can ADD again.
|
|
///
|
|
/// Does NOT touch `waiters` — that's the scheduler's bookkeeping; this
|
|
/// is purely the kernel-side cleanup.
|
|
pub fn epoll_deregister(&mut self, fd: RawFd) {
|
|
// EPOLL_CTL_DEL of an already-removed fd returns ENOENT; ignore.
|
|
unsafe {
|
|
libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut());
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Drop for IoThread {
|
|
fn drop(&mut self) {
|
|
// 1. Signal the epoll thread to exit by writing the shutdown pipe.
|
|
unsafe {
|
|
let buf: [u8; 1] = [0];
|
|
// Single byte; we don't care about EINTR retry here — worst
|
|
// case the epoll thread blocks until process exit, which is
|
|
// fine because we then close fds out from under it.
|
|
libc::write(self.shutdown_write, buf.as_ptr() as *const _, 1);
|
|
}
|
|
|
|
// 2. Hang up the pool's request channel so the pool thread exits.
|
|
let (dead_tx, _) = mpsc::channel::<Request>();
|
|
let real_tx = std::mem::replace(&mut self.tx, dead_tx);
|
|
drop(real_tx);
|
|
|
|
// 3. Join both threads.
|
|
if let Some(h) = self.epoll_thread.take() {
|
|
let _ = h.join();
|
|
}
|
|
if let Some(h) = self.pool_thread.take() {
|
|
let _ = h.join();
|
|
}
|
|
|
|
// 4. Close fds.
|
|
unsafe {
|
|
libc::close(self.epollfd);
|
|
libc::close(self.shutdown_read);
|
|
libc::close(self.shutdown_write);
|
|
libc::close(self.wake_read);
|
|
libc::close(self.wake_write);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Sentinel `epoll_event.u64` distinguishing the shutdown pipe from
|
|
/// registered actor fds. RawFd values fit in i32, so the high bits are
|
|
/// available for a marker; we use u64::MAX which can't be a valid fd.
|
|
const SHUTDOWN_EPOLL_TOKEN: u64 = u64::MAX;
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Pool loop
|
|
// ---------------------------------------------------------------------------
|
|
|
|
fn pool_loop(
|
|
rx: mpsc::Receiver<Request>,
|
|
completions: Arc<Mutex<VecDeque<Completion>>>,
|
|
wake_write: RawFd,
|
|
) {
|
|
while let Ok(Request { pid, work }) = rx.recv() {
|
|
let result: IoResult = match panic::catch_unwind(panic::AssertUnwindSafe(work)) {
|
|
Ok(r) => r,
|
|
Err(payload) => Err(payload),
|
|
};
|
|
completions
|
|
.lock()
|
|
.unwrap()
|
|
.push_back(Completion::Blocking { pid, result });
|
|
wake_scheduler(wake_write);
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Epoll loop
|
|
// ---------------------------------------------------------------------------
|
|
|
|
fn epoll_loop(
|
|
epollfd: RawFd,
|
|
completions: Arc<Mutex<VecDeque<Completion>>>,
|
|
wake_write: RawFd,
|
|
) {
|
|
// Buffer for epoll_wait. 64 is plenty for our scale; if a real load
|
|
// appears that needs more, this is a one-line change.
|
|
const MAX_EVENTS: usize = 64;
|
|
let mut events: [libc::epoll_event; MAX_EVENTS] = unsafe { std::mem::zeroed() };
|
|
|
|
loop {
|
|
let n = unsafe {
|
|
libc::epoll_wait(
|
|
epollfd,
|
|
events.as_mut_ptr(),
|
|
MAX_EVENTS as libc::c_int,
|
|
-1,
|
|
)
|
|
};
|
|
|
|
if n < 0 {
|
|
let e = unsafe { *libc::__errno_location() };
|
|
if e == libc::EINTR {
|
|
continue;
|
|
}
|
|
// Anything else here is a programming error (EBADF on epollfd
|
|
// after we've closed it from Drop — the close races with us).
|
|
// Treat as shutdown.
|
|
return;
|
|
}
|
|
|
|
let mut shutdown_requested = false;
|
|
let mut pushed_any = false;
|
|
{
|
|
let mut q = completions.lock().unwrap();
|
|
for ev in events.iter().take(n as usize) {
|
|
if ev.u64 == SHUTDOWN_EPOLL_TOKEN {
|
|
shutdown_requested = true;
|
|
continue;
|
|
}
|
|
let fd = ev.u64 as RawFd;
|
|
let evs = ev.events;
|
|
q.push_back(Completion::FdReady {
|
|
fd,
|
|
events: evs,
|
|
});
|
|
pushed_any = true;
|
|
}
|
|
}
|
|
|
|
if pushed_any {
|
|
wake_scheduler(wake_write);
|
|
}
|
|
if shutdown_requested {
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Write one byte to the scheduler's wake pipe. Retries on EINTR; ignores
|
|
/// EAGAIN (pipe full means there's already an outstanding wake we haven't
|
|
/// consumed yet, which is sufficient).
|
|
fn wake_scheduler(wake_write: RawFd) {
|
|
let buf: [u8; 1] = [0];
|
|
unsafe {
|
|
loop {
|
|
let n = libc::write(wake_write, buf.as_ptr() as *const _, 1);
|
|
if n < 0 {
|
|
let e = *libc::__errno_location();
|
|
if e == libc::EINTR {
|
|
continue;
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Pipe helpers (unchanged from v0.2)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
fn make_pipe() -> io::Result<(RawFd, RawFd)> {
|
|
let mut fds: [libc::c_int; 2] = [0; 2];
|
|
let r = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
|
|
if r != 0 {
|
|
return Err(io::Error::last_os_error());
|
|
}
|
|
Ok((fds[0], fds[1]))
|
|
}
|
|
|
|
/// Drain pending bytes from the wake pipe. The scheduler calls this after
|
|
/// a `poll` wakeup so the next idle call sees an empty pipe.
|
|
pub fn drain_wake_pipe(fd: RawFd) {
|
|
let mut buf = [0u8; 64];
|
|
loop {
|
|
let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) };
|
|
if n <= 0 {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Block on `fd` for up to `timeout`, returning when either there's data
|
|
/// to read or the timeout elapses. `None` for `timeout` means wait forever.
|
|
pub fn poll_wake(fd: RawFd, timeout: Option<std::time::Duration>) {
|
|
let timeout_ms: libc::c_int = match timeout {
|
|
None => -1,
|
|
Some(d) => {
|
|
let ms = d.as_millis();
|
|
if ms > i32::MAX as u128 {
|
|
i32::MAX
|
|
} else {
|
|
ms as i32
|
|
}
|
|
}
|
|
};
|
|
let mut pfd = libc::pollfd {
|
|
fd,
|
|
events: libc::POLLIN,
|
|
revents: 0,
|
|
};
|
|
loop {
|
|
let r = unsafe { libc::poll(&mut pfd as *mut _, 1, timeout_ms) };
|
|
if r < 0 {
|
|
let e = unsafe { *libc::__errno_location() };
|
|
if e == libc::EINTR {
|
|
continue;
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
}
|