Cargo.lock

@@ -45,6 +45,7 @@ name = "swactor"
 version = "0.1.0"
 dependencies = [
  "crossbeam-queue",
+ "crossbeam-utils",
  "getrandom",
 ]
 

Cargo.toml

@@ -16,6 +16,7 @@ stress = []  # Enable stress tests
 [dependencies]
 getrandom = { version = "0.2", optional = true }
 crossbeam-queue = "0.3.12"
+crossbeam-utils = "0.8.21"
 
 [[bin]]
 name = "bench"

WORKER.md

@@ -0,0 +1,113 @@
+struct MessageRing:
+    buffer: [u8; 4096]
+    head: AtomicUsize      # Sender writes (Release)
+    tail: AtomicUsize      # Worker reads (Acquire)
+
+struct LocalArena:
+    buffer: [u8; 262144]   # Raw message bytes from rings
+    bump: usize
+
+# Key addition: Bucket buffer (indices into arena, not copies)
+struct BucketBuffer:
+    # Pre-allocated array of slices. Max 64K actors, resize if needed.
+    # bucket[i] contains indices of messages for actor i.
+    buckets: Vec<Vec<Slice>>  # Or flat Vec with head/tail if arena-allocated
+    actor_order: Vec<ActorId> # Which actors have messages (for iteration)
+
+struct Worker:
+    worker_id: ID
+    inbox_rings: Vec<MessageRing>  # Per-sender rings
+    arena: LocalArena               # Contiguous message storage
+    buckets: BucketBuffer           # Grouped by actor
+    actor_table: Vec<Actor>         # ActorId -> Actor
+
+# Sender side (unchanged, 30 cycles)
+function send_message(sender, target_worker, actor_id, payload):
+    ring = sender.rings[target_worker]
+    offset = reserve_in_ring(ring, 8 + len(payload))
+    serialize(ring.buffer[offset:], actor_id, len(payload), payload)
+    ring.head.store(offset + 8 + len(payload), Release)
+
+# Worker side: Three-phase pipeline
+function worker_run(worker):
+    while true:
+        # PHASE 1: DRAIN (same as before, ~5 cycles per message)
+        # -----------------------------------------------
+        for ring in worker.inbox_rings:
+            head = ring.head.load(Acquire)
+            tail = ring.tail.load(Relaxed)
+            if head == tail: continue
+            
+            # Copy sequential chunk from ring -> arena (hardware prefetch)
+            size = head - tail
+            memcpy(worker.arena.buffer[worker.arena.bump:], 
+                   ring.buffer[tail:], size)
+            
+            # Parse boundaries while copying to avoid second pass
+            parse_and_bucket(worker.arena, worker.arena.bump, size, worker.buckets)
+            
+            worker.arena.bump += size
+            ring.tail.store(head, Relaxed)
+        
+        if worker.arena.bump == 0: 
+            cpu_relax()
+            continue
+        
+        # PHASE 2: RADIX BUCKET (O(N), deterministic ~300 cycles)
+        # ------------------------------------------------------
+        # We already built buckets during parse_and_bucket above, 
+        # but if we deferred parsing, do it now:
+        
+        # Option A: If parsed during drain (optimal)
+        # Buckets already filled with (offset, len) pairs pointing into arena
+        
+        # Option B: Linear scan to build buckets (if raw bytes in arena)
+        offset = 0
+        while offset < worker.arena.bump:
+            actor_id = read_u32(arena[offset:])
+            msg_len = read_u32(arena[offset+4:])
+            
+            # Append to actor's bucket (Vec push, amortized O(1))
+            # Each bucket entry: (offset, msg_len) = 16 bytes
+            worker.buckets.buckets[actor_id].append((offset+8, msg_len))
+            
+            # Track unique actors (optional, avoids empty bucket scans)
+            if worker.buckets.buckets[actor_id].len() == 1:
+                worker.buckets.actor_order.append(actor_id)
+            
+            offset += 8 + msg_len
+        
+        # PHASE 3: PROCESS BY ACTOR (hidden message fetch, hot actor state)
+        # ----------------------------------------------------------------
+        for actor_id in worker.buckets.actor_order:
+            actor = worker.actor_table[actor_id]  # First access: L3 miss (250 cycles)
+            
+            # Prefetch next actor's state while processing current (optional)
+            prefetch_actor(worker.buckets.actor_order, worker.actor_table)
+            
+            # Process all messages for this actor
+            # arena[slice] is L1 hit (12 cycles) - scanned sequentially within actor
+            for (msg_offset, msg_len) in worker.buckets.buckets[actor_id]:
+                msg_data = worker.arena.buffer[msg_offset : msg_offset+msg_len]
+                actor.process(msg_data)  # 100 cycles work
+            
+            # actor state stays in L1 for entire inner loop
+            
+        # PHASE 4: RESET (zero cost)
+        worker.arena.bump = 0
+        clear_buckets(worker.buckets)  # Just reset lengths, don't free
+
+# Helper: Parse during drain to avoid touching bytes twice
+function parse_and_bucket(arena, base_offset, size, buckets):
+    ptr = 0
+    while ptr < size:
+        actor_id = read_u32(arena[base_offset + ptr:])
+        msg_len = read_u32(arena[base_offset + ptr + 4:])
+        
+        # Append metadata to bucket (16 bytes: offset, len)
+        buckets.buckets[actor_id].append((base_offset + ptr + 8, msg_len))
+        
+        if buckets.buckets[actor_id].len() == 1:
+            buckets.actor_order.append(actor_id)
+        
+        ptr += 8 + msg_len

