zbq_sys/

lib.rs

//! Low-level unsafe FFI bindings for `zbq.h`.
//!
//! `zbq` is a Linux-specific single-producer / multi-consumer (SPMC) shared-memory
//! ring-buffer with optional file-descriptor passing.
//!
//! This crate compiles `zbq.h` (by defining `ZBQ_IMPLEMENTATION`) via the `cc`
//! build-dependency and exposes the raw C symbols through `extern "C"` blocks.
//! For safe, idiomatic Rust wrappers see the `zbq` crate.
#![warn(missing_docs)]
#![allow(non_camel_case_types, non_snake_case, non_upper_case_globals)]

use std::os::raw::{c_int, c_void};

// =============================================================================
// Constants
// =============================================================================

/// Maximum number of file descriptors that may be attached to a single message.
pub const ZBQ_MAX_FD_PER_MSG: u8 = 253;

/// Message alignment requirement in bytes.
///
/// Every committed run must be a multiple of this value and every
/// [`zbq_msg_header`] is physically placed at an offset satisfying this.
pub const ZBQ_MSG_ALIGN: usize = 16;

// =============================================================================
// Error codes
// =============================================================================

/// Result / error codes returned by the C API.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum zbq_error {
    /// Operation succeeded.
    ZBQ_OK = 0,
    /// Invalid argument or precondition violation.
    ZBQ_ERROR_INVALID = -1,
    /// Memory allocation failed.
    ZBQ_ERROR_NOMEM = -2,
    /// Spurious wake-up or try-again condition.
    ZBQ_ERROR_AGAIN = -3,
    /// Low-level I/O error (syscall failure, etc.).
    ZBQ_ERROR_IO = -4,
    /// Protocol / state mismatch.
    ZBQ_ERROR_PROTOCOL = -5,
    /// Ring overflow / would exceed capacity.
    ZBQ_ERROR_OVERFLOW = -6,
    /// No free slot or entry not found.
    ZBQ_ERROR_NOENT = -7,
}

// =============================================================================
// Wakeup mode
// =============================================================================

/// Consumer blocking strategy.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum zbq_wakeup_mode {
    /// Consumer blocks on the producer's broadcast futex.
    ZBQ_WAKEUP_PUSH = 0,
    /// Consumer self-manages polling (busy-poll or adaptive sleep).
    ZBQ_WAKEUP_PULL = 1,
}

// =============================================================================
// Commit flags
// =============================================================================

/// Flags that modify [`zbq_producer_commit`] behaviour.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum zbq_commit_flags {
    /// No special behaviour.
    ZBQ_COMMIT_NONE = 0,
    /// Force a consumer scan after committing the write.
    ZBQ_COMMIT_FORCE_SCAN = 1,
}

// =============================================================================
// Shared-memory layout structures
// =============================================================================

/// 256-byte control region, cache-line aligned.
///
/// This structure lives in the shared memory segment that all consumers map
/// read-only.  The producer writes to the fields below.
#[repr(C, align(64))]
pub struct zbq_control_region {
    /// Monotonically-increasing producer head (in bytes), atomically stored
    /// with release semantics on commit.
    pub producer_head: u64,

    _pad0: [u8; 56],

    /// Futex word incremented by the producer when new data is available.
    /// Push-mode consumers [`zbq_consumer_wait_broadcast`] on this word.
    pub broadcast_futex: u32,

    _pad1: [u8; 60],

    /// PID of the producer process.  Consumers use this to open a `pidfd`
    /// for liveness monitoring.
    pub producer_pid: i32,

    _pad2: [u8; 60],

    /// Total size of the data ring in bytes.
    pub ring_capacity: u64,
    /// Number of failed reservation attempts that triggers a forced consumer scan.
    pub clog_threshold: u64,
    /// Nanoseconds after which a consumer with no progress is considered hung.
    pub stall_timeout_ns: u64,
    reserved: [u8; 40],
}

