use crate::Router;
use std::cell::RefCell;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
static WORKER_ID_COUNTER: AtomicUsize = AtomicUsize::new(0);
#[inline]
pub fn next_worker_id() -> usize {
WORKER_ID_COUNTER.fetch_add(1, Ordering::Relaxed)
}
#[inline]
pub fn total_workers() -> usize {
WORKER_ID_COUNTER.load(Ordering::Relaxed)
}
thread_local! {
static WORKER_ROUTER: RefCell<Option<Arc<Router>>> = const { RefCell::new(None) };
static WORKER_ID: RefCell<Option<usize>> = const { RefCell::new(None) };
}
#[inline]
pub fn init_worker_router(router: Arc<Router>) {
WORKER_ROUTER.with(|r| {
*r.borrow_mut() = Some(router);
});
WORKER_ID.with(|id| {
if id.borrow().is_none() {
*id.borrow_mut() = Some(next_worker_id());
}
});
WORKER_STATS.record_init();
}
#[inline]
pub fn clear_worker_router() {
WORKER_ROUTER.with(|r| {
*r.borrow_mut() = None;
});
}
#[inline]
pub fn worker_id() -> Option<usize> {
WORKER_ID.with(|id| *id.borrow())
}
#[inline]
pub fn has_worker_router() -> bool {
WORKER_ROUTER.with(|r| r.borrow().is_some())
}
pub struct WorkerRouter;
impl WorkerRouter {
#[inline]
pub fn with<F, R>(f: F) -> R
where
F: FnOnce(&Router) -> R,
{
WORKER_ROUTER.with(|r| {
let router_ref = r.borrow();
let router = router_ref
.as_ref()
.expect("WorkerRouter not initialized. Call init_worker_router first.");
WORKER_STATS.record_access();
f(router)
})
}
#[inline]
pub fn try_with<F, R>(f: F) -> Option<R>
where
F: FnOnce(&Router) -> R,
{
WORKER_ROUTER.with(|r| {
let router_ref = r.borrow();
router_ref.as_ref().map(|router| {
WORKER_STATS.record_access();
f(router)
})
})
}
#[inline]
pub fn clone_arc() -> Option<Arc<Router>> {
WORKER_ROUTER.with(|r| {
let router_ref = r.borrow();
router_ref.as_ref().map(|router| {
WORKER_STATS.record_clone();
Arc::clone(router)
})
})
}
#[inline]
pub fn clone_arc_or_panic() -> Arc<Router> {
Self::clone_arc().expect("WorkerRouter not initialized")
}
}
#[derive(Debug, Clone)]
pub struct WorkerConfig {
pub num_workers: usize,
pub cpu_affinity: bool,
pub stack_size: Option<usize>,
pub name_prefix: String,
}
impl Default for WorkerConfig {
fn default() -> Self {
Self {
num_workers: 0, cpu_affinity: false,
stack_size: None,
name_prefix: "armature-worker".to_string(),
}
}
}
impl WorkerConfig {
pub fn new() -> Self {
Self::default()
}
#[inline]
pub fn workers(mut self, n: usize) -> Self {
self.num_workers = n;
self
}
#[inline]
pub fn with_cpu_affinity(mut self) -> Self {
self.cpu_affinity = true;
self
}
#[inline]
pub fn stack_size(mut self, size: usize) -> Self {
self.stack_size = Some(size);
self
}
#[inline]
pub fn name_prefix(mut self, prefix: impl Into<String>) -> Self {
self.name_prefix = prefix.into();
self
}
#[inline]
pub fn effective_workers(&self) -> usize {
if self.num_workers > 0 {
self.num_workers
} else {
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
}
}
}
#[derive(Debug, Clone)]
pub struct AffinityConfig {
pub enabled: bool,
pub cores: Vec<usize>,
pub mode: AffinityMode,
}
impl Default for AffinityConfig {
fn default() -> Self {
Self {
enabled: false,
cores: Vec::new(),
mode: AffinityMode::RoundRobin,
}
}
}
impl AffinityConfig {
pub fn new() -> Self {
Self::default()
}
#[inline]
pub fn enable(mut self) -> Self {
self.enabled = true;
self
}
#[inline]
pub fn disable(mut self) -> Self {
self.enabled = false;
self
}
#[inline]
pub fn cores(mut self, cores: Vec<usize>) -> Self {
self.cores = cores;
self
}
#[inline]
pub fn mode(mut self, mode: AffinityMode) -> Self {
self.mode = mode;
self
}
#[inline]
pub fn core_for_worker(&self, worker_id: usize) -> usize {
if self.cores.is_empty() {
let num_cores = num_cpus();
match self.mode {
AffinityMode::RoundRobin => worker_id % num_cores,
AffinityMode::Packed => worker_id.min(num_cores - 1),
AffinityMode::Spread => {
let stride = num_cores / 2;
(worker_id * stride.max(1)) % num_cores
}
}
} else {
self.cores[worker_id % self.cores.len()]
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AffinityMode {
RoundRobin,
Packed,
Spread,
}
#[inline]
pub fn num_cpus() -> usize {
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
}
#[inline]
pub fn num_physical_cpus() -> usize {
let total = num_cpus();
if total > 4 && total.is_multiple_of(2) {
total / 2
} else {
total
}
}
#[inline]
pub fn set_thread_affinity(core: usize) -> Result<(), AffinityError> {
#[cfg(target_os = "linux")]
{
set_thread_affinity_linux(core)
}
#[cfg(not(target_os = "linux"))]
{
let _ = core;
Ok(())
}
}
#[cfg(target_os = "linux")]
fn set_thread_affinity_linux(core: usize) -> Result<(), AffinityError> {
use std::mem;
let num_cores = num_cpus();
if core >= num_cores {
return Err(AffinityError::InvalidCore {
core,
max: num_cores - 1,
});
}
let mut mask: u64 = 0;
mask |= 1u64 << core;
unsafe {
let result = libc::sched_setaffinity(
0, mem::size_of::<u64>(),
&mask as *const u64 as *const libc::cpu_set_t,
);
if result == 0 {
AFFINITY_STATS.record_set(true);
Ok(())
} else {
AFFINITY_STATS.record_set(false);
Err(AffinityError::SystemError {
errno: *libc::__errno_location(),
})
}
}
}
#[derive(Debug, Clone)]
pub enum AffinityError {
InvalidCore { core: usize, max: usize },
SystemError { errno: i32 },
NotSupported,
}
impl std::fmt::Display for AffinityError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::InvalidCore { core, max } => {
write!(f, "Invalid core {}, max is {}", core, max)
}
Self::SystemError { errno } => {
write!(f, "System error: errno {}", errno)
}
Self::NotSupported => write!(f, "CPU affinity not supported on this platform"),
}
}
}
impl std::error::Error for AffinityError {}
#[inline]
pub fn get_thread_affinity() -> Result<Vec<usize>, AffinityError> {
#[cfg(target_os = "linux")]
{
get_thread_affinity_linux()
}
#[cfg(not(target_os = "linux"))]
{
Ok((0..num_cpus()).collect())
}
}
#[cfg(target_os = "linux")]
fn get_thread_affinity_linux() -> Result<Vec<usize>, AffinityError> {
use std::mem;
let mut mask: u64 = 0;
unsafe {
let result = libc::sched_getaffinity(
0,
mem::size_of::<u64>(),
&mut mask as *mut u64 as *mut libc::cpu_set_t,
);
if result == 0 {
let mut cores = Vec::new();
for i in 0..64 {
if (mask & (1u64 << i)) != 0 {
cores.push(i);
}
}
Ok(cores)
} else {
Err(AffinityError::SystemError {
errno: *libc::__errno_location(),
})
}
}
}
#[inline]
pub fn affinity_supported() -> bool {
cfg!(target_os = "linux")
}
#[inline]
pub fn init_worker_with_affinity(
worker_id: usize,
config: &AffinityConfig,
router: Arc<Router>,
) -> Result<(), AffinityError> {
if config.enabled && affinity_supported() {
let core = config.core_for_worker(worker_id);
set_thread_affinity(core)?;
}
init_worker_router(router);
Ok(())
}
#[derive(Debug, Default)]
pub struct AffinityStats {
successful: AtomicU64,
failed: AtomicU64,
}
impl AffinityStats {
pub fn new() -> Self {
Self::default()
}
#[inline]
fn record_set(&self, success: bool) {
if success {
self.successful.fetch_add(1, Ordering::Relaxed);
} else {
self.failed.fetch_add(1, Ordering::Relaxed);
}
}
pub fn successful(&self) -> u64 {
self.successful.load(Ordering::Relaxed)
}
pub fn failed(&self) -> u64 {
self.failed.load(Ordering::Relaxed)
}
pub fn success_rate(&self) -> f64 {
let total = self.successful() + self.failed();
if total > 0 {
(self.successful() as f64 / total as f64) * 100.0
} else {
0.0
}
}
}
static AFFINITY_STATS: AffinityStats = AffinityStats {
successful: AtomicU64::new(0),
failed: AtomicU64::new(0),
};
pub fn affinity_stats() -> &'static AffinityStats {
&AFFINITY_STATS
}
#[derive(Debug, Default)]
pub struct WorkerStats {
inits: AtomicU64,
accesses: AtomicU64,
clones: AtomicU64,
}
impl WorkerStats {
pub fn new() -> Self {
Self::default()
}
#[inline]
fn record_init(&self) {
self.inits.fetch_add(1, Ordering::Relaxed);
}
#[inline]
fn record_access(&self) {
self.accesses.fetch_add(1, Ordering::Relaxed);
}
#[inline]
fn record_clone(&self) {
self.clones.fetch_add(1, Ordering::Relaxed);
}
pub fn inits(&self) -> u64 {
self.inits.load(Ordering::Relaxed)
}
pub fn accesses(&self) -> u64 {
self.accesses.load(Ordering::Relaxed)
}
pub fn clones(&self) -> u64 {
self.clones.load(Ordering::Relaxed)
}
pub fn clone_avoidance_ratio(&self) -> f64 {
let accesses = self.accesses() as f64;
let clones = self.clones() as f64;
if accesses > 0.0 {
((accesses - clones) / accesses) * 100.0
} else {
0.0
}
}
}
static WORKER_STATS: WorkerStats = WorkerStats {
inits: AtomicU64::new(0),
accesses: AtomicU64::new(0),
clones: AtomicU64::new(0),
};
pub fn worker_stats() -> &'static WorkerStats {
&WORKER_STATS
}
#[derive(Debug, Clone)]
pub struct WorkerHandle {
pub id: usize,
pub name: String,
}
impl WorkerHandle {
pub fn new(id: usize, name_prefix: &str) -> Self {
Self {
id,
name: format!("{}-{}", name_prefix, id),
}
}
}
pub struct WorkerState<T: 'static> {
_marker: std::marker::PhantomData<T>,
}
thread_local! {
static WORKER_STATE: RefCell<WorkerStateStorage> = RefCell::new(WorkerStateStorage::new());
}
#[derive(Default)]
struct WorkerStateStorage {
data: std::collections::HashMap<std::any::TypeId, Box<dyn std::any::Any + Send>>,
}
impl WorkerStateStorage {
fn new() -> Self {
Self {
data: std::collections::HashMap::new(),
}
}
fn insert<T: 'static + Send>(&mut self, value: T) {
let type_id = std::any::TypeId::of::<T>();
self.data.insert(type_id, Box::new(value));
}
fn get<T: 'static>(&self) -> Option<&T> {
let type_id = std::any::TypeId::of::<T>();
self.data.get(&type_id).and_then(|b| b.downcast_ref::<T>())
}
fn get_mut<T: 'static>(&mut self) -> Option<&mut T> {
let type_id = std::any::TypeId::of::<T>();
self.data
.get_mut(&type_id)
.and_then(|b| b.downcast_mut::<T>())
}
fn remove<T: 'static>(&mut self) -> Option<T> {
let type_id = std::any::TypeId::of::<T>();
self.data
.remove(&type_id)
.and_then(|b| b.downcast::<T>().ok().map(|b| *b))
}
fn contains<T: 'static>(&self) -> bool {
let type_id = std::any::TypeId::of::<T>();
self.data.contains_key(&type_id)
}
fn clear(&mut self) {
self.data.clear();
}
}
impl<T: 'static + Send> WorkerState<T> {
#[inline]
pub fn init(value: T) {
WORKER_STATE.with(|storage| {
storage.borrow_mut().insert(value);
});
WORKER_STATE_STATS.record_init();
}
#[inline]
pub fn with<F, R>(f: F) -> R
where
F: FnOnce(&T) -> R,
{
WORKER_STATE.with(|storage| {
let storage_ref = storage.borrow();
let state = storage_ref
.get::<T>()
.expect("WorkerState not initialized for this type");
WORKER_STATE_STATS.record_access();
f(state)
})
}
#[inline]
pub fn with_mut<F, R>(f: F) -> R
where
F: FnOnce(&mut T) -> R,
{
WORKER_STATE.with(|storage| {
let mut storage_ref = storage.borrow_mut();
let state = storage_ref
.get_mut::<T>()
.expect("WorkerState not initialized for this type");
WORKER_STATE_STATS.record_access();
f(state)
})
}
#[inline]
pub fn try_with<F, R>(f: F) -> Option<R>
where
F: FnOnce(&T) -> R,
{
WORKER_STATE.with(|storage| {
let storage_ref = storage.borrow();
storage_ref.get::<T>().map(|state| {
WORKER_STATE_STATS.record_access();
f(state)
})
})
}
#[inline]
pub fn try_with_mut<F, R>(f: F) -> Option<R>
where
F: FnOnce(&mut T) -> R,
{
WORKER_STATE.with(|storage| {
let mut storage_ref = storage.borrow_mut();
storage_ref.get_mut::<T>().map(|state| {
WORKER_STATE_STATS.record_access();
f(state)
})
})
}
#[inline]
pub fn is_initialized() -> bool {
WORKER_STATE.with(|storage| storage.borrow().contains::<T>())
}
#[inline]
pub fn take() -> Option<T> {
WORKER_STATE.with(|storage| storage.borrow_mut().remove::<T>())
}
#[inline]
pub fn replace(value: T) -> Option<T> {
let old = Self::take();
Self::init(value);
old
}
}
#[inline]
pub fn init_worker_state<T: 'static + Send>(value: T) {
WorkerState::<T>::init(value);
}
pub fn clear_worker_state() {
WORKER_STATE.with(|storage| {
storage.borrow_mut().clear();
});
}
#[derive(Debug, Default)]
pub struct WorkerStateStats {
inits: AtomicU64,
accesses: AtomicU64,
}
impl WorkerStateStats {
pub fn new() -> Self {
Self::default()
}
#[inline]
fn record_init(&self) {
self.inits.fetch_add(1, Ordering::Relaxed);
}
#[inline]
fn record_access(&self) {
self.accesses.fetch_add(1, Ordering::Relaxed);
}
pub fn inits(&self) -> u64 {
self.inits.load(Ordering::Relaxed)
}
pub fn accesses(&self) -> u64 {
self.accesses.load(Ordering::Relaxed)
}
}
static WORKER_STATE_STATS: WorkerStateStats = WorkerStateStats {
inits: AtomicU64::new(0),
accesses: AtomicU64::new(0),
};
pub fn worker_state_stats() -> &'static WorkerStateStats {
&WORKER_STATE_STATS
}
pub struct StateFactory<T: Clone + Send + 'static> {
state: Arc<T>,
}
impl<T: Clone + Send + 'static> StateFactory<T> {
pub fn new(state: T) -> Self {
Self {
state: Arc::new(state),
}
}
pub fn from_arc(state: Arc<T>) -> Self {
Self { state }
}
pub fn init_for_worker(&self) {
let cloned = (*self.state).clone();
WorkerState::<T>::init(cloned);
}
pub fn shared(&self) -> &T {
&self.state
}
pub fn arc(&self) -> Arc<T> {
Arc::clone(&self.state)
}
}
impl<T: Clone + Send + 'static> Clone for StateFactory<T> {
fn clone(&self) -> Self {
Self {
state: Arc::clone(&self.state),
}
}
}
#[derive(Debug)]
pub struct WorkerCache<K, V>
where
K: std::hash::Hash + Eq + Clone,
{
data: std::collections::HashMap<K, V>,
max_entries: usize,
hits: u64,
misses: u64,
}
impl<K, V> WorkerCache<K, V>
where
K: std::hash::Hash + Eq + Clone,
{
pub fn new(max_entries: usize) -> Self {
Self {
data: std::collections::HashMap::with_capacity(max_entries),
max_entries,
hits: 0,
misses: 0,
}
}
pub fn get(&mut self, key: &K) -> Option<&V> {
if self.data.contains_key(key) {
self.hits += 1;
self.data.get(key)
} else {
self.misses += 1;
None
}
}
pub fn get_mut(&mut self, key: &K) -> Option<&mut V> {
if self.data.contains_key(key) {
self.hits += 1;
self.data.get_mut(key)
} else {
self.misses += 1;
None
}
}
pub fn insert(&mut self, key: K, value: V) -> Option<V> {
if self.data.len() >= self.max_entries
&& !self.data.contains_key(&key)
&& let Some(first_key) = self.data.keys().next().cloned()
{
self.data.remove(&first_key);
}
self.data.insert(key, value)
}
pub fn remove(&mut self, key: &K) -> Option<V> {
self.data.remove(key)
}
pub fn contains(&self, key: &K) -> bool {
self.data.contains_key(key)
}
pub fn clear(&mut self) {
self.data.clear();
self.hits = 0;
self.misses = 0;
}
pub fn len(&self) -> usize {
self.data.len()
}
pub fn is_empty(&self) -> bool {
self.data.is_empty()
}
pub fn hits(&self) -> u64 {
self.hits
}
pub fn misses(&self) -> u64 {
self.misses
}
pub fn hit_ratio(&self) -> f64 {
let total = self.hits + self.misses;
if total > 0 {
(self.hits as f64 / total as f64) * 100.0
} else {
0.0
}
}
}
impl<K, V> Default for WorkerCache<K, V>
where
K: std::hash::Hash + Eq + Clone,
{
fn default() -> Self {
Self::new(1000)
}
}
#[macro_export]
macro_rules! with_worker_router {
($router:ident, $body:block) => {{ $crate::worker::WorkerRouter::with(|$router| $body) }};
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_worker_id_generation() {
let id1 = next_worker_id();
let id2 = next_worker_id();
assert!(id2 > id1);
}
#[test]
fn test_worker_config_default() {
let config = WorkerConfig::default();
assert_eq!(config.num_workers, 0);
assert!(!config.cpu_affinity);
}
#[test]
fn test_worker_config_builder() {
let config = WorkerConfig::new()
.workers(4)
.with_cpu_affinity()
.name_prefix("test-worker");
assert_eq!(config.num_workers, 4);
assert!(config.cpu_affinity);
assert_eq!(config.name_prefix, "test-worker");
}
#[test]
fn test_effective_workers() {
let config = WorkerConfig::new().workers(8);
assert_eq!(config.effective_workers(), 8);
let auto_config = WorkerConfig::new();
assert!(auto_config.effective_workers() >= 1);
}
#[test]
fn test_affinity_config_default() {
let config = AffinityConfig::default();
assert!(!config.enabled);
assert!(config.cores.is_empty());
assert_eq!(config.mode, AffinityMode::RoundRobin);
}
#[test]
fn test_affinity_config_builder() {
let config = AffinityConfig::new()
.enable()
.cores(vec![0, 2, 4])
.mode(AffinityMode::Spread);
assert!(config.enabled);
assert_eq!(config.cores, vec![0, 2, 4]);
assert_eq!(config.mode, AffinityMode::Spread);
}
#[test]
fn test_core_for_worker_round_robin() {
let config = AffinityConfig::new()
.enable()
.mode(AffinityMode::RoundRobin);
let num_cores = num_cpus();
assert_eq!(config.core_for_worker(0), 0);
assert_eq!(config.core_for_worker(1), 1 % num_cores);
assert_eq!(config.core_for_worker(num_cores), 0);
}
#[test]
fn test_core_for_worker_specific_cores() {
let config = AffinityConfig::new().enable().cores(vec![0, 4, 8]);
assert_eq!(config.core_for_worker(0), 0);
assert_eq!(config.core_for_worker(1), 4);
assert_eq!(config.core_for_worker(2), 8);
assert_eq!(config.core_for_worker(3), 0); }
#[test]
fn test_num_cpus() {
let cpus = num_cpus();
assert!(cpus >= 1);
}
#[test]
fn test_num_physical_cpus() {
let physical = num_physical_cpus();
let total = num_cpus();
assert!(physical >= 1);
assert!(physical <= total);
}
#[test]
fn test_affinity_supported() {
let _ = affinity_supported();
}
#[test]
fn test_get_thread_affinity() {
let result = get_thread_affinity();
assert!(result.is_ok());
let cores = result.unwrap();
assert!(!cores.is_empty());
}
#[test]
fn test_affinity_stats() {
let stats = affinity_stats();
let _ = stats.successful();
let _ = stats.failed();
let _ = stats.success_rate();
}
#[test]
fn test_affinity_error_display() {
let err1 = AffinityError::InvalidCore { core: 100, max: 7 };
assert!(err1.to_string().contains("100"));
let err2 = AffinityError::NotSupported;
assert!(err2.to_string().contains("not supported"));
}
#[test]
fn test_worker_router_not_initialized() {
clear_worker_router();
assert!(!has_worker_router());
assert!(WorkerRouter::try_with(|_| ()).is_none());
assert!(WorkerRouter::clone_arc().is_none());
}
#[test]
fn test_worker_state_basic() {
clear_worker_state();
WorkerState::<u64>::init(42);
let value = WorkerState::<u64>::with(|v| *v);
assert_eq!(value, 42);
WorkerState::<u64>::with_mut(|v| *v += 1);
let value = WorkerState::<u64>::with(|v| *v);
assert_eq!(value, 43);
clear_worker_state();
}
#[test]
fn test_worker_state_multiple_types() {
clear_worker_state();
WorkerState::<u64>::init(100);
WorkerState::<String>::init("hello".to_string());
assert_eq!(WorkerState::<u64>::with(|v| *v), 100);
assert_eq!(WorkerState::<String>::with(|v| v.clone()), "hello");
clear_worker_state();
}
#[test]
fn test_worker_state_try_with() {
clear_worker_state();
assert!(WorkerState::<i32>::try_with(|_| ()).is_none());
WorkerState::<i32>::init(123);
assert!(WorkerState::<i32>::try_with(|v| *v).is_some());
assert_eq!(WorkerState::<i32>::try_with(|v| *v), Some(123));
clear_worker_state();
}
#[test]
fn test_worker_state_take() {
clear_worker_state();
WorkerState::<String>::init("test".to_string());
assert!(WorkerState::<String>::is_initialized());
let taken = WorkerState::<String>::take();
assert_eq!(taken, Some("test".to_string()));
assert!(!WorkerState::<String>::is_initialized());
clear_worker_state();
}
#[test]
fn test_worker_state_replace() {
clear_worker_state();
WorkerState::<u32>::init(10);
let old = WorkerState::<u32>::replace(20);
assert_eq!(old, Some(10));
assert_eq!(WorkerState::<u32>::with(|v| *v), 20);
clear_worker_state();
}
#[test]
fn test_worker_cache_basic() {
let mut cache = WorkerCache::<String, u32>::new(10);
cache.insert("key1".to_string(), 100);
cache.insert("key2".to_string(), 200);
assert_eq!(cache.get(&"key1".to_string()), Some(&100));
assert_eq!(cache.get(&"key3".to_string()), None);
assert_eq!(cache.len(), 2);
}
#[test]
fn test_worker_cache_eviction() {
let mut cache = WorkerCache::<u32, u32>::new(3);
cache.insert(1, 100);
cache.insert(2, 200);
cache.insert(3, 300);
assert_eq!(cache.len(), 3);
cache.insert(4, 400);
assert_eq!(cache.len(), 3);
assert!(cache.contains(&4));
}
#[test]
fn test_worker_cache_hit_ratio() {
let mut cache = WorkerCache::<u32, u32>::new(10);
cache.insert(1, 100);
cache.get(&1); cache.get(&1); cache.get(&2);
assert_eq!(cache.hits(), 2);
assert_eq!(cache.misses(), 1);
assert!((cache.hit_ratio() - 66.67).abs() < 1.0);
}
#[test]
fn test_state_factory() {
clear_worker_state();
let factory = StateFactory::new(vec![1, 2, 3]);
factory.init_for_worker();
WorkerState::<Vec<i32>>::with(|v| {
assert_eq!(v, &vec![1, 2, 3]);
});
clear_worker_state();
}
#[test]
fn test_worker_state_stats() {
let stats = worker_state_stats();
let _ = stats.inits();
let _ = stats.accesses();
}
#[test]
fn test_worker_router_initialization() {
let router = Arc::new(Router::new());
init_worker_router(router);
assert!(has_worker_router());
assert!(worker_id().is_some());
WorkerRouter::with(|r| {
assert!(r.routes.is_empty());
});
clear_worker_router();
}
#[test]
fn test_worker_handle() {
let handle = WorkerHandle::new(5, "test-worker");
assert_eq!(handle.id, 5);
assert_eq!(handle.name, "test-worker-5");
}
#[test]
fn test_worker_stats() {
let stats = worker_stats();
let _ = stats.inits();
let _ = stats.accesses();
let _ = stats.clones();
let _ = stats.clone_avoidance_ratio();
}
}