fixup! Change ChildPidWatcher to use pidfd and epoll

sporksmith · sporksmith · commit 5c1277184d4b · 2021-07-12T13:49:59.000-05:00
diff --git a/src/main/utility/childpid_watcher.rs b/src/main/utility/childpid_watcher.rs
@@ -10,6 +10,9 @@ use std::sync::Arc;
 use std::sync::Mutex;
 use std::thread;
 
+/// Utility for monitoring a set of child pid's, calling registered callbacks
+/// when one exits or is killed. Starts a background thread, which is shut down
+/// when the object is dropped.
 pub struct ChildPidWatcher {
     inner: Arc<Mutex<Inner>>,
     epoll: std::os::unix::io::RawFd,
@@ -69,7 +72,8 @@ fn is_zombie(pid: Pid) -> bool {
 }
 
 impl ChildPidWatcher {
-    /// Get a reference to the global singleton watcher.
+    /// Create a ChildPidWatcher. Spawns a background thread, which is joined
+    /// when the object is dropped.
     pub fn new() -> Self {
         let epoll = epoll_create1(EpollCreateFlags::empty()).unwrap();
         let command_notifier =
@@ -94,7 +98,7 @@ impl ChildPidWatcher {
             epoll,
         };
         let thread_handle = {
-            let inner = watcher.inner.clone();
+            let inner = Arc::clone(&watcher.inner);
             let epoll = watcher.epoll;
             thread::Builder::new()
                 .name("child-pid-watcher".into())
@@ -133,14 +137,22 @@ impl ChildPidWatcher {
                 Err(nix::Error::Sys(nix::errno::Errno::EAGAIN)) => true,
                 Err(e) => panic!("Unexpected error {:?}", e),
             });
-            // Run  commands
+            // Run commands
             std::mem::swap(&mut commands, &mut inner.commands);
             for cmd in commands.drain(..) {
                 match cmd {
                     Command::RunCallbacks(pid) => {
                         inner.run_callbacks_for_fd(epoll, pid);
                     }
-                    Command::Finish => done = true,
+                    Command::Finish => {
+                        done = true;
+                        // There could be more commands queued and/or more epoll
+                        // events ready, but it doesn't matter. We don't
+                        // guarantee to callers whether callbacks have run or
+                        // not after having sent `Finish`; only that no more
+                        // callbacks will run after the thread is joined.
+                        break;
+                    }
                 }
             }
         }
@@ -153,7 +165,7 @@ impl ChildPidWatcher {
     /// The returned handle is guaranteed to be non-zero.
     ///
     /// Panics if `pid` doesn't exist.
-    pub fn watch(&self, pid: Pid, callback: Box<dyn Send + FnOnce(Pid)>) -> WatchHandle {
+    pub fn watch(&self, pid: Pid, callback: impl Send + FnOnce(Pid) + 'static) -> WatchHandle {
         let mut inner = self.inner.lock().unwrap();
         if !inner.pidfds.contains_key(&pid) {
             let pidfd: RawFd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid.as_raw(), 0) }
@@ -174,7 +186,7 @@ impl ChildPidWatcher {
         let handle = inner.next_handle;
         inner.next_handle += 1;
         let callbacks = inner.callbacks.entry(pid).or_default();
-        callbacks.insert(handle, callback);
+        callbacks.insert(handle, Box::new(callback));
         handle
     }
 
@@ -441,10 +453,11 @@ mod export {
         data: *mut libc::c_void,
     ) -> WatchHandle {
         let data = SyncSendPointer(data);
-        unsafe { watcher.as_ref() }.unwrap().watch(
-            Pid::from_raw(pid),
-            Box::new(move |pid| callback(pid.into(), data.ptr())),
-        )
+        unsafe { watcher.as_ref() }
+            .unwrap()
+            .watch(Pid::from_raw(pid), move |pid| {
+                callback(pid.into(), data.ptr())
+            })
     }
 
     /// Unregisters a callback. After returning, the corresponding callback is
