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

refactor: remove 'waterlevel'

Browse source
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-02-09 14:15:13 +07:00
parent 3d2abae686
commit fe500e9e69
9 changed files with 44 additions and 175 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
benches/worker_benchmarks.rs
View file

@ -3,32 +3,6 @@ use std::collections::VecDeque;
use criterion::{
criterion_group, criterion_main, BenchmarkId, Criterion, Throughput,
};
use swactor::worker::drain_count;
// ---------------------------------------------------------------------------
// drain_count O(1) verification
// ---------------------------------------------------------------------------
fn bench_drain_count(c: &mut Criterion) {
let mut group = c.benchmark_group("drain_count");
// Below waterlevel
group.bench_function("below", |b| {
b.iter(|| std::hint::black_box(drain_count(50, 100)));
});
// At waterlevel
group.bench_function("at", |b| {
b.iter(|| std::hint::black_box(drain_count(100, 100)));
});
// Above waterlevel
group.bench_function("above", |b| {
b.iter(|| std::hint::black_box(drain_count(500, 100)));
});
group.finish();
}
// ---------------------------------------------------------------------------
// VecDeque push throughput (mirrors old mailbox_push)
@ -79,43 +53,9 @@ fn vecdeque_pop(c: &mut Criterion) {
group.finish();
}
// ---------------------------------------------------------------------------
// Simulated actor tick: drain_count + pop N from VecDeque
// ---------------------------------------------------------------------------
fn simulated_actor_tick(c: &mut Criterion) {
let mut group = c.benchmark_group("simulated_actor_tick");
for (wl, fill) in [(10, 5), (10, 10), (10, 50), (100, 200)] {
let param = format!("wl={wl},fill={fill}");
group.bench_function(BenchmarkId::from_parameter(&param), |b| {
b.iter_batched(
|| {
let mut q: VecDeque<u64> = VecDeque::new();
for i in 0..fill {
q.push_back(i as u64);
}
q
},
|mut q| {
let n = drain_count(q.len(), wl);
for _ in 0..n {
std::hint::black_box(q.pop_front());
}
},
criterion::BatchSize::SmallInput,
);
});
}
group.finish();
}
criterion_group!(
benches,
bench_drain_count,
vecdeque_push,
vecdeque_pop,
simulated_actor_tick,
);
criterion_main!(benches);

6
crates/swactor-python/src/lib.rs
View file

