From f6969e538b649377c01b23787066eb930e9e48f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 15:07:35 +0000 Subject: [PATCH] fix(io): deregister a stopped actor's fd wait on the unwind path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stopped actor unwinding out of wait_fd's park leaked its waiters entry and the kernel-side EPOLLONESHOT registration; the stale entry then failed every future wait_*() on that fd with AlreadyExists (the defensive bare DEL in epoll_register sits behind the contains_key check, so it never ran). Fix where the invariant breaks: a drop guard in wait_fd, armed after a successful register and forgotten on the normal wake (where FdReady already removed + DEL'd). On unwind it cleans up iff the entry is still this wait's (pid, epoch) — an entry consumed by a racing FdReady means the fd may carry another actor's fresh registration, which must be left alone. Closes the v0.2 fd-hygiene TODO. --- src/io.rs | 20 +++++++++++--------- src/scheduler.rs | 32 ++++++++++++++++++++++++++++++++ tests/io_epoll.rs | 43 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 86 insertions(+), 9 deletions(-) diff --git a/src/io.rs b/src/io.rs index f7e50ef..c9fd977 100644 --- a/src/io.rs +++ b/src/io.rs @@ -44,11 +44,13 @@ //! //! 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. +//! An actor stopped while waiting on an fd unwinds out of `wait_fd`'s park; +//! a drop guard there (armed after a successful register, forgotten on a +//! normal wake) removes the `waiters` entry iff it is still that wait's +//! `(pid, epoch)` and only then `EPOLL_CTL_DEL`s the fd — an entry already +//! consumed by a racing `FdReady` means the fd may carry someone else's +//! fresh registration, which must be left alone. `epoll_register` keeps a +//! defensive bare DEL before ADD as belt-and-braces. //! //! Buffers used with `read`/`write` should be on fds opened with //! `O_NONBLOCK`. If they aren't, the syscall may block the scheduler @@ -282,10 +284,10 @@ impl IoThread { )); } - // 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. + // Belt-and-braces: the unwind guard in `wait_fd` is responsible for + // cleaning up a stopped waiter's registration, but a bare DEL is + // harmless if the fd isn't registered (ENOENT) and removes any leak + // a path we haven't thought of might leave behind. unsafe { libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut()); } diff --git a/src/scheduler.rs b/src/scheduler.rs index 28f9c7f..1add4d3 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -396,7 +396,39 @@ fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::R let mut io = inner.io.lock().unwrap(); io.as_mut().expect("io thread not started").epoll_register(fd, me, epoch, readable, writable) })?; + + // If a terminal stop unwinds us out of the park below, the registration + // must not outlive us: a stale `waiters` entry fails every future + // `wait_*` on this fd with AlreadyExists, and the kernel-side ADD leaks + // until the fd happens to be reused. Clean up iff the entry is still + // ours — a `FdReady` racing the stop may have consumed it already (it + // removes + DELs under the io lock), after which the fd may even carry + // ANOTHER actor's fresh registration; in that case touch nothing. + struct Dereg { + fd: std::os::fd::RawFd, + me: Pid, + epoch: u32, + } + impl Drop for Dereg { + fn drop(&mut self) { + with_runtime(|inner| { + let mut io = inner.io.lock().unwrap(); + if let Some(io) = io.as_mut() { + if io.waiters.get(&self.fd) == Some(&(self.me, self.epoch)) { + io.waiters.remove(&self.fd); + io.epoll_deregister(self.fd); + } + } + }); + } + } + let guard = Dereg { fd, me, epoch }; park_current(); + // Normal wake: the FdReady path removed the entry and DEL'd the fd + // before the unpark, so the guard's check would be a guaranteed no-op — + // skip the io lock on the hot path. (Dereg owns nothing; forget leaks + // no resource.) + std::mem::forget(guard); Ok(()) } diff --git a/tests/io_epoll.rs b/tests/io_epoll.rs index a28516a..89cacf4 100644 --- a/tests/io_epoll.rs +++ b/tests/io_epoll.rs @@ -322,3 +322,46 @@ fn wait_writable_on_empty_pipe_returns_quickly() { elapsed ); } + +// --------------------------------------------------------------------------- +// Fd hygiene on actor death (v0.8) +// --------------------------------------------------------------------------- + +// An actor stopped while parked on an fd must not leak its `waiters` entry: +// before the unwind-path guard in wait_fd, the stale entry made every +// future wait_*() on that fd fail with AlreadyExists, forever. +#[test] +fn stopped_waiter_does_not_poison_the_fd() { + let outcome = Arc::new(StdMutex::new(None::)); + let outcome2 = outcome.clone(); + run(move || { + let p = Pipe::new(); + let (rfd, wfd) = (p.read, p.write); + + // First waiter parks on the (empty) pipe and is stopped in place. + let h = smarm::spawn(move || { + wait_readable(rfd).unwrap(); + unreachable!("the pipe is never written while this actor lives"); + }); + yield_now(); // let it reach the park + smarm::request_stop(h.pid()); + h.join().unwrap(); // Ok(()): stopped, not panicked + + // Second waiter on the SAME fd must be able to register... + let seen = Arc::new(AtomicU32::new(0)); + let seen2 = seen.clone(); + let h2 = smarm::spawn(move || { + wait_readable(rfd).unwrap(); + let mut buf = [0u8; 1]; + assert_eq!(raw_read(rfd, &mut buf), 1); + seen2.store(buf[0] as u32, Ordering::SeqCst); + }); + yield_now(); // ...and park (a failed register would panic the unwrap) + + // ...and actually be woken by readiness. + assert_eq!(raw_write(wfd, b"x"), 1); + h2.join().unwrap(); + *outcome2.lock().unwrap() = Some(seen.load(Ordering::SeqCst) as u8); + }); + assert_eq!(*outcome.lock().unwrap(), Some(b'x')); +}