palladium_runtime/multi_core/

transport.rs

use std::sync::Arc;

use palladium_actor::{AddrHash, Envelope, MessagePayload};
use palladium_transport::{
    InProcessTransport, MailboxSender, Transport, TransportError, TransportFuture,
};

use crate::placement::PlacementMap;
use crate::ring_buffer::{CacheLineAtomicBool, InterCoreQueue};

/// Cross-core routing transport backed by [`InterCoreQueue`] ring buffers.
///
/// When the destination actor lives on a different core, the message is pushed
/// to `rings[my_core][dest_core]` instead of being delivered directly. The
/// receiving core's event loop is responsible for draining its inbound rings.
///
/// Falls back to the local [`InProcessTransport`] when the destination is on
/// the same core or has no placement entry.
///
/// After each successful cross-core push the sender signals the destination
/// core's entry in the shared `has_work` flag array (cache-line padded to
/// prevent false sharing).  The drain task on the destination core uses this
/// flag to break out of its spin window early and to skip the expensive
/// `yield_now()` / Tokio wakeup when more inter-core work is already in flight.
pub struct InterCoreTransport {
    pub(crate) my_core: usize,
    pub(crate) rings: Arc<Vec<Vec<InterCoreQueue>>>,
    pub(crate) placement: Arc<PlacementMap>,
    pub(crate) local: Arc<InProcessTransport>,
    /// Per-core work-available flags shared across all `InterCoreTransport`
    /// instances.  `has_work[dst]` is set (Release) after a successful push
    /// into the `dst` core's inbound ring.
    pub(crate) has_work: Arc<Vec<CacheLineAtomicBool>>,
    /// Per-core idle state. `true` means destination core is idle/parked.
    /// Senders flip `true -> false` to gate wake signals to one per idle period.
    pub(crate) core_idle: Arc<Vec<CacheLineAtomicBool>>,
}

impl Transport for InterCoreTransport {
    fn try_send(&self, envelope: Envelope, payload: MessagePayload) -> Result<(), TransportError> {
        let dest_core = self
            .placement
            .get(&envelope.destination.path_hash())
            .map(|v| *v);

        if let Some(core) = dest_core {
            if core != self.my_core {
                let msg = palladium_transport::MailboxMessage { envelope, payload };
                return if self.rings[self.my_core][core].push(msg) {
                    // Idle->active transition gate: only signal has_work once
                    // per idle period for this destination core.
                    if self.core_idle[core].take() {
                        self.has_work[core].signal();
                    }
                    Ok(())
                } else {
                    Err(TransportError::MailboxFull)
                };
            }
        }
        // If same core or unknown, use local transport.
        self.local.try_send(envelope, payload)
    }

    fn send<'a>(&'a self, envelope: Envelope, payload: MessagePayload) -> TransportFuture<'a> {
        Box::pin(async move { self.try_send(envelope, payload) })
    }

    fn register(&self, _addr: AddrHash, _mailbox: MailboxSender) -> Result<(), TransportError> {
        // Registration is handled by the core-local InProcessTransport.
        Ok(())
    }

    fn unregister(&self, _addr: AddrHash) -> Result<(), TransportError> {
        Ok(())
    }

    fn can_route(&self, destination: AddrHash) -> bool {
        // InterCoreTransport can attempt to route any address by checking the placement map.
        // It acts as a fallback for non-local addresses.
        self.placement.contains_key(&destination.path_hash())
    }
}

impl InterCoreTransport {
    /// Create a new `InterCoreTransport` owning an N×N ring buffer matrix and
    /// a private `has_work` flag array.
    ///
    /// `capacity` is the number of messages each ring can hold before
    /// returning [`TransportError::MailboxFull`].
    ///
    /// When multiple transport instances need to **share** the same flag array
    /// (i.e. cross-core signalling in `MultiCoreEngine`), construct the struct
    /// directly using the public fields rather than calling this constructor.
    pub fn new(
        my_core: usize,
        num_cores: usize,
        capacity: usize,
        placement: Arc<PlacementMap>,
        local: Arc<InProcessTransport>,
    ) -> Self {
        let rings = (0..num_cores)
            .map(|_| {
                (0..num_cores)
                    .map(|_| InterCoreQueue::new(capacity))
                    .collect()
            })
            .collect();
        let has_work = Arc::new(
            (0..num_cores)
                .map(|_| CacheLineAtomicBool::new(false))
                .collect(),
        );
        let core_idle = Arc::new(
            (0..num_cores)
                .map(|_| CacheLineAtomicBool::new(true))
                .collect(),
        );
        Self {
            my_core,
            rings: Arc::new(rings),
            placement,
            local,
            has_work,
            core_idle,
        }
    }

    /// Deliver a message, routing cross-core via the ring buffer if needed.
    pub fn try_deliver(
        &self,
        envelope: palladium_actor::Envelope,
        payload: palladium_actor::MessagePayload,
    ) -> Result<(), TransportError> {
        let dest_core = self
            .placement
            .get(&envelope.destination.path_hash())
            .map(|v| *v);
        if let Some(core) = dest_core {
            if core != self.my_core {
                let msg = palladium_transport::MailboxMessage { envelope, payload };
                return if self.rings[self.my_core][core].push(msg) {
                    if self.core_idle[core].take() {
                        self.has_work[core].signal();
                    }
                    Ok(())
                } else {
                    Err(TransportError::MailboxFull)
                };
            }
        }
        self.local.try_deliver(envelope, payload)
    }
}