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 {
    use super::*;
    use crate::test_support::sfd;
    use core::num::{NonZeroU32, NonZeroUsize};
    use std::sync::Arc;
    use std::time::Duration;

    fn nz32(n: u32) -> NonZeroU32 {
        NonZeroU32::new(n).expect("n ≠ 0")
    }

    fn nzusize(n: usize) -> NonZeroUsize {
        NonZeroUsize::new(n).expect("n ≠ 0")
    }

    /// Assertion **statique** : `LockedIoUring` est `Send + Sync` (le `!Sync`
    /// d'`IoUring` est prouvé par le doctest `compile_fail` sur le type).
    #[test]
    fn locked_io_uring_is_send_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<LockedIoUring>();
        fn assert_send<T: Send>() {}
        assert_send::<IoUring>(); // IoUring est Send (déplaçable entre threads)
        assert_send::<RingPool>();
        assert_send::<SqpollIoUring>();
    }

    // ── LockedIoUring : accès concurrent multi-thread ─────────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring non supporté par Miri")]
    fn locked_io_uring_concurrent_nops() {
        let ring = Arc::new(LockedIoUring::new(nz32(16)).expect("LockedIoUring"));
        let threads: Vec<_> = (0..4)
            .map(|_| {
                let ring = Arc::clone(&ring);
                std::thread::spawn(move || {
                    // Round-trip **atomique** sous le verrou : soumission, attente
                    // et moisson dans une **seule** section critique (`with_lock`).
                    // Indispensable — enchaîner les raccourcis `submit_nop` /
                    // `submit_and_wait` / `wait_completion` prend et relâche le
                    // verrou **trois fois**, ce qui laisserait un autre thread
                    // moissonner *ce* CQE entre-temps (course de complétion, pas
                    // de course SQ/CQ/slab). En tenant le verrou de bout en bout,
                    // la complétion récupérée est **forcément** la sienne.
                    ring.with_lock(|r| {
                        let tok = r.submit_nop().expect("nop");
                        r.submit_and_wait(1).expect("submit_and_wait");
                        let c = r.wait_completion().expect("completion");
                        assert_eq!(c.token(), tok);
                        c.completed().is_ok()
                    })
                })
            })
            .collect();
        for t in threads {
            assert!(t.join().expect("thread"), "nop completed");
        }
        assert_eq!(ring.in_flight(), 0, "aucun slot perdu");
        // Raccourcis nommés (`&self`), en **mono-thread** : chacun prend puis
        // relâche le verrou ; sans concurrence, le token récupéré est le sien.
        let tok = ring.submit_nop().expect("nop via mirror");
        ring.submit_and_wait(1).expect("submit_and_wait via mirror");
        assert_eq!(ring.wait_completion().expect("c").token(), tok);
        assert_eq!(ring.in_flight(), 0);
    }

    // ── RingPool : ATTACH_WQ + routage msg_ring inter-ring ────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring non supporté par Miri")]
    fn ring_pool_routes_msg_ring_between_peers() {
        let pool = RingPool::new(nzusize(3), nz32(8)).expect("RingPool");
        // Poignée vers le worker 1 (pair), capturée avant distribution.
        let handle1 = pool.handle(1).expect("handle 1");
        assert!(pool.handle(3).is_none(), "worker hors borne");
        let mut rings = pool.into_rings();
        assert_eq!(rings.len(), 3, "un ring par worker");
        // worker 1 : attend (poll cq_ready) qu'un message arrive sur son CQ.
        let ring1 = rings.remove(1);
        let peer = std::thread::spawn(move || {
            let ring1 = ring1;
            for _ in 0..2000 {
                if ring1.cq_ready() > 0 {
                    return true; // message reçu (CQE posté par msg_ring du pair)
                }
                std::thread::sleep(Duration::from_millis(1));
            }
            false
        });
        // worker 0 : route un msg_ring vers le pair (handle 1).
        let mut ring0 = rings.remove(0);
        let tok = ring0
            .submit_message_ring_data_to(&handle1, 7, 0xABCD, MessageRingFlags::empty())
            .expect("msg_ring vers le pair");
        // La complétion de l'op msg_ring (côté émetteur) confirme l'envoi.
        ring0.submit_and_wait(1).expect("submit");
        let sent = ring0.wait_completion().expect("completion émetteur");
        assert_eq!(sent.token(), tok);
        sent.completed().expect("msg_ring envoyé");
        assert!(
            peer.join().expect("thread pair"),
            "le pair a reçu le message"
        );
    }

    // ── SqpollIoUring : soumission + réveil NEED_WAKEUP ───────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring SQPOLL non supporté par Miri")]
    fn sqpoll_submits_and_wakes_after_idle() {
        // idle court (10 ms) pour exercer le réveil NEED_WAKEUP après inactivité.
        let mut ring = match SqpollIoUring::new(nz32(8), Duration::from_millis(10), None) {
            Ok(r) => r,
            // SQPOLL peut être refusé selon la config (privilège/cgroup) → on signale.
            Err(e) => {
                assert!(
                    matches!(e, Errno::EPERM),
                    "SQPOLL indisponible : {e:?} (attendu EPERM si refusé)"
                );
                return;
            }
        };
        // 1ère soumission : thread SQPOLL éveillé (vient d'être créé) → pas de syscall.
        let tok = ring.submit_nop().expect("nop");
        ring.submit()
            .expect("submit (SQPOLL, sans enter en régime établi)");
        ring.submit_and_wait(1).expect("submit_and_wait");
        let c = ring.wait_completion().expect("completion");
        assert_eq!(c.token(), tok);
        c.completed().expect("nop ok");
        assert_eq!(ring.in_flight(), 0);
        // Laisse le thread SQPOLL s'endormir (> idle), puis re-soumets : la façade
        // doit détecter NEED_WAKEUP et réveiller le thread (chemin SQ_WAKEUP).
        std::thread::sleep(Duration::from_millis(60));
        let tok2 = ring.submit_nop().expect("nop 2");
        // submit() après idle : NEED_WAKEUP détecté → réveil (chemin SQ_WAKEUP).
        ring.submit().expect("submit (réveil NEED_WAKEUP)");
        ring.submit_and_wait(1).expect("submit_and_wait après idle");
        let c2 = ring.wait_completion().expect("completion 2");
        assert_eq!(c2.token(), tok2);
        c2.completed().expect("nop 2 ok (réveil NEED_WAKEUP)");
        // Le thread SQPOLL est maintenant **éveillé** : une rafale de soumissions
        // immédiates exerce le chemin **sans syscall** (`!NEED_WAKEUP`).
        for _ in 0..16 {
            let tok = ring.submit_nop().expect("nop rafale");
            ring.submit().expect("submit éveillé (sans enter)");
            ring.submit_and_wait(1).expect("drain");
            assert_eq!(ring.wait_completion().expect("c").token(), tok);
        }
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring SQPOLL non supporté par Miri")]
    fn sqpoll_with_cpu_affinity_round_trips() {
        // SQ_AFF : épingle le thread SQPOLL sur le CPU 0 (présent sur tout hôte).
        let mut ring = match SqpollIoUring::new(nz32(8), Duration::from_millis(100), Some(0)) {
            Ok(r) => r,
            Err(e) => {
                assert!(matches!(e, Errno::EPERM), "SQPOLL/SQ_AFF refusé : {e:?}");
                return;
            }
        };
        let tok = ring.submit_nop().expect("nop");
        let n = ring.submit_and_wait(1).expect("submit_and_wait");
        assert!(n <= 1);
        assert_eq!(ring.wait_completion().expect("c").token(), tok);
        assert_eq!(ring.in_flight(), 0);
    }

    // ── Thread-per-core : un ring par thread (modèle §2) ──────────────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring non supporté par Miri")]
    fn thread_per_core_each_thread_owns_its_ring() {
        // 3 threads, un IoUring déplacé (Send) dans chacun, charge indépendante,
        // aucun partage ⇒ aucun verrou.
        let threads: Vec<_> = (0..3)
            .map(|_| {
                std::thread::spawn(|| {
                    let mut ring = IoUring::new(nz32(8)).expect("ring");
                    let tok = ring.submit_nop().expect("nop");
                    ring.submit_and_wait(1).expect("submit");
                    let c = ring.wait_completion().expect("completion");
                    assert_eq!(c.token(), tok);
                    c.completed().is_ok()
                })
            })
            .collect();
        for t in threads {
            assert!(t.join().expect("thread"), "ring indépendant OK");
        }
    }

    // ── Couverture : mirrors LockedIoUring + read SQPOLL + Debug ──────────

    #[test]
    #[cfg_attr(miri, ignore = "io_uring non supporté par Miri")]
    fn locked_io_uring_read_write_and_debug() {
        let path = std::env::temp_dir().join(format!("air-iou3e-{}", std::process::id()));
        {
            std::fs::File::create(&path).expect("create");
        }
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .expect("open rw");
        // `from_builder` (couvre la construction via builder).
        let ring = LockedIoUring::from_builder(IoUringBuilder::new(nz32(8)))
            .expect("LockedIoUring from_builder");
        // Debug ne prend pas le verrou et n'affiche pas l'état du ring.
        assert_eq!(format!("{ring:?}"), "LockedIoUring { .. }");
        // submit_write puis submit() explicite (mirror) puis wait.
        ring.submit_write(sfd(&file), b"shared".to_vec(), Some(0))
            .expect("submit_write");
        assert!(ring.submit().expect("submit") >= 1);
        let (_buf, n) = ring
            .wait_completion()
            .expect("completion")
            .into_buffer_result()
            .expect("write ok");
        assert_eq!(n, 6);
        // submit_read (mirror) + try_completion.
        let tok = ring
            .submit_read(sfd(&file), vec![0u8; 6], Some(0))
            .expect("submit_read");
        ring.submit_and_wait(1).expect("submit_and_wait");
        // try_completion (mirror) récupère la complétion prête.
        let c = ring.try_completion().expect("try_completion prête");
        assert_eq!(c.token(), tok);
        let (buf, n) = c.into_buffer_result().expect("read ok");
        assert_eq!(n, 6);
        assert_eq!(&buf, b"shared");
        assert_eq!(ring.in_flight(), 0);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring non supporté par Miri")]
    fn ring_pool_and_sqpoll_debug_and_extras() {
        let pool = RingPool::new(nzusize(2), nz32(8)).expect("pool");
        assert!(format!("{pool:?}").contains("RingPool"));
        assert!(format!("{pool:?}").contains('2'), "workers affichés");
        drop(pool.into_rings());
        // SQPOLL : Debug + submit_read + submit_and_wait(0) (branche want == 0).
        let mut sq = match SqpollIoUring::new(nz32(8), Duration::from_millis(100), None) {
            Ok(r) => r,
            Err(e) => {
                assert!(matches!(e, Errno::EPERM), "SQPOLL refusé : {e:?}");
                return;
            }
        };
        assert_eq!(format!("{sq:?}"), "SqpollIoUring { .. }");
        // submit_and_wait(0) : publie (sans rien) puis retourne sans attendre.
        assert_eq!(sq.submit_and_wait(0).expect("submit_and_wait(0)"), 0);
        // submit_read sur un fichier.
        use std::io::Write as _;
        let path = std::env::temp_dir().join(format!("air-iou3e-sq-{}", std::process::id()));
        {
            let mut f = std::fs::File::create(&path).expect("create");
            f.write_all(b"sqpoll").expect("write");
        }
        let file = std::fs::File::open(&path).expect("open");
        let tok = sq
            .submit_read(sfd(&file), vec![0u8; 6], Some(0))
            .expect("submit_read SQPOLL");
        sq.submit_and_wait(1).expect("submit_and_wait");
        let completion = sq.wait_completion().expect("completion");
        assert_eq!(completion.token(), tok);
        let (buf, n) = completion.into_buffer_result().expect("read ok");
        assert_eq!(n, 6);
        assert_eq!(&buf, b"sqpoll");
        assert_eq!(sq.in_flight(), 0);
        std::fs::remove_file(&path).ok();
    }

    // ── SQPOLL : bras d'erreur d'`io_uring_enter` (simulateur) ────────────
    //
    // Les deux `io_uring_enter` de `SqpollIoUring` — réveil `SQ_WAKEUP`
    // (`submit`) et attente `GETEVENTS` (`submit_and_wait`) — passent par la
    // **même couture instrumentable** `syscall::enter` que les autres syscalls
    // io_uring (cf. `registration.rs`). Le simulateur y injecte un `-errno` :
    // l'errno transite par le **même** `errno_from_negative_syscall_ret` que le
    // vrai kernel et doit **remonter tel quel** (ADR-021 conv. 2 : `EINTR` non
    // retenté). On couvre ainsi les bras d'erreur de façon **déterministe**.

    #[test]
    #[cfg_attr(miri, ignore = "io_uring SQPOLL non supporté par Miri")]
    fn sqpoll_submit_propagates_enter_error_on_wakeup() {
        use super::syscall::sim::{self, Syscall};
        sim::clear();
        let mut ring = match SqpollIoUring::new(nz32(8), Duration::from_millis(10), None) {
            Ok(r) => r,
            Err(e) => {
                assert!(matches!(e, Errno::EPERM), "SQPOLL refusé : {e:?}");
                return;
            }
        };
        // Prépare un SQE **sans publier**, puis laisse le thread kernel SQPOLL
        // s'endormir : `idle` vaut 10 ms et on attend bien plus — il pose alors
        // `NEED_WAKEUP` sur la SQ. Délai **éprouvé**, identique à
        // `sqpoll_submits_and_wakes_after_idle` (vert en CI sur les deux arches).
        // `submit` empruntera donc le bras `SQ_WAKEUP` → `enter`. Pas de boucle
        // conditionnelle (qui laisserait un corps non exécuté si `NEED_WAKEUP`
        // était déjà posé) : si la précondition n'était pas tenue, le `submit`
        // ci-dessous renverrait `Ok` et l'`expect_err` échouerait bruyamment.
        let _token = ring.submit_nop().expect("nop");
        std::thread::sleep(Duration::from_millis(60));
        // Le simulateur fait échouer l'`enter(SQ_WAKEUP)` ⇒ l'errno doit remonter.
        sim::inject(Syscall::Enter, -4); // -EINTR
        let err = ring.submit().expect_err("enter SQ_WAKEUP en échec → Err");
        assert!(
            matches!(err, Errno::EINTR),
            "EINTR remonté tel quel (conv. 2 ADR-021) : {err:?}"
        );
        sim::clear();
    }

    #[test]
    #[cfg_attr(miri, ignore = "io_uring SQPOLL non supporté par Miri")]
    fn sqpoll_submit_and_wait_propagates_enter_error_on_getevents() {
        use super::syscall::sim::{self, Syscall};
        sim::clear();
        let mut ring = match SqpollIoUring::new(nz32(8), Duration::from_millis(100), None) {
            Ok(r) => r,
            Err(e) => {
                assert!(matches!(e, Errno::EPERM), "SQPOLL refusé : {e:?}");
                return;
            }
        };
        // **Aucun** SQE en attente : `submit` (appelé en tête de
        // `submit_and_wait`) court-circuite sur `to_submit == 0` (aucun
        // `enter`). Le **seul** `enter` émis est donc le `GETEVENTS` d'attente
        // → le simulateur le fait échouer et l'errno doit remonter.
        // Entièrement déterministe (aucune dépendance au timing du thread).
        sim::inject(Syscall::Enter, -4); // -EINTR
        let err = ring
            .submit_and_wait(1)
            .expect_err("enter GETEVENTS en échec → Err");
        assert!(
            matches!(err, Errno::EINTR),
            "EINTR remonté tel quel : {err:?}"
        );
        sim::clear();
    }
}