coreshift_core/

fd.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/

//! Owned file-descriptor primitives.
//!
//! [`Fd`] is the crate-wide owned descriptor type (move-only, closes on drop).
//! [`Token`] and [`Event`] describe readiness events delivered by the
//! [`crate::reactor::Reactor`].

use crate::CoreError;
use crate::error::syscall_ret;
use std::io::Error as IoError;
use std::time::Duration;

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

/// An owned file descriptor that closes on drop.
///
/// `Fd` is move-only. Constructing one from a raw descriptor transfers close
/// ownership to `Fd`; do not also close the raw descriptor elsewhere.
///
/// ### Fork Safety
/// `Fd` instances created by Core usually have `O_CLOEXEC` set. If the process
/// forks, the descriptor will be inherited by the child but will be closed
/// automatically upon `exec`. Callers that need a descriptor to survive `exec`
/// must clear the flag manually.
pub struct Fd(RawFd);

use std::os::unix::io::{AsRawFd, RawFd};

impl AsRawFd for Fd {
    fn as_raw_fd(&self) -> RawFd {
        self.0
    }
}

impl Fd {
    /// Wrap a raw file descriptor.
    ///
    /// # Errors
    /// Returns a [`CoreError`] if the descriptor is negative.
    #[inline(always)]
    pub(crate) fn new(fd: RawFd, op: &'static str) -> Result<Self, CoreError> {
        if fd < 0 {
            Err(CoreError::sys(errno(), op))
        } else {
            Ok(Self(fd))
        }
    }

    /// Wrap an owned raw file descriptor.
    ///
    /// # Safety
    /// The caller must guarantee `fd` is valid, open, and uniquely owned by the
    /// returned `Fd`. Passing a borrowed fd, or closing `fd` after this call,
    /// can cause double-close or use-after-close bugs.
    #[inline(always)]
    pub unsafe fn from_owned_raw_fd(fd: RawFd, op: &'static str) -> Result<Self, CoreError> {
        Self::new(fd, op)
    }

    /// Create a non-blocking `eventfd` with `EFD_CLOEXEC`.
    ///
    /// The descriptor is created with `FD_CLOEXEC` set.
    ///
    /// ### Errors
    /// - `EINVAL`: `init` is invalid.
    /// - `EMFILE`: Process limit on open file descriptors hit.
    /// - `ENFILE`: System-wide limit on open files hit.
    pub fn eventfd(init: u32) -> Result<Self, CoreError> {
        let fd = unsafe { libc::eventfd(init, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) };
        syscall_ret(fd, "eventfd")?;
        Self::new(fd, "eventfd")
    }

    /// Create a non-blocking `timerfd` using `CLOCK_MONOTONIC` with `TFD_CLOEXEC`.
    ///
    /// The descriptor is created with `FD_CLOEXEC` set.
    ///
    /// ### Errors
    /// - `EMFILE`: Process limit on open file descriptors hit.
    /// - `ENFILE`: System-wide limit on open files hit.
    /// - `ENOMEM`: Insufficient kernel memory.
    pub fn timerfd() -> Result<Self, CoreError> {
        let fd = unsafe {
            libc::timerfd_create(
                libc::CLOCK_MONOTONIC,
                libc::TFD_CLOEXEC | libc::TFD_NONBLOCK,
            )
        };
        syscall_ret(fd, "timerfd_create")?;
        Self::new(fd, "timerfd_create")
    }

    /// Access the underlying raw file descriptor.
    ///
    /// NOTE: This is an escape hatch for low-level interactions. Prefer using
    /// the safe methods on `Fd` or implementing `AsRawFd`.
    #[inline(always)]
    pub(crate) fn raw(&self) -> RawFd {
        self.0
    }

