io: Implement process wait timeouts

This implements set_timeout() for std::io::Process which will affect wait()
operations on the process. This follows the same pattern as the rest of the
timeouts emerging in std::io::net.

The implementation was super easy for everything except libnative on unix
(backwards from usual!), which required a good bit of signal handling. There's a
doc comment explaining the strategy in libnative. Internally, this also required
refactoring the "helper thread" implementation used by libnative to allow for an
extra helper thread (not just the timer).

This is a breaking change in terms of the io::Process API. It is now possible
for wait() to fail, and subsequently wait_with_output(). These two functions now
return IoResult<T> due to the fact that they can time out.

Additionally, the wait_with_output() function has moved from taking `&mut self`
to taking `self`. If a timeout occurs while waiting with output, the semantics
are undesirable in almost all cases if attempting to re-wait on the process.
Equivalent functionality can still be achieved by dealing with the output
handles manually.

[breaking-change]

cc #13523

8 files changed, 649 insertions, 222 deletions

diff --git a/src/libnative/io/c_unix.rs b/src/libnative/io/c_unix.rs
index abb22476e52..767090a10cd 100644
--- a/src/libnative/io/c_unix.rs
+++ b/src/libnative/io/c_unix.rs
@@ -10,7 +10,12 @@
 
 //! C definitions used by libnative that don't belong in liblibc
 
+#![allow(dead_code)]
+
 pub use self::select::fd_set;
+pub use self::signal::{sigaction, siginfo, sigset_t};
+pub use self::signal::{SA_ONSTACK, SA_RESTART, SA_RESETHAND, SA_NOCLDSTOP};
+pub use self::signal::{SA_NODEFER, SA_NOCLDWAIT, SA_SIGINFO, SIGCHLD};
 
 use libc;
 
@@ -34,6 +39,8 @@ pub static MSG_DONTWAIT: libc::c_int = 0x80;
 #[cfg(target_os = "android")]
 pub static MSG_DONTWAIT: libc::c_int = 0x40;
 
+pub static WNOHANG: libc::c_int = 1;
+
 extern {
     pub fn gettimeofday(timeval: *mut libc::timeval,
                         tzp: *libc::c_void) -> libc::c_int;
@@ -49,6 +56,17 @@ extern {
                       optlen: *mut libc::socklen_t) -> libc::c_int;
     pub fn ioctl(fd: libc::c_int, req: libc::c_ulong, ...) -> libc::c_int;
 
+
+    pub fn waitpid(pid: libc::pid_t, status: *mut libc::c_int,
+                   options: libc::c_int) -> libc::pid_t;
+
+    pub fn sigaction(signum: libc::c_int,
+                     act: *sigaction,
+                     oldact: *mut sigaction) -> libc::c_int;
+
+    pub fn sigaddset(set: *mut sigset_t, signum: libc::c_int) -> libc::c_int;
+    pub fn sigdelset(set: *mut sigset_t, signum: libc::c_int) -> libc::c_int;
+    pub fn sigemptyset(set: *mut sigset_t) -> libc::c_int;
 }
 
 #[cfg(target_os = "macos")]
@@ -81,3 +99,94 @@ mod select {
         set.fds_bits[fd / uint::BITS] |= 1 << (fd % uint::BITS);
     }
 }
