air_sys_syscall/io_uring/

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

//! **Opérations liées** (Temps 3c) : exprimer qu'une opération ne démarre
//! qu'après l'achèvement de la précédente, **sans aller-retour userspace**.
//! Sous-module `air-sys-syscall::io_uring::linked`. Repose sur les drapeaux SQE
//! `IOSQE_IO_LINK` / `IOSQE_IO_HARDLINK` (axe D) et intègre l'op `LINK_TIMEOUT`
//! (15). **Aucun register opcode.**
//!
//! Référence normative : `docs/specs/layer-0/io-uring-3c-linked.md`.
//!
//! ## Sémantique soft vs hard (point critique)
//!
//! L'ordre est garanti **dans les deux cas** ; ils diffèrent par la propagation
//! des erreurs :
//!
//! - **soft** ([`LinkedChainBuilder::then`], `IOSQE_IO_LINK`) : si un maillon se
//!   termine **en erreur**, la chaîne est **rompue** — les maillons restants sont
//!   complétés avec **`-ECANCELED`**. **Définition large de l'« erreur »** : un
//!   **`read`/`recv` court** (short read — moins d'octets que demandé) **compte
//!   comme erreur** et rompt la chaîne. *Piège classique* : `read(N) → process`
//!   casse si le `read` rend `< N` octets.
//! - **hard** ([`LinkedChainBuilder::then_hard`], `IOSQE_IO_HARDLINK`) : une
//!   erreur de **complétion** d'un maillon **n'interrompt pas** (suivants
//!   exécutés) ; implique l'ordre du soft. Ne protège **pas** d'un **échec de
//!   soumission** du parent (qui rompt quand même).
//!
//! ## Staging + réservation atomique + publication unique
//!
//! Pendant la durée du builder, le `&mut IoUring` est en **mode staging** : un
//! `submit_*` appelé dans la closure réserve un slot S1 et **écrit** un SQE (avec
//! le drapeau de lien posé sur le **prédécesseur**) **sans publier la queue**. La
//! publication n'a lieu qu'au [`LinkedChainBuilder::submit`] final (**un seul
//! store-release**, Temps 1 §3.2 → le kernel voit la chaîne complète d'un coup).
//! On réutilise ainsi les `submit_*` existants **sans duplication** de `prepare_*`.
//! Si un maillon ne peut être réservé (`EBUSY`), le builder **rembobine** (libère
//! les slots déjà réservés, rembobine la `tail`) avant de retourner — **jamais de
//! chaîne partielle publiée**.

use super::owned::OwnedOp;
use super::{IoUring, SubmissionToken, raw};
use air_sys_types::Errno;
use air_sys_types::io_uring::{TimeoutFlags, TimeoutSpec};
use alloc::boxed::Box;
use alloc::vec::Vec;

impl IoUring {
    /// Démarre la construction d'une **chaîne liée** (mode staging).
    ///
    /// Les `submit_*` appelés via le builder sont **mis en attente** (slot S1
    /// réservé + SQE écrit, lien posé sur le prédécesseur) sans publier la queue ;
    /// [`LinkedChainBuilder::submit`] publie la chaîne entière en une fois.
    pub fn link_chain(&mut self) -> LinkedChainBuilder<'_> {
        let start_tail = self.sq.local_tail();
        LinkedChainBuilder {
            ring: self,
            start_tail,
            tokens: Vec::new(),
            prev_index: None,
        }
    }

    /// `IORING_OP_LINK_TIMEOUT` (15) **staged** : borne le maillon précédent. Le
    /// `__kernel_timespec` est gardé en vie dans le slot S1 (comme `submit_timeout`).
    /// Réservé au [`LinkedChainBuilder`].
    fn submit_link_timeout_staged(
        &mut self,
        spec: TimeoutSpec,
        flags: TimeoutFlags,
    ) -> Result<SubmissionToken, Errno> {
        let ts = Box::new(super::async_ops::timespec_of(spec));
        let ts_ptr = core::ptr::from_ref(&*ts) as u64;
        let op_flags = flags.bits();
        self.submit_op(Some(OwnedOp::Timeout(ts)), |sqe| {
            sqe.opcode = raw::IORING_OP_LINK_TIMEOUT;
            sqe.fd = -1;
            sqe.addr_or_splice_off_in = ts_ptr;
            sqe.len = 1;
            sqe.op_flags = op_flags;
        })
    }
}

