From 9766cd644fb3a5c2838085f7d40ab37416f2cee8 Mon Sep 17 00:00:00 2001 From: Ivan Petkov Date: Sat, 31 Aug 2019 13:02:27 -0700 Subject: [PATCH] process: omit several future types in favor of async/await (#1526) --- tokio-net/src/process/mod.rs | 166 +++++++------------------------ tokio-net/tests/process_stdio.rs | 2 +- 2 files changed, 38 insertions(+), 130 deletions(-) diff --git a/tokio-net/src/process/mod.rs b/tokio-net/src/process/mod.rs index 62ffdc68d..b009b0e4b 100644 --- a/tokio-net/src/process/mod.rs +++ b/tokio-net/src/process/mod.rs @@ -107,7 +107,6 @@ //! be regained with the [`Child::forget`] method. use std::ffi::OsStr; -use std::fmt; use std::future::Future; use std::io; #[cfg(unix)] @@ -122,9 +121,7 @@ use std::task::Poll; use self::kill::Kill; use futures_core::TryFuture; -use futures_util::future; -use futures_util::future::FutureExt; -use futures_util::try_future::TryFutureExt; +use futures_util::try_future::try_join3; use tokio_io::{AsyncRead, AsyncReadExt, AsyncWrite}; #[path = "unix/mod.rs"] @@ -549,23 +546,19 @@ impl Command { /// 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. + /// 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 + /// If this 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. + /// This future will return an error if the child process cannot be spawned + /// or if there is an error while awaiting its status. /// /// # Examples /// @@ -577,12 +570,15 @@ impl Command { /// async fn run_ls() -> std::process::ExitStatus { /// Command::new("ls") /// .status() - /// .expect("ls command failed to start") /// .await /// .expect("ls command failed to run") /// } - pub fn status(&mut self) -> io::Result { - self.spawn().map(|mut child| { + pub fn status(&mut self) -> impl Future> { + let child = self.spawn(); + + async { + let mut child = 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. @@ -590,8 +586,8 @@ impl Command { child.stdout.take(); child.stderr.take(); - StatusAsync { inner: child } - }) + child.await + } } /// Executes the command as a child process, waiting for it to finish and @@ -604,7 +600,7 @@ impl Command { /// > `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 + /// child process's stdout/stderr. It will resolve to /// the `Output` type in the standard library, containing `stdout` and /// `stderr` as `Vec` along with an `ExitStatus` representing how the /// process exited. @@ -612,7 +608,7 @@ impl Command { /// 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 + /// If this future is dropped before the future resolves, then /// the child will be killed, if it was spawned. /// /// # Examples @@ -629,15 +625,13 @@ impl Command { /// .expect("ls command failed to run"); /// println!("stderr of ls: {:?}", output.stderr); /// } - pub fn output(&mut self) -> OutputAsync { + pub fn output(&mut self) -> impl Future> { self.std.stdout(Stdio::piped()); self.std.stderr(Stdio::piped()); - let inner = future::ready(self.spawn()).and_then(Child::wait_with_output); + let child = self.spawn(); - OutputAsync { - inner: inner.boxed(), - } + async { child?.wait_with_output().await } } } @@ -772,42 +766,26 @@ impl Child { /// 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_val = self.stdout.take(); - let stderr_val = self.stderr.take(); - let stdout_fut = async { - match stdout_val { - Some(mut io) => { - let mut vec = Vec::new(); - AsyncReadExt::read_to_end(&mut io, &mut vec).await?; - Ok(vec) - } - None => Ok(Vec::new()), + pub async fn wait_with_output(mut self) -> io::Result { + async fn read_to_end(io: Option) -> io::Result> { + let mut vec = Vec::new(); + if let Some(mut io) = io { + AsyncReadExt::read_to_end(&mut io, &mut vec).await?; } - }; - let stderr_fut = async { - match stderr_val { - Some(mut io) => { - let mut vec = Vec::new(); - AsyncReadExt::read_to_end(&mut io, &mut vec).await?; - Ok(vec) - } - None => Ok(Vec::new()), - } - }; - - WaitWithOutput { - inner: futures_util::try_future::try_join3(stdout_fut, stderr_fut, self) - .and_then(|(stdout, stderr, status)| { - future::ok(Output { - status, - stdout, - stderr, - }) - }) - .boxed(), + Ok(vec) } + + drop(self.stdin().take()); + 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?; + + Ok(Output { + status, + stdout, + stderr, + }) } /// Drop this `Child` without killing the underlying process. @@ -846,76 +824,6 @@ impl Future for Child { } } -/// 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: Pin> + Send>>, -} - -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 Output = io::Result; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - Pin::new(&mut self.inner).poll(cx) - } -} - -/// Future returned by the [`Command::status`] 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 Output = io::Result; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - Pin::new(&mut self.inner).poll(cx) - } -} - -/// Future returned by the [`Command::output`] 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: Pin> + Send>>, -} - -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 Output = io::Result; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - Pin::new(&mut self.inner).poll(cx) - } -} - /// The standard input stream for spawned children. /// /// This type implements the `AsyncWrite` trait to pass data to the stdin handle of diff --git a/tokio-net/tests/process_stdio.rs b/tokio-net/tests/process_stdio.rs index 5301ed16a..b7f7c2aff 100644 --- a/tokio-net/tests/process_stdio.rs +++ b/tokio-net/tests/process_stdio.rs @@ -139,7 +139,7 @@ async 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().expect("failed to spawn child"); + let child = cat().status(); with_timeout(child) .await