+
+#[cfg(target_os = "linux")]
+#[cfg(target_os = "android")]
+mod signal {
+    use libc;
+
+    pub static SA_NOCLDSTOP: libc::c_ulong = 0x00000001;
+    pub static SA_NOCLDWAIT: libc::c_ulong = 0x00000002;
+    pub static SA_NODEFER: libc::c_ulong = 0x40000000;
+    pub static SA_ONSTACK: libc::c_ulong = 0x08000000;
+    pub static SA_RESETHAND: libc::c_ulong = 0x80000000;
+    pub static SA_RESTART: libc::c_ulong = 0x10000000;
+    pub static SA_SIGINFO: libc::c_ulong = 0x00000004;
+    pub static SIGCHLD: libc::c_int = 17;
+
+    // This definition is not as accurate as it could be, {pid, uid, status} is
+    // actually a giant union. Currently we're only interested in these fields,
+    // however.
+    pub struct siginfo {
+        si_signo: libc::c_int,
+        si_errno: libc::c_int,
+        si_code: libc::c_int,
+        pub pid: libc::pid_t,
+        pub uid: libc::uid_t,
+        pub status: libc::c_int,
+    }
+
+    pub struct sigaction {
+        pub sa_handler: extern fn(libc::c_int),
+        pub sa_mask: sigset_t,
+        pub sa_flags: libc::c_ulong,
+        sa_restorer: *mut libc::c_void,
+    }
+
+    #[cfg(target_word_size = "32")]
+    pub struct sigset_t {
+        __val: [libc::c_ulong, ..32],
+    }
+    #[cfg(target_word_size = "64")]
+    pub struct sigset_t {
+        __val: [libc::c_ulong, ..16],
+    }
+}
+
+#[cfg(target_os = "macos")]
+#[cfg(target_os = "freebsd")]
+mod signal {
+    use libc;
+
+    pub static SA_ONSTACK: libc::c_int = 0x0001;
+    pub static SA_RESTART: libc::c_int = 0x0002;
+    pub static SA_RESETHAND: libc::c_int = 0x0004;
+    pub static SA_NOCLDSTOP: libc::c_int = 0x0008;
+    pub static SA_NODEFER: libc::c_int = 0x0010;
+    pub static SA_NOCLDWAIT: libc::c_int = 0x0020;
+    pub static SA_SIGINFO: libc::c_int = 0x0040;
+    pub static SIGCHLD: libc::c_int = 20;
+
+    #[cfg(target_os = "macos")]
+    pub type sigset_t = u32;
+    #[cfg(target_os = "freebsd")]
+    pub struct sigset_t {
+        bits: [u32, ..4],
+    }
+
+    // This structure has more fields, but we're not all that interested in
+    // them.
+    pub struct siginfo {
+        pub si_signo: libc::c_int,
+        pub si_errno: libc::c_int,
+        pub si_code: libc::c_int,
+        pub pid: libc::pid_t,
+        pub uid: libc::uid_t,
+        pub status: libc::c_int,
+    }
+
+    #[cfg(target_os = "macos")]
+    pub struct sigaction {
+        pub sa_handler: extern fn(libc::c_int),
+        sa_tramp: *mut libc::c_void,
+        pub sa_mask: sigset_t,
+        pub sa_flags: libc::c_int,
+    }
+
+    #[cfg(target_os = "freebsd")]
+    pub struct sigaction {
+        pub sa_handler: extern fn(libc::c_int),
+        pub sa_flags: libc::c_int,
+        pub sa_mask: sigset_t,
+    }
+}
diff --git a/src/libnative/io/helper_thread.rs b/src/libnative/io/helper_thread.rs
new file mode 100644
index 00000000000..2260d74e161
--- /dev/null
+++ b/src/libnative/io/helper_thread.rs
@@ -0,0 +1,205 @@
+// Copyright 2013-2014 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! Implementation of the helper thread for the timer module
+//!
+//! This module contains the management necessary for the timer worker thread.
+//! This thread is responsible for performing the send()s on channels for timers
+//! that are using channels instead of a blocking call.
+//!
+//! The timer thread is lazily initialized, and it's shut down via the
+//! `shutdown` function provided. It must be maintained as an invariant that
+//! `shutdown` is only called when the entire program is finished. No new timers
+//! can be created in the future and there must be no active timers at that
+//! time.
+
+#![macro_escape]
+
+use std::mem;
+use std::rt::bookkeeping;
+use std::rt;
+use std::ty::Unsafe;
+use std::unstable::mutex::StaticNativeMutex;
+
+use task;
+
+/// A structure for management of a helper thread.
+///
+/// This is generally a static structure which tracks the lifetime of a helper
+/// thread.
+///
+/// The fields of this helper are all public, but they should not be used, this
+/// is for static initialization.
+pub struct Helper<M> {
+    /// Internal lock which protects the remaining fields
+    pub lock: StaticNativeMutex,
+
+    // You'll notice that the remaining fields are Unsafe<T>, and this is
+    // because all helper thread operations are done through &self, but we need
+    // these to be mutable (once `lock` is held).
+
+    /// Lazily allocated channel to send messages to the helper thread.
+    pub chan: Unsafe<*mut Sender<M>>,
+
+    /// OS handle used to wake up a blocked helper thread
+    pub signal: Unsafe<uint>,
+
+    /// Flag if this helper thread has booted and been initialized yet.
+    pub initialized: Unsafe<bool>,
+}
+
+macro_rules! helper_init( (static mut $name:ident: Helper<$m:ty>) => (
+    static mut $name: Helper<$m> = Helper {
+        lock: ::std::unstable::mutex::NATIVE_MUTEX_INIT,
+        chan: ::std::ty::Unsafe {
+            value: 0 as *mut Sender<$m>,
+            marker1: ::std::kinds::marker::InvariantType,
+        },
+        signal: ::std::ty::Unsafe {
+            value: 0,
+            marker1: ::std::kinds::marker::InvariantType,
+        },
+        initialized: ::std::ty::Unsafe {
+            value: false,
+            marker1: ::std::kinds::marker::InvariantType,
+        },
+    };
+) )
+
+impl<M: Send> Helper<M> {
+    /// Lazily boots a helper thread, becoming a no-op if the helper has already
+    /// been spawned.
+    ///
+    /// This function will check to see if the thread has been initialized, and
+    /// if it has it returns quickly. If initialization has not happened yet,
+    /// the closure `f` will be run (inside of the initialization lock) and
+    /// passed to the helper thread in a separate task.
+    ///
+    /// This function is safe to be called many times.
+    pub fn boot<T: Send>(&'static self,
+                         f: || -> T,
+                         helper: fn(imp::signal, Receiver<M>, T)) {
+        unsafe {
+            let _guard = self.lock.lock();
+            if !*self.initialized.get() {
+                let (tx, rx) = channel();
+                *self.chan.get() = mem::transmute(box tx);
+                let (receive, send) = imp::new();
+                *self.signal.get() = send as uint;
+
+                let t = f();
+                task::spawn(proc() {
+                    bookkeeping::decrement();
+                    helper(receive, rx, t);
+                    self.lock.lock().signal()
+                });
+
+                rt::at_exit(proc() { self.shutdown() });
+                *self.initialized.get() = true;
+            }
+        }
+    }
+
+    /// Sends a message to a spawned worker thread.
+    ///
+    /// This is only valid if the worker thread has previously booted
+    pub fn send(&'static self, msg: M) {
+        unsafe {
+            let _guard = self.lock.lock();
+
+            // Must send and *then* signal to ensure that the child receives the
+            // message. Otherwise it could wake up and go to sleep before we
+            // send the message.
+            assert!(!self.chan.get().is_null());
+            (**self.chan.get()).send(msg);
+            imp::signal(*self.signal.get() as imp::signal);
+        }
+    }
+
+    fn shutdown(&'static self) {
+        unsafe {
+            // Shut down, but make sure this is done inside our lock to ensure
+            // that we'll always receive the exit signal when the thread
+            // returns.
+            let guard = self.lock.lock();
+
+            // Close the channel by destroying it
+            let chan: Box<Sender<M>> = mem::transmute(*self.chan.get());
+            *self.chan.get() = 0 as *mut Sender<M>;
+            drop(chan);
+            imp::signal(*self.signal.get() as imp::signal);
+
+            // Wait for the child to exit
+            guard.wait();
+            drop(guard);
+
+            // Clean up after ourselves
+            self.lock.destroy();
+            imp::close(*self.signal.get() as imp::signal);
+            *self.signal.get() = 0;
+        }
+    }
+}
+
+#[cfg(unix)]
+mod imp {
+    use libc;
+    use std::os;
+
+    use io::file::FileDesc;
+
+    pub type signal = libc::c_int;
+
+    pub fn new() -> (signal, signal) {
+        let pipe = os::pipe();
+        (pipe.input, pipe.out)
+    }
+
+    pub fn signal(fd: libc::c_int) {
+        FileDesc::new(fd, false).inner_write([0]).unwrap();
+    }
+
+    pub fn close(fd: libc::c_int) {
+        let _fd = FileDesc::new(fd, true);
+    }
+}
+
+#[cfg(windows)]
+mod imp {
+    use libc::{BOOL, LPCSTR, HANDLE, LPSECURITY_ATTRIBUTES, CloseHandle};
+    use std::ptr;
+    use libc;
+
+    pub type signal = HANDLE;
+
+    pub fn new() -> (HANDLE, HANDLE) {
+        unsafe {
+            let handle = CreateEventA(ptr::mut_null(), libc::FALSE, libc::FALSE,
+                                      ptr::null());
+            (handle, handle)
+        }
+    }
+
+    pub fn signal(handle: HANDLE) {
+        assert!(unsafe { SetEvent(handle) != 0 });
+    }
+
+    pub fn close(handle: HANDLE) {
+        assert!(unsafe { CloseHandle(handle) != 0 });
+    }
+
+    extern "system" {
+        fn CreateEventA(lpSecurityAttributes: LPSECURITY_ATTRIBUTES,
+                        bManualReset: BOOL,
+                        bInitialState: BOOL,
+                        lpName: LPCSTR) -> HANDLE;
+        fn SetEvent(hEvent: HANDLE) -> BOOL;
+    }
+}
diff --git a/src/libnative/io/mod.rs b/src/libnative/io/mod.rs
index f2c2c66e142..a9aca656319 100644
--- a/src/libnative/io/mod.rs
+++ b/src/libnative/io/mod.rs
@@ -40,6 +40,8 @@ use ai = std::io::net::addrinfo;
 pub use self::file::FileDesc;
 pub use self::process::Process;
 
