a10 0.4.2

//! The A10 I/O library. [^1]
//!
//! This library is meant as a low-level library safely exposing different OS's
//! abilities to perform non-blocking I/O.
//!
//! For simplicity this only has the following main types and a number of helper
//! types:
//!  * [`AsyncFd`] is a wrapper around a file descriptor that provides a safe
//!    API to perform I/O operations such as `read(2)` using [`Future`]s.
//!  * [`SubmissionQueue`] is needed to start operations, such as opening a file
//!    or new socket, but on its own can't do much.
//!  * [`Ring`] needs be [polled] so that the I/O operations can make progress.
//!
//! Some modules provide ways to create `AsyncFd`, e.g. [`socket`] or
//! [`OpenOptions`], others are simply a place to expose the [`Future`]s
//! supporting the I/O operations. The modules try to roughly follow the same
//! structure as that of the standard library.
//!
//! [polled]: Ring::poll
//! [`socket`]: net::socket
//! [`OpenOptions`]: fs::OpenOptions
//! [`Future`]: std::future::Future
//!
//! # Implementation Notes
//!
//! On Linux this uses io_uring, which is a completion based API. For the BSD
//! family of OS (FreeBSD, OpenBSD, NetBSD, etc.) and for the Apple family
//! (macOS, iOS, etc.) this uses kqueue, which is a poll based API.
//!
//! To support both the completion and poll based API most I/O operations need
//! ownership of the data, e.g. a buffer, so it can delay deallocation if
//! needed. [^2] The input data can be retrieved again by using the [`Extract`]
//! trait.
//!
//! Additional documentation can be found in the [`io_uring(7)`] and
//! [`kqueue(2)`] manuals.
//!
//! [`io_uring(7)`]: https://man7.org/linux/man-pages/man7/io_uring.7.html
//! [`kqueue(2)`]: https://man.freebsd.org/cgi/man.cgi?query=kqueue
//!
//! # Examples
//!
//! Examples can be found in the examples directory of the source code,
//! [available online on GitHub].
//!
//! [available online on GitHub]: https://github.com/Thomasdezeeuw/a10/tree/main/examples
//!
//! [^1]: The name A10 comes from the [A10 ring road around Amsterdam], which
//!       relates to the ring buffers that io_uring uses in its design.
//! [^2]: Delaying of the deallocation needs to happen for completion based APIs
//!       where an I/O operation `Future` is dropped before it's complete -- the
//!       OS will continue to use the resources, which would result in a
//!       use-after-free bug.
//!
//! [A10 ring road around Amsterdam]: https://en.wikipedia.org/wiki/A10_motorway_(Netherlands)

#![cfg_attr(feature = "nightly", feature(async_iterator, cfg_sanitize))]

#[cfg(not(any(
    target_os = "android",
    target_os = "dragonfly",
    target_os = "freebsd",
    target_os = "ios",
    target_os = "linux",
    target_os = "macos",
    target_os = "netbsd",
    target_os = "openbsd",
    target_os = "tvos",
    target_os = "visionos",
    target_os = "watchos",
)))]
compile_error!("OS not supported");

use std::fmt;
use std::sync::atomic::{AtomicU8, Ordering};
use std::time::Duration;

// This must come before the other modules for the documentation.
pub mod fd;

mod asan;
mod config;
mod msan;
mod op;
#[cfg(unix)]
mod unix;

pub mod extract;
pub mod fs;
pub mod io;
pub mod mem;
pub mod net;
pub mod pipe;
pub mod poll;
pub mod process;

#[cfg(any(target_os = "android", target_os = "linux"))]
mod io_uring;
#[cfg(any(
    target_os = "dragonfly",
    target_os = "freebsd",
    target_os = "ios",
    target_os = "macos",
    target_os = "netbsd",
    target_os = "openbsd",
    target_os = "tvos",
    target_os = "visionos",
    target_os = "watchos",
))]
mod kqueue;

#[cfg(any(target_os = "android", target_os = "linux"))]
mod inotify;

mod sys {
    #[cfg(any(target_os = "android", target_os = "linux"))]
    pub(crate) use crate::io_uring::*;
    #[cfg(any(
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "ios",
        target_os = "macos",
        target_os = "netbsd",
        target_os = "openbsd",
        target_os = "tvos",
        target_os = "visionos",
        target_os = "watchos",
    ))]
    pub(crate) use crate::kqueue::*;

    #[cfg(any(target_os = "android", target_os = "linux"))]
    pub(crate) use crate::inotify as fs_notify;
}

#[doc(inline)]
pub use config::Config;
#[doc(no_inline)]
pub use extract::Extract;
#[doc(no_inline)]
pub use fd::AsyncFd;

