air_sys_syscall/io_uring/

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

//! **Usage multi-thread** d'io_uring (Temps 3e). Sous-module
//! `air-sys-syscall::io_uring::shared`. Repose sur l'invariant **ADR-022
//! Décision 6** ([`IoUring`] est `Send` mais **`!Sync`**), sur `SETUP_SQPOLL`
//! (Temps 1), `ATTACH_WQ` (Temps 3a) et `msg_ring` (Temps 2c). **Aucun register
//! opcode propre.**
//!
//! Référence normative : `docs/specs/layer-0/io-uring-3e-shared.md`.
//!
//! ## Le point de départ : `Send` mais pas `Sync`
//!
//! Le protocole d'ordering du Temps 1 (§3.2) synchronise *userspace ↔ kernel*,
//! **pas** *userspace ↔ userspace* : têtes/queues SQ/CQ et **slab S1** ne sont
//! pas protégés contre des accès userspace concurrents. Le `!Sync` d'[`IoUring`]
//! **interdit par typage** le partage accidentel d'un ring (prouvé par un
//! doctest `compile_fail` sur [`LockedIoUring`]). Trois réponses, par ordre de
//! préférence Air :
//!
//! 1. **Thread-per-core (§2, recommandé)** — un [`IoUring`] par thread
//!    (`SINGLE_ISSUER | DEFER_TASKRUN`), **aucun verrou** ; communication par
//!    `msg_ring`. Aucun type nouveau : usage direct d'[`IoUring`], déplacé
//!    (`Send`) dans son thread. [`RingPool`] **organise** N tels rings.
//! 2. **[`LockedIoUring`] (§3)** — un ring partagé derrière un verrou
//!    (`Send + Sync`). Simple, mais le verrou est un **point de contention**.
//! 3. **[`SqpollIoUring`] (§5)** — un thread kernel scrute la SQ (latence vs CPU).

use super::{Completion, IoUring, IoUringBuilder, SetupFlags, SubmissionToken, raw, syscall};
use air_sys_types::Errno;
use air_sys_types::fd::BorrowedFd;
use air_sys_types::io_uring::MessageRingFlags;
use alloc::vec::Vec;
use core::num::{NonZeroU32, NonZeroUsize};
use core::time::Duration;

use crate::sync::FutexMutex;

// ───────────────────────────────────────────────────────────────────────────
// LockedIoUring — partage simple par verrou (§3)
// ───────────────────────────────────────────────────────────────────────────

/// Un [`IoUring`] **partagé** entre threads derrière un **verrou interne**
/// (`Send + Sync`). Le verrou sérialise les accès userspace au SQ/CQ/slab ; le
/// protocole d'ordering userspace↔kernel du Temps 1 reste inchangé dessous.
///
/// **Avertissement (Principe 5)** : le verrou est un **point de contention** —
/// sous forte charge multi-thread il sérialise tout. À réserver aux cas simples
/// ou peu sollicités ; pour la performance, préférer le **thread-per-core** (§2,
/// [`RingPool`]). **Verrou unique** d'abord (Principe 7) ; un affinage
/// (soumission/complétion séparées) n'est envisagé qu'**après mesure**.
///
/// **Sûreté** : `Send + Sync` par construction — [`IoUring`] est `Send`, donc
/// `FutexMutex<IoUring>` est `Send + Sync` :
///
/// ```
/// fn assert_send_sync<T: Send + Sync>() {}
/// assert_send_sync::<air_sys_syscall::io_uring::LockedIoUring>();
/// ```
///
/// alors qu'un [`IoUring`] nu n'est **pas** `Sync` (partage refusé par typage) :
///
/// ```compile_fail
/// fn assert_sync<T: Sync>() {}
/// assert_sync::<air_sys_syscall::io_uring::IoUring>(); // ERREUR : IoUring: !Sync
/// ```
pub struct LockedIoUring {
    /// `FutexMutex<IoUring>` : `IoUring` est `Send`, donc `FutexMutex<IoUring>`
    /// est `Send + Sync` → `LockedIoUring` devient partageable (`Sync`). Le
    /// verrou est un mutex futex-maison `std`-free (cf. [`crate::sync`]).
    inner: FutexMutex<IoUring>,
}

impl core::fmt::Debug for LockedIoUring {
    // `IoUring` n'est pas `Debug` (état de ring opaque) ; on n'affiche pas son
    // contenu (et on ne prend pas le verrou pour formater).
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.write_str("LockedIoUring { .. }")
    }
}

