process: fix panic from spurious pidfd wakeup (#7494)

tokio/src/io/poll_evented.rs

@@ -125,7 +125,7 @@ impl<E: Source> PollEvented<E> {
     }
 
     /// Returns a reference to the registration.
-    #[cfg(feature = "net")]
+    #[cfg(any(feature = "net", all(feature = "process", target_os = "linux")))]
     pub(crate) fn registration(&self) -> &Registration {
         &self.registration
     }
@@ -138,14 +138,6 @@ impl<E: Source> PollEvented<E> {
         Ok(inner)
     }
 
-    #[cfg(all(feature = "process", target_os = "linux"))]
-    pub(crate) fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        self.registration
-            .poll_read_ready(cx)
-            .map_err(io::Error::from)
-            .map_ok(|_| ())
-    }
-
     /// Re-register under new runtime with `interest`.
     #[cfg(all(feature = "process", target_os = "linux"))]
     pub(crate) fn reregister(&mut self, interest: Interest) -> io::Result<()> {

tokio/src/process/unix/pidfd_reaper.rs

@@ -19,7 +19,7 @@ use std::{
     pin::Pin,
     process::ExitStatus,
     sync::atomic::{AtomicBool, Ordering::Relaxed},
-    task::{ready, Context, Poll},
+    task::{Context, Poll},
 };
 
 #[derive(Debug)]
@@ -117,17 +117,21 @@ where
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = Pin::into_inner(self);
 
-        match ready!(this.pidfd.poll_read_ready(cx)) {
-            Err(err) if is_rt_shutdown_err(&err) => {
-                this.pidfd.reregister(Interest::READABLE)?;
-                ready!(this.pidfd.poll_read_ready(cx))?
+        match this.pidfd.registration().poll_read_ready(cx) {
+            Poll::Ready(Ok(evt)) => {
+                if let Some(exit_code) = this.inner.try_wait()? {
+                    return Poll::Ready(Ok(exit_code));
+                }
+                this.pidfd.registration().clear_readiness(evt);
             }
-            res => res?,
-        }
-        Poll::Ready(Ok(this
-            .inner
-            .try_wait()?
-            .expect("pidfd is ready to read, the process should have exited")))
+            Poll::Ready(Err(err)) if is_rt_shutdown_err(&err) => {}
+            Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+            Poll::Pending => return Poll::Pending,
+        };
+
+        this.pidfd.reregister(Interest::READABLE)?;
+        cx.waker().wake_by_ref();
+        Poll::Pending
     }
 }
 

tokio/tests/process_issue_7144.rs

@@ -0,0 +1,28 @@
+#![cfg(feature = "process")]
+#![warn(rust_2018_idioms)]
+#![cfg(target_os = "linux")]
+#![cfg(not(miri))]
+
+use tokio::process::Command;
+use tokio::time::{sleep, Duration};
+
+#[tokio::test]
+async fn issue_7144() {
+    let mut threads = vec![];
+    for _ in 0..20 {
+        threads.push(tokio::spawn(test_one()));
+    }
+    for thread in threads {
+        thread.await.unwrap();
+    }
+}
+
+async fn test_one() {
+    let mut t = Command::new("strace")
+        .args("-o /dev/null -D sleep 5".split(' '))
+        .spawn()
+        .unwrap();
+    sleep(Duration::from_millis(100)).await;
+    unsafe { libc::kill(t.id().unwrap() as _, libc::SIGINT) };
+    t.wait().await.unwrap();
+}