    /// Duplicate this descriptor, returning a new owned `Fd`.
    ///
    /// Both descriptors share the same open file description, so a `dup` of
    /// an eventfd remains a single signalable object: a write on either copy
    /// is observed on the other. Useful for fan-out wakeups (hub keeps one
    /// copy, a worker reactor owns another).
    ///
    /// ### Errors
    /// - `EBADF`: The source descriptor is invalid.
    /// - `EMFILE`: The process file descriptor limit is reached.
    pub fn dup(&self) -> Result<Self, CoreError> {
        let r = loop {
            let d = unsafe { libc::dup(self.0) };
            if d < 0 && errno() == libc::EINTR {
                continue;
            }
            break d;
        };
        if r < 0 {
            let e = errno();
            Err(CoreError::sys(e, "dup"))
        } else {
            // SAFETY: `r` is a freshly duplicated owned descriptor.
            unsafe { Self::from_owned_raw_fd(r, "dup") }
        }
    }

    /// Perform a `dup2` syscall.
    ///
    /// ### Errors
    /// - `EBADF`: The source or target file descriptor is invalid.
    /// - `EMFILE`: The target descriptor exceeds the process limit.
    pub fn dup2(&self, target: RawFd) -> Result<(), CoreError> {
        loop {
            let r = unsafe { libc::dup2(self.0, target) };
            if r < 0 {
                let e = errno();
                if e == libc::EINTR {
                    continue;
                }
                return syscall_ret(r, "dup2");
            }
            return Ok(());
        }
    }

    /// Set the `O_NONBLOCK` flag on the descriptor.
    ///
    /// ### Errors
    /// - `EBADF`: The file descriptor is invalid.
    pub fn set_nonblock(&self) -> Result<(), CoreError> {
        let flags = unsafe { libc::fcntl(self.0, libc::F_GETFL) };
        syscall_ret(flags, "fcntl(F_GETFL)")?;
        let r = unsafe { libc::fcntl(self.0, libc::F_SETFL, flags | libc::O_NONBLOCK) };
        syscall_ret(r, "fcntl(F_SETFL)")
    }

    /// Set the `FD_CLOEXEC` flag on the descriptor.
    ///
    /// ### Errors
    /// - `EBADF`: The file descriptor is invalid.
    pub fn set_cloexec(&self) -> Result<(), CoreError> {
        let flags = unsafe { libc::fcntl(self.0, libc::F_GETFD) };
        syscall_ret(flags, "fcntl(F_GETFD)")?;
        let r = unsafe { libc::fcntl(self.0, libc::F_SETFD, flags | libc::FD_CLOEXEC) };
        syscall_ret(r, "fcntl(F_SETFD)")
    }

    /// Read bytes into a mutable slice.
    ///
    /// Returns `Ok(None)` if the operation would block (`EAGAIN`).
    ///
    /// ### Edge Cases
    /// - **Zero-length read**: Returns `Ok(Some(0))` immediately.
    /// - **Partial read**: Returns the number of bytes actually read.
    ///
    /// ### Errors
    /// - `EBADF`: The file descriptor is invalid or not open for reading.
    /// - `EFAULT`: `buf` points outside the process's address space.
    /// - `EIO`: Low-level I/O error.
    pub fn read_slice(&self, buf: &mut [u8]) -> Result<Option<usize>, CoreError> {
        self.read_raw(buf.as_mut_ptr(), buf.len())
    }

    /// Seek to an absolute file offset.
    ///
    /// ### Errors
    /// - `EBADF`: The file descriptor is not seekable.
    /// - `EINVAL`: `offset` is invalid.
    /// - `EOVERFLOW`: The resulting offset exceeds the off_t range.
    pub fn seek_set(&self, offset: i64) -> Result<u64, CoreError> {
        loop {
            let pos = unsafe { libc::lseek(self.0, offset as libc::off_t, libc::SEEK_SET) };
            if pos < 0 {
                let e = errno();
                if e == libc::EINTR {
                    continue;
                }
                return Err(CoreError::sys(e, "lseek"));
            }
            return Ok(pos as u64);
        }
    }