/// Ring.
///
/// The API on this type is quite minimal. It provides access to the
/// [`SubmissionQueue`], which is used to perform I/O operations. And it exposes
/// [`Ring::poll`] needs to be called to make progress on the operations and
/// mark the [`Future`]s are ready to poll.
#[derive(Debug)]
pub struct Ring {
    cq: sys::Completions,
    sq: sys::Submissions,
}

impl Ring {
    /// Configure a `Ring`.
    pub const fn config<'r>() -> Config<'r> {
        Config {
            sys: crate::sys::Config::new(),
        }
    }

    /// Create a new `Ring` with the default configuration.
    ///
    /// For more configuration options see [`Config`].
    #[doc(alias = "io_uring_setup")]
    #[doc(alias = "kqueue")]
    pub fn new() -> io::Result<Ring> {
        Ring::config().build()
    }

    /// Returns the `SubmissionQueue` used by this ring.
    ///
    /// The submission queue can be used to queue asynchronous I/O operations.
    pub fn sq(&self) -> SubmissionQueue {
        SubmissionQueue(self.sq.clone())
    }

    /// Poll the ring for completions.
    ///
    /// This will wake all completed [`Future`]s with the result of their
    /// operations.
    ///
    /// [`Future`]: std::future::Future
    #[doc(alias = "io_uring_enter")]
    #[doc(alias = "kevent")]
    pub fn poll(&mut self, timeout: Option<Duration>) -> io::Result<()> {
        self.cq.poll(self.sq.shared(), timeout)
    }

    /// Returns an [`AsyncIterator`] that returns whenever this `Ring` is ready
    /// be [polled].
    ///
    /// The submission queue `sq` MUST not be attached to this ring.
    ///
    /// [`AsyncIterator`]: std::async_iter::AsyncIterator
    /// [polled]: Ring::poll
    ///
    /// # Examples
    ///
    /// ```
    /// # async fn run() -> std::io::Result<()> {
    /// let mut main_ring = a10::Ring::new()?;
    /// let other_ring = a10::Ring::new()?;
    ///
    /// // This AsyncIterator will return something once other_ring has an event to be polled.
    /// let other_ring_pollable = other_ring.pollable(main_ring.sq());
    ///
    /// // This will wake other_ring_pollable once other_ring has an event.
    /// main_ring.poll(None)?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// Don't pass the the submission queue of the ring itself, it doesn't work.
    ///
    /// ```should_panic
    /// # fn main() -> std::io::Result<()> {
    /// let ring = a10::Ring::new()?;
    /// let bad = ring.pollable(ring.sq()); // BAD: this doesn't work.
    /// # Ok(())
    /// # }
    /// ```
    pub fn pollable(&self, sq: SubmissionQueue) -> poll::Pollable {
        assert!(
            self.sq != sq.0,
            "can't wait on pollable with sq of the same ring"
        );
        let state = poll::PollableState::new(self.sq());
        poll::Pollable::new(sq, state, ())
    }
}

/// Queue to submit asynchronous operations to.
///
/// This type doesn't have many public methods, but is used by all I/O types to
/// queue asynchronous operations. The queue can be acquired by using
/// [`Ring::sq`].
///
/// The submission queue can be shared by cloning it, it's a cheap operation.
///
/// The submission queue can be compared equality, when it returns true it means
/// both will submit to the same ring.
#[derive(Clone, Debug, Eq, PartialEq)]
#[repr(transparent)]
pub struct SubmissionQueue(sys::Submissions);

impl SubmissionQueue {
    /// Wake the connected [`Ring`].
    ///
    /// All this does is interrupt a call to [`Ring::poll`].
    ///
    /// # Notes
    ///
    /// When using [`Config::single_issuer`] (io_uring only) this will always
    /// return `EEXIST` (when calling on a different thread).
    pub fn wake(&self) {
        if let Err(err) = self.0.wake() {
            log::warn!("failed to wake a10::Ring: {err}");
        }
    }

    pub(crate) fn submissions(&self) -> &sys::Submissions {
        &self.0
    }

    #[cfg(any(target_os = "android", target_os = "linux"))]
    pub(crate) fn from_ref<'a>(submissions: &'a sys::Submissions) -> &'a SubmissionQueue {
        // SAFETY: `SubmissionQueue` and `sys::Submissions` have the same layout
        // due to `repr(transparent)`.
        unsafe { &*std::ptr::from_ref(submissions).cast::<SubmissionQueue>() }
    }

    /// Returns itself.
    ///
    /// Used by the operation macro to be generic over `SubmissionQueue` and
    /// `AsyncFd`.
    pub(crate) const fn sq(&self) -> &SubmissionQueue {
        self
    }
}

/// Helper macro to execute a system call that returns an `io::Result`.
macro_rules! syscall {
    ($fn: ident ( $($arg: expr),* $(,)? ) ) => {{
        #[allow(unused_unsafe)]
        let res = unsafe { libc::$fn($( $arg, )*) };
        if res == -1 {
            ::std::result::Result::Err(::std::io::Error::last_os_error())
        } else {
            ::std::result::Result::Ok(res)
        }
    }};
}