/// 4096-byte per-consumer metadata page, page aligned.
///
/// A consumer has read-write access to exactly **one** such page; the
/// producer retains read-write access to all of them.
#[repr(C, align(4096))]
pub struct zbq_consumer_metadata {
    /// Tail offset known to the consumer, atomically published with release.
    pub tail_offset: u64,
    /// Timestamp relative to `join_baseline` of the last progress update.
    pub progress_timestamp: u64,
    /// Baseline timestamp captured at consumer join time.
    pub join_baseline: u64,
    reserved: [u8; 4072],
}

/// 16-byte packed message header, 16-byte aligned.
///
/// Every committed run starts with this header followed by `payload_len` bytes
/// of payload.  `total_len` (used by the consumer view) is
/// `sizeof(zbq_msg_header) + payload_len`.
#[repr(C, align(16))]
pub struct zbq_msg_header {
    /// Length in bytes of the payload that follows this header.
    pub payload_len: u32,
    /// Number of file descriptors attached to this message (0..253).
    pub fd_count: u8,
    /// Message-specific flags (reserved, currently ignored on the wire).
    pub flags: u8,
    /// Reserved for future use.
    pub reserved: u16,
    /// Sequence ID used to correlate the ring entry with FD records sent
    /// over the per-consumer side socket.
    pub seq_id: u64,
}

// =============================================================================
// Parameter & view structures
// =============================================================================

/// Parameters supplied to [`zbq_producer_create`].
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct zbq_producer_params {
    /// Size of the data ring.  Must be non-zero and a multiple of 4096.
    pub ring_capacity: u64,
    /// Clog threshold (failed reservation attempts) before a forced scan.
    pub clog_threshold: u64,
    /// Nanoseconds after which a consumer is evicted if no progress is seen.
    pub stall_timeout_ns: u64,
    /// Maximum number of simultaneous consumer slots.
    pub max_consumers: u32,
}

/// Result of a producer reservation query.
///
/// Describes a physically contiguous span of the ring that the producer is
/// allowed to write into.  The run never wraps around the physical end of the
/// buffer.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct zbq_contiguous_run {
    /// Pointer to the first writable byte inside the producer's ring mapping.
    pub base: *mut c_void,
    /// Largest number of bytes that may be written starting at `base`.
    pub max_bytes: u64,
    /// Logical byte offset of `base` (equal to the producer head at query time).
    pub offset: u64,
}

/// Read-only view of a message inside the ring.
///
/// Returned by [`zbq_consumer_peek`].
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct zbq_msg_view {
    /// Pointer to the message header inside the read-only ring mapping.
    pub header: *const zbq_msg_header,
    /// Pointer to the payload bytes that follow the header.
    pub payload: *const c_void,
    /// Length of the payload in bytes.
    pub payload_len: u64,
    /// Logical offset of this message in the ring.
    pub logical_offset: u64,
    /// Total bytes occupied by this message (`header + payload`).
    pub total_len: u64,
}

/// Payload sent from producer to consumer during the join handshake.
#[repr(C)]
pub struct zbq_join_payload {
    /// Size of the data ring.
    pub ring_capacity: u64,
    /// Baseline timestamp used for progress measurements.
    pub join_baseline: u64,
    /// Initial head offset (consumer starts reading from here).
    pub initial_head: u64,
    /// PID of the producer process.
    pub producer_pid: i32,
    /// Slot index assigned to this consumer.
    pub slot_index: u32,
    /// Wakeup mode (`ZBQ_WAKEUP_PUSH` or `ZBQ_WAKEUP_PULL`).
    pub wakeup_mode: u32,
    reserved: [u8; 28],
}

// =============================================================================
// Opaque handles
// =============================================================================

/// Opaque handle for a ring-buffer producer.
///
/// Created by [`zbq_producer_create`] and destroyed by [`zbq_producer_destroy`].
/// Do not attempt to allocate or inspect this type directly.
#[repr(C)]
pub struct zbq_producer {
    _private: [u8; 0],
}

