Use the waiter backend on Windows

.github/workflows/ci.yml

@@ -46,6 +46,7 @@ jobs:
       - run: cargo test
         env:
           RUSTFLAGS: ${{ env.RUSTFLAGS }} --cfg async_process_force_signal_backend
+        if: matrix.os != 'windows-latest'
 
   test-android:
     runs-on: ubuntu-latest

Cargo.toml

@@ -16,38 +16,37 @@ exclude = ["/.*"]
 
 [dependencies]
 async-lock = "3.0.0"
+async-io = "2.1.0"
 cfg-if = "1.0"
 event-listener = "5.1.0"
 futures-lite = "2.0.0"
 tracing = { version = "0.1.40", default-features = false }
 
 [target.'cfg(unix)'.dependencies]
-async-io = "2.1.0"
 async-signal = "0.2.3"
 rustix = { version = "0.38", default-features = false, features = ["std", "fs"] }
 
-[target.'cfg(target_os = "linux")'.dependencies]
+[target.'cfg(any(windows, target_os = "linux"))'.dependencies]
 async-channel = "2.0.0"
 async-task = "4.7.0"
 
 [target.'cfg(all(unix, not(target_os = "linux")))'.dependencies]
 rustix = { version = "0.38", default-features = false, features = ["std", "fs", "process"] }
 
 [target.'cfg(windows)'.dependencies]
-async-channel = "2.0.0"
 blocking = "1.0.0"
 
-[target.'cfg(windows)'.dependencies.windows-sys]
+[lints.rust]
+unexpected_cfgs = { level = "warn", check-cfg = ['cfg(async_process_force_signal_backend)'] }
+
+[dev-dependencies]
+async-executor = "1.5.1"
+
+[target.'cfg(windows)'.dev-dependencies.windows-sys]
 version = "0.59"
 default-features = false
 features = [
     "Win32_Foundation",
     "Win32_System_Threading",
 ]
 
-[lints.rust]
-unexpected_cfgs = { level = "warn", check-cfg = ['cfg(async_process_force_signal_backend)'] }
-
-[dev-dependencies]
-async-executor = "1.5.1"
-async-io = "2.1.0"

src/reaper/mod.rs

@@ -12,22 +12,29 @@
 #![allow(irrefutable_let_patterns)]
 
 /// Enable the waiting reaper.
-#[cfg(target_os = "linux")]
+#[cfg(any(windows, target_os = "linux"))]
 macro_rules! cfg_wait {
     ($($tt:tt)*) => {$($tt)*};
 }
 
 /// Enable the waiting reaper.
-#[cfg(not(target_os = "linux"))]
+#[cfg(not(any(windows, target_os = "linux")))]
 macro_rules! cfg_wait {
     ($($tt:tt)*) => {};
 }
 
 /// Enable signals.
+#[cfg(not(windows))]
 macro_rules! cfg_signal {
     ($($tt:tt)*) => {$($tt)*};
 }
 
+/// Enable signals.
+#[cfg(windows)]
+macro_rules! cfg_signal {
+    ($($tt:tt)*) => {};
+}
+
 cfg_wait! {
     mod wait;
 }
@@ -41,31 +48,34 @@ use std::sync::Mutex;
 
 /// The underlying system reaper.
 pub(crate) enum Reaper {
-    #[cfg(target_os = "linux")]
+    #[cfg(any(windows, target_os = "linux"))]
     /// The reaper based on the wait backend.
     Wait(wait::Reaper),
 
     /// The reaper based on the signal backend.
+    #[cfg(not(windows))]
     Signal(signal::Reaper),
 }
 
 /// The wrapper around a child.
 pub(crate) enum ChildGuard {
-    #[cfg(target_os = "linux")]
+    #[cfg(any(windows, target_os = "linux"))]
     /// The child guard based on the wait backend.
     Wait(wait::ChildGuard),
 
     /// The child guard based on the signal backend.
+    #[cfg(not(windows))]
     Signal(signal::ChildGuard),
 }
 
 /// A lock on the reaper.
 pub(crate) enum Lock {
-    #[cfg(target_os = "linux")]
+    #[cfg(any(windows, target_os = "linux"))]
     /// The wait-based reaper needs no lock.
     Wait,
 
     /// The lock for the signal-based reaper.
+    #[cfg(not(windows))]
     Signal(signal::Lock),
 }
 

src/reaper/signal.rs

@@ -1,8 +1,9 @@
 //! A version of the reaper that waits for a signal to check for process progress.
 
 use async_lock::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard};
+use async_signal::{Signal, Signals};
 use event_listener::Event;
-use futures_lite::future;
+use futures_lite::{future, prelude::*};
 
 use std::io;
 use std::mem;
@@ -141,103 +142,27 @@ impl ChildGuard {
     }
 }
 
