ipc-ring48 1.0.1

//! `ipc-ring48` is a bounded 48-byte POSIX shared-memory SPSC queue.
//!
//! The queue has one producer, one consumer, fixed 48-byte slots, and a
//! lock-free hot path. `push` returns [`PushError::Full`] when the queue is
//! full. `pop` returns `None` when the queue is empty.

use libc::{
    MAP_FAILED, MAP_SHARED, O_CREAT, O_EXCL, O_RDWR, PROT_READ, PROT_WRITE, c_void, close, fstat,
    ftruncate, mmap, munmap, shm_open, shm_unlink,
};
use std::ffi::CString;
use std::fmt;
use std::io;
use std::mem::{size_of, zeroed};
use std::ptr::{self, addr_of};
use std::sync::atomic::{AtomicU64, Ordering};

/// POSIX shared-memory object name used by `ipc-ring48`.
pub const SHM_NAME: &str = "/ipc_ring48_queue";

/// Fixed payload size carried by every queue slot.
pub const PAYLOAD_SIZE: usize = 48;

const MAGIC: u64 = 0x4950_4352_494E_4734; // "IPCRING4"
const VERSION: u64 = 1;
const FLAGS: u64 = 0;

#[repr(C, align(64))]
pub struct SharedHeader {
    pub magic: u64,
    pub version: u64,
    pub region_size: u64,
    pub payload_size: u64,
    pub capacity: u64,
    pub flags: u64,
    pub producer_pid: u64,
    pub consumer_pid: u64,
}

#[repr(C, align(64))]
pub struct CounterLine {
    pub value: AtomicU64,
    pub reserved: [u8; 56],
}

#[repr(C, align(64))]
pub struct Slot48 {
    pub payload: [u8; PAYLOAD_SIZE],
    pub reserved: [u8; 16],
}

const HEADER_SIZE: usize = size_of::<SharedHeader>();
const COUNTER_SIZE: usize = size_of::<CounterLine>();
const SLOT_SIZE: usize = size_of::<Slot48>();
const BASE_SIZE: usize = HEADER_SIZE + COUNTER_SIZE + COUNTER_SIZE;

/// Error returned by [`Producer::push`] when the queue is full.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PushError {
    Full,
}

impl fmt::Display for PushError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            PushError::Full => write!(f, "queue full"),
        }
    }
}

impl std::error::Error for PushError {}

/// Snapshot of queue state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QueueStats {
    pub capacity: u64,
    pub len: u64,
    pub head: u64,
    pub tail: u64,
    pub region_size: u64,
}

struct Mapping {
    ptr: *mut u8,
    len: usize,
    fd: i32,
}

