air_sys_syscall/

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

//! Famille `futex` — blocage/réveil noyau sur un mot 32 bits (`futex(2)`).
//!
//! Ajoutée par **ADR-048** (descellement unique de la couche 0 : `std`-free +
//! `futex(2)`, re-sceau `couche-0-v1.6`) pour fournir la primitive de
//! synchronisation `std`-free dont dépendent la libc Air (`air-thread`, couche 1)
//! **et** le mutex interne de l'état io_uring partagé de la couche 0 (qui remplace
//! ainsi son dernier `std::sync::Mutex`). Cf. `docs/specs/layer-0/family-futex.md`.
//!
//! Doctrine « kernel = bible », alignée sur la famille `poll` (ADR-044) :
//!
//! - **`futex(2)` classique** (`SYS_futex`) sur un [`AtomicU32`] de l'appelant —
//!   distinct du futex *via io_uring* (trop lourd pour un `Mutex` basique, cf.
//!   ADR-048 alternatives écartées).
//! - **Une fonction typée par opération** (ADR-021 §3) : [`futex_wait`] et
//!   [`futex_wake`]. Aucun multiplexeur `futex(op, …)` exposé ; l'`op` brut reste
//!   un détail interne du wrapper asm.
//! - **`EINTR` remonté, jamais de retry** (ADR-021 §2) : une attente interrompue
//!   par un signal renvoie [`Errno::EINTR`] **tel quel** ; la reprise (échéance
//!   recalculée) vit en couche 1. Le wrapper ne contient **aucune** boucle.
//! - **Portée et compte typés** (pas de drapeau brut) : [`FutexScope`] (privé par
//!   défaut) et [`FutexWakeCount`] (borné à `INT_MAX` par construction).
//! - **Pas d'allocation** (Principe 4) : le mot est emprunté (`&AtomicU32`), le
//!   timeout est un `KernelTimespec` de pile (réutilisé de `poll`). Errno =
//!   `NonZeroI32`.

use core::num::NonZeroI32;
use core::sync::atomic::AtomicU32;
use core::time::Duration;

use air_sys_types::{Errno, FutexScope, FutexWakeCount};

use crate::poll::{KernelTimespec, kernel_timespec_from_duration_clamped};

/// `FUTEX_WAIT` (opération 0) : attend tant que `*uaddr == val`.
const FUTEX_WAIT: u32 = 0;
/// `FUTEX_WAKE` (opération 1) : réveille jusqu'à `val` waiters.
const FUTEX_WAKE: u32 = 1;
/// `FUTEX_PRIVATE_FLAG` (`0x80` = 128) : restreint le futex au processus courant
/// (clé basée sur l'adresse virtuelle, pas l'inode du mapping) — plus rapide.
const FUTEX_PRIVATE_FLAG: u32 = 128;

/// Compose le code d'opération `futex(2)` à partir de l'opération de base et de la
/// portée typée. Détail **interne** : la surface publique reste deux fonctions
/// dédiées (ADR-021 §3), jamais un `op` exposé.
const fn futex_op(base: u32, scope: FutexScope) -> u32 {
    match scope {
        // `base | FUTEX_PRIVATE_FLAG` : `bitor`, pas d'arithmétique (aucun overflow).
        FutexScope::Private => base | FUTEX_PRIVATE_FLAG,
        FutexScope::Shared => base,
    }
}

