compio_driver/

lib.rs

//! Platform-specific drivers.
//!
//! Some types differ by compilation target.

#![cfg_attr(docsrs, feature(doc_cfg))]
#![cfg_attr(feature = "once_cell_try", feature(once_cell_try))]
#![allow(unused_features)]
#![warn(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]
#![doc(
    html_logo_url = "https://github.com/compio-rs/compio-logo/raw/refs/heads/master/generated/colored-bold.svg"
)]
#![doc(
    html_favicon_url = "https://github.com/compio-rs/compio-logo/raw/refs/heads/master/generated/colored-bold.svg"
)]

use std::{
    io,
    num::NonZero,
    task::{Poll, Waker},
    time::Duration,
};

use compio_buf::BufResult;
use compio_log::instrument;

mod control;
mod macros;
mod panic;

mod key;
pub use key::Key;

mod asyncify;
pub use asyncify::*;

mod fd;
pub use fd::*;

mod driver_type;
pub use driver_type::*;

mod sys;
pub use sys::{op, *};

mod cancel;
pub use cancel::*;

mod buffer_pool;
pub use buffer_pool::{BoxAllocator, BufferAllocator, BufferPool, BufferRef};

use crate::{
    buffer_pool::{BufferAlloc, BufferPoolRoot},
    key::ErasedKey,
    panic::resume_unwind_io,
    sys::op::OpCodeFlag,
};

/// The return type of [`Proactor::push`].
#[derive(Debug)]
pub enum PushEntry<K, R> {
    /// The operation is pushed to the submission queue.
    Pending(K),
    /// The operation is ready and returns.
    Ready(R),
}

impl<K, R> PushEntry<K, R> {
    /// Get if the current variant is [`PushEntry::Ready`].
    pub const fn is_ready(&self) -> bool {
        matches!(self, Self::Ready(_))
    }

    /// Take the ready variant if exists.
    pub fn take_ready(self) -> Option<R> {
        match self {
            Self::Pending(_) => None,
            Self::Ready(res) => Some(res),
        }
    }

    /// Map the [`PushEntry::Pending`] branch.
    pub fn map_pending<L>(self, f: impl FnOnce(K) -> L) -> PushEntry<L, R> {
        match self {
            Self::Pending(k) => PushEntry::Pending(f(k)),
            Self::Ready(r) => PushEntry::Ready(r),
        }
    }

    /// Map the [`PushEntry::Ready`] branch.
    pub fn map_ready<S>(self, f: impl FnOnce(R) -> S) -> PushEntry<K, S> {
        match self {
            Self::Pending(k) => PushEntry::Pending(k),
            Self::Ready(r) => PushEntry::Ready(f(r)),
        }
    }
}

/// Low-level actions of completion-based IO.
/// It owns the operations to keep the driver safe.
pub struct Proactor {
    driver: Driver,
    buffer_pool: BufferPoolState,
}

enum BufferPoolState {
    Uninit {
        allocator: BufferAlloc,
        num_of_bufs: u16,
        buffer_len: usize,
        flags: u16,
    },
    Init(BufferPoolRoot),
}

impl BufferPoolState {
    fn get(&mut self, driver: &mut Driver) -> io::Result<BufferPool> {
        loop {
            match self {
                BufferPoolState::Uninit {
                    allocator,
                    num_of_bufs,
                    buffer_len,
                    flags,
                } => {
                    *self = BufferPoolState::Init(BufferPoolRoot::new(
                        driver,
                        *allocator,
                        *num_of_bufs,
                        *buffer_len,
                        *flags,
                    )?);
                }
                BufferPoolState::Init(root) => return Ok(root.get_pool()),
            }
        }
    }
}

impl Drop for Proactor {
    fn drop(&mut self) {
        let BufferPoolState::Init(buffer_pool) = &mut self.buffer_pool else {
            return;
        };
        debug_assert!(buffer_pool.is_unique()); // Just in case. Shouldn't happen
        _ = unsafe { buffer_pool.release(&mut self.driver) };
    }
}

assert_not_impl!(Proactor, Send);
assert_not_impl!(Proactor, Sync);