    /// Write bytes from a slice.
    ///
    /// Returns `Ok(None)` if the operation would block (`EAGAIN`).
    ///
    /// ### Edge Cases
    /// - **Zero-length write**: Returns `Ok(Some(0))` immediately.
    /// - **Partial write**: Returns the number of bytes actually written.
    ///
    /// ### Errors
    /// - `EBADF`: The file descriptor is invalid or not open for writing.
    /// - `EFAULT`: `buf` points outside the process's address space.
    /// - `EPIPE`: The reading end of a pipe or socket was closed.
    pub fn write_slice(&self, buf: &[u8]) -> Result<Option<usize>, CoreError> {
        self.write_raw(buf.as_ptr(), buf.len())
    }

    /// Read a native-endian `u64`, blocking until data is available.
    ///
    /// Unlike `read_u64`, this never returns `Ok(None)` — it retries on `EINTR`
    /// and returns `Err` only on a hard I/O failure. Intended for **blocking**
    /// eventfds used as inter-thread notification primitives.
    ///
    /// CORE-M10: [`Fd::eventfd`] always creates a *non-blocking* descriptor
    /// (`EFD_NONBLOCK`); on such an fd this returns `Err(EAGAIN)` when empty.
    /// This method's contract assumes a genuinely blocking fd (e.g. the
    /// `libc::eventfd` in `task_stack.rs`), and it bounds the EINTR retry so a
    /// signal storm cannot spin the calling thread forever.
    pub fn read_u64_blocking(&self) -> Result<u64, CoreError> {
        let mut bytes = [0u8; std::mem::size_of::<u64>()];
        let mut eintr_retries = 0;
        loop {
            let n =
                unsafe { libc::read(self.0, bytes.as_mut_ptr() as *mut libc::c_void, bytes.len()) };
            if n == bytes.len() as isize {
                return Ok(u64::from_ne_bytes(bytes));
            }
            if n < 0 {
                let e = errno();
                if e == libc::EINTR {
                    eintr_retries += 1;
                    if eintr_retries >= MAX_BLOCKING_READ_EINTR_RETRIES {
                        return Err(CoreError::sys(
                            libc::EINTR,
                            "read_u64_blocking:eintr_exhausted",
                        ));
                    }
                    continue;
                }
                return Err(CoreError::sys(e, "read_u64_blocking"));
            }
            return Err(CoreError::sys(libc::EIO, "read_u64_blocking:short_read"));
        }
    }

    /// Read a native-endian `u64`.
    ///
    /// Returns `Ok(None)` if the operation would block (`EAGAIN`).
    pub fn read_u64(&self) -> Result<Option<u64>, CoreError> {
        let mut bytes = [0u8; std::mem::size_of::<u64>()];
        match self.read_slice(&mut bytes)? {
            Some(n) if n == bytes.len() => Ok(Some(u64::from_ne_bytes(bytes))),
            Some(_) => Err(CoreError::sys(libc::EIO, "read_u64")),
            None => Ok(None),
        }
    }

    /// Write a native-endian `u64`.
    ///
    /// Returns `Ok(None)` if the operation would block (`EAGAIN`).
    pub fn write_u64(&self, value: u64) -> Result<Option<usize>, CoreError> {
        self.write_slice(&value.to_ne_bytes())
    }

    /// Arm or disarm a one-shot `timerfd`.
    ///
    /// Passing `None` disarms the timer. Zero durations are rounded up to one
    /// nanosecond so the timer still expires.
    ///
    /// ### Errors
    /// - `EBADF`: The file descriptor is invalid.
    /// - `EINVAL`: The duration is invalid or not supported by the kernel.
    pub fn set_timer_oneshot(&self, delay: Option<Duration>) -> Result<(), CoreError> {
        let mut spec: libc::itimerspec = unsafe { std::mem::zeroed() };
        if let Some(delay) = delay {
            let delay = delay.max(Duration::from_nanos(1));
            spec.it_value.tv_sec = delay.as_secs() as libc::time_t;
            spec.it_value.tv_nsec = delay.subsec_nanos() as libc::c_long;
        }

        let ret = unsafe { libc::timerfd_settime(self.raw(), 0, &spec, std::ptr::null_mut()) };
        syscall_ret(ret, "timerfd_settime")
    }

