port windows impl and add missing drop guard

dignifiedquire · dignifiedquire · commit 7971002ae428 · 2020-03-09T13:34:56.000+01:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -34,7 +34,15 @@ default = [
   "pin-project-lite",
 ]
 docs = ["attributes", "unstable", "default"]
-unstable = ["std", "broadcaster"]
+unstable = [
+  "std",
+  "broadcaster",
+  "crossbeam-queue",
+  "async-signals",
+  "libc",
+  "lazy_static",
+  "async-macros",
+]
 attributes = ["async-attributes"]
 std = [
   "alloc",
@@ -46,11 +54,7 @@ std = [
   "once_cell",
   "pin-utils",
   "slab",
-  "crossbeam-queue",
-  "async-signals",
-  "libc",
-  "lazy_static",
-  "async-macros",
+
 ]
 alloc = [
   "futures-core/alloc",
@@ -78,11 +82,18 @@ pin-project-lite = { version = "0.1.2", optional = true }
 pin-utils = { version = "0.1.0-alpha.4", optional = true }
 slab = { version = "0.4.2", optional = true }
 crossbeam-queue = { version = "0.2.1", optional = true }
-async-signals = { version = "0.1.0", optional = true }
 lazy_static = { version = "1.4.0", optional = true }
 async-macros = { version = "2.0.0", optional = true }
+
+[target.'cfg(not(target_os="windows"))'.dependencies]
+async-signals = { version = "0.1.0", optional = true }
 libc = { version = "0.2.67", optional = true }
 
+[target.'cfg(target_os="windows")'.dependencies]
+winapi = { version = "0.3.8", features = ["threadpoollegacyapiset"] }
+mio-named-pipes = { version = "0.1.6" }
+futures-channel = "0.3.4"
+
 [dev-dependencies]
 femme = "1.3.0"
 rand = "0.7.3"
diff --git a/src/lib.rs b/src/lib.rs
@@ -270,7 +270,6 @@ cfg_default! {
     pub mod fs;
     pub mod path;
     pub mod net;
-    pub mod process;
 }
 
 cfg_unstable! {
@@ -280,6 +279,7 @@ cfg_unstable! {
     mod option;
     mod string;
     mod collections;
+    pub mod process;
 }
 
 cfg_unstable_default! {
diff --git a/src/process/kill.rs b/src/process/kill.rs
@@ -7,8 +7,8 @@ use std::task::{Context, Poll};
 /// A drop guard which ensures the child process is killed on drop to maintain
 /// the contract of dropping a Future leads to "cancellation".
 #[derive(Debug)]
-pub(crate) struct ChildDropGuard<T: Kill> {
-    inner: T,
+pub(super) struct ChildDropGuard<T: Kill> {
+    pub(super) inner: T,
     kill_on_drop: bool,
 }
 
@@ -61,7 +61,7 @@ impl<T: Future + Kill + Unpin> Future for ChildDropGuard<T> {
 }
 
 /// An interface for killing a running process.
-pub(crate) trait Kill {
+pub(super) trait Kill {
     /// Forcefully kill the process.
     fn kill(&mut self) -> io::Result<()>;
 }
diff --git a/src/process/mod.rs b/src/process/mod.rs
@@ -141,22 +141,17 @@ mod imp;
 
 mod kill;
 
-use kill::Kill;
+use kill::{ChildDropGuard, Kill};
 
 /// Representation of a running or exited child process.
 ///
 /// This structure is used to represent and manage child processes. A child
 /// process is created via the [`Command`] struct, which configures the
 /// spawning process and can itself be constructed using a builder-style
 /// interface.
-///
-/// There is no implementation of [`Drop`] for child processes,
-/// so if you do not ensure the `Child` has exited then it will continue to
-/// run, even after the `Child` handle to the child process has gone out of
-/// scope.
 #[derive(Debug)]
 pub struct Child {
-    child: imp::Child,
+    child: ChildDropGuard<imp::Child>,
 
     /// The handle for writing to the child's standard input (stdin), if it has
     /// been captured.
@@ -172,7 +167,7 @@ pub struct Child {
 impl Child {
     /// Returns the OS-assigned process identifier associated with this child.
     pub fn id(&self) -> u32 {
-        self.child.id()
+        self.child.inner.id()
     }
 
     /// Forces the child process to exit. If the child has already exited, an [`InvalidInput`]
diff --git a/src/process/unix/mod.rs b/src/process/unix/mod.rs
@@ -44,8 +44,9 @@ use self::{
 use crate::net::driver::Watcher;
 use crate::prelude::*;
 use crate::process::{
-    kill::Kill, Child as SpawnedChild, ChildStderr as SpawnedChildStderr,
-    ChildStdin as SpawnedChildStdin, ChildStdout as SpawnedChildStdout, ExitStatus,
+    kill::{ChildDropGuard, Kill},
+    Child as SpawnedChild, ChildStderr as SpawnedChildStderr, ChildStdin as SpawnedChildStdin,
+    ChildStdout as SpawnedChildStdout, ExitStatus,
 };
 
 lazy_static::lazy_static! {
@@ -111,9 +112,9 @@ pub(crate) fn spawn_child(cmd: &mut process::Command) -> io::Result<SpawnedChild
 
     let signal = Signals::new(&[libc::SIGCHLD])?;
     Ok(SpawnedChild {
-        child: Child {
+        child: ChildDropGuard::new(Child {
             inner: Reaper::new(child, GlobalOrphanQueue, signal),
-        },
+        }),
         stdin: stdin.map(|stdin| SpawnedChildStdin { inner: stdin }),
         stdout: stdout.map(|stdout| SpawnedChildStdout { inner: stdout }),
         stderr: stderr.map(|stderr| SpawnedChildStderr { inner: stderr }),
diff --git a/src/process/windows/mod.rs b/src/process/windows/mod.rs
@@ -1,4 +1,4 @@
-//! Windows asynchronous process handling.
+//! Windows asynchronous process handling.ven
 //!
 //! Like with Unix we don't actually have a way of registering a process with an
 //! IOCP object. As a result we similarly need another mechanism for getting a
@@ -15,31 +15,34 @@
 //! `RegisterWaitForSingleObject` and then wait on the other end of the oneshot
 //! from then on out.
 
-extern crate winapi;
-extern crate mio_named_pipes;
-
 use std::fmt;
-use std::io;
+use std::future::Future;
+use std::io::{self, Read, Write};
 use std::os::windows::prelude::*;
 use std::os::windows::process::ExitStatusExt;
-use std::process::{self, ExitStatus};
+use std::pin::Pin;
+use std::process;
 use std::ptr;
-
-use futures::future::Fuse;
-use futures::sync::oneshot;
-use futures::{Future, Poll, Async};
-use kill::Kill;
-use self::mio_named_pipes::NamedPipe;
-use self::winapi::shared::minwindef::*;
-use self::winapi::shared::winerror::*;
-use self::winapi::um::handleapi::*;
-use self::winapi::um::processthreadsapi::*;
-use self::winapi::um::synchapi::*;
-use self::winapi::um::threadpoollegacyapiset::*;
-use self::winapi::um::winbase::*;
-use self::winapi::um::winnt::*;
-use super::SpawnedChild;
-use tokio_reactor::{Handle, PollEvented};
+use std::task::{Context, Poll};
+
+use futures_channel::oneshot::{channel as oneshot, Receiver, Sender};
+use futures_io::{AsyncRead, AsyncWrite};
+use mio_named_pipes::NamedPipe;
+use winapi::shared::minwindef::*;
+use winapi::shared::winerror::*;
+use winapi::um::handleapi::*;
+use winapi::um::processthreadsapi::*;
+use winapi::um::synchapi::*;
+use winapi::um::threadpoollegacyapiset::*;
+use winapi::um::winbase::*;
+use winapi::um::winnt::*;
+
+use crate::net::driver::Watcher;
+use crate::process::{
+    kill::{ChildDropGuard, Kill},
+    Child as SpawnedChild, ChildStderr as SpawnedChildStderr, ChildStdin as SpawnedChildStdin,
+    ChildStdout as SpawnedChildStdout, ExitStatus,
+};
 
 #[must_use = "futures do nothing unless polled"]
 pub struct Child {
@@ -48,7 +51,7 @@ pub struct Child {
 }
 
 impl fmt::Debug for Child {
-    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         fmt.debug_struct("Child")
             .field("pid", &self.id())
             .field("child", &self.child)
@@ -58,28 +61,28 @@ impl fmt::Debug for Child {
 }
 
 struct Waiting {
-    rx: Fuse<oneshot::Receiver<()>>,
+    rx: Receiver<()>,
     wait_object: HANDLE,
-    tx: *mut Option<oneshot::Sender<()>>,
+    tx: *mut Option<Sender<()>>,
 }
 
 unsafe impl Sync for Waiting {}
 unsafe impl Send for Waiting {}
 
-pub(crate) fn spawn_child(cmd: &mut process::Command, handle: &Handle) -> io::Result<SpawnedChild> {
+pub(crate) fn spawn_child(cmd: &mut process::Command) -> io::Result<SpawnedChild> {
     let mut child = cmd.spawn()?;
-    let stdin = stdio(child.stdin.take(), handle)?;
-    let stdout = stdio(child.stdout.take(), handle)?;
-    let stderr = stdio(child.stderr.take(), handle)?;
+    let stdin = stdio(child.stdin.take())?;
+    let stdout = stdio(child.stdout.take())?;
+    let stderr = stdio(child.stderr.take())?;
 
     Ok(SpawnedChild {
-        child: Child {
+        child: ChildDropGuard::new(Child {
             child,
             waiting: None,
-        },
-        stdin,
-        stdout,
-        stderr,
+        }),
+        stdin: stdin.map(|stdin| SpawnedChildStdin { inner: stdin }),
+        stdout: stdout.map(|stdout| SpawnedChildStdout { inner: stdout }),
+        stderr: stderr.map(|stderr| SpawnedChildStderr { inner: stderr }),
     })
 }
 
@@ -96,42 +99,47 @@ impl Kill for Child {
 }
 
 impl Future for Child {
-    type Item = ExitStatus;
-    type Error = io::Error;
+    type Output = io::Result<ExitStatus>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<<Self as Future>::Output> {
         loop {
             if let Some(ref mut w) = self.waiting {
-                match w.rx.poll().expect("should not be canceled") {
-                    Async::Ready(()) => {}
-                    Async::NotReady => return Ok(Async::NotReady),
+                match Pin::new(&mut w.rx).poll(cx) {
+                    Poll::Ready(Ok(())) => {}
+                    Poll::Ready(Err(_)) => panic!("cancellation was not supposed to happen"),
+                    Poll::Pending => return Poll::Pending,
                 }
-                let status = try!(try_wait(&self.child)).expect("not ready yet");
-                return Ok(status.into())
+                return match try_wait(&self.child) {
+                    Ok(status) => Poll::Ready(Ok(status.expect("not ready yet"))),
+                    Err(err) => Poll::Ready(Err(err)),
+                };
             }
 
-            if let Some(e) = try!(try_wait(&self.child)) {
-                return Ok(e.into())
+            match try_wait(&self.child) {
+                Ok(Some(e)) => return Poll::Ready(Ok(e)),
+                Ok(None) => {}
+                Err(err) => return Poll::Ready(Err(err)),
             }
-            let (tx, rx) = oneshot::channel();
+            let (tx, rx) = oneshot();
             let ptr = Box::into_raw(Box::new(Some(tx)));
             let mut wait_object = ptr::null_mut();
             let rc = unsafe {
-                RegisterWaitForSingleObject(&mut wait_object,
-                                            self.child.as_raw_handle(),
-                                            Some(callback),
-                                            ptr as *mut _,
-                                            INFINITE,
-                                            WT_EXECUTEINWAITTHREAD |
-                                              WT_EXECUTEONLYONCE)
+                RegisterWaitForSingleObject(
+                    &mut wait_object,
+                    self.child.as_raw_handle(),
+                    Some(callback),
+                    ptr as *mut _,
+                    INFINITE,
+                    WT_EXECUTEINWAITTHREAD | WT_EXECUTEONLYONCE,
+                )
             };
             if rc == 0 {
                 let err = io::Error::last_os_error();
                 drop(unsafe { Box::from_raw(ptr) });
-                return Err(err)
+                return Poll::Ready(Err(err));
             }
             self.waiting = Some(Waiting {
-                rx: rx.fuse(),
+                rx,
                 wait_object,
                 tx: ptr,
             });
@@ -151,9 +159,8 @@ impl Drop for Waiting {
     }
 }
 
-unsafe extern "system" fn callback(ptr: PVOID,
-                                   _timer_fired: BOOLEAN) {
-    let complete = &mut *(ptr as *mut Option<oneshot::Sender<()>>);
+unsafe extern "system" fn callback(ptr: PVOID, _timer_fired: BOOLEAN) {
+    let complete = &mut *(ptr as *mut Option<Sender<()>>);
     let _ = complete.take().unwrap().send(());
 }
 
@@ -174,19 +181,57 @@ pub fn try_wait(child: &process::Child) -> io::Result<Option<ExitStatus>> {
     }
 }
 
-pub type ChildStdin = PollEvented<NamedPipe>;
-pub type ChildStdout = PollEvented<NamedPipe>;
-pub type ChildStderr = PollEvented<NamedPipe>;
+impl AsyncWrite for SpawnedChildStdin {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        self.inner.poll_write_with_mut(cx, |inner| inner.write(buf))
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.inner.poll_write_with_mut(cx, |inner| inner.flush())
+    }
+
+    fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl AsyncRead for SpawnedChildStdout {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.inner.poll_read_with_mut(cx, |inner| inner.read(buf))
+    }
+}
+
+impl AsyncRead for SpawnedChildStderr {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.inner.poll_read_with_mut(cx, |inner| inner.read(buf))
+    }
+}
+
+pub type ChildStdin = Watcher<NamedPipe>;
+pub type ChildStdout = Watcher<NamedPipe>;
+pub type ChildStderr = Watcher<NamedPipe>;
 
-fn stdio<T>(option: Option<T>, handle: &Handle)
-            -> io::Result<Option<PollEvented<NamedPipe>>>
-    where T: IntoRawHandle,
+fn stdio<T>(option: Option<T>) -> io::Result<Option<Watcher<NamedPipe>>>
+where
+    T: IntoRawHandle,
 {
     let io = match option {
         Some(io) => io,
         None => return Ok(None),
     };
     let pipe = unsafe { NamedPipe::from_raw_handle(io.into_raw_handle()) };
-    let io = try!(PollEvented::new_with_handle(pipe, handle));
+    let io = Watcher::new(pipe);
     Ok(Some(io))
 }
diff --git a/src/sync/channel.rs b/src/sync/channel.rs

Original file line number	Diff line number	Diff line change
`@@ -270,7 +270,6 @@ cfg_default! {`
`270`	`270`	`pub mod fs;`
`271`	`271`	`pub mod path;`
`272`	`272`	`pub mod net;`
`273`		`- pub mod process;`
`274`	`273`	`}`
`275`	`274`
`276`	`275`	`cfg_unstable! {`
`@@ -280,6 +279,7 @@ cfg_unstable! {`
`280`	`279`	`mod option;`
`281`	`280`	`mod string;`
`282`	`281`	`mod collections;`
	`282`	`+ pub mod process;`
`283`	`283`	`}`
`284`	`284`
`285`	`285`	`cfg_unstable_default! {`
Original file line number	Diff line number	Diff line change
`@@ -7,8 +7,8 @@ use std::task::{Context, Poll};`
`7`	`7`	`/// A drop guard which ensures the child process is killed on drop to maintain`
`8`	`8`	`/// the contract of dropping a Future leads to "cancellation".`
`9`	`9`	`#[derive(Debug)]`
`10`		`-pub(crate) struct ChildDropGuard<T: Kill> {`
`11`		`- inner: T,`
	`10`	`+pub(super) struct ChildDropGuard<T: Kill> {`
	`11`	`+ pub(super) inner: T,`
`12`	`12`	`kill_on_drop: bool,`
`13`	`13`	`}`
`14`	`14`
`@@ -61,7 +61,7 @@ impl<T: Future + Kill + Unpin> Future for ChildDropGuard<T> {`
`61`	`61`	`}`
`62`	`62`
`63`	`63`	`/// An interface for killing a running process.`
`64`		`-pub(crate) trait Kill {`
	`64`	`+pub(super) trait Kill {`
`65`	`65`	`/// Forcefully kill the process.`
`66`	`66`	`fn kill(&mut self) -> io::Result<()>;`
`67`	`67`	`}`