Skip to content

Commit

Permalink
process: add Child::{wait,try_wait} (#2796)
Browse files Browse the repository at this point in the history
* add Child::try_wait to mirror the std API
* replace Future impl on Child with `.wait()` method to bring our
  APIs closer to those in std and it allow us to
  internally fuse the future so that repeated calls to `wait` result in
  the same value (similar to std) without forcing the caller to fuse the
  outer future
* Also change `Child::id` to return an Option result to avoid
  allowing the caller to accidentally use the pid on Unix systems after
  the child has been reaped
* Also remove deprecated Child methods
  • Loading branch information
ipetkov committed Sep 7, 2020
1 parent d74eabc commit 842d556
Show file tree
Hide file tree
Showing 9 changed files with 146 additions and 88 deletions.
25 changes: 24 additions & 1 deletion tests-integration/tests/process_stdio.rs
Expand Up @@ -72,7 +72,7 @@ async fn feed_cat(mut cat: Child, n: usize) -> io::Result<ExitStatus> {
};

// Compose reading and writing concurrently.
future::join3(write, read, cat)
future::join3(write, read, cat.wait())
.map(|(_, _, status)| status)
.await
}
Expand Down Expand Up @@ -125,3 +125,26 @@ async fn status_closes_any_pipes() {

assert_ok!(child.await);
}

