air_sys_syscall/ebpf/

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

//! **Le consommateur d'un tampon circulaire BPF** (`BPF_MAP_TYPE_RINGBUF`).
//!
//! Un programme eBPF écrit des enregistrements (`bpf_ringbuf_output`) ; l'espace utilisateur
//! les lit en **projetant** la carte, sans appel système par enregistrement. Le protocole est
//! celui de `Documentation/bpf/ringbuf.rst` et de `kernel/bpf/ringbuf.c` :
//!
//! ```text
//! offset 0          une page      position CONSOMMÉE (u64) — écrite par nous
//! offset page       une page      position PRODUITE (u64)  — écrite par le noyau
//! offset 2 × page   2 × taille    les données, projetées DEUX FOIS à la suite
//! ```
//!
//! La double projection fait qu'un enregistrement à cheval sur la fin de la zone reste
//! **contigu** : on le lit d'un seul tenant. Chaque enregistrement commence par un en-tête de
//! 8 octets — sa longueur sur 32 bits, dont deux bits d'état (`BUSY` : en cours d'écriture ;
//! `DISCARD` : jeté), puis 32 bits propres au noyau —, et sa taille totale est arrondie à 8.
//!
//! # Pourquoi en couche 0
//!
//! Lire une mémoire que le noyau écrit **en même temps** demande des accès atomiques sur des
//! adresses brutes : c'est intrinsèquement `unsafe`, et l'`unsafe` d'Air vit en couche 0, nulle
//! part ailleurs. Au-dessus, [`BpfRingBufferConsumer::consume`] ne rend que des tranches
//! d'octets déjà publiées.
//!
//! # L'ordre des accès
//!
//! - la position produite se lit en **acquérir** : ce qui la précède est alors visible ;
//! - la longueur d'un en-tête se lit en **acquérir** : le noyau l'écrit en **libérer** quand il
//!   publie l'enregistrement, et c'est ce qui rend ses données visibles ;
//! - la position consommée s'écrit en **libérer** : le noyau ne réutilise la place qu'après.

use core::sync::atomic::{AtomicU32, AtomicU64, Ordering};

use air_sys_types::Errno;
use air_sys_types::fd::BorrowedFd;
use air_sys_types::mem::{MapFlags, ProtectionFlags};

use crate::mem::{Mapping, mmap_file};

/// L'enregistrement est en cours d'écriture : on s'arrête avant lui.
const BUSY: u32 = 1 << 31;
/// L'enregistrement a été jeté : on le saute sans le rendre.
const DISCARD: u32 = 1 << 30;
/// Taille de l'en-tête d'un enregistrement.
const HEADER_SIZE: u64 = 8;
/// La plus petite page que le noyau connaisse.
const MIN_PAGE_SIZE: usize = 4096;

/// **Le consommateur d'un tampon circulaire BPF**, projeté en mémoire.
#[derive(Debug)]
pub struct BpfRingBufferConsumer {
    /// La page de la position consommée, en lecture et écriture.
    consumer: Mapping,
    /// La page de la position produite, puis les données projetées deux fois, en lecture seule.
    producer: Mapping,
    /// Taille d'une page : le décalage des données dans `producer`.
    page_size: usize,
    /// `taille des données - 1` : la taille est une puissance de deux.
    mask: u64,
}

impl BpfRingBufferConsumer {
    /// Projette le tampon `map`, de `data_size` octets de données, sur une machine dont la page
    /// fait `page_size` octets.
    ///
    /// # Errors
    ///
    /// `EINVAL` si `page_size` n'est pas une puissance de deux d'au moins 4 Kio, ou si
    /// `data_size` n'est pas une puissance de deux multiple de la page — ce que le noyau exige
    /// d'un tampon circulaire ; l'erreur de `mmap(2)` sinon (`map` n'est pas un tampon circulaire,
    /// taille fausse…).
    pub fn new(map: BorrowedFd<'_>, data_size: usize, page_size: usize) -> Result<Self, Errno> {
        let valid = page_size.is_power_of_two()
            && page_size >= MIN_PAGE_SIZE
            && data_size.is_power_of_two()
            && data_size >= page_size;
        let producer_length = data_size
            .checked_mul(2)
            .and_then(|data| data.checked_add(page_size))
            .filter(|_| valid)
            .ok_or(Errno::EINVAL)?;
        // `usize` → `u64` ne perd rien sur les cibles d'Air (64 bits) : le repli n'est jamais pris.
        let producer_offset = u64::try_from(page_size).unwrap_or(u64::MAX);
        let mask = u64::try_from(data_size)
            .unwrap_or(u64::MAX)
            .saturating_sub(1);
        let consumer = mmap_file(
            map,
            page_size,
            0,
            ProtectionFlags::READ | ProtectionFlags::WRITE,
            MapFlags::SHARED,
        )?;
        let producer = mmap_file(
            map,
            producer_length,
            producer_offset,
            ProtectionFlags::READ,
            MapFlags::SHARED,
        )?;
        Ok(Self {
            consumer,
            producer,
            page_size,
            mask,
        })
    }

