process: [WIP] Actually be non-blocking

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" }

src/bin/cat.rs

@@ -15,4 +15,5 @@ fn main() {
         }
         stdout.write(line.as_bytes()).unwrap();
     }
+    stdout.flush().unwrap();
 }

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;

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>(T);
 
+impl<T> RawFdWrap<T> {
+    fn new(fd: T) -> io::Result<Self>
+        where T: AsRawFd {
+
+        try!(set_nonblock(&fd));
+        Ok(RawFdWrap(fd))
+    }
+}
+
+impl<T> io::Read for RawFdWrap<T> where T: io::Read {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        self.0.read(bytes)
+    }
+}
+
+impl<T> io::Write for RawFdWrap<T> where T: io::Write {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        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<T> {
     io: PollEvented<RawFdWrap<T>>,
 }
@@ -37,29 +74,34 @@ 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 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<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)
+        self.io.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)
+        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<T>(option: &mut Option<T>, handle: &Handle) -> Result<Option<StdStream<
     where T: AsRawFd {
 
     option.take().map_or(Ok(None), |stream| {
-        PollEvented::new(RawFdWrap(stream), handle).map(|io| Some(StdStream { io: io }))
+        PollEvented::new(try!(RawFdWrap::new(stream)), handle).map(|io| {
+            Some(StdStream { io: io })
+        })
     })
 }
 

tests/stdio.rs

@@ -2,10 +2,13 @@ extern crate futures;
 #[macro_use]
 extern crate tokio_core;
 extern crate tokio_process;
+#[macro_use]
+extern crate log;
+extern crate env_logger;
 
 use std::env;
-use std::io;
-use std::process::Stdio;
+use std::io::{self, Write};
+use std::process::{Stdio, ExitStatus};
 
 use futures::{Future, BoxFuture};
 use futures::stream::{self, Stream};
@@ -23,13 +26,15 @@ fn cat(handle: &Handle) -> Command {
     cmd
 }
 
-fn feed_cat(cat: &mut Child, n: usize) -> BoxFuture<(), io::Error> {
+fn feed_cat(mut cat: Child, n: usize) -> BoxFuture<ExitStatus, io::Error> {
     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));
 }