diff --git a/Cargo.toml b/Cargo.toml index 56a75ac42..1ab0f7964 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ An implementation of an asynchronous process management backed futures. [dependencies] tokio-core = "0.1" -futures = "0.1" +futures = "0.1.7" mio = "0.6" log = "0.3" @@ -22,12 +22,8 @@ env_logger = { version = "0.3", default-features = false } [target.'cfg(windows)'.dependencies] winapi = "0.2" kernel32-sys = "0.2" -mio-named-pipes = { git = 'https://github.com/alexcrichton/mio-named-pipes' } +mio-named-pipes = "0.1" [target.'cfg(unix)'.dependencies] libc = "0.2" tokio-signal = "0.1" - -[replace] -"mio:0.6.1" = { git = "https://github.com/alexcrichton/mio", branch = "custom-iocp" } -"tokio-core:0.1.1" = { git = "https://github.com/tokio-rs/tokio-core" } diff --git a/src/lib.rs b/src/lib.rs index 534197229..ecaff5b09 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,35 +1,96 @@ -//! An implementation of process management for Tokio. +//! An implementation of asynchronous process management for Tokio. //! -//! This crate provides `Future` implementations for spawning and waiting -//! on child processes. These implementations are powered by system APIs on -//! Windows and by signals on Unix systems. +//! This crate provides a `CommandExt` trait to enhance the functionality of the +//! `Command` type in the standard library. The three methods provided by this +//! trait mirror the "spawning" methods in the standard library. The +//! `CommandExt` trait in this crate, though, returns "future aware" types that +//! interoperate with Tokio. The asynchronous process support is provided +//! through signal handling on Unix and system APIs on Windows. //! -//! # Usage +//! # Examples //! -//! To achieve efficient polling of running child processes, we will need to -//! set up an event loop from `tokio-core`: -// FIXME: add warning that on Unix systems the *first* event loop can't go away? +//! Here's an example program which will spawn `echo hello world` and then wait +//! for it using an event loop. //! //! ```no_run //! extern crate futures; //! extern crate tokio_core; //! extern crate tokio_process; //! +//! use std::process::Command; +//! //! use futures::Future; //! use tokio_core::reactor::Core; -//! use tokio_process::Command; +//! use tokio_process::CommandExt; //! //! fn main() { -//! let mut event_loop = Core::new().expect("failed to init event loop!"); -//! let mut cmd = Command::new("echo", &event_loop.handle()); -//! cmd.args(&["hello", "world"]); +//! // Create our own local event loop +//! let mut core = Core::new().unwrap(); //! -//! match event_loop.run(cmd.spawn().flatten()) { -//! Ok(status) => println!("exited successfully: {}", status.success()), -//! Err(e) => panic!("failed to run command: {}", e), +//! // Use the standard library's `Command` type to build a process and +//! // then execute it via the `CommandExt` trait. +//! let child = Command::new("echo").arg("hello").arg("world") +//! .spawn_async(&core.handle()); +//! +//! // Make sure our child succeeded in spawning +//! let child = child.expect("failed to spawn"); +//! +//! match core.run(child) { +//! Ok(status) => println!("exit status: {}", status), +//! Err(e) => panic!("failed to wait for exit: {}", e), //! } //! } //! ``` +//! +//! Next, let's take a look at an example where we not only spawn `echo hello +//! world` but we also capture its output. +//! +//! ```no_run +//! extern crate futures; +//! extern crate tokio_core; +//! extern crate tokio_process; +//! +//! use std::process::Command; +//! +//! use futures::Future; +//! use tokio_core::reactor::Core; +//! use tokio_process::CommandExt; +//! +//! fn main() { +//! let mut core = Core::new().unwrap(); +//! +//! // Like above, but use `output_async` which returns a future instead of +//! // immediately returning the `Child`. +//! let output = Command::new("echo").arg("hello").arg("world") +//! .output_async(&core.handle()); +//! let output = core.run(output).expect("failed to collect output"); +//! +//! assert!(output.status.success()); +//! assert_eq!(output.stdout, b"hello world\n"); +//! } +//! ``` +//! +//! # Caveats +//! +//! While similar to the standard library, this crate's `Child` type differs +//! importantly in the behavior of `drop`. In the standard library, a child +//! process will continue running after the instance of `std::process::Child` +//! is dropped. In this crate, however, because `tokio_process::Child` is a +//! future of the child's `ExitStatus`, a child process is terminated if +//! `tokio_process::Child` is dropped. The behavior of the standard library can +//! be regained with the `Child::forget` method. +//! +//! As a final caveat, currently this crate relies on the `tokio-signal` crate +//! and therefore inherits its current restriction. Namely, once a child has +//! been spawned onto an event loop then *that event loop must stay alive for +//! any spawned child in the future to make progress*. In other words, once +//! you spawn a child onto an event loop, you should ensure that the event loop +//! keeps running for the duration of the program if there are multiple event +//! loops. Unfortunately this makes testing particularly tricky, but you can +//! work around this with an initial event loop that just runs forever in the +//! background. + +#![deny(missing_docs)] #[macro_use] extern crate futures; @@ -38,13 +99,13 @@ extern crate mio; #[macro_use] extern crate log; -use std::ffi::OsStr; use std::io::{self, Read, Write}; -use std::path::Path; -use std::process::{self, ExitStatus}; +use std::process::{ExitStatus, Command, Output, Stdio}; -use futures::{Future, Poll}; +use futures::{Future, Poll, IntoFuture}; +use futures::future::{Flatten, FutureResult, Either, ok}; use tokio_core::reactor::Handle; +use tokio_core::io::{IoFuture, read_to_end}; #[path = "unix.rs"] #[cfg(unix)] @@ -54,167 +115,210 @@ mod imp; #[cfg(windows)] mod imp; -pub struct Command { - inner: process::Command, - #[allow(dead_code)] - handle: Handle, +/// Extensions provided by this crate to the `Command` type in the standard +/// library. +/// +/// This crate primarily enhances the standard library's `Command` type with +/// asynchronous capabilities. The currently three blocking functions in the +/// standard library, `spawn`, `status`, and `output`, all have asynchronous +/// versions through this trait. +/// +/// Note that the `Child` type spawned is specific to this crate, and that the +/// I/O handles created from this crate are all asynchronous as well (differing +/// from their `std` counterparts). +pub trait CommandExt { + /// Executes the command as a child process, returning a handle to it. + /// + /// By default, stdin, stdout and stderr are inherited from the parent. + /// + /// This method will spawn the child process synchronously and return a + /// handle to a future-aware child process. The `Child` returned implements + /// `Future` itself to acquire the `ExitStatus` of the child, and otherwise + /// the `Child` has methods to acquire handles to the stdin, stdout, and + /// stderr streams. + /// + /// The `handle` specified to this method must be a handle to a valid event + /// loop, and all I/O this child does will be associated with the specified + /// event loop. + fn spawn_async(&mut self, handle: &Handle) -> io::Result; + + /// Executes a command as a child process, waiting for it to finish and + /// collecting its exit status. + /// + /// By default, stdin, stdout and stderr are inherited from the parent. + /// + /// The `StatusAsync` future returned will resolve to the `ExitStatus` + /// type in the standard library representing how the process exited. If + /// output handles are set to a pipe then they will be immediately closed + /// after the child is spawned. + /// + /// The `handle` specified must be a handle to a valid event loop, and all + /// I/O this child does will be associated with the specified event loop. + /// + /// If the `OutputAsync` future is dropped before the future resolves, then + /// the child will be killed, if it was spawned. + fn status_async(&mut self, handle: &Handle) -> StatusAsync; + + /// Executes the command as a child process, waiting for it to finish and + /// collecting all of its output. + /// + /// > **Note**: this method, unlike the standard library, will + /// > unconditionally configure the stdout/stderr handles to be pipes, even + /// > if they have been previously configured. If this is not desired then + /// > the `spawn_async` method should be used in combination with the + /// > `wait_with_output` method on child. + /// + /// This method will return a future representing the collection of the + /// child process's stdout/stderr. The `OutputAsync` future will resolve to + /// the `Output` type in the standard library, containing `stdout` and + /// `stderr` as `Vec` along with an `ExitStatus` representing how the + /// process exited. + /// + /// The `handle` specified must be a handle to a valid event loop, and all + /// I/O this child does will be associated with the specified event loop. + /// + /// If the `OutputAsync` future is dropped before the future resolves, then + /// the child will be killed, if it was spawned. + fn output_async(&mut self, handle: &Handle) -> OutputAsync; } -/// A future that represents a spawned child process. -/// -/// This future is created by the `Command::spawn` method. -/// -/// If the caller does not care about the intermediate handle to a spawned -/// child, this future can be `flatten`ed to directly compute the child's -/// exit status. -pub struct Spawn { - inner: Box>, + +impl CommandExt for Command { + fn spawn_async(&mut self, handle: &Handle) -> io::Result { + let mut child = Child { + child: imp::Child::new(try!(self.spawn()), handle), + stdin: None, + stdout: None, + stderr: None, + kill_on_drop: true, + }; + child.stdin = try!(child.child.register_stdin(handle)).map(|io| { + ChildStdin { inner: io } + }); + child.stdout = try!(child.child.register_stdout(handle)).map(|io| { + ChildStdout { inner: io } + }); + child.stderr = try!(child.child.register_stderr(handle)).map(|io| { + ChildStderr { inner: io } + }); + Ok(child) + } + + fn status_async(&mut self, handle: &Handle) -> StatusAsync { + StatusAsync { + inner: self.spawn_async(handle).into_future().flatten(), + } + } + + fn output_async(&mut self, handle: &Handle) -> OutputAsync { + self.stdout(Stdio::piped()); + self.stderr(Stdio::piped()); + OutputAsync { + inner: self.spawn_async(handle).into_future().and_then(|c| { + c.wait_with_output() + }).boxed(), + } + } } -/// A future that represents the exit status of a running or exited child process. +/// Representation of a child process spawned onto an event loop. /// -/// This future is created by successfully polling the `Spawn` future. +/// This type is also a future which will yield the `ExitStatus` of the +/// underlying child process. A `Child` here also provides access to information +/// like the OS-assigned identifier and the stdio streams. /// -/// # Note -/// -/// Take note that there is no implementation of `Drop` for this future, -/// 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. +/// > **Note**: The behavior of `drop` on a child in this crate is *different +/// > than the behavior of the standard library*. If a `tokio_process::Child` is +/// > dropped before the process finishes then the process will be terminated. +/// > In the standard library, however, the process continues executing. This is +/// > done because futures in general take `drop` as a sign of cancellation, and +/// > this `Child` is itself a future. If you'd like to run a process in the +/// > background, though, you may use the `forget` method. pub struct Child { - inner: imp::Child, + child: imp::Child, + kill_on_drop: bool, stdin: Option, stdout: Option, stderr: Option, } -pub struct ChildStdin { - inner: imp::ChildStdin, -} - -pub struct ChildStdout { - inner: imp::ChildStdout, -} - -pub struct ChildStderr { - inner: imp::ChildStderr, -} - -impl Command { - pub fn new>(exe: T, handle: &Handle) -> Command { - Command::_new(exe.as_ref(), handle) - } - - fn _new(exe: &OsStr, handle: &Handle) -> Command { - Command { - inner: process::Command::new(exe), - handle: handle.clone(), - } - } - - pub fn arg>(&mut self, arg: S) -> &mut Command { - self._arg(arg.as_ref()) - } - - fn _arg(&mut self, arg: &OsStr) -> &mut Command { - self.inner.arg(arg); - self - } - - pub fn args>(&mut self, args: &[S]) -> &mut Command { - for arg in args { - self._arg(arg.as_ref()); - } - self - } - - pub fn env(&mut self, key: K, val: V) -> &mut Command - where K: AsRef, V: AsRef - { - self._env(key.as_ref(), val.as_ref()) - } - - fn _env(&mut self, key: &OsStr, val: &OsStr) -> &mut Command { - self.inner.env(key, val); - self - } - - pub fn env_remove>(&mut self, key: K) -> &mut Command { - self._env_remove(key.as_ref()) - } - - fn _env_remove(&mut self, key: &OsStr) -> &mut Command { - self.inner.env_remove(key); - self - } - - pub fn env_clear(&mut self) -> &mut Command { - self.inner.env_clear(); - self - } - - pub fn current_dir>(&mut self, dir: P) -> &mut Command { - self._current_dir(dir.as_ref()) - } - - fn _current_dir(&mut self, dir: &Path) -> &mut Command { - self.inner.current_dir(dir); - self - } - - pub fn stdin(&mut self, cfg: process::Stdio) -> &mut Self { - self.inner.stdin(cfg); - self - } - - pub fn stdout(&mut self, cfg: process::Stdio) -> &mut Self { - self.inner.stdout(cfg); - self - } - pub fn stderr(&mut self, cfg: process::Stdio) -> &mut Self { - self.inner.stderr(cfg); - self - } - - pub fn spawn(self) -> Spawn { - Spawn { - inner: Box::new(imp::spawn(self)), - } - } -} - -impl Future for Spawn { - type Item = Child; - type Error = io::Error; - - fn poll(&mut self) -> Poll { - self.inner.poll() - } -} - impl Child { /// Returns the OS-assigned process identifier associated with this child. pub fn id(&self) -> u32 { - self.inner.id() + self.child.id() } - /// Forces the child to exit. This is equivalent to sending a - /// SIGKILL on unix platforms. + /// Forces the child to exit. + /// + /// This is equivalent to sending a SIGKILL on unix platforms. pub fn kill(&mut self) -> io::Result<()> { - self.inner.kill() + self.child.kill() } + /// Returns a handle for writing to the child's stdin, if it has been + /// captured pub fn stdin(&mut self) -> &mut Option { &mut self.stdin } + /// Returns a handle for writing to the child's stdout, if it has been + /// captured pub fn stdout(&mut self) -> &mut Option { &mut self.stdout } + /// Returns a handle for writing to the child's stderr, if it has been + /// captured pub fn stderr(&mut self) -> &mut Option { &mut self.stderr } + + /// Returns a future that will resolve to an `Output`, containing the exit + /// status, stdout, and stderr of the child process. + /// + /// The returned future will simultaneously waits for the child to exit and + /// collect all remaining output on the stdout/stderr handles, returning an + /// `Output` instance. + /// + /// The stdin handle to the child process, if any, will be closed before + /// waiting. This helps avoid deadlock: it ensures that the child does not + /// block waiting for input from the parent, while the parent waits for the + /// child to exit. + /// + /// By default, stdin, stdout and stderr are inherited from the parent. In + /// order to capture the output into this `Output` it is necessary to create + /// new pipes between parent and child. Use `stdout(Stdio::piped())` or + /// `stderr(Stdio::piped())`, respectively, when creating a `Command`. + pub fn wait_with_output(mut self) -> WaitWithOutput { + drop(self.stdin().take()); + let stdout = match self.stdout().take() { + Some(io) => Either::A(read_to_end(io, Vec::new()).map(|p| p.1)), + None => Either::B(ok(Vec::new())), + }; + let stderr = match self.stderr().take() { + Some(io) => Either::A(read_to_end(io, Vec::new()).map(|p| p.1)), + None => Either::B(ok(Vec::new())), + }; + + WaitWithOutput { + inner: self.join(stdout).join(stderr).map(|((status, stdout), stderr)| { + Output { + status: status, + stdout: stdout, + stderr: stderr, + } + }).boxed() + } + } + + /// Drop this `Child` without killing the underlying process. + /// + /// Normally a `Child` is killed if it's still alive when dropped, but this + /// method will ensure that the child may continue running once the `Child` + /// instance is dropped. + pub fn forget(mut self) { + self.kill_on_drop = false; + } } impl Future for Child { @@ -222,10 +326,100 @@ impl Future for Child { type Error = io::Error; fn poll(&mut self) -> Poll { + self.child.poll_exit() + } +} + +impl Drop for Child { + fn drop(&mut self) { + if self.kill_on_drop { + drop(self.kill()); + } + } +} + +/// Future returned from the `Child::wait_with_output` method. +/// +/// This future will resolve to the standard library's `Output` type which +/// contains the exit status, stdout, and stderr of a child process. +pub struct WaitWithOutput { + inner: IoFuture, +} + +impl Future for WaitWithOutput { + type Item = Output; + type Error = io::Error; + + fn poll(&mut self) -> Poll { self.inner.poll() } } +/// Future returned by the `CommandExt::status_async` method. +/// +/// This future is used to conveniently spawn a child and simply wait for its +/// exit status. This future will resolves to the `ExitStatus` type in the +/// standard library. +pub struct StatusAsync { + inner: Flatten>, +} + +impl Future for StatusAsync { + type Item = ExitStatus; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.inner.poll() + } +} + +/// Future returned by the `CommandExt::output_async` method. +/// +/// This future is mostly equivalent to spawning a process and then calling +/// `wait_with_output` on it internally. This can be useful to simply spawn a +/// process, collecting all of its output and its exit status. +pub struct OutputAsync { + inner: IoFuture, +} + +impl Future for OutputAsync { + type Item = Output; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.inner.poll() + } +} + +/// The standard input stream for spawned children. +/// +/// This type implements the `Write` trait to pass data to the stdin handle of +/// a child process. Note that this type is also "futures aware" meaning that it +/// is both (a) nonblocking and (b) will panic if used off of a future's task. +pub struct ChildStdin { + inner: imp::ChildStdin, +} + +/// The standard output stream for spawned children. +/// +/// This type implements the `Read` trait to read data from the stdout handle +/// of a child process. Note that this type is also "futures aware" meaning +/// that it is both (a) nonblocking and (b) will panic if used off of a +/// future's task. +pub struct ChildStdout { + inner: imp::ChildStdout, +} + +/// The standard error stream for spawned children. +/// +/// This type implements the `Read` trait to read data from the stderr handle +/// of a child process. Note that this type is also "futures aware" meaning +/// that it is both (a) nonblocking and (b) will panic if used off of a +/// future's task. +pub struct ChildStderr { + inner: imp::ChildStderr, +} + impl Write for ChildStdin { fn write(&mut self, bytes: &[u8]) -> io::Result { self.inner.write(bytes) diff --git a/src/unix.rs b/src/unix.rs index 9c74f182e..b29e9b1f7 100644 --- a/src/unix.rs +++ b/src/unix.rs @@ -1,3 +1,26 @@ +//! Unix handling of child processes +//! +//! Right now the only "fancy" thing about this is how we implement the +//! `Future` implementation on `Child` to get the exit status. Unix offers +//! no way to register a child with epoll, and the only real way to get a +//! notification when a process exits is the SIGCHLD signal. +//! +//! Signal handling in general is *super* hairy and complicated, and it's even +//! more complicated here with the fact that signals are coalesced, so we may +//! not get a SIGCHLD-per-child. +//! +//! Our best approximation here is to check *all spawned processes* for all +//! SIGCHLD signals received. To do that we create a `Signal`, implemented in +//! the `tokio-signal` crate, which is a stream over signals being received. +//! +//! Later when we poll the process's exit status we simply check to see if a +//! SIGCHLD has happened since we last checked, and while that returns "yes" we +//! keep trying. +//! +//! Note that this means that this isn't really scalable, but then again +//! processes in general aren't scalable (e.g. millions) so it shouldn't be that +//! bad in theory... + extern crate libc; extern crate tokio_signal; @@ -5,104 +28,64 @@ use std::io; use std::os::unix::prelude::*; use std::process::{self, ExitStatus}; -use futures::stream::Stream; -use futures::{Future, Poll, Async}; -use tokio_core::reactor::{Handle,PollEvented}; +use futures::future::FlattenStream; +use futures::{Future, Poll, Async, Stream}; +use mio::unix::EventedFd; +use mio::{Evented, PollOpt, Ready, Token}; +use mio; use self::libc::c_int; use self::tokio_signal::unix::Signal; - -use mio; -use mio::{Evented, PollOpt, Ready, Token}; -use mio::unix::EventedFd; - -use Command; +use tokio_core::io::IoFuture; +use tokio_core::reactor::{Handle, PollEvented}; pub struct Child { - child: process::Child, + inner: process::Child, reaped: bool, - sigchld: Signal, -} - -/// Spawns a new child process. -/// -/// Right now the only "fancy" thing about this is how we implement the -/// `Future` implementation on `Child` to get the exit status. Unix offers -/// no way to register a child with epoll, and the only real way to get a -/// notification when a process exits is the SIGCHLD signal. -/// -/// Signal handling in general is *super* hairy and complicated, and it's even -/// more complicated here with the fact that signals are coalesced, so we may -/// not get a SIGCHLD-per-child. -/// -/// Our best approximation here is to check *all spawned processes* for all -/// SIGCHLD signals received. To do that we create a `Signal`, implemented in -/// the `tokio-signal` crate, which is a stream over signals being received. -/// -/// Later when we poll the process's exit status we simply check to see if a -/// SIGCHLD has happened since we last checked, and while that returns "yes" we -/// keep trying. -/// -/// Note that this means that this isn't really scalable, but then again -/// processes in general aren't scalable (e.g. millions) so it shouldn't be that -/// bad in theory... -pub fn spawn(mut cmd: Command) -> Box> { - struct KillOnDrop(Option); - - impl Drop for KillOnDrop { - fn drop(&mut self) { - if let Some(mut c) = self.0.take() { - drop(c.kill()); - } - } - } - - Box::new(Signal::new(libc::SIGCHLD, &cmd.handle).and_then(move |sigchld| { - cmd.inner.spawn().and_then(|mut c| { - let stdin = c.stdin.take(); - let stdout = c.stdout.take(); - let stderr = c.stderr.take(); - let mut c = KillOnDrop(Some(c)); - let stdin = try!(stdio(stdin, &cmd.handle)); - let stdout = try!(stdio(stdout, &cmd.handle)); - let stderr = try!(stdio(stderr, &cmd.handle)); - Ok(::Child { - inner: Child { - child: c.0.take().unwrap(), - reaped: false, - sigchld: sigchld, - }, - stdin: stdin.map(|io| ::ChildStdin { inner: io }), - stdout: stdout.map(|io| ::ChildStdout { inner: io }), - stderr: stderr.map(|io| ::ChildStderr { inner: io }), - }) - }) - })) + sigchld: FlattenStream>, } impl Child { + pub fn new(inner: process::Child, handle: &Handle) -> Child { + Child { + inner: inner, + reaped: false, + sigchld: Signal::new(libc::SIGCHLD, handle).flatten_stream(), + } + } + + pub fn register_stdin(&mut self, handle: &Handle) + -> io::Result> { + stdio(self.inner.stdin.take(), handle) + } + + pub fn register_stdout(&mut self, handle: &Handle) + -> io::Result> { + stdio(self.inner.stdout.take(), handle) + } + + pub fn register_stderr(&mut self, handle: &Handle) + -> io::Result> { + stdio(self.inner.stderr.take(), handle) + } + pub fn id(&self) -> u32 { - self.child.id() + self.inner.id() } pub fn kill(&mut self) -> io::Result<()> { if self.reaped { Ok(()) } else { - self.child.kill() + self.inner.kill() } } -} -impl Future for Child { - type Item = ExitStatus; - type Error = io::Error; - - fn poll(&mut self) -> Poll { + pub fn poll_exit(&mut self) -> Poll { assert!(!self.reaped); loop { // Ensure that once we've successfully waited we won't try to // `kill` above. - if let Some(e) = try!(try_wait(&self.child)) { + if let Some(e) = try!(self.try_wait()) { self.reaped = true; return Ok(e.into()) } @@ -118,24 +101,24 @@ impl Future for Child { } } } -} -pub fn try_wait(child: &process::Child) -> io::Result> { - let id = child.id() as c_int; - let mut status = 0; - loop { - match unsafe { libc::waitpid(id, &mut status, libc::WNOHANG) } { - 0 => return Ok(None), - n if n < 0 => { - let err = io::Error::last_os_error(); - if err.kind() == io::ErrorKind::Interrupted { - continue + fn try_wait(&self) -> io::Result> { + let id = self.id() as c_int; + let mut status = 0; + loop { + match unsafe { libc::waitpid(id, &mut status, libc::WNOHANG) } { + 0 => return Ok(None), + n if n < 0 => { + let err = io::Error::last_os_error(); + if err.kind() == io::ErrorKind::Interrupted { + continue + } + return Err(err) + } + n => { + assert_eq!(n, id); + return Ok(Some(ExitStatus::from_raw(status))) } - return Err(err) - } - n => { - assert_eq!(n, id); - return Ok(Some(ExitStatus::from_raw(status))) } } } diff --git a/src/windows.rs b/src/windows.rs index ccea91c77..7feaa6e05 100644 --- a/src/windows.rs +++ b/src/windows.rs @@ -1,3 +1,20 @@ +//! Windows asynchronous process handling. +//! +//! 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 +//! signal when a process has exited. For now this is implemented with the +//! `RegisterWaitForSingleObject` function in the kernel32.dll. +//! +//! This strategy is the same that libuv takes and essentially just queues up a +//! wait for the process in a kernel32-specific thread pool. Once the object is +//! notified (e.g. the process exits) then we have a callback that basically +//! just completes a `Oneshot`. +//! +//! The `poll_exit` implementation will attempt to wait for the process in a +//! nonblocking fashion, but failing that it'll fire off a +//! `RegisterWaitForSingleObject` and then wait on the other end of the oneshot +//! from then on out. + extern crate winapi; extern crate kernel32; extern crate mio_named_pipes; @@ -7,11 +24,9 @@ use std::os::windows::prelude::*; use std::os::windows::process::ExitStatusExt; use std::process::{self, ExitStatus}; -use tokio_core::reactor::{PollEvented, Handle}; -use futures::{self, Future, Poll, Async, Oneshot, Complete, oneshot, Fuse}; +use futures::{Future, Poll, Async, Oneshot, Complete, oneshot, Fuse}; use self::mio_named_pipes::NamedPipe; - -use Command; +use tokio_core::reactor::{PollEvented, Handle}; pub struct Child { child: process::Child, @@ -27,39 +42,29 @@ struct Waiting { unsafe impl Sync for Waiting {} unsafe impl Send for Waiting {} -pub fn spawn(mut cmd: Command) -> Box> { - struct KillOnDrop(Option); - - impl Drop for KillOnDrop { - fn drop(&mut self) { - if let Some(mut c) = self.0.take() { - drop(c.kill()); - } +impl Child { + pub fn new(child: process::Child, _handle: &Handle) -> Child { + Child { + child: child, + waiting: None, } } - Box::new(futures::done(cmd.inner.spawn().and_then(|mut c| { - let stdin = c.stdin.take(); - let stdout = c.stdout.take(); - let stderr = c.stderr.take(); - let mut c = KillOnDrop(Some(c)); - let stdin = try!(stdio(stdin, &cmd.handle)); - let stdout = try!(stdio(stdout, &cmd.handle)); - let stderr = try!(stdio(stderr, &cmd.handle)); + pub fn register_stdin(&mut self, handle: &Handle) + -> io::Result> { + stdio(self.child.stdin.take(), handle) + } - Ok(::Child { - inner: Child { - child: c.0.take().unwrap(), - waiting: None, - }, - stdin: stdin.map(|io| ::ChildStdin { inner: io }), - stdout: stdout.map(|io| ::ChildStdout { inner: io }), - stderr: stderr.map(|io| ::ChildStderr { inner: io }), - }) - }))) -} + pub fn register_stdout(&mut self, handle: &Handle) + -> io::Result> { + stdio(self.child.stdout.take(), handle) + } + + pub fn register_stderr(&mut self, handle: &Handle) + -> io::Result> { + stdio(self.child.stderr.take(), handle) + } -impl Child { pub fn id(&self) -> u32 { self.child.id() } @@ -67,13 +72,8 @@ impl Child { pub fn kill(&mut self) -> io::Result<()> { self.child.kill() } -} -impl Future for Child { - type Item = ExitStatus; - type Error = io::Error; - - fn poll(&mut self) -> Poll { + pub fn poll_exit(&mut self) -> Poll { loop { if let Some(ref mut w) = self.waiting { match w.rx.poll().expect("should not be canceled") { @@ -100,8 +100,9 @@ impl Future for Child { winapi::WT_EXECUTEONLYONCE) }; if rc == 0 { + let err = io::Error::last_os_error(); drop(unsafe { Box::from_raw(ptr) }); - return Err(io::Error::last_os_error()) + return Err(err) } self.waiting = Some(Waiting { rx: rx.fuse(), diff --git a/tests/smoke.rs b/tests/smoke.rs index b156920a6..142e1a444 100644 --- a/tests/smoke.rs +++ b/tests/smoke.rs @@ -6,9 +6,10 @@ use std::env; use std::sync::mpsc::channel; use std::sync::{Once, ONCE_INIT}; use std::thread; +use std::process::Command; -use tokio_core::reactor::{Core, Handle}; -use tokio_process::Command; +use tokio_core::reactor::Core; +use tokio_process::CommandExt; static INIT: Once = ONCE_INIT; @@ -17,8 +18,8 @@ fn init() { let (tx, rx) = channel(); thread::spawn(move || { let mut lp = Core::new().unwrap(); - let cmd = exit(&lp.handle()); - let mut child = lp.run(cmd.spawn()).unwrap(); + let mut cmd = exit(); + let mut child = cmd.spawn_async(&lp.handle()).unwrap(); drop(child.kill()); lp.run(child).unwrap(); tx.send(()).unwrap(); @@ -28,14 +29,14 @@ fn init() { }); } -fn exit(handle: &Handle) -> Command { +fn exit() -> Command { let mut me = env::current_exe().unwrap(); me.pop(); if me.ends_with("deps") { me.pop(); } me.push("exit"); - Command::new(me, handle) + Command::new(me) } #[test] @@ -43,9 +44,9 @@ fn simple() { init(); let mut lp = Core::new().unwrap(); - let mut cmd = exit(&lp.handle()); + let mut cmd = exit(); cmd.arg("2"); - let mut child = lp.run(cmd.spawn()).unwrap(); + let mut child = cmd.spawn_async(&lp.handle()).unwrap(); let id = child.id(); assert!(id > 0); let status = lp.run(&mut child).unwrap(); diff --git a/tests/stdio.rs b/tests/stdio.rs index aac0df345..71d103861 100644 --- a/tests/stdio.rs +++ b/tests/stdio.rs @@ -8,22 +8,22 @@ extern crate env_logger; use std::env; use std::io; -use std::process::{Stdio, ExitStatus}; +use std::process::{Stdio, ExitStatus, Command}; use futures::{Future, BoxFuture}; use futures::stream::{self, Stream}; -use tokio_core::io::{read_until, write_all}; -use tokio_core::reactor::{Core, Handle}; -use tokio_process::{Command, Child}; +use tokio_core::io::{read_until, write_all, read_to_end}; +use tokio_core::reactor::Core; +use tokio_process::{CommandExt, Child}; -fn cat(handle: &Handle) -> Command { +fn cat() -> Command { let mut path = env::current_exe().unwrap(); path.pop(); if path.ends_with("deps") { path.pop(); } path.push("cat"); - let mut cmd = Command::new(path, handle); + let mut cmd = Command::new(path); cmd.stdin(Stdio::piped()) .stdout(Stdio::piped()); cmd @@ -85,16 +85,29 @@ fn feed_cat(mut cat: Child, n: usize) -> BoxFuture { /// concurrently; otherwise this would deadlock. /// /// - We read the same lines from the child that we fed it. -// +/// /// - The child does produce EOF on stdout after the last line. fn feed_a_lot() { let _ = ::env_logger::init(); let mut lp = Core::new().unwrap(); - let cmd = cat(&lp.handle()); - let child = cmd.spawn().and_then(|child| { - feed_cat(child, 10000) - }); - let status = lp.run(child).unwrap(); + let child = cat().spawn_async(&lp.handle()).unwrap(); + let status = lp.run(feed_cat(child, 10000)).unwrap(); assert_eq!(status.code(), Some(0)); } + +#[test] +fn drop_kills() { + let _ = ::env_logger::init(); + + let mut lp = Core::new().unwrap(); + let mut child = cat().spawn_async(&lp.handle()).unwrap(); + let stdin = child.stdin().take().unwrap(); + let stdout = child.stdout().take().unwrap(); + drop(child); + + let (_, output) = lp.run(read_to_end(stdout, Vec::new())).unwrap(); + assert_eq!(output.len(), 0); + let err = lp.run(write_all(stdin, b"1234")).err().unwrap(); + assert_eq!(err.kind(), io::ErrorKind::BrokenPipe); +}