crossio_kqueue/

kqueue.rs

use std::{
    io,
    mem,
    os::fd::RawFd,
    time::Duration,
};

use crossio_core::{
    Backend,
    BackendConfig,
    CrossioError,
    Event,
    Events,
    Interest,
    Token,
};

use libc::{
    c_void,
    kevent,
    kqueue,
    timespec,
    EV_ADD,
    EV_CLEAR,
    EV_DELETE,
    EV_ENABLE,
    EV_EOF,
    EV_ERROR,
    EVFILT_READ,
    EVFILT_WRITE,
};

/// BSD/macOS kqueue-backed implementation of the `Backend` trait.
pub struct KqueueBackend {
    kq: RawFd,
    config: BackendConfig,
}

impl KqueueBackend {
    fn new_kqueue() -> Result<RawFd, CrossioError> {
        // SAFETY: kqueue takes no parameters and returns a file descriptor on
        // success or -1 on failure.
        let fd = unsafe { kqueue() };
        if fd == -1 {
            return Err(CrossioError::Io(io::Error::last_os_error()));
        }

        Ok(fd)
    }

    fn to_timespec(timeout: Option<Duration>) -> Option<timespec> {
        timeout.map(|d| timespec {
            tv_sec: d.as_secs() as libc::time_t,
            tv_nsec: (d.subsec_nanos()) as libc::c_long,
        })
    }

    fn apply_changes(&self, changes: &mut [kevent]) -> Result<(), CrossioError> {
        if changes.is_empty() {
            return Ok(());
        }

        // SAFETY: `changes.as_ptr()` points to `changes.len()` valid kevent
        // structures. The timeout is NULL, so this call returns immediately
        // after applying the changelist.
        let res = unsafe {
            libc::kevent(
                self.kq,
                changes.as_ptr(),
                changes.len() as i32,
                std::ptr::null_mut(),
                0,
                std::ptr::null(),
            )
        };

        if res == -1 {
            return Err(CrossioError::Io(io::Error::last_os_error()));
        }

        Ok(())
    }

    fn interest_to_changes(
        fd: RawFd,
        token: Token,
        interest: Interest,
        flags: i16,
        out: &mut Vec<kevent>,
    ) {
        let udata = token.as_usize() as *mut c_void;

        unsafe {
            if interest.contains(Interest::READABLE) {
                let mut ev: kevent = mem::zeroed();
                libc::EV_SET(
                    &mut ev,
                    fd as libc::uintptr_t,
                    EVFILT_READ,
                    flags as u16,
                    0,
                    0,
                    udata,
                );
                out.push(ev);
            }

            if interest.contains(Interest::WRITABLE) {
                let mut ev: kevent = mem::zeroed();
                libc::EV_SET(
                    &mut ev,
                    fd as libc::uintptr_t,
                    EVFILT_WRITE,
                    flags as u16,
                    0,
                    0,
                    udata,
                );
                out.push(ev);
            }
        }
    }

    fn kevent_to_event(ev: &kevent) -> Option<Event> {
        let token = Token::from_usize(ev.udata as usize);
        let mut interest = Interest::empty();

        if ev.filter == EVFILT_READ {
            interest |= Interest::READABLE;
        }
        if ev.filter == EVFILT_WRITE {
            interest |= Interest::WRITABLE;
        }

        if (ev.flags & EV_EOF) != 0 || (ev.flags & EV_ERROR) != 0 {
            interest |= Interest::CLOSED;
        }

        if interest.is_empty() {
            None
        } else {
            Some(Event::new(token, interest))
        }
    }
}

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

impl Backend for KqueueBackend {
    type RawSource = RawFd;

    fn new(config: BackendConfig) -> Result<Self, CrossioError> {
        let kq = Self::new_kqueue()?;
        Ok(Self { kq, config })
    }