    /// Read bytes into a raw buffer.
    ///
    /// Internal callers must ensure `buf` points to a valid writable region of
    /// at least `count` bytes.
    ///
    /// Returns `Ok(None)` if the operation would block (`EAGAIN`).
    pub(crate) fn read_raw(&self, buf: *mut u8, count: usize) -> Result<Option<usize>, CoreError> {
        loop {
            let n = unsafe { libc::read(self.0, buf as *mut libc::c_void, count) };
            if n < 0 {
                let e = errno();
                if e == libc::EINTR {
                    continue;
                }
                if e == libc::EAGAIN || e == libc::EWOULDBLOCK {
                    return Ok(None);
                }
                return Err(CoreError::sys(e, "read"));
            }
            return Ok(Some(n as usize));
        }
    }

    /// Write bytes from a raw buffer.
    ///
    /// Internal callers must ensure `buf` points to a valid readable region of
    /// at least `count` bytes.
    ///
    /// Returns `Ok(None)` if the operation would block (`EAGAIN`).
    pub(crate) fn write_raw(
        &self,
        buf: *const u8,
        count: usize,
    ) -> Result<Option<usize>, CoreError> {
        loop {
            let n = unsafe { libc::write(self.0, buf as *const libc::c_void, count) };
            if n < 0 {
                let e = errno();
                if e == libc::EINTR {
                    continue;
                }
                if e == libc::EAGAIN || e == libc::EWOULDBLOCK {
                    return Ok(None);
                }
                return Err(CoreError::sys(e, "write"));
            }
            return Ok(Some(n as usize));
        }
    }
}

impl Drop for Fd {
    fn drop(&mut self) {
        if self.0 >= 0 {
            unsafe {
                libc::close(self.0);
            }
        }
    }
}

/// An opaque token representing a registered file descriptor.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct Token(pub(crate) u64);

#[allow(dead_code)]
impl Token {
    #[inline(always)]
    pub(crate) fn new(val: u64) -> Self {
        Self(val)
    }

    #[inline(always)]
    pub(crate) fn val(&self) -> u64 {
        self.0
    }
}

/// A readiness event generated by the reactor.
#[derive(Clone, Copy, Debug)]
pub struct Event {
    /// Token associated with the ready descriptor.
    pub token: Token,
    /// Descriptor is ready for reading (`EPOLLIN`).
    pub readable: bool,
    /// Descriptor has priority data or an exceptional condition (`EPOLLPRI`).
    pub priority: bool,
    /// Descriptor is ready for writing (`EPOLLOUT`).
    pub writable: bool,
    /// Indicates an error condition (`EPOLLERR`).
    ///
    /// NOTE: For edge-triggered readiness, an error condition often means both
    /// readable and writable are set to ensure the handler drains the FD.
    pub error: bool,
    /// Indicates a remote hangup (`EPOLLHUP`).
    pub hangup: bool,
}

const _: () = assert!(std::mem::size_of::<Event>() == 16);
const _: () = assert!(std::mem::align_of::<Event>() == 8);

/// CORE-M10: cap on consecutive `EINTR` retries inside
/// [`read_u64_blocking`](Fd::read_u64_blocking). A real signal storm would
/// otherwise spin the caller forever; this turns a pathological storm into a
/// surfaced error while remaining generous enough for legitimate interruption.
const MAX_BLOCKING_READ_EINTR_RETRIES: u32 = 10_000;