diff --git a/Cargo.toml b/Cargo.toml index 112201de5..8269fabc7 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/bin/cat.rs b/src/bin/cat.rs new file mode 100644 index 000000000..8118d2bea --- /dev/null +++ b/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(); + } +} diff --git a/src/lib.rs b/src/lib.rs index b8b44d46c..c72887cd8 100644 --- a/src/lib.rs +++ b/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 { + &mut self.inner.stdin + } + + pub fn stdout(&mut self) -> &mut Option { + &mut self.inner.stdout + } + + pub fn stderr(&mut self) -> &mut Option { + &mut self.inner.stderr + } } impl Future for Child { diff --git a/src/unix.rs b/src/unix.rs index fbc780c7a..aa79141c2 100644 --- a/src/unix.rs +++ b/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, + pub stdout: Option, + pub stderr: Option, +} + +struct RawFdWrap(T); + +pub struct StdStream { + io: PollEvented>, +} + +pub type ChildStdin = StdStream; +pub type ChildStdout = StdStream; +pub type ChildStderr = StdStream; + +impl Evented for RawFdWrap 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 io::Read for StdStream where T: io::Read { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.io.get_mut().0.read(buf) + } +} + +impl io::Write for StdStream where T: io::Write { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.io.get_mut().0.write(buf) + } + fn flush(&mut self) -> io::Result<()> { + self.io.get_mut().0.flush() + } +} + +fn stdio(option: &mut Option, handle: &Handle) -> Result>, 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> { 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, + }) }) })) } diff --git a/tests/stdio.rs b/tests/stdio.rs new file mode 100644 index 000000000..4a551b63e --- /dev/null +++ b/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)); +}