impl LockedIoUring {
    /// Raccourci : `LockedIoUring::from_builder(IoUringBuilder::new(entries))`.
    ///
    /// # Errors
    ///
    /// Voir [`IoUringBuilder::build`].
    pub fn new(entries: NonZeroU32) -> Result<Self, Errno> {
        Ok(Self {
            inner: FutexMutex::new(IoUring::new(entries)?),
        })
    }

    /// Construit à partir d'un [`IoUringBuilder`] (flags, SQPOLL, etc.).
    ///
    /// # Errors
    ///
    /// Voir [`IoUringBuilder::build`].
    pub fn from_builder(builder: IoUringBuilder) -> Result<Self, Errno> {
        Ok(Self {
            inner: FutexMutex::new(builder.build()?),
        })
    }

    /// Exécute `f` sur le ring sous-jacent **sous le verrou** (accès exclusif
    /// sérialisé). C'est la primitive complète : toute opération d'[`IoUring`]
    /// (les ~50 `submit_*`, les accesseurs de complétion) est accessible ainsi en
    /// `&self`, sans recopier 50 forwarders. Les méthodes nommées ci-dessous en
    /// sont de simples raccourcis pour les cas les plus fréquents.
    ///
    /// Le verrou est **sans empoisonnement** ([`FutexMutex`] n'implémente pas le
    /// poisoning) : un panic d'un autre thread sous le verrou ne le rend pas
    /// inutilisable (cohérent avec la future synchro sans poisoning d'`air-thread`).
    pub fn with_lock<R>(&self, f: impl FnOnce(&mut IoUring) -> R) -> R {
        let mut guard = self.inner.lock();
        f(&mut guard)
    }

    /// `submit_read` (Temps 2a) sous le verrou.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit_read`].
    pub fn submit_read(
        &self,
        fd: BorrowedFd<'_>,
        buffer: Vec<u8>,
        offset: Option<u64>,
    ) -> Result<SubmissionToken, Errno> {
        self.with_lock(|ring| ring.submit_read(fd, buffer, offset))
    }

    /// `submit_write` (Temps 2a) sous le verrou.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit_write`].
    pub fn submit_write(
        &self,
        fd: BorrowedFd<'_>,
        buffer: Vec<u8>,
        offset: Option<u64>,
    ) -> Result<SubmissionToken, Errno> {
        self.with_lock(|ring| ring.submit_write(fd, buffer, offset))
    }

    /// `submit_nop` (Temps 1) sous le verrou.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit_nop`].
    pub fn submit_nop(&self) -> Result<SubmissionToken, Errno> {
        self.with_lock(|ring| ring.submit_nop())
    }

    /// `submit` (Temps 1) sous le verrou.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit`].
    pub fn submit(&self) -> Result<u32, Errno> {
        self.with_lock(|ring| ring.submit())
    }

    /// `submit_and_wait` (Temps 1) sous le verrou.
    ///
    /// **Attention** : l'attente bloque **avec le verrou tenu** — un seul thread
    /// attend à la fois. Pour de l'attente concurrente, préférer le thread-per-core.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit_and_wait`].
    pub fn submit_and_wait(&self, want: u32) -> Result<u32, Errno> {
        self.with_lock(|ring| ring.submit_and_wait(want))
    }

    /// `wait_completion` (Temps 1) sous le verrou.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::wait_completion`].
    pub fn wait_completion(&self) -> Result<Completion, Errno> {
        self.with_lock(|ring| ring.wait_completion())
    }

    /// `try_completion` (Temps 1) sous le verrou.
    #[must_use]
    pub fn try_completion(&self) -> Option<Completion> {
        self.with_lock(|ring| ring.try_completion())
    }

    /// Nombre d'opérations en vol (sous le verrou).
    #[must_use]
    pub fn in_flight(&self) -> u32 {
        self.with_lock(|ring| ring.in_flight())
    }
}

// ───────────────────────────────────────────────────────────────────────────
// RingPool — assistant thread-per-core (§4)
// ───────────────────────────────────────────────────────────────────────────