/// Link to online manual.
#[rustfmt::skip]
macro_rules! man_link {
    ($syscall: tt ( $section: tt ) ) => {
        concat!(
            "\n\nAdditional documentation can be found in the ",
            "[`", stringify!($syscall), "(", stringify!($section), ")`]",
            "(https://man7.org/linux/man-pages/man", stringify!($section), "/", stringify!($syscall), ".", stringify!($section), ".html)",
            " manual.\n"
        )
    };
}

macro_rules! new_flag {
    (
        $(
        $(#[$type_meta:meta])*
        $type_vis: vis struct $type_name: ident ( $type_repr: ty ) $( impl BitOr $( $type_or: ty )* )? {
            $(
            $(#[$value_meta:meta])*
            $value_name: ident = $libc: ident :: $value_type: ident,
            )*
        }
        )+
    ) => {
        $(
        $(#[$type_meta])*
        #[derive(Copy, Clone, Eq, PartialEq)]
        #[repr(transparent)]
        $type_vis struct $type_name(pub(crate) $type_repr);

        impl $type_name {
            $(
            $(#[$value_meta])*
            #[allow(trivial_numeric_casts, clippy::cast_sign_loss)]
            $type_vis const $value_name: $type_name = $type_name($libc::$value_type as $type_repr);
            )*

            // Need the value_meta to set the cfg attribute, but that also
            // includes documentation, which we can ignore.
            #[allow(unused_doc_comments, dead_code)]
            pub(crate) const ALL_VALUES: &[$type_name] = &[
                $(
                $(#[$value_meta])*
                $type_name::$value_name,
                )*
            ];
        }

        $crate::debug_detail!(impl match for $type_name($type_repr), $( $(#[$value_meta])* $libc::$value_type ),*);

        $(
        impl std::ops::BitOr for $type_name {
            type Output = Self;

            fn bitor(self, rhs: Self) -> Self::Output {
                $type_name(self.0 | rhs.0)
            }
        }

        $(
        impl std::ops::BitOr<$type_or> for $type_name {
            type Output = Self;

            #[allow(clippy::cast_sign_loss)]
            fn bitor(self, rhs: $type_or) -> Self::Output {
                $type_name(self.0 | rhs as $type_repr)
            }
        }
        )*
        )?
        )+
    };
}

macro_rules! debug_detail {
    (
        // Match a value exactly.
        match $type: ident ($event_type: ty),
        $( $( #[$meta: meta] )* $libc: ident :: $flag: ident ),+ $(,)?
    ) => {
        #[repr(transparent)]
        struct $type($event_type);

        impl $type {
            #[allow(dead_code)]
            fn from_ref(value: &$event_type) -> &$type {
                // SAFETY: layout is the same due to repr(transparent).
                unsafe { &*::std::ptr::from_ref(value).cast::<$type>() }
            }
        }

        $crate::debug_detail!(impl match for $type($event_type), $( $(#[$meta])* $libc::$flag ),*);
    };
    (
        // Match a value exactly.
        impl match for $type: ident ($type_repr: ty),
        $( $( #[$meta: meta] )* $libc: ident :: $flag: ident ),* $(,)?
    ) => {
        impl ::std::fmt::Debug for $type {
            #[allow(trivial_numeric_casts, unreachable_patterns, unreachable_code, unused_doc_comments, clippy::bad_bit_mask)]
            fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
                $(
                $(#[$meta])*
                pub(super) const $flag: $type_repr = $libc :: $flag as $type_repr;
                )*

                f.write_str(match self.0 {
                    $(
                    $(#[$meta])*
                    $flag => stringify!($flag),
                    )*
                    value => return value.fmt(f),
                })
            }
        }
    };
    (
        // Integer bitset.
        bitset $type: ident ($event_type: ty),
        $( $( #[$meta: meta] )* $libc: ident :: $flag: ident ),+ $(,)?
    ) => {
        #[repr(transparent)]
        struct $type($event_type);

        impl $type {
            #[allow(dead_code)]
            fn from_ref(value: &$event_type) -> &$type {
                // SAFETY: layout is the same due to repr(transparent).
                unsafe { &*::std::ptr::from_ref(value).cast::<$type>() }
            }
        }

        $crate::debug_detail!(impl bitset for $type($event_type), $( $(#[$meta])* $libc::$flag ),*);
    };
    (
        // Integer bitset.
        impl bitset for $type: ident ($event_type: ty),
        $( $( #[$meta: meta] )* $libc: ident :: $flag: ident ),+ $(,)?
    ) => {
        impl ::std::fmt::Debug for $type {
            fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
                #[allow(unused_mut)]
                let mut written_one = false;
                $(
                    $(#[$meta])*
                    #[allow(clippy::bad_bit_mask)] // Apparently some flags are zero.
                    {
                        if self.0 & $libc :: $flag != 0 {
                            if !written_one {
                                write!(f, "{}", stringify!($flag))?;
                                written_one = true;
                            } else {
                                write!(f, "|{}", stringify!($flag))?;
                            }
                        }
                    }
                )+
                if !written_one {
                    write!(f, "(empty)")
                } else {
                    Ok(())
                }
            }
        }
    };
}

use {debug_detail, man_link, new_flag, syscall};

/// Lock `mutex` clearing any poison set.
fn lock<'a, T>(mutex: &'a std::sync::Mutex<T>) -> std::sync::MutexGuard<'a, T> {
    match mutex.lock() {
        Ok(guard) => guard,
        Err(err) => {
            mutex.clear_poison();
            err.into_inner()
        }
    }
}

/// Same as [`lock`], but doesn't block if the mutex is locked.
#[cfg(any(target_os = "android", target_os = "linux"))]
fn try_lock<'a, T>(mutex: &'a std::sync::Mutex<T>) -> Option<std::sync::MutexGuard<'a, T>> {
    match mutex.try_lock() {
        Ok(guard) => Some(guard),
        Err(std::sync::TryLockError::Poisoned(err)) => {
            mutex.clear_poison();
            Some(err.into_inner())
        }
        Err(std::sync::TryLockError::WouldBlock) => None,
    }
}

/// Get mutable access to the lock's data.
#[cfg(any(target_os = "android", target_os = "linux"))]
fn get_mut<'a, T>(mutex: &'a mut std::sync::Mutex<T>) -> &'a mut T {
    match mutex.get_mut() {
        Ok(guard) => guard,
        Err(err) => err.into_inner(),
    }
}

/// Trait to work with results for singleshot (`io::Result`) and multishot
/// (`Option<io::Result>`) operations.
// Replace this with std::ops::FromResidual once stable.
#[allow(unused)]
trait OpPollResult<T> {
    fn from_ok(ok: T) -> Self;
    fn from_err(err: io::Error) -> Self;
    fn from_res(res: io::Result<T>) -> Self;
    fn done() -> Self;
}

impl<T> OpPollResult<T> for io::Result<T> {
    fn from_ok(ok: T) -> Self {
        Ok(ok)
    }

    fn from_err(err: io::Error) -> Self {
        Err(err)
    }

    fn from_res(res: io::Result<T>) -> Self {
        res
    }

    fn done() -> Self {
        unreachable!()
    }
}

impl<T> OpPollResult<T> for Option<io::Result<T>> {
    fn from_ok(ok: T) -> Self {
        Some(Ok(ok))
    }

    fn from_err(err: io::Error) -> Self {
        Some(Err(err))
    }

    fn from_res(res: io::Result<T>) -> Self {
        Some(res)
    }

    fn done() -> Self {
        None
    }
}

/// Polling state of a [`Ring`].
pub(crate) struct PollingState(AtomicU8);

const IS_POLLING: u8 = 0b01;
#[allow(unused)] // Used in an assert below, but rustc doesn't seem to care.
const NOT_POLLING: u8 = 0b00;
const IS_AWOKEN: u8 = 0b10;
const NOT_AWOKEN: u8 = 0b00;

impl PollingState {
    pub(crate) const fn new() -> PollingState {
        PollingState(AtomicU8::new(0))
    }

    /// Set the state to currently (not) polling.
    ///
    /// Returns a boolean indicating if the poll call should continue like
    /// normal, or should be cut short (e.g. with a zero timeout).
    #[allow(clippy::cast_lossless)]
    pub(crate) fn set_polling(&self, is_polling: bool) -> bool {
        const _BOOL_CAST_CHECK_TRUE: () = assert!(true as u8 == IS_POLLING);
        const _BOOL_CAST_CHECK_FALSE: () = assert!(false as u8 == NOT_POLLING);
        let state = self.0.swap(is_polling as u8 | NOT_AWOKEN, Ordering::AcqRel);
        (state & IS_AWOKEN) != 0
    }

    /// Set the state to wake the polling thread.
    ///
    /// Returns a boolean indicating if the caller should submit an event to
    /// wake up the polling thread.
    pub(crate) fn wake(&self) -> bool {
        let state = self.0.fetch_or(IS_AWOKEN, Ordering::AcqRel);
        state == (IS_POLLING | NOT_AWOKEN)
    }
}

impl fmt::Debug for PollingState {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let state = self.0.load(Ordering::Relaxed);
        f.debug_struct("PollingState")
            .field("polling", &((state & IS_POLLING) != 0))
            .field("awoken", &((state & IS_AWOKEN) != 0))
            .finish()
    }
}