1 files changed, 13 insertions, 46 deletions

diff --git a/src/process/unix/reap.rs b/src/process/unix/reap.rs
index 8963805..de483c4 100644
--- a/src/process/unix/reap.rs
+++ b/src/process/unix/reap.rs
@@ -1,6 +1,6 @@
 use crate::process::imp::orphan::{OrphanQueue, Wait};
 use crate::process::kill::Kill;
-use crate::signal::unix::Signal;
+use crate::signal::unix::InternalStream;
 
 use std::future::Future;
 use std::io;
@@ -23,17 +23,6 @@ where
     signal: S,
 }
 
-// Work around removal of `futures_core` dependency
-pub(crate) trait Stream: Unpin {
-    fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<()>>;
-}
-
-impl Stream for Signal {
-    fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<()>> {
-        Signal::poll_recv(self, cx)
-    }
-}
-
 impl<W, Q, S> Deref for Reaper<W, Q, S>
 where
     W: Wait + Unpin,
@@ -63,7 +52,7 @@ where
         self.inner.as_ref().expect("inner has gone away")
     }
 
-    fn inner_mut(&mut self) -> &mut W {
+    pub(crate) fn inner_mut(&mut self) -> &mut W {
         self.inner.as_mut().expect("inner has gone away")
     }
 }
@@ -72,7 +61,7 @@ impl<W, Q, S> Future for Reaper<W, Q, S>
 where
     W: Wait + Unpin,
     Q: OrphanQueue<W> + Unpin,
-    S: Stream,
+    S: InternalStream,
 {
     type Output = io::Result<ExitStatus>;
 
@@ -80,10 +69,8 @@ where
         loop {
             // If the child hasn't exited yet, then it's our responsibility to
             // ensure the current task gets notified when it might be able to
-            // make progress.
-            //
-            // As described in `spawn` above, we just indicate that we can
-            // next make progress once a SIGCHLD is received.
+            // make progress. We can use the delivery of a SIGCHLD signal as a
+            // sign that we can potentially make progress.
             //
             // However, we will register for a notification on the next signal
             // BEFORE we poll the child. Otherwise it is possible that the child
@@ -99,7 +86,6 @@ where
             // should not cause significant issues with parent futures.
             let registered_interest = self.signal.poll_recv(cx).is_pending();
 
-            self.orphan_queue.reap_orphans();
             if let Some(status) = self.inner_mut().try_wait()? {
                 return Poll::Ready(Ok(status));
             }
@@ -147,8 +133,9 @@ where
 mod test {
     use super::*;
 
+    use crate::process::unix::orphan::test::MockQueue;
+    use crate::sync::mpsc::error::TryRecvError;
     use futures::future::FutureExt;
-    use std::cell::{Cell, RefCell};
     use std::os::unix::process::ExitStatusExt;
     use std::process::ExitStatus;
     use std::task::Context;
@@ -211,7 +198,7 @@ mod test {
         }
     }
 
-    impl Stream for MockStream {
+    impl InternalStream for MockStream {
         fn poll_recv(&mut self, _cx: &mut Context<'_>) -> Poll<Option<()>> {
             self.total_polls += 1;
             match self.values.remove(0) {
@@ -219,29 +206,9 @@ mod test {
                 None => Poll::Pending,
             }
         }
-    }
 
-    struct MockQueue<W> {
-        all_enqueued: RefCell<Vec<W>>,
-        total_reaps: Cell<usize>,
-    }
-
-    impl<W> MockQueue<W> {
-        fn new() -> Self {
-            Self {
-                all_enqueued: RefCell::new(Vec::new()),
-                total_reaps: Cell::new(0),
-            }
-        }
-    }
-
-    impl<W: Wait> OrphanQueue<W> for MockQueue<W> {
-        fn push_orphan(&self, orphan: W) {
-            self.all_enqueued.borrow_mut().push(orphan);
-        }
-
-        fn reap_orphans(&self) {
-            self.total_reaps.set(self.total_reaps.get() + 1);
+        fn try_recv(&mut self) -> Result<(), TryRecvError> {
+            unimplemented!();
         }
     }
 
@@ -262,7 +229,7 @@ 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!(1, grim.orphan_queue.total_reaps.get());
+        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
@@ -270,7 +237,7 @@ 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!(3, grim.orphan_queue.total_reaps.get());
+        assert_eq!(0, grim.orphan_queue.total_reaps.get());
         assert!(grim.orphan_queue.all_enqueued.borrow().is_empty());
 
         // Exited
@@ -283,7 +250,7 @@ mod test {
         }
         assert_eq!(4, grim.signal.total_polls);
         assert_eq!(4, grim.total_waits);
-        assert_eq!(4, grim.orphan_queue.total_reaps.get());
+        assert_eq!(0, grim.orphan_queue.total_reaps.get());
         assert!(grim.orphan_queue.all_enqueued.borrow().is_empty());
     }