coreshift_core/reactor/

mod.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/

//! Asynchronous event reactor.
//!
//! This module provides a lightweight wrapper around Linux `epoll` for
//! multiplexing I/O events. It is optimized for edge-triggered monitoring.
//! It is intentionally explicit about Linux readiness semantics rather than
//! hiding them behind a higher-level async runtime abstraction.

pub use crate::fd::{Event, Fd, Token};

use crate::CoreError;
use crate::error::syscall_ret;
use std::io::Error as IoError;
use std::os::unix::io::RawFd;

#[inline(always)]
fn errno() -> i32 {
    IoError::last_os_error().raw_os_error().unwrap_or(0)
}

/// A lightweight epoll reactor using edge-triggered monitoring (EPOLLET).
///
/// ### Edge-Triggered Contract
/// Because this reactor uses EPOLLET, all handlers MUST drain their respective
/// read or write sources until they receive an `EAGAIN` / `EWOULDBLOCK` error
/// (represented as `Ok(None)` in the `Fd` helpers).
///
/// Failure to drain a source will result in missing future readiness events
/// for that file descriptor until it is re-registered or another event occurs.
///
/// ### Fork Safety
/// The `Reactor` owns an `epoll` descriptor which is `O_CLOEXEC`. After an
/// `exec` call in a child process, the reactor and all its registrations are
/// lost. If the child continues without `exec`, it shares the same epoll
/// instance, which is generally unsafe and requires careful coordination.
///
/// # Example
/// ```no_run
/// # use coreshift_core::reactor::{Reactor, Fd, Event};
/// # fn example(fd: Fd) -> Result<(), Box<dyn std::error::Error>> {
/// let mut reactor = Reactor::new()?;
/// let token = reactor.add(&fd, true, false)?;
///
/// let mut events = Vec::new();
/// loop {
///     reactor.wait(&mut events, 64, -1)?;
///     for ev in &events {
///         if ev.token == token {
///             // Drain fd...
///         }
///     }
/// }
/// # Ok(())
/// # }
/// ```
pub struct Reactor {
    epfd: RawFd,
    next_token: u64,
    events_buf: Vec<libc::epoll_event>,
    signalfd: Option<Fd>,
    signalfd_previous_mask: Option<libc::sigset_t>,
    /// Thread that called [`Self::setup_signalfd`]; its signal mask is only
    /// restored in `Drop` when the drop runs on the same thread.
    signalfd_setup_thread: Option<libc::pthread_t>,
    /// Token for the signalfd (if initialized).
    sigchld_token: Option<Token>,
    /// Token for the inotify fd (if initialized).
    inotify_token: Option<Token>,
}

impl Reactor {
    /// Create a new epoll reactor.
    ///
    /// ### Errors
    /// - `EMFILE`: Process limit on open file descriptors hit.
    /// - `ENFILE`: System-wide limit on open files hit.
    /// - `ENOMEM`: Insufficient kernel memory.
    pub fn new() -> Result<Self, CoreError> {
        let epfd = unsafe { libc::epoll_create1(libc::EPOLL_CLOEXEC) };
        syscall_ret(epfd, "epoll_create1")?;
        Ok(Self {
            epfd,
            next_token: 1,
            events_buf: Vec::with_capacity(64),
            signalfd: None,
            signalfd_previous_mask: None,
            signalfd_setup_thread: None,
            sigchld_token: None,
            inotify_token: None,
        })
    }

    /// Initialize inotify and add it to the reactor.
    ///
    /// ### Errors
    /// - `EMFILE`: Process limit on open file descriptors hit.
    /// - `ENFILE`: System-wide limit on open files hit.
    /// - `ENOMEM`: Insufficient kernel memory.
    /// - `EPERM`: Permission denied to create inotify instance.
    pub fn setup_inotify(&mut self) -> Result<(Fd, Token), CoreError> {
        let fd = unsafe { libc::inotify_init1(libc::IN_CLOEXEC | libc::IN_NONBLOCK) };
        syscall_ret(fd, "inotify_init1")?;

        let fd_obj = Fd::new(fd, "inotify")?;
        let token = self.add(&fd_obj, true, false)?;
        self.inotify_token = Some(token);

        Ok((fd_obj, token))
    }