/// `futex(2)` `FUTEX_WAIT` : bloque le thread courant tant que `word == expected`.
///
/// L'appelant fournit un mot futex [`AtomicU32`] et la valeur `expected` qu'il
/// **vient** d'observer. Le noyau revérifie `*word == expected` **atomiquement**
/// avant de s'endormir : si la valeur a déjà changé entre la lecture de l'appelant
/// et l'entrée dans le syscall, l'attente est inutile et renvoie immédiatement
/// [`Errno::EAGAIN`] (c'est l'élimination de la course « réveil perdu »).
///
/// - `timeout = None`     → attente **indéfinie** (timespec `NULL`).
/// - `timeout = Some(d)`  → attente bornée **relative** (`FUTEX_WAIT` mesure contre
///   `CLOCK_MONOTONIC`) ; [`Duration::ZERO`] = sondage immédiat.
/// - `scope`              → [`FutexScope::Private`] (défaut, intra-processus) ou
///   [`FutexScope::Shared`] (mémoire partagée). Doit être **cohérent** avec le
///   `futex_wake` correspondant.
///
/// Retour : `Ok(())` quand le thread a été **réveillé** par un `futex_wake` (ou un
/// réveil spurious — l'appelant relit toujours `*word` après retour).
///
/// # Errors
///
/// - [`Errno::EAGAIN`] : `*word != expected` au moment de l'appel — **aucune
///   attente** n'a eu lieu (réveil perdu évité).
/// - [`Errno::EINTR`] : un signal a interrompu l'attente — **remonté tel quel**
///   (ADR-021 §2), aucune reprise automatique en couche 0.
/// - [`Errno::ETIMEDOUT`] : l'échéance `timeout` a expiré avant tout réveil.
/// - [`Errno::EFAULT`] : pointeur invalide — **inatteignable** via cette API safe
///   (`word` est une référence vivante, `timeout` un local vivant ou `NULL`).
pub fn futex_wait(
    word: &AtomicU32,
    expected: u32,
    timeout: Option<Duration>,
    scope: FutexScope,
) -> Result<(), Errno> {
    // Échéance : `None` → pointeur NULL (attente infinie) ; `Some` → adresse d'un
    // `KernelTimespec` local vivant pour toute la durée de l'appel (réutilisé de la
    // famille `poll` : même `repr(C)`, même conversion défensive — Principe 2).
    let ts = timeout.map(kernel_timespec_from_duration_clamped);
    let timeout_ptr: u64 = match &ts {
        Some(t) => {
            let p: *const KernelTimespec = t;
            p as u64
        }
        None => 0,
    };

    let op = futex_op(FUTEX_WAIT, scope);
    let uaddr = word.as_ptr() as u64;

    // SAFETY: SYS_futex / FUTEX_WAIT. `uaddr` pointe l'`AtomicU32` emprunté par
    // l'appelant (vivant pour la durée de l'appel) ; le noyau y **lit** 4 octets
    // (comparaison à `expected`), n'écrit rien dans la mémoire utilisateur.
    // `timeout_ptr` pointe un `KernelTimespec` local vivant (16 octets, lu seul si
    // non-NULL) ou vaut 0 (NULL = attente infinie). `op`/`expected` sont des scalaires
    // passés par registre. Aucun pointeur ne survit à l'appel.
    let ret = unsafe { raw_syscall_futex_wait(uaddr, op, expected, timeout_ptr) };

    // COUVERTURE (ADR-035 — cf. docs/COVERAGE-EXCEPTIONS.md, section `futex`). Le bras
    // d'erreur est **couvert** par `EAGAIN` (`*word != expected`, déterministe) et
    // `ETIMEDOUT` (échéance courte, déterministe). `EINTR` (signal pendant l'attente)
    // partage ce même bras mais n'est pas injectable de façon déterministe en CI sans
    // trampoline de restorer de signal x86_64 (cf. précédent `ppoll`) — il est donc
    // documenté STRUCTURAL sans test dédié ; le décodage d'errno lui-même est couvert
    // par `errno_from_negative_syscall_ret_maps_known`.
    if ret < 0 {
        return Err(errno_from_negative_syscall_ret(ret));
    }
    Ok(())
}

