corcovado 0.2.5

use std::collections::HashMap;
#[cfg(not(target_os = "netbsd"))]
use std::os::raw::{c_int, c_short};
use std::os::unix::io::AsRawFd;
use std::os::unix::io::RawFd;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use std::{cmp, fmt, ptr};

use libc::{self, time_t};

use event_imp::{self as event, Event};
use sys::unix::io::set_cloexec;
use sys::unix::{cvt, UnixReady};
use {io, PollOpt, Ready, Token};

/// Each Selector has a globally unique(ish) ID associated with it. This ID
/// gets tracked by `TcpStream`, `TcpListener`, etc... when they are first
/// registered with the `Selector`. If a type that is previously associated with
/// a `Selector` attempts to register itself with a different `Selector`, the
/// operation will return with an error. This matches windows behavior.
static NEXT_ID: AtomicUsize = AtomicUsize::new(0);

#[cfg(not(target_os = "netbsd"))]
type Filter = c_short;
#[cfg(not(target_os = "netbsd"))]
type UData = *mut ::libc::c_void;
#[cfg(not(target_os = "netbsd"))]
type Count = c_int;

#[cfg(target_os = "netbsd")]
type Filter = u32;
#[cfg(target_os = "netbsd")]
type UData = ::libc::intptr_t;
#[cfg(target_os = "netbsd")]
type Count = usize;

macro_rules! kevent {
    ($id: expr, $filter: expr, $flags: expr, $data: expr) => {
        libc::kevent {
            ident: $id as ::libc::uintptr_t,
            filter: $filter as Filter,
            flags: $flags,
            fflags: 0,
            data: 0,
            udata: $data as UData,
        }
    };
}

pub struct Selector {
    id: usize,
    kq: RawFd,
}

impl Selector {
    pub fn new() -> io::Result<Selector> {
        // offset by 1 to avoid choosing 0 as the id of a selector
        let id = NEXT_ID.fetch_add(1, Ordering::Relaxed) + 1;
        let kq = unsafe { cvt(libc::kqueue())? };
        drop(set_cloexec(kq));

        Ok(Selector { id, kq })
    }

    #[allow(unused)]
    pub fn id(&self) -> usize {
        self.id
    }

    pub fn select(
        &self,
        evts: &mut Events,
        awakener: Token,
        timeout: Option<Duration>,
    ) -> io::Result<bool> {
        let timeout = timeout.map(|to| {
            libc::timespec {
                tv_sec: cmp::min(to.as_secs(), time_t::MAX as u64) as time_t,
                // `Duration::subsec_nanos` is guaranteed to be less than one
                // billion (the number of nanoseconds in a second), making the
                // cast to i32 safe. The cast itself is needed for platforms
                // where C's long is only 32 bits.
                tv_nsec: libc::c_long::from(to.subsec_nanos() as i32),
            }
        });
        let timeout = timeout
            .as_ref()
            .map(|s| s as *const _)
            .unwrap_or(ptr::null_mut());

        evts.clear();
        unsafe {
            let cnt = cvt(libc::kevent(
                self.kq,
                ptr::null(),
                0,
                evts.sys_events.0.as_mut_ptr(),
                evts.sys_events.0.capacity() as Count,
                timeout,
            ))?;
            evts.sys_events.0.set_len(cnt as usize);
            Ok(evts.coalesce(awakener))
        }
    }