    fn register(
        &self,
        source: Self::RawSource,
        token: Token,
        interest: Interest,
    ) -> Result<(), CrossioError> {
        let mut changes = Vec::new();
        let flags = (EV_ADD | EV_ENABLE | EV_CLEAR) as i16;
        Self::interest_to_changes(source, token, interest, flags, &mut changes);
        self.apply_changes(&mut changes)
    }

    fn reregister(
        &self,
        source: Self::RawSource,
        token: Token,
        interest: Interest,
    ) -> Result<(), CrossioError> {
        let mut changes = Vec::new();
        let flags = (EV_ADD | EV_ENABLE | EV_CLEAR) as i16;
        Self::interest_to_changes(source, token, interest, flags, &mut changes);
        self.apply_changes(&mut changes)
    }

    fn deregister(&self, source: Self::RawSource) -> Result<(), CrossioError> {
        let mut changes = Vec::new();
        let token = Token::from_usize(0);
        let interest = Interest::readable_writable();
        let flags = (EV_DELETE) as i16;
        Self::interest_to_changes(source, token, interest, flags, &mut changes);
        self.apply_changes(&mut changes)
    }

    fn poll(
        &self,
        events: &mut Events,
        timeout: Option<Duration>,
    ) -> Result<usize, CrossioError> {
        events.clear();

        let capacity = self.config.event_batch_size().max(1);
        let mut evlist: Vec<kevent> = Vec::with_capacity(capacity);

        let ts_opt = Self::to_timespec(timeout.or(self.config.default_timeout()));
        let timeout_ptr = ts_opt
            .as_ref()
            .map(|ts| ts as *const timespec)
            .unwrap_or(std::ptr::null());

        // SAFETY: `evlist` has sufficient capacity for `capacity` events. On
        // success, kevent initializes the first `n` entries.
        let n = unsafe {
            libc::kevent(
                self.kq,
                std::ptr::null(),
                0,
                evlist.as_mut_ptr(),
                capacity as i32,
                timeout_ptr,
            )
        };

        if n == -1 {
            return Err(CrossioError::Io(io::Error::last_os_error()));
        }

        unsafe {
            evlist.set_len(n as usize);
        }

        for ev in &evlist {
            if let Some(e) = Self::kevent_to_event(ev) {
                events.push(e);
            }
        }

        Ok(events.len())
    }
}

#[cfg(test)]
mod tests {
    use super::KqueueBackend;
    use crossio_core::{
        registration::{Registration, Source},
        BackendConfig,
        Interest,
        Reactor,
        Token,
    };
    use std::{
        os::fd::RawFd,
        time::Duration,
    };

    struct FdSource(RawFd);

    impl Source<KqueueBackend> for FdSource {
        fn raw_source(&self) -> RawFd {
            self.0
        }
    }

    #[test]
    fn kqueue_reports_readable_pipe() {
        let mut fds = [0; 2];
        let res = unsafe { libc::pipe(fds.as_mut_ptr()) };
        assert_eq!(res, 0, "pipe creation failed");

        let read_fd = fds[0];
        let write_fd = fds[1];

        let config = BackendConfig::default().with_event_batch_size(16);
        let mut reactor = Reactor::<KqueueBackend>::with_config(config)
            .expect("failed to create kqueue reactor");

        let source = FdSource(read_fd);
        let token = Token::from_usize(1);
        let registration = Registration::new(token, Interest::READABLE);

        reactor
            .register(&source, registration)
            .expect("failed to register pipe with kqueue");

        let buf = [42u8];
        let written = unsafe {
            libc::write(write_fd, buf.as_ptr() as *const _, buf.len())
        };
        assert_eq!(written, buf.len() as isize, "write to pipe failed");

        let events = reactor
            .poll(Some(Duration::from_millis(100)))
            .expect("kqueue poll failed");

        let has_readable = events
            .iter()
            .any(|ev| ev.token() == token && ev.readiness().contains(Interest::READABLE));

        assert!(has_readable, "kqueue did not report readable pipe");

        unsafe {
            libc::close(read_fd);
            libc::close(write_fd);
        }
    }
}