palladium_runtime/

placement.rs

use std::sync::atomic::{AtomicU64, Ordering};

use dashmap::DashMap;
use palladium_actor::AddrHash;

// ── Placement ─────────────────────────────────────────────────────────────────

/// Specifies where a newly spawned actor should be placed.
#[derive(Debug, Clone, PartialEq)]
pub enum Placement {
    /// Engine chooses the core using `hash(ActorPath) % num_cores`.
    Auto,
    /// Pin the actor to a specific core index.
    ///
    /// Panics at spawn time if `core_id >= num_cores`.
    Core(usize),
    /// Place the actor on the same core as an existing actor.
    ///
    /// Returns `SpawnError::ColocateTargetNotFound` if the target
    /// `AddrHash` is not registered in the placement map.
    ColocateWith(AddrHash),
}

// ── PlacementMap ──────────────────────────────────────────────────────────────

/// Shared, concurrent map from path hash (u64) to core index.
///
/// Written only at actor spawn/stop time (rare).
/// Read on every cross-core send (frequent).
/// [`DashMap`] provides shard-based concurrent access with no global lock.
pub type PlacementMap = DashMap<u64, usize>;

// ── default_core_for ──────────────────────────────────────────────────────────

/// Hash-based default placement: distributes actors across cores by taking the
/// path hash bits of the `AddrHash` modulo `num_cores`.
pub fn default_core_for(hash: AddrHash, num_cores: usize) -> usize {
    debug_assert!(num_cores > 0, "num_cores must be > 0");
    (hash.path_hash() as usize) % num_cores
}

// ── CoreStats ─────────────────────────────────────────────────────────────────

/// Per-core statistics counters.
///
/// `#[repr(align(64))]` ensures each instance occupies its own cache line,
/// preventing false sharing when cores update their own stats concurrently.
#[repr(align(64))]
pub struct CoreStats {
    pub messages_processed: AtomicU64,
    pub actors_spawned: AtomicU64,
    pub mailbox_drops: AtomicU64,
}

impl CoreStats {
    pub fn new() -> Self {
        Self {
            messages_processed: AtomicU64::new(0),
            actors_spawned: AtomicU64::new(0),
            mailbox_drops: AtomicU64::new(0),
        }
    }

    pub fn messages_processed(&self) -> u64 {
        self.messages_processed.load(Ordering::Relaxed)
    }

    pub fn actors_spawned(&self) -> u64 {
        self.actors_spawned.load(Ordering::Relaxed)
    }

    pub fn mailbox_drops(&self) -> u64 {
        self.mailbox_drops.load(Ordering::Relaxed)
    }
}

impl Default for CoreStats {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn core_stats_cache_line_aligned() {
        assert_eq!(std::mem::align_of::<CoreStats>(), 64);
    }

    #[test]
    fn core_stats_counters_start_at_zero() {
        let s = CoreStats::new();
        assert_eq!(s.messages_processed(), 0);
        assert_eq!(s.actors_spawned(), 0);
        assert_eq!(s.mailbox_drops(), 0);
    }

    #[test]
    fn core_stats_increments() {
        let s = CoreStats::new();
        s.messages_processed.fetch_add(5, Ordering::Relaxed);
        s.actors_spawned.fetch_add(2, Ordering::Relaxed);
        s.mailbox_drops.fetch_add(1, Ordering::Relaxed);
        assert_eq!(s.messages_processed(), 5);
        assert_eq!(s.actors_spawned(), 2);
        assert_eq!(s.mailbox_drops(), 1);
    }

    #[test]
    fn default_core_for_distributes_evenly() {
        // Generate 10_000 synthetic AddrHashes and verify none is placed
        // out of range, and the distribution across 4 cores is roughly even
        // (within ±30% of expected 2500 per core).
        const N: usize = 10_000;
        const NUM_CORES: usize = 4;
        let expected = N / NUM_CORES;
        let tolerance = expected * 30 / 100; // ±30%

        let mut counts = [0usize; NUM_CORES];
        for i in 0..N {
            let bytes = i.to_le_bytes();
            let hash = AddrHash::synthetic(&bytes);
            let core = default_core_for(hash, NUM_CORES);
            assert!(core < NUM_CORES, "core {core} out of range");
            counts[core] += 1;
        }

        for (c, &count) in counts.iter().enumerate() {
            assert!(
                (expected as isize - count as isize).unsigned_abs() <= tolerance,
                "core {c} count {count} is outside expected range [{}, {}]",
                expected - tolerance,
                expected + tolerance,
            );
        }
    }

    #[test]
    fn default_core_for_single_core_always_zero() {
        for i in 0..100u64 {
            let bytes = i.to_le_bytes();
            let hash = AddrHash::synthetic(&bytes);
            assert_eq!(default_core_for(hash, 1), 0);
        }
    }

    #[test]
    fn placement_map_insert_lookup_remove() {
        let map: PlacementMap = DashMap::new();
        let addr = AddrHash::synthetic(b"test-actor");

        map.insert(addr.path_hash(), 2);
        assert_eq!(map.get(&addr.path_hash()).map(|v| *v), Some(2));

        map.remove(&addr.path_hash());
        assert!(map.get(&addr.path_hash()).is_none());
    }

    #[test]
    fn placement_map_concurrent_reads() {
        use std::sync::Arc;

        let map = Arc::new(PlacementMap::new());

        // Pre-populate 100 actors on core 0.
        let hashes: Vec<AddrHash> = (0..100u64)
            .map(|i| AddrHash::synthetic(&i.to_le_bytes()))
            .collect();
        for (i, &hash) in hashes.iter().enumerate() {
            map.insert(hash.path_hash(), i % 4);
        }

        // 8 reader threads each do 1000 lookups — no panics or data races.
        let hashes = Arc::new(hashes);
        let threads: Vec<_> = (0..8)
            .map(|_| {
                let m = Arc::clone(&map);
                let h = Arc::clone(&hashes);
                std::thread::spawn(move || {
                    for i in 0..1000usize {
                        let hash = h[i % 100];
                        let _ = m.get(&hash.path_hash()).map(|v| *v);
                    }
                })
            })
            .collect();

        for t in threads {
            t.join().unwrap();
        }
    }
}