    pub fn register(
        &self,
        fd: RawFd,
        token: Token,
        interests: Ready,
        opts: PollOpt,
    ) -> io::Result<()> {
        trace!("registering; token={:?}; interests={:?}", token, interests);

        let flags = if opts.contains(PollOpt::edge()) {
            libc::EV_CLEAR
        } else {
            0
        } | if opts.contains(PollOpt::oneshot()) {
            libc::EV_ONESHOT
        } else {
            0
        } | libc::EV_RECEIPT;

        unsafe {
            let r = if interests.contains(Ready::readable()) {
                libc::EV_ADD
            } else {
                libc::EV_DELETE
            };
            let w = if interests.contains(Ready::writable()) {
                libc::EV_ADD
            } else {
                libc::EV_DELETE
            };
            let mut changes = [
                kevent!(fd, libc::EVFILT_READ, flags | r, usize::from(token)),
                kevent!(fd, libc::EVFILT_WRITE, flags | w, usize::from(token)),
            ];

            cvt(libc::kevent(
                self.kq,
                changes.as_ptr(),
                changes.len() as Count,
                changes.as_mut_ptr(),
                changes.len() as Count,
                ::std::ptr::null(),
            ))?;

            for change in changes.iter() {
                debug_assert_eq!(change.flags & libc::EV_ERROR, libc::EV_ERROR);

                // Test to see if an error happened
                if change.data == 0 {
                    continue;
                }

                // Older versions of OSX (10.11 and 10.10 have been witnessed)
                // can return EPIPE when registering a pipe file descriptor
                // where the other end has already disappeared. For example code
                // that creates a pipe, closes a file descriptor, and then
                // registers the other end will see an EPIPE returned from
                // `register`.
                //
                // It also turns out that kevent will still report events on the
                // file descriptor, telling us that it's readable/hup at least
                // after we've done this registration. As a result we just
                // ignore `EPIPE` here instead of propagating it.
                //
                // More info can be found at carllerche/mio#582
                if change.data as i32 == libc::EPIPE
                    && change.filter == libc::EVFILT_WRITE as Filter
                {
                    continue;
                }

                // ignore ENOENT error for EV_DELETE
                let orig_flags = if change.filter == libc::EVFILT_READ as Filter {
                    r
                } else {
                    w
                };
                if change.data as i32 == libc::ENOENT && orig_flags & libc::EV_DELETE != 0
                {
                    continue;
                }

                return Err(::std::io::Error::from_raw_os_error(change.data as i32));
            }
            Ok(())
        }
    }

    pub fn reregister(
        &self,
        fd: RawFd,
        token: Token,
        interests: Ready,
        opts: PollOpt,
    ) -> io::Result<()> {
        // Just need to call register here since EV_ADD is a mod if already
        // registered
        self.register(fd, token, interests, opts)
    }

    pub fn deregister(&self, fd: RawFd) -> io::Result<()> {
        unsafe {
            // EV_RECEIPT is a nice way to apply changes and get back per-event results while not
            // draining the actual changes.
            let filter = libc::EV_DELETE | libc::EV_RECEIPT;
            #[cfg(not(target_os = "netbsd"))]
            let mut changes = [
                kevent!(fd, libc::EVFILT_READ, filter, ptr::null_mut()),
                kevent!(fd, libc::EVFILT_WRITE, filter, ptr::null_mut()),
            ];

            #[cfg(target_os = "netbsd")]
            let mut changes = [
                kevent!(fd, libc::EVFILT_READ, filter, 0),
                kevent!(fd, libc::EVFILT_WRITE, filter, 0),
            ];

            cvt(libc::kevent(
                self.kq,
                changes.as_ptr(),
                changes.len() as Count,
                changes.as_mut_ptr(),
                changes.len() as Count,
                ::std::ptr::null(),
            ))
            .map(|_| ())?;

            if changes[0].data as i32 == libc::ENOENT
                && changes[1].data as i32 == libc::ENOENT
            {
                return Err(::std::io::Error::from_raw_os_error(changes[0].data as i32));
            }
            for change in changes.iter() {
                debug_assert_eq!(libc::EV_ERROR & change.flags, libc::EV_ERROR);
                if change.data != 0 && change.data as i32 != libc::ENOENT {
                    return Err(::std::io::Error::from_raw_os_error(
                        changes[0].data as i32,
                    ));
                }
            }
            Ok(())
        }
    }
}

impl fmt::Debug for Selector {
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        fmt.debug_struct("Selector")
            .field("id", &self.id)
            .field("kq", &self.kq)
            .finish()
    }
}

impl AsRawFd for Selector {
    fn as_raw_fd(&self) -> RawFd {
        self.kq
    }
}

