process: Add Windows support for stdio streams

Cargo.toml

@@ -15,17 +15,19 @@ tokio-core = "0.1"
 futures = "0.1"
 mio = "0.6"
 log = "0.3"
-env_logger = "0.3"
+
+[dev-dependencies]
+env_logger = { version = "0.3", default-features = false }
 
 [target.'cfg(windows)'.dependencies]
 winapi = "0.2"
 kernel32-sys = "0.2"
+mio-named-pipes = { git = 'https://github.com/alexcrichton/mio-named-pipes' }
 
 [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" }
+"mio:0.6.1" = { git = "https://github.com/alexcrichton/mio", branch = "custom-iocp" }
+"tokio-core:0.1.1" = { git = "https://github.com/tokio-rs/tokio-core" }

src/lib.rs

@@ -6,7 +6,7 @@ extern crate mio;
 extern crate log;
 
 use std::ffi::OsStr;
-use std::io;
+use std::io::{self, Read, Write};
 use std::path::Path;
 use std::process::{self, ExitStatus};
 
@@ -21,9 +21,6 @@ mod imp;
 #[cfg(windows)]
 mod imp;
 
-pub use imp::ChildStdin;
-pub use imp::ChildStdout;
-
 pub struct Command {
     inner: process::Command,
     #[allow(dead_code)]
@@ -36,6 +33,21 @@ pub struct Spawn {
 
 pub struct Child {
     inner: imp::Child,
+    stdin: Option<ChildStdin>,
+    stdout: Option<ChildStdout>,
+    stderr: Option<ChildStderr>,
+}
+
+pub struct ChildStdin {
+    inner: imp::ChildStdin,
+}
+
+pub struct ChildStdout {
+    inner: imp::ChildStdout,
+}
+
+pub struct ChildStderr {
+    inner: imp::ChildStderr,
 }
 
 impl Command {
@@ -116,7 +128,7 @@ impl Command {
 
     pub fn spawn(self) -> Spawn {
         Spawn {
-            inner: Box::new(imp::spawn(self).map(|c| Child { inner: c })),
+            inner: Box::new(imp::spawn(self)),
         }
     }
 }
@@ -139,16 +151,16 @@ impl Child {
         self.inner.kill()
     }
 
-    pub fn stdin(&mut self) -> &mut Option<imp::ChildStdin> {
-        &mut self.inner.stdin
+    pub fn stdin(&mut self) -> &mut Option<ChildStdin> {
+        &mut self.stdin
     }
 
-    pub fn stdout(&mut self) -> &mut Option<imp::ChildStdout> {
-        &mut self.inner.stdout
+    pub fn stdout(&mut self) -> &mut Option<ChildStdout> {
+        &mut self.stdout
     }
 
-    pub fn stderr(&mut self) -> &mut Option<imp::ChildStderr> {
-        &mut self.inner.stderr
+    pub fn stderr(&mut self) -> &mut Option<ChildStderr> {
+        &mut self.stderr
     }
 }
 
@@ -160,3 +172,25 @@ impl Future for Child {
         self.inner.poll()
     }
 }
+
+impl Write for ChildStdin {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        self.inner.write(bytes)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+impl Read for ChildStdout {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        self.inner.read(bytes)
+    }
+}
+
+impl Read for ChildStderr {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        self.inner.read(bytes)
+    }
+}

src/unix.rs

@@ -1,5 +1,4 @@
 extern crate libc;
-extern crate nix;
 extern crate tokio_signal;
 
 use std::io;
@@ -10,8 +9,6 @@ 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;
@@ -24,95 +21,6 @@ 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);
-
-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>>,
-}
-
-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<()> {
-        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<()> {
-        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.read(buf)
-    }
-}
-
-impl<T> io::Write for StdStream<T> where T: io::Write {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.io.write(buf)
-    }
-    fn flush(&mut self) -> io::Result<()> {
-        self.io.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(try!(RawFdWrap::new(stream)), handle).map(|io| {
-            Some(StdStream { io: io })
-        })
-    })
 }
 
 /// Spawns a new child process.