    /// Initialize signalfd for SIGCHLD and add it to the reactor.
    ///
    /// SIGCHLD is blocked on the calling thread (the thread that will read the
    /// signalfd) so child exits arrive via the signalfd instead of the default
    /// disposition. The previous mask of the calling thread is restored when
    /// the reactor is dropped **on the same thread** that called this method;
    /// dropping from another thread leaves the block in place rather than
    /// clobbering that thread's mask.
    ///
    /// ### Multi-threaded delivery
    /// `pthread_sigmask` affects only the calling thread. In a multi-threaded
    /// process every thread that must not consume SIGCHLD — at minimum the
    /// thread driving this reactor — needs SIGCHLD blocked. Block it early
    /// (e.g. in `main` before spawning threads) so worker threads inherit the
    /// mask and the signalfd never loses a child-exit notification to a
    /// sibling thread.
    ///
    /// ### Errors
    /// - `EBADF`: The provided file descriptor is invalid.
    /// - `EINVAL`: Signal mask is invalid or already set up.
    /// - `EMFILE`: Process limit on open file descriptors hit.
    pub fn setup_signalfd(&mut self) -> Result<Token, CoreError> {
        if self.signalfd.is_some() {
            return Err(CoreError::sys(
                libc::EINVAL,
                "setup_signalfd already initialized",
            ));
        }

        let mut mask: libc::sigset_t = unsafe { std::mem::zeroed() };
        unsafe { libc::sigemptyset(&mut mask) };
        unsafe { libc::sigaddset(&mut mask, libc::SIGCHLD) };

        let mut previous_mask: libc::sigset_t = unsafe { std::mem::zeroed() };
        let r = unsafe { libc::pthread_sigmask(libc::SIG_BLOCK, &mask, &mut previous_mask) };
        if r != 0 {
            return Err(CoreError::sys(r, "pthread_sigmask(SIG_BLOCK)"));
        }

        let sfd = unsafe { libc::signalfd(-1, &mask, libc::SFD_NONBLOCK | libc::SFD_CLOEXEC) };
        if let Err(err) = syscall_ret(sfd, "signalfd") {
            let _ = unsafe {
                libc::pthread_sigmask(libc::SIG_SETMASK, &previous_mask, std::ptr::null_mut())
            };
            return Err(err);
        }

        let fd = Fd::new(sfd, "signalfd")?;
        let token = match self.add(&fd, true, false) {
            Ok(token) => token,
            Err(err) => {
                let _ = unsafe {
                    libc::pthread_sigmask(libc::SIG_SETMASK, &previous_mask, std::ptr::null_mut())
                };
                return Err(err);
            }
        };

        self.signalfd = Some(fd);
        self.signalfd_previous_mask = Some(previous_mask);
        self.signalfd_setup_thread = Some(unsafe { libc::pthread_self() });
        self.sigchld_token = Some(token);

        Ok(token)
    }

    /// Drain the internal signalfd buffer.
    pub fn drain_signalfd(&self) -> Result<(), CoreError> {
        if let Some(fd) = &self.signalfd {
            let mut buf = [0u8; std::mem::size_of::<libc::signalfd_siginfo>()];
            loop {
                match fd.read_slice(&mut buf) {
                    Ok(Some(n)) if n < buf.len() => break,
                    Ok(Some(_)) => continue,
                    Ok(None) => break,
                    Err(e) => return Err(e),
                }
            }
        }
        Ok(())
    }

    /// Register a file descriptor with the reactor.
    ///
    /// This assigns a new unique token for the descriptor and enables
    /// edge-triggered monitoring.
    #[inline(always)]
    pub fn add(&mut self, fd: &Fd, readable: bool, writable: bool) -> Result<Token, CoreError> {
        let token = Token(self.next_token);
        self.next_token += 1;
        self.add_with_token(fd.raw(), token, readable, writable, false)?;
        Ok(token)
    }

    /// Register a file descriptor for priority readiness (EPOLLPRI).
    #[inline(always)]
    pub fn add_priority(&mut self, fd: &Fd) -> Result<Token, CoreError> {
        let token = Token(self.next_token);
        self.next_token += 1;
        self.add_with_token(fd.raw(), token, false, false, true)?;
        Ok(token)
    }

    /// Register a file descriptor with custom epoll flags.
    ///
    /// This allows registration with flags like `EPOLLONESHOT` or explicit
    /// control over `EPOLLET`.
    ///
    /// # Example
    /// ```no_run
    /// # use coreshift_core::reactor::{Reactor, Fd};
    /// let mut reactor = Reactor::new().unwrap();
    /// let fd = Fd::eventfd(0).unwrap();
    /// reactor.add_with_flags(&fd, (libc::EPOLLIN | libc::EPOLLONESHOT) as u32).unwrap();
    /// ```
    #[inline(always)]
    pub fn add_with_flags(&mut self, fd: &Fd, flags: u32) -> Result<Token, CoreError> {
        let token = Token(self.next_token);
        self.next_token += 1;
        let mut ev = libc::epoll_event {
            events: flags,
            u64: token.0,
        };
        let r = unsafe { libc::epoll_ctl(self.epfd, libc::EPOLL_CTL_ADD, fd.raw(), &mut ev) };
        syscall_ret(r, "epoll_ctl_add")?;
        Ok(token)
    }