impl Proactor {
    /// Create [`Proactor`] with 1024 entries.
    pub fn new() -> io::Result<Self> {
        Self::builder().build()
    }

    /// Create [`ProactorBuilder`] to config the proactor.
    pub fn builder() -> ProactorBuilder {
        ProactorBuilder::new()
    }

    fn with_builder(builder: &ProactorBuilder) -> io::Result<Self> {
        Ok(Self {
            driver: Driver::new(builder)?,
            buffer_pool: BufferPoolState::Uninit {
                allocator: builder.buffer_pool_allocator,
                num_of_bufs: builder.buffer_pool_size,
                buffer_len: builder.buffer_pool_buffer_len,
                flags: builder.buffer_pool_flag,
            },
        })
    }

    /// Get a default [`Extra`] for underlying driver.
    pub fn default_extra(&self) -> Extra {
        Extra::new(&self.driver)
    }

    /// The current driver type.
    pub fn driver_type(&self) -> DriverType {
        self.driver.driver_type()
    }

    /// Attach an fd to the driver.
    ///
    /// ## Platform specific
    /// * IOCP: it will be attached to the completion port. An fd could only be
    ///   attached to one driver, and could only be attached once, even if you
    ///   `try_clone` it.
    /// * io-uring & polling: it will do nothing but return `Ok(())`.
    pub fn attach(&mut self, fd: RawFd) -> io::Result<()> {
        self.driver.attach(fd)
    }

    /// Cancel an operation with the pushed [`Key`].
    ///
    /// Returns the result if the key is unique and the operation is completed.
    ///
    /// The cancellation is not reliable. The underlying operation may continue,
    /// but just don't return from [`Proactor::poll`].
    pub fn cancel<T: OpCode>(&mut self, key: Key<T>) -> Option<BufResult<usize, T>> {
        instrument!(compio_log::Level::DEBUG, "cancel", ?key);
        if key.set_cancelled() {
            return None;
        }
        if key.is_unique() && key.has_result() {
            let (res, buf) = key.take_result().into_parts();
            Some(BufResult(resume_unwind_io(res), buf))
        } else {
            self.driver.cancel(key.erase());
            None
        }
    }

    /// Cancel an operation with a [`Cancel`] token.
    ///
    /// Returns if a cancellation has been issued.
    ///
    /// The cancellation is not reliable. The underlying operation may continue,
    /// but just don't return from [`Proactor::pop`]. This will do nothing if
    /// the operation has already been completed or cancelled before.
    pub fn cancel_token(&mut self, token: Cancel) -> bool {
        instrument!(compio_log::Level::DEBUG, "cancel_token", ?token);

        let Some(key) = token.upgrade() else {
            return false;
        };
        if key.set_cancelled() || key.has_result() {
            return false;
        }
        self.driver.cancel(key);
        true
    }

    /// Create a [`Cancel`] that can be used to cancel the operation even
    /// without the key.
    ///
    /// This acts like a weak reference to the [`Key`], but can only be used to
    /// cancel the operation with [`Proactor::cancel_token`]. Extra copy of
    /// [`Key`] may cause [`Proactor::pop`] to panic while keys registered
    /// as [`Cancel`] will not. So this is useful in cases where you're not sure
    /// if the operation will be cancelled.
    pub fn register_cancel<T: OpCode>(&mut self, key: &Key<T>) -> Cancel {
        Cancel::new(key)
    }