/// `futex(2)` `FUTEX_WAKE` : réveille jusqu'à `count` waiters bloqués sur `word`.
///
/// - `word`  : le **même** mot futex que celui sur lequel les waiters se sont
///   endormis (le noyau identifie la file d'attente par l'adresse).
/// - `count` : [`FutexWakeCount`] — typiquement [`FutexWakeCount::ONE`] (relâche de
///   `Mutex`) ou [`FutexWakeCount::ALL`] (`Condvar::notify_all`). Borné à `INT_MAX`
///   par construction.
/// - `scope` : doit correspondre à celui des `futex_wait` ciblés.
///
/// L'appelant publie en général la nouvelle valeur dans `word` (`store`) **avant**
/// d'appeler `futex_wake`, pour que les waiters relisant `*word` voient l'état à jour.
///
/// Retour : le **nombre de waiters effectivement réveillés** (`0` si aucun n'était
/// en attente — ce n'est pas une erreur).
///
/// # Errors
///
/// - [`Errno::EFAULT`] : pointeur invalide — **inatteignable** via cette API safe
///   (`word` est une référence vivante). `FUTEX_WAKE` ne lit ni n'écrit le contenu
///   du mot (clé d'adresse uniquement) et ne prend aucun timeout.
pub fn futex_wake(
    word: &AtomicU32,
    count: FutexWakeCount,
    scope: FutexScope,
) -> Result<u32, Errno> {
    let op = futex_op(FUTEX_WAKE, scope);
    let uaddr = word.as_ptr() as u64;
    // `count.get()` est garanti `≤ INT_MAX` par construction (newtype) : le `val`
    // passé au noyau reste un `int` positif, sans conversion qui échoue.
    let val = count.get();

    // SAFETY: SYS_futex / FUTEX_WAKE. `uaddr` pointe l'`AtomicU32` emprunté par
    // l'appelant (vivant) ; `FUTEX_WAKE` n'utilise que l'**adresse** comme clé de file
    // — aucune lecture ni écriture du contenu du mot, aucun pointeur `timeout`.
    // `op`/`val` sont des scalaires passés par registre. Aucun pointeur ne survit.
    let ret = unsafe { raw_syscall_futex_wake(uaddr, op, val) };

    // COUVERTURE (ADR-035, STRUCTURAL — cf. docs/COVERAGE-EXCEPTIONS.md, section
    // `futex`). `FUTEX_WAKE` sur un `&AtomicU32` valide ne peut échouer que par
    // `EFAULT` (pointeur invalide — exclu par construction) : le bras `ret < 0` est
    // inatteignable via cette API safe. Le chemin succès (compte de réveils) est
    // couvert par le test multi-thread.
    if ret < 0 {
        return Err(errno_from_negative_syscall_ret(ret));
    }
    // `ret ≥ 0` est le nombre de waiters réveillés (`≤` nombre de waiters, lui-même
    // borné par les threads vivants) : conversion défensive `i64 → u32` (Principe 2,
    // pas d'`as`). Le bras d'erreur est inatteignable (`ret` tient toujours dans un
    // `u32`) — STRUCTURAL (cf. docs/COVERAGE-EXCEPTIONS.md).
    u32::try_from(ret).map_err(|_| Errno::EINVAL)
}

// ─────────────────────────────────────────────────────────────────────────
// Helper commun : conversion d'une valeur de retour syscall négative en Errno.
// ─────────────────────────────────────────────────────────────────────────

fn errno_from_negative_syscall_ret(ret: i64) -> Errno {
    debug_assert!(ret < 0 && ret > -4096);
    #[allow(clippy::cast_possible_truncation)]
    let raw = ret.wrapping_neg() as i32;
    let nz = NonZeroI32::new(raw).expect("errno strictement positif par construction");
    Errno::from_nonzero(nz)
}

// ─────────────────────────────────────────────────────────────────────────
// asm! wrappers x86_64 / aarch64 — une fonction dédiée par opération.
// Signature noyau : futex(uaddr, op, val, timeout, uaddr2, val3).
// FUTEX_WAIT utilise `timeout` ; uaddr2/val3 sont nuls (non-PI, non-REQUEUE).
// FUTEX_WAKE n'utilise que uaddr/op/val.
// ─────────────────────────────────────────────────────────────────────────

#[cfg(target_arch = "x86_64")]
#[inline]
unsafe fn raw_syscall_futex_wait(uaddr: u64, op: u32, val: u32, timeout: u64) -> i64 {
    let ret: i64;
    // SAFETY: SYS_futex (x86_64 = 202). Le noyau **lit** 4 octets à `uaddr`
    // (comparaison à `val`) et 16 octets à `timeout` si non-NULL ; aucune écriture
    // mémoire utilisateur → `readonly` valide. `uaddr2`/`val3` sont nuls.
    unsafe {
        core::arch::asm!(
            "syscall",
            in("rax") 202_i64,
            in("rdi") uaddr,
            in("rsi") u64::from(op),
            in("rdx") u64::from(val),
            in("r10") timeout,
            in("r8") 0_u64,
            in("r9") 0_u64,
            lateout("rax") ret,
            lateout("rcx") _,
            lateout("r11") _,
            options(nostack, preserves_flags, readonly),
        );
    }
    ret
}

#[cfg(target_arch = "aarch64")]
#[inline]
unsafe fn raw_syscall_futex_wait(uaddr: u64, op: u32, val: u32, timeout: u64) -> i64 {
    let ret: i64;
    // SAFETY: SYS_futex (aarch64 = 98). Mêmes accès mémoire que sur x86_64 : lecture
    // seule de `*uaddr` (4 octets) et `*timeout` (16 octets, si non-NULL).
    unsafe {
        core::arch::asm!(
            "svc 0",
            in("x8") 98_i64,
            inout("x0") uaddr => ret,
            in("x1") u64::from(op),
            in("x2") u64::from(val),
            in("x3") timeout,
            in("x4") 0_u64,
            in("x5") 0_u64,
            options(nostack, preserves_flags, readonly),
        );
    }
    ret
}