/// Constructeur d'une **chaîne d'opérations liées** sur un [`IoUring`] en mode
/// staging (cf. module). Emprunte le ring `&mut` pour toute sa durée (aucune
/// soumission directe ne peut s'intercaler).
pub struct LinkedChainBuilder<'ring> {
    /// Ring emprunté exclusivement (staging).
    ring: &'ring mut IoUring,
    /// `tail` locale avant le premier maillon (borne de rollback).
    start_tail: u32,
    /// Jetons des maillons stagés, dans l'ordre.
    tokens: Vec<SubmissionToken>,
    /// Index (dans l'anneau SQ) du **dernier** maillon stagé, pour y poser le
    /// drapeau de lien au maillon suivant. `None` avant le premier.
    prev_index: Option<u32>,
}

impl LinkedChainBuilder<'_> {
    /// Cœur commun : stage un maillon (`op` = un `submit_*` normal). Si
    /// `link_flag` est `Some`, le pose sur le **prédécesseur**. En cas d'échec de
    /// réservation, **rembobine** la chaîne et propage l'erreur.
    fn stage<F>(mut self, link_flag: Option<u8>, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        let index = self.ring.sq.local_tail();
        match op(self.ring) {
            Ok(token) => {
                if let (Some(flag), Some(prev)) = (link_flag, self.prev_index) {
                    // SAFETY: `prev` désigne le SQE du maillon précédent, stagé
                    // pendant ce builder et **non publié** ; accès exclusif (`&mut`).
                    unsafe {
                        self.ring.sq.or_sqe_flags(prev, flag);
                    }
                }
                self.tokens.push(token);
                self.prev_index = Some(index);
                Ok(self)
            }
            Err(error) => {
                self.rollback();
                Err(error)
            }
        }
    }

    /// **Premier maillon** : `op` appelle un `submit_*` normal, mis en attente
    /// (staging). Ne pose aucun drapeau de lien (pas de prédécesseur).
    ///
    /// # Errors
    ///
    /// L'erreur du `submit_*` (ex. [`Errno::EBUSY`] si la SQ/le slab sont pleins) ;
    /// la chaîne est alors rembobinée.
    pub fn first<F>(self, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        self.stage(None, op)
    }

    /// Maillon **soft-lié** au précédent (`IOSQE_IO_LINK` posé sur le précédent) :
    /// une **erreur de complétion** du précédent (`-errno`, **short read inclus**)
    /// rompt la chaîne (suivants `-ECANCELED`).
    ///
    /// # Errors
    ///
    /// Voir [`LinkedChainBuilder::first`].
    pub fn then<F>(self, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        self.stage(Some(raw::IOSQE_IO_LINK), op)
    }

    /// Maillon **hard-lié** au précédent (`IOSQE_IO_HARDLINK` posé sur le
    /// précédent) : une erreur de complétion du précédent **n'interrompt pas** la
    /// chaîne (l'ordre reste garanti).
    ///
    /// # Errors
    ///
    /// Voir [`LinkedChainBuilder::first`].
    pub fn then_hard<F>(self, op: F) -> Result<Self, Errno>
    where
        F: FnOnce(&mut IoUring) -> Result<SubmissionToken, Errno>,
    {
        self.stage(Some(raw::IOSQE_IO_HARDLINK), op)
    }

    /// Borne dans le temps le **maillon précédent** via un `LINK_TIMEOUT` (op 15) :
    /// si le timeout expire avant la fin du maillon, le maillon est annulé
    /// (`-ECANCELED`) et la chaîne rompue (le timeout complète) ; si le maillon
    /// finit avant, le timeout est annulé. **Unique** manière sûre d'émettre un
    /// `LINK_TIMEOUT` (émission isolée refusée, Temps 2c).
    ///
    /// # Errors
    ///
    /// [`Errno::EINVAL`] s'il n'y a **aucun** maillon à borner ; sinon l'erreur du
    /// staging (chaîne rembobinée).
    pub fn with_link_timeout(
        mut self,
        spec: TimeoutSpec,
        flags: TimeoutFlags,
    ) -> Result<Self, Errno> {
        let Some(prev) = self.prev_index else {
            // Pas de maillon précédent à borner.
            self.rollback();
            return Err(Errno::EINVAL);
        };
        let index = self.ring.sq.local_tail();
        match self.ring.submit_link_timeout_staged(spec, flags) {
            Ok(token) => {
                // SAFETY: `prev` (maillon borné) est stagé non publié ; on le lie
                // (`IO_LINK`) au `LINK_TIMEOUT` qui le suit immédiatement. Accès
                // exclusif (`&mut`).
                unsafe {
                    self.ring.sq.or_sqe_flags(prev, raw::IOSQE_IO_LINK);
                }
                self.tokens.push(token);
                self.prev_index = Some(index);
                Ok(self)
            }
            Err(error) => {
                self.rollback();
                Err(error)
            }
        }
    }

    /// Publie la chaîne **entière, atomiquement** (un seul store-release + un
    /// `io_uring_enter`). Rend les jetons dans l'ordre de soumission.
    ///
    /// # Errors
    ///
    /// Les erreurs de [`IoUring::submit`] ([`Errno::EINTR`] remonté tel quel, etc.).
    /// La réservation des slots ayant déjà réussi au staging, l'atomicité de la
    /// chaîne est garantie en amont (jamais de chaîne partielle).
    pub fn submit(mut self) -> Result<ChainTokens, Errno> {
        self.ring.submit()?;
        Ok(ChainTokens {
            tokens: core::mem::take(&mut self.tokens),
        })
    }

    /// Rembobine le staging : libère les slots déjà réservés (idempotent côté
    /// slab) et rembobine la `tail` locale → aucun SQE de la chaîne ne sera publié.
    fn rollback(&mut self) {
        let tokens = core::mem::take(&mut self.tokens);
        for token in tokens {
            self.ring.slab.release(token);
        }
        self.ring.sq.rewind_to(self.start_tail);
        self.prev_index = None;
    }
}

