From 61701bd7f29c013fd52054c2f4372380c99b1388 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Tue, 3 Mar 2026 00:48:03 +0700 Subject: [PATCH] feat: no_std core runtime --- Cargo.lock | 5 + Cargo.toml | 2 + crates/core/Cargo.toml | 4 + crates/core/src/lib.rs | 273 +++++++++++++++++++++++++++++++++++++++++ src/actor.rs | 125 +++++++++++-------- src/channel.rs | 2 +- src/compat.rs | 190 ++++++++++++++++++++++++++++ src/config.rs | 9 +- src/delivery.rs | 58 +++++++-- src/error.rs | 24 +++- src/extension.rs | 5 +- src/lib.rs | 20 ++- src/runtime.rs | 91 +++++++++----- src/stats.rs | 7 +- src/sync_impl.rs | 230 ++++++++++++++++++++++++++++++++++ src/transport.rs | 2 +- src/worker.rs | 159 ++++++++++++++---------- 17 files changed, 1032 insertions(+), 174 deletions(-) create mode 100644 crates/core/Cargo.toml create mode 100644 crates/core/src/lib.rs create mode 100644 src/compat.rs create mode 100644 src/sync_impl.rs diff --git a/Cargo.lock b/Cargo.lock index 3079515..790fb63 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5324,10 +5324,15 @@ dependencies = [ "proptest", "proptest-state-machine", "serde", + "swactor-core", "tracing", "web-time 0.2.4", ] +[[package]] +name = "swactor-core" +version = "0.1.0" + [[package]] name = "swactor-datastore" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index b940538..f3051b9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ ".", + "crates/core", "crates/bindings/python", "crates/bindings/wasm-runtime", "crates/simulation", @@ -41,6 +42,7 @@ transport = [ wasm = ["no_random", "dep:web-time"] # browser/wasm32 target support [dependencies] +swactor-core = { path = "crates/core" } getrandom = { version = "0.2", optional = true } serde = { version = "1", features = ["derive"], optional = true } tracing = { version = "0.1", optional = true } diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml new file mode 100644 index 0000000..e912133 --- /dev/null +++ b/crates/core/Cargo.toml @@ -0,0 +1,4 @@ +[package] +name = "swactor-core" +version = "0.1.0" +edition = "2024" diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs new file mode 100644 index 0000000..937bc62 --- /dev/null +++ b/crates/core/src/lib.rs @@ -0,0 +1,273 @@ +//! Pure, synchronous, deterministic core for the swactor actor framework. +//! +//! This crate contains the decision logic that maps 1:1 to a TLA+ specification. +//! Every public function is a pure function: `fn(inputs) -> output` with no side +//! effects, no IO, no allocation, and no panics. +//! +//! The runtime crate (`swactor`) calls these functions to make decisions; this crate +//! never calls back into the runtime. If you can't trivially transliterate a +//! function's signature into TLA+, it doesn't belong here. + +#![no_std] +#![deny(unsafe_code)] + +// ─── Actor Lifecycle Phase ─────────────────────────────────────────────────── + +/// The lifecycle phase of an actor slot. +/// +/// Replaces the four boolean fields (`started`, `poisoned`, `stopping`, `suspended`) +/// with a single state machine. Every phase transition goes through +/// [`lifecycle_transition`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ActorPhase { + /// Actor has been inserted into the pool but `on_start` has not been called yet. + Unstarted, + /// Actor is running normally — processing messages from its mailbox. + Running, + /// Actor is suspended — messages continue to queue but are not processed. + Suspended, + /// Actor has been requested to stop gracefully. `on_stop` will be called during cleanup. + Stopping, + /// Actor panicked during `on_start` or message handling. No further processing occurs. + Poisoned, +} + +// ─── Stop / Exit Reasons ───────────────────────────────────────────────────── + +/// Reason an actor was removed from the runtime. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum StopReason { + /// Graceful stop (via `ctx.stop_self()` or `Runtime::stop_actor()`). + Normal, + /// Actor panicked and could not be restarted. + Panicked, + /// Actor stopped with a typed exit value (via `Ctx::stop_with`). + Completed, +} + +/// Why an actor exited — delivered to watchers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ExitReason { + /// Actor was explicitly stopped or removed from the pool. + Stopped, + /// Actor panicked during message handling. + Panicked, + /// The node hosting the actor left the cluster (SWIM Dead). + NodeDown, + /// Actor stopped with a typed exit value (via `Ctx::stop_with`). + Completed, +} + +// ─── Mailbox ───────────────────────────────────────────────────────────────── + +/// What to do when a bounded mailbox is full. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MailboxOverflow { + /// Drop the incoming message (newest). The message is silently discarded. + DropNewest, + /// Drop the oldest message in the queue to make room for the new one. + DropOldest, +} + +/// Result of the mailbox capacity check. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MailboxDecision { + /// The mailbox has room (or is unbounded). Accept the message. + Accept, + /// The mailbox is full and the policy is DropNewest — reject the incoming message. + RejectNewest, + /// The mailbox is full and the policy is DropOldest — evict the front of the queue. + EvictOldest, +} + +// ─── Lifecycle Events ──────────────────────────────────────────────────────── + +/// Events that drive the actor lifecycle state machine. +/// +/// Each variant corresponds to a runtime action that may cause a phase transition. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LifecycleEvent { + /// `on_start` completed successfully. + Started, + /// `on_start` panicked. + StartPanicked, + /// A message handler panicked. + MessagePanicked, + /// A graceful stop was requested (via `ctx.stop_self()` or `StopSignal`). + StopRequested, + /// A stop-with-value was requested (via `ctx.stop_with()`). + StopWithRequested, + /// The actor was suspended (via `ctx.suspend_self()`). + SuspendRequested, + /// A suspended actor was resumed (via `ResumeSignal`). + Resumed, +} + +// ─── Deliver Decision ──────────────────────────────────────────────────────── + +/// What should happen when a message arrives at an actor in a particular phase. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DeliverDecision { + /// Enqueue the message in the actor's mailbox (subject to overflow policy). + Enqueue, + /// The actor is dead/stopping — discard the message silently. + Discard, + /// The actor is suspended — enqueue but do not process until resumed. + Queue, +} + +// ─── Pure Functions ───────────────────────────────────────────────────────── + +// TLA+: LifecycleTransition +/// The actor lifecycle state machine. +/// +/// Given the current phase and an event, returns the new phase. Every phase +/// transition in the runtime must go through this function. +/// +/// Illegal transitions (e.g., `Resumed` on a non-suspended actor) return the +/// phase unchanged — the runtime is responsible for not issuing nonsensical +/// events, but the core never panics. +#[must_use] +pub fn lifecycle_transition(phase: ActorPhase, event: LifecycleEvent) -> ActorPhase { + match (phase, event) { + // Unstarted: waiting for on_start result + (ActorPhase::Unstarted, LifecycleEvent::Started) => ActorPhase::Running, + (ActorPhase::Unstarted, LifecycleEvent::StartPanicked) => ActorPhase::Poisoned, + (ActorPhase::Unstarted, LifecycleEvent::StopRequested) => ActorPhase::Stopping, + (ActorPhase::Unstarted, LifecycleEvent::StopWithRequested) => ActorPhase::Stopping, + (ActorPhase::Unstarted, LifecycleEvent::SuspendRequested) => ActorPhase::Suspended, + (ActorPhase::Unstarted, LifecycleEvent::MessagePanicked) => phase, + (ActorPhase::Unstarted, LifecycleEvent::Resumed) => phase, + + // Running: normal operation + (ActorPhase::Running, LifecycleEvent::MessagePanicked) => ActorPhase::Poisoned, + (ActorPhase::Running, LifecycleEvent::StopRequested) => ActorPhase::Stopping, + (ActorPhase::Running, LifecycleEvent::StopWithRequested) => ActorPhase::Stopping, + (ActorPhase::Running, LifecycleEvent::SuspendRequested) => ActorPhase::Suspended, + (ActorPhase::Running, LifecycleEvent::Started) => phase, + (ActorPhase::Running, LifecycleEvent::StartPanicked) => phase, + (ActorPhase::Running, LifecycleEvent::Resumed) => phase, + + // Suspended: waiting for resume or stop + (ActorPhase::Suspended, LifecycleEvent::Resumed) => ActorPhase::Running, + (ActorPhase::Suspended, LifecycleEvent::StopRequested) => ActorPhase::Stopping, + (ActorPhase::Suspended, LifecycleEvent::StopWithRequested) => ActorPhase::Stopping, + (ActorPhase::Suspended, LifecycleEvent::Started) => phase, + (ActorPhase::Suspended, LifecycleEvent::StartPanicked) => phase, + (ActorPhase::Suspended, LifecycleEvent::MessagePanicked) => phase, + (ActorPhase::Suspended, LifecycleEvent::SuspendRequested) => phase, + + // Terminal states: Stopping and Poisoned absorb all events + (ActorPhase::Stopping, _) => ActorPhase::Stopping, + (ActorPhase::Poisoned, _) => ActorPhase::Poisoned, + } +} + +// TLA+: MailboxAccept +/// Decide what to do with an incoming message given mailbox state and policy. +/// +/// `capacity` of 0 means unbounded (always accept). When the mailbox is full, +/// the overflow policy determines whether the newest or oldest message is dropped. +#[must_use] +pub fn mailbox_accept( + mailbox_len: usize, + capacity: usize, + policy: MailboxOverflow, +) -> MailboxDecision { + if capacity == 0 || mailbox_len < capacity { + MailboxDecision::Accept + } else { + match policy { + MailboxOverflow::DropNewest => MailboxDecision::RejectNewest, + MailboxOverflow::DropOldest => MailboxDecision::EvictOldest, + } + } +} + +// TLA+: ShouldTickActor +/// Whether an actor in the given phase should be ticked (i.e., have messages processed). +/// +/// Returns `false` for `Poisoned`, `Stopping`, and `Suspended` — these actors +/// either cannot process messages or must not. +#[must_use] +pub fn should_tick_actor(phase: ActorPhase) -> bool { + match phase { + ActorPhase::Unstarted | ActorPhase::Running => true, + ActorPhase::Suspended | ActorPhase::Stopping | ActorPhase::Poisoned => false, + } +} + +// TLA+: BudgetExhausted +/// Whether the per-actor message budget has been exhausted. +/// +/// `budget` of 0 means unlimited — never exhausted. +#[must_use] +pub fn budget_exhausted(processed: usize, budget: usize) -> bool { + budget > 0 && processed >= budget +} + +// TLA+: CleanupStopReason +/// Determine the stop reason for a dead actor given its phase and whether +/// it has a typed exit value. +/// +/// - `Poisoned` → `StopReason::Panicked` +/// - Has exit value → `StopReason::Completed` +/// - Otherwise → `StopReason::Normal` +#[must_use] +pub fn cleanup_stop_reason(phase: ActorPhase, has_exit_value: bool) -> StopReason { + if phase == ActorPhase::Poisoned { + StopReason::Panicked + } else if has_exit_value { + StopReason::Completed + } else { + StopReason::Normal + } +} + +// ─── Invariant Predicates ─────────────────────────────────────────────────── + +// TLA+: PhaseIsTerminal +/// Returns `true` if the phase is terminal — the actor will not process any +/// further messages and is awaiting cleanup. +#[must_use] +pub fn phase_is_terminal(phase: ActorPhase) -> bool { + match phase { + ActorPhase::Stopping | ActorPhase::Poisoned => true, + ActorPhase::Unstarted | ActorPhase::Running | ActorPhase::Suspended => false, + } +} + +// TLA+: ValidPhaseTransition +/// Returns `true` only for transitions that exist in the lifecycle state machine. +/// +/// This is an invariant predicate — useful for property-based testing and Kani +/// harnesses to verify that no illegal transition is ever taken. +#[must_use] +pub fn valid_phase_transition(from: ActorPhase, to: ActorPhase) -> bool { + match (from, to) { + // Self-transitions are always valid (no-op events) + (a, b) if a == b => true, + + // From Unstarted + (ActorPhase::Unstarted, ActorPhase::Running) => true, + (ActorPhase::Unstarted, ActorPhase::Poisoned) => true, + (ActorPhase::Unstarted, ActorPhase::Stopping) => true, + (ActorPhase::Unstarted, ActorPhase::Suspended) => true, + + // From Running + (ActorPhase::Running, ActorPhase::Poisoned) => true, + (ActorPhase::Running, ActorPhase::Stopping) => true, + (ActorPhase::Running, ActorPhase::Suspended) => true, + + // From Suspended + (ActorPhase::Suspended, ActorPhase::Running) => true, + (ActorPhase::Suspended, ActorPhase::Stopping) => true, + + // Terminal states never transition out + (ActorPhase::Stopping, _) => false, + (ActorPhase::Poisoned, _) => false, + + // Everything else is invalid + _ => false, + } +} diff --git a/src/actor.rs b/src/actor.rs index 0493ff3..198556f 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,7 +1,9 @@ -use std::any::Any; -use std::any::TypeId; -use std::collections::{HashMap, HashSet}; -use std::sync::Arc; +#[cfg(not(feature = "std"))] +use crate::compat::{Box, String, Vec}; + +use core::any::{Any, TypeId}; +use crate::compat::{Arc, TypeIdMap, TypeIdSet}; +use crate::delivery::AddrSet; use crate::Error; @@ -36,8 +38,8 @@ impl PartialEq for ExitValue { impl Eq for ExitValue {} -impl std::fmt::Debug for ExitValue { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl core::fmt::Debug for ExitValue { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { f.write_str("ExitValue(..)") } } @@ -55,36 +57,60 @@ pub struct SystemInfo { pub uptime_ms: u64, } -/// Why an actor exited. -#[derive(Debug, Clone, PartialEq, Eq)] -#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] -pub enum ExitReason { - /// Actor was explicitly stopped or removed from the pool. - Stopped, - /// Actor panicked during message handling. - Panicked, - /// The node hosting the actor left the cluster (SWIM Dead). - NodeDown, - /// Actor stopped with a typed exit value (via [`Ctx::stop_with`]). - Completed, -} +pub use swactor_core::ExitReason; /// Delivered to watchers when a watched actor exits. /// /// Implements `Message` (Clone + Send + Sync + 'static) so it can be /// delivered through normal mailbox channels. #[derive(Debug, Clone)] -#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] pub struct ActorExited { /// The address of the actor that died. pub addr: ActorAddress, /// Why it exited. pub reason: ExitReason, /// Typed exit value if the actor called [`Ctx::stop_with`]. - #[cfg_attr(feature = "serde", serde(skip))] pub exit_value: Option, } +#[cfg(feature = "serde")] +impl serde::Serialize for ActorExited { + fn serialize(&self, serializer: S) -> Result { + use serde::ser::SerializeStruct; + let mut s = serializer.serialize_struct("ActorExited", 2)?; + s.serialize_field("addr", &self.addr)?; + let reason_str = match self.reason { + ExitReason::Stopped => "Stopped", + ExitReason::Panicked => "Panicked", + ExitReason::NodeDown => "NodeDown", + ExitReason::Completed => "Completed", + }; + s.serialize_field("reason", reason_str)?; + s.end() + } +} + +#[cfg(feature = "serde")] +impl<'de> serde::Deserialize<'de> for ActorExited { + fn deserialize>(deserializer: D) -> Result { + #[derive(serde::Deserialize)] + struct Helper { + addr: ActorAddress, + reason: HelperExitReason, + } + #[derive(serde::Deserialize)] + enum HelperExitReason { Stopped, Panicked, NodeDown, Completed } + let h = Helper::deserialize(deserializer)?; + let reason = match h.reason { + HelperExitReason::Stopped => ExitReason::Stopped, + HelperExitReason::Panicked => ExitReason::Panicked, + HelperExitReason::NodeDown => ExitReason::NodeDown, + HelperExitReason::Completed => ExitReason::Completed, + }; + Ok(ActorExited { addr: h.addr, reason, exit_value: None }) + } +} + impl PartialEq for ActorExited { fn eq(&self, other: &Self) -> bool { self.addr == other.addr && self.reason == other.reason @@ -140,9 +166,9 @@ pub struct ActorAddress(pub [u8; 32]); /// Custom Hash: only hash the first 8 bytes since all 32 are random. /// SipHash on 8 bytes is ~3x faster than on 32 bytes, with identical /// collision properties (2^64 possible values from cryptographic randomness). -impl std::hash::Hash for ActorAddress { +impl core::hash::Hash for ActorAddress { #[inline] - fn hash(&self, state: &mut H) { + fn hash(&self, state: &mut H) { // SAFETY: ActorAddress is always 32 bytes, so [..8] is valid. state.write_u64(u64::from_ne_bytes( self.0[..8].try_into().unwrap(), @@ -150,8 +176,20 @@ impl std::hash::Hash for ActorAddress { } } -impl std::fmt::Display for ActorAddress { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl PartialOrd for ActorAddress { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for ActorAddress { + fn cmp(&self, other: &Self) -> core::cmp::Ordering { + self.0.cmp(&other.0) + } +} + +impl core::fmt::Display for ActorAddress { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { for b in &self.0[..8] { write!(f, "{:02x}", b)?; } @@ -177,14 +215,14 @@ impl ActorAddress { /// can clone individual entries cheaply (Arc bump) for copy-on-write overrides. #[derive(Clone, Default)] pub struct Environment { - inner: Arc>>, + inner: Arc>>, } impl Environment { /// Create an empty environment. pub fn new() -> Self { Self { - inner: Arc::new(HashMap::new()), + inner: Arc::new(TypeIdMap::new()), } } @@ -223,14 +261,14 @@ impl Environment { /// /// Allows inserting/replacing typed values before freezing into an immutable `Environment`. pub struct EnvironmentBuilder { - map: HashMap>, + map: TypeIdMap>, } impl EnvironmentBuilder { /// Create an empty builder. pub fn new() -> Self { Self { - map: HashMap::new(), + map: TypeIdMap::new(), } } @@ -304,14 +342,14 @@ impl LogicalName { #[derive(Clone, Debug)] pub struct ServiceBinding { pub addr: ActorAddress, - _marker: std::marker::PhantomData, + _marker: core::marker::PhantomData, } impl ServiceBinding { pub fn new(addr: ActorAddress) -> Self { Self { addr, - _marker: std::marker::PhantomData, + _marker: core::marker::PhantomData, } } } @@ -328,11 +366,11 @@ impl ServiceBinding { /// Built via fluent API: `CapabilitySet::new().with_send(addr).with_spawn()`. #[derive(Clone, Default)] pub struct CapabilitySet { - send_any: HashSet, - send_typed: HashSet<(TypeId, ActorAddress)>, + send_any: AddrSet, + send_typed: crate::compat::TypedAddrSet, can_spawn: bool, - service_types: HashSet, - monitor_targets: HashSet, + service_types: TypeIdSet, + monitor_targets: AddrSet, } impl CapabilitySet { @@ -386,8 +424,8 @@ impl CapabilitySet { } } -impl std::fmt::Debug for CapabilitySet { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl core::fmt::Debug for CapabilitySet { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { f.debug_struct("CapabilitySet") .field("send_any", &self.send_any.len()) .field("send_typed", &self.send_typed.len()) @@ -442,7 +480,7 @@ where let msg = match msg.downcast::() { Ok(typed) => { self.inner.handle(ctx, *typed); - return Some(std::any::type_name::()); + return Some(core::any::type_name::()); } Err(msg) => msg, }; @@ -484,16 +522,7 @@ impl MonitorRef { } } -/// Reason an actor was removed from the runtime. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum StopReason { - /// Graceful stop (via `ctx.stop_self()` or `Runtime::stop_actor()`). - Normal, - /// Actor panicked and could not be restarted. - Panicked, - /// Actor stopped with a typed exit value (via [`Ctx::stop_with`]). - Completed, -} +pub use swactor_core::StopReason; /// Death notification delivered as a normal message when a monitored actor dies. /// diff --git a/src/channel.rs b/src/channel.rs index 88b536f..3010c2a 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use crate::compat::Arc; use crossbeam_queue::{ArrayQueue, SegQueue}; diff --git a/src/compat.rs b/src/compat.rs new file mode 100644 index 0000000..c400035 --- /dev/null +++ b/src/compat.rs @@ -0,0 +1,190 @@ +// Compatibility module: re-exports types from std or core/alloc depending on feature flags. +// +// Core types (Any, fmt, atomics, etc.) are imported directly from `core::` in each file. +// This module provides conditional re-exports for types that differ between std and no_std. + +// --- alloc types (conditional on std feature) --- +#[cfg(feature = "std")] +pub use std::sync::Arc; +#[cfg(not(feature = "std"))] +pub use alloc::sync::Arc; + +#[cfg(feature = "std")] +pub use std::collections::VecDeque; +#[cfg(not(feature = "std"))] +pub use alloc::collections::VecDeque; + +// Vec, String, Box — in std prelude normally, but need explicit import in no_std. +// vec! and format! macros are available via `extern crate alloc` in lib.rs. +#[cfg(not(feature = "std"))] +pub use alloc::{boxed::Box, string::String, string::ToString, vec::Vec}; + +// --- sync primitives (std vs spin-based) --- +#[cfg(feature = "std")] +pub use std::sync::{RwLock, OnceLock}; +#[cfg(not(feature = "std"))] +pub use crate::sync_impl::SpinRwLock as RwLock; + +// --- collections (std vs alloc-based alternatives) --- +#[cfg(feature = "std")] +pub use std::collections::HashMap; +#[cfg(not(feature = "std"))] +pub use alloc::collections::BTreeMap as HashMap; + +// --- TypeId-keyed map/set --- +// +// TypeId does not implement Ord, so BTreeMap cannot be used in no_std. +// In std mode: thin wrappers around HashMap/HashSet. +// In no_std mode: linear-scan Vec wrappers (<10 entries typical). + +use core::any::TypeId as StdTypeId; + +/// A map keyed by `TypeId`. +/// std: HashMap. no_std: linear-scan Vec. +#[derive(Clone)] +pub struct TypeIdMap { + #[cfg(feature = "std")] + inner: std::collections::HashMap, + #[cfg(not(feature = "std"))] + inner: Vec<(StdTypeId, V)>, +} + +impl TypeIdMap { + pub fn new() -> Self { + Self { + #[cfg(feature = "std")] + inner: std::collections::HashMap::new(), + #[cfg(not(feature = "std"))] + inner: Vec::new(), + } + } + + #[cfg(feature = "std")] + pub fn get(&self, key: &StdTypeId) -> Option<&V> { + self.inner.get(key) + } + + #[cfg(not(feature = "std"))] + pub fn get(&self, key: &StdTypeId) -> Option<&V> { + self.inner.iter().find(|(k, _)| k == key).map(|(_, v)| v) + } + + #[cfg(feature = "std")] + pub fn contains_key(&self, key: &StdTypeId) -> bool { + self.inner.contains_key(key) + } + + #[cfg(not(feature = "std"))] + pub fn contains_key(&self, key: &StdTypeId) -> bool { + self.inner.iter().any(|(k, _)| k == key) + } + + #[cfg(feature = "std")] + pub fn insert(&mut self, key: StdTypeId, value: V) -> Option { + self.inner.insert(key, value) + } + + #[cfg(not(feature = "std"))] + pub fn insert(&mut self, key: StdTypeId, value: V) -> Option { + for entry in &mut self.inner { + if entry.0 == key { + let old = core::mem::replace(&mut entry.1, value); + return Some(old); + } + } + self.inner.push((key, value)); + None + } + + pub fn len(&self) -> usize { + self.inner.len() + } + + pub fn is_empty(&self) -> bool { + self.inner.is_empty() + } +} + +impl Default for TypeIdMap { + fn default() -> Self { + Self::new() + } +} + +/// A set of `TypeId` values. +/// std: HashSet. no_std: linear-scan Vec. +#[derive(Clone, Default)] +pub struct TypeIdSet { + #[cfg(feature = "std")] + inner: std::collections::HashSet, + #[cfg(not(feature = "std"))] + inner: Vec, +} + +impl TypeIdSet { + #[allow(dead_code)] + pub fn new() -> Self { + Self::default() + } + + pub fn insert(&mut self, id: StdTypeId) -> bool { + #[cfg(feature = "std")] + { self.inner.insert(id) } + #[cfg(not(feature = "std"))] + { + if self.inner.contains(&id) { + false + } else { + self.inner.push(id); + true + } + } + } + + pub fn contains(&self, id: &StdTypeId) -> bool { + self.inner.contains(id) + } + + pub fn len(&self) -> usize { + self.inner.len() + } +} + +/// A set of `(TypeId, ActorAddress)` pairs. +/// Used for typed send capabilities. std: HashSet. no_std: linear-scan Vec. +#[derive(Clone, Default)] +pub struct TypedAddrSet { + #[cfg(feature = "std")] + inner: std::collections::HashSet<(StdTypeId, crate::actor::ActorAddress)>, + #[cfg(not(feature = "std"))] + inner: Vec<(StdTypeId, crate::actor::ActorAddress)>, +} + +impl TypedAddrSet { + #[allow(dead_code)] + pub fn new() -> Self { + Self::default() + } + + pub fn insert(&mut self, pair: (StdTypeId, crate::actor::ActorAddress)) -> bool { + #[cfg(feature = "std")] + { self.inner.insert(pair) } + #[cfg(not(feature = "std"))] + { + if self.inner.contains(&pair) { + false + } else { + self.inner.push(pair); + true + } + } + } + + pub fn contains(&self, pair: &(StdTypeId, crate::actor::ActorAddress)) -> bool { + self.inner.contains(pair) + } + + pub fn len(&self) -> usize { + self.inner.len() + } +} diff --git a/src/config.rs b/src/config.rs index 85ddee4..a0d1e5a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,11 +1,4 @@ -/// What to do when a bounded mailbox is full. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum MailboxOverflow { - /// Drop the incoming message (newest). The message is silently discarded. - DropNewest, - /// Drop the oldest message in the queue to make room for the new one. - DropOldest, -} +pub use swactor_core::MailboxOverflow; /// The tunable settings for the runtime. pub struct RuntimeConfig { diff --git a/src/delivery.rs b/src/delivery.rs index a0e6ea1..69c91b3 100644 --- a/src/delivery.rs +++ b/src/delivery.rs @@ -1,8 +1,16 @@ -use std::any::Any; -use std::collections::{HashMap, HashSet}; -use std::hash::{BuildHasher, Hasher}; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, OnceLock, RwLock}; +#[cfg(not(feature = "std"))] +use crate::compat::{Box, Vec}; + +use core::any::Any; +#[cfg(feature = "std")] +use core::hash::{BuildHasher, Hasher}; +#[cfg(feature = "std")] +use std::collections::HashMap; +use core::sync::atomic::{AtomicUsize, Ordering}; +use crate::compat::{Arc, RwLock}; +#[cfg(feature = "std")] +use crate::compat::OnceLock; +#[cfg(feature = "std")] use std::thread::Thread; use crate::actor::{ActorAddress, Message, SpawnRequest}; @@ -21,8 +29,10 @@ use crate::Error; /// /// This is safe because the input is already random (uniform distribution), /// so additional mixing would be redundant. +#[cfg(feature = "std")] pub struct AddrHasher(u64); +#[cfg(feature = "std")] impl Hasher for AddrHasher { #[inline] fn finish(&self) -> u64 { @@ -41,9 +51,11 @@ impl Hasher for AddrHasher { } /// BuildHasher for creating AddrHasher instances. +#[cfg(feature = "std")] #[derive(Default, Clone)] pub struct AddrBuildHasher; +#[cfg(feature = "std")] impl BuildHasher for AddrBuildHasher { type Hasher = AddrHasher; @@ -55,10 +67,18 @@ impl BuildHasher for AddrBuildHasher { /// HashMap optimized for ActorAddress keys. /// Uses identity hashing since ActorAddress bytes are already random. +/// In no_std: BTreeMap using ActorAddress's Ord impl. +#[cfg(feature = "std")] pub type AddrMap = HashMap; +#[cfg(not(feature = "std"))] +pub type AddrMap = alloc::collections::BTreeMap; /// HashSet optimized for ActorAddress keys. -pub type AddrSet = HashSet; +/// In no_std: BTreeSet using ActorAddress's Ord impl. +#[cfg(feature = "std")] +pub type AddrSet = std::collections::HashSet; +#[cfg(not(feature = "std"))] +pub type AddrSet = alloc::collections::BTreeSet; // ─── Address Map Types ─────────────────────────────────────────────────────── @@ -80,12 +100,20 @@ pub(crate) struct AddressMap { } impl AddressMap { + #[cfg(feature = "std")] pub fn with_capacity(cap: usize) -> Self { Self { inner: RwLock::new(HashMap::with_capacity_and_hasher(cap, AddrBuildHasher)), } } + #[cfg(not(feature = "std"))] + pub fn with_capacity(_cap: usize) -> Self { + Self { + inner: RwLock::new(AddrMap::new()), + } + } + pub fn insert(&self, addr: ActorAddress, worker: WorkerId) { self.inner.write().unwrap().insert(addr, worker); } @@ -200,18 +228,26 @@ pub(crate) struct InboxRegistry { } impl InboxRegistry { + #[cfg(feature = "std")] pub fn new() -> Self { Self { senders: RwLock::new(HashMap::with_hasher(AddrBuildHasher)), } } + #[cfg(not(feature = "std"))] + pub fn new() -> Self { + Self { + senders: RwLock::new(AddrMap::new()), + } + } + pub fn register(&self, addr: ActorAddress, sender: Arc) { self.senders.write().unwrap().insert(addr, sender); } /// Check if an address is registered without consuming a message. - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] pub fn contains(&self, addr: &ActorAddress) -> bool { self.senders.read().unwrap().contains_key(addr) } @@ -242,14 +278,16 @@ pub(crate) struct TickContext<'a> { pub(crate) extension: Option<&'a dyn crate::extension::RuntimeExtension>, pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>, /// Thread handles for waking parked workers on cross-worker sends. + #[cfg(feature = "std")] pub(crate) worker_threads: &'a [OnceLock], /// Per-worker stats for summing total_actors across workers. pub(crate) worker_stats: &'a [Arc], /// Runtime creation time for computing uptime_ms. + #[cfg(feature = "std")] pub(crate) created_at: crate::Instant, - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] pub(crate) codec_registry: Option<&'a crate::transport::CodecRegistry>, - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] pub(crate) transport_router: Option<&'a crate::transport::TransportRouter>, } @@ -261,7 +299,7 @@ impl<'a> TickContext<'a> { addr: ActorAddress, msg: Box, ) -> Result<(), Error> { - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] { if self.inbox_registry.contains(&addr) { return self.inbox_registry.try_deliver(addr, msg); diff --git a/src/error.rs b/src/error.rs index 3294a58..914335c 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,3 +1,6 @@ +#[cfg(not(feature = "std"))] +use crate::compat::{Box, String, ToString}; + /// Simple, ergonomic, local `Error` type. /// # Usage /// ``` @@ -13,16 +16,29 @@ /// } /// ``` #[derive(Debug)] -pub struct Error(Box); +pub struct Error(Box); + +/// Small wrapper so `String` can become `Box` in both +/// std and no_std (where String doesn't auto-impl core::error::Error via blanket). +#[derive(Debug)] +struct StringError(String); + +impl core::fmt::Display for StringError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.write_str(&self.0) + } +} + +impl core::error::Error for StringError {} impl> From for Error { fn from(value: T) -> Self { - Error(value.as_ref().to_string().into()) + Error(Box::new(StringError(value.as_ref().to_string()))) } } -impl std::fmt::Display for Error { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl core::fmt::Display for Error { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { write!(f, "{}", self.0) } } diff --git a/src/extension.rs b/src/extension.rs index e9ca2e4..10799c2 100644 --- a/src/extension.rs +++ b/src/extension.rs @@ -1,4 +1,7 @@ -use std::any::Any; +#[cfg(not(feature = "std"))] +use crate::compat::{Box, Vec}; + +use core::any::Any; use crate::actor::{ActorAddress, Environment, ExitValue, StopReason}; diff --git a/src/lib.rs b/src/lib.rs index 4b32d66..4954fe6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,12 @@ +#![cfg_attr(not(feature = "std"), no_std)] + +#[cfg(not(feature = "std"))] +extern crate alloc; + +pub(crate) mod compat; +#[cfg(not(feature = "std"))] +pub(crate) mod sync_impl; + pub mod actor; pub mod extension; pub mod worker; @@ -10,7 +19,9 @@ pub(crate) mod error; pub use error::Error; // Re-export identity hashing types for ActorAddress-keyed collections. -pub use delivery::{AddrBuildHasher, AddrMap, AddrSet}; +#[cfg(feature = "std")] +pub use delivery::AddrBuildHasher; +pub use delivery::{AddrMap, AddrSet}; pub mod config; pub(crate) mod delivery; @@ -18,7 +29,7 @@ pub mod stats; pub mod runtime; -#[cfg(feature = "transport")] +#[cfg(all(feature = "transport", feature = "std"))] pub mod transport; #[cfg(feature = "std")] @@ -26,9 +37,10 @@ pub mod std; // Platform-aware Instant: web_time on wasm, std::time on native. // web_time is a no-op re-export of std::time::Instant on non-wasm targets. +// In no_std (without wasm), Instant is not available — timing returns 0. #[cfg(feature = "wasm")] pub(crate) use web_time::Instant; -#[cfg(not(feature = "wasm"))] +#[cfg(all(not(feature = "wasm"), feature = "std"))] pub(crate) use ::std::time::Instant; #[cfg(feature = "getrandom")] @@ -36,7 +48,7 @@ pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() } -#[cfg(all(feature = "no_random", not(feature = "getrandom")))] +#[cfg(not(feature = "getrandom"))] pub(crate) fn get_random(buf: &mut [u8]) { use core::sync::atomic::{AtomicUsize, Ordering}; diff --git a/src/runtime.rs b/src/runtime.rs index e75726a..b77577a 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,11 +1,16 @@ -use std::any::Any; -use std::cell::RefCell; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, OnceLock}; -#[cfg(not(target_arch = "wasm32"))] +#[cfg(not(feature = "std"))] +use crate::compat::{Box, Vec}; + +use core::any::Any; +use core::cell::RefCell; +use core::sync::atomic::{AtomicBool, Ordering}; +use crate::compat::Arc; +#[cfg(feature = "std")] +use crate::compat::OnceLock; +#[cfg(all(not(target_arch = "wasm32"), feature = "std"))] use std::thread::{self, JoinHandle}; +#[cfg(feature = "std")] use std::thread::Thread; -use crate::Instant; use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Environment, ExitValue, Message, ResumeSignal, SpawnRequest, StopSignal, StopWithSignal, SystemInfo}; use crate::channel::{Receiver, Sender}; @@ -69,13 +74,13 @@ impl Ask { } /// Handle for dealing with a runtime that has started via the `Runtime::run()` method. -#[cfg(not(target_arch = "wasm32"))] +#[cfg(all(not(target_arch = "wasm32"), feature = "std"))] pub struct RuntimeHandle { pub runtime: Arc, threads: Vec>, } -#[cfg(not(target_arch = "wasm32"))] +#[cfg(all(not(target_arch = "wasm32"), feature = "std"))] impl RuntimeHandle { pub fn join(self) { for handle in self.threads { @@ -110,11 +115,13 @@ pub struct Runtime { /// Workers available for tick(). run() drains this and moves workers to threads. tick_workers: RefCell>, /// Thread handles for waking parked workers. Set by workers on startup via OnceLock. + #[cfg(feature = "std")] worker_threads: Arc>>, - created_at: Instant, - #[cfg(feature = "transport")] + #[cfg(feature = "std")] + created_at: crate::Instant, + #[cfg(all(feature = "transport", feature = "std"))] codec_registry: Option>, - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] transport_router: Option>, } @@ -129,8 +136,8 @@ unsafe impl Sync for Runtime {} #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] pub struct RuntimeAddress(pub [u8; 32]); -impl std::fmt::Display for RuntimeAddress { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl core::fmt::Display for RuntimeAddress { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { for b in &self.0[..8] { write!(f, "{:02x}", b)?; } @@ -155,6 +162,7 @@ impl RuntimeAddress { pub struct ExternalSender { address_map: Arc, transfer_txs: Vec>, + #[cfg(feature = "std")] worker_threads: Arc>>, } @@ -163,6 +171,7 @@ impl Clone for ExternalSender { Self { address_map: self.address_map.clone(), transfer_txs: self.transfer_txs.clone(), + #[cfg(feature = "std")] worker_threads: self.worker_threads.clone(), } } @@ -182,6 +191,7 @@ impl ExternalSender { Some(wid) => { self.transfer_txs[wid.as_usize()] .send(Envelope::new(addr, Box::new(msg))); + #[cfg(feature = "std")] notify_worker(&self.worker_threads, wid.as_usize()); Ok(()) } @@ -232,6 +242,7 @@ impl Runtime { let placement = Placement::new(num_workers, worker_stats.clone()); + #[cfg(feature = "std")] let worker_threads: Arc>> = Arc::new((0..num_workers).map(|_| OnceLock::new()).collect()); @@ -247,11 +258,13 @@ impl Runtime { worker_stats, stats_hook: None, tick_workers: RefCell::new(workers), + #[cfg(feature = "std")] worker_threads, - created_at: Instant::now(), - #[cfg(feature = "transport")] + #[cfg(feature = "std")] + created_at: crate::Instant::now(), + #[cfg(all(feature = "transport", feature = "std"))] codec_registry: None, - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] transport_router: None, }; @@ -369,6 +382,7 @@ impl Runtime { ExternalSender { address_map: self.address_map.clone(), transfer_txs: self.transfer_txs.iter().cloned().collect(), + #[cfg(feature = "std")] worker_threads: self.worker_threads.clone(), } } @@ -383,12 +397,14 @@ impl Runtime { config: &self.config, extension: self.extension.as_deref(), stats_hook: self.stats_hook.as_deref(), + #[cfg(feature = "std")] worker_threads: &self.worker_threads, worker_stats: &self.worker_stats, + #[cfg(feature = "std")] created_at: self.created_at, - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] codec_registry: self.codec_registry.as_deref(), - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] transport_router: self.transport_router.as_deref(), } } @@ -414,7 +430,7 @@ impl Runtime { /// In single-threaded mode, one background thread is spawned. /// /// Not available on wasm32 — use the browser crate's Web Worker-based run instead. - #[cfg(not(target_arch = "wasm32"))] + #[cfg(all(not(target_arch = "wasm32"), feature = "std"))] pub fn run(self) -> Result { self.is_running.store(true, Ordering::Release); @@ -464,7 +480,10 @@ impl Runtime { .map(|ws| ws.drain_tick_timings()) .collect(); + #[cfg(feature = "std")] let uptime_ms = self.created_at.elapsed().as_millis() as u64; + #[cfg(not(feature = "std"))] + let uptime_ms = 0u64; RuntimeStats { num_workers, uptime_ms, actors, workers, actor_details: Vec::new(), tick_timings } } @@ -480,6 +499,7 @@ impl Runtime { Some(wid) => { self.transfer_txs[wid.as_usize()] .send(Envelope::new(addr, Box::new(StopSignal))); + #[cfg(feature = "std")] notify_worker(&self.worker_threads, wid.as_usize()); Ok(()) } @@ -494,9 +514,12 @@ impl Runtime { self.is_running.store(false, Ordering::Release); // Wake all parked workers so they see the shutdown flag immediately - for thread in self.worker_threads.iter() { - if let Some(t) = thread.get() { - t.unpark(); + #[cfg(feature = "std")] + { + for thread in self.worker_threads.iter() { + if let Some(t) = thread.get() { + t.unpark(); + } } } } @@ -509,13 +532,13 @@ impl Runtime { } /// Set the codec registry for remote transport. - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] pub fn set_codec_registry(&mut self, registry: Arc) { self.codec_registry = Some(registry); } /// Set the transport router for remote message delivery. - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] pub fn set_transport_router(&mut self, router: Arc) { self.transport_router = Some(router); } @@ -524,7 +547,7 @@ impl Runtime { /// /// Used by [`CodecRegistry::receive`](crate::transport::CodecRegistry::receive) /// to inject incoming messages from remote runtimes. - #[cfg(feature = "transport")] + #[cfg(all(feature = "transport", feature = "std"))] pub fn deliver_raw( &self, addr: ActorAddress, @@ -543,6 +566,7 @@ impl Runtime { /// Wake a parked worker thread so it can process new work. /// No-op if the thread handle hasn't been registered yet (single-threaded tick mode). +#[cfg(feature = "std")] #[inline] pub(crate) fn notify_worker(threads: &[OnceLock], wid: usize) { if let Some(t) = threads.get(wid).and_then(|o| o.get()) { @@ -557,6 +581,7 @@ impl ContextInner for Runtime { Some(wid) => { self.transfer_txs[wid.as_usize()] .send(Envelope::new(addr, msg)); + #[cfg(feature = "std")] notify_worker(&self.worker_threads, wid.as_usize()); Ok(()) } @@ -569,6 +594,7 @@ impl ContextInner for Runtime { self.address_map.insert(request.addr, worker_id); self.spawn_txs[worker_id.as_usize()] .send(request); + #[cfg(feature = "std")] notify_worker(&self.worker_threads, worker_id.as_usize()); } @@ -577,6 +603,7 @@ impl ContextInner for Runtime { if let Some(wid) = self.address_map.lookup(&addr) { self.transfer_txs[wid.as_usize()] .send(Envelope::new(addr, Box::new(StopSignal))); + #[cfg(feature = "std")] notify_worker(&self.worker_threads, wid.as_usize()); } } @@ -585,19 +612,22 @@ impl ContextInner for Runtime { if let Some(wid) = self.address_map.lookup(&addr) { self.transfer_txs[wid.as_usize()] .send(Envelope::new(addr, Box::new(StopWithSignal(value)))); + #[cfg(feature = "std")] notify_worker(&self.worker_threads, wid.as_usize()); } } - fn request_suspend(&self, addr: ActorAddress) { + fn request_suspend(&self, _addr: ActorAddress) { // Outside worker context — not supported (suspend is per-actor, from handler) - eprintln!("swactor: request_suspend called outside worker context for {addr} — ignored"); + #[cfg(feature = "std")] + eprintln!("swactor: request_suspend called outside worker context for {_addr} — ignored"); } fn request_resume(&self, addr: ActorAddress) { if let Some(wid) = self.address_map.lookup(&addr) { self.transfer_txs[wid.as_usize()] .send(Envelope::new(addr, Box::new(ResumeSignal))); + #[cfg(feature = "std")] notify_worker(&self.worker_threads, wid.as_usize()); } } @@ -605,6 +635,7 @@ impl ContextInner for Runtime { fn post_worker_request(&self, _request: Box) { // Worker requests (e.g., timers) are per-worker; posting from outside // a worker context (e.g., rt.spawn() callback) is not supported. + #[cfg(feature = "std")] eprintln!("swactor: post_worker_request called outside worker context — ignored"); } @@ -617,11 +648,15 @@ impl ContextInner for Runtime { let total_actors: usize = self.worker_stats.iter() .map(|ws| ws.num_actors.load(Ordering::Relaxed)) .sum(); + #[cfg(feature = "std")] + let uptime_ms = self.created_at.elapsed().as_millis() as u64; + #[cfg(not(feature = "std"))] + let uptime_ms = 0u64; SystemInfo { worker_id: 0, num_workers, total_actors, - uptime_ms: self.created_at.elapsed().as_millis() as u64, + uptime_ms, } } } diff --git a/src/stats.rs b/src/stats.rs index 436e869..c375e0c 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -1,4 +1,7 @@ -use std::sync::atomic::{AtomicU64, AtomicUsize}; +#[cfg(not(feature = "std"))] +use crate::compat::{String, Vec}; + +use core::sync::atomic::{AtomicU64, AtomicUsize}; use crossbeam_queue::ArrayQueue; @@ -83,7 +86,7 @@ impl WorkerStats { /// Create a point-in-time snapshot as a [`WorkerInfo`]. pub fn snapshot(&self, id: usize) -> WorkerInfo { - use std::sync::atomic::Ordering::Relaxed; + use core::sync::atomic::Ordering::Relaxed; WorkerInfo { id, num_actors: self.num_actors.load(Relaxed), diff --git a/src/sync_impl.rs b/src/sync_impl.rs new file mode 100644 index 0000000..c931ffd --- /dev/null +++ b/src/sync_impl.rs @@ -0,0 +1,230 @@ +//! Spin-based sync primitives for `#[cfg(not(feature = "std"))]`. +//! +//! These are minimal implementations sufficient for the swactor runtime +//! in single-threaded cooperative (no_std) environments. They use atomic +//! spin loops and are NOT suitable for highly contended multi-threaded use. +#![allow(dead_code)] + +use core::cell::UnsafeCell; +use core::ops::{Deref, DerefMut}; +use core::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering}; + +// ─── SpinMutex ────────────────────────────────────────────────────────────── + +/// A simple spin-lock mutex. +/// +/// Returns `Result` from `lock()` to match `std::sync::Mutex` API +/// (the codebase calls `.lock().unwrap()`). The `Err` variant is never produced. +pub struct SpinMutex { + locked: AtomicBool, + data: UnsafeCell, +} + +unsafe impl Send for SpinMutex {} +unsafe impl Sync for SpinMutex {} + +impl SpinMutex { + pub const fn new(value: T) -> Self { + Self { + locked: AtomicBool::new(false), + data: UnsafeCell::new(value), + } + } + + pub fn lock(&self) -> Result, core::convert::Infallible> { + while self + .locked + .compare_exchange_weak(false, true, Ordering::Acquire, Ordering::Relaxed) + .is_err() + { + core::hint::spin_loop(); + } + Ok(SpinMutexGuard { lock: self }) + } +} + +pub struct SpinMutexGuard<'a, T> { + lock: &'a SpinMutex, +} + +impl Deref for SpinMutexGuard<'_, T> { + type Target = T; + fn deref(&self) -> &T { + unsafe { &*self.lock.data.get() } + } +} + +impl DerefMut for SpinMutexGuard<'_, T> { + fn deref_mut(&mut self) -> &mut T { + unsafe { &mut *self.lock.data.get() } + } +} + +impl Drop for SpinMutexGuard<'_, T> { + fn drop(&mut self) { + self.lock.locked.store(false, Ordering::Release); + } +} + +// ─── SpinRwLock ───────────────────────────────────────────────────────────── + +/// A simple spin-lock reader-writer lock. +/// +/// State encoding in `state: AtomicUsize`: +/// - 0 = unlocked +/// - 1..usize::MAX-1 = N active readers +/// - usize::MAX = writer holds the lock +const WRITER: usize = usize::MAX; + +pub struct SpinRwLock { + state: AtomicUsize, + data: UnsafeCell, +} + +unsafe impl Send for SpinRwLock {} +unsafe impl Sync for SpinRwLock {} + +impl SpinRwLock { + pub const fn new(value: T) -> Self { + Self { + state: AtomicUsize::new(0), + data: UnsafeCell::new(value), + } + } + + pub fn read(&self) -> Result, core::convert::Infallible> { + loop { + let s = self.state.load(Ordering::Relaxed); + if s == WRITER { + core::hint::spin_loop(); + continue; + } + if self + .state + .compare_exchange_weak(s, s + 1, Ordering::Acquire, Ordering::Relaxed) + .is_ok() + { + return Ok(SpinRwLockReadGuard { lock: self }); + } + core::hint::spin_loop(); + } + } + + pub fn write(&self) -> Result, core::convert::Infallible> { + loop { + if self + .state + .compare_exchange_weak(0, WRITER, Ordering::Acquire, Ordering::Relaxed) + .is_ok() + { + return Ok(SpinRwLockWriteGuard { lock: self }); + } + core::hint::spin_loop(); + } + } +} + +pub struct SpinRwLockReadGuard<'a, T> { + lock: &'a SpinRwLock, +} + +impl Deref for SpinRwLockReadGuard<'_, T> { + type Target = T; + fn deref(&self) -> &T { + unsafe { &*self.lock.data.get() } + } +} + +impl Drop for SpinRwLockReadGuard<'_, T> { + fn drop(&mut self) { + self.lock.state.fetch_sub(1, Ordering::Release); + } +} + +pub struct SpinRwLockWriteGuard<'a, T> { + lock: &'a SpinRwLock, +} + +impl Deref for SpinRwLockWriteGuard<'_, T> { + type Target = T; + fn deref(&self) -> &T { + unsafe { &*self.lock.data.get() } + } +} + +impl DerefMut for SpinRwLockWriteGuard<'_, T> { + fn deref_mut(&mut self) -> &mut T { + unsafe { &mut *self.lock.data.get() } + } +} + +impl Drop for SpinRwLockWriteGuard<'_, T> { + fn drop(&mut self) { + self.lock.state.store(0, Ordering::Release); + } +} + +// ─── SpinOnceLock ─────────────────────────────────────────────────────────── + +const EMPTY: u8 = 0; +const INITIALIZING: u8 = 1; +const READY: u8 = 2; + +/// A spin-based `OnceLock` matching the `std::sync::OnceLock` API surface. +pub struct SpinOnceLock { + state: AtomicU8, + data: UnsafeCell>, +} + +unsafe impl Send for SpinOnceLock {} +unsafe impl Sync for SpinOnceLock {} + +impl SpinOnceLock { + pub const fn new() -> Self { + Self { + state: AtomicU8::new(EMPTY), + data: UnsafeCell::new(None), + } + } + + pub fn get(&self) -> Option<&T> { + if self.state.load(Ordering::Acquire) == READY { + unsafe { (*self.data.get()).as_ref() } + } else { + None + } + } + + pub fn set(&self, value: T) -> Result<(), T> { + if self + .state + .compare_exchange(EMPTY, INITIALIZING, Ordering::Acquire, Ordering::Relaxed) + .is_ok() + { + unsafe { *self.data.get() = Some(value) }; + self.state.store(READY, Ordering::Release); + Ok(()) + } else { + Err(value) + } + } + + pub fn get_or_init T>(&self, f: F) -> &T { + if self.state.load(Ordering::Acquire) == READY { + return unsafe { (*self.data.get()).as_ref().unwrap() }; + } + if self + .state + .compare_exchange(EMPTY, INITIALIZING, Ordering::Acquire, Ordering::Relaxed) + .is_ok() + { + unsafe { *self.data.get() = Some(f()) }; + self.state.store(READY, Ordering::Release); + } else { + while self.state.load(Ordering::Acquire) != READY { + core::hint::spin_loop(); + } + } + unsafe { (*self.data.get()).as_ref().unwrap() } + } +} diff --git a/src/transport.rs b/src/transport.rs index 42ed64c..230b982 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -8,7 +8,7 @@ //! - **[`Codec`]**: HOW bytes are encoded — gRPC/protobuf, bincode, custom, etc. //! - **[`Transport`]**: WHERE bytes are sent — in-memory, gRPC channel, etc. -use std::any::{Any, TypeId}; +use core::any::{Any, TypeId}; use std::collections::HashMap; use std::sync::{Arc, RwLock}; diff --git a/src/worker.rs b/src/worker.rs index 50ce227..f3548b3 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -1,15 +1,22 @@ -use std::any::Any; -use std::cell::RefCell; -use std::collections::{HashMap, VecDeque}; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +#[cfg(not(feature = "std"))] +use crate::compat::{Box, Vec}; + +use core::any::Any; +use core::cell::RefCell; +use core::sync::atomic::Ordering; +#[cfg(feature = "std")] +use core::sync::atomic::AtomicBool; +use crate::compat::{Arc, HashMap, VecDeque}; +#[cfg(feature = "std")] use std::thread; +#[cfg(feature = "std")] use crate::Instant; use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, Environment, ExitValue, ResumeSignal, SpawnRequest, StopReason, StopSignal, StopWithSignal, SystemInfo}; use crate::channel::Receiver; use crate::config::MailboxOverflow; -use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId}; +use crate::delivery::{AddrMap, Envelope, TickContext, WorkerId}; +use swactor_core::{ActorPhase, LifecycleEvent, lifecycle_transition}; use crate::stats::{ActorSnapshot, TickTiming, WorkerStats}; use crate::Error; @@ -29,6 +36,7 @@ fn route_to_pool_or_remote( match tc.address_map.lookup(&dest) { Some(wid) => { tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg)); + #[cfg(feature = "std")] crate::runtime::notify_worker(tc.worker_threads, wid.as_usize()); } None => { @@ -86,7 +94,11 @@ impl Worker { let mut spawn_count: usize = 0; while let Some(mut req) = self.spawn_rx.try_recv() { if let Some(ext) = tc.extension { - req.env = ext.on_spawn(req.addr, req.parent, req.env, tc.created_at.elapsed().as_millis() as u64); + #[cfg(feature = "std")] + let uptime_ms = tc.created_at.elapsed().as_millis() as u64; + #[cfg(not(feature = "std"))] + let uptime_ms = 0u64; + req.env = ext.on_spawn(req.addr, req.parent, req.env, uptime_ms); } self.pool.insert(req); #[cfg(feature = "tracing")] @@ -169,10 +181,12 @@ impl Worker { } let mut did_work = false; + #[cfg(feature = "std")] let t0 = Instant::now(); // 1. Drain spawn queue → add actors to pool did_work |= self.drain_spawns(tc); + #[cfg(feature = "std")] let t1 = Instant::now(); // 2. Drain transfer queue → deliver envelopes to actors @@ -182,6 +196,7 @@ impl Worker { self.pool.deliver(&dest, payload); did_work = true; } + #[cfg(feature = "std")] let t2 = Instant::now(); // 2.5. Fire per-worker extension (e.g., timers) → deliver before tick_all @@ -218,6 +233,7 @@ impl Worker { did_work = true; } } + #[cfg(feature = "std")] let t3 = Instant::now(); #[cfg(feature = "tracing")] @@ -232,6 +248,7 @@ impl Worker { // 4. Drain spawn queue again — actors spawned during step 3 // must be in the pool before pending_local delivery. did_work |= self.drain_spawns(tc); + #[cfg(feature = "std")] let t4 = Instant::now(); // 5. Drain pending_local buffer → deliver to local actors @@ -250,6 +267,7 @@ impl Worker { } } + #[cfg(feature = "std")] let t5 = Instant::now(); // 6. Publish stats (skip entirely when idle to avoid allocation + mutex) @@ -268,10 +286,12 @@ impl Worker { } } + #[cfg(feature = "std")] let t6 = Instant::now(); // Record tick timing let timing = TickTiming { + #[cfg(feature = "std")] phase_us: [ t1.duration_since(t0).as_micros() as u64, t2.duration_since(t1).as_micros() as u64, @@ -280,6 +300,8 @@ impl Worker { t5.duration_since(t4).as_micros() as u64, t6.duration_since(t5).as_micros() as u64, ], + #[cfg(not(feature = "std"))] + phase_us: [0; 6], messages_processed: processed, did_work, }; @@ -303,6 +325,7 @@ impl Worker { did_work } + #[cfg(feature = "std")] pub(crate) fn run(&mut self, tc: &TickContext, is_running: &AtomicBool) { #[cfg(feature = "tracing")] let _span = tracing::info_span!("worker.run", worker_id = self.id.0).entered(); @@ -343,6 +366,7 @@ impl ContextInner for WorkerContext<'_> { Some(wid) => { self.stats.cross_sends.fetch_add(1, Ordering::Relaxed); self.tc.transfer_txs[wid.as_usize()].send(Envelope::new(addr, msg)); + #[cfg(feature = "std")] crate::runtime::notify_worker(self.tc.worker_threads, wid.as_usize()); Ok(()) } @@ -358,6 +382,7 @@ impl ContextInner for WorkerContext<'_> { self.tc.address_map.insert(request.addr, worker_id); self.tc.spawn_txs[worker_id.as_usize()] .send(request); + #[cfg(feature = "std")] crate::runtime::notify_worker(self.tc.worker_threads, worker_id.as_usize()); } @@ -392,11 +417,15 @@ impl ContextInner for WorkerContext<'_> { let total_actors: usize = self.tc.worker_stats.iter() .map(|ws| ws.num_actors.load(Ordering::Relaxed)) .sum(); + #[cfg(feature = "std")] + let uptime_ms = self.tc.created_at.elapsed().as_millis() as u64; + #[cfg(not(feature = "std"))] + let uptime_ms = 0u64; SystemInfo { worker_id: self.worker_id.0, num_workers, total_actors, - uptime_ms: self.tc.created_at.elapsed().as_millis() as u64, + uptime_ms, } } } @@ -404,13 +433,8 @@ impl ContextInner for WorkerContext<'_> { struct ActorSlot { mailbox: VecDeque>, actor: Box, - poisoned: bool, - /// Graceful stop requested (via StopSignal). - stopping: bool, - /// Whether on_start has been called for this actor. - started: bool, - /// Actor is suspended — messages queue but are not processed. - suspended: bool, + /// Lifecycle phase — replaces the old `started`, `poisoned`, `stopping`, `suspended` booleans. + phase: ActorPhase, last_msg_type: Option<&'static str>, messages_processed: u64, /// Per-message-type counters (bounded to 32 entries). @@ -438,7 +462,10 @@ pub(crate) struct ActorPool { impl ActorPool { pub fn new(default_mailbox_capacity: usize, default_overflow_policy: MailboxOverflow) -> Self { Self { - actors: HashMap::with_hasher(AddrBuildHasher), + #[cfg(feature = "std")] + actors: std::collections::HashMap::with_hasher(crate::delivery::AddrBuildHasher), + #[cfg(not(feature = "std"))] + actors: AddrMap::new(), default_mailbox_capacity, default_overflow_policy, drops_this_tick: 0, @@ -451,10 +478,7 @@ impl ActorPool { self.actors.insert(req.addr, ActorSlot { mailbox: VecDeque::with_capacity(prealloc), actor: req.actor, - poisoned: false, - stopping: false, - started: false, - suspended: false, + phase: ActorPhase::Unstarted, last_msg_type: None, messages_processed: 0, msg_type_counts: HashMap::new(), @@ -472,13 +496,13 @@ impl ActorPool { if let Some(slot) = self.actors.get_mut(addr) { // Intercept control signals for suspended actors: they skip tick_all // so we must handle resume/stop at delivery time. - if slot.suspended { + if slot.phase == ActorPhase::Suspended { if msg.is::() { - slot.suspended = false; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::Resumed); return true; } if msg.is::() { - slot.stopping = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StopRequested); slot.mailbox.clear(); return true; } @@ -486,21 +510,20 @@ impl ActorPool { if let Ok(sig) = msg.downcast::() { slot.exit_value = Some(sig.0); } - slot.stopping = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StopWithRequested); slot.mailbox.clear(); return true; } } - if slot.mailbox_capacity > 0 && slot.mailbox.len() >= slot.mailbox_capacity { - match slot.overflow_policy { - MailboxOverflow::DropNewest => { - self.drops_this_tick += 1; - return true; - } - MailboxOverflow::DropOldest => { - slot.mailbox.pop_front(); - self.drops_this_tick += 1; - } + match swactor_core::mailbox_accept(slot.mailbox.len(), slot.mailbox_capacity, slot.overflow_policy) { + swactor_core::MailboxDecision::Accept => {} + swactor_core::MailboxDecision::RejectNewest => { + self.drops_this_tick += 1; + return true; + } + swactor_core::MailboxDecision::EvictOldest => { + slot.mailbox.pop_front(); + self.drops_this_tick += 1; } } slot.mailbox.push_back(msg); @@ -512,7 +535,7 @@ impl ActorPool { /// Take and reset the drop counter for this tick. pub fn take_drops(&mut self) -> usize { - std::mem::replace(&mut self.drops_this_tick, 0) + core::mem::replace(&mut self.drops_this_tick, 0) } /// Tick all actors in the pool. Returns the number of messages processed. @@ -530,14 +553,11 @@ impl ActorPool { ) -> usize { let mut count = 0; for (&addr, slot) in self.actors.iter_mut() { - if slot.poisoned || slot.stopping { - // Discard all messages for poisoned/stopping actors - slot.mailbox.clear(); - continue; - } - - // Skip suspended actors — messages keep queueing - if slot.suspended { + if !swactor_core::should_tick_actor(slot.phase) { + // Discard mailbox for poisoned/stopping; suspended keeps queueing + if slot.phase == ActorPhase::Poisoned || slot.phase == ActorPhase::Stopping { + slot.mailbox.clear(); + } continue; } @@ -554,26 +574,30 @@ impl ActorPool { let ctx = Ctx::new(inner, addr, slot.parent_addr, slot.env.clone(), snap_processed, snap_depth, snap_type_counts); // Call on_start once, before first message - if !slot.started { + if slot.phase == ActorPhase::Unstarted { + #[cfg(feature = "std")] let start_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { slot.actor.on_start(&ctx); })); - slot.started = true; + #[cfg(not(feature = "std"))] + let start_result: Result<(), ()> = { slot.actor.on_start(&ctx); Ok(()) }; if start_result.is_err() { stats.panics.fetch_add(1, Ordering::Relaxed); + #[cfg(feature = "std")] eprintln!("swactor: actor {addr} panicked in on_start — poisoned"); #[cfg(feature = "tracing")] tracing::error!(actor_addr = %addr, "actor.on_start_panicked"); - slot.poisoned = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StartPanicked); slot.mailbox.clear(); continue; } + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::Started); // Check if on_start requested stop or stop_with { let stops = stop_requests.borrow(); if !stops.is_empty() && stops.contains(&addr) { drop(stops); - slot.stopping = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StopRequested); stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); // Check for stop_with value @@ -591,7 +615,7 @@ impl ActorPool { if let Some(pos) = sws.iter().position(|(a, _)| *a == addr) { let (_, val) = sws.swap_remove(pos); slot.exit_value = Some(val); - slot.stopping = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StopWithRequested); stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); continue; @@ -602,7 +626,7 @@ impl ActorPool { let suspends = suspend_requests.borrow(); if !suspends.is_empty() && suspends.contains(&addr) { drop(suspends); - slot.suspended = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::SuspendRequested); continue; } } @@ -612,7 +636,7 @@ impl ActorPool { while let Some(msg) = slot.mailbox.pop_front() { // Intercept StopSignal (from external runtime.stop_actor) if msg.is::() { - slot.stopping = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StopRequested); stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); #[cfg(feature = "tracing")] @@ -625,15 +649,18 @@ impl ActorPool { if let Ok(sig) = msg.downcast::() { slot.exit_value = Some(sig.0); } - slot.stopping = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StopWithRequested); stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); break; } + #[cfg(feature = "std")] let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { slot.actor.handle_any(&ctx, msg) })); + #[cfg(not(feature = "std"))] + let result: Result, ()> = Ok(slot.actor.handle_any(&ctx, msg)); match result { Ok(None) => { stats.type_mismatches.fetch_add(1, Ordering::Relaxed); @@ -641,10 +668,11 @@ impl ActorPool { Err(_) => { stats.panics.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); + #[cfg(feature = "std")] eprintln!("swactor: actor {addr} panicked — poisoned, future messages will be discarded"); #[cfg(feature = "tracing")] tracing::error!(actor_addr = %addr, "actor.panicked"); - slot.poisoned = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::MessagePanicked); slot.mailbox.clear(); break; } @@ -675,7 +703,7 @@ impl ActorPool { slot.exit_value = Some(val); } drop(sws); - slot.stopping = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::StopRequested); stats.stops.fetch_add(1, Ordering::Relaxed); slot.mailbox.clear(); break; @@ -688,12 +716,12 @@ impl ActorPool { let suspends = suspend_requests.borrow(); if !suspends.is_empty() && suspends.contains(&addr) { drop(suspends); - slot.suspended = true; + slot.phase = lifecycle_transition(slot.phase, LifecycleEvent::SuspendRequested); break; // stop processing this actor's messages this tick } } - if budget > 0 && actor_count >= budget { + if swactor_core::budget_exhausted(actor_count, budget) { break; } } @@ -723,21 +751,15 @@ impl ActorPool { let dead_addrs: Vec = self .actors .iter() - .filter(|(_, slot)| slot.poisoned || slot.stopping) + .filter(|(_, slot)| slot.phase == ActorPhase::Poisoned || slot.phase == ActorPhase::Stopping) .map(|(&addr, _)| addr) .collect(); let mut dead = Vec::with_capacity(dead_addrs.len()); for addr in dead_addrs { if let Some(mut slot) = self.actors.remove(&addr) { - let reason = if slot.poisoned { - StopReason::Panicked - } else if slot.exit_value.is_some() { - StopReason::Completed - } else { - StopReason::Normal - }; + let reason = swactor_core::cleanup_stop_reason(slot.phase, slot.exit_value.is_some()); // Call on_stop for gracefully stopping actors only - if slot.stopping && !slot.poisoned { + if slot.phase == ActorPhase::Stopping { let mut type_counts: Vec<(&'static str, u64)> = slot.msg_type_counts.iter().map(|(&k, &v)| (k, v)).collect(); type_counts.sort_by(|a, b| b.1.cmp(&a.1)); @@ -747,9 +769,12 @@ impl ActorPool { slot.mailbox.len(), type_counts, ); - let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + #[cfg(feature = "std")] + { let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { slot.actor.on_stop(&ctx); - })); + })); } + #[cfg(not(feature = "std"))] + { slot.actor.on_stop(&ctx); } } dead.push((addr, reason, slot.exit_value.take())); // slot is dropped here — actor resources freed @@ -770,7 +795,7 @@ impl ActorPool { mailbox_depth: slot.mailbox.len(), last_msg_type: slot.last_msg_type, messages_processed: slot.messages_processed, - poisoned: slot.poisoned, + poisoned: slot.phase == ActorPhase::Poisoned, message_type_counts: type_counts, } }));