/// Opaque handle for a ring-buffer consumer.
///
/// Created by [`zbq_consumer_join`] and destroyed by [`zbq_consumer_destroy`].
/// Do not attempt to allocate or inspect this type directly.
#[repr(C)]
pub struct zbq_consumer {
    _private: [u8; 0],
}

// =============================================================================
// Compile-time layout checks (must match the C compile-time assertions)
// =============================================================================

const _: () = assert!(core::mem::size_of::<zbq_control_region>() == 256);
const _: () = assert!(core::mem::align_of::<zbq_control_region>() == 64);
const _: () = assert!(core::mem::size_of::<zbq_consumer_metadata>() == 4096);
const _: () = assert!(core::mem::align_of::<zbq_consumer_metadata>() == 4096);
const _: () = assert!(core::mem::size_of::<zbq_msg_header>() == 16);
const _: () = assert!(core::mem::align_of::<zbq_msg_header>() == 16);

// =============================================================================
// Producer API
// =============================================================================

extern "C" {
    /// Create a new producer ring-buffer.
    ///
    /// `params->ring_capacity` must be a non-zero multiple of 4096 and
    /// `params->max_consumers` must be non-zero.
    ///
    /// On success the newly-allocated handle is written to `*out` and
    /// `ZBQ_OK` is returned.
    pub fn zbq_producer_create(
        out: *mut *mut zbq_producer,
        params: *const zbq_producer_params,
    ) -> c_int;

    /// Destroy a producer and release all associated resources (mappings, pidfds,
    /// sockets, slots).
    ///
    /// Passing `NULL` is a no-op.
    pub fn zbq_producer_destroy(prod: *mut zbq_producer);

    /// Complete the join handshake for an incoming consumer connection.
    ///
    /// `conn_fd` is an already-accepted Unix-domain socket descriptor.
    /// `mode` is the desired consumer wakeup mode.
    ///
    /// On success the assigned consumer slot index is written to `*out_slot_id`.
    /// The caller retains ownership of `conn_fd` and should close it after this
    /// call returns.
    pub fn zbq_producer_handshake(
        prod: *mut zbq_producer,
        conn_fd: c_int,
        mode: zbq_wakeup_mode,
        out_slot_id: *mut u32,
    ) -> c_int;

    /// Query the largest physically contiguous writable run available to the
    /// producer.
    ///
    /// If the queue is fully clogged this returns a run with `max_bytes == 0`.
    pub fn zbq_producer_query(prod: *mut zbq_producer) -> zbq_contiguous_run;

    /// Commit `written_bytes` previously written into the contiguous run
    /// returned by the most recent [`zbq_producer_query`].
    ///
    /// Returns `true` on success and `false` if `written_bytes` exceeds the
    /// queried `max_bytes` or if the producer handle is invalid.
    pub fn zbq_producer_commit(
        prod: *mut zbq_producer,
        written_bytes: u64,
        flags: zbq_commit_flags,
    ) -> bool;

    /// Atomically increment and return the next sequence ID for FD-bearing
    /// messages.
    ///
    /// The caller should store the returned value into the message header
    /// before calling [`zbq_producer_commit`].
    pub fn zbq_producer_next_seq_id(prod: *mut zbq_producer) -> u64;

    /// Broadcast file descriptors to all push-mode consumers.
    ///
    /// Sends an `SCM_RIGHTS` record on each consumer's side socket containing
    /// exactly `fd_count` descriptors and the matching `seq_id`.
    ///
    /// This function is a no-op (zero overhead) when `fd_count == 0`.
    pub fn zbq_producer_broadcast_fds(
        prod: *mut zbq_producer,
        seq_id: u64,
        fds: *const c_int,
        fd_count: u8,
    ) -> c_int;

    /// Scan for dead or hung consumers and evict them.
    ///
    /// Normally invoked automatically when the clog threshold is reached, but
    /// may be called explicitly when `ZBQ_COMMIT_FORCE_SCAN` is used or by
    /// application logic.
    pub fn zbq_producer_scan(prod: *mut zbq_producer) -> c_int;
}