impl Drop for Selector {
    fn drop(&mut self) {
        unsafe {
            let _ = libc::close(self.kq);
        }
    }
}

pub struct Events {
    sys_events: KeventList,
    events: Vec<Event>,
    event_map: HashMap<Token, usize>,
}

struct KeventList(Vec<libc::kevent>);

unsafe impl Send for KeventList {}
unsafe impl Sync for KeventList {}

impl Events {
    pub fn with_capacity(cap: usize) -> Events {
        Events {
            sys_events: KeventList(Vec::with_capacity(cap)),
            events: Vec::with_capacity(cap),
            event_map: HashMap::with_capacity(cap),
        }
    }

    #[inline]
    pub fn len(&self) -> usize {
        self.events.len()
    }

    #[inline]
    pub fn capacity(&self) -> usize {
        self.events.capacity()
    }

    #[inline]
    pub fn is_empty(&self) -> bool {
        self.events.is_empty()
    }

    pub fn get(&self, idx: usize) -> Option<Event> {
        self.events.get(idx).cloned()
    }

    fn coalesce(&mut self, awakener: Token) -> bool {
        let mut ret = false;
        self.events.clear();
        self.event_map.clear();

        for e in self.sys_events.0.iter() {
            let token = Token(e.udata as usize);
            let len = self.events.len();

            if token == awakener {
                // TODO: Should this return an error if event is an error. It
                // is not critical as spurious wakeups are permitted.
                ret = true;
                continue;
            }

            let idx = *self.event_map.entry(token).or_insert(len);

            if idx == len {
                // New entry, insert the default
                self.events.push(Event::new(Ready::empty(), token));
            }

            if e.flags & libc::EV_ERROR != 0 {
                event::kind_mut(&mut self.events[idx]).insert(*UnixReady::error());
            }

            if e.filter == libc::EVFILT_READ as Filter {
                event::kind_mut(&mut self.events[idx]).insert(Ready::readable());
            } else if e.filter == libc::EVFILT_WRITE as Filter {
                event::kind_mut(&mut self.events[idx]).insert(Ready::writable());
            }
            #[cfg(any(
                target_os = "dragonfly",
                target_os = "freebsd",
                target_os = "ios",
                target_os = "macos"
            ))]
            {
                if e.filter == libc::EVFILT_AIO {
                    event::kind_mut(&mut self.events[idx]).insert(UnixReady::aio());
                }
            }
            #[cfg(target_os = "freebsd")]
            {
                if e.filter == libc::EVFILT_LIO {
                    event::kind_mut(&mut self.events[idx]).insert(UnixReady::lio());
                }
            }
        }

        ret
    }

    pub fn push_event(&mut self, event: Event) {
        self.events.push(event);
    }

    pub fn clear(&mut self) {
        self.sys_events.0.truncate(0);
        self.events.truncate(0);
        self.event_map.clear();
    }
}

impl fmt::Debug for Events {
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        fmt.debug_struct("Events")
            .field("len", &self.sys_events.0.len())
            .finish()
    }
}

#[test]
fn does_not_register_rw() {
    use unix::EventedFd;
    use {Poll, PollOpt, Ready, Token};

    let kq = unsafe { libc::kqueue() };
    let kqf = EventedFd(&kq);
    let poll = Poll::new().unwrap();

    // registering kqueue fd will fail if write is requested (On anything but some versions of OS
    // X)
    poll.register(
        &kqf,
        Token(1234),
        Ready::readable(),
        PollOpt::edge() | PollOpt::oneshot(),
    )
    .unwrap();
}

#[cfg(any(
    target_os = "dragonfly",
    target_os = "freebsd",
    target_os = "ios",
    target_os = "macos"
))]
#[test]
fn test_coalesce_aio() {
    let mut events = Events::with_capacity(1);
    events
        .sys_events
        .0
        .push(kevent!(0x1234, libc::EVFILT_AIO, 0, 42));
    events.coalesce(Token(0));
    assert!(events.events[0].readiness() == UnixReady::aio().into());
    assert!(events.events[0].token() == Token(42));
}