impl Mapping {
    fn create_or_open(capacity: usize) -> io::Result<Self> {
        validate_capacity(capacity)?;

        let expected_region_size = region_size_for_capacity(capacity)?;
        let name = CString::new(SHM_NAME).unwrap();

        let mut created = false;

        let fd = unsafe { shm_open(name.as_ptr(), O_CREAT | O_EXCL | O_RDWR, 0o600) };

        let fd = if fd >= 0 {
            created = true;
            fd
        } else {
            let err = io::Error::last_os_error();

            if err.raw_os_error() == Some(libc::EEXIST) {
                let fd = unsafe { shm_open(name.as_ptr(), O_RDWR, 0o600) };

                if fd < 0 {
                    return Err(io::Error::last_os_error());
                }

                fd
            } else {
                return Err(err);
            }
        };

        let region_size = if created {
            let rc = unsafe { ftruncate(fd, expected_region_size as libc::off_t) };

            if rc != 0 {
                let err = io::Error::last_os_error();

                unsafe {
                    close(fd);
                }

                return Err(err);
            }

            expected_region_size
        } else {
            file_size(fd)?
        };

        let raw = unsafe {
            mmap(
                ptr::null_mut(),
                region_size,
                PROT_READ | PROT_WRITE,
                MAP_SHARED,
                fd,
                0,
            )
        };

        if raw == MAP_FAILED {
            let err = io::Error::last_os_error();

            unsafe {
                close(fd);
            }

            return Err(err);
        }

        let mapping = Self {
            ptr: raw as *mut u8,
            len: region_size,
            fd,
        };

        if created {
            mapping.initialise(capacity)?;
        } else {
            mapping.validate()?;

            if mapping.capacity() != capacity as u64 {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidInput,
                    "existing queue capacity does not match requested capacity",
                ));
            }
        }

        Ok(mapping)
    }

    fn open_existing() -> io::Result<Self> {
        let name = CString::new(SHM_NAME).unwrap();

        let fd = unsafe { shm_open(name.as_ptr(), O_RDWR, 0o600) };

        if fd < 0 {
            return Err(io::Error::last_os_error());
        }

        let region_size = file_size(fd)?;

        let raw = unsafe {
            mmap(
                ptr::null_mut(),
                region_size,
                PROT_READ | PROT_WRITE,
                MAP_SHARED,
                fd,
                0,
            )
        };

        if raw == MAP_FAILED {
            let err = io::Error::last_os_error();

            unsafe {
                close(fd);
            }

            return Err(err);
        }

        let mapping = Self {
            ptr: raw as *mut u8,
            len: region_size,
            fd,
        };

        mapping.validate()?;

        Ok(mapping)
    }

    fn initialise(&self, capacity: usize) -> io::Result<()> {
        validate_capacity(capacity)?;

        unsafe {
            ptr::write_bytes(self.ptr, 0, self.len);

            let header = self.header_mut();

            (*header).magic = MAGIC;
            (*header).version = VERSION;
            (*header).region_size = self.len as u64;
            (*header).payload_size = PAYLOAD_SIZE as u64;
            (*header).capacity = capacity as u64;
            (*header).flags = FLAGS;
            (*header).producer_pid = libc::getpid() as u64;
            (*header).consumer_pid = 0;

            (*self.head()).value = AtomicU64::new(0);
            (*self.tail()).value = AtomicU64::new(0);
        }

        Ok(())
    }

    fn validate(&self) -> io::Result<()> {
        unsafe {
            let header = self.header();

            let magic = ptr::read_volatile(addr_of!((*header).magic));
            let version = ptr::read_volatile(addr_of!((*header).version));
            let region_size = ptr::read_volatile(addr_of!((*header).region_size));
            let payload_size = ptr::read_volatile(addr_of!((*header).payload_size));
            let capacity = ptr::read_volatile(addr_of!((*header).capacity));

            if magic != MAGIC {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "shared memory magic mismatch",
                ));
            }

            if version != VERSION {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "shared memory version mismatch",
                ));
            }

            if region_size != self.len as u64 {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "shared memory region size mismatch",
                ));
            }

            if payload_size != PAYLOAD_SIZE as u64 {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "shared memory payload size mismatch",
                ));
            }

            let capacity_usize = usize::try_from(capacity).map_err(|_| {
                io::Error::new(io::ErrorKind::InvalidData, "capacity does not fit usize")
            })?;

            validate_capacity(capacity_usize)?;

            let expected_region_size = region_size_for_capacity(capacity_usize)?;

            if expected_region_size != self.len {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "capacity does not match region size",
                ));
            }
        }

        Ok(())
    }

    fn push(&self, value: [u8; PAYLOAD_SIZE]) -> Result<(), PushError> {
        unsafe {
            let head = &(*self.head()).value;
            let tail = &(*self.tail()).value;

            let current_head = head.load(Ordering::Relaxed);
            let current_tail = tail.load(Ordering::Acquire);
            let capacity = self.capacity();

            if current_head.wrapping_sub(current_tail) == capacity {
                return Err(PushError::Full);
            }

            let index = current_head & (capacity - 1);
            let slot = self.slot(index);

            ptr::copy_nonoverlapping(value.as_ptr(), (*slot).payload.as_mut_ptr(), PAYLOAD_SIZE);

            head.store(current_head.wrapping_add(1), Ordering::Release);

            Ok(())
        }
    }

    fn pop(&self) -> Option<[u8; PAYLOAD_SIZE]> {
        unsafe {
            let head = &(*self.head()).value;
            let tail = &(*self.tail()).value;

            let current_tail = tail.load(Ordering::Relaxed);
            let current_head = head.load(Ordering::Acquire);

            if current_tail == current_head {
                return None;
            }

            let capacity = self.capacity();
            let index = current_tail & (capacity - 1);
            let slot = self.slot(index);

            let mut out = [0u8; PAYLOAD_SIZE];

            ptr::copy_nonoverlapping((*slot).payload.as_ptr(), out.as_mut_ptr(), PAYLOAD_SIZE);

            tail.store(current_tail.wrapping_add(1), Ordering::Release);

            Some(out)
        }
    }

    fn stats(&self) -> QueueStats {
        unsafe {
            let head = (*self.head()).value.load(Ordering::Acquire);
            let tail = (*self.tail()).value.load(Ordering::Acquire);
            let capacity = self.capacity();

            QueueStats {
                capacity,
                len: head.wrapping_sub(tail),
                head,
                tail,
                region_size: self.len as u64,
            }
        }
    }

    fn set_producer_pid(&self) {
        unsafe {
            (*self.header_mut()).producer_pid = libc::getpid() as u64;
        }
    }

    fn set_consumer_pid(&self) {
        unsafe {
            (*self.header_mut()).consumer_pid = libc::getpid() as u64;
        }
    }

    fn capacity(&self) -> u64 {
        unsafe { ptr::read_volatile(addr_of!((*self.header()).capacity)) }
    }

    unsafe fn header(&self) -> *const SharedHeader {
        self.ptr as *const SharedHeader
    }

    unsafe fn header_mut(&self) -> *mut SharedHeader {
        self.ptr as *mut SharedHeader
    }

    unsafe fn head(&self) -> *mut CounterLine {
        unsafe { self.ptr.add(HEADER_SIZE) as *mut CounterLine }
    }

    unsafe fn tail(&self) -> *mut CounterLine {
        unsafe { self.ptr.add(HEADER_SIZE + COUNTER_SIZE) as *mut CounterLine }
    }

    unsafe fn slot(&self, index: u64) -> *mut Slot48 {
        unsafe { self.ptr.add(BASE_SIZE + index as usize * SLOT_SIZE) as *mut Slot48 }
    }
}

