diff --git a/Cargo.toml b/Cargo.toml index 8269fabc7..da60131b7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,8 @@ An implementation of an asynchronous process management backed futures. tokio-core = "0.1" futures = "0.1" mio = "0.6" +log = "0.3" +env_logger = "0.3" [target.'cfg(windows)'.dependencies] winapi = "0.2" @@ -21,4 +23,9 @@ kernel32-sys = "0.2" [target.'cfg(unix)'.dependencies] libc = "0.2" +nix = "0.6" tokio-signal = "0.1" + +[replace] +"mio:0.6.1" = { path = "mio" } +"tokio-core:0.1.1" = { path = "tokio-core" } diff --git a/src/bin/cat.rs b/src/bin/cat.rs index 8118d2bea..b982fceaa 100644 --- a/src/bin/cat.rs +++ b/src/bin/cat.rs @@ -15,4 +15,5 @@ fn main() { } stdout.write(line.as_bytes()).unwrap(); } + stdout.flush().unwrap(); } diff --git a/src/lib.rs b/src/lib.rs index c72887cd8..66b186a8e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,6 +2,8 @@ extern crate futures; extern crate tokio_core; extern crate mio; +#[macro_use] +extern crate log; use std::ffi::OsStr; use std::io; diff --git a/src/unix.rs b/src/unix.rs index aa79141c2..928e0ea7e 100644 --- a/src/unix.rs +++ b/src/unix.rs @@ -1,5 +1,5 @@ -extern crate mio; extern crate libc; +extern crate nix; extern crate tokio_signal; use std::io; @@ -10,9 +10,12 @@ use futures::stream::Stream; use futures::{Future, Poll, Async}; use tokio_core::reactor::{Handle,PollEvented}; use self::libc::c_int; +use self::nix::fcntl::FcntlArg::F_SETFL; +use self::nix::fcntl::{fcntl, O_NONBLOCK}; use self::tokio_signal::unix::Signal; -use mio::{Evented,PollOpt,Ready,Token}; +use mio; +use mio::{Evented, PollOpt, Ready, Token}; use mio::unix::EventedFd; use Command; @@ -28,6 +31,40 @@ pub struct Child { struct RawFdWrap(T); +impl RawFdWrap { + fn new(fd: T) -> io::Result + where T: AsRawFd { + + try!(set_nonblock(&fd)); + Ok(RawFdWrap(fd)) + } +} + +impl io::Read for RawFdWrap where T: io::Read { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.0.read(bytes) + } +} + +impl io::Write for RawFdWrap where T: io::Write { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.write(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } +} + +fn from_nix_error(err: nix::Error) -> io::Error { + io::Error::from_raw_os_error(err.errno() as i32) +} + +fn set_nonblock(s: &AsRawFd) -> io::Result<()> { + fcntl(s.as_raw_fd(), F_SETFL(O_NONBLOCK)).map_err(from_nix_error) + .map(|_| ()) +} + pub struct StdStream { io: PollEvented>, } @@ -37,29 +74,34 @@ 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 register(&self, poll: &mio::Poll, token: Token, interest: Ready, opts: PollOpt) + -> io::Result<()> { + debug!("Evented::register({:?}, {:?}, {:?}", token, interest, opts); + EventedFd(&self.0.as_raw_fd()).register(poll, token, interest | Ready::hup(), 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 reregister(&self, poll: &mio::Poll, token: Token, interest: Ready, opts: PollOpt) + -> io::Result<()> { + debug!("Evented::reregister({:?}, {:?}, {:?}", token, interest, opts); + EventedFd(&self.0.as_raw_fd()).reregister(poll, token, interest | Ready::hup(), opts) } fn deregister(&self, poll: &mio::Poll) -> io::Result<()> { + debug!("Evented::deregister()"); 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) + self.io.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) + self.io.write(buf) } fn flush(&mut self) -> io::Result<()> { - self.io.get_mut().0.flush() + self.io.flush() } } @@ -67,7 +109,9 @@ fn stdio(option: &mut Option, handle: &Handle) -> Result Command { cmd } -fn feed_cat(cat: &mut Child, n: usize) -> BoxFuture<(), io::Error> { +fn feed_cat(mut cat: Child, n: usize) -> BoxFuture { let stdin = cat.stdin().take().unwrap(); let stdout = cat.stdout().take().unwrap(); + debug!("starting to feed"); // 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| { + debug!("sending line {} to child", i); write_all(stdin, format!("line {}\n", i).into_bytes()).map(|(writer, _)| writer) }).map(|_| {}); @@ -39,7 +44,10 @@ fn feed_cat(cat: &mut Child, n: usize) -> BoxFuture<(), io::Error> { 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; + debug!("starting read from child"); read_until(reader, b'\n', Vec::new()).and_then(move |(reader, vec)| { + debug!("read line {} from child ({} bytes, done: {})", i, vec.len(), done); + io::stdout().flush().unwrap(); match (done, vec.len()) { (false, 0) => { Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")) @@ -60,7 +68,7 @@ fn feed_cat(cat: &mut Child, n: usize) -> BoxFuture<(), io::Error> { }) }); // Compose reading and writing concurrently. - write.join(read).map(|_| {}).boxed() + write.join(read).and_then(|_| cat).boxed() } #[test] @@ -75,11 +83,14 @@ fn feed_cat(cat: &mut Child, n: usize) -> BoxFuture<(), io::Error> { /// - 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() { +fn feed_a_lot() { + let _ = ::env_logger::init(); + 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(); + let child = cmd.spawn().and_then(|child| { + feed_cat(child, 10000) + }); + let status = lp.run(child).unwrap(); assert_eq!(status.code(), Some(0)); }