@@ -137,19 +45,35 @@ fn stdio<T>(option: &mut Option<T>, handle: &Handle) -> Result<Option<StdStream<
 /// Note that this means that this isn't really scalable, but then again
 /// processes in general aren't scalable (e.g. millions) so it shouldn't be that
 /// bad in theory...
-pub fn spawn(mut cmd: Command) -> Box<Future<Item=Child, Error=io::Error>> {
+pub fn spawn(mut cmd: Command) -> Box<Future<Item=::Child, Error=io::Error>> {
+    struct KillOnDrop(Option<process::Child>);
+
+    impl Drop for KillOnDrop {
+        fn drop(&mut self) {
+            if let Some(mut c) = self.0.take() {
+                drop(c.kill());
+            }
+        }
+    }
+
     Box::new(Signal::new(libc::SIGCHLD, &cmd.handle).and_then(move |sigchld| {
         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,
-                stdin: stdin,
-                stdout: stdout,
-                stderr: stderr,
+            let stdin = c.stdin.take();
+            let stdout = c.stdout.take();
+            let stderr = c.stderr.take();
+            let mut c = KillOnDrop(Some(c));
+            let stdin = try!(stdio(stdin, &cmd.handle));
+            let stdout = try!(stdio(stdout, &cmd.handle));
+            let stderr = try!(stdio(stderr, &cmd.handle));
+            Ok(::Child {
+                inner: Child {
+                    child: c.0.take().unwrap(),
+                    reaped: false,
+                    sigchld: sigchld,
+                },
+                stdin: stdin.map(|io| ::ChildStdin { inner: io }),
+                stdout: stdout.map(|io| ::ChildStdout { inner: io }),
+                stderr: stderr.map(|io| ::ChildStderr { inner: io }),
             })
         })
     }))
@@ -216,3 +140,80 @@ pub fn try_wait(child: &process::Child) -> io::Result<Option<ExitStatus>> {
         }
     }
 }
+
+pub struct Fd<T>(T);
+
+impl<T: io::Read> io::Read for Fd<T> {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        self.0.read(bytes)
+    }
+}
+
+impl<T: io::Write> io::Write for Fd<T> {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        self.0.write(bytes)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.0.flush()
+    }
+}
+
+pub type ChildStdin = PollEvented<Fd<process::ChildStdin>>;
+pub type ChildStdout = PollEvented<Fd<process::ChildStdout>>;
+pub type ChildStderr = PollEvented<Fd<process::ChildStderr>>;
+
+impl<T> Evented for Fd<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 | 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 | Ready::hup(),
+                                                  opts)
+    }
+
+    fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
+        EventedFd(&self.0.as_raw_fd()).deregister(poll)
+    }
+}
+
+fn stdio<T>(option: Option<T>, handle: &Handle)
+            -> io::Result<Option<PollEvented<Fd<T>>>>
+    where T: AsRawFd
+{
+    let io = match option {
+        Some(io) => io,
+        None => return Ok(None),
+    };
+
+    // Set the fd to nonblocking before we pass it to the event loop
+    unsafe {
+        let fd = io.as_raw_fd();
+        let r = libc::fcntl(fd, libc::F_GETFL);
+        if r == -1 {
+            return Err(io::Error::last_os_error())
+        }
+        let r = libc::fcntl(fd, libc::F_SETFL, r | libc::O_NONBLOCK);
+        if r == -1 {
+            return Err(io::Error::last_os_error())
+        }
+    }
+    let io = try!(PollEvented::new(Fd(io), handle));
+    Ok(Some(io))
+}

src/windows.rs

@@ -1,12 +1,15 @@
 extern crate winapi;
 extern crate kernel32;
+extern crate mio_named_pipes;
 
 use std::io;
 use std::os::windows::prelude::*;
 use std::os::windows::process::ExitStatusExt;
 use std::process::{self, ExitStatus};
 