#[cfg(target_arch = "x86_64")]
#[inline]
unsafe fn raw_syscall_futex_wake(uaddr: u64, op: u32, val: u32) -> i64 {
    let ret: i64;
    // SAFETY: SYS_futex (x86_64 = 202). `FUTEX_WAKE` n'accède **pas** au contenu de
    // `*uaddr` (clé d'adresse uniquement) : aucun accès mémoire utilisateur observable
    // → `nomem` valide. `timeout`/`uaddr2`/`val3` sont nuls (ignorés par `FUTEX_WAKE`).
    unsafe {
        core::arch::asm!(
            "syscall",
            in("rax") 202_i64,
            in("rdi") uaddr,
            in("rsi") u64::from(op),
            in("rdx") u64::from(val),
            in("r10") 0_u64,
            in("r8") 0_u64,
            in("r9") 0_u64,
            lateout("rax") ret,
            lateout("rcx") _,
            lateout("r11") _,
            options(nostack, preserves_flags, nomem),
        );
    }
    ret
}

#[cfg(target_arch = "aarch64")]
#[inline]
unsafe fn raw_syscall_futex_wake(uaddr: u64, op: u32, val: u32) -> i64 {
    let ret: i64;
    // SAFETY: SYS_futex (aarch64 = 98). Comme sur x86_64 : `FUTEX_WAKE` n'accède pas
    // au contenu mémoire utilisateur (clé d'adresse) → `nomem` valide.
    unsafe {
        core::arch::asm!(
            "svc 0",
            in("x8") 98_i64,
            inout("x0") uaddr => ret,
            in("x1") u64::from(op),
            in("x2") u64::from(val),
            in("x3") 0_u64,
            in("x4") 0_u64,
            in("x5") 0_u64,
            options(nostack, preserves_flags, nomem),
        );
    }
    ret
}

#[cfg(all(test, target_os = "linux", not(loom)))]
mod tests {
    use super::{FUTEX_PRIVATE_FLAG, FUTEX_WAIT, FUTEX_WAKE, futex_op, futex_wait, futex_wake};
    use air_sys_types::{Errno, FutexScope, FutexWakeCount};
    use core::num::NonZeroI32;
    use core::sync::atomic::{AtomicU32, Ordering};
    use core::time::Duration;
    use std::sync::Arc;
    use std::thread;

    /// `futex_wait` revient immédiatement par `EAGAIN` si la valeur observée ne
    /// correspond plus — aucune attente, aucun blocage (course « réveil perdu »).
    #[test]
    fn futex_wait_eagain_on_value_mismatch() {
        let word = AtomicU32::new(1);
        // On prétend attendre la valeur 0 alors que `word == 1` : EAGAIN immédiat.
        let err = futex_wait(&word, 0, None, FutexScope::Private).unwrap_err();
        assert_eq!(err, Errno::EAGAIN);
    }

    /// `futex_wait` avec une portée `Shared` exerce l'autre bras de `futex_op` et
    /// renvoie aussi `EAGAIN` sur valeur divergente (mot privé à la pile du test,
    /// mais le drapeau de portée est bien transmis sans drapeau brut).
    #[test]
    fn futex_wait_shared_scope_eagain_on_mismatch() {
        let word = AtomicU32::new(42);
        let err = futex_wait(&word, 7, None, FutexScope::Shared).unwrap_err();
        assert_eq!(err, Errno::EAGAIN);
    }

    /// `futex_wait` avec un court timeout sur un mot qui ne change jamais expire en
    /// `ETIMEDOUT` (échéance relative `CLOCK_MONOTONIC`).
    #[test]
    fn futex_wait_times_out() {
        let word = AtomicU32::new(0);
        let err = futex_wait(
            &word,
            0,
            Some(Duration::from_millis(20)),
            FutexScope::Private,
        )
        .unwrap_err();
        assert_eq!(err, Errno::ETIMEDOUT);
    }

    /// `Duration::ZERO` = sondage immédiat : un mot inchangé expire aussitôt.
    #[test]
    fn futex_wait_zero_timeout_polls_immediately() {
        let word = AtomicU32::new(5);
        let err = futex_wait(&word, 5, Some(Duration::ZERO), FutexScope::Private).unwrap_err();
        assert_eq!(err, Errno::ETIMEDOUT);
    }

