diff --git a/Cargo.toml b/Cargo.toml index da60131b7..56a75ac42 100644 --- a/Cargo.toml +++ b/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" } diff --git a/src/lib.rs b/src/lib.rs index 66b186a8e..137e475ad 100644 --- a/src/lib.rs +++ b/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, + stdout: Option, + stderr: Option, +} + +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 { - &mut self.inner.stdin + pub fn stdin(&mut self) -> &mut Option { + &mut self.stdin } - pub fn stdout(&mut self) -> &mut Option { - &mut self.inner.stdout + pub fn stdout(&mut self) -> &mut Option { + &mut self.stdout } - pub fn stderr(&mut self) -> &mut Option { - &mut self.inner.stderr + pub fn stderr(&mut self) -> &mut Option { + &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 { + 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 { + self.inner.read(bytes) + } +} + +impl Read for ChildStderr { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.inner.read(bytes) + } +} diff --git a/src/unix.rs b/src/unix.rs index 928e0ea7e..9c74f182e 100644 --- a/src/unix.rs +++ b/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, - pub stdout: Option, - pub stderr: Option, -} - -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>, -} - -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<()> { - 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 io::Read for StdStream where T: io::Read { - fn read(&mut self, buf: &mut [u8]) -> io::Result { - self.io.read(buf) - } -} - -impl io::Write for StdStream where T: io::Write { - fn write(&mut self, buf: &[u8]) -> io::Result { - self.io.write(buf) - } - fn flush(&mut self) -> io::Result<()> { - self.io.flush() - } -} - -fn stdio(option: &mut Option, handle: &Handle) -> Result>, 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(option: &mut Option, handle: &Handle) -> Result Box> { +pub fn spawn(mut cmd: Command) -> Box> { + struct KillOnDrop(Option); + + 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> { } } } + +pub struct Fd(T); + +impl io::Read for Fd { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + self.0.read(bytes) + } +} + +impl io::Write for Fd { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.write(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } +} + +pub type ChildStdin = PollEvented>; +pub type ChildStdout = PollEvented>; +pub type ChildStderr = PollEvented>; + +impl Evented for Fd 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(option: Option, handle: &Handle) + -> io::Result>>> + 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)) +} diff --git a/src/windows.rs b/src/windows.rs index 48667f8b8..ccea91c77 100644 --- a/src/windows.rs +++ b/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> { - Box::new(futures::done(cmd.inner.spawn().map(|c| { - Child { - child: c, - waiting: None, +pub fn spawn(mut cmd: Command) -> Box> { + struct KillOnDrop(Option); + + 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> { } } } + +pub type ChildStdin = PollEvented; +pub type ChildStdout = PollEvented; +pub type ChildStderr = PollEvented; + +fn stdio(option: Option, handle: &Handle) + -> io::Result>> + 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)) +} diff --git a/tests/stdio.rs b/tests/stdio.rs index 1a8224f45..56b4e46f2 100644 --- a/tests/stdio.rs +++ b/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 { 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 { } }) }); + // Compose reading and writing concurrently. write.join(read).and_then(|_| cat).boxed() }