+use tokio_core::reactor::{PollEvented, Handle};
 use futures::{self, Future, Poll, Async, Oneshot, Complete, oneshot, Fuse};
+use self::mio_named_pipes::NamedPipe;
 
 use Command;
 
@@ -24,12 +27,35 @@ struct Waiting {
 unsafe impl Sync for Waiting {}
 unsafe impl Send for Waiting {}
 
-pub fn spawn(mut cmd: Command) -> Box<Future<Item=Child, Error=io::Error>> {
-    Box::new(futures::done(cmd.inner.spawn().map(|c| {
-        Child {
-            child: c,
-            waiting: None,
+pub fn spawn(mut cmd: Command) -> Box<Future<Item=::Child, Error=io::Error>> {
+    struct KillOnDrop(Option<process::Child>);
+
+    impl Drop for KillOnDrop {
+        fn drop(&mut self) {
+            if let Some(mut c) = self.0.take() {
+                drop(c.kill());
+            }
         }
+    }
+
+    Box::new(futures::done(cmd.inner.spawn().and_then(|mut c| {
+        let stdin = c.stdin.take();
+        let stdout = c.stdout.take();
+        let stderr = c.stderr.take();
+        let mut c = KillOnDrop(Some(c));
+        let stdin = try!(stdio(stdin, &cmd.handle));
+        let stdout = try!(stdio(stdout, &cmd.handle));
+        let stderr = try!(stdio(stderr, &cmd.handle));
+
+        Ok(::Child {
+            inner: Child {
+                child: c.0.take().unwrap(),
+                waiting: None,
+            },
+            stdin: stdin.map(|io| ::ChildStdin { inner: io }),
+            stdout: stdout.map(|io| ::ChildStdout { inner: io }),
+            stderr: stderr.map(|io| ::ChildStderr { inner: io }),
+        })
     })))
 }
 
@@ -121,3 +147,20 @@ pub fn try_wait(child: &process::Child) -> io::Result<Option<ExitStatus>> {
         }
     }
 }
+
+pub type ChildStdin = PollEvented<NamedPipe>;
+pub type ChildStdout = PollEvented<NamedPipe>;
+pub type ChildStderr = PollEvented<NamedPipe>;
+
+fn stdio<T>(option: Option<T>, handle: &Handle)
+            -> io::Result<Option<PollEvented<NamedPipe>>>
+    where T: IntoRawHandle,
+{
+    let io = match option {
+        Some(io) => io,
+        None => return Ok(None),
+    };
+    let pipe = unsafe { NamedPipe::from_raw_handle(io.into_raw_handle()) };
+    let io = try!(PollEvented::new(pipe, handle));
+    Ok(Some(io))
+}

tests/stdio.rs

@@ -7,7 +7,7 @@ extern crate log;
 extern crate env_logger;
 
 use std::env;
-use std::io::{self, Write};
+use std::io;
 use std::process::{Stdio, ExitStatus};
 
 use futures::{Future, BoxFuture};
@@ -22,7 +22,7 @@ fn cat(handle: &Handle) -> Command {
     path.push("cat");
     let mut cmd = Command::new(path, handle);
     cmd.stdin(Stdio::piped())
-        .stdout(Stdio::piped());
+       .stdout(Stdio::piped());
     cmd
 }
 
@@ -35,19 +35,19 @@ fn feed_cat(mut cat: Child, n: usize) -> BoxFuture<ExitStatus, io::Error> {
     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(|_| {});
+        write_all(stdin, format!("line {}\n", i).into_bytes()).map(|p| p.0)
+    }).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 expected_numbers = stream::iter((0..n + 1).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();
+            debug!("read line {} from child ({} bytes, done: {})",
+                   i, vec.len(), done);
             match (done, vec.len()) {
                 (false, 0) => {
                     Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe"))
@@ -67,6 +67,7 @@ fn feed_cat(mut cat: Child, n: usize) -> BoxFuture<ExitStatus, io::Error> {
             }
         })
     });
+
     // Compose reading and writing concurrently.
     write.join(read).and_then(|_| cat).boxed()
 }