/// Assistant **thread-per-core** : crée N [`IoUring`] cohérents en **partageant
/// le pool io-wq** (`ATTACH_WQ`, Temps 3a) — le nombre total de threads workers
/// kernel est **borné** quel que soit N (**crucial sur matériel modeste**, Pi 4)
/// — et en **enregistrant les ring fds** (Temps 3a) pour le routage `msg_ring`.
///
/// `RingPool` **organise, ne partage pas** : chaque [`IoUring`] rendu reste
/// `Send`/`!Sync` et est **déplacé** dans son thread, jamais partagé.
pub struct RingPool {
    /// Un ring possédé par worker ; le ring 0 détient le pool io-wq, les autres
    /// s'y attachent. Ring fds enregistrés (msg_ring découplé).
    rings: Vec<IoUring>,
}

impl core::fmt::Debug for RingPool {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("RingPool")
            .field("workers", &self.rings.len())
            .finish_non_exhaustive()
    }
}

impl RingPool {
    /// Crée `workers` rings : le **premier** crée le pool io-wq, les **suivants**
    /// s'y attachent (`ATTACH_WQ`) → nombre total de threads workers kernel borné.
    /// Chaque ring fd est enregistré (`REGISTER_RING_FDS`, Temps 3a).
    ///
    /// # Errors
    ///
    /// Erreurs d'`io_uring_setup`/`io_uring_register` (cf. [`IoUringBuilder::build`],
    /// [`IoUring::register_ring_fd`]).
    pub fn new(workers: NonZeroUsize, entries: NonZeroU32) -> Result<Self, Errno> {
        let count = workers.get();
        let mut rings: Vec<IoUring> = Vec::with_capacity(count);
        // Premier ring : crée le pool io-wq.
        rings.push(IoUring::new(entries)?);
        // Suivants : s'attachent au pool du premier (workers kernel bornés).
        for _ in 1..count {
            let attached = {
                let first = &rings[0];
                IoUringBuilder::new(entries)
                    .attach_work_queue(first)
                    .build()?
            };
            rings.push(attached);
        }
        // Enregistre chaque ring fd (routage msg_ring découplé, Temps 3a).
        for ring in &mut rings {
            ring.register_ring_fd()?;
        }
        Ok(Self { rings })
    }

    /// Distribue les rings : un [`IoUring`] **possédé** par worker, à **déplacer**
    /// dans chaque thread.
    #[must_use]
    pub fn into_rings(self) -> Vec<IoUring> {
        self.rings
    }

    /// Poignée de routage vers le ring du worker `worker` (cible `msg_ring`).
    /// `None` si `worker` est hors borne.
    #[must_use]
    pub fn handle(&self, worker: usize) -> Option<RingHandle> {
        self.rings.get(worker).map(|ring| RingHandle {
            target_fd: ring.fd_raw(),
        })
    }
}

/// Référence à un ring **pair**, utilisable comme **cible de `msg_ring`**
/// ([`IoUring::submit_message_ring_data_to`]). Porte le FD du ring pair (valide
/// tant que ce ring vit dans son thread — l'appelant garantit sa vivacité).
#[derive(Debug, Clone, Copy)]
pub struct RingHandle {
    /// FD du ring cible (process-global ; valable cross-thread).
    target_fd: i32,
}

impl IoUring {
    /// `msg_ring` (Temps 2c, `MSG_DATA`) ciblant un **pair** par sa [`RingHandle`]
    /// (FD), pour réveiller/transmettre une donnée vers un ring **d'un autre
    /// thread** (modèle thread-per-core / [`RingPool`]). Variante de
    /// [`IoUring::submit_message_ring_data`] (qui prend une `&IoUring`, même
    /// contexte).
    ///
    /// # Errors
    ///
    /// [`Errno::EBUSY`] si la SQ/le slab sont pleins ; à la complétion, `EBADF`
    /// si le ring cible n'existe plus.
    pub fn submit_message_ring_data_to(
        &mut self,
        target: &RingHandle,
        res: i32,
        user_data: u64,
        flags: MessageRingFlags,
    ) -> Result<SubmissionToken, Errno> {
        let target_fd = target.target_fd;
        let len = res.cast_unsigned();
        let op_flags = flags.bits();
        self.submit_op(None, |sqe| {
            sqe.opcode = raw::IORING_OP_MSG_RING;
            sqe.fd = target_fd;
            sqe.addr_or_splice_off_in = raw::IORING_MSG_DATA;
            sqe.len = len;
            sqe.off_or_addr2 = user_data;
            sqe.op_flags = op_flags;
        })
    }
}