    /// **Vide** le tampon : rend chaque enregistrement publié à `handler`, dans l'ordre, puis
    /// avance la position consommée. S'arrête sur un enregistrement encore en cours d'écriture.
    /// Rend le nombre d'enregistrements rendus (les jetés ne comptent pas).
    pub fn consume(&mut self, mut handler: impl FnMut(&[u8])) -> usize {
        let consumer = self.consumer.as_ptr().cast::<u64>();
        let producer = self.producer.as_ptr().cast::<u64>();
        // SAFETY: `consumer` et `producer` sont les débuts de deux projections vivantes, d'au
        // moins une page chacune, alignées sur la page — donc sur 8 octets ; le noyau y tient un
        // `u64` qu'il ne lit et n'écrit qu'atomiquement. `data` est le début de la zone de
        // données, projetée deux fois, soit `2 × (mask + 1)` octets lisibles après lui.
        unsafe {
            let data = self.producer.as_ptr().add(self.page_size);
            drain(
                AtomicU64::from_ptr(consumer.cast_mut()),
                AtomicU64::from_ptr(producer.cast_mut()),
                data,
                self.mask,
                &mut handler,
            )
        }
    }
}

/// Le cœur du protocole, sur des pointeurs : c'est ce qui permet de l'éprouver sur une mémoire
/// fabriquée, avec des en-têtes que le noyau ne produirait qu'en course (`BUSY`).
///
/// Le gestionnaire est un `dyn` : il n'existe qu'**une** copie de ce protocole, quelle que soit
/// la fermeture de l'appelant — et ce sont ses branches, toutes, que les tests éprouvent.
///
/// # Safety
///
/// `data` doit désigner `2 × (mask + 1)` octets lisibles, aligné sur 8 ; `mask + 1` doit être
/// une puissance de deux ; les enregistrements entre `consumer` et `producer` doivent suivre le
/// protocole du noyau (en-tête aligné sur 8, longueur ne dépassant pas `mask + 1`).
pub(crate) unsafe fn drain(
    consumer: &AtomicU64,
    producer: &AtomicU64,
    data: *const u8,
    mask: u64,
    handler: &mut dyn FnMut(&[u8]),
) -> usize {
    let mut delivered = 0_usize;
    let mut position = consumer.load(Ordering::Acquire);
    loop {
        let published = producer.load(Ordering::Acquire);
        if position >= published {
            return delivered;
        }
        // `position & mask` est au plus `mask`, qui vient d'un `usize` : la conversion tient.
        let offset = usize::try_from(position & mask).unwrap_or(0);
        // SAFETY: `offset <= mask`, et la zone fait `2 × (mask + 1)` octets : l'en-tête, aligné
        // sur 8, y est entier ; le noyau n'y écrit qu'atomiquement.
        let header = unsafe { AtomicU32::from_ptr(data.add(offset).cast::<u32>().cast_mut()) };
        let length = header.load(Ordering::Acquire);
        if length & BUSY != 0 {
            return delivered;
        }
        let size = length & !(BUSY | DISCARD);
        // Une longueur plus grande que la zone : le protocole est rompu, on n'avance plus.
        if u64::from(size) > mask {
            return delivered;
        }
        if length & DISCARD == 0 {
            // `size <= mask`, qui vient d'un `usize` : la conversion tient.
            let size = usize::try_from(size).unwrap_or(0);
            // SAFETY: l'enregistrement est publié (ni `BUSY` ni jeté) : le noyau ne l'écrira plus
            // avant que la position consommée ne l'ait dépassé. Il commence 8 octets après son
            // en-tête et tient dans la zone grâce à la double projection (`offset + 8 + size
            // <= 2 × (mask + 1)`).
            let record = unsafe { core::slice::from_raw_parts(data.add(offset).add(8), size) };
            handler(record);
            delivered = delivered.saturating_add(1);
        }
        let advance = (u64::from(size)
            .saturating_add(HEADER_SIZE)
            .saturating_add(7))
            & !7;
        position = position.saturating_add(advance);
        consumer.store(position, Ordering::Release);
    }
}

#[cfg(test)]
mod tests;