@ -227,8 +227,6 @@ pub struct PyRuntimeConfig {
#[pyo3(get, set)]
actor_max_messages: usize,
#[pyo3(get, set)]
mailbox_waterlevel: usize,
#[pyo3(get, set)]
spin_threshold: u32,
#[pyo3(get, set)]
yield_threshold: u32,
@ -246,7 +244,6 @@ impl PyRuntimeConfig {
num_threads = 1,
max_actors = 1_000,
actor_max_messages = 1_000,
mailbox_waterlevel = 10,
spin_threshold = 64,
yield_threshold = 256,
sleep_increment_us = 50,
@ -256,7 +253,6 @@ impl PyRuntimeConfig {
num_threads: usize,
max_actors: usize,
actor_max_messages: usize,
mailbox_waterlevel: usize,
spin_threshold: u32,
yield_threshold: u32,
sleep_increment_us: u64,
@ -266,7 +262,6 @@ impl PyRuntimeConfig {
num_threads,
max_actors,
actor_max_messages,
mailbox_waterlevel,
spin_threshold,
yield_threshold,
sleep_increment_us,
@ -281,7 +276,6 @@ impl From<PyRuntimeConfig> for RuntimeConfig {
num_threads: py.num_threads,
max_actors: py.max_actors,
actor_max_messages: py.actor_max_messages,
mailbox_waterlevel: py.mailbox_waterlevel,
backoff_policy: BackoffPolicy {
spin_threshold: py.spin_threshold,
yield_threshold: py.yield_threshold,

5
fuzz/fuzz_targets/fuzz_runtime.rs
View file

@ -210,7 +210,6 @@ enum RawAction {
#[derive(Debug, Arbitrary)]
struct FuzzInput {
max_actors: u8,
mailbox_waterlevel: u8,
scenarios: Vec<Scenario>,
}
@ -773,7 +772,6 @@ impl fmt::Debug for FuzzState {
fuzz_target!(|input: FuzzInput| {
let max_actors = (input.max_actors as usize).max(1).min(200);
let mailbox_waterlevel = (input.mailbox_waterlevel as usize).max(1).min(50);
let interval = log_interval();
let run = if interval > 0 {
@ -783,7 +781,6 @@ fuzz_target!(|input: FuzzInput| {
let config = RuntimeConfig {
max_actors,
mailbox_waterlevel,
num_threads: 1,
..Default::default()
};
@ -806,7 +803,7 @@ fuzz_target!(|input: FuzzInput| {
let depth: usize = stats.workers.iter().map(|w| w.mailbox_depth).sum();
let processed: u64 = stats.workers.iter().map(|w| w.messages_processed).sum();
eprintln!("\
\n=== Run #{run} | cap={max_actors} waterlevel={mailbox_waterlevel} ===
\n=== Run #{run} | cap={max_actors} ===
{trace}\
--- {spawned} spawned, {sent} sent, {recv} received, \
{alive} alive, {depth} queued, {processed} processed ---\n",

1
src/actor.rs
View file

@ -64,7 +64,6 @@ where
pub trait ContextInner {
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error>;
fn mailbox_waterlevel(&self) -> usize;
}
/// Actor syscall interface — passed to `ActorInterface::handle()`.

5
src/config.rs
View file

@ -28,7 +28,6 @@ pub struct RuntimeConfig {
pub max_actors: usize,
pub actor_max_messages: usize,
pub num_threads: usize,
pub mailbox_waterlevel: usize,
pub backoff_policy: BackoffPolicy,
}
@ -40,16 +39,12 @@ const DEFAULT_MAX_ACTORS: usize = 1_000;
/// 1_000 * 16kB = 16MB
const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
const DEFAULT_MAILBOX_WATERLEVEL: usize = 10;
impl Default for RuntimeConfig {
fn default() -> Self {
Self {
max_actors: DEFAULT_MAX_ACTORS,
actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES,
num_threads: 1,
// Clamp to minimum 2: drain_count(1,1)==0 causes livelock
mailbox_waterlevel: DEFAULT_MAILBOX_WATERLEVEL.max(2),
backoff_policy: BackoffPolicy::default(),
}
}

19
src/crate_test/mod.rs
View file

@ -253,13 +253,8 @@ fn wrong_type_silently_dropped() {
}
#[test]
fn backpressure_drains_half() {
let config = RuntimeConfig {
mailbox_waterlevel: 4,
..Default::default()
};
run_with(config, &[
fn all_messages_drain_in_one_tick() {
run(&[
Step::Spawn(1),
Step::Tick,
@ -268,15 +263,7 @@ fn backpressure_drains_half() {
Step::Send(1, 4), Step::Send(1, 5), Step::Send(1, 6), Step::Send(1, 7),
Step::Send(1, 8), Step::Send(1, 9),
// Tick 1: 10 in mailbox, drain_count(10, 4) = 5
Step::Tick,
Step::Expect { pool_len: 1, depth: 5, processed: 5 },
// Tick 2: 5 remaining, drain_count(5, 4) = 2
Step::Tick,
Step::Expect { pool_len: 1, depth: 3, processed: 7 },
// Tick 3: 3 remaining, drain_count(3, 4) = 3 (below waterlevel → all)
// All 10 processed in a single tick
Step::Tick,
Step::Expect { pool_len: 1, depth: 0, processed: 10 },
Step::ExpectHandled(1, 10),

91
src/runtime.rs
View file

@ -66,15 +66,12 @@ pub struct Runtime {
placement: Placement,
is_running: AtomicBool,
worker_stats: Vec<Arc<WorkerStats>>,
/// Single-threaded mode: worker stored inline
single_worker: Option<RefCell<Worker>>,
/// Multi-threaded mode: workers waiting to be assigned to threads by run()
pending_workers: Option<Vec<Worker>>,
/// Workers available for tick(). run() drains this and moves workers to threads.
tick_workers: RefCell<Vec<Worker>>,
}
// Safety: RefCell<Worker> is only accessed from the thread that owns the Runtime
// in single-threaded mode. In multi-threaded mode, single_worker is None and
// pending_workers is consumed by run() before Arc sharing.
// Safety: RefCell<Vec<Worker>> is only accessed from the owning thread via tick().
// After run() the RefCell is empty and not accessed by worker threads.
unsafe impl Sync for Runtime {}
impl Runtime {
@ -111,35 +108,16 @@ impl Runtime {
workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats));
}
if config.num_threads < 2 {
// Single-threaded: store one worker inline
let worker = workers.remove(0);
Self {
config,
address_map,
inbox_registry,
transfer_txs,
spawn_txs,
placement,
is_running: AtomicBool::new(false),
worker_stats,
single_worker: Some(RefCell::new(worker)),
pending_workers: None,
}
} else {
// Multi-threaded: stash workers for run()
Self {
config,
address_map,
inbox_registry,
transfer_txs,
spawn_txs,
placement,
is_running: AtomicBool::new(false),
worker_stats,
single_worker: None,
pending_workers: Some(workers),
}
Self {
config,
address_map,
inbox_registry,
transfer_txs,
spawn_txs,
placement,
is_running: AtomicBool::new(false),
worker_stats,
tick_workers: RefCell::new(workers),
}
}
@ -179,17 +157,23 @@ impl Runtime {
}
/// Drive one tick of the single-threaded worker.
///
/// Panics if called on a multi-threaded runtime — use `run()` instead.
pub fn tick(&self) {
if let Some(ref worker) = self.single_worker {
let tc = TickContext {
address_map: &self.address_map,
transfer_txs: &self.transfer_txs,
spawn_txs: &self.spawn_txs,
placement: &self.placement,
inbox_registry: &self.inbox_registry,
config: &self.config,
};
worker.borrow_mut().tick_once(&tc);
assert!(
self.config.num_threads < 2,
"tick() is only valid for single-threaded runtimes; use run() for multi-threaded"
);
let tc = TickContext {
address_map: &self.address_map,
transfer_txs: &self.transfer_txs,
spawn_txs: &self.spawn_txs,
placement: &self.placement,
inbox_registry: &self.inbox_registry,
config: &self.config,
};
for worker in self.tick_workers.borrow_mut().iter_mut() {
worker.tick_once(&tc);
}
}
@ -198,17 +182,10 @@ impl Runtime {
///
/// Works in both single-threaded and multi-threaded configurations.
/// In single-threaded mode, one background thread is spawned.
pub fn run(mut self) -> Result<RuntimeHandle, Error> {
pub fn run(self) -> Result<RuntimeHandle, Error> {
self.is_running.store(true, Ordering::Release);
let mut workers: Vec<Worker> = Vec::new();
if let Some(w) = self.single_worker.take() {
workers.push(w.into_inner());
}
if let Some(ws) = self.pending_workers.take() {
workers.extend(ws);
}
let workers: Vec<Worker> = self.tick_workers.replace(Vec::new());
let rt = Arc::new(self);
let mut handles: Vec<JoinHandle<()>> = Vec::with_capacity(workers.len());
@ -294,8 +271,4 @@ impl ContextInner for Runtime {
.try_send((addr, actor))
.map_err(|_| Error::from("Spawn queue full"))
}
fn mailbox_waterlevel(&self) -> usize {
self.config.mailbox_waterlevel
}
}

31
src/worker.rs
View file

@ -159,20 +159,6 @@ impl ContextInner for WorkerContext<'_> {
.map_err(|_| Error::from("Spawn queue full"))
}
fn mailbox_waterlevel(&self) -> usize {
self.tc.config.mailbox_waterlevel
}
}
/// How many messages to process this tick:
/// - `len < waterlevel` → process all (`len`)
/// - `len >= waterlevel` → process half (`len >> 1`)
pub fn drain_count(len: usize, waterlevel: usize) -> usize {
if len < waterlevel {
len
} else {
len >> 1
}
}
struct ActorSlot {
@ -218,16 +204,15 @@ impl ActorPool {
pub fn tick_all(&mut self, inner: &dyn ContextInner) -> usize {
let mut count = 0;
for (&addr, slot) in self.actors.iter_mut() {
let len = slot.mailbox.len();
let n = drain_count(len, inner.mailbox_waterlevel());
if n > 0 {
let ctx = Ctx::new(inner, addr);
for _ in 0..n {
if let Some(msg) = slot.mailbox.pop_front() {
slot.actor.handle_any(&ctx, msg);
count += 1;
}
let ctx = Ctx::new(inner, addr);
while let Some(msg) = slot.mailbox.pop_front() {
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
slot.actor.handle_any(&ctx, msg);
}));
if result.is_err() {
eprintln!("swactor: actor {addr} panicked in handler");
}
count += 1;
}
}
count

1
tests/test_python.py
View file

@ -38,7 +38,6 @@ class TestRuntimeConfig(unittest.TestCase):
self.assertEqual(cfg.num_threads, 1)
self.assertEqual(cfg.max_actors, 1000)
self.assertEqual(cfg.actor_max_messages, 1000)
self.assertEqual(cfg.mailbox_waterlevel, 10)
self.assertEqual(cfg.spin_threshold, 64)
self.assertEqual(cfg.yield_threshold, 256)
self.assertEqual(cfg.sleep_increment_us, 50)