    #[inline(always)]
    pub(crate) fn add_with_token(
        &mut self,
        raw_fd: RawFd,
        token: Token,
        readable: bool,
        writable: bool,
        priority: bool,
    ) -> Result<(), CoreError> {
        let mut events = libc::EPOLLET as u32;
        if readable {
            events |= libc::EPOLLIN as u32;
        }
        if writable {
            events |= libc::EPOLLOUT as u32;
        }
        if priority {
            events |= libc::EPOLLPRI as u32;
        }
        let mut ev = libc::epoll_event {
            events,
            u64: token.0,
        };
        let r = unsafe { libc::epoll_ctl(self.epfd, libc::EPOLL_CTL_ADD, raw_fd, &mut ev) };
        syscall_ret(r, "epoll_ctl_add")?;
        Ok(())
    }

    /// Remove a file descriptor from the reactor.
    #[inline(always)]
    pub fn del(&self, fd: &Fd) -> Result<(), CoreError> {
        self.del_raw(fd.raw())
    }

    /// Remove a raw descriptor from the reactor.
    ///
    /// NOTE: This is an escape hatch for low-level interactions. Prefer using
    /// [`del`](Self::del).
    #[inline(always)]
    pub(crate) fn del_raw(&self, raw: RawFd) -> Result<(), CoreError> {
        loop {
            let ret = unsafe {
                libc::epoll_ctl(self.epfd, libc::EPOLL_CTL_DEL, raw, std::ptr::null_mut())
            };
            if ret == -1 {
                let e = errno();
                if e == libc::EINTR {
                    continue;
                }
                return Err(CoreError::sys(e, "epoll_ctl_del"));
            }
            return Ok(());
        }
    }

    /// Wait for events.
    ///
    /// This function blocks until at least one event is ready or the timeout
    /// expires. Ready events are appended to the `buffer`.
    ///
    /// ### Timeout Contract
    /// - `-1`: Block indefinitely until an event occurs or a signal interrupts.
    /// - `0`: Return immediately, even if no events are ready.
    /// - `> 0`: Wait for up to the specified number of milliseconds.
    ///
    /// Returns the number of events received.
    #[inline(always)]
    pub fn wait(
        &mut self,
        buffer: &mut Vec<Event>,
        max_events: usize,
        timeout: i32,
    ) -> Result<usize, CoreError> {
        buffer.clear();

        if max_events == 0 {
            return Ok(0);
        }

        // Ensure buffer has enough capacity
        if buffer.capacity() < max_events {
            buffer.reserve(max_events.saturating_sub(buffer.len()));
        }

        if self.events_buf.capacity() < max_events {
            self.events_buf
                .reserve(max_events.saturating_sub(self.events_buf.len()));
        }

        let n = unsafe {
            libc::epoll_wait(
                self.epfd,
                self.events_buf.as_mut_ptr(),
                max_events as i32,
                timeout,
            )
        };

        if n > 0 {
            unsafe {
                self.events_buf.set_len(n as usize);
            }
            for i in 0..n as usize {
                let ev = self.events_buf[i];
                let is_read = (ev.events & libc::EPOLLIN as u32) != 0;
                let is_priority = (ev.events & libc::EPOLLPRI as u32) != 0;
                let is_write = (ev.events & libc::EPOLLOUT as u32) != 0;
                let is_err = (ev.events & libc::EPOLLERR as u32) != 0;
                let is_hup = (ev.events & libc::EPOLLHUP as u32) != 0;

                buffer.push(Event {
                    token: Token(ev.u64),
                    readable: is_read || is_err,
                    priority: is_priority || is_err,
                    writable: is_write || is_err,
                    error: is_err,
                    hangup: is_hup,
                });
            }
            return Ok(n as usize);
        }

        if n < 0 {
            let e = errno();
            if e == libc::EINTR {
                return Ok(0);
            }
            return Err(CoreError::sys(e, "epoll_wait"));
        }
        Ok(0)
    }

    /// Return the raw epoll file descriptor.
    ///
    /// NOTE: This is an escape hatch for low-level interactions.
    #[allow(dead_code)]
    pub(crate) fn fd(&self) -> RawFd {
        self.epfd
    }
}

impl Drop for Reactor {
    fn drop(&mut self) {
        // Only restore the calling thread's mask when the drop happens on the
        // thread that originally blocked SIGCHLD. Restoring on a different
        // thread would clobber that thread's mask; leaving the block in place
        // on the setup thread is harmless (SIGCHLD is silently ignored while
        // blocked and the signalfd fd is closed with the reactor).
        let same_thread = self
            .signalfd_setup_thread
            .is_some_and(|t| unsafe { libc::pthread_equal(t, libc::pthread_self()) } != 0);
        if same_thread && let Some(mask) = self.signalfd_previous_mask.take() {
            let _ =
                unsafe { libc::pthread_sigmask(libc::SIG_SETMASK, &mask, std::ptr::null_mut()) };
        }
        if self.epfd >= 0 {
            unsafe {
                libc::close(self.epfd);
            }
        }
    }
}