    /// Push an operation into the driver, and return the unique key [`Key`],
    /// associated with it.
    pub fn push<T: sys::OpCode + 'static>(
        &mut self,
        op: T,
    ) -> PushEntry<Key<T>, BufResult<usize, T>> {
        self.push_with_extra(op, self.default_extra())
    }

    /// Push an operation into the driver with user-defined [`Extra`], and
    /// return the unique key [`Key`], associated with it.
    pub fn push_with_extra<T: sys::OpCode + 'static>(
        &mut self,
        op: T,
        extra: Extra,
    ) -> PushEntry<Key<T>, BufResult<usize, T>> {
        let key = Key::new(op, extra, self.driver_type());
        match self.driver.push(key.clone().erase()) {
            Poll::Pending => PushEntry::Pending(key),
            Poll::Ready(res) => {
                key.set_result(res);
                PushEntry::Ready(key.take_result())
            }
        }
    }

    /// Flush the pushed operations to the kernel. It is roughly equivalent to
    /// calling [`poll`](Proactor::poll) with `Some(Duration::ZERO)` on
    /// io-uring, but is only needed if you're waiting on the driver fd with an
    /// external event loop.
    ///
    /// The return value indicates if the driver is in notified state, which
    /// means the driver is already notified by the user and should not be
    /// waited infinitely.
    ///
    /// This method resets the internal notified state, so that the waker to the
    /// driver will wake up the driver fd through syscalls after this method is
    /// called.
    pub fn flush(&mut self) -> bool {
        self.driver.flush()
    }

    /// Poll the driver and get completed entries.
    /// You need to call [`Proactor::pop`] to get the pushed
    /// operations.
    pub fn poll(&mut self, timeout: Option<Duration>) -> io::Result<()> {
        self.driver.poll(timeout)
    }

    /// Get the pushed operations from the completion entries.
    ///
    /// # Panics
    ///
    /// This function will panic if the [`Key`] is not unique or if the
    /// operation is blocking and it panicked in the thread pool.
    pub fn pop<T: OpCode>(&mut self, key: Key<T>) -> PushEntry<Key<T>, BufResult<usize, T>> {
        instrument!(compio_log::Level::DEBUG, "pop", ?key);
        if key.has_result() {
            let (res, buf) = key.take_result().into_parts();
            PushEntry::Ready(BufResult(resume_unwind_io(res), buf))
        } else {
            PushEntry::Pending(key)
        }
    }

    /// Get the pushed operations from the completion entries along the
    /// [`Extra`] associated.
    ///
    /// # Panics
    ///
    /// This function will panic if the [`Key`] is not unique or if the
    /// operation is blocking and it panicked in the thread pool.
    pub fn pop_with_extra<T: OpCode>(
        &mut self,
        key: Key<T>,
    ) -> PushEntry<Key<T>, (BufResult<usize, T>, Extra)> {
        instrument!(compio_log::Level::DEBUG, "pop", ?key);
        if key.has_result() {
            let extra = key.swap_extra(self.default_extra());
            let (res, buf) = key.take_result().into_parts();
            PushEntry::Ready((BufResult(resume_unwind_io(res), buf), extra))
        } else {
            PushEntry::Pending(key)
        }
    }

    /// Get one completion entry for a multishot operation. If it returns
    /// [`None`], the user should call [`Proactor::pop_with_extra`] to get the
    /// final result of the operation.
    pub fn pop_multishot<T: OpCode>(&mut self, key: &Key<T>) -> Option<BufResult<usize, Extra>> {
        instrument!(compio_log::Level::DEBUG, "pop_multishot", ?key);
        self.driver.pop_multishot(key)
    }

    /// Update the waker of the specified op.
    pub fn update_waker<T>(&mut self, op: &Key<T>, waker: &Waker) {
        op.set_waker(waker);
    }

    /// Create a waker to interrupt the inner driver.
    pub fn waker(&self) -> Waker {
        self.driver.waker()
    }

    /// Register file descriptors for fixed-file operations with io_uring.
    ///
    /// This only works on `io_uring` driver. It will return an [`Unsupported`]
    /// error on other drivers.
    ///
    /// [`Unsupported`]: std::io::ErrorKind::Unsupported
    pub fn register_files(&self, fds: &[RawFd]) -> io::Result<()> {
        fn unsupported(_: &[RawFd]) -> io::Error {
            io::Error::new(
                io::ErrorKind::Unsupported,
                "Fixed-file registration is only supported on io-uring driver",
            )
        }

        #[cfg(io_uring)]
        match self.driver.as_iour() {
            Some(iour) => iour.register_files(fds),
            None => Err(unsupported(fds)),
        }

        #[cfg(not(io_uring))]
        Err(unsupported(fds))
    }

    /// Unregister previously registered file descriptors.
    ///
    /// This only works on `io_uring` driver. It will return an [`Unsupported`]
    /// error on other drivers.
    ///
    /// [`Unsupported`]: std::io::ErrorKind::Unsupported
    pub fn unregister_files(&self) -> io::Result<()> {
        fn unsupported() -> io::Error {
            io::Error::new(
                io::ErrorKind::Unsupported,
                "Fixed-file unregistration is only supported on io-uring driver",
            )
        }

        #[cfg(io_uring)]
        match self.driver.as_iour() {
            Some(iour) => iour.unregister_files(),
            None => Err(unsupported()),
        }

        #[cfg(not(io_uring))]
        Err(unsupported())
    }

    /// Register a new personality in io-uring driver.
    ///
    /// Returns the personality id, which can be used with
    /// [`Extra::set_personality`] to set the personality for an operation.
    ///
    /// This only works on `io_uring` driver. It will return an [`Unsupported`]
    /// error on other drivers. See [`Submitter::register_personality`] for
    /// more.
    ///
    /// [`Unsupported`]: std::io::ErrorKind::Unsupported
    /// [`Submitter::register_personality`]: https://docs.rs/io-uring/latest/io_uring/struct.Submitter.html#method.register_personality
    pub fn register_personality(&self) -> io::Result<u16> {
        fn unsupported() -> io::Error {
            io::Error::new(
                io::ErrorKind::Unsupported,
                "Personality is only supported on io-uring driver",
            )
        }

        #[cfg(io_uring)]
        match self.driver.as_iour() {
            Some(iour) => iour.register_personality(),
            None => Err(unsupported()),
        }

        #[cfg(not(io_uring))]
        Err(unsupported())
    }

    /// Unregister the given personality in io-uring driver.
    ///
    /// This only works on `io_uring` driver. It will return an [`Unsupported`]
    /// error on other drivers. See [`Submitter::unregister_personality`] for
    /// more.
    ///
    /// [`Unsupported`]: std::io::ErrorKind::Unsupported
    /// [`Submitter::unregister_personality`]: https://docs.rs/io-uring/latest/io_uring/struct.Submitter.html#method.unregister_personality
    pub fn unregister_personality(&self, personality: u16) -> io::Result<()> {
        fn unsupported(_: u16) -> io::Error {
            io::Error::new(
                io::ErrorKind::Unsupported,
                "Personality is only supported on io-uring driver",
            )
        }

        #[cfg(io_uring)]
        match self.driver.as_iour() {
            Some(iour) => iour.unregister_personality(personality),
            None => Err(unsupported(personality)),
        }

        #[cfg(not(io_uring))]
        Err(unsupported(personality))
    }

    /// Get the buffer pool of the driver.
    ///
    /// This will lazily initialize the pool at the first time it's accessed,
    /// and future access to the pool will be cheap and infallible.
    pub fn buffer_pool(&mut self) -> io::Result<BufferPool> {
        self.buffer_pool.get(&mut self.driver)
    }
}

