crossio-epoll 0.1.0

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

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

/// Linux epoll-backed implementation of the `Backend` trait.
pub struct EpollBackend {
    epoll_fd: RawFd,
    config: BackendConfig,
}

impl EpollBackend {
    fn interest_to_epoll(interest: Interest) -> u32 {
        let mut flags: u32 = 0;

        if interest.contains(Interest::READABLE) {
            flags |= libc::EPOLLIN as u32;
        }
        if interest.contains(Interest::WRITABLE) {
            flags |= libc::EPOLLOUT as u32;
        }
        if interest.contains(Interest::PRIORITY) {
            flags |= libc::EPOLLPRI as u32;
        }
        if interest.contains(Interest::CLOSED) {
            flags |= libc::EPOLLRDHUP as u32;
        }

        flags
    }

    fn epoll_to_interest(events: u32) -> Interest {
        let mut interest = Interest::empty();

        if events & (libc::EPOLLIN as u32) != 0 {
            interest |= Interest::READABLE;
        }
        if events & (libc::EPOLLOUT as u32) != 0 {
            interest |= Interest::WRITABLE;
        }
        if events & (libc::EPOLLPRI as u32) != 0 {
            interest |= Interest::PRIORITY;
        }
        if events & (libc::EPOLLRDHUP as u32) != 0
            || events & (libc::EPOLLHUP as u32) != 0
            || events & (libc::EPOLLERR as u32) != 0
        {
            interest |= Interest::CLOSED;
        }

        interest
    }

    fn make_timeout_ms(&self, timeout: Option<Duration>) -> i32 {
        let effective = timeout.or(self.config.default_timeout());
        match effective {
            None => -1,
            Some(d) => {
                let ms = d.as_millis().min(i32::MAX as u128) as i32;
                ms
            }
        }
    }
}

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

impl Backend for EpollBackend {
    type RawSource = RawFd;

    fn new(config: BackendConfig) -> Result<Self, CrossioError> {
        // SAFETY: epoll_create1 is called with a valid flag; on success it
        // returns a file descriptor referring to a new epoll instance.
        let fd = unsafe { libc::epoll_create1(libc::EPOLL_CLOEXEC) };
        if fd < 0 {
            let err = io::Error::last_os_error();
            return Err(CrossioError::Io(err));
        }

        Ok(Self { epoll_fd: fd, config })
    }

    fn register(
        &self,
        source: Self::RawSource,
        token: Token,
        interest: Interest,
    ) -> Result<(), CrossioError> {
        let mut ev = libc::epoll_event {
            events: Self::interest_to_epoll(interest),
            u64: token.as_usize() as u64,
        };

        // SAFETY: epoll_fd and source are valid file descriptors; &mut ev
        // points to a properly initialized epoll_event.
        let res = unsafe {
            libc::epoll_ctl(
                self.epoll_fd,
                libc::EPOLL_CTL_ADD,
                source,
                &mut ev as *mut libc::epoll_event,
            )
        };

        if res < 0 {
            let err = io::Error::last_os_error();

            // Translate common resource-exhaustion errors into CrossioError::ResourceLimit.
            if matches!(err.raw_os_error(), Some(libc::EMFILE) | Some(libc::ENFILE) | Some(libc::ENOSPC)) {
                return Err(CrossioError::ResourceLimit);
            }

            return Err(CrossioError::Io(err));
        }

        Ok(())
    }

    fn reregister(
        &self,
        source: Self::RawSource,
        token: Token,
        interest: Interest,
    ) -> Result<(), CrossioError> {
        let mut ev = libc::epoll_event {
            events: Self::interest_to_epoll(interest),
            u64: token.as_usize() as u64,
        };

        let res = unsafe {
            libc::epoll_ctl(
                self.epoll_fd,
                libc::EPOLL_CTL_MOD,
                source,
                &mut ev as *mut libc::epoll_event,
            )
        };

        if res < 0 {
            let err = io::Error::last_os_error();
            return Err(CrossioError::Io(err));
        }

        Ok(())
    }

    fn deregister(&self, source: Self::RawSource) -> Result<(), CrossioError> {
        let res = unsafe {
            libc::epoll_ctl(
                self.epoll_fd,
                libc::EPOLL_CTL_DEL,
                source,
                std::ptr::null_mut(),
            )
        };

        if res < 0 {
            let err = io::Error::last_os_error();
            return Err(CrossioError::Io(err));
        }

        Ok(())
    }

    fn poll(
        &self,
        events: &mut Events,
        timeout: Option<Duration>,
    ) -> Result<usize, CrossioError> {
        let timeout_ms = self.make_timeout_ms(timeout);
        let capacity = self.config.event_batch_size().max(1);

        let mut raw_events: Vec<libc::epoll_event> = Vec::with_capacity(capacity);

        let nfds = unsafe {
            libc::epoll_wait(
                self.epoll_fd,
                raw_events.as_mut_ptr(),
                capacity as i32,
                timeout_ms,
            )
        };

        if nfds < 0 {
            let err = io::Error::last_os_error();

            // Treat EINTR as "no events" rather than an error so callers can retry.
            if matches!(err.raw_os_error(), Some(libc::EINTR)) {
                return Ok(0);
            }

            return Err(CrossioError::Io(err));
        }

        unsafe {
            raw_events.set_len(nfds as usize);
        }

        for ev in &raw_events {
            let interest = EpollBackend::epoll_to_interest(ev.events);
            let token = Token::from_usize(ev.u64 as usize);
            events.push(Event::new(token, interest));
        }

        Ok(nfds as usize)
    }
}

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

    struct FdSource(RawFd);

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

    #[test]
    fn epoll_reports_readable_pipe() {
        // SAFETY: pipe allocates a pair of file descriptors on success.
        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::<EpollBackend>::with_config(config)
            .expect("failed to create epoll 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 epoll");

        // Make the read end readable by writing a single byte.
        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("epoll poll failed");

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

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

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