diff --git a/tokio/src/process/unix/driver.rs b/tokio/src/process/unix/driver.rs index 110b4846b..43b2efa0a 100644 --- a/tokio/src/process/unix/driver.rs +++ b/tokio/src/process/unix/driver.rs @@ -3,11 +3,8 @@ //! Process driver use crate::park::Park; -use crate::process::unix::orphan::ReapOrphanQueue; use crate::process::unix::GlobalOrphanQueue; -use crate::signal::unix::driver::Driver as SignalDriver; -use crate::signal::unix::{signal_with_handle, SignalKind}; -use crate::sync::watch; +use crate::signal::unix::driver::{Driver as SignalDriver, Handle as SignalHandle}; use std::io; use std::time::Duration; @@ -16,51 +13,20 @@ use std::time::Duration; #[derive(Debug)] pub(crate) struct Driver { park: SignalDriver, - inner: CoreDriver, GlobalOrphanQueue>, -} - -#[derive(Debug)] -struct CoreDriver { - sigchild: S, - orphan_queue: Q, -} - -trait HasChanged { - fn has_changed(&mut self) -> bool; -} - -impl HasChanged for watch::Receiver { - fn has_changed(&mut self) -> bool { - self.try_has_changed().and_then(Result::ok).is_some() - } -} - -// ===== impl CoreDriver ===== - -impl CoreDriver -where - S: HasChanged, - Q: ReapOrphanQueue, -{ - fn process(&mut self) { - if self.sigchild.has_changed() { - self.orphan_queue.reap_orphans(); - } - } + signal_handle: SignalHandle, } // ===== impl Driver ===== impl Driver { /// Creates a new signal `Driver` instance that delegates wakeups to `park`. - pub(crate) fn new(park: SignalDriver) -> io::Result { - let sigchild = signal_with_handle(SignalKind::child(), park.handle())?; - let inner = CoreDriver { - sigchild, - orphan_queue: GlobalOrphanQueue, - }; + pub(crate) fn new(park: SignalDriver) -> Self { + let signal_handle = park.handle(); - Ok(Self { park, inner }) + Self { + park, + signal_handle, + } } } @@ -76,13 +42,13 @@ impl Park for Driver { fn park(&mut self) -> Result<(), Self::Error> { self.park.park()?; - self.inner.process(); + GlobalOrphanQueue::reap_orphans(&self.signal_handle); Ok(()) } fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> { self.park.park_timeout(duration)?; - self.inner.process(); + GlobalOrphanQueue::reap_orphans(&self.signal_handle); Ok(()) } @@ -90,43 +56,3 @@ impl Park for Driver { self.park.shutdown() } } - -#[cfg(test)] -mod test { - use super::*; - use crate::process::unix::orphan::test::MockQueue; - - struct MockStream { - total_try_recv: usize, - values: Vec>, - } - - impl MockStream { - fn new(values: Vec>) -> Self { - Self { - total_try_recv: 0, - values, - } - } - } - - impl HasChanged for MockStream { - fn has_changed(&mut self) -> bool { - self.total_try_recv += 1; - self.values.remove(0).is_some() - } - } - - #[test] - fn no_reap_if_no_signal() { - let mut driver = CoreDriver { - sigchild: MockStream::new(vec![None]), - orphan_queue: MockQueue::<()>::new(), - }; - - driver.process(); - - assert_eq!(1, driver.sigchild.total_try_recv); - assert_eq!(0, driver.orphan_queue.total_reaps.get()); - } -} diff --git a/tokio/src/process/unix/mod.rs b/tokio/src/process/unix/mod.rs index 852a191b0..fab63dd3d 100644 --- a/tokio/src/process/unix/mod.rs +++ b/tokio/src/process/unix/mod.rs @@ -24,7 +24,7 @@ pub(crate) mod driver; pub(crate) mod orphan; -use orphan::{OrphanQueue, OrphanQueueImpl, ReapOrphanQueue, Wait}; +use orphan::{OrphanQueue, OrphanQueueImpl, Wait}; mod reap; use reap::Reaper; @@ -32,6 +32,7 @@ use reap::Reaper; use crate::io::PollEvented; use crate::process::kill::Kill; use crate::process::SpawnedChild; +use crate::signal::unix::driver::Handle as SignalHandle; use crate::signal::unix::{signal, Signal, SignalKind}; use mio::event::Source; @@ -73,9 +74,9 @@ impl fmt::Debug for GlobalOrphanQueue { } } -impl ReapOrphanQueue for GlobalOrphanQueue { - fn reap_orphans(&self) { - ORPHAN_QUEUE.reap_orphans() +impl GlobalOrphanQueue { + fn reap_orphans(handle: &SignalHandle) { + ORPHAN_QUEUE.reap_orphans(handle) } } diff --git a/tokio/src/process/unix/orphan.rs b/tokio/src/process/unix/orphan.rs index 8a1e12783..07f0dcfe4 100644 --- a/tokio/src/process/unix/orphan.rs +++ b/tokio/src/process/unix/orphan.rs @@ -1,6 +1,9 @@ +use crate::loom::sync::{Mutex, MutexGuard}; +use crate::signal::unix::driver::Handle as SignalHandle; +use crate::signal::unix::{signal_with_handle, SignalKind}; +use crate::sync::watch; use std::io; use std::process::ExitStatus; -use std::sync::Mutex; /// An interface for waiting on a process to exit. pub(crate) trait Wait { @@ -20,21 +23,8 @@ impl Wait for &mut T { } } -/// An interface for reaping a set of orphaned processes. -pub(crate) trait ReapOrphanQueue { - /// Attempts to reap every process in the queue, ignoring any errors and - /// enqueueing any orphans which have not yet exited. - fn reap_orphans(&self); -} - -impl ReapOrphanQueue for &T { - fn reap_orphans(&self) { - (**self).reap_orphans() - } -} - /// An interface for queueing up an orphaned process so that it can be reaped. -pub(crate) trait OrphanQueue: ReapOrphanQueue { +pub(crate) trait OrphanQueue { /// Adds an orphan to the queue. fn push_orphan(&self, orphan: T); } @@ -48,50 +38,91 @@ impl> OrphanQueue for &O { /// An implementation of `OrphanQueue`. #[derive(Debug)] pub(crate) struct OrphanQueueImpl { + sigchild: Mutex>>, queue: Mutex>, } impl OrphanQueueImpl { pub(crate) fn new() -> Self { Self { + sigchild: Mutex::new(None), queue: Mutex::new(Vec::new()), } } #[cfg(test)] fn len(&self) -> usize { - self.queue.lock().unwrap().len() + self.queue.lock().len() } -} -impl OrphanQueue for OrphanQueueImpl { - fn push_orphan(&self, orphan: T) { - self.queue.lock().unwrap().push(orphan) + pub(crate) fn push_orphan(&self, orphan: T) + where + T: Wait, + { + self.queue.lock().push(orphan) } -} -impl ReapOrphanQueue for OrphanQueueImpl { - fn reap_orphans(&self) { - let mut queue = self.queue.lock().unwrap(); - let queue = &mut *queue; + /// Attempts to reap every process in the queue, ignoring any errors and + /// enqueueing any orphans which have not yet exited. + pub(crate) fn reap_orphans(&self, handle: &SignalHandle) + where + T: Wait, + { + // If someone else is holding the lock, they will be responsible for draining + // the queue as necessary, so we can safely bail if that happens + if let Some(mut sigchild_guard) = self.sigchild.try_lock() { + match &mut *sigchild_guard { + Some(sigchild) => { + if sigchild.try_has_changed().and_then(Result::ok).is_some() { + drain_orphan_queue(self.queue.lock()); + } + } + None => { + let queue = self.queue.lock(); - for i in (0..queue.len()).rev() { - match queue[i].try_wait() { - Ok(None) => {} - Ok(Some(_)) | Err(_) => { - // The stdlib handles interruption errors (EINTR) when polling a child process. - // All other errors represent invalid inputs or pids that have already been - // reaped, so we can drop the orphan in case an error is raised. - queue.swap_remove(i); + // Be lazy and only initialize the SIGCHLD listener if there + // are any orphaned processes in the queue. + if !queue.is_empty() { + // An errors shouldn't really happen here, but if it does it + // means that the signal driver isn't running, in + // which case there isn't anything we can + // register/initialize here, so we can try again later + if let Ok(sigchild) = signal_with_handle(SignalKind::child(), &handle) { + *sigchild_guard = Some(sigchild); + drain_orphan_queue(queue); + } + } } } } } } +fn drain_orphan_queue(mut queue: MutexGuard<'_, Vec>) +where + T: Wait, +{ + for i in (0..queue.len()).rev() { + match queue[i].try_wait() { + Ok(None) => {} + Ok(Some(_)) | Err(_) => { + // The stdlib handles interruption errors (EINTR) when polling a child process. + // All other errors represent invalid inputs or pids that have already been + // reaped, so we can drop the orphan in case an error is raised. + queue.swap_remove(i); + } + } + } + + drop(queue); +} + #[cfg(all(test, not(loom)))] pub(crate) mod test { use super::*; + use crate::io::driver::Driver as IoDriver; + use crate::signal::unix::driver::{Driver as SignalDriver, Handle as SignalHandle}; + use crate::sync::watch; use std::cell::{Cell, RefCell}; use std::io; use std::os::unix::process::ExitStatusExt; @@ -100,14 +131,12 @@ pub(crate) mod test { pub(crate) struct MockQueue { pub(crate) all_enqueued: RefCell>, - pub(crate) total_reaps: Cell, } impl MockQueue { pub(crate) fn new() -> Self { Self { all_enqueued: RefCell::new(Vec::new()), - total_reaps: Cell::new(0), } } } @@ -118,12 +147,6 @@ pub(crate) mod test { } } - impl ReapOrphanQueue for MockQueue { - fn reap_orphans(&self) { - self.total_reaps.set(self.total_reaps.get() + 1); - } - } - struct MockWait { total_waits: Rc>, num_wait_until_status: usize, @@ -191,27 +214,107 @@ pub(crate) mod test { assert_eq!(orphanage.len(), 4); - orphanage.reap_orphans(); + drain_orphan_queue(orphanage.queue.lock()); assert_eq!(orphanage.len(), 2); assert_eq!(first_waits.get(), 1); assert_eq!(second_waits.get(), 1); assert_eq!(third_waits.get(), 1); assert_eq!(fourth_waits.get(), 1); - orphanage.reap_orphans(); + drain_orphan_queue(orphanage.queue.lock()); assert_eq!(orphanage.len(), 1); assert_eq!(first_waits.get(), 1); assert_eq!(second_waits.get(), 2); assert_eq!(third_waits.get(), 2); assert_eq!(fourth_waits.get(), 1); - orphanage.reap_orphans(); + drain_orphan_queue(orphanage.queue.lock()); assert_eq!(orphanage.len(), 0); assert_eq!(first_waits.get(), 1); assert_eq!(second_waits.get(), 2); assert_eq!(third_waits.get(), 3); assert_eq!(fourth_waits.get(), 1); - orphanage.reap_orphans(); // Safe to reap when empty + // Safe to reap when empty + drain_orphan_queue(orphanage.queue.lock()); + } + + #[test] + fn no_reap_if_no_signal_received() { + let (tx, rx) = watch::channel(()); + + let handle = SignalHandle::default(); + + let orphanage = OrphanQueueImpl::new(); + *orphanage.sigchild.lock() = Some(rx); + + let orphan = MockWait::new(2); + let waits = orphan.total_waits.clone(); + orphanage.push_orphan(orphan); + + orphanage.reap_orphans(&handle); + assert_eq!(waits.get(), 0); + + orphanage.reap_orphans(&handle); + assert_eq!(waits.get(), 0); + + tx.send(()).unwrap(); + orphanage.reap_orphans(&handle); + assert_eq!(waits.get(), 1); + } + + #[test] + fn no_reap_if_signal_lock_held() { + let handle = SignalHandle::default(); + + let orphanage = OrphanQueueImpl::new(); + let signal_guard = orphanage.sigchild.lock(); + + let orphan = MockWait::new(2); + let waits = orphan.total_waits.clone(); + orphanage.push_orphan(orphan); + + orphanage.reap_orphans(&handle); + assert_eq!(waits.get(), 0); + + drop(signal_guard); + } + + #[test] + fn does_not_register_signal_if_queue_empty() { + let signal_driver = IoDriver::new().and_then(SignalDriver::new).unwrap(); + let handle = signal_driver.handle(); + + let orphanage = OrphanQueueImpl::new(); + assert!(orphanage.sigchild.lock().is_none()); // Sanity + + // No register when queue empty + orphanage.reap_orphans(&handle); + assert!(orphanage.sigchild.lock().is_none()); + + let orphan = MockWait::new(2); + let waits = orphan.total_waits.clone(); + orphanage.push_orphan(orphan); + + orphanage.reap_orphans(&handle); + assert!(orphanage.sigchild.lock().is_some()); + assert_eq!(waits.get(), 1); // Eager reap when registering listener + } + + #[test] + fn does_nothing_if_signal_could_not_be_registered() { + let handle = SignalHandle::default(); + + let orphanage = OrphanQueueImpl::new(); + assert!(orphanage.sigchild.lock().is_none()); + + let orphan = MockWait::new(2); + let waits = orphan.total_waits.clone(); + orphanage.push_orphan(orphan); + + // Signal handler has "gone away", nothing to register or reap + orphanage.reap_orphans(&handle); + assert!(orphanage.sigchild.lock().is_none()); + assert_eq!(waits.get(), 0); } } diff --git a/tokio/src/process/unix/reap.rs b/tokio/src/process/unix/reap.rs index 5dc95e5b8..f7f4d3cc7 100644 --- a/tokio/src/process/unix/reap.rs +++ b/tokio/src/process/unix/reap.rs @@ -224,7 +224,6 @@ mod test { assert!(grim.poll_unpin(&mut context).is_pending()); assert_eq!(1, grim.signal.total_polls); assert_eq!(1, grim.total_waits); - assert_eq!(0, grim.orphan_queue.total_reaps.get()); assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); // Not yet exited, couldn't register interest the first time @@ -232,7 +231,6 @@ mod test { assert!(grim.poll_unpin(&mut context).is_pending()); assert_eq!(3, grim.signal.total_polls); assert_eq!(3, grim.total_waits); - assert_eq!(0, grim.orphan_queue.total_reaps.get()); assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); // Exited @@ -245,7 +243,6 @@ mod test { } assert_eq!(4, grim.signal.total_polls); assert_eq!(4, grim.total_waits); - assert_eq!(0, grim.orphan_queue.total_reaps.get()); assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); } @@ -260,7 +257,6 @@ mod test { grim.kill().unwrap(); assert_eq!(1, grim.total_kills); - assert_eq!(0, grim.orphan_queue.total_reaps.get()); assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); } @@ -276,7 +272,6 @@ mod test { drop(grim); - assert_eq!(0, queue.total_reaps.get()); assert!(queue.all_enqueued.borrow().is_empty()); } @@ -294,7 +289,6 @@ mod test { let grim = Reaper::new(&mut mock, &queue, MockStream::new(vec![])); drop(grim); - assert_eq!(0, queue.total_reaps.get()); assert_eq!(1, queue.all_enqueued.borrow().len()); } diff --git a/tokio/src/runtime/driver.rs b/tokio/src/runtime/driver.rs index a0e8e2362..7e459779b 100644 --- a/tokio/src/runtime/driver.rs +++ b/tokio/src/runtime/driver.rs @@ -23,7 +23,7 @@ cfg_io_driver! { let io_handle = io_driver.handle(); let (signal_driver, signal_handle) = create_signal_driver(io_driver)?; - let process_driver = create_process_driver(signal_driver)?; + let process_driver = create_process_driver(signal_driver); (Either::A(process_driver), Some(io_handle), signal_handle) } else { @@ -80,7 +80,7 @@ cfg_not_signal_internal! { cfg_process_driver! { type ProcessDriver = crate::process::unix::driver::Driver; - fn create_process_driver(signal_driver: SignalDriver) -> io::Result { + fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver { crate::process::unix::driver::Driver::new(signal_driver) } } @@ -89,8 +89,8 @@ cfg_not_process_driver! { cfg_io_driver! { type ProcessDriver = SignalDriver; - fn create_process_driver(signal_driver: SignalDriver) -> io::Result { - Ok(signal_driver) + fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver { + signal_driver } } } diff --git a/tokio/src/signal/unix.rs b/tokio/src/signal/unix.rs index 9f8d25691..f96b2f4c2 100644 --- a/tokio/src/signal/unix.rs +++ b/tokio/src/signal/unix.rs @@ -227,7 +227,7 @@ fn action(globals: Pin<&'static Globals>, signal: libc::c_int) { /// /// This will register the signal handler if it hasn't already been registered, /// returning any error along the way if that fails. -fn signal_enable(signal: SignalKind, handle: Handle) -> io::Result<()> { +fn signal_enable(signal: SignalKind, handle: &Handle) -> io::Result<()> { let signal = signal.0; if signal < 0 || signal_hook_registry::FORBIDDEN.contains(&signal) { return Err(Error::new( @@ -357,7 +357,7 @@ pub struct Signal { /// * If the signal is one of /// [`signal_hook::FORBIDDEN`](fn@signal_hook_registry::register#panics) pub fn signal(kind: SignalKind) -> io::Result { - let rx = signal_with_handle(kind, Handle::current())?; + let rx = signal_with_handle(kind, &Handle::current())?; Ok(Signal { inner: RxFuture::new(rx), @@ -366,7 +366,7 @@ pub fn signal(kind: SignalKind) -> io::Result { pub(crate) fn signal_with_handle( kind: SignalKind, - handle: Handle, + handle: &Handle, ) -> io::Result> { // Turn the signal delivery on once we are ready for it signal_enable(kind, handle)?; @@ -462,14 +462,14 @@ mod tests { #[test] fn signal_enable_error_on_invalid_input() { - signal_enable(SignalKind::from_raw(-1), Handle::default()).unwrap_err(); + signal_enable(SignalKind::from_raw(-1), &Handle::default()).unwrap_err(); } #[test] fn signal_enable_error_on_forbidden_input() { signal_enable( SignalKind::from_raw(signal_hook_registry::FORBIDDEN[0]), - Handle::default(), + &Handle::default(), ) .unwrap_err(); }