impl AsRawFd for Proactor {
    fn as_raw_fd(&self) -> RawFd {
        self.driver.as_raw_fd()
    }
}

/// An completed entry returned from kernel.
///
/// This represents the ownership of [`Key`] passed into the kernel is given
/// back from it to the driver.
#[derive(Debug)]
pub(crate) struct Entry {
    key: ErasedKey,
    result: io::Result<usize>,

    #[cfg(io_uring)]
    flags: u32,
}

unsafe impl Send for Entry {}
unsafe impl Sync for Entry {}

impl Entry {
    pub(crate) fn new(key: ErasedKey, result: io::Result<usize>) -> Self {
        #[cfg(not(io_uring))]
        {
            Self { key, result }
        }
        #[cfg(io_uring)]
        {
            Self {
                key,
                result,
                flags: 0,
            }
        }
    }

    #[allow(dead_code)]
    pub fn user_data(&self) -> usize {
        self.key.as_raw()
    }

    #[allow(dead_code)]
    pub fn into_key(self) -> ErasedKey {
        self.key
    }

    #[cfg(io_uring)]
    pub fn flags(&self) -> u32 {
        self.flags
    }

    #[cfg(io_uring)]
    pub(crate) fn set_flags(&mut self, flags: u32) {
        self.flags = flags;
    }