impl Drop for Mapping {
    fn drop(&mut self) {
        unsafe {
            munmap(self.ptr as *mut c_void, self.len);
            close(self.fd);
        }
    }
}

/// Queue producer.
///
/// A process holding `Producer` owns pushes into the queue.
pub struct Producer {
    mapping: Mapping,
}

impl Producer {
    /// Create the queue if missing, or open the existing queue when capacity matches.
    ///
    /// Capacity must be a non-zero power of two.
    pub fn create_or_open(capacity: usize) -> io::Result<Self> {
        let mapping = Mapping::create_or_open(capacity)?;
        mapping.set_producer_pid();

        Ok(Self { mapping })
    }

    /// Open an existing queue as producer.
    pub fn open() -> io::Result<Self> {
        let mapping = Mapping::open_existing()?;
        mapping.set_producer_pid();

        Ok(Self { mapping })
    }

    /// Push one 48-byte value.
    ///
    /// Returns [`PushError::Full`] when the bounded queue has no free slot.
    pub fn push(&self, value: [u8; PAYLOAD_SIZE]) -> Result<(), PushError> {
        self.mapping.push(value)
    }

    /// Queue capacity in slots.
    pub fn capacity(&self) -> u64 {
        self.mapping.stats().capacity
    }

    /// Current queue length in slots.
    pub fn len(&self) -> u64 {
        self.mapping.stats().len
    }

