process: Initial commit

.gitignore

@@ -0,0 +1,2 @@
+target
+Cargo.lock

Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "tokio-process"
+version = "0.1.0"
+authors = ["Alex Crichton <[email protected]>"]
+
+[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" }

src/bin/exit.rs

@@ -0,0 +1,3 @@
+fn main() {
+    std::process::exit(std::env::args().nth(1).unwrap().parse().unwrap());
+}

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<Future<Item=Child, Error=io::Error>>,
+}
+
+pub struct Child {
+    inner: imp::Child,
+}
+
+impl Command {
+    pub fn new<T: AsRef<OsStr>>(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<S: AsRef<OsStr>>(&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<S: AsRef<OsStr>>(&mut self, args: &[S]) -> &mut Command {
+        for arg in args {
+            self._arg(arg.as_ref());
+        }
+        self
+    }
+
+    pub fn env<K, V>(&mut self, key: K, val: V) -> &mut Command
+        where K: AsRef<OsStr>, V: AsRef<OsStr>
+    {
+        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<K: AsRef<OsStr>>(&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<P: AsRef<Path>>(&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<Child, io::Error> {
+        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<ExitStatus, io::Error> {
+        self.inner.poll()
+    }
+}

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<Future<Item=Child, Error=io::Error>> {
+    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<ExitStatus, io::Error> {
+        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<Option<ExitStatus>> {
+    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)))
+            }
+        }
+    }
+}

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());
+}