palladium_runtime/

ring_buffer.rs

use palladium_transport::MailboxMessage;
use std::sync::atomic::{AtomicBool, Ordering};

// ── CacheLineAtomicBool ───────────────────────────────────────────────────────

/// A single-bit flag padded to a full cache line (64 bytes) to prevent false
/// sharing when an array of flags is accessed concurrently from different cores.
///
/// Used for the per-core `has_work` array: the sender core sets the destination
/// core's flag after a ring push; the drain task reads it to skip ring scanning
/// when there is no pending inter-core traffic.
#[repr(align(64))]
pub struct CacheLineAtomicBool {
    pub value: AtomicBool,
}

impl CacheLineAtomicBool {
    pub const fn new(v: bool) -> Self {
        Self {
            value: AtomicBool::new(v),
        }
    }

    /// Signal that work is available (Release so ring push is visible to reader).
    #[inline]
    pub fn signal(&self) {
        self.value.store(true, Ordering::Release);
    }

    /// Set the flag to `true` with Release ordering.
    #[inline]
    pub fn set_true(&self) {
        self.value.store(true, Ordering::Release);
    }

    /// Set the flag to `false` with Release ordering.
    #[inline]
    pub fn set_false(&self) {
        self.value.store(false, Ordering::Release);
    }

    /// Load the flag with Acquire ordering (paired with `signal`).
    #[inline]
    pub fn is_set(&self) -> bool {
        self.value.load(Ordering::Acquire)
    }

    /// Atomically read and clear the flag (AcqRel).
    #[inline]
    pub fn take(&self) -> bool {
        self.value.swap(false, Ordering::AcqRel)
    }
}

// ── InterCoreQueue ────────────────────────────────────────────────────────────

/// Cache-line-padded SPSC inter-core message queue.
///
/// One instance per (source_core, dest_core) pair.  The `#[repr(align(64))]`
/// ensures that adjacent queue entries in a `Vec<InterCoreQueue>` do not share
/// a cache line, preventing false sharing between cores that access different
/// pairs.
///
/// Uses [`crossbeam_queue::ArrayQueue`] for wait-free push/pop.  `push`
/// returns `false` when the queue is at capacity — callers should treat this
/// as a backpressure signal.
#[repr(align(64))]
pub struct InterCoreQueue {
    inner: crossbeam_queue::ArrayQueue<MailboxMessage>,
}

impl InterCoreQueue {
    pub fn new(capacity: usize) -> Self {
        Self {
            inner: crossbeam_queue::ArrayQueue::new(capacity),
        }
    }

    /// Push a message. Returns `false` if the queue is full.
    pub fn push(&self, msg: MailboxMessage) -> bool {
        self.inner.push(msg).is_ok()
    }

    /// Pop the next message, or `None` if empty.
    pub fn pop(&self) -> Option<MailboxMessage> {
        self.inner.pop()
    }

    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    pub fn len(&self) -> usize {
        self.inner.len()
    }

    pub fn capacity(&self) -> usize {
        self.inner.capacity()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use palladium_actor::{AddrHash, Envelope, MessagePayload};

    fn make_msg(tag: u32) -> MailboxMessage {
        let src = AddrHash::synthetic(b"src");
        let dst = AddrHash::synthetic(b"dst");
        MailboxMessage {
            envelope: Envelope::new(src, dst, tag as u64, 0),
            payload: MessagePayload::local(tag),
        }
    }

    #[test]
    fn ring_buffer_push_pop_fifo() {
        let q = InterCoreQueue::new(8);
        assert!(q.push(make_msg(1)));
        assert!(q.push(make_msg(2)));
        assert!(q.push(make_msg(3)));

        let m1 = q.pop().unwrap();
        let m2 = q.pop().unwrap();
        let m3 = q.pop().unwrap();

        // FIFO: pop order matches push order.
        assert_eq!(m1.envelope.type_tag, 1);
        assert_eq!(m2.envelope.type_tag, 2);
        assert_eq!(m3.envelope.type_tag, 3);
        assert!(q.pop().is_none());
    }

    #[test]
    fn ring_buffer_bounded_capacity_returns_false_when_full() {
        let q = InterCoreQueue::new(4);
        assert!(q.push(make_msg(0)));
        assert!(q.push(make_msg(1)));
        assert!(q.push(make_msg(2)));
        assert!(q.push(make_msg(3)));

        // Queue is now at capacity; next push must fail.
        assert!(!q.push(make_msg(4)), "push should fail when queue is full");
        assert_eq!(q.len(), 4);
    }

    #[test]
    fn ring_buffer_is_empty_and_len() {
        let q = InterCoreQueue::new(4);
        assert!(q.is_empty());
        assert_eq!(q.len(), 0);

        q.push(make_msg(7));
        assert!(!q.is_empty());
        assert_eq!(q.len(), 1);

        q.pop();
        assert!(q.is_empty());
    }

    #[test]
    fn ring_buffer_cache_line_aligned() {
        // Struct must be at least one cache line (64 bytes) aligned.
        // crossbeam's ArrayQueue is internally 128-byte aligned, so we may get
        // more than 64 — that's fine and still prevents false sharing.
        assert!(std::mem::align_of::<InterCoreQueue>() >= 64);
    }

    /// Two threads: producer pushes 1000 messages; consumer pops all of them.
    /// Verifies the queue is safe across threads (MailboxMessage: Send).
    #[test]
    fn ring_buffer_cross_thread_send_recv() {
        use std::sync::Arc;

        const N: u32 = 1000;
        const CAP: usize = 128;
        let q = Arc::new(InterCoreQueue::new(CAP));

        let q_prod = Arc::clone(&q);
        let producer = std::thread::spawn(move || {
            let mut sent = 0u32;
            while sent < N {
                if q_prod.push(make_msg(sent)) {
                    sent += 1;
                } else {
                    std::hint::spin_loop();
                }
            }
        });

        let mut received = Vec::with_capacity(N as usize);
        while received.len() < N as usize {
            if let Some(msg) = q.pop() {
                received.push(msg.envelope.type_tag);
            } else {
                std::hint::spin_loop();
            }
        }

        producer.join().unwrap();
        assert_eq!(received.len(), N as usize);
        // FIFO: values arrive in order 0..N.
        for (i, &tag) in received.iter().enumerate() {
            assert_eq!(tag, i as u64, "FIFO violated at index {i}");
        }
    }
}