feat: Use the waiter backend on Windows

In async-process, we have a backend that assumes that child processes
are object that can be `.await`ed on, rather than just being dependent
on signals. At the moment it is only used with Linux and pidfd. Now, it
is used with Windows and the waitable process backend.

At the moment, the backend for `Waitable` in `async-io` is just backed
by a blocking threadpool. However it may also be possible to have it use
IOCP too with little extra overhead. See smol-rs/polling#141 for more
information.

As a side effect, this removes our dependency on `windows-sys`.

Signed-off-by: John Nunley <dev@notgull.net>

.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,38 @@ 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"
+async-io = "2.1.0"
+
+[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,
+                #[allow(unreachable_patterns)]
                 _ => 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
+        }
     }
 }