    pub fn notify(self) {
        #[cfg(io_uring)]
        self.key.borrow().extra_mut().set_flags(self.flags());
        self.key.set_result(self.result);
    }
}

#[derive(Debug, Clone)]
enum ThreadPoolBuilder {
    Create { limit: usize, recv_limit: Duration },
    Reuse(AsyncifyPool),
}

impl Default for ThreadPoolBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl ThreadPoolBuilder {
    pub fn new() -> Self {
        Self::Create {
            limit: 256,
            recv_limit: Duration::from_secs(60),
        }
    }

    pub fn create_or_reuse(&self) -> AsyncifyPool {
        match self {
            Self::Create { limit, recv_limit } => AsyncifyPool::new(*limit, *recv_limit),
            Self::Reuse(pool) => pool.clone(),
        }
    }
}

/// Builder for [`Proactor`].
#[derive(Debug, Clone)]
pub struct ProactorBuilder {
    capacity: u32,
    pool_builder: ThreadPoolBuilder,
    sqpoll_idle: Option<Duration>,
    sqpoll_cpu: Option<u32>,
    cqsize: Option<u32>,
    single_issuer: bool,
    coop_taskrun: bool,
    taskrun_flag: bool,
    defer_taskrun: bool,
    eventfd: Option<RawFd>,
    driver_type: Option<DriverType>,
    op_flags: OpCodeFlag,
    buffer_pool_size: u16,
    buffer_pool_flag: u16,
    buffer_pool_buffer_len: usize,
    buffer_pool_allocator: BufferAlloc,
}

// SAFETY: `RawFd` is thread safe.
unsafe impl Send for ProactorBuilder {}
unsafe impl Sync for ProactorBuilder {}

impl Default for ProactorBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl ProactorBuilder {
    /// Create the builder with default config.
    pub fn new() -> Self {
        Self {
            capacity: 1024,
            pool_builder: ThreadPoolBuilder::new(),
            sqpoll_idle: None,
            sqpoll_cpu: None,
            cqsize: None,
            single_issuer: false,
            coop_taskrun: false,
            taskrun_flag: false,
            defer_taskrun: false,
            eventfd: None,
            driver_type: None,
            op_flags: OpCodeFlag::empty(),
            buffer_pool_size: 8,
            buffer_pool_flag: 0,
            buffer_pool_buffer_len: 8192,
            buffer_pool_allocator: BufferAlloc::new::<BoxAllocator>(),
        }
    }

    /// Set the capacity of the inner event queue or submission queue, if
    /// exists. The default value is 1024.
    pub fn capacity(&mut self, capacity: u32) -> &mut Self {
        self.capacity = capacity;
        self
    }

    /// Set the completion queue size of io-uring driver. The value should be
    /// greater than `capacity`.
    pub fn cqsize(&mut self, cqsize: u32) -> &mut Self {
        self.cqsize = Some(cqsize);
        self
    }

    /// Set the thread number limit of the inner thread pool, if exists. The
    /// default value is 256.
    ///
    /// It will be ignored if `reuse_thread_pool` is set.
    ///
    /// Warning: some operations don't work if the limit is set to zero:
    /// * `Asyncify` needs thread pool.
    /// * Operations except `Recv*`, `Send*`, `Connect`, `Accept` may need
    ///   thread pool.
    pub fn thread_pool_limit(&mut self, value: usize) -> &mut Self {
        if let ThreadPoolBuilder::Create { limit, .. } = &mut self.pool_builder {
            *limit = value;
        }
        self
    }

    /// Set the waiting timeout of the inner thread, if exists. The default is
    /// 60 seconds.
    ///
    /// It will be ignored if `reuse_thread_pool` is set.
    pub fn thread_pool_recv_timeout(&mut self, timeout: Duration) -> &mut Self {
        if let ThreadPoolBuilder::Create { recv_limit, .. } = &mut self.pool_builder {
            *recv_limit = timeout;
        }
        self
    }

    /// Set to reuse an existing [`AsyncifyPool`] in this proactor.
    pub fn reuse_thread_pool(&mut self, pool: AsyncifyPool) -> &mut Self {
        self.pool_builder = ThreadPoolBuilder::Reuse(pool);
        self
    }