src/channel/mod.rs

@@ -1,3 +1,5 @@
+pub mod spsc;
+
 use std::{collections::VecDeque, sync::{Arc, Mutex}};
 
 use crossbeam_queue::ArrayQueue;

src/channel/spsc.rs

@@ -0,0 +1,198 @@
+use crossbeam_utils::CachePadded;
+use std::cell::UnsafeCell;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+const CACHELINE: usize = 64;
+
+struct Ring<const N: usize> {
+    head: CachePadded<AtomicUsize>,
+    tail: CachePadded<AtomicUsize>,
+    buf: UnsafeCell<[u8; N]>,
+}
+
+unsafe impl<const N: usize> Send for Ring<N> {}
+unsafe impl<const N: usize> Sync for Ring<N> {}
+
+pub struct Producer<const N: usize> {
+    inner: Arc<Ring<N>>,
+    head: usize, // cached local copy
+    tail: usize, // cached local copy
+}
+
+pub struct Consumer<const N: usize> {
+    inner: Arc<Ring<N>>,
+    head: usize,
+    tail: usize,
+}
+
+unsafe impl<const N: usize> Send for Producer<N> {}
+unsafe impl<const N: usize> Send for Consumer<N> {}
+
+pub fn channel<const N: usize>() -> (Producer<N>, Consumer<N>) {
+    assert!(N.is_power_of_two(), "capacity must be power of 2");
+
+    let inner = Arc::new(Ring {
+        head: CachePadded::new(AtomicUsize::new(0)),
+        tail: CachePadded::new(AtomicUsize::new(0)),
+        buf: UnsafeCell::new([0u8; N]),
+    });
+
+    let producer = Producer {
+        inner: inner.clone(),
+        head: 0,
+        tail: 0,
+    };
+    let consumer = Consumer {
+        inner,
+        head: 0,
+        tail: 0,
+    };
+    (producer, consumer)
+}
+
+impl<const N: usize> Producer<N> {
+    const MASK: usize = N - 1;
+
+    /// Copy `src` into circular buffer at `pos`, handling wraparound.
+    #[inline]
+    fn write_at(buf: &mut [u8; N], pos: usize, src: &[u8]) {
+        let start = pos & Self::MASK;
+        let end = start + src.len();
+
+        if end <= N {
+            // No wrap: single copy
+            buf[start..end].copy_from_slice(src);
+        } else {
+            // Wrap: split into two copies
+            let first = N - start;
+            buf[start..].copy_from_slice(&src[..first]);
+            buf[..src.len() - first].copy_from_slice(&src[first..]);
+        }
+    }
+
+    /// Attempts to write a byte slice into the buffer, returning the number of bytes written
+    pub fn try_write(&mut self, data: &[u8]) -> usize {
+        let needed = 4 + data.len();
+
+        // Refresh cached tail if we think we're full
+        let available = N - self.head.wrapping_sub(self.tail);
+        if available < needed {
+            self.tail = self.inner.tail.load(Ordering::Acquire);
+            let available = N - self.head.wrapping_sub(self.tail);
+            if available < needed {
+                return 0;
+            }
+        }
+
+        let buf = unsafe { &mut *self.inner.buf.get() };
+
+        // Write length prefix (4 bytes LE)
+        Self::write_at(buf, self.head, &(data.len() as u32).to_le_bytes());
+
+        // Write payload
+        Self::write_at(buf, self.head + 4, data);
+
+        self.head = self.head.wrapping_add(needed);
+        self.inner.head.store(self.head, Ordering::Release);
+
+        data.len()
+    }
+}
+
+impl<const N: usize> Consumer<N> {
+    const MASK: usize = N - 1;
+
+    #[inline]
+    fn read_at(buf: &[u8; N], pos: usize, dst: &mut [u8]) {
+        let start = pos & Self::MASK;
+        let end = start + dst.len();
+
+        if end <= N {
+            dst.copy_from_slice(&buf[start..end]);
+        } else {
+            let first = N - start;
+            let second = end - N;
+            dst[..first].copy_from_slice(&buf[start..]);
+            dst[first..].copy_from_slice(&buf[..second]);
+        }
+    }
+
+    /// Pop next message into provided buffer. Returns message length, or None if empty.
+    /// Panics if buffer is too small for the message.
+    pub fn pop_into(&mut self, dst: &mut [u8]) -> Option<usize> {
+        let filled = self.head.wrapping_sub(self.tail);
+        if filled < 4 {
+            self.head = self.inner.head.load(Ordering::Acquire);
+            let filled = self.head.wrapping_sub(self.tail);
+            if filled < 4 {
+                return None;
+            }
+        }
+
+        let buf = unsafe { &*self.inner.buf.get() };
+
+        let mut len_bytes = [0u8; 4];
+        Self::read_at(buf, self.tail, &mut len_bytes);
+        let len = u32::from_le_bytes(len_bytes) as usize;
+
+        let filled = self.head.wrapping_sub(self.tail);
+        if filled < 4 + len {
+            self.head = self.inner.head.load(Ordering::Acquire);
+            let filled = self.head.wrapping_sub(self.tail);
+            if filled < 4 + len {
+                return None;
+            }
+        }
+
+        Self::read_at(buf, self.tail + 4, &mut dst[..len]);
+
+        self.tail = self.tail.wrapping_add(4 + len);
+        self.inner.tail.store(self.tail, Ordering::Release);
+
+        Some(len)
+    }
+
+    pub fn is_empty(&self) -> bool {
+        let head = self.inner.head.load(Ordering::Acquire);
+        head == self.tail
+    }
+}
+
+// ============ Demo ============
+#[test]
+fn sanity() {
+    use std::thread;
+
+    let (mut tx, mut rx) = channel::<4096>();
+
+    let num_messages = 1000;
+    let producer = thread::spawn(move || {
+        for i in 0..num_messages {
+            let msg = format!("message {}", i);
+            while tx.try_write(msg.as_bytes()) == 0 {
+                std::hint::spin_loop();
+            }
+        }
+        tx.try_write(b"DONE");
+    });
+
+    let consumer = thread::spawn(move || {
+        let mut count = 0;
+        loop {
+            let mut buf = vec![];
+            if let Some(_) = rx.pop_into(&mut buf) {
+                if buf == b"DONE" {
+                    break;
+                }
+                count += 1;
+            } else {
+                std::hint::spin_loop();
+            }
+        }
+        assert_eq!(num_messages, count, "failed to process all messages");
+    });
+
+    producer.join().unwrap();
+    consumer.join().unwrap();
+}