// ───────────────────────────────────────────────────────────────────────────
// SqpollIoUring — thread kernel de poll de la SQ (§5)
// ───────────────────────────────────────────────────────────────────────────

/// [`IoUring`] créé avec `IORING_SETUP_SQPOLL` : un **thread kernel** scrute la
/// SQ. En régime établi, la **soumission ne fait aucun `io_uring_enter`** (juste
/// la publication release) → latence minimale.
///
/// **Coût** : un **cœur kernel** dédié à scruter. Excellent sous forte charge où
/// le CPU abonde ; **mauvais choix sur Pi 4** (cœurs rares). À réserver aux
/// profils où la latence prime (mesure, Principe 5).
pub struct SqpollIoUring {
    inner: IoUring,
}

impl core::fmt::Debug for SqpollIoUring {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.write_str("SqpollIoUring { .. }")
    }
}

impl SqpollIoUring {
    /// Crée un ring `SQPOLL`. `idle` : délai d'inactivité avant que le thread
    /// kernel ne s'endorme (`NEED_WAKEUP`). `cpu` : épinglage optionnel du thread
    /// (`SQ_AFF`).
    ///
    /// # Errors
    ///
    /// [`Errno::EPERM`] si `SQPOLL` n'est pas autorisé par la config ; autres
    /// erreurs de [`IoUringBuilder::build`].
    pub fn new(entries: NonZeroU32, idle: Duration, cpu: Option<u32>) -> Result<Self, Errno> {
        let mut flags = SetupFlags::SQPOLL;
        let mut builder = IoUringBuilder::new(entries);
        if let Some(cpu) = cpu {
            // `SQ_AFF` : épingle le thread kernel sur `cpu`.
            flags |= SetupFlags::SQ_AFF;
            builder = builder.with_sqpoll_cpu(cpu);
        }
        builder = builder.with_flags(flags).with_sqpoll_idle(idle);
        Ok(Self {
            inner: builder.build()?,
        })
    }

    /// Soumet les SQE en attente. En `SQPOLL`, **aucun `io_uring_enter`** n'est
    /// émis si le thread kernel est éveillé (il consommera la SQ) ; s'il s'est
    /// endormi (`IORING_SQ_NEED_WAKEUP`), la façade le **réveille** automatiquement
    /// via `io_uring_enter(.., SQ_WAKEUP)` — **seule magie cachée**, indispensable
    /// et documentée. Retourne le nombre de SQE en attente.
    ///
    /// # Errors
    ///
    /// [`Errno::EINTR`] remonté tel quel (ADR-021 conv. 2) ; autres errno d'`enter`.
    pub fn submit(&mut self) -> Result<u32, Errno> {
        let to_submit = self.inner.sq.publish_and_pending();
        if to_submit == 0 {
            return Ok(0);
        }
        if self.inner.sq.flags() & raw::IORING_SQ_NEED_WAKEUP != 0 {
            // Thread SQPOLL endormi : le réveiller.
            // SAFETY: `fd` est un ring fd valide ; pas d'argument étendu.
            let ret = unsafe {
                syscall::enter(
                    self.inner.fd_raw(),
                    to_submit,
                    0,
                    raw::IORING_ENTER_SQ_WAKEUP,
                    0,
                    0,
                )
            };
            if ret < 0 {
                return Err(raw::errno_from_negative_syscall_ret(ret));
            }
            Ok(u32::try_from(ret).unwrap_or(0))
        } else {
            // Thread SQPOLL éveillé : il consommera la SQ, aucun syscall requis.
            Ok(to_submit)
        }
    }