    /// Force reuse the thread pool for each proactor created by this builder,
    /// even `reuse_thread_pool` is not set.
    pub fn force_reuse_thread_pool(&mut self) -> &mut Self {
        self.reuse_thread_pool(self.create_or_get_thread_pool());
        self
    }

    /// Create or reuse the thread pool from the config.
    pub fn create_or_get_thread_pool(&self) -> AsyncifyPool {
        self.pool_builder.create_or_reuse()
    }

    /// Set `io-uring` sqpoll idle duration,
    ///
    /// This will also enable io-uring's sqpoll feature.
    ///
    /// # Notes
    ///
    /// - Only effective when the `io-uring` feature is enabled
    /// - `idle` must be >= 1ms, otherwise sqpoll idle will be set to 0 ms
    /// - `idle` will be rounded down
    pub fn sqpoll_idle(&mut self, idle: Duration) -> &mut Self {
        self.sqpoll_idle = Some(idle);
        self
    }

    /// Set CPU affinity for the `io-uring` SQPOLL thread when SQPOLL is
    /// enabled.
    ///
    /// This is only applied when SQPOLL is enabled with
    /// [`sqpoll_idle`](Self::sqpoll_idle).
    ///
    /// # Notes
    ///
    /// - Only effective when the `io-uring` feature is enabled
    /// - `cpu` must be less than the number of cpus in the system, otherwise it
    ///   will return an error when building the proactor.
    pub fn sqpoll_cpu(&mut self, cpu: u32) -> &mut Self {
        self.sqpoll_cpu = Some(cpu);
        self
    }

    /// Set the `io-uring` single issuer hint.
    ///
    /// # Notes
    ///
    /// - Available since Linux Kernel 6.0.
    /// - Only effective when the `io-uring` feature is enabled
    pub fn single_issuer(&mut self, enable: bool) -> &mut Self {
        self.single_issuer = enable;
        self
    }

    /// Optimize performance for most cases, especially compio is a single
    /// thread runtime.
    ///
    /// However, it can't run with sqpoll feature.
    ///
    /// # Notes
    ///
    /// - Available since Linux Kernel 5.19.
    /// - Only effective when the `io-uring` feature is enabled
    pub fn coop_taskrun(&mut self, enable: bool) -> &mut Self {
        self.coop_taskrun = enable;
        self
    }

    /// Allows io-uring driver to know if any cqe's are available when try to
    /// push an sqe to the submission queue.
    ///
    /// This should be enabled with [`coop_taskrun`](Self::coop_taskrun)
    ///
    /// # Notes
    ///
    /// - Available since Linux Kernel 5.19.
    /// - Only effective when the `io-uring` feature is enabled
    pub fn taskrun_flag(&mut self, enable: bool) -> &mut Self {
        self.taskrun_flag = enable;
        self
    }

    /// Defer io-uring task work until the driver enters the kernel to wait for
    /// completions.
    ///
    /// This is only applied when [`single_issuer`](Self::single_issuer) is
    /// enabled. The kernel requires `IORING_SETUP_SINGLE_ISSUER` for
    /// `IORING_SETUP_DEFER_TASKRUN`.
    ///
    /// # Notes
    ///
    /// - Available since Linux Kernel 6.1.
    /// - Only effective when the `io-uring` feature is enabled
    pub fn defer_taskrun(&mut self, enable: bool) -> &mut Self {
        self.defer_taskrun = enable;
        self
    }

    /// Register an eventfd to io-uring.
    ///
    /// # Notes
    ///
    /// - Only effective when the `io-uring` feature is enabled
    pub fn register_eventfd(&mut self, fd: RawFd) -> &mut Self {
        self.eventfd = Some(fd);
        self
    }

