From cf107315bc8c69ac06fe7811412cc247653ebb54 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Fri, 6 Feb 2026 18:16:09 +0700 Subject: [PATCH] feat: separation of runtime and worker thread logic Logically separate the runtime and worker thread, to make it more amenable for testing, visualization, reasoning, etc. --- src/runtime.rs | 10 +++++++- src/worker.rs | 65 +++++++++++++++++++++----------------------------- 2 files changed, 36 insertions(+), 39 deletions(-) diff --git a/src/runtime.rs b/src/runtime.rs index c2d034c..d667df6 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -259,7 +259,15 @@ impl Runtime { for mut worker in workers { let rt_clone = rt.clone(); let handle = thread::spawn(move || { - worker_loop(&mut worker, &rt_clone); + let tc = TickContext { + address_map: &rt_clone.address_map, + transfer_txs: &rt_clone.transfer_txs, + spawn_txs: &rt_clone.spawn_txs, + placement: &rt_clone.placement, + inbox_registry: &rt_clone.inbox_registry, + config: &rt_clone.config, + }; + worker.run(&tc, &rt_clone.is_running, &rt_clone.config.backoff_policy); }); handles.push(handle); } diff --git a/src/worker.rs b/src/worker.rs index 18a965d..1ca7c0f 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -1,12 +1,14 @@ use std::any::Any; use std::cell::RefCell; use std::collections::{HashMap, VecDeque}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread; use crate::actor::{ActorAddress, AnyActor, Message}; use crate::address_map::{AddressMap, Placement, WorkerId}; use crate::channel::{Receiver, Sender}; -use crate::config::RuntimeConfig; -use crate::runtime::{ContextInner, Envelope, InboxRegistry, Runtime}; +use crate::config::{BackoffPolicy, RuntimeConfig}; +use crate::runtime::{ContextInner, Envelope, InboxRegistry}; use crate::Error; /// Shared state passed to tick_once — single thin pointer avoids register spill. @@ -90,6 +92,29 @@ impl Worker { did_work } + + pub(crate) fn run(&mut self, tc: &TickContext, is_running: &AtomicBool, backoff: &BackoffPolicy) { + let mut idle_count: u32 = 0; + while is_running.load(Ordering::Acquire) { + let did_work = self.tick_once(tc); + if did_work { + idle_count = 0; + } else { + idle_count = idle_count.saturating_add(1); + if idle_count < backoff.spin_threshold { + // Hot spin + } else if idle_count < backoff.yield_threshold { + thread::yield_now(); + } else { + let micros = std::cmp::min( + (idle_count - backoff.yield_threshold) as u64 * backoff.sleep_increment_us, + backoff.sleep_max_us, + ); + thread::sleep(std::time::Duration::from_micros(micros)); + } + } + } + } } /// The `ContextInner` impl for worker threads. @@ -231,39 +256,3 @@ impl Mailbox { } } -/// Worker thread loop for multi-threaded runtime. -/// Uses spin → yield → park backoff to reduce CPU usage when idle. -fn worker_loop(worker: &mut Worker, rt: &Runtime) { - let tc = TickContext { - address_map: &rt.address_map, - transfer_txs: &rt.transfer_txs, - spawn_txs: &rt.spawn_txs, - placement: &rt.placement, - inbox_registry: &rt.inbox_registry, - config: &rt.config, - }; - let bp = &rt.config.backoff_policy; - let mut idle_count: u32 = 0; - - while rt.is_running.load(Ordering::Acquire) { - let did_work = worker.tick_once(&tc); - - if did_work { - idle_count = 0; - } else { - idle_count = idle_count.saturating_add(1); - if idle_count < bp.spin_threshold { - // Hot spin — no hint, keep polling fast - } else if idle_count < bp.yield_threshold { - thread::yield_now(); - } else { - // Park: sleep briefly, cap at configured max - let micros = std::cmp::min( - (idle_count - bp.yield_threshold) as u64 * bp.sleep_increment_us, - bp.sleep_max_us, - ); - thread::sleep(std::time::Duration::from_micros(micros)); - } - } - } -} \ No newline at end of file