    /// Returns `true` when the queue is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Current queue statistics.
    pub fn stats(&self) -> QueueStats {
        self.mapping.stats()
    }
}

/// Queue consumer.
///
/// A process holding `Consumer` owns pops from the queue.
pub struct Consumer {
    mapping: Mapping,
}

impl Consumer {
    /// Open an existing queue as consumer.
    pub fn open() -> io::Result<Self> {
        let mapping = Mapping::open_existing()?;
        mapping.set_consumer_pid();

        Ok(Self { mapping })
    }

    /// Pop one 48-byte value.
    ///
    /// Returns `None` when the queue is empty.
    pub fn pop(&self) -> Option<[u8; PAYLOAD_SIZE]> {
        self.mapping.pop()
    }

    /// Queue capacity in slots.
    pub fn capacity(&self) -> u64 {
        self.mapping.stats().capacity
    }

    /// Current queue length in slots.
    pub fn len(&self) -> u64 {
        self.mapping.stats().len
    }

    /// Returns `true` when the queue is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Current queue statistics.
    pub fn stats(&self) -> QueueStats {
        self.mapping.stats()
    }
}

/// Open the queue and return statistics.
pub fn stats() -> io::Result<QueueStats> {
    Ok(Mapping::open_existing()?.stats())
}

/// Remove the POSIX shared-memory object.
///
/// Removing an already absent object is treated as success.
pub fn unlink() -> io::Result<()> {
    let name = CString::new(SHM_NAME).unwrap();

    let rc = unsafe { shm_unlink(name.as_ptr()) };

    if rc != 0 {
        let err = io::Error::last_os_error();

        if err.raw_os_error() == Some(libc::ENOENT) {
            return Ok(());
        }

        return Err(err);
    }

    Ok(())
}

fn validate_capacity(capacity: usize) -> io::Result<()> {
    if capacity == 0 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "capacity must be greater than zero",
        ));
    }

    if !capacity.is_power_of_two() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "capacity must be a power of two",
        ));
    }

    Ok(())
}

fn region_size_for_capacity(capacity: usize) -> io::Result<usize> {
    let slots_size = capacity.checked_mul(SLOT_SIZE).ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "capacity overflows region size",
        )
    })?;

    BASE_SIZE
        .checked_add(slots_size)
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "region size overflow"))
}

fn file_size(fd: i32) -> io::Result<usize> {
    let mut stat_buf: libc::stat = unsafe { zeroed() };

    let rc = unsafe { fstat(fd, &mut stat_buf) };

    if rc != 0 {
        return Err(io::Error::last_os_error());
    }

    if stat_buf.st_size <= 0 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "shared memory object has invalid size",
        ));
    }

    usize::try_from(stat_buf.st_size).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidData,
            "shared memory object size does not fit usize",
        )
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn capacity_must_be_non_zero() {
        let err = validate_capacity(0).expect_err("zero capacity is invalid");
        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
    }

    #[test]
    fn capacity_must_be_power_of_two() {
        let err = validate_capacity(3).expect_err("non-power-of-two capacity is invalid");
        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
    }

    #[test]
    fn power_of_two_capacity_is_valid() {
        validate_capacity(1024).expect("power-of-two capacity is valid");
    }

    #[test]
    fn region_size_calculation_matches_layout() {
        let capacity = 1024;
        let expected = size_of::<SharedHeader>()
            + size_of::<CounterLine>()
            + size_of::<CounterLine>()
            + capacity * size_of::<Slot48>();

        assert_eq!(region_size_for_capacity(capacity).unwrap(), expected);
    }

    #[test]
    fn slot_layout_is_one_cacheline() {
        assert_eq!(size_of::<Slot48>(), 64);
        assert_eq!(PAYLOAD_SIZE, 48);
    }
}