fix(io): deregister a stopped actor's fd wait on the unwind path
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.
This commit is contained in:
@@ -44,11 +44,13 @@
|
|||||||
//!
|
//!
|
||||||
//! Fd hygiene
|
//! Fd hygiene
|
||||||
//! ==========
|
//! ==========
|
||||||
//! If an actor dies while waiting on an fd, the registration is leaked
|
//! An actor stopped while waiting on an fd unwinds out of `wait_fd`'s park;
|
||||||
//! (the fd stays in the epollfd, armed). EPOLLONESHOT bounds the damage:
|
//! a drop guard there (armed after a successful register, forgotten on a
|
||||||
//! at most one stale wakeup, after which the kernel disarms. The stale
|
//! normal wake) removes the `waiters` entry iff it is still that wait's
|
||||||
//! wakeup hits a dead pid in `waiters` and is dropped. Acceptable for v0.2;
|
//! `(pid, epoch)` and only then `EPOLL_CTL_DEL`s the fd — an entry already
|
||||||
//! a future pass should DEL on actor death.
|
//! 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
|
//! Buffers used with `read`/`write` should be on fds opened with
|
||||||
//! `O_NONBLOCK`. If they aren't, the syscall may block the scheduler
|
//! `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
|
// Belt-and-braces: the unwind guard in `wait_fd` is responsible for
|
||||||
// fd, the kernel-side registration was leaked (we don't walk all
|
// cleaning up a stopped waiter's registration, but a bare DEL is
|
||||||
// waiters on actor death). A bare DEL is harmless if the fd isn't
|
// harmless if the fd isn't registered (ENOENT) and removes any leak
|
||||||
// registered (ENOENT), and removes any leak.
|
// a path we haven't thought of might leave behind.
|
||||||
unsafe {
|
unsafe {
|
||||||
libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut());
|
libc::epoll_ctl(self.epollfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut());
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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();
|
let mut io = inner.io.lock().unwrap();
|
||||||
io.as_mut().expect("io thread not started").epoll_register(fd, me, epoch, readable, writable)
|
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();
|
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(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -322,3 +322,46 @@ fn wait_writable_on_empty_pipe_returns_quickly() {
|
|||||||
elapsed
|
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::<u8>));
|
||||||
|
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'));
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user