#[tokio::test]
async fn try_wait() {
let mut child = cat().spawn().unwrap();

let id = child.id().expect("missing id");
assert!(id > 0);

assert_eq!(None, assert_ok!(child.try_wait()));

// Drop the child's stdio handles so it can terminate
drop(child.stdin.take());
drop(child.stderr.take());
drop(child.stdout.take());

assert_ok!(child.wait().await);

// test that the `.try_wait()` method is fused just like the stdlib
assert!(assert_ok!(child.try_wait()).unwrap().success());

// Can't get id after process has exited
assert_eq!(child.id(), None);
}
3 changes: 0 additions & 3 deletions tokio/Cargo.toml
Expand Up @@ -63,9 +63,7 @@ process = [
"mio-named-pipes",
"signal",
"winapi/consoleapi",
"winapi/minwindef",
"winapi/threadpoollegacyapiset",
"winapi/winerror",
]
# Includes basic task execution capabilities
rt-core = ["slab"]
Expand All @@ -81,7 +79,6 @@ signal = [
"mio-uds",
"signal-hook-registry",
"winapi/consoleapi",
"winapi/minwindef",
]
stream = ["futures-core"]
sync = ["fnv"]
Expand Down
147 changes: 96 additions & 51 deletions tokio/src/process/mod.rs
Expand Up @@ -18,16 +18,15 @@
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! // The usage is the same as with the standard library's `Command` type, however the value
//! // returned from `spawn` is a `Result` containing a `Future`.
//! let child = Command::new("echo").arg("hello").arg("world")
//! .spawn();
//! // The usage is similar as with the standard library's `Command` type
//! let mut child = Command::new("echo")
//! .arg("hello")
//! .arg("world")
//! .spawn()
//! .expect("failed to spawn");
//!
//! // Make sure our child succeeded in spawning and process the result
//! let future = child.expect("failed to spawn");
//!
//! // Await until the future (and the command) completes
//! let status = future.await?;
//! // Await until the command completes
//! let status = child.wait().await?;
//! println!("the command exited with: {}", status);
//! Ok(())
//! }
Expand Down Expand Up @@ -83,8 +82,8 @@
//!
//! // Ensure the child process is spawned in the runtime so it can
//! // make progress on its own while we await for any output.
//! tokio::spawn(async {
//! let status = child.await
//! tokio::spawn(async move {
//! let status = child.wait().await
//! .expect("child process encountered an error");
//!
//! println!("child status was: {}", status);
Expand Down Expand Up @@ -555,16 +554,17 @@ impl Command {
/// Command::new("ls")
/// .spawn()
/// .expect("ls command failed to start")
/// .wait()
/// .await
/// .expect("ls command failed to run")
/// }
/// ```
pub fn spawn(&mut self) -> io::Result<Child> {
imp::spawn_child(&mut self.std).map(|spawned_child| Child {
child: ChildDropGuard {
child: FusedChild::Child(ChildDropGuard {
inner: spawned_child.child,
kill_on_drop: self.kill_on_drop,
},
}),
stdin: spawned_child.stdin.map(|inner| ChildStdin { inner }),
stdout: spawned_child.stdout.map(|inner| ChildStdout { inner }),
stderr: spawned_child.stderr.map(|inner| ChildStderr { inner }),
Expand Down Expand Up @@ -615,7 +615,7 @@ impl Command {
child.stdout.take();
child.stderr.take();

child.await
child.wait().await
}
}

Expand Down Expand Up @@ -725,12 +725,16 @@ where
}
}

/// Keeps track of the exit status of a child process without worrying about
/// polling the underlying futures even after they have completed.
#[derive(Debug)]
enum FusedChild {
Child(ChildDropGuard<imp::Child>),
Done(ExitStatus),
}

/// 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.
///
/// # Caveats
/// Similar to the behavior to the standard library, and unlike the futures
/// paradigm of dropping-implies-cancellation, a spawned process will, by
Expand All @@ -739,10 +743,9 @@ where
/// The `Command::kill_on_drop` method can be used to modify this behavior
/// and kill the child process if the `Child` wrapper is dropped before it
/// has exited.
#[must_use = "futures do nothing unless polled"]
#[derive(Debug)]
pub struct Child {
child: ChildDropGuard<imp::Child>,
child: FusedChild,

/// The handle for writing to the child's standard input (stdin), if it has
/// been captured.
Expand All @@ -758,9 +761,17 @@ pub struct Child {
}

impl Child {
/// Returns the OS-assigned process identifier associated with this child.
pub fn id(&self) -> u32 {
self.child.inner.id()
/// Returns the OS-assigned process identifier associated with this child
/// while it is still running.
///
/// Once the child has been polled to completion this will return `None`.
/// This is done to avoid confusion on platforms like Unix where the OS
/// identifier could be reused once the process has completed.
pub fn id(&self) -> Option<u32> {
match &self.child {
FusedChild::Child(child) => Some(child.inner.id()),
FusedChild::Done(_) => None,
}
}

/// Forces the child to exit.
Expand All @@ -783,35 +794,77 @@ impl Child {
/// let mut child = Command::new("sleep").arg("1").spawn().unwrap();
/// tokio::spawn(async move { send.send(()) });
/// tokio::select! {
/// _ = &mut child => {}
/// _ = child.wait() => {}
/// _ = recv => {
/// &mut child.kill();
/// child.kill().expect("kill failed");
/// // NB: await the child here to avoid a zombie process on Unix platforms
/// child.await.unwrap();
/// child.wait().await.unwrap();
/// }
/// }
/// }

pub fn kill(&mut self) -> io::Result<()> {
self.child.kill()
}

#[doc(hidden)]
#[deprecated(note = "please use `child.stdin` instead")]
pub fn stdin(&mut self) -> &mut Option<ChildStdin> {
&mut self.stdin
match &mut self.child {
FusedChild::Child(child) => child.kill(),
FusedChild::Done(_) => Err(io::Error::new(
io::ErrorKind::InvalidInput,
"invalid argument: can't kill an exited process",
)),
}
}

#[doc(hidden)]
#[deprecated(note = "please use `child.stdout` instead")]
pub fn stdout(&mut self) -> &mut Option<ChildStdout> {
&mut self.stdout
/// Waits for the child to exit completely, returning the status that it
/// exited with. This function will continue to have the same return value
/// after it has been called at least once.
///
/// 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.
pub async fn wait(&mut self) -> io::Result<ExitStatus> {
match &mut self.child {
FusedChild::Done(exit) => Ok(*exit),
FusedChild::Child(child) => {
let ret = child.await;

if let Ok(exit) = ret {
self.child = FusedChild::Done(exit);
}

ret
}
}
}

#[doc(hidden)]
#[deprecated(note = "please use `child.stderr` instead")]
pub fn stderr(&mut self) -> &mut Option<ChildStderr> {
&mut self.stderr
/// Attempts to collect the exit status of the child if it has already
/// exited.
///
/// This function will not block the calling thread and will only
/// check to see if the child process has exited or not. If the child has
/// exited then on Unix the process ID is reaped. This function is
/// guaranteed to repeatedly return a successful exit status so long as the
/// child has already exited.
///
/// If the child has exited, then `Ok(Some(status))` is returned. If the
/// exit status is not available at this time then `Ok(None)` is returned.
/// If an error occurs, then that error is returned.
///
/// Note that unlike `wait`, this function will not attempt to drop stdin,
/// nor will it wake the current task if the child exits.
pub fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
match &mut self.child {
FusedChild::Done(exit) => Ok(Some(*exit)),
FusedChild::Child(guard) => {
let ret = guard.inner.try_wait();

if let Ok(Some(exit)) = ret {
// Avoid the overhead of trying to kill a reaped process
guard.kill_on_drop = false;
self.child = FusedChild::Done(exit);
}

ret
}
}
}

/// Returns a future that will resolve to an `Output`, containing the exit
Expand Down Expand Up @@ -845,7 +898,7 @@ impl Child {
let stdout_fut = read_to_end(self.stdout.take());
let stderr_fut = read_to_end(self.stderr.take());

let (status, stdout, stderr) = try_join3(self, stdout_fut, stderr_fut).await?;
let (status, stdout, stderr) = try_join3(self.wait(), stdout_fut, stderr_fut).await?;

Ok(Output {
status,
Expand All @@ -855,14 +908,6 @@ impl Child {
}
}

impl Future for Child {
type Output = io::Result<ExitStatus>;

fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
Pin::new(&mut self.child).poll(cx)
}
}

/// The standard input stream for spawned children.
///
/// This type implements the `AsyncWrite` trait to pass data to the stdin handle of
Expand Down
4 changes: 4 additions & 0 deletions tokio/src/process/unix/mod.rs
Expand Up @@ -117,6 +117,10 @@ impl Child {
pub(crate) fn id(&self) -> u32 {
self.inner.id()
}

pub(crate) fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
self.inner.inner_mut().try_wait()
}
}

impl Kill for Child {
Expand Down
2 changes: 1 addition & 1 deletion tokio/src/process/unix/reap.rs
Expand Up @@ -63,7 +63,7 @@ where
self.inner.as_ref().expect("inner has gone away")
}

fn inner_mut(&mut self) -> &mut W {
pub(crate) fn inner_mut(&mut self) -> &mut W {
self.inner.as_mut().expect("inner has gone away")
}
}
Expand Down
34 changes: 8 additions & 26 deletions tokio/src/process/windows.rs
Expand Up @@ -24,20 +24,15 @@ use mio_named_pipes::NamedPipe;
use std::fmt;
use std::future::Future;
use std::io;
use std::os::windows::prelude::*;
use std::os::windows::process::ExitStatusExt;
use std::os::windows::prelude::{AsRawHandle, FromRawHandle, IntoRawHandle};
use std::pin::Pin;
use std::process::{Child as StdChild, Command as StdCommand, ExitStatus};
use std::ptr;
use std::task::Context;
use std::task::Poll;
use winapi::shared::minwindef::FALSE;
use winapi::shared::winerror::WAIT_TIMEOUT;
use winapi::um::handleapi::INVALID_HANDLE_VALUE;
use winapi::um::processthreadsapi::GetExitCodeProcess;
use winapi::um::synchapi::WaitForSingleObject;
use winapi::um::threadpoollegacyapiset::UnregisterWaitEx;
use winapi::um::winbase::{RegisterWaitForSingleObject, INFINITE, WAIT_OBJECT_0};
use winapi::um::winbase::{RegisterWaitForSingleObject, INFINITE};
use winapi::um::winnt::{BOOLEAN, HANDLE, PVOID, WT_EXECUTEINWAITTHREAD, WT_EXECUTEONLYONCE};

#[must_use = "futures do nothing unless polled"]
Expand Down Expand Up @@ -86,6 +81,10 @@ impl Child {
pub(crate) fn id(&self) -> u32 {
self.child.id()
}

pub(crate) fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
self.child.try_wait()
}
}

impl Kill for Child {
Expand All @@ -106,11 +105,11 @@ impl Future for Child {
Poll::Ready(Err(_)) => panic!("should not be canceled"),
Poll::Pending => return Poll::Pending,
}
let status = try_wait(&inner.child)?.expect("not ready yet");
let status = inner.try_wait()?.expect("not ready yet");
return Poll::Ready(Ok(status));
}

if let Some(e) = try_wait(&inner.child)? {
if let Some(e) = inner.try_wait()? {
return Poll::Ready(Ok(e));
}
let (tx, rx) = oneshot::channel();
Expand Down Expand Up @@ -157,23 +156,6 @@ unsafe extern "system" fn callback(ptr: PVOID, _timer_fired: BOOLEAN) {
let _ = complete.take().unwrap().send(());
}

pub(crate) fn try_wait(child: &StdChild) -> 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(crate) type ChildStdin = PollEvented<NamedPipe>;
pub(crate) type ChildStdout = PollEvented<NamedPipe>;
pub(crate) type ChildStderr = PollEvented<NamedPipe>;
Expand Down
2 changes: 1 addition & 1 deletion tokio/tests/process_issue_2174.rs
Expand Up @@ -40,7 +40,7 @@ async fn issue_2174() {

// Kill the child process.
child.kill().unwrap();
let _ = child.await;
let _ = child.wait().await;

assert_err!(handle.await);
}

0 comments on commit 842d556

Please sign in to comment.