diff --git a/tokio-process/CHANGELOG.md b/tokio-process/CHANGELOG.md new file mode 100644 index 000000000..61bb06dfa --- /dev/null +++ b/tokio-process/CHANGELOG.md @@ -0,0 +1,71 @@ +## 0.2.4 - 2019-06-21 +### Fixed +* Proccesses "leaked" via `Child::forget` now reaped rather than left as zombies +for the duration of the parent process. +* Dropping a `Child` process no longer blocks the caller until the process fully +exits. This avoids a pathological deadlock if the kernel doesn't kill the child. + +### Changed +* Updated the example program for reading lines from a child process to be more +flexible to be copy/pasted and iterated upon. + +## 0.2.3 - 2018-11-01 +### Added +* `ChildStd{in, out, err}` now implement `AsRawFd`/`AsRawHandle` on Unix/Windows +systems, respectively. + +## 0.2.2 - 2018-05-27 +### Fixed +- Fixed a pathological situation where a signal could be missed if it arrived +after polling the child but before registering for a new notification + +## 0.2.1 - 2018-05-18 +### Changed +- **Breaking**: asynchronous spawning of a child process now requires using a +reactor handle from the `tokio` crate instead of the `tokio-core` crate +- Child processes may be spawned without specifying a `tokio` handle at all +(the current/default reactor handle will be used) +### Removed +- **Breaking**: removed all previously deprecated items + +## 0.1.6 - 2018-05-09 +### Fixed +- On Unix systems, any child processes that are `kill`ed (or implicitly killed +via dropping the child without calling `forget`) are no longer left in a zombie +state, which allows the OS to reclaim the process. + +## 0.1.5 - 2018-01-03 +### Changed +- Minimum required version of `winapi` has been bumped to `0.3`. + +## 0.1.4 - 2017-06-25 +### Fixed +- Added missing `Debug` impls on all types. +- Added missing `must_use` annotations on all futures. +- Ensure `status_async` closes child's stdio handles after spawning in order +to prevent potential deadlocks when attempting to interact with any pipes held +by the parent process. + +## 0.1.3 - 2017-03-15 +### Changed +- Minimum required version of `futures` has been bumped to `0.1.11`. +- Minimum required version of `mio` has been bumped to `0.6.5`. +- Minimum required version of `tokio-core` has been bumped to `0.1.6`. + +## 0.1.2 - 2017-01-24 +### Changed +- Minimum required version of `tokio-signal` has been bumped to `0.1.2`. +### Fixed +- The event loop which spawns the first async child no longer needs to be kept +alive for subsequent child spawns to make progress. + +## 0.1.1 - 2016-12-19 +### Added +- Support performing async I/O operations on the child's stdio handles. +### Changed +- Functionality has been reimplemented as the `CommandExt` extension trait +(implemented directly on `std::process::Command`) instead of going through +the locally vendored `Command` type. + +## 0.1.0 - 2016-09-10 +- First release! diff --git a/tokio-process/Cargo.toml b/tokio-process/Cargo.toml new file mode 100644 index 000000000..2c3112b84 --- /dev/null +++ b/tokio-process/Cargo.toml @@ -0,0 +1,55 @@ +[package] +name = "tokio-process" +# When releasing to crates.io: +# - Update html_root_url. +# - Update CHANGELOG.md. +# - Create "X.Y.Z" git tag. +version = "0.2.4" +edition = "2018" +authors = ["Tokio Contributors "] +license = "MIT" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://github.com/tokio-rs/tokio" +documentation = "https://docs.rs/tokio-process/0.2.4/tokio_process" +description = """ +An implementation of an asynchronous process management backed futures. +""" +categories = ["asynchronous"] + +[dependencies] +futures = "0.1.11" +tokio-io = "0.1" +tokio-reactor = "0.1" + +[dev-dependencies] +failure = "0.1" +log = "0.4" + +[dev-dependencies.tokio] +version = "0.1" +default-features = false +features = ["rt-full"] + +[target.'cfg(windows)'.dependencies] +mio-named-pipes = "0.1" + +[target.'cfg(windows)'.dependencies.winapi] +version = "0.3" +features = [ + "handleapi", + "winerror", + "minwindef", + "processthreadsapi", + "synchapi", + "threadpoollegacyapiset", + "winbase", + "winnt", +] + +[target.'cfg(unix)'.dependencies] +crossbeam-queue = "0.1.2" +lazy_static = "1.3" +libc = "0.2" +log = "0.4" +mio = "0.6.5" +tokio-signal = "0.2.5" diff --git a/tokio-process/LICENSE b/tokio-process/LICENSE new file mode 100644 index 000000000..cdb28b4b5 --- /dev/null +++ b/tokio-process/LICENSE @@ -0,0 +1,25 @@ +Copyright (c) 2019 Tokio Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/tokio-process/README.md b/tokio-process/README.md new file mode 100644 index 000000000..9a252d9de --- /dev/null +++ b/tokio-process/README.md @@ -0,0 +1,48 @@ +# tokio-process + +An implementation of process management for Tokio + +[Documentation](https://docs.rs/tokio-process/0.2.4/tokio_process) + +## Usage + +First, add this to your `Cargo.toml`: + +```toml +[dependencies] +tokio-process = "0.2" +``` + +Next you can use this in conjunction with the `tokio` and `futures` crates: + +```rust,no_run +use std::process::Command; + +use futures::Future; +use tokio_process::CommandExt; + +fn main() { + // 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(); + + // Make sure our child succeeded in spawning and process the result + let future = child + .expect("failed to spawn") + .map(|status| println!("exit status: {}", status)) + .map_err(|e| panic!("failed to wait for exit: {}", e)); + + // Send the future to the tokio runtime for execution + tokio::run(future) +} +``` + +## License + +This project is licensed under the [MIT license](./LICENSE). + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, shall be licensed as MIT, without any additional +terms or conditions. diff --git a/tokio-process/src/bin/cat.rs b/tokio-process/src/bin/cat.rs new file mode 100644 index 000000000..7acfeb695 --- /dev/null +++ b/tokio-process/src/bin/cat.rs @@ -0,0 +1,19 @@ +// A cat-like utility that can be used as a subprocess to test I/O +// stream communication. +use std::io; +use std::io::Write; + +fn main() { + let stdin = io::stdin(); + let mut stdout = io::stdout(); + let mut line = String::new(); + loop { + line.clear(); + stdin.read_line(&mut line).unwrap(); + if line.is_empty() { + break; + } + stdout.write_all(line.as_bytes()).unwrap(); + } + stdout.flush().unwrap(); +} diff --git a/tokio-process/src/bin/exit.rs b/tokio-process/src/bin/exit.rs new file mode 100644 index 000000000..8db0a4a65 --- /dev/null +++ b/tokio-process/src/bin/exit.rs @@ -0,0 +1,5 @@ +#[allow(dead_code)] + +fn main() { + std::process::exit(std::env::args().nth(1).unwrap().parse().unwrap()); +} diff --git a/tokio-process/src/kill.rs b/tokio-process/src/kill.rs new file mode 100644 index 000000000..25d7d9a5d --- /dev/null +++ b/tokio-process/src/kill.rs @@ -0,0 +1,13 @@ +use std::io; + +/// An interface for killing a running process. +pub(crate) trait Kill { + /// Forcefully kill the process. + fn kill(&mut self) -> io::Result<()>; +} + +impl<'a, T: 'a + Kill> Kill for &'a mut T { + fn kill(&mut self) -> io::Result<()> { + (**self).kill() + } +} diff --git a/tokio-process/src/lib.rs b/tokio-process/src/lib.rs new file mode 100644 index 000000000..e9e963689 --- /dev/null +++ b/tokio-process/src/lib.rs @@ -0,0 +1,870 @@ +//! An implementation of asynchronous process management for Tokio. +//! +//! 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. +//! +//! # Examples +//! +//! 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; +//! extern crate tokio_process; +//! +//! use std::process::Command; +//! +//! use futures::Future; +//! use tokio_process::CommandExt; +//! +//! fn main() { +//! // 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(); +//! +//! // Make sure our child succeeded in spawning and process the result +//! let future = child.expect("failed to spawn") +//! .map(|status| println!("exit status: {}", status)) +//! .map_err(|e| panic!("failed to wait for exit: {}", e)); +//! +//! // Send the future to the tokio runtime for execution +//! tokio::run(future) +//! } +//! ``` +//! +//! 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; +//! extern crate tokio_process; +//! +//! use std::process::Command; +//! +//! use futures::Future; +//! use tokio_process::CommandExt; +//! +//! fn main() { +//! // 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(); +//! +//! let future = output.map_err(|e| panic!("failed to collect output: {}", e)) +//! .map(|output| { +//! assert!(output.status.success()); +//! assert_eq!(output.stdout, b"hello world\n"); +//! }); +//! +//! tokio::run(future); +//! } +//! ``` +//! +//! We can also read input line by line. +//! +//! ```no_run +//! extern crate failure; +//! extern crate futures; +//! extern crate tokio; +//! extern crate tokio_process; +//! extern crate tokio_io; +//! +//! use failure::Error; +//! use futures::{Future, Stream}; +//! use std::io::BufReader; +//! use std::process::{Command, Stdio}; +//! use tokio_process::{Child, ChildStdout, CommandExt}; +//! +//! fn lines_stream(child: &mut Child) -> impl Stream + Send + 'static { +//! let stdout = child.stdout().take() +//! .expect("child did not have a handle to stdout"); +//! +//! tokio_io::io::lines(BufReader::new(stdout)) +//! // Convert any io::Error into a failure::Error for better flexibility +//! .map_err(|e| Error::from(e)) +//! // We print each line we've received here as an example of a way we can +//! // do something with the data. This can be changed to map the data to +//! // something else, or to consume it differently. +//! .inspect(|line| println!("Line: {}", line)) +//! } +//! +//! fn main() { +//! // Lazily invoke any code so it can run directly within the tokio runtime +//! tokio::run(futures::lazy(|| { +//! let mut cmd = Command::new("cat"); +//! +//! // Specify that we want the command's standard output piped back to us. +//! // By default, standard input/output/error will be inherited from the +//! // current process (for example, this means that standard input will +//! // come from the keyboard and standard output/error will go directly to +//! // the terminal if this process is invoked from the command line). +//! cmd.stdout(Stdio::piped()); +//! +//! let mut child = cmd.spawn_async() +//! .expect("failed to spawn command"); +//! +//! let lines = lines_stream(&mut child); +//! +//! // Spawning into the tokio runtime requires that the future's Item and +//! // Error are both `()`. This is because tokio doesn't know what to do +//! // with any results or errors, so it requires that we've handled them! +//! // +//! // We can replace these sample usages of the child's exit status (or +//! // an encountered error) perform some different actions if needed! +//! // For example, log the error, or send a message on a channel, etc. +//! let child_future = child +//! .map(|status| println!("child status was: {}", status)) +//! .map_err(|e| panic!("error while running child: {}", e)); +//! +//! // Ensure the child process can live on within the runtime, otherwise +//! // the process will get killed if this handle is dropped +//! tokio::spawn(child_future); +//! +//! // Return a future to tokio. This is the same as calling using +//! // `tokio::spawn` above, but without having to return a dummy future +//! // here. +//! lines +//! // Convert the stream of values into a future which will resolve +//! // once the entire stream has been consumed. In this example we +//! // don't need to do anything with the data within the `for_each` +//! // call, but you can extend this to do something else (keep in mind +//! // that the stream will not produce items until the future returned +//! // from the closure resolves). +//! .for_each(|_| Ok(())) +//! // Similarly we "handle" any errors that arise, as required by tokio. +//! .map_err(|e| panic!("error while processing lines: {}", e)) +//! })); +//! } +//! ``` +//! +//! # 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. + +#![warn(missing_debug_implementations)] +#![deny(missing_docs)] +#![doc(html_root_url = "https://docs.rs/tokio-process/0.2")] + +extern crate futures; +extern crate tokio_io; +extern crate tokio_reactor; + +#[cfg(unix)] +#[macro_use] +extern crate lazy_static; +#[cfg(unix)] +#[macro_use] +extern crate log; + +use std::io::{self, Read, Write}; +use std::process::{Command, ExitStatus, Output, Stdio}; + +use crate::kill::Kill; +use futures::future::{ok, Either}; +use futures::{Async, Future, IntoFuture, Poll}; +use std::fmt; +use tokio_io::io::read_to_end; +use tokio_io::{AsyncRead, AsyncWrite, IoFuture}; +use tokio_reactor::Handle; + +#[path = "unix/mod.rs"] +#[cfg(unix)] +mod imp; + +#[path = "windows.rs"] +#[cfg(windows)] +mod imp; + +mod kill; + +/// 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. + /// + /// All I/O this child does will be associated with the current default + /// event loop. + fn spawn_async(&mut self) -> io::Result { + self.spawn_async_with_handle(&Handle::default()) + } + + /// 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_with_handle(&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 + /// any input/output handles are set to a pipe then they will be immediately + /// closed after the child is spawned. + /// + /// All I/O this child does will be associated with the current default + /// event loop. + /// + /// If the `StatusAsync` future is dropped before the future resolves, then + /// the child will be killed, if it was spawned. + /// + /// # Errors + /// + /// This function will return an error immediately if the child process + /// cannot be spawned. Otherwise errors obtained while waiting for the child + /// are returned through the `StatusAsync` future. + fn status_async(&mut self) -> io::Result { + self.status_async_with_handle(&Handle::default()) + } + + /// 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 + /// any input/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 `StatusAsync` future is dropped before the future resolves, then + /// the child will be killed, if it was spawned. + /// + /// # Errors + /// + /// This function will return an error immediately if the child process + /// cannot be spawned. Otherwise errors obtained while waiting for the child + /// are returned through the `StatusAsync` future. + fn status_async_with_handle(&mut self, handle: &Handle) -> io::Result; + + /// 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. + /// + /// All I/O this child does will be associated with the current default + /// 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) -> OutputAsync { + self.output_async_with_handle(&Handle::default()) + } + + /// 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_with_handle(&mut self, handle: &Handle) -> OutputAsync; +} + +struct SpawnedChild { + child: imp::Child, + stdin: Option, + stdout: Option, + stderr: Option, +} + +impl CommandExt for Command { + fn spawn_async_with_handle(&mut self, handle: &Handle) -> io::Result { + imp::spawn_child(self, handle).map(|spawned_child| Child { + child: ChildDropGuard::new(spawned_child.child), + stdin: spawned_child.stdin.map(|inner| ChildStdin { inner }), + stdout: spawned_child.stdout.map(|inner| ChildStdout { inner }), + stderr: spawned_child.stderr.map(|inner| ChildStderr { inner }), + }) + } + + fn status_async_with_handle(&mut self, handle: &Handle) -> io::Result { + self.spawn_async_with_handle(handle).map(|mut child| { + // Ensure we close any stdio handles so we can't deadlock + // waiting on the child which may be waiting to read/write + // to a pipe we're holding. + child.stdin.take(); + child.stdout.take(); + child.stderr.take(); + + StatusAsync { inner: child } + }) + } + + fn output_async_with_handle(&mut self, handle: &Handle) -> OutputAsync { + self.stdout(Stdio::piped()); + self.stderr(Stdio::piped()); + + let inner = self + .spawn_async_with_handle(handle) + .into_future() + .and_then(Child::wait_with_output); + + OutputAsync { + inner: Box::new(inner), + } + } +} + +/// 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)] +struct ChildDropGuard { + inner: T, + kill_on_drop: bool, +} + +impl ChildDropGuard { + fn new(inner: T) -> Self { + Self { + inner, + kill_on_drop: true, + } + } + + fn forget(&mut self) { + self.kill_on_drop = false; + } +} + +impl Kill for ChildDropGuard { + fn kill(&mut self) -> io::Result<()> { + let ret = self.inner.kill(); + + if ret.is_ok() { + self.kill_on_drop = false; + } + + ret + } +} + +impl Drop for ChildDropGuard { + fn drop(&mut self) { + if self.kill_on_drop { + drop(self.kill()); + } + } +} + +impl Future for ChildDropGuard { + type Item = T::Item; + type Error = T::Error; + + fn poll(&mut self) -> Poll { + let ret = self.inner.poll(); + + if let Ok(Async::Ready(_)) = ret { + // Avoid the overhead of trying to kill a reaped process + self.kill_on_drop = false; + } + + ret + } +} + +/// Representation of a child process spawned onto an event loop. +/// +/// 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**: 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. +#[must_use = "futures do nothing unless polled"] +#[derive(Debug)] +pub struct Child { + child: ChildDropGuard, + stdin: Option, + stdout: Option, + stderr: Option, +} + +impl Child { + /// Returns the OS-assigned process identifier associated with this child. + pub fn id(&self) -> u32 { + self.child.inner.id() + } + + /// Forces the child to exit. + /// + /// This is equivalent to sending a SIGKILL on unix platforms. + pub fn kill(&mut self) -> io::Result<()> { + 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: Box::new( + self.join3(stdout, stderr) + .map(|(status, stdout, stderr)| Output { + status, + stdout, + stderr, + }), + ), + } + } + + /// 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. + /// + /// > **Note**: this method may leak OS resources depending on your platform. + /// > To ensure resources are eventually cleaned up, consider sending the + /// > `Child` instance into an event loop as an alternative to this method. + /// + /// ```no_run + /// # extern crate futures; + /// # extern crate tokio; + /// # extern crate tokio_process; + /// # + /// # use std::process::Command; + /// # + /// # use futures::Future; + /// # use tokio_process::CommandExt; + /// # + /// # fn main() { + /// let child = Command::new("echo").arg("hello").arg("world") + /// .spawn_async() + /// .expect("failed to spawn"); + /// + /// let do_cleanup = child.map(|_| ()) // Ignore result + /// .map_err(|_| ()); // Ignore errors + /// + /// tokio::spawn(do_cleanup); + /// # } + /// ``` + pub fn forget(mut self) { + self.child.forget(); + } +} + +impl Future for Child { + type Item = ExitStatus; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.child.poll() + } +} + +/// 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. +#[must_use = "futures do nothing unless polled"] +pub struct WaitWithOutput { + inner: IoFuture, +} + +impl fmt::Debug for WaitWithOutput { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("WaitWithOutput") + .field("inner", &"..") + .finish() + } +} + +impl Future for WaitWithOutput { + type Item = Output; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.inner.poll() + } +} + +#[doc(hidden)] +#[deprecated(note = "renamed to `StatusAsync`", since = "0.2.1")] +pub type StatusAsync2 = StatusAsync; + +/// 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. +#[must_use = "futures do nothing unless polled"] +#[derive(Debug)] +pub struct StatusAsync { + inner: Child, +} + +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. +#[must_use = "futures do nothing unless polled"] +pub struct OutputAsync { + inner: IoFuture, +} + +impl fmt::Debug for OutputAsync { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("OutputAsync") + .field("inner", &"..") + .finish() + } +} + +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. +#[derive(Debug)] +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. +#[derive(Debug)] +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. +#[derive(Debug)] +pub struct ChildStderr { + inner: imp::ChildStderr, +} + +impl Write for ChildStdin { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.inner.write(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + +impl AsyncWrite for ChildStdin { + fn shutdown(&mut self) -> Poll<(), io::Error> { + self.inner.shutdown() + } +} + +impl Read for ChildStdout { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.inner.read(bytes) + } +} + +impl AsyncRead for ChildStdout {} + +impl Read for ChildStderr { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.inner.read(bytes) + } +} + +impl AsyncRead for ChildStderr {} + +#[cfg(unix)] +mod sys { + use super::{ChildStderr, ChildStdin, ChildStdout}; + use std::os::unix::io::{AsRawFd, RawFd}; + + impl AsRawFd for ChildStdin { + fn as_raw_fd(&self) -> RawFd { + self.inner.get_ref().as_raw_fd() + } + } + + impl AsRawFd for ChildStdout { + fn as_raw_fd(&self) -> RawFd { + self.inner.get_ref().as_raw_fd() + } + } + + impl AsRawFd for ChildStderr { + fn as_raw_fd(&self) -> RawFd { + self.inner.get_ref().as_raw_fd() + } + } +} + +#[cfg(windows)] +mod sys { + use super::{ChildStderr, ChildStdin, ChildStdout}; + use std::os::windows::io::{AsRawHandle, RawHandle}; + + impl AsRawHandle for ChildStdin { + fn as_raw_handle(&self) -> RawHandle { + self.inner.get_ref().as_raw_handle() + } + } + + impl AsRawHandle for ChildStdout { + fn as_raw_handle(&self) -> RawHandle { + self.inner.get_ref().as_raw_handle() + } + } + + impl AsRawHandle for ChildStderr { + fn as_raw_handle(&self) -> RawHandle { + self.inner.get_ref().as_raw_handle() + } + } +} + +#[cfg(test)] +mod test { + use super::ChildDropGuard; + use crate::kill::Kill; + use futures::{Async, Future, Poll}; + use std::io; + + struct Mock { + num_kills: usize, + num_polls: usize, + poll_result: Poll<(), ()>, + } + + impl Mock { + fn new() -> Self { + Self::with_result(Ok(Async::NotReady)) + } + + fn with_result(result: Poll<(), ()>) -> Self { + Self { + num_kills: 0, + num_polls: 0, + poll_result: result, + } + } + } + + impl Kill for Mock { + fn kill(&mut self) -> io::Result<()> { + self.num_kills += 1; + Ok(()) + } + } + + impl Future for Mock { + type Item = (); + type Error = (); + + fn poll(&mut self) -> Poll { + self.num_polls += 1; + self.poll_result + } + } + + #[test] + fn kills_on_drop() { + let mut mock = Mock::new(); + + { + let guard = ChildDropGuard::new(&mut mock); + drop(guard); + } + + assert_eq!(1, mock.num_kills); + assert_eq!(0, mock.num_polls); + } + + #[test] + fn no_kill_if_already_killed() { + let mut mock = Mock::new(); + + { + let mut guard = ChildDropGuard::new(&mut mock); + let _ = guard.kill(); + drop(guard); + } + + assert_eq!(1, mock.num_kills); + assert_eq!(0, mock.num_polls); + } + + #[test] + fn no_kill_if_reaped() { + let mut mock_pending = Mock::with_result(Ok(Async::NotReady)); + let mut mock_reaped = Mock::with_result(Ok(Async::Ready(()))); + let mut mock_err = Mock::with_result(Err(())); + + { + let mut guard = ChildDropGuard::new(&mut mock_pending); + let _ = guard.poll(); + + let mut guard = ChildDropGuard::new(&mut mock_reaped); + let _ = guard.poll(); + + let mut guard = ChildDropGuard::new(&mut mock_err); + let _ = guard.poll(); + } + + assert_eq!(1, mock_pending.num_kills); + assert_eq!(1, mock_pending.num_polls); + + assert_eq!(0, mock_reaped.num_kills); + assert_eq!(1, mock_reaped.num_polls); + + assert_eq!(1, mock_err.num_kills); + assert_eq!(1, mock_err.num_polls); + } + + #[test] + fn no_kill_on_forget() { + let mut mock = Mock::new(); + + { + let mut guard = ChildDropGuard::new(&mut mock); + guard.forget(); + drop(guard); + } + + assert_eq!(0, mock.num_kills); + assert_eq!(0, mock.num_polls); + } +} diff --git a/tokio-process/src/unix/mod.rs b/tokio-process/src/unix/mod.rs new file mode 100644 index 000000000..adf0f4ba2 --- /dev/null +++ b/tokio-process/src/unix/mod.rs @@ -0,0 +1,221 @@ +//! 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 mio; +extern crate tokio_signal; + +mod orphan; +mod reap; + +use self::mio::event::Evented; +use self::mio::unix::{EventedFd, UnixReady}; +use self::mio::{Poll as MioPoll, PollOpt, Ready, Token}; +use self::orphan::{AtomicOrphanQueue, OrphanQueue, Wait}; +use self::reap::Reaper; +use self::tokio_signal::unix::Signal; +use super::SpawnedChild; +use crate::kill::Kill; +use futures::future::FlattenStream; +use futures::{Future, Poll}; +use std::fmt; +use std::io; +use std::os::unix::io::{AsRawFd, RawFd}; +use std::process::{self, ExitStatus}; +use tokio_io::IoFuture; +use tokio_reactor::{Handle, PollEvented}; + +impl Wait for process::Child { + fn id(&self) -> u32 { + self.id() + } + + fn try_wait(&mut self) -> io::Result> { + self.try_wait() + } +} + +impl Kill for process::Child { + fn kill(&mut self) -> io::Result<()> { + self.kill() + } +} + +lazy_static! { + static ref ORPHAN_QUEUE: AtomicOrphanQueue = AtomicOrphanQueue::new(); +} + +struct GlobalOrphanQueue; + +impl fmt::Debug for GlobalOrphanQueue { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + ORPHAN_QUEUE.fmt(fmt) + } +} + +impl OrphanQueue for GlobalOrphanQueue { + fn push_orphan(&self, orphan: process::Child) { + ORPHAN_QUEUE.push_orphan(orphan) + } + + fn reap_orphans(&self) { + ORPHAN_QUEUE.reap_orphans() + } +} + +#[must_use = "futures do nothing unless polled"] +pub struct Child { + inner: Reaper>>, +} + +impl fmt::Debug for Child { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Child") + .field("pid", &self.inner.id()) + .finish() + } +} + +pub(crate) fn spawn_child(cmd: &mut process::Command, handle: &Handle) -> io::Result { + 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 signal = Signal::with_handle(libc::SIGCHLD, handle).flatten_stream(); + Ok(SpawnedChild { + child: Child { + inner: Reaper::new(child, GlobalOrphanQueue, signal), + }, + stdin, + stdout, + stderr, + }) +} + +impl Child { + pub fn id(&self) -> u32 { + self.inner.id() + } +} + +impl Kill for Child { + fn kill(&mut self) -> io::Result<()> { + self.inner.kill() + } +} + +impl Future for Child { + type Item = ExitStatus; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.inner.poll() + } +} + +#[derive(Debug)] +pub struct Fd(T); + +impl io::Read for Fd { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.0.read(bytes) + } +} + +impl io::Write for Fd { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.write(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } +} + +impl AsRawFd for Fd +where + T: AsRawFd, +{ + fn as_raw_fd(&self) -> RawFd { + self.0.as_raw_fd() + } +} + +pub type ChildStdin = PollEvented>; +pub type ChildStdout = PollEvented>; +pub type ChildStderr = PollEvented>; + +impl Evented for Fd +where + T: AsRawFd, +{ + fn register( + &self, + poll: &MioPoll, + token: Token, + interest: Ready, + opts: PollOpt, + ) -> io::Result<()> { + EventedFd(&self.as_raw_fd()).register(poll, token, interest | UnixReady::hup(), opts) + } + + fn reregister( + &self, + poll: &MioPoll, + token: Token, + interest: Ready, + opts: PollOpt, + ) -> io::Result<()> { + EventedFd(&self.as_raw_fd()).reregister(poll, token, interest | UnixReady::hup(), opts) + } + + fn deregister(&self, poll: &MioPoll) -> io::Result<()> { + EventedFd(&self.as_raw_fd()).deregister(poll) + } +} + +fn stdio(option: Option, handle: &Handle) -> io::Result>>> +where + T: AsRawFd, +{ + let io = match option { + Some(io) => io, + None => return Ok(None), + }; + + // Set the fd to nonblocking before we pass it to the event loop + unsafe { + let fd = io.as_raw_fd(); + let r = libc::fcntl(fd, libc::F_GETFL); + if r == -1 { + return Err(io::Error::last_os_error()); + } + let r = libc::fcntl(fd, libc::F_SETFL, r | libc::O_NONBLOCK); + if r == -1 { + return Err(io::Error::last_os_error()); + } + } + let io = PollEvented::new_with_handle(Fd(io), handle)?; + Ok(Some(io)) +} diff --git a/tokio-process/src/unix/orphan.rs b/tokio-process/src/unix/orphan.rs new file mode 100644 index 000000000..6f456ce0e --- /dev/null +++ b/tokio-process/src/unix/orphan.rs @@ -0,0 +1,193 @@ +extern crate crossbeam_queue; + +use self::crossbeam_queue::SegQueue; +use std::io; +use std::process::ExitStatus; + +/// An interface for waiting on a process to exit. +pub(crate) trait Wait { + /// Get the identifier for this process or diagnostics. + fn id(&self) -> u32; + /// Try waiting for a process to exit in a non-blocking manner. + fn try_wait(&mut self) -> io::Result>; +} + +impl<'a, T: 'a + Wait> Wait for &'a mut T { + fn id(&self) -> u32 { + (**self).id() + } + + fn try_wait(&mut self) -> io::Result> { + (**self).try_wait() + } +} + +/// An interface for queueing up an orphaned process so that it can be reaped. +pub(crate) trait OrphanQueue { + /// Add an orphan to the queue. + fn push_orphan(&self, orphan: T); + /// Attempt to reap every process in the queue, ignoring any errors and + /// enqueueing any orphans which have not yet exited. + fn reap_orphans(&self); +} + +impl<'a, T, O: 'a + OrphanQueue> OrphanQueue for &'a O { + fn push_orphan(&self, orphan: T) { + (**self).push_orphan(orphan); + } + + fn reap_orphans(&self) { + (**self).reap_orphans() + } +} + +/// An atomic implementation of `OrphanQueue`. +#[derive(Debug)] +pub(crate) struct AtomicOrphanQueue { + queue: SegQueue, +} + +impl AtomicOrphanQueue { + pub(crate) fn new() -> Self { + Self { + queue: SegQueue::new(), + } + } +} + +impl OrphanQueue for AtomicOrphanQueue { + fn push_orphan(&self, orphan: T) { + self.queue.push(orphan) + } + + fn reap_orphans(&self) { + let len = self.queue.len(); + + if len == 0 { + return; + } + + let mut orphans = Vec::with_capacity(len); + while let Ok(mut orphan) = self.queue.pop() { + match orphan.try_wait() { + Ok(Some(_)) => {} + Err(e) => error!( + "leaking orphaned process {} due to try_wait() error: {}", + orphan.id(), + e, + ), + + // Still not done yet, we need to put it back in the queue + // when were done draining it, so that we don't get stuck + // in an infinite loop here + Ok(None) => orphans.push(orphan), + } + } + + for orphan in orphans { + self.queue.push(orphan); + } + } +} + +#[cfg(test)] +mod test { + use super::Wait; + use super::{AtomicOrphanQueue, OrphanQueue}; + use std::cell::Cell; + use std::io; + use std::os::unix::process::ExitStatusExt; + use std::process::ExitStatus; + use std::rc::Rc; + + struct MockWait { + total_waits: Rc>, + num_wait_until_status: usize, + return_err: bool, + } + + impl MockWait { + fn new(num_wait_until_status: usize) -> Self { + Self { + total_waits: Rc::new(Cell::new(0)), + num_wait_until_status, + return_err: false, + } + } + + fn with_err() -> Self { + Self { + total_waits: Rc::new(Cell::new(0)), + num_wait_until_status: 0, + return_err: true, + } + } + } + + impl Wait for MockWait { + fn id(&self) -> u32 { + 42 + } + + fn try_wait(&mut self) -> io::Result> { + let waits = self.total_waits.get(); + + let ret = if self.num_wait_until_status == waits { + if self.return_err { + Ok(Some(ExitStatus::from_raw(0))) + } else { + Err(io::Error::new(io::ErrorKind::Other, "mock err")) + } + } else { + Ok(None) + }; + + self.total_waits.set(waits + 1); + ret + } + } + + #[test] + fn drain_attempts_a_single_reap_of_all_queued_orphans() { + let first_orphan = MockWait::new(0); + let second_orphan = MockWait::new(1); + let third_orphan = MockWait::new(2); + let fourth_orphan = MockWait::with_err(); + + let first_waits = first_orphan.total_waits.clone(); + let second_waits = second_orphan.total_waits.clone(); + let third_waits = third_orphan.total_waits.clone(); + let fourth_waits = fourth_orphan.total_waits.clone(); + + let orphanage = AtomicOrphanQueue::new(); + orphanage.push_orphan(first_orphan); + orphanage.push_orphan(third_orphan); + orphanage.push_orphan(second_orphan); + orphanage.push_orphan(fourth_orphan); + + assert_eq!(orphanage.queue.len(), 4); + + orphanage.reap_orphans(); + assert_eq!(orphanage.queue.len(), 2); + assert_eq!(first_waits.get(), 1); + assert_eq!(second_waits.get(), 1); + assert_eq!(third_waits.get(), 1); + assert_eq!(fourth_waits.get(), 1); + + orphanage.reap_orphans(); + assert_eq!(orphanage.queue.len(), 1); + assert_eq!(first_waits.get(), 1); + assert_eq!(second_waits.get(), 2); + assert_eq!(third_waits.get(), 2); + assert_eq!(fourth_waits.get(), 1); + + orphanage.reap_orphans(); + assert_eq!(orphanage.queue.len(), 0); + assert_eq!(first_waits.get(), 1); + assert_eq!(second_waits.get(), 2); + assert_eq!(third_waits.get(), 3); + assert_eq!(fourth_waits.get(), 1); + + orphanage.reap_orphans(); // Safe to reap when empty + } +} diff --git a/tokio-process/src/unix/reap.rs b/tokio-process/src/unix/reap.rs new file mode 100644 index 000000000..125aaf30f --- /dev/null +++ b/tokio-process/src/unix/reap.rs @@ -0,0 +1,318 @@ +use super::orphan::{OrphanQueue, Wait}; +use crate::kill::Kill; +use futures::{Async, Future, Poll, Stream}; +use std::io; +use std::ops::Deref; +use std::process::ExitStatus; + +/// Orchestrates between registering interest for receiving signals when a +/// child process has exited, and attempting to poll for process completion. +#[derive(Debug)] +pub(crate) struct Reaper +where + W: Wait, + Q: OrphanQueue, +{ + inner: Option, + orphan_queue: Q, + signal: S, +} + +impl Deref for Reaper +where + W: Wait, + Q: OrphanQueue, +{ + type Target = W; + + fn deref(&self) -> &Self::Target { + self.inner() + } +} + +impl Reaper +where + W: Wait, + Q: OrphanQueue, +{ + pub(crate) fn new(inner: W, orphan_queue: Q, signal: S) -> Self { + Self { + inner: Some(inner), + orphan_queue, + signal, + } + } + + fn inner(&self) -> &W { + self.inner.as_ref().expect("inner has gone away") + } + + fn inner_mut(&mut self) -> &mut W { + self.inner.as_mut().expect("inner has gone away") + } +} + +impl Future for Reaper +where + W: Wait, + Q: OrphanQueue, + S: Stream, +{ + type Item = ExitStatus; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + loop { + // If the child hasn't exited yet, then it's our responsibility to + // ensure the current task gets notified when it might be able to + // make progress. + // + // As described in `spawn` above, we just indicate that we can + // next make progress once a SIGCHLD is received. + // + // However, we will register for a notification on the next signal + // BEFORE we poll the child. Otherwise it is possible that the child + // can exit and the signal can arrive after we last polled the child, + // but before we've registered for a notification on the next signal + // (this can cause a deadlock if there are no more spawned children + // which can generate a different signal for us). A side effect of + // pre-registering for signal notifications is that when the child + // exits, we will have already registered for an additional + // notification we don't need to consume. If another signal arrives, + // this future's task will be notified/woken up again. Since the + // futures model allows for spurious wake ups this extra wakeup + // should not cause significant issues with parent futures. + let registered_interest = self.signal.poll()?.is_not_ready(); + + self.orphan_queue.reap_orphans(); + if let Some(status) = self.inner_mut().try_wait()? { + return Ok(Async::Ready(status)); + } + + // If our attempt to poll for the next signal was not ready, then + // we've arranged for our task to get notified and we can bail out. + if registered_interest { + return Ok(Async::NotReady); + } else { + // Otherwise, if the signal stream delivered a signal to us, we + // won't get notified at the next signal, so we'll loop and try + // again. + continue; + } + } + } +} + +impl Kill for Reaper +where + W: Kill + Wait, + Q: OrphanQueue, +{ + fn kill(&mut self) -> io::Result<()> { + self.inner_mut().kill() + } +} + +impl Drop for Reaper +where + W: Wait, + Q: OrphanQueue, +{ + fn drop(&mut self) { + if let Ok(Some(_)) = self.inner_mut().try_wait() { + return; + } + + let orphan = self.inner.take().unwrap(); + self.orphan_queue.push_orphan(orphan); + } +} + +#[cfg(test)] +mod test { + use super::*; + use futures::{Async, Poll, Stream}; + use std::cell::{Cell, RefCell}; + use std::os::unix::process::ExitStatusExt; + use std::process::ExitStatus; + + #[derive(Debug)] + struct MockWait { + total_kills: usize, + total_waits: usize, + num_wait_until_status: usize, + status: ExitStatus, + } + + impl MockWait { + fn new(status: ExitStatus, num_wait_until_status: usize) -> Self { + Self { + total_kills: 0, + total_waits: 0, + num_wait_until_status, + status, + } + } + } + + impl Wait for MockWait { + fn id(&self) -> u32 { + 0 + } + + fn try_wait(&mut self) -> io::Result> { + let ret = if self.num_wait_until_status == self.total_waits { + Some(self.status) + } else { + None + }; + + self.total_waits += 1; + Ok(ret) + } + } + + impl Kill for MockWait { + fn kill(&mut self) -> io::Result<()> { + self.total_kills += 1; + Ok(()) + } + } + + struct MockStream { + total_polls: usize, + values: Vec>, + } + + impl MockStream { + fn new(values: Vec>) -> Self { + Self { + total_polls: 0, + values, + } + } + } + + impl Stream for MockStream { + type Item = (); + type Error = io::Error; + + fn poll(&mut self) -> Poll, Self::Error> { + self.total_polls += 1; + match self.values.remove(0) { + Some(()) => Ok(Async::Ready(Some(()))), + None => Ok(Async::NotReady), + } + } + } + + struct MockQueue { + all_enqueued: RefCell>, + total_reaps: Cell, + } + + impl MockQueue { + fn new() -> Self { + Self { + all_enqueued: RefCell::new(Vec::new()), + total_reaps: Cell::new(0), + } + } + } + + impl OrphanQueue for MockQueue { + fn push_orphan(&self, orphan: W) { + self.all_enqueued.borrow_mut().push(orphan); + } + + fn reap_orphans(&self) { + self.total_reaps.set(self.total_reaps.get() + 1); + } + } + + #[test] + fn reaper() { + let exit = ExitStatus::from_raw(0); + let mock = MockWait::new(exit, 3); + let mut grim = Reaper::new( + mock, + MockQueue::new(), + MockStream::new(vec![None, Some(()), None, None, None]), + ); + + // Not yet exited, interest registered + assert_eq!(Async::NotReady, grim.poll().expect("failed to wait")); + assert_eq!(1, grim.signal.total_polls); + assert_eq!(1, grim.total_waits); + assert_eq!(1, grim.orphan_queue.total_reaps.get()); + assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); + + // Not yet exited, couldn't register interest the first time + // but managed to register interest the second time around + assert_eq!(Async::NotReady, grim.poll().expect("failed to wait")); + assert_eq!(3, grim.signal.total_polls); + assert_eq!(3, grim.total_waits); + assert_eq!(3, grim.orphan_queue.total_reaps.get()); + assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); + + // Exited + assert_eq!(Async::Ready(exit), grim.poll().expect("failed to wait")); + assert_eq!(4, grim.signal.total_polls); + assert_eq!(4, grim.total_waits); + assert_eq!(4, grim.orphan_queue.total_reaps.get()); + assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); + } + + #[test] + fn kill() { + let exit = ExitStatus::from_raw(0); + let mut grim = Reaper::new( + MockWait::new(exit, 0), + MockQueue::new(), + MockStream::new(vec![None]), + ); + + grim.kill().unwrap(); + assert_eq!(1, grim.total_kills); + assert_eq!(0, grim.orphan_queue.total_reaps.get()); + assert!(grim.orphan_queue.all_enqueued.borrow().is_empty()); + } + + #[test] + fn drop_reaps_if_possible() { + let exit = ExitStatus::from_raw(0); + let mut mock = MockWait::new(exit, 0); + + { + let queue = MockQueue::new(); + + let grim = Reaper::new(&mut mock, &queue, MockStream::new(vec![])); + + drop(grim); + + assert_eq!(0, queue.total_reaps.get()); + assert!(queue.all_enqueued.borrow().is_empty()); + } + + assert_eq!(1, mock.total_waits); + assert_eq!(0, mock.total_kills); + } + + #[test] + fn drop_enqueues_orphan_if_wait_fails() { + let exit = ExitStatus::from_raw(0); + let mut mock = MockWait::new(exit, 2); + + { + let queue = MockQueue::<&mut MockWait>::new(); + let grim = Reaper::new(&mut mock, &queue, MockStream::new(vec![])); + drop(grim); + + assert_eq!(0, queue.total_reaps.get()); + assert_eq!(1, queue.all_enqueued.borrow().len()); + } + + assert_eq!(1, mock.total_waits); + assert_eq!(0, mock.total_kills); + } +} diff --git a/tokio-process/src/windows.rs b/tokio-process/src/windows.rs new file mode 100644 index 000000000..0d43f29fa --- /dev/null +++ b/tokio-process/src/windows.rs @@ -0,0 +1,192 @@ +//! 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 mio_named_pipes; +extern crate winapi; + +use std::fmt; +use std::io; +use std::os::windows::prelude::*; +use std::os::windows::process::ExitStatusExt; +use std::process::{self, ExitStatus}; +use std::ptr; + +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 futures::future::Fuse; +use futures::sync::oneshot; +use futures::{Async, Future, Poll}; +use kill::Kill; +use tokio_reactor::{Handle, PollEvented}; + +#[must_use = "futures do nothing unless polled"] +pub struct Child { + child: process::Child, + waiting: Option, +} + +impl fmt::Debug for Child { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Child") + .field("pid", &self.id()) + .field("child", &self.child) + .field("waiting", &"..") + .finish() + } +} + +struct Waiting { + rx: Fuse>, + wait_object: HANDLE, + tx: *mut Option>, +} + +unsafe impl Sync for Waiting {} +unsafe impl Send for Waiting {} + +pub(crate) fn spawn_child(cmd: &mut process::Command, handle: &Handle) -> io::Result { + 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)?; + + Ok(SpawnedChild { + child: Child { + child, + waiting: None, + }, + stdin, + stdout, + stderr, + }) +} + +impl Child { + pub fn id(&self) -> u32 { + self.child.id() + } +} + +impl Kill for Child { + 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 { + 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), + } + let status = try_wait(&self.child)?.expect("not ready yet"); + return Ok(status.into()); + } + + if let Some(e) = try_wait(&self.child)? { + return Ok(e.into()); + } + let (tx, rx) = oneshot::channel(); + 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, + ) + }; + if rc == 0 { + let err = io::Error::last_os_error(); + drop(unsafe { Box::from_raw(ptr) }); + return Err(err); + } + self.waiting = Some(Waiting { + rx: rx.fuse(), + wait_object, + tx: ptr, + }); + } + } +} + +impl Drop for Waiting { + fn drop(&mut self) { + unsafe { + let rc = UnregisterWaitEx(self.wait_object, INVALID_HANDLE_VALUE); + if rc == 0 { + panic!("failed to unregister: {}", io::Error::last_os_error()); + } + drop(Box::from_raw(self.tx)); + } + } +} + +unsafe extern "system" fn callback(ptr: PVOID, _timer_fired: BOOLEAN) { + let complete = &mut *(ptr as *mut Option>); + let _ = complete.take().unwrap().send(()); +} + +pub fn try_wait(child: &process::Child) -> io::Result> { + unsafe { + match WaitForSingleObject(child.as_raw_handle(), 0) { + WAIT_OBJECT_0 => {} + WAIT_TIMEOUT => return Ok(None), + _ => return Err(io::Error::last_os_error()), + } + let mut status = 0; + let rc = GetExitCodeProcess(child.as_raw_handle(), &mut status); + if rc == FALSE { + Err(io::Error::last_os_error()) + } else { + Ok(Some(ExitStatus::from_raw(status))) + } + } +} + +pub type ChildStdin = PollEvented; +pub type ChildStdout = PollEvented; +pub type ChildStderr = PollEvented; + +fn stdio(option: Option, handle: &Handle) -> io::Result>> +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 = PollEvented::new_with_handle(pipe, handle)?; + Ok(Some(io)) +} diff --git a/tokio-process/tests/issue_42.rs b/tokio-process/tests/issue_42.rs new file mode 100644 index 000000000..0b34efcf1 --- /dev/null +++ b/tokio-process/tests/issue_42.rs @@ -0,0 +1,51 @@ +#![cfg(unix)] + +extern crate futures; +extern crate tokio_process; + +use futures::{stream, Future, IntoFuture, Stream}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::thread; +use std::time::Duration; +use tokio_process::CommandExt; + +fn run_test() { + let finished = Arc::new(AtomicBool::new(false)); + let finished_clone = finished.clone(); + + thread::spawn(move || { + let _ = stream::iter_ok(0..2) + .map(|i| { + Command::new("echo") + .arg(format!("I am spawned process #{}", i)) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn_async() + .into_future() + .flatten() + }) + .buffered(2) + .collect() + .wait(); + + finished_clone.store(true, Ordering::SeqCst); + }); + + thread::sleep(Duration::from_millis(100)); + assert!( + finished.load(Ordering::SeqCst), + "FINISHED flag not set, maybe we deadlocked?" + ); +} + +#[test] +fn issue_42() { + let max = 10; + for i in 0..max { + println!("running {}/{}", i, max); + run_test() + } +} diff --git a/tokio-process/tests/smoke.rs b/tokio-process/tests/smoke.rs new file mode 100644 index 000000000..9d11002ad --- /dev/null +++ b/tokio-process/tests/smoke.rs @@ -0,0 +1,22 @@ +extern crate tokio_process; + +use tokio_process::CommandExt; + +mod support; + +#[test] +fn simple() { + let mut cmd = support::cmd("exit"); + cmd.arg("2"); + + let mut child = cmd.spawn_async().unwrap(); + + let id = child.id(); + assert!(id > 0); + + let status = support::run_with_timeout(&mut child).expect("failed to run future"); + assert_eq!(status.code(), Some(2)); + + assert_eq!(child.id(), id); + drop(child.kill()); +} diff --git a/tokio-process/tests/stdio.rs b/tokio-process/tests/stdio.rs new file mode 100644 index 000000000..647350f72 --- /dev/null +++ b/tokio-process/tests/stdio.rs @@ -0,0 +1,113 @@ +extern crate futures; +#[macro_use] +extern crate log; +extern crate tokio_io; +extern crate tokio_process; + +use std::io; +use std::process::{Command, ExitStatus, Stdio}; + +use futures::future::Future; +use futures::stream::{self, Stream}; +use tokio_io::io::{read_until, write_all}; +use tokio_process::{Child, CommandExt}; + +mod support; + +fn cat() -> Command { + let mut cmd = support::cmd("cat"); + cmd.stdin(Stdio::piped()).stdout(Stdio::piped()); + cmd +} + +fn feed_cat(mut cat: Child, n: usize) -> Box> { + let stdin = cat.stdin().take().unwrap(); + let stdout = cat.stdout().take().unwrap(); + + debug!("starting to feed"); + // Produce n lines on the child's stdout. + let numbers = stream::iter_ok(0..n); + let write = numbers + .fold(stdin, |stdin, i| { + debug!("sending line {} to child", i); + write_all(stdin, format!("line {}\n", i).into_bytes()).map(|p| p.0) + }) + .map(|_| ()); + + // Try to read `n + 1` lines, ensuring the last one is empty + // (i.e. EOF is reached after `n` lines. + let reader = io::BufReader::new(stdout); + let expected_numbers = stream::iter_ok(0..=n); + let read = expected_numbers.fold((reader, 0), move |(reader, i), _| { + let done = i >= n; + debug!("starting read from child"); + read_until(reader, b'\n', Vec::new()).and_then(move |(reader, vec)| { + debug!( + "read line {} from child ({} bytes, done: {})", + i, + vec.len(), + done + ); + match (done, vec.len()) { + (false, 0) => Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")), + (true, n) if n != 0 => Err(io::Error::new(io::ErrorKind::Other, "extraneous data")), + _ => { + let s = std::str::from_utf8(&vec).unwrap(); + let expected = format!("line {}\n", i); + if done || s == expected { + Ok((reader, i + 1)) + } else { + Err(io::Error::new(io::ErrorKind::Other, "unexpected data")) + } + } + } + }) + }); + + // Compose reading and writing concurrently. + Box::new(write.join(read).and_then(|_| cat)) +} + +/// Check for the following properties when feeding stdin and +/// consuming stdout of a cat-like process: +/// +/// - A number of lines that amounts to a number of bytes exceeding a +/// typical OS buffer size can be fed to the child without +/// deadlock. This tests that we also consume the stdout +/// 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. +#[test] +fn feed_a_lot() { + let child = cat().spawn_async().unwrap(); + let status = support::run_with_timeout(feed_cat(child, 10000)).unwrap(); + assert_eq!(status.code(), Some(0)); +} + +#[test] +fn wait_with_output_captures() { + let mut child = cat().spawn_async().unwrap(); + let stdin = child.stdin().take().unwrap(); + let out = child.wait_with_output(); + + let future = write_all(stdin, b"1234").map(|p| p.1).join(out); + let ret = support::run_with_timeout(future).unwrap(); + let (written, output) = ret; + + assert!(output.status.success()); + assert_eq!(output.stdout, written); + assert_eq!(output.stderr.len(), 0); +} + +#[test] +fn status_closes_any_pipes() { + // Cat will open a pipe between the parent and child. + // If `status_async` doesn't ensure the handles are closed, + // we would end up blocking forever (and time out). + let child = cat().status_async().expect("failed to spawn child"); + + support::run_with_timeout(child) + .expect("time out exceeded! did we get stuck waiting on the child?"); +} diff --git a/tokio-process/tests/support/mod.rs b/tokio-process/tests/support/mod.rs new file mode 100644 index 000000000..0e9ce6845 --- /dev/null +++ b/tokio-process/tests/support/mod.rs @@ -0,0 +1,43 @@ +extern crate futures; +extern crate tokio; + +use self::futures::Future; +use self::tokio::timer::Timeout; +use std::env; +use std::process::Command; +use std::time::Duration; + +pub use self::tokio::runtime::current_thread::Runtime as CurrentThreadRuntime; + +pub fn cmd(s: &str) -> Command { + let mut me = env::current_exe().unwrap(); + me.pop(); + if me.ends_with("deps") { + me.pop(); + } + me.push(s); + Command::new(me) +} + +pub fn with_timeout(future: F) -> impl Future { + Timeout::new(future, Duration::from_secs(3)).map_err(|e| { + if e.is_timer() { + panic!("failed to register timer"); + } else if e.is_elapsed() { + panic!("timed out") + } else { + e.into_inner().expect("missing inner error") + } + }) +} + +pub fn run_with_timeout(future: F) -> Result +where + F: Future, +{ + // NB: Timeout requires a timer registration which is provided by + // tokio's `current_thread::Runtime`, but isn't available by just using + // tokio's default CurrentThread executor which powers `current_thread::block_on_all`. + let mut rt = CurrentThreadRuntime::new().expect("failed to get runtime"); + rt.block_on(with_timeout(future)) +}