process: Add support for stdio streams

Cargo.toml

@@ -13,6 +13,7 @@ An implementation of an asynchronous process management backed futures.
 [dependencies]
 tokio-core = "0.1"
 futures = "0.1"
+mio = "0.6"
 
 [target.'cfg(windows)'.dependencies]
 winapi = "0.2"

src/bin/cat.rs

@@ -0,0 +1,18 @@
+// 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.len() == 0 {
+            break;
+        }
+        stdout.write(line.as_bytes()).unwrap();
+    }
+}

src/lib.rs

@@ -1,6 +1,7 @@
 #[macro_use]
 extern crate futures;
 extern crate tokio_core;
+extern crate mio;
 
 use std::ffi::OsStr;
 use std::io;
@@ -18,6 +19,9 @@ mod imp;
 #[cfg(windows)]
 mod imp;
 
+pub use imp::ChildStdin;
+pub use imp::ChildStdout;
+
 pub struct Command {
     inner: process::Command,
     #[allow(dead_code)]
@@ -94,6 +98,20 @@ impl Command {
         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).map(|c| Child { inner: c })),
@@ -118,6 +136,18 @@ impl Child {
     pub fn kill(&mut self) -> io::Result<()> {
         self.inner.kill()
     }
+
+    pub fn stdin(&mut self) -> &mut Option<imp::ChildStdin> {
+        &mut self.inner.stdin
+    }
+
+    pub fn stdout(&mut self) -> &mut Option<imp::ChildStdout> {
+        &mut self.inner.stdout
+    }
+
+    pub fn stderr(&mut self) -> &mut Option<imp::ChildStderr> {
+        &mut self.inner.stderr
+    }
 }
 
 impl Future for Child {

src/unix.rs

@@ -1,3 +1,4 @@
+extern crate mio;
 extern crate libc;
 extern crate tokio_signal;
 
@@ -7,15 +8,67 @@ use std::process::{self, ExitStatus};
 
 use futures::stream::Stream;
 use futures::{Future, Poll, Async};
+use tokio_core::reactor::{Handle,PollEvented};
 use self::libc::c_int;
 use self::tokio_signal::unix::Signal;
 
+use mio::{Evented,PollOpt,Ready,Token};
+use mio::unix::EventedFd;
+
 use Command;
 
 pub struct Child {
     child: process::Child,
     reaped: bool,
     sigchld: Signal,
+    pub stdin: Option<ChildStdin>,
+    pub stdout: Option<ChildStdout>,
+    pub stderr: Option<ChildStderr>,
+}
+
+struct RawFdWrap<T>(T);
+
+pub struct StdStream<T> {
+    io: PollEvented<RawFdWrap<T>>,
+}
+
+pub type ChildStdin = StdStream<process::ChildStdin>;
+pub type ChildStdout = StdStream<process::ChildStdout>;
+pub type ChildStderr = StdStream<process::ChildStderr>;
+
+impl<T> Evented for RawFdWrap<T> where T: AsRawFd {
+    fn register(&self, poll: &mio::Poll, token: Token, interest: Ready, opts: PollOpt) -> io::Result<()> {
+        EventedFd(&self.0.as_raw_fd()).register(poll, token, interest, opts)
+    }
+    fn reregister(&self, poll: &mio::Poll, token: Token, interest: Ready, opts: PollOpt) -> io::Result<()> {
+        EventedFd(&self.0.as_raw_fd()).reregister(poll, token, interest, opts)
+    }
+    fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
+        EventedFd(&self.0.as_raw_fd()).deregister(poll)
+    }
+}
+
+impl<T> io::Read for StdStream<T> where T: io::Read {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.io.get_mut().0.read(buf)
+    }
+}
+
+impl<T> io::Write for StdStream<T> where T: io::Write {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.io.get_mut().0.write(buf)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        self.io.get_mut().0.flush()
+    }
+}
+
+fn stdio<T>(option: &mut Option<T>, handle: &Handle) -> Result<Option<StdStream<T>>, io::Error>
+    where T: AsRawFd {
+
+    option.take().map_or(Ok(None), |stream| {
+        PollEvented::new(RawFdWrap(stream), handle).map(|io| Some(StdStream { io: io }))
+    })
 }
 
 /// Spawns a new child process.
@@ -42,12 +95,18 @@ pub struct Child {
 /// bad in theory...
 pub fn spawn(mut cmd: Command) -> Box<Future<Item=Child, Error=io::Error>> {
     Box::new(Signal::new(libc::SIGCHLD, &cmd.handle).and_then(move |sigchld| {
-        cmd.inner.spawn().map(|c| {
-            Child {
+        cmd.inner.spawn().and_then(|mut c| {
+            let stdin = try!(stdio(&mut c.stdin, &cmd.handle));
+            let stdout = try!(stdio(&mut c.stdout, &cmd.handle));
+            let stderr = try!(stdio(&mut c.stderr, &cmd.handle));
+            Ok(Child {
                 child: c,
                 reaped: false,
-                sigchld: sigchld
-            }
+                sigchld: sigchld,
+                stdin: stdin,
+                stdout: stdout,
+                stderr: stderr,
+            })
         })
     }))
 }

tests/stdio.rs

@@ -0,0 +1,85 @@
+extern crate futures;
+#[macro_use]
+extern crate tokio_core;
+extern crate tokio_process;
+
+use std::env;
+use std::io;
+use std::process::Stdio;
+
+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};
+
+fn cat(handle: &Handle) -> Command {
+    let mut path = env::current_exe().unwrap();
+    path.pop();
+    path.push("cat");
+    let mut cmd = Command::new(path, handle);
+    cmd.stdin(Stdio::piped())
+        .stdout(Stdio::piped());
+    cmd
+}
+
+fn feed_cat(cat: &mut Child, n: usize) -> BoxFuture<(), io::Error> {
+    let stdin = cat.stdin().take().unwrap();
+    let stdout = cat.stdout().take().unwrap();
+
+    // Produce n lines on the child's stdout.
+    let numbers = stream::iter((0..n).into_iter().map(Ok));
+    let write = numbers.fold(stdin, |stdin, i| {
+        write_all(stdin, format!("line {}\n", i).into_bytes()).map(|(writer, _)| writer)
+    }).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((0..n + 1).into_iter().map(Ok));
+    let read = expected_numbers.fold((reader, 0), move |(reader, i), _| {
+        let done = i >= n;
+        read_until(reader, b'\n', Vec::new()).and_then(move |(reader, vec)| {
+            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.
+    write.join(read).map(|_| {}).boxed()
+}
+
+#[test]
+/// 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.
+fn cat_loop() {
+    let mut lp = Core::new().unwrap();
+    let cmd = cat(&lp.handle());
+    let mut child = lp.run(cmd.spawn()).unwrap();
+    lp.run(feed_cat(&mut child, 10000)).unwrap();
+    let status = lp.run(&mut child).unwrap();
+    assert_eq!(status.code(), Some(0));
+}