diff --git a/apps/nesinit/src/reap.rs b/apps/nesinit/src/reap.rs index 99856420..c85fb8e3 100644 --- a/apps/nesinit/src/reap.rs +++ b/apps/nesinit/src/reap.rs @@ -7,6 +7,7 @@ use std::collections::HashMap; use std::io; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use nesprotocol::lifecycle::Exit; @@ -83,7 +84,45 @@ pub fn refuse_oom_kill() -> io::Result<()> { /// component exists to report. So the reaper is the only waiter, and this is /// how it hands an exit to whoever asked for one. #[derive(Clone, Default)] -pub struct Waiters(Arc>>>); +pub struct Waiters(Arc>>); + +/// The registry's half of one watched child. +struct Watch { + exit: oneshot::Sender, + running: Arc, +} + +/// A child something is waiting for: its pid, its exit, and whether the pid +/// still means what it meant. +/// +/// The flag is the point. A pid is only a name for a process until that +/// process is reaped, after which the kernel may hand the same number to +/// something else — so a caller that kept a pid and signals it later can hit a +/// process it has never heard of. Anything that signals asks here first. +pub struct Watched { + pub pid: i32, + running: Arc, + exit: Option>, +} + +impl Watched { + /// The exit, which only one caller may hold — whoever takes it is the one + /// that reports it. What is left behind is the pid and whether it still + /// means this child, which is what a signaller needs. + pub fn take_exit(&mut self) -> Option> { + self.exit.take() + } + + /// Whether the pid is still this child's. + /// + /// False from the moment its exit is delivered. There is a window between + /// the reap and the delivery in which this still says true; closing it + /// entirely needs a handle the kernel keeps for us rather than a number, + /// and the number is what the rest of this component has. + pub fn running(&self) -> bool { + self.running.load(Ordering::SeqCst) + } +} impl Waiters { pub fn new() -> Self { @@ -96,29 +135,44 @@ impl Waiters { /// that exits immediately is reaped by a thread that has to take the same /// lock to deliver the exit, so it waits until the slot exists rather than /// finding none and dropping it. - pub fn watch(&self, spawn: F) -> io::Result> + pub fn watch(&self, spawn: F) -> io::Result where F: FnOnce() -> io::Result, { let mut waiting = self.lock(); let pid = spawn()?; let (sender, receiver) = oneshot::channel(); - waiting.insert(pid, sender); - Ok(receiver) + let running = Arc::new(AtomicBool::new(true)); + waiting.insert( + pid, + Watch { + exit: sender, + running: running.clone(), + }, + ); + Ok(Watched { + pid, + running, + exit: Some(receiver), + }) } /// Hand an exit to whoever is waiting for that pid. `false` if nobody is, /// which is the common case: most of what an init reaps is an orphan /// nothing asked about. pub fn deliver(&self, pid: i32, exit: Exit) -> bool { - let Some(sender) = self.lock().remove(&pid) else { + let Some(watch) = self.lock().remove(&pid) else { return false; }; + // Before the exit goes anywhere: the pid has been reaped, so the + // number is free for the kernel to give to something else and nothing + // may signal it from here on. + watch.running.store(false, Ordering::SeqCst); // An error here means the caller stopped waiting, which is its right. - sender.send(exit).is_ok() + watch.exit.send(exit).is_ok() } - fn lock(&self) -> std::sync::MutexGuard<'_, HashMap>> { + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { self.0 .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index 9fb814f7..0ef5433f 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -14,7 +14,7 @@ use nesprotocol::lifecycle::{Exec, Exit, Mount}; use std::os::unix::process::CommandExt; -use crate::reap::Waiters; +use crate::reap::{Waiters, Watched}; /// Why something could not be done, in the words the operating system used. /// @@ -55,14 +55,17 @@ pub trait Workload { /// The workload as a local process. pub struct Process { waiters: Waiters, - pid: Option, + running: Option, } impl Process { /// Started through the reaper's registry, because the reaper is the only /// thing in this component that may call `wait`. pub fn new(waiters: Waiters) -> Self { - Self { waiters, pid: None } + Self { + waiters, + running: None, + } } /// Send a signal to the workload and nothing else. @@ -70,11 +73,18 @@ impl Process { /// Nothing here ever signals the process group or every process: the order /// a shutdown promises is only true if the workload can be stopped alone. pub fn signal(&self, signal: libc::c_int) { - let Some(pid) = self.pid else { return }; + let Some(watched) = &self.running else { return }; + // Nothing is signalled once the exit has been delivered. The pid was + // freed by the reap that produced it, and the kernel is entitled to + // give that number to something else — a stop or a kill aimed at it + // then lands on a process nobody meant. + if !watched.running() { + return; + } // SAFETY: two integers, and it cannot touch this process's memory. A // pid that has already gone fails with ESRCH, which is exactly the // idempotence the callers of this rely on. - unsafe { libc::kill(pid, signal) }; + unsafe { libc::kill(watched.pid, signal) }; } /// Wait up to `grace` for the workload to leave, without a runtime. @@ -83,7 +93,13 @@ impl Process { /// one pid rather than for any child, so a slow service cannot be mistaken /// for the workload still running. pub fn await_exit(&self, grace: std::time::Duration) -> bool { - let Some(pid) = self.pid else { return true }; + let Some(watched) = &self.running else { + return true; + }; + if !watched.running() { + return true; // already reaped, and its exit already reported + } + let pid = watched.pid; let deadline = std::time::Instant::now() + grace; loop { let mut status: libc::c_int = 0; @@ -151,21 +167,18 @@ impl Workload for Process { }); } - let mut pid = None; - let exited = self + let mut watched = self .waiters - .watch(|| { - let child = command.spawn()?; - let started = child.id() as i32; - pid = Some(started); - Ok(started) - }) + .watch(|| Ok(command.spawn()?.id() as i32)) .map_err(|error| Failure::new(error.to_string()))?; - self.pid = pid; + + // The caller gets the exit and reports it; this handle keeps the pid + // and whether that pid is still this child's. + let exit = watched.take_exit().expect("a new watch has its exit"); + self.running = Some(watched); Ok(Box::pin(async move { - exited - .await + exit.await .map_err(|_| io::Error::other("the workload's exit was not delivered")) })) } diff --git a/apps/nesinit/tests/reaping.rs b/apps/nesinit/tests/reaping.rs index 3ee12fc9..66f0a30b 100644 --- a/apps/nesinit/tests/reaping.rs +++ b/apps/nesinit/tests/reaping.rs @@ -115,9 +115,10 @@ fn an_exit_reaches_whoever_asked_for_it() { let _alone = alone(); let waiters = Waiters::new(); - let mut exited = waiters + let mut watched = waiters .watch(|| Ok(fork_child(|| unsafe { libc::_exit(9) }))) .expect("the child never started"); + let mut exited = watched.take_exit().unwrap(); let deadline = Instant::now() + Duration::from_secs(5); while Instant::now() < deadline { @@ -154,7 +155,7 @@ fn an_exit_that_happens_before_the_caller_is_registered_is_not_lost() { } }); - let mut exited = waiters + let mut watched = waiters .watch(|| { let pid = fork_child(|| unsafe { libc::_exit(5) }); // Long enough for the child to exit and the reaper to reach it. @@ -162,6 +163,7 @@ fn an_exit_that_happens_before_the_caller_is_registered_is_not_lost() { Ok(pid) }) .expect("the child never started"); + let mut exited = watched.take_exit().unwrap(); let deadline = Instant::now() + Duration::from_secs(5); let exit = loop { @@ -179,3 +181,37 @@ fn an_exit_that_happens_before_the_caller_is_registered_is_not_lost() { stop.store(true, Ordering::Relaxed); reaper.join().unwrap(); } + +#[test] +fn a_pid_stops_being_the_workload_the_moment_it_is_reaped() { + let _alone = alone(); + let waiters = Waiters::new(); + + let mut watched = waiters + .watch(|| Ok(fork_child(|| unsafe { libc::_exit(0) }))) + .expect("the child never started"); + let mut exited = watched.take_exit().unwrap(); + assert!( + watched.running(), + "a child that has not been reaped is still itself" + ); + + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + for (pid, exit) in reap_exited() { + waiters.deliver(pid, exit); + } + if exited.try_recv().is_ok() { + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + + // The pid was freed by the reap that produced that exit, and the kernel is + // entitled to hand the number to something else. Signalling it after this + // point is signalling a stranger. + assert!( + !watched.running(), + "a reaped pid is still being treated as the workload's", + ); +}