From abedaf6307cc130faf1e7d0a9cf2c8a731f37b30 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Thu, 5 Feb 2026 10:44:48 +0700 Subject: [PATCH] feat: workstealing, less contention for multithreaded --- Cargo.lock | 27 ++++++ Cargo.toml | 2 + DESIGN.md | 55 ------------ WORKER.md | 113 ------------------------- src/channel/mod.rs | 25 ++++-- src/channel/spsc.rs | 198 -------------------------------------------- src/lib.rs | 1 - src/router.rs | 2 +- src/runtime.rs | 166 +++++++++++++++++++++++++++++-------- src/worker.rs | 46 ---------- 10 files changed, 180 insertions(+), 455 deletions(-) delete mode 100644 DESIGN.md delete mode 100644 WORKER.md delete mode 100644 src/channel/spsc.rs delete mode 100644 src/worker.rs diff --git a/Cargo.lock b/Cargo.lock index 1e6027b..561715c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,6 +8,25 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -40,13 +59,21 @@ version = "0.2.180" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc" +[[package]] +name = "smallvec" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" + [[package]] name = "swactor" version = "0.1.0" dependencies = [ + "crossbeam-deque", "crossbeam-queue", "crossbeam-utils", "getrandom", + "smallvec", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index d2ebf83..6471309 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,8 +15,10 @@ stress = [] # Enable stress tests [dependencies] getrandom = { version = "0.2", optional = true } +crossbeam-deque = "0.8" crossbeam-queue = "0.3.12" crossbeam-utils = "0.8.21" +smallvec = "1.13" [[bin]] name = "bench" diff --git a/DESIGN.md b/DESIGN.md deleted file mode 100644 index 93f57e6..0000000 --- a/DESIGN.md +++ /dev/null @@ -1,55 +0,0 @@ -# Design goals -Get as much usability and speed as possible while keeping line count low. Aim for no footguns, ability to plug in -logic easily, and run near anywhere. We may make this a `![no_std]` library, but the MVP will use the -memory allocator and threading provided by the rust standard library. - -We are not building a new erlang/BEAM. Minimal feature set means spawning actor processes, not having supervisiors, lots of process -monitoring tools, prempting, etc. - -# FIXME -This is slightly obsolete. Was necessary in order to concentrate on getting the basic skeleton up, now it's a distraction and unreliable. -After getting the benches/etc finished, move this into an `ARCHITECTURE.md` file, have it make sense. - - -## Actor model - -An actor has: - - - An inbox: - this is a mpsc channel that the runtime/router dumps messages into and the actor consumes when the runtime loads it - Implemented as a barebones atomic ring buffer. The router is responsible for inserting messages. - - - an outbox channel connection: - this is a mpmc channel that is implemented by the runtime and router. Actors on this specific channel put responses and outgoing messages into this channel, to be routed to the given address. - - - a growable and mutable state: - An actor owns some, from the runtime perspective, type erased bytes. The actor when processing messages can access its own state, but no other task can. This includes viewing. - - - a set of functions for processing messages: - When the runtime loads the actor, it locks the inbox and attempts to process the messages therein. - -## Runtime - -In order for an actor to consume and send messages, it is processed by a runtime. The runtime, in order to negotiate messages between -actors, possesses a router. - -A runtime has: - - An actor processing thread(s): - the processor will mark an actor as busy, load its state and inbox, and begin consuming messages from the inbox. The number of messages consumed is determined by the runtime. A good start is a backpressure strategy: after loading, process messages until mailbox is empty or size drops below a threshold (e.g., "drain to 50%"). - - - A message router: - the router is responsible for ensuring messages posted by actors get delivered to the appropriate inbox. - - - An atomic ring buffer containing thread-safe references to actors that are not currently loaded. Actors are popped off the buffer, messages are - processed, and the reference is returned to the buffer/queue before the next actor is loaded. - -## Router - -The router is the engine for message delivery. It posesses: - - - An actor address book: - The address book maps actor ids to `Sender` references that can be used to deliver messages to the actor inbox. - - - Its own inbox: - The router possesses its own mpsc queue where references to messages are stored. The router will process this queue by dereferencing and writing directly into the recipient's inbox buffer. - diff --git a/WORKER.md b/WORKER.md deleted file mode 100644 index 9000a8d..0000000 --- a/WORKER.md +++ /dev/null @@ -1,113 +0,0 @@ -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> # Or flat Vec with head/tail if arena-allocated - actor_order: Vec # Which actors have messages (for iteration) - -struct Worker: - worker_id: ID - inbox_rings: Vec # Per-sender rings - arena: LocalArena # Contiguous message storage - buckets: BucketBuffer # Grouped by actor - actor_table: Vec # 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 \ No newline at end of file diff --git a/src/channel/mod.rs b/src/channel/mod.rs index 93b865e..fbccdb5 100644 --- a/src/channel/mod.rs +++ b/src/channel/mod.rs @@ -1,19 +1,21 @@ -pub mod spsc; -use std::{collections::VecDeque, sync::{Arc, Mutex}}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; -use crossbeam_queue::ArrayQueue; +use crossbeam_queue::{ArrayQueue, SegQueue}; pub struct HybridChannel { ring: ArrayQueue, - overflow: Mutex>, + overflow: SegQueue, + overflow_len: AtomicUsize, } impl HybridChannel { pub fn new(capacity: usize) -> Self { Self { ring: ArrayQueue::new(capacity), - overflow: Mutex::new(VecDeque::new()), + overflow: SegQueue::new(), + overflow_len: AtomicUsize::new(0), } } @@ -21,7 +23,8 @@ impl HybridChannel { match self.ring.push(value) { Ok(()) => Ok(()), Err(v) => { - self.overflow.lock().unwrap().push_back(v); + self.overflow.push(v); + self.overflow_len.fetch_add(1, Ordering::Relaxed); Ok(()) } } @@ -32,11 +35,17 @@ impl HybridChannel { return Some(value); } - self.overflow.lock().unwrap().pop_front() + match self.overflow.pop() { + Some(value) => { + self.overflow_len.fetch_sub(1, Ordering::Relaxed); + Some(value) + } + None => None, + } } pub fn len(&self) -> usize { - self.ring.len() + self.overflow.lock().unwrap().len() + self.ring.len() + self.overflow_len.load(Ordering::Relaxed) } } diff --git a/src/channel/spsc.rs b/src/channel/spsc.rs deleted file mode 100644 index a44c87e..0000000 --- a/src/channel/spsc.rs +++ /dev/null @@ -1,198 +0,0 @@ -use crossbeam_utils::CachePadded; -use std::cell::UnsafeCell; -use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; - -const CACHELINE: usize = 64; - -struct Ring { - head: CachePadded, - tail: CachePadded, - buf: UnsafeCell<[u8; N]>, -} - -unsafe impl Send for Ring {} -unsafe impl Sync for Ring {} - -pub struct Producer { - inner: Arc>, - head: usize, // cached local copy - tail: usize, // cached local copy -} - -pub struct Consumer { - inner: Arc>, - head: usize, - tail: usize, -} - -unsafe impl Send for Producer {} -unsafe impl Send for Consumer {} - -pub fn channel() -> (Producer, Consumer) { - 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 Producer { - 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 Consumer { - 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 { - 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(); -} diff --git a/src/lib.rs b/src/lib.rs index 0a3122c..1d67009 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,7 +6,6 @@ pub use error::Error; mod router; pub mod runtime; -mod worker; #[cfg(feature = "getrandom")] pub(crate) fn get_random(buf: &mut [u8]) { diff --git a/src/router.rs b/src/router.rs index 34157f4..d015fe1 100644 --- a/src/router.rs +++ b/src/router.rs @@ -21,7 +21,7 @@ impl SenderT for Sender { if let Some(msg) = envelope.downcast_ref::() { // FIXME: we are directly cloning the contents of the Arc pointer here // Do we want to? Should we provide another way? - // + // // The standard concept of an actor has message and state // isolation, so we should leave this as is. However, we should // make it clear and obvious this pathway is the heavy, contained diff --git a/src/runtime.rs b/src/runtime.rs index 7a9f768..1a5df3b 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,6 +1,9 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::thread::{self, JoinHandle}; +use std::time::Duration; + +use crossbeam_deque::{Injector, Steal, Stealer, Worker}; use crate::channel::HybridChannel; use crate::{ @@ -56,10 +59,24 @@ impl Default for RuntimeConfig { } } +/// Per-worker state for work-stealing scheduler +struct WorkerState { + local: Worker>, // FIFO for fairness + id: usize, +} + +/// Shared scheduler state for work-stealing +struct SchedulerState { + injector: Injector>, // For spawn() + stealers: Vec>>, // For work-stealing + is_running: AtomicBool, +} + /// The `Runtime` struct is the primary gateway for interacting with the framework. pub struct Runtime { config: RuntimeConfig, - actor_queue: HybridChannel>, + actor_queue: HybridChannel>, // Used for single-threaded mode + scheduler: Option>, // Used for multi-threaded mode router_interface: Sender, router: Option>, // `None` if single-threaded @@ -113,6 +130,7 @@ impl Runtime { Self { config, actor_queue, + scheduler: None, // Initialized in run() for multi-threaded mode router_interface: router_sender, is_running: AtomicBool::new(false), router: router_option, @@ -133,9 +151,16 @@ impl Runtime { Error::from("Runtime error: failed to add actor to router. Router inbox full") })?; - self.actor_queue - .push(Box::new(Actor::new(inbox, actor))) - .map_err(|_| Error::from("Runtime error: Failed to spawn actor. Queue full."))?; + let boxed_actor: Box = Box::new(Actor::new(inbox, actor)); + + // Multi-threaded: use injector, single-threaded: use shared queue + if let Some(ref scheduler) = self.scheduler { + scheduler.injector.push(boxed_actor); + } else { + self.actor_queue + .push(boxed_actor) + .map_err(|_| Error::from("Runtime error: Failed to spawn actor. Queue full."))?; + } Ok(addr) } @@ -193,48 +218,56 @@ impl Runtime { .take() .expect("Router must be present for multi-threaded runtime"); + let num_workers = self.config.num_threads - 1; + + // Create workers and collect stealers for work-stealing + let mut workers = Vec::with_capacity(num_workers); + let mut stealers = Vec::with_capacity(num_workers); + for id in 0..num_workers { + let worker = WorkerState { + local: Worker::new_fifo(), // FIFO for fairness + id, + }; + stealers.push(worker.local.stealer()); + workers.push(worker); + } + + let scheduler = Arc::new(SchedulerState { + injector: Injector::new(), + stealers, + is_running: AtomicBool::new(true), + }); + + // Transfer any actors spawned before run() to the injector + while let Some(actor) = self.actor_queue.pop() { + scheduler.injector.push(actor); + } + + self.scheduler = Some(scheduler.clone()); let rt = Arc::new(self); let mut handles: Vec> = vec![]; // Router thread owns the router directly - no synchronization needed + // Process multiple ticks per cycle to maximize throughput + const ROUTER_BATCH_SIZE: usize = 64; let router_handle = { let ctx = rt.clone(); + let sched = scheduler.clone(); thread::spawn(move || { - while ctx.is_running.load(Ordering::Acquire) { - router.tick(&ctx); - thread::yield_now(); + while sched.is_running.load(Ordering::Acquire) { + for _ in 0..ROUTER_BATCH_SIZE { + router.tick(&ctx); + } } }) }; handles.push(router_handle); - // Spawn worker threads - let num_workers = rt.config.num_threads - 1; - for _ in 0..num_workers { + // Spawn worker threads, each owns its WorkerState + for worker in workers { let ctx = rt.clone(); - let handle = thread::spawn(move || { - while ctx.is_running.load(Ordering::Acquire) { - if let Some(mut actor) = ctx.actor_queue.pop() { - actor.tick(&ctx); - // FIXME: Justify this loop. It is here to prevent panics when the - // actor queue is full, but results in a spinlock. - loop { - match ctx.actor_queue.push(actor) { - Ok(()) => break, - Err(a) => { - actor = a; - if !ctx.is_running.load(Ordering::Acquire) { - break; - } - thread::yield_now(); - } - } - } - } else { - thread::yield_now(); - } - } - }); + let sched = scheduler.clone(); + let handle = thread::spawn(move || worker_loop(ctx, sched, worker)); handles.push(handle); } @@ -256,5 +289,72 @@ impl Runtime { /// Signal all workers to stop pub fn shutdown(&self) { self.is_running.store(false, Ordering::Release); + // Also stop the scheduler if multi-threaded + if let Some(ref scheduler) = self.scheduler { + scheduler.is_running.store(false, Ordering::Release); + } + } +} + +/// Work-stealing worker loop for multi-threaded runtime +fn worker_loop(ctx: Arc, scheduler: Arc, worker: WorkerState) { + const TICKS_PER_ACTOR: usize = 4; + const INJECTOR_CHECK_INTERVAL: usize = 64; // Check injector every N iterations + let mut spin_count: usize = 0; + let mut iteration: usize = 0; + + while scheduler.is_running.load(Ordering::Acquire) { + iteration = iteration.wrapping_add(1); + + // Priority 1: Local queue + let mut actor = worker.local.pop(); + + // Priority 2: Periodically check injector for new actors + // This ensures newly spawned actors get picked up even when workers have local work + if actor.is_none() || (iteration % INJECTOR_CHECK_INTERVAL == 0) { + if let Steal::Success(a) = scheduler.injector.steal_batch_and_pop(&worker.local) { + if actor.is_some() { + // We already had an actor, push the stolen one to local + worker.local.push(a); + } else { + actor = Some(a); + } + } + } + + // Priority 3: Steal from peer workers when idle + if actor.is_none() { + for (i, stealer) in scheduler.stealers.iter().enumerate() { + if i == worker.id { + continue; + } + if let Steal::Success(a) = stealer.steal_batch_and_pop(&worker.local) { + actor = Some(a); + break; + } + } + } + + // Process or backoff + match actor { + Some(mut actor) => { + spin_count = 0; + for _ in 0..TICKS_PER_ACTOR { + actor.tick(&ctx); + } + worker.local.push(actor); + } + None => { + // Exponential backoff + spin_count = spin_count.saturating_add(1); + if spin_count < 10 { + std::hint::spin_loop(); + } else if spin_count < 100 { + thread::yield_now(); + } else { + thread::sleep(Duration::from_micros(10)); + } + } + } } } diff --git a/src/worker.rs b/src/worker.rs deleted file mode 100644 index 4a0efd5..0000000 --- a/src/worker.rs +++ /dev/null @@ -1,46 +0,0 @@ -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>, - arena: LocalArena, - actor_table: Vec, -} - -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>, // Or flat Vec with head/tail if arena-allocated - actor_order: Vec, // Which actors have messages (for iteration) -}