zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: workstealing, less contention for multithreaded

Browse source
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-02-05 10:44:48 +07:00
parent ed26f0c080
commit abedaf6307
10 changed files with 180 additions and 455 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
Cargo.lock generated
View file

@ -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]]

2
Cargo.toml
View file

@ -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"

55
DESIGN.md
View file

@ -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.

113
WORKER.md
View file

@ -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<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

25
src/channel/mod.rs
View file

@ -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<T> {
ring: ArrayQueue<T>,
overflow: Mutex<VecDeque<T>>,
overflow: SegQueue<T>,
overflow_len: AtomicUsize,
}
impl<T> HybridChannel<T> {
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<T> HybridChannel<T> {
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<T> HybridChannel<T> {
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)
}
}

198
src/channel/spsc.rs
View file

@ -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<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();
}

1
src/lib.rs
View file

@ -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]) {

160
src/runtime.rs
View file

@ -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<Box<dyn AnyActor>>, // FIFO for fairness
id: usize,
}
/// Shared scheduler state for work-stealing
struct SchedulerState {
injector: Injector<Box<dyn AnyActor>>, // For spawn()
stealers: Vec<Stealer<Box<dyn AnyActor>>>, // 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<Box<dyn AnyActor>>,
actor_queue: HybridChannel<Box<dyn AnyActor>>, // Used for single-threaded mode
scheduler: Option<Arc<SchedulerState>>, // Used for multi-threaded mode
router_interface: Sender<RouterMessage>,
router: Option<Actor<Router>>, // `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")
})?;
let boxed_actor: Box<dyn AnyActor> = 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(Box::new(Actor::new(inbox, actor)))
.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<JoinHandle<()>> = 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) {
while sched.is_running.load(Ordering::Acquire) {
for _ in 0..ROUTER_BATCH_SIZE {
router.tick(&ctx);
thread::yield_now();
}
}
})
};
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<Runtime>, scheduler: Arc<SchedulerState>, 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));
}
}
}
}
}

46
src/worker.rs
View file

@ -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<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)
}