    /// `futex_wake` sur un mot sans aucun waiter réveille `0` thread — ce n'est pas
    /// une erreur (chemin succès du décodage du compte).
    #[test]
    fn futex_wake_without_waiters_returns_zero() {
        let word = AtomicU32::new(0);
        let woken = futex_wake(&word, FutexWakeCount::ALL, FutexScope::Private).expect("wake");
        assert_eq!(woken, 0);
    }

    /// Rendez-vous réel : un thread `futex_wait` (timeout infini) sur un mot, le
    /// thread principal change la valeur puis `futex_wake` → le waiter se réveille,
    /// et `futex_wake` rapporte **exactement 1** réveil.
    #[test]
    fn futex_wake_releases_one_waiter() {
        let word = Arc::new(AtomicU32::new(0));
        let word_waiter = Arc::clone(&word);

        let waiter = thread::spawn(move || {
            // Boucle de mutex idiomatique : tant que la valeur vaut 0, on attend.
            // `futex_wait` peut renvoyer `Ok(())` (réveil) ou `EAGAIN` (la valeur a
            // déjà changé) ; les deux sortent quand `word != 0`. Aucun retry sur
            // `EINTR` ici n'est nécessaire : le test ne s'envoie pas de signal.
            loop {
                if word_waiter.load(Ordering::Acquire) != 0 {
                    return;
                }
                match futex_wait(&word_waiter, 0, None, FutexScope::Private) {
                    Ok(()) | Err(Errno::EAGAIN) => {}
                    Err(other) => panic!("futex_wait inattendu : {other:?}"),
                }
            }
        });

        // Laisse le waiter atteindre `futex_wait` (réduit la fenêtre où la valeur
        // change avant l'entrée dans le syscall — non requis pour la justesse).
        thread::sleep(Duration::from_millis(50));
        word.store(1, Ordering::Release);

        // Réveille jusqu'à tous les waiters ; il y en a exactement un. On reboucle
        // tant que le compte est 0 (le waiter peut ne pas encore être endormi).
        let mut total_woken = 0_u32;
        while total_woken == 0 {
            total_woken =
                futex_wake(&word, FutexWakeCount::ALL, FutexScope::Private).expect("wake");
            if total_woken == 0 {
                thread::sleep(Duration::from_millis(5));
            }
        }
        assert_eq!(total_woken, 1, "un seul waiter doit être réveillé");
        waiter.join().expect("join waiter");
    }

    /// Exerce explicitement `futex_op` (les deux portées, les deux opérations de
    /// base) — le drapeau privé n'est posé que pour `Private`.
    #[test]
    fn futex_op_composes_private_flag() {
        assert_eq!(
            futex_op(FUTEX_WAIT, FutexScope::Private),
            FUTEX_WAIT | FUTEX_PRIVATE_FLAG
        );
        assert_eq!(futex_op(FUTEX_WAIT, FutexScope::Shared), FUTEX_WAIT);
        assert_eq!(
            futex_op(FUTEX_WAKE, FutexScope::Private),
            FUTEX_WAKE | FUTEX_PRIVATE_FLAG
        );
        assert_eq!(futex_op(FUTEX_WAKE, FutexScope::Shared), FUTEX_WAKE);
    }

    #[test]
    fn errno_from_negative_syscall_ret_maps_known() {
        assert_eq!(super::errno_from_negative_syscall_ret(-11), Errno::EAGAIN);
        assert_eq!(
            super::errno_from_negative_syscall_ret(-110),
            Errno::ETIMEDOUT
        );
        // Cohérence du décodage : -EINTR (4) → Errno::EINTR (remonté sans retry).
        let nz = NonZeroI32::new(4).unwrap();
        assert_eq!(Errno::from_nonzero(nz), Errno::EINTR);
    }

    // Couvrent les deux opérandes du `debug_assert!(ret < 0 && ret > -4096)` de
    // `errno_from_negative_syscall_ret` (`ret = 0` ; `ret = -5000`).
    #[test]
    #[should_panic(expected = "ret < 0 && ret > -4096")]
    fn errno_from_negative_syscall_ret_panics_on_non_negative() {
        let _ = super::errno_from_negative_syscall_ret(0);
    }

    #[test]
    #[should_panic(expected = "ret < 0 && ret > -4096")]
    fn errno_from_negative_syscall_ret_panics_below_errno_range() {
        let _ = super::errno_from_negative_syscall_ret(-5000);
    }
}