-cfg_if::cfg_if! {
-    if #[cfg(windows)] {
-        use async_channel::{Sender, Receiver, bounded};
-        use std::ffi::c_void;
-        use std::os::windows::io::AsRawHandle;
-        use std::ptr;
-
-        use windows_sys::Win32::{
-            Foundation::{BOOLEAN, HANDLE},
-            System::Threading::{
-                RegisterWaitForSingleObject, INFINITE, WT_EXECUTEINWAITTHREAD, WT_EXECUTEONLYONCE,
-            },
-        };
-
-        /// Waits for the next SIGCHLD signal.
-        struct Pipe {
-            /// The sender channel for the SIGCHLD signal.
-            sender: Sender<()>,
-
-            /// The receiver channel for the SIGCHLD signal.
-            receiver: Receiver<()>,
-        }
-
-        impl Pipe {
-            /// Creates a new pipe.
-            fn new() -> io::Result<Pipe> {
-                let (sender, receiver) = bounded(1);
-                Ok(Pipe {
-                    sender,
-                    receiver
-                })
-            }
-
-            /// Waits for the next SIGCHLD signal.
-            async fn wait(&self) {
-                self.receiver.recv().await.ok();
-            }
-
-            /// Register a process object into this pipe.
-            fn register(&self, child: &std::process::Child) -> io::Result<()> {
-                // Called when a child exits.
-                #[allow(clippy::infallible_destructuring_match)]
-                unsafe extern "system" fn callback(_: *mut c_void, _: BOOLEAN) {
-                    let reaper = match &crate::Reaper::get().sys {
-                        super::Reaper::Signal(reaper) => reaper,
-                    };
-
-                    reaper.pipe.sender.try_send(()).ok();
-                }
-
-                // Register this child process to invoke `callback` on exit.
-                let mut wait_object = ptr::null_mut();
-                let ret = unsafe {
-                    RegisterWaitForSingleObject(
-                        &mut wait_object,
-                        child.as_raw_handle() as HANDLE,
-                        Some(callback),
-                        std::ptr::null_mut(),
-                        INFINITE,
-                        WT_EXECUTEINWAITTHREAD | WT_EXECUTEONLYONCE,
-                    )
-                };
-
-                if ret == 0 {
-                    Err(io::Error::last_os_error())
-                } else {
-                    Ok(())
-                }
-            }
-        }
-    } else if #[cfg(unix)] {
-        use async_signal::{Signal, Signals};
-        use futures_lite::prelude::*;
-
-        /// Waits for the next SIGCHLD signal.
-        struct Pipe {
-            /// The iterator over SIGCHLD signals.
-            signals: Signals,
-        }
+/// Waits for the next SIGCHLD signal.
+struct Pipe {
+    /// The iterator over SIGCHLD signals.
+    signals: Signals,
+}
 
-        impl Pipe {
-            /// Creates a new pipe.
-            fn new() -> io::Result<Pipe> {
-                Ok(Pipe {
-                    signals: Signals::new(Some(Signal::Child))?,
-                })
-            }
+impl Pipe {
+    /// Creates a new pipe.
+    fn new() -> io::Result<Pipe> {
+        Ok(Pipe {
+            signals: Signals::new(Some(Signal::Child))?,
+        })
+    }
 
-            /// Waits for the next SIGCHLD signal.
-            async fn wait(&self) {
-                (&self.signals).next().await;
-            }
+    /// Waits for the next SIGCHLD signal.
+    async fn wait(&self) {
+        (&self.signals).next().await;
+    }
 
-            /// Register a process object into this pipe.
-            fn register(&self, _child: &std::process::Child) -> io::Result<()> {
-                Ok(())
-            }
-        }
+    /// Register a process object into this pipe.
+    fn register(&self, _child: &std::process::Child) -> io::Result<()> {
+        Ok(())
     }
 }

src/reaper/wait.rs

@@ -2,7 +2,8 @@
 //!
 //! This uses:
 //!
-//! - pidfd on Linux/Android
+//! - pidfd on Linux
+//! - Waitable objects on Windows
 
 use async_channel::{Receiver, Sender};
 use async_task::Runnable;
@@ -69,6 +70,7 @@ impl Reaper {
             // Get the inner child value.
             let inner = match &mut child.inner {
                 super::ChildGuard::Wait(inner) => inner,
+                #[cfg(not(windows))]
                 _ => unreachable!(),
             };
 
@@ -181,5 +183,43 @@ cfg_if::cfg_if! {
             // Tell if it was okay or not.
             result.is_ok()
         }
+    } else if #[cfg(windows)] {
+        use async_io::os::windows::Waitable;
+
+        /// Waitable version of `std::process::Child`.
+        struct WaitableChild {
+            inner: Waitable<std::process::Child>,
+        }
+
+        impl WaitableChild {
+            fn new(child: std::process::Child) -> io::Result<Self> {
+                Ok(Self {
+                    inner: Waitable::new(child)?
+                })
+            }
+
+            fn get_mut(&mut self) -> &mut std::process::Child {
+                // SAFETY: We never move the child out.
+                unsafe {
+                    self.inner.get_mut()
+                }
+            }
+
+            fn poll_wait(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<std::process::ExitStatus>> {
+                loop {
+                    if let Some(status) = self.get_mut().try_wait()? {
+                        return Poll::Ready(Ok(status));
+                    }
+
+                    // Wait for us to become readable.
+                    futures_lite::ready!(self.inner.poll_ready(cx))?;
+                }
+            }
+        }
+
+        /// Tell if we are able to use this backend.
+        pub(crate) fn available() -> bool {
+            true
+        }
     }
 }