// =============================================================================
// Consumer API
// =============================================================================

extern "C" {
    /// Join a ring-buffer as a consumer.
    ///
    /// `rendezvous_fd` must be a connected Unix-domain socket to the producer.
    /// `mode` selects push (futex) or pull (self-poll) behaviour.
    ///
    /// On success the consumer handle is written to `*out` and `ZBQ_OK` is
    /// returned.  The caller retains ownership of `rendezvous_fd`.
    pub fn zbq_consumer_join(
        rendezvous_fd: c_int,
        mode: zbq_wakeup_mode,
        out: *mut *mut zbq_consumer,
    ) -> c_int;

    /// Destroy a consumer and release all associated resources.
    ///
    /// Passing `NULL` is a no-op.
    pub fn zbq_consumer_destroy(cons: *mut zbq_consumer);

    /// Check whether the producer process is still alive.
    ///
    /// This polls the `pidfd` opened on the producer at join time.
    pub fn zbq_consumer_is_producer_alive(cons: *mut zbq_consumer) -> bool;

    /// Load the current producer head with acquire semantics.
    pub fn zbq_consumer_acquire_head(cons: *const zbq_consumer) -> u64;

    /// Inspect the message at the consumer's current tail.
    ///
    /// If data is available, `out` is populated with a [`zbq_msg_view`]
    /// pointing into the read-only ring mapping.
    ///
    /// Returns `ZBQ_ERROR_AGAIN` when the caller has reached the current
    /// producer head and there is no new data yet.
    pub fn zbq_consumer_peek(cons: *mut zbq_consumer, out: *mut zbq_msg_view) -> c_int;

    /// **Do not use this on untrusted producers.**
    ///
    /// Attempt to receive the file descriptors attached to the message with the
    /// given `expected_seq_id` over the consumer's side socket.
    ///
    /// `out_fds` must point to an array with room for at least
    /// [`ZBQ_MAX_FD_PER_MSG`] elements.  The actual number received is written
    /// to `*out_fd_count`.
    pub fn zbq_consumer_recv_fds(
        cons: *mut zbq_consumer,
        expected_seq_id: u64,
        out_fds: *mut c_int,
        out_fd_count: *mut u8,
    ) -> c_int;

    /// Atomically publish the new consumer tail and progress timestamp.
    ///
    /// `new_tail` must not exceed the value returned by the most recent
    /// [`zbq_consumer_acquire_head`].  This must be called after processing
    /// a batch of messages to release ring space back to the producer.
    pub fn zbq_consumer_publish(cons: *mut zbq_consumer, new_tail: u64);

    /// Read the current broadcast futex token.
    ///
    /// Push-mode consumers should load this before calling
    /// [`zbq_consumer_wait_broadcast`] and reload it after waking up.
    pub fn zbq_consumer_get_futex_token(cons: *const zbq_consumer) -> u32;

    /// Block until the producer commits new data (push mode).
    ///
    /// `token` should be the value obtained from
    /// [`zbq_consumer_get_futex_token`] immediately before this call.
    ///
    /// Returns `ZBQ_ERROR_AGAIN` on spurious wake-up — the caller should
    /// re-read the token and retry.
    pub fn zbq_consumer_wait_broadcast(cons: *mut zbq_consumer, token: u32) -> c_int;

    /// Load the consumer's local tail with acquire semantics.
    pub fn zbq_consumer_tail(cons: *const zbq_consumer) -> u64;

    /// Return a read-only pointer to the shared control region.
    pub fn zbq_consumer_ctrl(cons: *const zbq_consumer) -> *const zbq_control_region;

    /// Return a mutable pointer to this consumer's private metadata page.
    pub fn zbq_consumer_meta(cons: *mut zbq_consumer) -> *mut zbq_consumer_metadata;
}