    /// `submit_nop` (Temps 1) : prépare un SQE (sans publier — [`Self::submit`] ou
    /// [`Self::submit_and_wait`] publie).
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit_nop`].
    pub fn submit_nop(&mut self) -> Result<SubmissionToken, Errno> {
        self.inner.submit_nop()
    }

    /// `submit_read` (Temps 2a) en mode `SQPOLL`.
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::submit_read`].
    pub fn submit_read(
        &mut self,
        fd: BorrowedFd<'_>,
        buffer: Vec<u8>,
        offset: Option<u64>,
    ) -> Result<SubmissionToken, Errno> {
        self.inner.submit_read(fd, buffer, offset)
    }

    /// Publie (avec réveil `SQPOLL` si nécessaire) **puis** attend au moins `want`
    /// complétions (`io_uring_enter(GETEVENTS)`). L'attente requiert un syscall
    /// (la soumission, elle, peut n'en demander aucun).
    ///
    /// # Errors
    ///
    /// [`Errno::EINTR`] remonté tel quel ; autres errno d'`enter`.
    pub fn submit_and_wait(&mut self, want: u32) -> Result<u32, Errno> {
        self.submit()?;
        if want == 0 {
            return Ok(0);
        }
        // SAFETY: `fd` valide ; on attend `want` complétions, pas d'argument étendu.
        let ret = unsafe {
            syscall::enter(
                self.inner.fd_raw(),
                0,
                want,
                raw::IORING_ENTER_GETEVENTS,
                0,
                0,
            )
        };
        if ret < 0 {
            return Err(raw::errno_from_negative_syscall_ret(ret));
        }
        Ok(u32::try_from(ret).unwrap_or(0))
    }

    /// `wait_completion` (Temps 1).
    ///
    /// # Errors
    ///
    /// Voir [`IoUring::wait_completion`].
    pub fn wait_completion(&mut self) -> Result<Completion, Errno> {
        self.inner.wait_completion()
    }

    /// Nombre d'opérations en vol.
    #[must_use]
    pub fn in_flight(&self) -> u32 {
        self.inner.in_flight()
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Preuve loom — discipline de verrou de LockedIoUring (sous `cfg(loom)`)
// ───────────────────────────────────────────────────────────────────────────
//
// Exécution : `RUSTFLAGS="--cfg loom" cargo test -p air-sys-syscall --lib shared::loom_proof`.
// `LockedIoUring` enveloppe l'`IoUring` (SQ/CQ/slab non protégés) dans un `Mutex` :
// la sûreté repose entièrement sur l'**exclusion mutuelle** du verrou. Comme un
// `IoUring` réel n'est pas constructible sous `loom::model` (pas de syscall), on
// modélise la **discipline** : deux threads accèdent à un état partagé gardé
// (représentant SQ-tail/slab) UNIQUEMENT sous le verrou. loom explore tous les
// entrelacements et prouve l'absence de course (cellule gardée) et de mise à
// jour perdue. Si l'accès gardé fuyait hors du verrou, loom le détecterait.
#[cfg(loom)]
mod loom_proof {
    use loom::cell::UnsafeCell;
    use loom::sync::{Arc, Mutex};

    /// Deux threads incrémentent un état partagé **sous le verrou** : aucune mise
    /// à jour perdue (somme exacte), aucun accès concurrent à la cellule.
    #[test]
    fn mutex_serializes_shared_ring_state() {
        loom::model(|| {
            // `Mutex<()>` garde l'accès à la cellule (état userspace du ring).
            let state = Arc::new((Mutex::new(()), UnsafeCell::new(0u32)));

            let writer = {
                let state = Arc::clone(&state);
                loom::thread::spawn(move || {
                    let (lock, cell) = &*state;
                    let _guard = lock.lock().unwrap();
                    // SAFETY: accès exclusif garanti par le verrou tenu ; loom
                    // vérifie qu'aucun autre thread ne touche la cellule sans le
                    // verrou (modèle de `LockedIoUring::with_lock`).
                    cell.with_mut(|p| unsafe { *p += 1 });
                })
            };

            {
                let (lock, cell) = &*state;
                let _guard = lock.lock().unwrap();
                // SAFETY: idem — sous le verrou, accès exclusif à la cellule.
                cell.with_mut(|p| unsafe { *p += 1 });
            }

            writer.join().unwrap();

            let (_lock, cell) = &*state;
            // SAFETY: les deux threads ont terminé (join) ; lecture finale exclusive.
            let total = cell.with(|p| unsafe { *p });
            assert_eq!(total, 2, "aucune mise à jour perdue sous le verrou");
        });
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Tests
// ───────────────────────────────────────────────────────────────────────────
//
// Intégration kernel (io_uring non modélisé par Miri) → `#[cfg_attr(miri, ignore)]`.
// La sûreté multi-thread est prouvée à **trois** niveaux : (1) `!Sync` d'`IoUring`
// et `Send + Sync` de `LockedIoUring` par **doctests `compile_fail`/compile-pass**
// (sur le type `LockedIoUring`) + assertion statique ci-dessous ; (2) **loom** sur
// la discipline de verrou (`mod loom_proof`, sous `cfg(loom)`) ; (3) tests
// d'intégration concurrents réels (threads).

#[cfg(test)]
mod tests;