refactor: major library changes #5
2 changed files with 36 additions and 39 deletions
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<M: Message> Mailbox<M> {
|
|||
}
|
||||
}
|
||||
|
||||
/// 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));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue