tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-18 00:00:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge tokio-process into tokio

Browse Source 
Original repo can be found at https://github.com/alexcrichton/tokio-process/
This commit is contained in:
Ivan Petkov
2019-07-21 11:08:16 -07:00
parent d9688bc094 c6defbce4b
commit a3b8d82711
16 changed files with 2259 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-process/CHANGELOG.md
+71
View File
@@ -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!
tokio-process/Cargo.toml
+55
View File
@@ -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 <[email protected]>"]
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"
tokio-process/LICENSE
+25
View File
@@ -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.
tokio-process/README.md
+48
View File
@@ -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.
tokio-process/src/bin/cat.rs
+19
View File
@@ -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();
}
tokio-process/src/bin/exit.rs
+5
View File
@@ -0,0 +1,5 @@
#[allow(dead_code)]
fn main() {
std::process::exit(std::env::args().nth(1).unwrap().parse().unwrap());
}
tokio-process/src/kill.rs
+13
View File
@@ -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()
}
}
tokio-process/src/lib.rs
+870
View File
@@ -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<Item = String, Error = Error> + 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<Child> {
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<Child>;
/// 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<StatusAsync> {
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<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<u8>` 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<u8>` 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<imp::ChildStdin>,
stdout: Option<imp::ChildStdout>,
stderr: Option<imp::ChildStderr>,
}
impl CommandExt for Command {
fn spawn_async_with_handle(&mut self, handle: &Handle) -> io::Result<Child> {
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<StatusAsync> {
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<T: Kill> {
inner: T,
kill_on_drop: bool,
}
impl<T: Kill> ChildDropGuard<T> {
fn new(inner: T) -> Self {
Self {
inner,
kill_on_drop: true,
}
}
fn forget(&mut self) {
self.kill_on_drop = false;
}
}
impl<T: Kill> Kill for ChildDropGuard<T> {
fn kill(&mut self) -> io::Result<()> {
let ret = self.inner.kill();
if ret.is_ok() {
self.kill_on_drop = false;
}
ret
}
}
impl<T: Kill> Drop for ChildDropGuard<T> {
fn drop(&mut self) {
if self.kill_on_drop {
drop(self.kill());
}
}
}
impl<T: Future + Kill> Future for ChildDropGuard<T> {
type Item = T::Item;
type Error = T::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
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<imp::Child>,
stdin: Option<ChildStdin>,
stdout: Option<ChildStdout>,
stderr: Option<ChildStderr>,
}
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<ChildStdin> {
&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<ChildStdout> {
&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<ChildStderr> {
&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<ExitStatus, io::Error> {
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<Output>,
}
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<Output, io::Error> {
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<ExitStatus, io::Error> {
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<Output>,
}
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<Output, io::Error> {
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<usize> {
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<usize> {
self.inner.read(bytes)
}
}
impl AsyncRead for ChildStdout {}
impl Read for ChildStderr {
fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
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::Item, Self::Error> {
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);
}
}
tokio-process/src/unix/mod.rs
+221
View File
@@ -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<Option<ExitStatus>> {
self.try_wait()
}
}
impl Kill for process::Child {
fn kill(&mut self) -> io::Result<()> {
self.kill()
}
}
lazy_static! {
static ref ORPHAN_QUEUE: AtomicOrphanQueue<process::Child> = AtomicOrphanQueue::new();
}
struct GlobalOrphanQueue;
impl fmt::Debug for GlobalOrphanQueue {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
ORPHAN_QUEUE.fmt(fmt)
}
}
impl OrphanQueue<process::Child> 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<process::Child, GlobalOrphanQueue, FlattenStream<IoFuture<Signal>>>,
}
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<SpawnedChild> {
let mut child = cmd.spawn()?;
let stdin = stdio(child.stdin.take(), handle)?;
let stdout = stdio(child.stdout.take(), handle)?;
let stderr = stdio(child.stderr.take(), handle)?;
let 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::Item, Self::Error> {
self.inner.poll()
}
}
#[derive(Debug)]
pub struct Fd<T>(T);
impl<T: io::Read> io::Read for Fd<T> {
fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
self.0.read(bytes)
}
}
impl<T: io::Write> io::Write for Fd<T> {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.0.write(bytes)
}
fn flush(&mut self) -> io::Result<()> {
self.0.flush()
}
}
impl<T> AsRawFd for Fd<T>
where
T: AsRawFd,
{
fn as_raw_fd(&self) -> RawFd {
self.0.as_raw_fd()
}
}
pub type ChildStdin = PollEvented<Fd<process::ChildStdin>>;
pub type ChildStdout = PollEvented<Fd<process::ChildStdout>>;
pub type ChildStderr = PollEvented<Fd<process::ChildStderr>>;
impl<T> Evented for Fd<T>
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<T>(option: Option<T>, handle: &Handle) -> io::Result<Option<PollEvented<Fd<T>>>>
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))
}
tokio-process/src/unix/orphan.rs
+193
View File
@@ -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<Option<ExitStatus>>;
}
impl<'a, T: 'a + Wait> Wait for &'a mut T {
fn id(&self) -> u32 {
(**self).id()
}
fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
(**self).try_wait()
}
}
/// An interface for queueing up an orphaned process so that it can be reaped.
pub(crate) trait OrphanQueue<T> {
/// 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<T>> OrphanQueue<T> 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<T> {
queue: SegQueue<T>,
}
impl<T> AtomicOrphanQueue<T> {
pub(crate) fn new() -> Self {
Self {
queue: SegQueue::new(),
}
}
}
impl<T: Wait> OrphanQueue<T> for AtomicOrphanQueue<T> {
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<Cell<usize>>,
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<Option<ExitStatus>> {
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
}
}
tokio-process/src/unix/reap.rs
+318
View File
@@ -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<W, Q, S>
where
W: Wait,
Q: OrphanQueue<W>,
{
inner: Option<W>,
orphan_queue: Q,
signal: S,
}
impl<W, Q, S> Deref for Reaper<W, Q, S>
where
W: Wait,
Q: OrphanQueue<W>,
{
type Target = W;
fn deref(&self) -> &Self::Target {
self.inner()
}
}
impl<W, Q, S> Reaper<W, Q, S>
where
W: Wait,
Q: OrphanQueue<W>,
{
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<W, Q, S> Future for Reaper<W, Q, S>
where
W: Wait,
Q: OrphanQueue<W>,
S: Stream<Error = io::Error>,
{
type Item = ExitStatus;
type Error = io::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
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<W, Q, S> Kill for Reaper<W, Q, S>
where
W: Kill + Wait,
Q: OrphanQueue<W>,
{
fn kill(&mut self) -> io::Result<()> {
self.inner_mut().kill()
}
}
impl<W, Q, S> Drop for Reaper<W, Q, S>
where
W: Wait,
Q: OrphanQueue<W>,
{
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<Option<ExitStatus>> {
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<Option<()>>,
}
impl MockStream {
fn new(values: Vec<Option<()>>) -> Self {
Self {
total_polls: 0,
values,
}
}
}
impl Stream for MockStream {
type Item = ();
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
self.total_polls += 1;
match self.values.remove(0) {
Some(()) => Ok(Async::Ready(Some(()))),
None => Ok(Async::NotReady),
}
}
}
struct MockQueue<W> {
all_enqueued: RefCell<Vec<W>>,
total_reaps: Cell<usize>,
}
impl<W> MockQueue<W> {
fn new() -> Self {
Self {
all_enqueued: RefCell::new(Vec::new()),
total_reaps: Cell::new(0),
}
}
}
impl<W: Wait> OrphanQueue<W> for MockQueue<W> {
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);
}
}
tokio-process/src/windows.rs
+192
View File
@@ -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<Waiting>,
}
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<oneshot::Receiver<()>>,
wait_object: HANDLE,
tx: *mut Option<oneshot::Sender<()>>,
}
unsafe impl Sync for Waiting {}
unsafe impl Send for Waiting {}
pub(crate) fn spawn_child(cmd: &mut process::Command, handle: &Handle) -> io::Result<SpawnedChild> {
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<Self::Item, Self::Error> {
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<oneshot::Sender<()>>);
let _ = complete.take().unwrap().send(());
}
pub fn try_wait(child: &process::Child) -> io::Result<Option<ExitStatus>> {
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<NamedPipe>;
pub type ChildStdout = PollEvented<NamedPipe>;
pub type ChildStderr = PollEvented<NamedPipe>;
fn stdio<T>(option: Option<T>, handle: &Handle) -> io::Result<Option<PollEvented<NamedPipe>>>
where
T: IntoRawHandle,
{
let io = match option {
Some(io) => io,
None => return Ok(None),
};
let pipe = unsafe { NamedPipe::from_raw_handle(io.into_raw_handle()) };
let io = PollEvented::new_with_handle(pipe, handle)?;
Ok(Some(io))
}
tokio-process/tests/issue_42.rs
+51
View File
@@ -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()
}
}
tokio-process/tests/smoke.rs
+22
View File
@@ -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());
}
tokio-process/tests/stdio.rs
+113
View File
@@ -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<Future<Item = ExitStatus, Error = io::Error>> {
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?");
}
tokio-process/tests/support/mod.rs
+43
View File
@@ -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<F: Future>(future: F) -> impl Future<Item = F::Item, Error = F::Error> {
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<F>(future: F) -> Result<F::Item, F::Error>
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))
}