process: avoid redundant effort to reap orphan processes (#3743)

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<watch::Receiver<()>, GlobalOrphanQueue>,
-}
-
-#[derive(Debug)]
-struct CoreDriver<S, Q> {
-    sigchild: S,
-    orphan_queue: Q,
-}
-
-trait HasChanged {
-    fn has_changed(&mut self) -> bool;
-}
-
-impl<T> HasChanged for watch::Receiver<T> {
-    fn has_changed(&mut self) -> bool {
-        self.try_has_changed().and_then(Result::ok).is_some()
-    }
-}
-
-// ===== impl CoreDriver =====
-
-impl<S, Q> CoreDriver<S, Q>
-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<Self> {
-        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<Option<()>>,
-    }
-
-    impl MockStream {
-        fn new(values: Vec<Option<()>>) -> 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());
-    }
-}

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)
     }
 }
 

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<T: Wait> 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<T: ReapOrphanQueue> 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<T>: ReapOrphanQueue {
+pub(crate) trait OrphanQueue<T> {
     /// Adds an orphan to the queue.
     fn push_orphan(&self, orphan: T);
 }
@@ -48,50 +38,91 @@ impl<T, O: OrphanQueue<T>> OrphanQueue<T> for &O {
 /// An implementation of `OrphanQueue`.
 #[derive(Debug)]
 pub(crate) struct OrphanQueueImpl<T> {
+    sigchild: Mutex<Option<watch::Receiver<()>>>,
     queue: Mutex<Vec<T>>,
 }
 
 impl<T> OrphanQueueImpl<T> {
     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<T: Wait> OrphanQueue<T> for OrphanQueueImpl<T> {
-    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<T: Wait> ReapOrphanQueue for OrphanQueueImpl<T> {
-    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<T>(mut queue: MutexGuard<'_, Vec<T>>)
+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<W> {
         pub(crate) all_enqueued: RefCell<Vec<W>>,
-        pub(crate) total_reaps: Cell<usize>,
     }
 
     impl<W> MockQueue<W> {
         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<W> ReapOrphanQueue for MockQueue<W> {
-        fn reap_orphans(&self) {
-            self.total_reaps.set(self.total_reaps.get() + 1);
-        }
-    }
-
     struct MockWait {
         total_waits: Rc<Cell<usize>>,
         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);
     }
 }

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());
         }
 

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<ProcessDriver> {
+    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<ProcessDriver> {
-            Ok(signal_driver)
+        fn create_process_driver(signal_driver: SignalDriver) -> ProcessDriver {
+            signal_driver
         }
     }
 }

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<Signal> {
-    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<Signal> {
 
 pub(crate) fn signal_with_handle(
     kind: SignalKind,
-    handle: Handle,
+    handle: &Handle,
 ) -> io::Result<watch::Receiver<()>> {
     // 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();
     }