commit 97508096fa5671d64eece96b974cd8d989e4a752 Author: Alex Crichton Date: Wed Sep 7 00:13:11 2016 -0700 process: Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 000000000..a9d37c560 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +target +Cargo.lock diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 000000000..2d3bc940a --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "tokio-process" +version = "0.1.0" +authors = ["Alex Crichton "] + +[dependencies] +tokio-core = { git = "https://github.com/tokio-rs/tokio-core" } +futures = { git = "https://github.com/alexcrichton/futures-rs" } + +[target.'cfg(unix)'.dependencies] +libc = "0.2" +tokio-signal = { git = "https://github.com/alexcrichton/tokio-signal" } diff --git a/src/bin/exit.rs b/src/bin/exit.rs new file mode 100644 index 000000000..73ce00d17 --- /dev/null +++ b/src/bin/exit.rs @@ -0,0 +1,3 @@ +fn main() { + std::process::exit(std::env::args().nth(1).unwrap().parse().unwrap()); +} diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 000000000..6e499b9fa --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,124 @@ +#[macro_use] +extern crate futures; +extern crate tokio_core; + +use std::ffi::OsStr; +use std::io; +use std::path::Path; +use std::process::{self, ExitStatus}; + +use futures::{Future, Poll}; +use tokio_core::LoopHandle; + +#[path = "unix.rs"] +mod imp; + +pub struct Command { + inner: process::Command, + handle: LoopHandle, +} + +pub struct Spawn { + inner: Box>, +} + +pub struct Child { + inner: imp::Child, +} + +impl Command { + pub fn new>(exe: T, handle: &LoopHandle) -> Command { + Command::_new(exe.as_ref(), handle) + } + + fn _new(exe: &OsStr, handle: &LoopHandle) -> Command { + Command { + inner: process::Command::new(exe), + handle: handle.clone(), + } + } + + pub fn arg>(&mut self, arg: S) -> &mut Command { + self._arg(arg.as_ref()) + } + + fn _arg(&mut self, arg: &OsStr) -> &mut Command { + self.inner.arg(arg); + self + } + + pub fn args>(&mut self, args: &[S]) -> &mut Command { + for arg in args { + self._arg(arg.as_ref()); + } + self + } + + pub fn env(&mut self, key: K, val: V) -> &mut Command + where K: AsRef, V: AsRef + { + self._env(key.as_ref(), val.as_ref()) + } + + fn _env(&mut self, key: &OsStr, val: &OsStr) -> &mut Command { + self.inner.env(key, val); + self + } + + pub fn env_remove>(&mut self, key: K) -> &mut Command { + self._env_remove(key.as_ref()) + } + + fn _env_remove(&mut self, key: &OsStr) -> &mut Command { + self.inner.env_remove(key); + self + } + + pub fn env_clear(&mut self) -> &mut Command { + self.inner.env_clear(); + self + } + + pub fn current_dir>(&mut self, dir: P) -> &mut Command { + self._current_dir(dir.as_ref()) + } + + fn _current_dir(&mut self, dir: &Path) -> &mut Command { + self.inner.current_dir(dir); + self + } + + pub fn spawn(self) -> Spawn { + Spawn { + inner: Box::new(imp::spawn(self).map(|c| Child { inner: c })), + } + } +} + +impl Future for Spawn { + type Item = Child; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.inner.poll() + } +} + +impl Child { + pub fn id(&self) -> u32 { + self.inner.id() + } + + pub fn kill(&mut self) -> io::Result<()> { + self.inner.kill() + } +} + +impl Future for Child { + type Item = ExitStatus; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + self.inner.poll() + } +} diff --git a/src/unix.rs b/src/unix.rs new file mode 100644 index 000000000..fbc780c7a --- /dev/null +++ b/src/unix.rs @@ -0,0 +1,115 @@ +extern crate libc; +extern crate tokio_signal; + +use std::io; +use std::os::unix::prelude::*; +use std::process::{self, ExitStatus}; + +use futures::stream::Stream; +use futures::{Future, Poll, Async}; +use self::libc::c_int; +use self::tokio_signal::unix::Signal; + +use Command; + +pub struct Child { + child: process::Child, + reaped: bool, + sigchld: Signal, +} + +/// Spawns a new child process. +/// +/// Right now the only "fancy" thing about this is how we implement the +/// `Future` implementation on `Child` to get the exit status. Unix offers +/// no way to register a child with epoll, and the only real way to get a +/// notification when a process exits is the SIGCHLD signal. +/// +/// Signal handling in general is *super* hairy and complicated, and it's even +/// more complicated here with the fact that signals are coalesced, so we may +/// not get a SIGCHLD-per-child. +/// +/// Our best approximation here is to check *all spawned processes* for all +/// SIGCHLD signals received. To do that we create a `Signal`, implemented in +/// the `tokio-signal` crate, which is a stream over signals being received. +/// +/// Later when we poll the process's exit status we simply check to see if a +/// SIGCHLD has happened since we last checked, and while that returns "yes" we +/// keep trying. +/// +/// 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> { + Box::new(Signal::new(libc::SIGCHLD, &cmd.handle).and_then(move |sigchld| { + cmd.inner.spawn().map(|c| { + Child { + child: c, + reaped: false, + sigchld: sigchld + } + }) + })) +} + +impl Child { + pub fn id(&self) -> u32 { + self.child.id() + } + + pub fn kill(&mut self) -> io::Result<()> { + if self.reaped { + Ok(()) + } else { + self.child.kill() + } + } +} + +impl Future for Child { + type Item = ExitStatus; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + assert!(!self.reaped); + loop { + // Ensure that once we've successfully waited we won't try to + // `kill` above. + if let Some(e) = try!(try_wait(&self.child)) { + self.reaped = true; + return Ok(e.into()) + } + + // If the child hasn't exited yet, then it's our responsibility to + // ensure the current task gets notified when it might be able to + // make progress. + // + // As described in `spawn` above, we just indicate that we can + // next make progress once a SIGCHLD is received. + if try!(self.sigchld.poll()).is_not_ready() { + return Ok(Async::NotReady) + } + } + } +} + +pub fn try_wait(child: &process::Child) -> io::Result> { + let id = child.id() as c_int; + let mut status = 0; + loop { + match unsafe { libc::waitpid(id, &mut status, libc::WNOHANG) } { + 0 => return Ok(None), + n if n < 0 => { + let err = io::Error::last_os_error(); + if err.kind() == io::ErrorKind::Interrupted { + continue + } + return Err(err) + } + n => { + assert_eq!(n, id); + return Ok(Some(ExitStatus::from_raw(status))) + } + } + } +} diff --git a/tests/smoke.rs b/tests/smoke.rs new file mode 100644 index 000000000..4e0a65006 --- /dev/null +++ b/tests/smoke.rs @@ -0,0 +1,54 @@ +extern crate futures; +extern crate tokio_core; +extern crate tokio_process; + +use std::env; +use std::sync::mpsc::channel; +use std::sync::{Once, ONCE_INIT}; +use std::thread; + +use futures::Future; +use futures::stream::Stream; +use tokio_core::{Loop, LoopHandle}; +use tokio_process::Command; + +static INIT: Once = ONCE_INIT; + +fn init() { + INIT.call_once(|| { + let (tx, rx) = channel(); + thread::spawn(move || { + let mut lp = Loop::new().unwrap(); + let cmd = exit(&lp.handle()); + let mut child = lp.run(cmd.spawn()).unwrap(); + drop(child.kill()); + lp.run(child).unwrap(); + tx.send(()).unwrap(); + drop(lp.run(futures::empty::<(), ()>())); + }); + rx.recv().unwrap(); + }); +} + +fn exit(handle: &LoopHandle) -> Command { + let mut me = env::current_exe().unwrap(); + me.pop(); + me.push("exit"); + Command::new(me, handle) +} + +#[test] +fn simple() { + init(); + + let mut lp = Loop::new().unwrap(); + let mut cmd = exit(&lp.handle()); + cmd.arg("2"); + let mut child = lp.run(cmd.spawn()).unwrap(); + let id = child.id(); + assert!(id > 0); + let status = lp.run(&mut child).unwrap(); + assert_eq!(status.code(), Some(2)); + assert_eq!(child.id(), id); + assert!(child.kill().is_ok()); +}