src/lib.rs

@@ -6,6 +6,7 @@ pub use error::Error;
 
 mod router;
 pub mod runtime;
+mod worker;
 
 #[cfg(feature = "getrandom")]
 pub(crate) fn get_random(buf: &mut [u8]) {

src/worker.rs

@@ -0,0 +1,46 @@
+type WorkerId = usize;
+type ActorId = usize;
+
+struct Actor(u8);
+
+const MESSAGE_RING_BUFFER_SIZE: usize = 4096;
+const LOCAL_ARENA_BUFFER_SIZE: usize = 262144;
+// hitting the maximum would imply reading nothing but length prefixes from the ring channel
+const MAX_MESSAGES_PER_DRAIN: usize = MESSAGE_RING_BUFFER_SIZE / 4;
+
+use crate::channel::spsc::Consumer as RingBuffer;
+
+#[repr(align(64))]
+struct LocalArena {
+    data: [u8; LOCAL_ARENA_BUFFER_SIZE],
+    offsets: [u32; MAX_MESSAGES_PER_DRAIN],
+}
+struct Worker {
+    id: WorkerId,
+    inbox_rings: Vec<RingBuffer<MESSAGE_RING_BUFFER_SIZE>>,
+    arena: LocalArena,
+    actor_table: Vec<Actor>,
+}
+
+impl Worker {
+    pub fn run(&mut self) {
+        // cannot overflow the arena buffer
+        debug_assert!(self.inbox_rings.len() * MESSAGE_RING_BUFFER_SIZE < LOCAL_ARENA_BUFFER_SIZE);
+
+        loop {
+            // drain messages into our local memory arena buffer
+            for ring in &self.inbox_rings {
+                // logic here
+            }
+        }
+    }
+}
+
+
+
+struct BucketBuffer {
+    // Pre-allocated array of slices. Max 64K actors, resize if needed.
+    // bucket[i] contains indices of messages for actor i.
+    buckets: Vec<Vec<u8>>,     // Or flat Vec with head/tail if arena-allocated
+    actor_order: Vec<ActorId>, // Which actors have messages (for iteration)
+}