    /// Set which io-uring [`OpCode`] must be supported by the driver.
    ///
    /// Support for io-uring opcodes varies by kernel version. Setting this
    /// will force the driver to check for support of the specified opcodes, and
    /// when any of them are not supported:
    ///
    /// - Fallback to `polling` driver if it is enabled, or
    /// - Return an [`Unsupported`] error when building the proactor otherwise.
    ///
    /// # Notes
    ///
    /// - Only effective when the `io-uring` feature is enabled
    /// - [`OpCodeFlag`] is a bitflag struct, you can combine multiple opcodes
    ///   with bitwise OR or use [`OpCodeFlag::all`] to require all opcodes to
    ///   be supported.
    ///
    /// [`Unsupported`]: std::io::ErrorKind::Unsupported
    pub fn detect_opcode_support(&mut self, flags: OpCodeFlag) -> &mut Self {
        self.op_flags = flags;
        self
    }

    /// Force a driver type to use.
    ///
    /// It is ignored if the fusion driver is disabled.
    pub fn driver_type(&mut self, t: DriverType) -> &mut Self {
        self.driver_type = Some(t);
        self
    }

    /// Number of buffers in the buffer pool.
    ///
    /// `size` will be rounded up if it's not power of 2.
    ///
    /// Default to be `8`.
    pub fn buffer_pool_size(&mut self, size: NonZero<u16>) -> &mut Self {
        self.buffer_pool_size = size.get();
        self
    }

    /// Flag to be used to initialize buffer pool.
    ///
    /// This is only supported on io-uring driver.
    ///
    /// Default to be `0`.
    pub fn buffer_pool_flag(&mut self, flag: u16) -> &mut Self {
        self.buffer_pool_flag = flag;
        self
    }

    /// Length of each buffer pool's buffer.
    ///
    /// Default to be `8192`.
    pub fn buffer_pool_buffer_len(&mut self, size: usize) -> &mut Self {
        self.buffer_pool_buffer_len = size;
        self
    }

    /// Set the allocator for buffer pool.
    ///
    /// This is different from the std's unstable `Allocator` trait: it's purely
    /// static and doesn't take an instance at all. This means implementation
    /// should be global (e.g., `Global`, `malloc` or `mmap`).
    ///
    /// Default to [`BoxAllocator`].
    ///
    /// # Note
    ///
    /// Default allocator performs [very poor] when using managed i/o on Zen 3,
    /// possibly due to [a bug related to FSRM]. If you observe such a problem,
    /// try swap the allocator to a mmap-based one may solve it.
    ///
    /// [very poor]: https://github.com/compio-rs/compio/issues/472
    /// [a bug related to FSRM]: https://bugs.launchpad.net/ubuntu/+source/glibc/+bug/2030515
    pub fn buffer_pool_allocator<A: BufferAllocator>(&mut self) -> &mut Self {
        self.buffer_pool_allocator = BufferAlloc::new::<A>();
        self
    }

    /// Build the [`Proactor`].
    pub fn build(&self) -> io::Result<Proactor> {
        Proactor::with_builder(self)
    }
}

mod seal {
    use std::io;

    use compio_buf::BufResult;

    pub(crate) trait Seal {}

    impl Seal for io::Error {}
    impl<T> Seal for io::Result<T> {}
    impl<T, B> Seal for BufResult<T, B> {}
}

/// Extension trait for [`io::Error`] and results with it.
#[allow(private_bounds)]
pub trait ErrorExt: seal::Seal {
    #[doc(hidden)]
    fn as_io_error(&self) -> Option<&io::Error>;

    /// Whether the error or result is cancelled.
    fn is_cancelled(&self) -> bool {
        #[cfg(unix)]
        const CANCEL_ERROR: i32 = libc::ECANCELED;
        #[cfg(windows)]
        const CANCEL_ERROR: i32 = windows_sys::Win32::Foundation::ERROR_OPERATION_ABORTED as _;

        self.as_io_error()
            .and_then(io::Error::raw_os_error)
            .is_some_and(|e| e == CANCEL_ERROR)
    }
}

impl ErrorExt for io::Error {
    fn as_io_error(&self) -> Option<&io::Error> {
        Some(self)
    }
}

impl<T> ErrorExt for io::Result<T> {
    fn as_io_error(&self) -> Option<&io::Error> {
        self.as_ref().err()
    }
}

impl<T, B> ErrorExt for BufResult<T, B> {
    fn as_io_error(&self) -> Option<&io::Error> {
        self.0.as_io_error()
    }
}