+mod helper_thread;
+
 // Native I/O implementations
 pub mod addrinfo;
 pub mod net;
@@ -75,8 +77,6 @@ pub mod pipe;
 #[cfg(unix)]    #[path = "c_unix.rs"]  mod c;
 #[cfg(windows)] #[path = "c_win32.rs"] mod c;
 
-mod timer_helper;
-
 pub type IoResult<T> = Result<T, IoError>;
 
 fn unimpl() -> IoError {
diff --git a/src/libnative/io/process.rs b/src/libnative/io/process.rs
index 14ea1f12a5c..799db64e688 100644
--- a/src/libnative/io/process.rs
+++ b/src/libnative/io/process.rs
@@ -8,20 +8,27 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use std::io;
 use libc::{pid_t, c_void, c_int};
 use libc;
+use std::io;
+use std::mem;
 use std::os;
 use std::ptr;
 use std::rt::rtio;
 use p = std::io::process;
 
+
 use super::IoResult;
 use super::file;
+use super::util;
 
-#[cfg(windows)] use std::mem;
 #[cfg(windows)] use std::strbuf::StrBuf;
-#[cfg(not(windows))] use super::retry;
+#[cfg(unix)] use super::c;
+#[cfg(unix)] use super::retry;
+#[cfg(unix)] use io::helper_thread::Helper;
+
+#[cfg(unix)]
+helper_init!(static mut HELPER: Helper<Req>)
 
 /**
  * A value representing a child process.
@@ -44,6 +51,14 @@ pub struct Process {
 
     /// Manually delivered signal
     exit_signal: Option<int>,
+
+    /// Deadline after which wait() will return
+    deadline: u64,
+}
+
+#[cfg(unix)]
+enum Req {
+    NewChild(libc::pid_t, Sender<p::ProcessExit>, u64),
 }
 
 impl Process {
@@ -116,6 +131,7 @@ impl Process {
                         handle: res.handle,
                         exit_code: None,
                         exit_signal: None,
+                        deadline: 0,
                     },
                     ret_io))
             }
@@ -131,11 +147,15 @@ impl Process {
 impl rtio::RtioProcess for Process {
     fn id(&self) -> pid_t { self.pid }
 
-    fn wait(&mut self) -> p::ProcessExit {
+    fn set_timeout(&mut self, timeout: Option<u64>) {
+        self.deadline = timeout.map(|i| i + ::io::timer::now()).unwrap_or(0);
+    }
+
+    fn wait(&mut self) -> IoResult<p::ProcessExit> {
         match self.exit_code {
-            Some(code) => code,
+            Some(code) => Ok(code),
             None => {
-                let code = waitpid(self.pid);
+                let code = try!(waitpid(self.pid, self.deadline));
                 // On windows, waitpid will never return a signal. If a signal
                 // was successfully delivered to the process, however, we can
                 // consider it as having died via a signal.
@@ -145,7 +165,7 @@ impl rtio::RtioProcess for Process {
                     Some(..) => code,
                 };
                 self.exit_code = Some(code);
-                code
+                Ok(code)
             }
         }
     }
@@ -762,61 +782,301 @@ fn translate_status(status: c_int) -> p::ProcessExit {
  * operate on a none-existent process or, even worse, on a newer process
  * with the same id.
  */
-fn waitpid(pid: pid_t) -> p::ProcessExit {
-    return waitpid_os(pid);
-
-    #[cfg(windows)]
-    fn waitpid_os(pid: pid_t) -> p::ProcessExit {
-        use libc::types::os::arch::extra::DWORD;
-        use libc::consts::os::extra::{
-            SYNCHRONIZE,
-            PROCESS_QUERY_INFORMATION,
-            FALSE,
-            STILL_ACTIVE,
-            INFINITE,
-            WAIT_FAILED
-        };
-        use libc::funcs::extra::kernel32::{
-            OpenProcess,
-            GetExitCodeProcess,
-            CloseHandle,
-            WaitForSingleObject
-        };
+#[cfg(windows)]
+fn waitpid(pid: pid_t, deadline: u64) -> IoResult<p::ProcessExit> {
+    use libc::types::os::arch::extra::DWORD;
+    use libc::consts::os::extra::{
+        SYNCHRONIZE,
+        PROCESS_QUERY_INFORMATION,
+        FALSE,
+        STILL_ACTIVE,
+        INFINITE,
+        WAIT_TIMEOUT,
+        WAIT_OBJECT_0,
+    };
+    use libc::funcs::extra::kernel32::{
+        OpenProcess,
+        GetExitCodeProcess,
+        CloseHandle,
+        WaitForSingleObject,
+    };
 
-        unsafe {
+    unsafe {
+        let process = OpenProcess(SYNCHRONIZE | PROCESS_QUERY_INFORMATION,
+                                  FALSE,
+                                  pid as DWORD);
+        if process.is_null() {
+            return Err(super::last_error())
+        }
 
-            let process = OpenProcess(SYNCHRONIZE | PROCESS_QUERY_INFORMATION,
-                                      FALSE,
-                                      pid as DWORD);
-            if process.is_null() {
-                fail!("failure in OpenProcess: {}", os::last_os_error());
+        loop {
+            let mut status = 0;
+            if GetExitCodeProcess(process, &mut status) == FALSE {
+                let err = Err(super::last_error());
+                assert!(CloseHandle(process) != 0);
+                return err;
             }
-
-            loop {
-                let mut status = 0;
-                if GetExitCodeProcess(process, &mut status) == FALSE {
-                    assert!(CloseHandle(process) != 0);
-                    fail!("failure in GetExitCodeProcess: {}", os::last_os_error());
-                }
-                if status != STILL_ACTIVE {
+            if status != STILL_ACTIVE {
+                assert!(CloseHandle(process) != 0);
+                return Ok(p::ExitStatus(status as int));
+            }
+            let interval = if deadline == 0 {
+                INFINITE
+            } else {
+                let now = ::io::timer::now();
+                if deadline < now {0} else {(deadline - now) as u32}
+            };
+            match WaitForSingleObject(process, interval) {
+                WAIT_OBJECT_0 => {}
+                WAIT_TIMEOUT => {
                     assert!(CloseHandle(process) != 0);
-                    return p::ExitStatus(status as int);
+                    return Err(util::timeout("process wait timed out"))
                 }
-                if WaitForSingleObject(process, INFINITE) == WAIT_FAILED {
+                _ => {
+                    let err = Err(super::last_error());
                     assert!(CloseHandle(process) != 0);
-                    fail!("failure in WaitForSingleObject: {}", os::last_os_error());
+                    return err
                 }
             }
         }
     }
+}
 
-    #[cfg(unix)]
-    fn waitpid_os(pid: pid_t) -> p::ProcessExit {
-        use libc::funcs::posix01::wait;
-        let mut status = 0 as c_int;
-        match retry(|| unsafe { wait::waitpid(pid, &mut status, 0) }) {
+#[cfg(unix)]
+fn waitpid(pid: pid_t, deadline: u64) -> IoResult<p::ProcessExit> {
+    use std::cmp;
+    use std::comm;
+
+    static mut WRITE_FD: libc::c_int = 0;
+
+    let mut status = 0 as c_int;
+    if deadline == 0 {
+        return match retry(|| unsafe { c::waitpid(pid, &mut status, 0) }) {
             -1 => fail!("unknown waitpid error: {}", super::last_error()),
-            _ => translate_status(status),
+            _ => Ok(translate_status(status)),
+        }
+    }
+
+    // On unix, wait() and its friends have no timeout parameters, so there is
+    // no way to time out a thread in wait(). From some googling and some
+    // thinking, it appears that there are a few ways to handle timeouts in
+    // wait(), but the only real reasonable one for a multi-threaded program is
+    // to listen for SIGCHLD.
+    //
+    // With this in mind, the waiting mechanism with a timeout barely uses
+    // waitpid() at all. There are a few times that waitpid() is invoked with
+    // WNOHANG, but otherwise all the necessary blocking is done by waiting for
+    // a SIGCHLD to arrive (and that blocking has a timeout). Note, however,
+    // that waitpid() is still used to actually reap the child.
+    //
+    // Signal handling is super tricky in general, and this is no exception. Due
+    // to the async nature of SIGCHLD, we use the self-pipe trick to transmit
+    // data out of the signal handler to the rest of the application. The first
+    // idea would be to have each thread waiting with a timeout to read this
+    // output file descriptor, but a write() is akin to a signal(), not a
+    // broadcast(), so it would only wake up one thread, and possibly the wrong
+    // thread. Hence a helper thread is used.
+    //
+    // The helper thread here is responsible for farming requests for a
+    // waitpid() with a timeout, and then processing all of the wait requests.
+    // By guaranteeing that only this helper thread is reading half of the
+    // self-pipe, we're sure that we'll never lose a SIGCHLD. This helper thread
+    // is also responsible for select() to wait for incoming messages or
+    // incoming SIGCHLD messages, along with passing an appropriate timeout to
+    // select() to wake things up as necessary.
+    //
+    // The ordering of the following statements is also very purposeful. First,
+    // we must be guaranteed that the helper thread is booted and available to
+    // receive SIGCHLD signals, and then we must also ensure that we do a
+    // nonblocking waitpid() at least once before we go ask the sigchld helper.
+    // This prevents the race where the child exits, we boot the helper, and
+    // then we ask for the child's exit status (never seeing a sigchld).
+    //
+    // The actual communication between the helper thread and this thread is
+    // quite simple, just a channel moving data around.
+
+    unsafe { HELPER.boot(register_sigchld, waitpid_helper) }
+
+    match waitpid_nowait(pid) {
+        Some(ret) => return Ok(ret),
+        None => {}
+    }
+
+    let (tx, rx) = channel();
+    unsafe { HELPER.send(NewChild(pid, tx, deadline)); }
+    return match rx.recv_opt() {
+        Ok(e) => Ok(e),
+        Err(()) => Err(util::timeout("wait timed out")),
+    };
+
+    // Register a new SIGCHLD handler, returning the reading half of the
+    // self-pipe plus the old handler registered (return value of sigaction).
+    fn register_sigchld() -> (libc::c_int, c::sigaction) {
+        unsafe {
+            let mut old: c::sigaction = mem::init();
+            let mut new: c::sigaction = mem::init();
+            new.sa_handler = sigchld_handler;
+            new.sa_flags = c::SA_NOCLDSTOP;
+            assert_eq!(c::sigaction(c::SIGCHLD, &new, &mut old), 0);
+
+            let mut pipes = [0, ..2];
+            assert_eq!(libc::pipe(pipes.as_mut_ptr()), 0);
+            util::set_nonblocking(pipes[0], true).unwrap();
+            util::set_nonblocking(pipes[1], true).unwrap();
+            WRITE_FD = pipes[1];
+            (pipes[0], old)
+        }
+    }
+
+    // Helper thread for processing SIGCHLD messages
+    fn waitpid_helper(input: libc::c_int,
+                      messages: Receiver<Req>,
+                      (read_fd, old): (libc::c_int, c::sigaction)) {
+        util::set_nonblocking(input, true).unwrap();
+        let mut set: c::fd_set = unsafe { mem::init() };
+        let mut tv: libc::timeval;
+        let mut active = Vec::<(libc::pid_t, Sender<p::ProcessExit>, u64)>::new();
+        let max = cmp::max(input, read_fd) + 1;
+
+        'outer: loop {
+            // Figure out the timeout of our syscall-to-happen. If we're waiting
+            // for some processes, then they'll have a timeout, otherwise we
+            // wait indefinitely for a message to arrive.
+            //
+            // FIXME: sure would be nice to not have to scan the entire array
+            let min = active.iter().map(|a| *a.ref2()).enumerate().min_by(|p| {
+                p.val1()
+            });
+            let (p, idx) = match min {
+                Some((idx, deadline)) => {
+                    let now = ::io::timer::now();
+                    let ms = if now < deadline {deadline - now} else {0};
+                    tv = util::ms_to_timeval(ms);
+                    (&tv as *_, idx)
+                }
+                None => (ptr::null(), -1),
+            };
+
+            // Wait for something to happen
+            c::fd_set(&mut set, input);
+            c::fd_set(&mut set, read_fd);
+            match unsafe { c::select(max, &set, ptr::null(), ptr::null(), p) } {
+                // interrupted, retry
+                -1 if os::errno() == libc::EINTR as int => continue,
+
+                // We read something, break out and process
+                1 | 2 => {}
+
+                // Timeout, the pending request is removed
+                0 => {
+                    drop(active.remove(idx));
+                    continue
+                }
+
+                n => fail!("error in select {} ({})", os::errno(), n),
+            }
+
+            // Process any pending messages
+            if drain(input) {
+                loop {
+                    match messages.try_recv() {
+                        Ok(NewChild(pid, tx, deadline)) => {
+                            active.push((pid, tx, deadline));
+                        }
+                        Err(comm::Disconnected) => {
+                            assert!(active.len() == 0);
+                            break 'outer;
+                        }
+                        Err(comm::Empty) => break,
+                    }
+                }
+            }
+
+            // If a child exited (somehow received SIGCHLD), then poll all
+            // children to see if any of them exited.
+            //
+            // We also attempt to be responsible netizens when dealing with
+            // SIGCHLD by invoking any previous SIGCHLD handler instead of just
+            // ignoring any previous SIGCHLD handler. Note that we don't provide
+            // a 1:1 mapping of our handler invocations to the previous handler
+            // invocations because we drain the `read_fd` entirely. This is
+            // probably OK because the kernel is already allowed to coalesce
+            // simultaneous signals, we're just doing some extra coalescing.
+            //
+            // Another point of note is that this likely runs the signal handler
+            // on a different thread than the one that received the signal. I
+            // *think* this is ok at this time.
+            //
+            // The main reason for doing this is to allow stdtest to run native
+            // tests as well. Both libgreen and libnative are running around
+            // with process timeouts, but libgreen should get there first
+            // (currently libuv doesn't handle old signal handlers).
+            if drain(read_fd) {
+                let i: uint = unsafe { mem::transmute(old.sa_handler) };
+                if i != 0 {
+                    assert!(old.sa_flags & c::SA_SIGINFO == 0);
+                    (old.sa_handler)(c::SIGCHLD);
+                }
+
+                // FIXME: sure would be nice to not have to scan the entire
+                //        array...
+                active.retain(|&(pid, ref tx, _)| {
+                    match waitpid_nowait(pid) {
+                        Some(msg) => { tx.send(msg); false }
+                        None => true,
+                    }
+                });
+            }
+        }
+
+        // Once this helper thread is done, we re-register the old sigchld
+        // handler and close our intermediate file descriptors.
+        unsafe {
+            assert_eq!(c::sigaction(c::SIGCHLD, &old, ptr::mut_null()), 0);
+            let _ = libc::close(read_fd);
+            let _ = libc::close(WRITE_FD);
+            WRITE_FD = -1;
+        }
+    }
+
+    // Drain all pending data from the file descriptor, returning if any data
+    // could be drained. This requires that the file descriptor is in
+    // nonblocking mode.
+    fn drain(fd: libc::c_int) -> bool {
+        let mut ret = false;
+        loop {
+            let mut buf = [0u8, ..1];
+            match unsafe {
+                libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void,
+                           buf.len() as libc::size_t)
+            } {
+                n if n > 0 => { ret = true; }
+                0 => return true,
+                -1 if util::wouldblock() => return ret,
+                n => fail!("bad read {} ({})", os::last_os_error(), n),
+            }
+        }
+    }
+
+    // Signal handler for SIGCHLD signals, must be async-signal-safe!
+    //
+    // This function will write to the writing half of the "self pipe" to wake
+    // up the helper thread if it's waiting. Note that this write must be
+    // nonblocking because if it blocks and the reader is the thread we
+    // interrupted, then we'll deadlock.
+    //
+    // When writing, if the write returns EWOULDBLOCK then we choose to ignore
+    // it. At that point we're guaranteed that there's something in the pipe
+    // which will wake up the other end at some point, so we just allow this
+    // signal to be coalesced with the pending signals on the pipe.
+    extern fn sigchld_handler(_signum: libc::c_int) {
+        let mut msg = 1;
+        match unsafe {
+            libc::write(WRITE_FD, &mut msg as *mut _ as *libc::c_void, 1)
+        } {
+            1 => {}
+            -1 if util::wouldblock() => {} // see above comments
+            n => fail!("bad error on write fd: {} {}", n, os::errno()),
         }
     }
 }
@@ -830,10 +1090,9 @@ fn waitpid_nowait(pid: pid_t) -> Option<p::ProcessExit> {
 
     #[cfg(unix)]
     fn waitpid_os(pid: pid_t) -> Option<p::ProcessExit> {
-        use libc::funcs::posix01::wait;
         let mut status = 0 as c_int;
         match retry(|| unsafe {
-            wait::waitpid(pid, &mut status, libc::WNOHANG)
+            c::waitpid(pid, &mut status, c::WNOHANG)
         }) {
             n if n == pid => Some(translate_status(status)),
             0 => None,
diff --git a/src/libnative/io/timer_helper.rs b/src/libnative/io/timer_helper.rs
deleted file mode 100644
index 95b2620f3c7..00000000000
--- a/src/libnative/io/timer_helper.rs
+++ /dev/null
@@ -1,149 +0,0 @@
-// Copyright 2013-2014 The Rust Project Developers. See the COPYRIGHT
-// file at the top-level directory of this distribution and at
-// http://rust-lang.org/COPYRIGHT.
-//
-// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
-// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
-// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
-// option. This file may not be copied, modified, or distributed
-// except according to those terms.
-
-//! Implementation of the helper thread for the timer module
-//!
-//! This module contains the management necessary for the timer worker thread.
-//! This thread is responsible for performing the send()s on channels for timers
-//! that are using channels instead of a blocking call.
-//!
-//! The timer thread is lazily initialized, and it's shut down via the
-//! `shutdown` function provided. It must be maintained as an invariant that
-//! `shutdown` is only called when the entire program is finished. No new timers
-//! can be created in the future and there must be no active timers at that
-//! time.
-
-use std::mem;
-use std::rt::bookkeeping;
-use std::rt;
-use std::unstable::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT};
-
-use io::timer::{Req, Shutdown};
-use task;
-
-// You'll note that these variables are *not* protected by a lock. These
-// variables are initialized with a Once before any Timer is created and are
-// only torn down after everything else has exited. This means that these
-// variables are read-only during use (after initialization) and both of which
-// are safe to use concurrently.
-static mut HELPER_CHAN: *mut Sender<Req> = 0 as *mut Sender<Req>;
-static mut HELPER_SIGNAL: imp::signal = 0 as imp::signal;
-
-static mut TIMER_HELPER_EXIT: StaticNativeMutex = NATIVE_MUTEX_INIT;
-
-pub fn boot(helper: fn(imp::signal, Receiver<Req>)) {
-    static mut LOCK: StaticNativeMutex = NATIVE_MUTEX_INIT;
-    static mut INITIALIZED: bool = false;
-
-    unsafe {
-        let mut _guard = LOCK.lock();
-        if !INITIALIZED {
-            let (tx, rx) = channel();
-            // promote this to a shared channel
-            drop(tx.clone());
-            HELPER_CHAN = mem::transmute(box tx);
-            let (receive, send) = imp::new();
-            HELPER_SIGNAL = send;
-
-            task::spawn(proc() {
-                bookkeeping::decrement();
-                helper(receive, rx);
-                TIMER_HELPER_EXIT.lock().signal()
-            });
-
-            rt::at_exit(proc() { shutdown() });
-            INITIALIZED = true;
-        }
-    }
-}
-
-pub fn send(req: Req) {
-    unsafe {
-        assert!(!HELPER_CHAN.is_null());
-        (*HELPER_CHAN).send(req);
-        imp::signal(HELPER_SIGNAL);
-    }
-}
-
-fn shutdown() {
-    // Request a shutdown, and then wait for the task to exit
-    unsafe {
-        let guard = TIMER_HELPER_EXIT.lock();
-        send(Shutdown);
-        guard.wait();
-        drop(guard);
-        TIMER_HELPER_EXIT.destroy();
-    }
-
-
-    // Clean up after ther helper thread
-    unsafe {
-        imp::close(HELPER_SIGNAL);
-        let _chan: Box<Sender<Req>> = mem::transmute(HELPER_CHAN);
-        HELPER_CHAN = 0 as *mut Sender<Req>;
-        HELPER_SIGNAL = 0 as imp::signal;
-    }
-}
-
-#[cfg(unix)]
-mod imp {
-    use libc;
-    use std::os;
-
-    use io::file::FileDesc;
-
-    pub type signal = libc::c_int;
-
-    pub fn new() -> (signal, signal) {
-        let pipe = os::pipe();
-        (pipe.input, pipe.out)
-    }
-
-    pub fn signal(fd: libc::c_int) {
-        FileDesc::new(fd, false).inner_write([0]).unwrap();
-    }
-
-    pub fn close(fd: libc::c_int) {
-        let _fd = FileDesc::new(fd, true);
-    }
-}
-
-#[cfg(windows)]
-mod imp {
-    use libc::{BOOL, LPCSTR, HANDLE, LPSECURITY_ATTRIBUTES, CloseHandle};
-    use std::ptr;
-    use libc;
-
-    pub type signal = HANDLE;
-
-    pub fn new() -> (HANDLE, HANDLE) {
-        unsafe {
-            let handle = CreateEventA(ptr::mut_null(), libc::FALSE, libc::FALSE,
-                                      ptr::null());
-            (handle, handle)
-        }
-    }
-
-    pub fn signal(handle: HANDLE) {
-        assert!(unsafe { SetEvent(handle) != 0 });
-    }
-
-    pub fn close(handle: HANDLE) {
-        assert!(unsafe { CloseHandle(handle) != 0 });
-    }
-
-    extern "system" {
-        fn CreateEventA(lpSecurityAttributes: LPSECURITY_ATTRIBUTES,
-                        bManualReset: BOOL,
-                        bInitialState: BOOL,
-                        lpName: LPCSTR) -> HANDLE;
-        fn SetEvent(hEvent: HANDLE) -> BOOL;
-    }
-}
diff --git a/src/libnative/io/timer_unix.rs b/src/libnative/io/timer_unix.rs
index e008e6fb9e9..2c5b7984827 100644
--- a/src/libnative/io/timer_unix.rs
+++ b/src/libnative/io/timer_unix.rs
@@ -52,11 +52,14 @@ use std::os;
 use std::ptr;
 use std::rt::rtio;
 use std::sync::atomics;
+use std::comm;
 
 use io::IoResult;
 use io::c;
 use io::file::FileDesc;
-use io::timer_helper;
+use io::helper_thread::Helper;
+
+helper_init!(static mut HELPER: Helper<Req>)
 
 pub struct Timer {
     id: uint,
@@ -79,9 +82,6 @@ pub enum Req {
     // Remove a timer based on its id and then send it back on the channel
     // provided
     RemoveTimer(uint, Sender<Box<Inner>>),
-
-    // Shut down the loop and then ACK this channel once it's shut down
-    Shutdown,
 }
 
 // returns the current time (in milliseconds)
@@ -93,7 +93,7 @@ pub fn now() -> u64 {
     }
 }
 
-fn helper(input: libc::c_int, messages: Receiver<Req>) {
+fn helper(input: libc::c_int, messages: Receiver<Req>, _: ()) {
     let mut set: c::fd_set = unsafe { mem::init() };
 
     let mut fd = FileDesc::new(input, true);
@@ -163,7 +163,7 @@ fn helper(input: libc::c_int, messages: Receiver<Req>) {
             1 => {
                 loop {
                     match messages.try_recv() {
-                        Ok(Shutdown) => {
+                        Err(comm::Disconnected) => {
                             assert!(active.len() == 0);
                             break 'outer;
                         }
@@ -202,7 +202,7 @@ fn helper(input: libc::c_int, messages: Receiver<Req>) {
 
 impl Timer {
     pub fn new() -> IoResult<Timer> {
-        timer_helper::boot(helper);
+        unsafe { HELPER.boot(|| {}, helper); }
 
         static mut ID: atomics::AtomicUint = atomics::INIT_ATOMIC_UINT;
         let id = unsafe { ID.fetch_add(1, atomics::Relaxed) };
@@ -235,7 +235,7 @@ impl Timer {
             Some(i) => i,
             None => {
                 let (tx, rx) = channel();
-                timer_helper::send(RemoveTimer(self.id, tx));
+                unsafe { HELPER.send(RemoveTimer(self.id, tx)); }
                 rx.recv()
             }
         }
@@ -261,7 +261,7 @@ impl rtio::RtioTimer for Timer {
         inner.interval = msecs;
         inner.target = now + msecs;
 
-        timer_helper::send(NewTimer(inner));
+        unsafe { HELPER.send(NewTimer(inner)); }
         return rx;
     }
 
@@ -275,7 +275,7 @@ impl rtio::RtioTimer for Timer {
         inner.interval = msecs;
         inner.target = now + msecs;
 
-        timer_helper::send(NewTimer(inner));
+        unsafe { HELPER.send(NewTimer(inner)); }
         return rx;
     }
 }
diff --git a/src/libnative/io/timer_win32.rs b/src/libnative/io/timer_win32.rs
index 15e6e62421a..e7130de05c2 100644
--- a/src/libnative/io/timer_win32.rs
+++ b/src/libnative/io/timer_win32.rs
@@ -23,10 +23,13 @@
 use libc;
 use std::ptr;
 use std::rt::rtio;
+use std::comm;
 
-use io::timer_helper;
+use io::helper_thread::Helper;
 use io::IoResult;
 
+helper_init!(static mut HELPER: Helper<Req>)
+
 pub struct Timer {
     obj: libc::HANDLE,
     on_worker: bool,
@@ -35,10 +38,9 @@ pub struct Timer {
 pub enum Req {
     NewTimer(libc::HANDLE, Sender<()>, bool),
     RemoveTimer(libc::HANDLE, Sender<()>),
-    Shutdown,
 }
 
-fn helper(input: libc::HANDLE, messages: Receiver<Req>) {
+fn helper(input: libc::HANDLE, messages: Receiver<Req>, _: ()) {
     let mut objs = vec![input];
     let mut chans = vec![];
 
@@ -67,12 +69,12 @@ fn helper(input: libc::HANDLE, messages: Receiver<Req>) {
                             None => {}
                         }
                     }
-                    Ok(Shutdown) => {
+                    Err(comm::Disconnected) => {
                         assert_eq!(objs.len(), 1);
                         assert_eq!(chans.len(), 0);
                         break 'outer;
                     }
-                    _ => break
+                    Err(..) => break
                 }
             }
         } else {
@@ -102,7 +104,7 @@ pub fn now() -> u64 {
 
 impl Timer {
     pub fn new() -> IoResult<Timer> {
-        timer_helper::boot(helper);
+        unsafe { HELPER.boot(|| {}, helper) }
 
         let obj = unsafe {
             imp::CreateWaitableTimerA(ptr::mut_null(), 0, ptr::null())
@@ -124,7 +126,7 @@ impl Timer {
         if !self.on_worker { return }
 
         let (tx, rx) = channel();
-        timer_helper::send(RemoveTimer(self.obj, tx));
+        unsafe { HELPER.send(RemoveTimer(self.obj, tx)) }
         rx.recv();
 
         self.on_worker = false;
@@ -157,7 +159,7 @@ impl rtio::RtioTimer for Timer {
                                   ptr::mut_null(), 0)
         }, 1);
 
-        timer_helper::send(NewTimer(self.obj, tx, true));
+        unsafe { HELPER.send(NewTimer(self.obj, tx, true)) }
         self.on_worker = true;
         return rx;
     }
@@ -173,7 +175,7 @@ impl rtio::RtioTimer for Timer {
                                   ptr::null(), ptr::mut_null(), 0)
         }, 1);
 
-        timer_helper::send(NewTimer(self.obj, tx, false));
+        unsafe { HELPER.send(NewTimer(self.obj, tx, false)) }
         self.on_worker = true;
 
         return rx;
diff --git a/src/libnative/lib.rs b/src/libnative/lib.rs
index 05cf415ec78..8ba06133369 100644
--- a/src/libnative/lib.rs
+++ b/src/libnative/lib.rs
@@ -55,6 +55,7 @@
 // NB this crate explicitly does *not* allow glob imports, please seriously
 //    consider whether they're needed before adding that feature here (the
 //    answer is that you don't need them)
+#![feature(macro_rules)]
 
 extern crate libc;