commit 50973e0734b50a6a99f944d35b53d82c63a852ff
Author: Alex Crichton <alex@alexcrichton.com>
Date:   Tue Sep 6 23:00:17 2016 -0700

    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..268883199
--- /dev/null
+++ b/Cargo.toml
@@ -0,0 +1,13 @@
+[package]
+name = "tokio-signal"
+version = "0.1.0"
+authors = ["Alex Crichton <alex@alexcrichton.com>"]
+
+[dependencies]
+tokio-core = { git = "https://github.com/tokio-rs/tokio-core" }
+futures = { git = "https://github.com/alexcrichton/futures-rs" }
+
+[target.'cfg(unix)'.dependencies]
+tokio-uds = { git = "https://github.com/tokio-rs/tokio-uds" }
+libc = "0.2"
+mio = "0.6"
diff --git a/src/lib.rs b/src/lib.rs
new file mode 100644
index 000000000..2d304ecd0
--- /dev/null
+++ b/src/lib.rs
@@ -0,0 +1,5 @@
+#[macro_use]
+extern crate futures;
+extern crate tokio_core;
+
+pub mod unix;
diff --git a/src/unix.rs b/src/unix.rs
new file mode 100644
index 000000000..7d1f30477
--- /dev/null
+++ b/src/unix.rs
@@ -0,0 +1,338 @@
+#![cfg(unix)]
+
+extern crate libc;
+extern crate mio;
+extern crate tokio_uds;
+
+use std::cell::RefCell;
+use std::io::{self, Write, Read};
+use std::mem;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Once, ONCE_INIT, Mutex};
+
+use futures::stream::{Stream, Fuse};
+use futures::{self, Future, Complete, Oneshot, Poll, Async};
+use self::libc::c_int;
+use self::tokio_uds::UnixStream;
+use tokio_core::io::IoFuture;
+use tokio_core::{LoopHandle, Sender, Receiver, ReadinessStream};
+
+static INIT: Once = ONCE_INIT;
+static mut GLOBAL_STATE: *mut GlobalState = 0 as *mut _;
+
+pub struct Signal {
+    signum: c_int,
+    reg: ReadinessStream<MyRegistration>,
+    _finished: Complete<()>,
+}
+
+struct GlobalState {
+    write: UnixStream,
+    tx: Mutex<Sender<Message>>,
+    signals: [GlobalSignalState; 32],
+}
+
+struct GlobalSignalState {
+    ready: AtomicBool,
+    prev: libc::sigaction,
+}
+
+enum Message {
+    NewSignal(c_int, Complete<io::Result<Signal>>),
+}
+
+struct DriverTask {
+    handle: LoopHandle,
+    read: UnixStream,
+    rx: Fuse<Receiver<Message>>,
+    signals: [SignalState; 32],
+}
+
+struct SignalState {
+    registered: bool,
+    tasks: Vec<(RefCell<Oneshot<()>>, mio::SetReadiness)>,
+}
+
+impl Signal {
+    // TODO: document coalescing (happens everywhere)
+    // TODO: document multiple event loops (first must stay alive)
+    pub fn new(signum: c_int, handle: &LoopHandle) -> IoFuture<Signal> {
+        let mut init = None;
+        INIT.call_once(|| {
+            init = Some(global_init(handle));
+        });
+        let new_signal = futures::lazy(move || {
+            let (tx, rx) = futures::oneshot();
+            let msg = Message::NewSignal(signum, tx);
+            let res = unsafe {
+                (*GLOBAL_STATE).tx.lock().unwrap().send(msg)
+            };
+            res.expect("failed to request a new signal stream, did the \
+                        first event loop go away?");
+            rx.then(|r| r.unwrap())
+        });
+        match init {
+            Some(init) => init.and_then(|()| new_signal).boxed(),
+            None => new_signal.boxed(),
+        }
+    }
+}
+
+impl Stream for Signal {
+    type Item = c_int;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Option<c_int>, io::Error> {
+        try_ready!(self.reg.poll_read());
+        self.reg.get_ref()
+                .inner.borrow()
+                .as_ref().unwrap().1
+                .set_readiness(mio::Ready::none())
+                .expect("failed to set readiness");
+        Ok(Async::Ready(Some(self.signum)))
+    }
+}
+
+fn global_init(handle: &LoopHandle) -> IoFuture<()> {
+    let handle = handle.clone();
+    let (tx, rx) = handle.clone().channel();
+    let io = rx.join(UnixStream::pair(handle.clone()));
+    io.map(move |(rx, (read, write))| {
+        unsafe {
+            let state = Box::new(GlobalState {
+                write: write,
+                signals: {
+                    fn new() -> GlobalSignalState {
+                        GlobalSignalState {
+                            ready: AtomicBool::new(false),
+                            prev: unsafe { mem::zeroed() },
+                        }
+                    }
+                    [
+                        new(), new(), new(), new(), new(), new(), new(), new(),
+                        new(), new(), new(), new(), new(), new(), new(), new(),
+                        new(), new(), new(), new(), new(), new(), new(), new(),
+                        new(), new(), new(), new(), new(), new(), new(), new(),
+                    ]
+                },
+                tx: Mutex::new(tx.clone()),
+            });
+            GLOBAL_STATE = Box::into_raw(state);
+
+            handle.clone().spawn(|_| {
+                DriverTask {
+                    handle: handle,
+                    rx: rx.fuse(),
+                    read: read,
+                    signals: {
+                        fn new() -> SignalState {
+                            SignalState { registered: false, tasks: Vec::new() }
+                        }
+                        [
+                            new(), new(), new(), new(), new(), new(), new(), new(),
+                            new(), new(), new(), new(), new(), new(), new(), new(),
+                            new(), new(), new(), new(), new(), new(), new(), new(),
+                            new(), new(), new(), new(), new(), new(), new(), new(),
+                        ]
+                    },
+                }
+            });
+        }
+    }).boxed()
+}
+
+impl Future for DriverTask {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<(), ()> {
+        self.check_signal_drops();
+        self.check_messages();
+        self.check_signals();
+
+        // TODO: when to finish this task?
+        Ok(Async::NotReady)
+    }
+}
+
+impl DriverTask {
+    fn check_signal_drops(&mut self) {
+        for signal in self.signals.iter_mut() {
+            signal.tasks.retain(|task| {
+                !task.0.borrow_mut().poll().is_err()
+            });
+        }
+    }
+
+    fn check_messages(&mut self) {
+        loop {
+            // Acquire the next message
+            let message = match self.rx.poll() {
+                Ok(Async::Ready(Some(e))) => e,
+                Ok(Async::Ready(None)) |
+                Ok(Async::NotReady) => break,
+                Err(e) => panic!("error on rx: {}", e),
+            };
+            let (sig, complete) = match message {
+                Message::NewSignal(sig, complete) => (sig, complete),
+            };
+
+            // If the signal's too large, then we return an error, otherwise we
+            // use this index to look at the signal slot.
+            //
+            // If the signal wasn't previously registered then we do so now.
+            let signal = match self.signals.get_mut(sig as usize) {
+                Some(signal) => signal,
+                None => {
+                    complete.complete(Err(io::Error::new(io::ErrorKind::Other,
+                                                         "signum too large")));
+                    continue
+                }
+            };
+            if !signal.registered {
+                unsafe {
+                    let mut new: libc::sigaction = mem::zeroed();
+                    new.sa_sigaction = handler as usize;
+                    new.sa_flags = libc::SA_RESTART | libc::SA_SIGINFO;
+                    let mut prev = mem::zeroed();
+                    if libc::sigaction(sig, &new, &mut prev) != 0 {
+                        complete.complete(Err(io::Error::last_os_error()));
+                        continue
+                    }
+                    signal.registered = true;
+                }
+            }
+
+            // Acquire the (registration, set_readiness) pair by... assuming
+            // we're on the event loop (true because of the spawn above).
+            let reg = MyRegistration { inner: RefCell::new(None) };
+            let mut new = ReadinessStream::new(self.handle.clone(), reg);
+            let reg = match new.poll() {
+                Ok(Async::Ready(reg)) => reg,
+                Ok(Async::NotReady) => panic!("should be on event loop"),
+                Err(e) => {
+                    complete.complete(Err(e));
+                    continue
+                }
+            };
+
+            // Create the `Signal` to pass back and then also keep a handle to
+            // the `SetReadiness` for ourselves internally.
+            let (tx, rx) = futures::oneshot();
+            let ready = reg.get_ref().inner.borrow_mut().as_mut().unwrap().1.clone();
+            complete.complete(Ok(Signal {
+                signum: sig,
+                reg: reg,
+                _finished: tx,
+            }));
+            signal.tasks.push((RefCell::new(rx), ready));
+        }
+    }
+
+    fn check_signals(&mut self) {
+        // Drain all data from the pipe
+        let mut buf = [0; 32];
+        let mut any = false;
+        loop {
+            match self.read.read(&mut buf) {
+                Ok(0) => {  // EOF == something happened
+                    any = true;
+                    break
+                }
+                Ok(..) => any = true,   // data read, but keep draining
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
+                Err(e) => panic!("bad read: {}", e),
+            }
+        }
+
+        // If nothing happened, no need to check the signals
+        if !any {
+            return
+        }
+
+        for (i, slot) in self.signals.iter().enumerate() {
+            // No need to go farther if we haven't even registered a signal
+            if !slot.registered {
+                continue
+            }
+
+            // See if this signal actually happened since we last checked
+            unsafe {
+                if !(*GLOBAL_STATE).signals[i].ready.swap(false, Ordering::SeqCst) {
+                    continue
+                }
+            }
+
+            // Wake up all the tasks waiting on this signal
+            for task in slot.tasks.iter() {
+                task.1.set_readiness(mio::Ready::readable())
+                      .expect("failed to set readiness");
+            }
+        }
+    }
+}
+
+extern fn handler(signum: c_int,
+                  info: *mut libc::siginfo_t,
+                  ptr: *mut libc::c_void) {
+    type FnSigaction = extern fn(c_int, *mut libc::siginfo_t, *mut libc::c_void);
+    type FnHandler = extern fn(c_int);
+
+    unsafe {
+        let state = match (*GLOBAL_STATE).signals.get(signum as usize) {
+            Some(state) => state,
+            None => return,
+        };
+
+        if !state.ready.swap(true, Ordering::SeqCst) {
+            match (&(*GLOBAL_STATE).write).write(&[1]) {
+                Ok(..) => {}
+                Err(e) => {
+                    if e.kind() != io::ErrorKind::WouldBlock {
+                        panic!("bad error on write fd: {}", e)
+                    }
+                }
+            }
+        }
+
+        let fnptr = state.prev.sa_sigaction;
+        if fnptr == 0 || fnptr == libc::SIG_DFL || fnptr == libc::SIG_IGN {
+            return
+        }
+        if state.prev.sa_flags & libc::SA_SIGINFO == 0 {
+            let action = mem::transmute::<usize, FnHandler>(fnptr);
+            action(signum)
+        } else {
+            let action = mem::transmute::<usize, FnSigaction>(fnptr);
+            action(signum, info, ptr)
+        }
+    }
+}
+
+struct MyRegistration {
+    inner: RefCell<Option<(mio::Registration, mio::SetReadiness)>>,
+}
+
+impl mio::Evented for MyRegistration {
+    fn register(&self,
+                poll: &mio::Poll,
+                token: mio::Token,
+                events: mio::Ready,
+                opts: mio::PollOpt) -> io::Result<()> {
+        let reg = mio::Registration::new(poll, token, events, opts);
+        *self.inner.borrow_mut() = Some(reg);
+        Ok(())
+    }
+
+    fn reregister(&self,
+                  _poll: &mio::Poll,
+                  _token: mio::Token,
+                  _events: mio::Ready,
+                  _opts: mio::PollOpt) -> io::Result<()> {
+        Ok(())
+    }
+
+    fn deregister(&self, _poll: &mio::Poll) -> io::Result<()> {
+        Ok(())
+    }
+}
diff --git a/tests/signal.rs b/tests/signal.rs
new file mode 100644
index 000000000..ef7ccb771
--- /dev/null
+++ b/tests/signal.rs
@@ -0,0 +1,97 @@
+#![cfg(unix)]
+
+extern crate futures;
+extern crate libc;
+extern crate tokio_core;
+extern crate tokio_signal;
+
+use std::sync::mpsc::channel;
+use std::sync::{Once, ONCE_INIT, Mutex, MutexGuard};
+use std::thread;
+use std::time::Duration;
+
+use futures::Future;
+use futures::stream::Stream;
+use tokio_core::Loop;
+use tokio_signal::unix::Signal;
+
+static INIT: Once = ONCE_INIT;
+static mut LOCK: *mut Mutex<()> = 0 as *mut _;
+
+fn lock() -> MutexGuard<'static, ()> {
+    unsafe {
+        INIT.call_once(|| {
+            LOCK = Box::into_raw(Box::new(Mutex::new(())));
+            let (tx, rx) = channel();
+            thread::spawn(move || {
+                let mut lp = Loop::new().unwrap();
+                let handle = lp.handle();
+                let _signal = lp.run(Signal::new(libc::SIGALRM, &handle)).unwrap();
+                tx.send(()).unwrap();
+                drop(lp.run(futures::empty::<(), ()>()));
+            });
+            rx.recv().unwrap();
+        });
+        (*LOCK).lock().unwrap()
+    }
+}
+
+#[test]
+fn simple() {
+    let _lock = lock();
+
+    let mut lp = Loop::new().unwrap();
+    let handle = lp.handle();
+    let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap();
+    unsafe {
+        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
+    }
+    lp.run(signal.into_future()).ok().unwrap();
+}
+
+#[test]
+fn notify_both() {
+    let _lock = lock();
+
+    let mut lp = Loop::new().unwrap();
+    let handle = lp.handle();
+    let signal1 = lp.run(Signal::new(libc::SIGUSR2, &handle)).unwrap();
+    let signal2 = lp.run(Signal::new(libc::SIGUSR2, &handle)).unwrap();
+    unsafe {
+        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR2), 0);
+    }
+    lp.run(signal1.into_future().join(signal2.into_future())).ok().unwrap();
+}
+
+#[test]
+fn drop_then_get_a_signal() {
+    let _lock = lock();
+
+    let mut lp = Loop::new().unwrap();
+    let handle = lp.handle();
+    let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap();
+    drop(signal);
+    unsafe {
+        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
+    }
+    let timeout = lp.handle().timeout(Duration::from_millis(1));
+    lp.run(timeout.and_then(|t| t)).unwrap();
+}
+
+#[test]
+fn twice() {
+    let _lock = lock();
+
+    let mut lp = Loop::new().unwrap();
+    let handle = lp.handle();
+    let signal = lp.run(Signal::new(libc::SIGUSR1, &handle)).unwrap();
+    unsafe {
+        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
+    }
+    let (num, signal) = lp.run(signal.into_future()).ok().unwrap();
+    assert_eq!(num, Some(libc::SIGUSR1));
+    unsafe {
+        assert_eq!(libc::kill(libc::getpid(), libc::SIGUSR1), 0);
+    }
+    lp.run(signal.into_future()).ok().unwrap();
+}