/// Jetons des maillons d'une chaîne, **dans l'ordre de soumission** (corrèle
/// chaque complétion à son maillon).
#[derive(Debug, Clone)]
pub struct ChainTokens {
    tokens: Vec<SubmissionToken>,
}

impl ChainTokens {
    /// Jetons des maillons, dans l'ordre de soumission.
    #[must_use]
    pub fn tokens(&self) -> &[SubmissionToken] {
        &self.tokens
    }

    /// Jeton du maillon `index` (ordre de soumission), le cas échéant.
    #[must_use]
    pub fn get(&self, index: usize) -> Option<SubmissionToken> {
        self.tokens.get(index).copied()
    }

    /// Nombre de maillons de la chaîne.
    #[must_use]
    pub fn len(&self) -> usize {
        self.tokens.len()
    }

    /// `true` si la chaîne est vide.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.tokens.is_empty()
    }

    /// Consomme et rend les jetons possédés.
    #[must_use]
    pub fn into_vec(self) -> Vec<SubmissionToken> {
        self.tokens
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Tests
// ───────────────────────────────────────────────────────────────────────────
//
// Les chaînes touchent `io_uring_enter` (non modélisé par Miri) →
// `#[cfg_attr(miri, ignore)]`. Les accesseurs de `ChainTokens` (logique pure)
// tournent **sous Miri**. Les erreurs déterministes sont injectées via
// `submit_nop_with_result` (`IORING_NOP_INJECT_RESULT`) ; le piège du **short
// read** est testé sur un pipe réel.

#[cfg(test)]
mod tests;