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

process: Rewrite the crate with an extension trait

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