diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md deleted file mode 100644 index a93831e..0000000 --- a/ARCHITECTURE.md +++ /dev/null @@ -1,199 +0,0 @@ -# Swactor Architecture - -## System Diagram - -``` -┌─────────────────────────────────────────────────────────────────────────────┐ -│ Runtime │ -│ (composes everything) │ -│ │ -│ ┌───────────────────────────────────────────────────────────────────────┐ │ -│ │ Address Map │ │ -│ │ ActorAddress → WorkerId │ │ -│ │ (shared across all workers, read-heavy) │ │ -│ └──────┬──────────────────┬──────────────────────┬─────────────────────┘ │ -│ │ │ │ │ -│ ▼ ▼ ▼ │ -│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ -│ │ Worker 0 │ │ Worker 1 │ ... │ Worker N │ │ -│ │ (thread) │ │ (thread) │ │ (thread) │ │ -│ │ │ │ │ │ │ │ -│ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ -│ │ │ Actor A │ │ │ │ Actor C │ │ │ │ Actor E │ │ │ -│ │ │ [═════] │ │ │ │ [═════] │ │ │ │ [═════] │ │ │ -│ │ │ mailbox │ │ │ │ mailbox │ │ │ │ mailbox │ │ │ -│ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │ -│ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ -│ │ │ Actor B │ │ │ │ Actor D │ │ │ │ Actor F │ │ │ -│ │ │ [═════] │ │ │ │ [═════] │ │ │ │ [═════] │ │ │ -│ │ │ mailbox │ │ │ │ mailbox │ │ │ │ mailbox │ │ │ -│ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │ -│ │ │ │ │ │ │ │ -│ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ -│ │ │Transfer │◄├────├─┤Transfer │◄├───────├─┤Transfer │ │ │ -│ │ │ Queue │ │ │ │ Queue │ │ │ │ Queue │ │ │ -│ │ │ (MPSC) │─├────├►│ (MPSC) │─├───────├►│ (MPSC) │ │ │ -│ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │ -│ └─────────────┘ └─────────────┘ └─────────────┘ │ -│ │ -└─────────────────────────────────────────────────────────────────────────────┘ - - - ══ = VecDeque (no atomics) -``` - -## Message Flow - -``` - SAME WORKER (fast path — zero atomics) - ═══════════════════════════════════════ - - Actor A Actor B - handle() { mailbox (VecDeque) - ctx.send(addr_B, msg) ▲ - │ │ - ├─ address_map[addr_B] │ - │ → Worker 0 (that's me!) │ - │ │ - └─ mailbox_B.push(msg) ──────┘ - } no atomics, no envelope - - - CROSS-WORKER (one atomic hop) - ═══════════════════════════════ - - Actor A (Worker 0) Worker 1 Actor C (Worker 1) - handle() { transfer queue mailbox (VecDeque) - ctx.send(addr_C, msg) ▲ ▲ - │ │ │ - ├─ address_map[addr_C] │ │ - │ → Worker 1 (not me) │ │ - │ │ │ - └─ envelope(addr_C, msg) ──────┘ │ - (atomic push) │ │ - └── worker 1 pops ─────┘ - and distributes - (local, no atomic) - } -``` - -## Worker Loop - -``` - ┌─────────────────────────────────────────────┐ - │ Worker Thread │ - │ │ - │ loop { │ - │ ┌──────────────────────────────────────┐ │ - │ │ 1. DRAIN TRANSFER QUEUE │ │ - │ │ while let Some((addr, env)) = │ │ - │ │ transfer_queue.pop() │ │ - │ │ { │ │ - │ │ local_actors[addr].mailbox │ │ - │ │ .push(env.unpack()) │ │ - │ │ } │ │ - │ └──────────────────────────────────────┘ │ - │ ┌──────────────────────────────────────┐ │ - │ │ 2. TICK ACTORS │ │ - │ │ for actor in &mut actor_pool { │ │ - │ │ let n = drain_count(actor); │ │ - │ │ for _ in 0..n { │ │ - │ │ let msg = actor.mailbox.pop();│ │ - │ │ actor.handle(&ctx, msg); │ │ - │ │ } │ │ - │ │ } │ │ - │ └──────────────────────────────────────┘ │ - │ ┌──────────────────────────────────────┐ │ - │ │ 3. IDLE? │ │ - │ │ if no messages processed: │ │ - │ │ spin → yield → park │ │ - │ └──────────────────────────────────────┘ │ - │ } │ - └───────────────────────────────────────────────┘ -``` - -## File Tree - -``` -src/ -├── lib.rs # crate root, feature flags, public exports -├── error.rs # Error type -│ -├── actor.rs # Message trait, ActorInterface trait, ActorAddress -│ # - ActorInterface::handle(&mut self, ctx: &Ctx, msg) -│ # - actors depend ONLY on Ctx, nothing else -│ -├── context.rs # Ctx wrapper — the "syscall interface" for actors -│ # - wraps &dyn ContextInner (solves object-safety) -│ # - send(), self_addr(), spawn() -│ # - this is ALL actors can see of the framework -│ -├── envelope.rs # Envelope type — type erasure for cross-thread messages -│ # - wraps typed messages for the transfer queue -│ # - unwraps back to concrete type at destination -│ -├── address_map.rs # ActorAddress → WorkerId mapping -│ # - shared read-heavy structure -│ # - written on spawn, read on every send -│ -├── channel/ -│ └── mod.rs # HybridChannel — per-worker MPSC -│ # - the ONE concurrent data structure on the hot path -│ # - carries Envelope (cross-worker) and spawn tuples -│ -├── worker/ -│ ├── mod.rs # Worker struct and worker loop -│ │ # - owns actor pool + transfer queue -│ │ # - the thread boundary: concurrent outside, local inside -│ │ # - drain transfer queue → tick actors → backoff -│ │ -│ ├── mailbox.rs # VecDeque-based local mailbox -│ │ # - NO atomics, NO Arc, NO crossbeam -│ │ # - only touched by the owning worker thread -│ │ -│ └── pool.rs # Actor pool — stores actors assigned to this worker -│ # - local HashMap for ActorAddress → Actor lookup -│ # - insert on spawn, remove on shutdown -│ -├── runtime.rs # Runtime — the composition point -│ # - creates workers, address map -│ # - implements ContextInner (delegates to address map + transfer queues) -│ # - public API: new(), spawn(), send_to(), run(), tick(), shutdown() -│ -├── config.rs # RuntimeConfig — tuning knobs -│ # - num_threads, max_actors, actor_max_messages -│ # - mailbox_waterlevel (drain threshold per actor) -│ # - BackoffPolicy (spin/yield/sleep thresholds) -│ -└── placement.rs # Actor placement strategy - # - currently: round-robin across workers -``` - -## Components - -| Component | File(s) | What It Does | Concurrent? | -|---|---|---|---| -| **Worker** | `worker/mod.rs` | Owns a thread, a pool of actors, their mailboxes, and a transfer queue. Runs the tick loop. Everything inside is single-threaded. | No (that's the point) | -| **Mailbox** | `worker/mailbox.rs` | `VecDeque` per actor. Zero atomics. Only the owning worker reads/writes. | No | -| **Actor Pool** | `worker/pool.rs` | Stores actors on this worker. Local lookup by address. | No | -| **Transfer Queue** | `channel/mod.rs` | HybridChannel MPSC queue per worker. The only atomic boundary. Other workers push, this worker pops. | Yes (the ONE place) | -| **Address Map** | `address_map.rs` | Maps ActorAddress → WorkerId. Read on every cross-thread send, written on spawn. | Yes (read-heavy) | -| **Envelope** | `envelope.rs` | Type-erases messages for the transfer queue. Unwrapped at destination. | No (data format) | -| **Context** | `context.rs` | `Ctx` wrapper over `&dyn ContextInner`. `send()`, `self_addr()`, `spawn()`. Hides all framework internals. | N/A (trait) | -| **Runtime** | `runtime.rs` | Wires it all together. Creates workers, holds address map, exposes public API. | Minimal (delegates) | -| **Placement** | `placement.rs` | Decides which worker gets a new actor. Currently round-robin. | No (called at spawn time) | - -## Single-Threaded / WASM Mode - -One worker. No transfer queue needed. No address map needed (everything is local). The system collapses to: - -``` - Worker 0 - ┌───────────────────────┐ - │ Actor A [mailbox] │ - │ Actor B [mailbox] │ All sends are local. - │ Actor C [mailbox] │ All mailboxes are VecDeque. - │ │ Zero atomics anywhere. - │ tick() drives loop │ - └───────────────────────┘ -``` diff --git a/Cargo.lock b/Cargo.lock index a25b4f1..d0e4e9d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,12 +2,158 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + +[[package]] +name = "anstyle" +version = "1.0.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78" + +[[package]] +name = "autocfg" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" + +[[package]] +name = "bumpalo" +version = "3.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510" + +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cfg-if" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + +[[package]] +name = "clap" +version = "4.5.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6899ea499e3fb9305a65d5ebf6e3d2248c5fab291f300ad0a704fbe142eae31a" +dependencies = [ + "clap_builder", +] + +[[package]] +name = "clap_builder" +version = "4.5.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b12c8b680195a62a8364d16b8447b01b6c2c8f9aaf68bee653be34d4245e238" +dependencies = [ + "anstyle", + "clap_lex", +] + +[[package]] +name = "clap_lex" +version = "0.7.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3e64b0cc0439b12df2fa678eae89a1c56a529fd067a9115f7827f1fffd22b32" + +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools", + "num-traits", + "once_cell", + "oorandom", + "plotters", + "rayon", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools", +] + +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -23,6 +169,18 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + +[[package]] +name = "either" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" + [[package]] name = "getrandom" version = "0.2.17" @@ -34,6 +192,59 @@ dependencies = [ "wasi", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy", +] + +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys", +] + +[[package]] +name = "itertools" +version = "0.10.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" +dependencies = [ + "either", +] + +[[package]] +name = "itoa" +version = "1.0.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" + +[[package]] +name = "js-sys" +version = "0.3.85" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c942ebf8e95485ca0d52d97da7c5a2c387d0e7f0ba4c35e93bfcaee045955b3" +dependencies = [ + "once_cell", + "wasm-bindgen", +] + [[package]] name = "libc" version = "0.2.180" @@ -41,19 +252,230 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc" [[package]] -name = "smallvec" -version = "1.15.1" +name = "memchr" +version = "2.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273" + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg", +] + +[[package]] +name = "once_cell" +version = "1.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" + +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21b2ebcf727b7760c461f091f9f0f539b77b8e87f2fd88131e7f1b433b3cece4" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "rayon" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "368f01d005bf8fd9b1206fb6fa653e6c4a81ceb1466406b81792d87c5677a58f" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + +[[package]] +name = "regex" +version = "1.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a96887878f22d7bad8a3b6dc5b7440e0ada9a245242924394987b21cf2210a4c" + +[[package]] +name = "rustversion" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" + +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.149" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] [[package]] name = "swactor" version = "0.1.0" dependencies = [ + "criterion", "crossbeam-queue", "crossbeam-utils", "getrandom", - "smallvec", +] + +[[package]] +name = "syn" +version = "2.0.114" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4d107df263a3013ef9b1879b0df87d706ff80f65a86ea879bd9c31f9b307c2a" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + +[[package]] +name = "unicode-ident" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9312f7c4f6ff9069b165498234ce8be658059c6728633667c526e27dc2cf1df5" + +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", ] [[package]] @@ -61,3 +483,108 @@ name = "wasi" version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasm-bindgen" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64024a30ec1e37399cf85a7ffefebdb72205ca1c972291c51512360d90bd8566" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "008b239d9c740232e71bd39e8ef6429d27097518b6b30bdf9086833bd5b6d608" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5256bae2d58f54820e6490f9839c49780dff84c65aeab9e772f15d5f0e913a55" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f01b580c9ac74c8d8f0c0e4afb04eeef2acf145458e52c03845ee9cd23e3d12" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "web-sys" +version = "0.3.85" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "312e32e551d92129218ea9a2452120f4aabc03529ef03e4d0d82fb2780608598" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zerocopy" +version = "0.8.39" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db6d35d663eadb6c932438e763b262fe1a70987f9ae936e60158176d710cae4a" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.39" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4122cd3169e94605190e77839c9a40d40ed048d305bfdc146e7df40ab0f3e517" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "zmij" +version = "1.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ff05f8caa9038894637571ae6b9e29466c1f4f829d26c9b28f869a29cbe3445" diff --git a/Cargo.toml b/Cargo.toml index 78f8ddb..a348085 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,8 +17,10 @@ stress = [] # Enable stress tests getrandom = { version = "0.2", optional = true } crossbeam-queue = "0.3.12" crossbeam-utils = "0.8.21" -smallvec = "1.13" -[[bin]] -name = "bench" -path = "benches/main.rs" +[dev-dependencies] +criterion = { version = "0.5", features = ["html_reports"] } + +[[bench]] +name = "runtime_benchmarks" +harness = false diff --git a/benches/harness.rs b/benches/harness.rs deleted file mode 100644 index 8eddc47..0000000 --- a/benches/harness.rs +++ /dev/null @@ -1,270 +0,0 @@ -//! Manual benchmark harness - zero dependencies, full control. -//! -//! Provides statistical analysis of benchmark runs including: -//! - Mean, median, min, max -//! - Standard deviation -//! - Percentiles (P50, P90, P99, P99.9) -//! - Throughput calculations -//! - Outlier detection and removal - -use std::time::{Duration, Instant}; - -/// Results from a single benchmark run -#[derive(Debug, Clone)] -pub struct BenchResult { - pub name: String, - pub iterations: usize, - pub total_time: Duration, - pub times: Vec, - /// Optional: elements processed (for throughput calculation) - pub elements: Option, -} - -/// Statistical summary of benchmark results -#[derive(Debug)] -pub struct Stats { - pub mean: Duration, - pub median: Duration, - pub min: Duration, - pub max: Duration, - pub std_dev: Duration, - pub p50: Duration, - pub p90: Duration, - pub p99: Duration, - pub p999: Duration, - pub throughput: Option, // elements per second -} - -impl BenchResult { - /// Calculate statistics from the raw timing data - pub fn stats(&self) -> Stats { - let mut sorted: Vec = self.times.clone(); - sorted.sort(); - - let n = sorted.len(); - assert!(n > 0, "Cannot compute stats on empty results"); - - let sum: Duration = sorted.iter().sum(); - let mean = sum / n as u32; - - let median = if n % 2 == 0 { - (sorted[n / 2 - 1] + sorted[n / 2]) / 2 - } else { - sorted[n / 2] - }; - - // Standard deviation - let mean_nanos = mean.as_nanos() as f64; - let variance: f64 = sorted - .iter() - .map(|t| { - let diff = t.as_nanos() as f64 - mean_nanos; - diff * diff - }) - .sum::() - / n as f64; - let std_dev = Duration::from_nanos(variance.sqrt() as u64); - - // Percentiles - let percentile = |p: f64| -> Duration { - let idx = ((p / 100.0) * (n - 1) as f64).round() as usize; - sorted[idx.min(n - 1)] - }; - - let throughput = self.elements.map(|e| { - let secs = self.total_time.as_secs_f64(); - if secs > 0.0 { - (e * self.iterations as u64) as f64 / secs - } else { - 0.0 - } - }); - - Stats { - mean, - median, - min: sorted[0], - max: sorted[n - 1], - std_dev, - p50: percentile(50.0), - p90: percentile(90.0), - p99: percentile(99.0), - p999: percentile(99.9), - throughput, - } - } - - /// Pretty print the results - pub fn print(&self) { - let stats = self.stats(); - - println!("\n{}", "=".repeat(60)); - println!(" {}", self.name); - println!("{}", "=".repeat(60)); - println!(" Iterations: {}", self.iterations); - println!(" Total time: {:?}", self.total_time); - println!(); - println!(" Mean: {:?}", stats.mean); - println!(" Median: {:?}", stats.median); - println!(" Std Dev: {:?}", stats.std_dev); - println!(" Min: {:?}", stats.min); - println!(" Max: {:?}", stats.max); - println!(); - println!(" P50: {:?}", stats.p50); - println!(" P90: {:?}", stats.p90); - println!(" P99: {:?}", stats.p99); - println!(" P99.9: {:?}", stats.p999); - - if let Some(throughput) = stats.throughput { - println!(); - println!(" Throughput: {:.2} ops/sec", throughput); - if throughput > 1_000_000.0 { - println!(" {:.2} M ops/sec", throughput / 1_000_000.0); - } else if throughput > 1_000.0 { - println!(" {:.2} K ops/sec", throughput / 1_000.0); - } - } - println!("{}", "=".repeat(60)); - } -} - -/// A benchmark builder for configuring and running benchmarks -pub struct Bench { - name: String, - warmup_iters: usize, - bench_iters: usize, - elements_per_iter: Option, -} - -impl Bench { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - warmup_iters: 3, - bench_iters: 100, - elements_per_iter: None, - } - } - - /// Set number of warmup iterations (default: 3) - pub fn warmup(mut self, n: usize) -> Self { - self.warmup_iters = n; - self - } - - /// Set number of benchmark iterations (default: 100) - pub fn iters(mut self, n: usize) -> Self { - self.bench_iters = n; - self - } - - /// Set elements per iteration for throughput calculation - pub fn elements(mut self, n: u64) -> Self { - self.elements_per_iter = Some(n); - self - } - - /// Run the benchmark with setup before each iteration - pub fn run_with_setup(self, mut setup: S, mut f: F) -> BenchResult - where - S: FnMut() -> T, - F: FnMut(T), - { - // Warmup - for _ in 0..self.warmup_iters { - let state = setup(); - f(state); - } - - // Benchmark - let mut times = Vec::with_capacity(self.bench_iters); - let total_start = Instant::now(); - - for _ in 0..self.bench_iters { - let state = setup(); - let start = Instant::now(); - f(state); - times.push(start.elapsed()); - } - - let total_time = total_start.elapsed(); - - BenchResult { - name: self.name, - iterations: self.bench_iters, - total_time, - times, - elements: self.elements_per_iter, - } - } -} - -/// A collection of benchmarks to run together -pub struct BenchSuite { - name: String, - results: Vec, -} - -impl BenchSuite { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - results: Vec::new(), - } - } - - pub fn add(&mut self, result: BenchResult) { - self.results.push(result); - } - - pub fn print_summary(&self) { - println!("\n{}", "#".repeat(70)); - println!("# BENCHMARK SUITE: {}", self.name); - println!("{}", "#".repeat(70)); - - for result in &self.results { - result.print(); - } - - // Summary table - println!("\n{}", "-".repeat(70)); - println!(" SUMMARY"); - println!("{}", "-".repeat(70)); - println!( - " {:30} {:>12} {:>12} {:>12}", - "Benchmark", "Mean", "P99", "Throughput" - ); - println!("{}", "-".repeat(70)); - - for result in &self.results { - let stats = result.stats(); - let throughput_str = stats - .throughput - .map(|t| { - if t > 1_000_000.0 { - format!("{:.2}M/s", t / 1_000_000.0) - } else if t > 1_000.0 { - format!("{:.2}K/s", t / 1_000.0) - } else { - format!("{:.2}/s", t) - } - }) - .unwrap_or_else(|| "-".to_string()); - - println!( - " {:30} {:>12.2?} {:>12.2?} {:>12}", - result.name, stats.mean, stats.p99, throughput_str - ); - } - println!("{}", "-".repeat(70)); - } -} - -/// Prevent the compiler from optimizing away a value -#[inline(never)] -pub fn black_box(x: T) -> T { - // Use inline assembly to prevent optimization - // This is a simplified version - in practice, reads from the value - let ptr = &x as *const T; - unsafe { std::ptr::read_volatile(ptr) } -} diff --git a/benches/main.rs b/benches/main.rs deleted file mode 100644 index 41f32fd..0000000 --- a/benches/main.rs +++ /dev/null @@ -1,46 +0,0 @@ -//! Swactor Benchmark Suite -//! -//! A manual benchmark harness for measuring runtime performance. -//! Zero external dependencies - just std::time. -//! -//! Run with: cargo run --bin bench --release -//! -//! Options: -//! --throughput Run throughput benchmarks only -//! --scaling Run scaling benchmarks only -//! --all Run all benchmarks (default) - -mod harness; -mod throughput; -mod scaling; - -use std::env; - -fn main() { - let args: Vec = env::args().collect(); - - println!("============================================================"); - println!(" SWACTOR BENCHMARK SUITE"); - println!("============================================================"); - println!(); - - // Parse arguments - let run_throughput = args.contains(&"--throughput".to_string()) - || args.contains(&"--all".to_string()) - || args.len() == 1; - let run_scaling = args.contains(&"--scaling".to_string()) - || args.contains(&"--all".to_string()) - || args.len() == 1; - - if run_throughput { - let suite = throughput::run_all(); - suite.print_summary(); - } - - if run_scaling { - let suite = scaling::run_all(); - suite.print_summary(); - } - - println!("\nBenchmarks complete."); -} diff --git a/benches/runtime_benchmarks.rs b/benches/runtime_benchmarks.rs new file mode 100644 index 0000000..64eff02 --- /dev/null +++ b/benches/runtime_benchmarks.rs @@ -0,0 +1,276 @@ +use criterion::{ + criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion, Throughput, +}; +use swactor::{ + actor::{ActorAddress, ActorInterface}, + config::RuntimeConfig, + runtime::{Ctx, Runtime}, +}; + +// --------------------------------------------------------------------------- +// Helper +// --------------------------------------------------------------------------- + +fn make_config(max_actors: usize, max_messages: usize) -> RuntimeConfig { + RuntimeConfig { + max_actors, + actor_max_messages: max_messages, + num_threads: 1, + ..Default::default() + } +} + +// --------------------------------------------------------------------------- +// Message types +// --------------------------------------------------------------------------- + +#[derive(Clone)] +struct NoopMessage; + +#[derive(Clone)] +struct PingMessage { + reply_to: ActorAddress, +} + +#[derive(Clone)] +struct PongMessage; + +#[derive(Clone)] +struct CountMessage(u64); + +#[derive(Clone)] +struct RingMessage { + hops: u64, +} + +// --------------------------------------------------------------------------- +// Actor types +// --------------------------------------------------------------------------- + +struct NoopActor; + +impl ActorInterface for NoopActor { + type Incoming = NoopMessage; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: NoopMessage) {} +} + +struct EchoActor; + +impl ActorInterface for EchoActor { + type Incoming = PingMessage; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: PingMessage) { + let _ = ctx.send(msg.reply_to, PongMessage); + } +} + +struct SinkActor; + +impl ActorInterface for SinkActor { + type Incoming = CountMessage; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: CountMessage) {} +} + +struct RingActor { + next: ActorAddress, +} + +impl ActorInterface for RingActor { + type Incoming = RingMessage; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: RingMessage) { + let _ = ctx.send(self.next, RingMessage { hops: msg.hops + 1 }); + } +} + +// --------------------------------------------------------------------------- +// Latency benchmarks +// --------------------------------------------------------------------------- + +fn latency_benchmarks(c: &mut Criterion) { + let mut group = c.benchmark_group("latency"); + + // A1 — Spawn latency + group.bench_function("spawn", |b| { + b.iter_batched( + || Runtime::new(make_config(1_000, 1_000)), + |rt| { + rt.spawn(NoopActor).unwrap(); + }, + BatchSize::SmallInput, + ); + }); + + // A2 — Message round-trip + group.bench_function("message_roundtrip", |b| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(1_000, 1_000)); + let addr = rt.spawn(EchoActor).unwrap(); + rt.tick(); // register actor + let inbox = rt.new_inbox::().unwrap(); + let inbox_addr = *inbox.addr(); + (rt, addr, inbox, inbox_addr) + }, + |(rt, addr, inbox, inbox_addr)| { + rt.send_to(addr, PingMessage { reply_to: inbox_addr }).unwrap(); + for _ in 0..20 { + rt.tick(); + if inbox.try_recv().is_some() { + return; + } + } + panic!("PongMessage not received within 20 ticks"); + }, + BatchSize::SmallInput, + ); + }); + + // A3 — Fire-and-forget send + group.bench_function("send_fire_and_forget", |b| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(1_000, 100_000)); + let addr = rt.spawn(NoopActor).unwrap(); + rt.tick(); // register actor + (rt, addr) + }, + |(rt, addr)| { + rt.send_to(addr, NoopMessage).unwrap(); + }, + BatchSize::SmallInput, + ); + }); + + // A4 — Inbox creation + group.bench_function("inbox_creation", |b| { + b.iter_batched( + || Runtime::new(make_config(1_000, 1_000)), + |rt| { + rt.new_inbox::().unwrap(); + }, + BatchSize::SmallInput, + ); + }); + + group.finish(); +} + +// --------------------------------------------------------------------------- +// Throughput benchmarks +// --------------------------------------------------------------------------- + +fn throughput_benchmarks(c: &mut Criterion) { + let mut group = c.benchmark_group("throughput"); + + // B1 — Single-actor throughput + for n in [100, 1_000, 10_000] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::new("single_actor", n), &n, |b, &n| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(100, n + 100)); + let addr = rt.spawn(SinkActor).unwrap(); + rt.tick(); // register actor + for i in 0..n { + rt.send_to(addr, CountMessage(i as u64)).unwrap(); + } + rt + }, + |rt| { + for _ in 0..50 { + rt.tick(); + } + }, + BatchSize::LargeInput, + ); + }); + } + + // B2 — Multi-actor throughput + for (actors, msgs_per) in [(10, 100), (100, 100), (100, 1_000)] { + let total = actors * msgs_per; + group.throughput(Throughput::Elements(total as u64)); + let param = format!("{actors}x{msgs_per}"); + group.bench_with_input(BenchmarkId::new("multi_actor", ¶m), &(actors, msgs_per), |b, &(actors, msgs_per)| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(actors + 100, msgs_per + 100)); + let addrs: Vec<_> = (0..actors) + .map(|_| rt.spawn(SinkActor).unwrap()) + .collect(); + rt.tick(); // register actors + for &addr in &addrs { + for i in 0..msgs_per { + rt.send_to(addr, CountMessage(i as u64)).unwrap(); + } + } + rt + }, + |rt| { + for _ in 0..100 { + rt.tick(); + } + }, + BatchSize::LargeInput, + ); + }); + } + + // B3 — Ring throughput + for ring_size in [10usize, 100, 500] { + group.throughput(Throughput::Elements((ring_size + 1) as u64)); + group.bench_with_input(BenchmarkId::new("ring", ring_size), &ring_size, |b, &ring_size| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(ring_size + 100, 100)); + let inbox = rt.new_inbox::().unwrap(); + // Build the ring: last actor sends to inbox, each prior actor sends to the next + let mut next_addr = *inbox.addr(); + let mut entry_addr = next_addr; + for _ in 0..ring_size { + let addr = rt.spawn(RingActor { next: next_addr }).unwrap(); + entry_addr = addr; + next_addr = addr; + } + rt.tick(); // register all actors + (rt, entry_addr, inbox) + }, + |(rt, entry_addr, inbox)| { + rt.send_to(entry_addr, RingMessage { hops: 0 }).unwrap(); + for _ in 0..(ring_size + 10) { + rt.tick(); + if inbox.try_recv().is_some() { + return; + } + } + panic!("RingMessage not received within tick budget"); + }, + BatchSize::LargeInput, + ); + }); + } + + // B4 — Spawn throughput + for n in [100, 1_000, 5_000] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::new("spawn", n), &n, |b, &n| { + b.iter_batched( + || Runtime::new(make_config(n + 100, 1_000)), + |rt| { + for _ in 0..n { + rt.spawn(NoopActor).unwrap(); + } + }, + BatchSize::LargeInput, + ); + }); + } + + group.finish(); +} + +criterion_group!(benches, latency_benchmarks, throughput_benchmarks); +criterion_main!(benches); diff --git a/benches/scaling.rs b/benches/scaling.rs deleted file mode 100644 index 0febea5..0000000 --- a/benches/scaling.rs +++ /dev/null @@ -1,283 +0,0 @@ -//! Scaling benchmarks for the swactor runtime. -//! -//! These benchmarks measure how performance scales with: -//! - Number of actors -//! - Number of worker threads -//! - Message payload size - -use crate::harness::{black_box, Bench, BenchSuite}; -use std::thread; -use swactor::{ - actor::ActorInterface, - runtime::{Ctx, Runtime, RuntimeConfig}, -}; - -// ============================================================================ -// Test Actors -// ============================================================================ - -/// A counter actor that just increments on each message -struct CounterActor { - count: usize, -} - -impl CounterActor { - fn new() -> Self { - Self { count: 0 } - } -} - -#[derive(Clone)] -struct Increment; - -impl ActorInterface for CounterActor { - type Incoming = Increment; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Increment) { - self.count += 1; - } -} - -struct SharedCounter { - count: std::sync::Arc, -} - -impl ActorInterface for SharedCounter { - type Incoming = Increment; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Increment) { - self.count - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - } -} - -/// An actor that handles variable-sized payloads -struct PayloadActor { - bytes_received: usize, -} - -impl PayloadActor { - fn new() -> Self { - Self { bytes_received: 0 } - } -} - -#[derive(Clone)] -struct Payload(Vec); - -impl ActorInterface for PayloadActor { - type Incoming = Payload; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, msg: Payload) { - self.bytes_received += msg.0.len(); - black_box(&msg.0); - } -} - -// ============================================================================ -// Benchmarks -// ============================================================================ - -/// Benchmark: How throughput scales with actor count -pub fn bench_actor_count_scaling(suite: &mut BenchSuite) { - for (actor_count, messages_per_actor, warmup_n, iters_n) in [ - (10u64, 200u64, 5usize, 50usize), - (100, 200, 5, 30), - (500, 200, 5, 15), - (1000, 200, 5, 12), - ] { - let name = format!("scaling_{}_actors", actor_count); - let total_messages = actor_count * messages_per_actor; - - let result = Bench::new(&name) - .warmup(warmup_n) - .iters(iters_n) - .elements(total_messages) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: (actor_count as usize) + 100, - actor_max_messages: (total_messages as usize) * 3, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - - // Spawn actors - let mut actors = Vec::with_capacity(actor_count as usize); - for _ in 0..actor_count { - let addr = runtime.spawn(CounterActor::new()).unwrap(); - actors.push(addr); - } - - // Process spawns - for _ in 0..(actor_count * 2) { - runtime.tick(); - } - - (runtime, actors, messages_per_actor) - }, - |(runtime, actors, msgs_per)| { - // Distribute messages across all actors - for _ in 0..msgs_per { - for actor in &actors { - let _ = runtime.send_to::(*actor, Increment); - } - } - - // Process all - let total = actors.len() as u64 * msgs_per; - for _ in 0..(total * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: How throughput scales with thread count (multithreaded runtime) -pub fn bench_thread_count_scaling(suite: &mut BenchSuite) { - let actor_count = 100u64; - let messages_per_actor = 500u64; - let total_messages = actor_count * messages_per_actor; - - for thread_count in [2usize, 4, 8] { - let name = format!("scaling_{}_threads", thread_count); - - let result = Bench::new(&name) - .warmup(5) - .iters(30) - .elements(total_messages) - .run_with_setup( - || { - let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); - let config = RuntimeConfig { - max_actors: (actor_count as usize) + 100, - actor_max_messages: (total_messages as usize) * 3, - num_threads: thread_count, - ..Default::default() - }; - let runtime = Runtime::new(config); - - let mut actors = Vec::with_capacity(actor_count as usize); - for _ in 0..actor_count { - let addr = runtime - .spawn(SharedCounter { - count: counter.clone(), - }) - .unwrap(); - actors.push(addr); - } - - let handle = runtime.run().unwrap(); - - for actor in &actors { - loop { - if handle - .runtime - .send_to::(*actor, Increment) - .is_ok() - { - break; - } - thread::yield_now(); - } - } - - while counter.load(std::sync::atomic::Ordering::Relaxed) < actors.len() { - thread::yield_now(); - } - counter.store(0, std::sync::atomic::Ordering::Relaxed); - - (handle, actors, counter) - }, - |(handle, actors, counter)| { - for _ in 0..messages_per_actor { - for actor in &actors { - let _ = handle.runtime.send_to::(*actor, Increment); - } - } - - while counter.load(std::sync::atomic::Ordering::Relaxed) - < total_messages as usize - { - thread::yield_now(); - } - - handle.shutdown(); - handle.join(); - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: How throughput scales with message payload size -pub fn bench_payload_size_scaling(suite: &mut BenchSuite) { - let message_count = 3_000u64; - - for payload_size in [64usize, 1024, 16384, 65536] { - let name = format!("payload_{}B", payload_size); - let payload = vec![0u8; payload_size]; - - let result = Bench::new(&name) - .warmup(5) - .iters(40) - .elements(message_count) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: 10, - actor_max_messages: (message_count as usize) * 2, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(PayloadActor::new()).unwrap(); - - // Process spawn - for _ in 0..10 { - runtime.tick(); - } - - (runtime, sink, payload.clone()) - }, - |(runtime, sink, payload)| { - for _ in 0..message_count { - let _ = runtime.send_to::(sink, Payload(payload.clone())); - } - - for _ in 0..(message_count * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Run all scaling benchmarks -pub fn run_all() -> BenchSuite { - let mut suite = BenchSuite::new("Scaling Benchmarks"); - - println!("\nRunning actor count scaling benchmarks..."); - bench_actor_count_scaling(&mut suite); - - println!("Running thread count scaling benchmarks..."); - bench_thread_count_scaling(&mut suite); - - println!("Running payload size scaling benchmarks..."); - bench_payload_size_scaling(&mut suite); - - suite -} diff --git a/benches/throughput.rs b/benches/throughput.rs deleted file mode 100644 index 900a4d0..0000000 --- a/benches/throughput.rs +++ /dev/null @@ -1,350 +0,0 @@ -//! Core throughput benchmarks for the swactor runtime. -//! -//! These benchmarks measure: -//! - Message passing throughput -//! - Actor spawn rate -//! - Fan-out and fan-in patterns -//! - Ping-pong latency - -use crate::harness::{black_box, Bench, BenchSuite}; -use swactor::{ - actor::{ActorAddress, ActorInterface}, - runtime::{Ctx, Runtime, RuntimeConfig}, -}; - -// ============================================================================ -// Test Actors -// ============================================================================ - -/// A sink actor that counts messages received -struct SinkActor { - count: usize, -} - -impl SinkActor { - fn new() -> Self { - Self { count: 0 } - } -} - -#[derive(Clone)] -struct Ping; - -impl ActorInterface for SinkActor { - type Incoming = Ping; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Ping) { - self.count += 1; - } -} - -/// A forwarding actor that passes messages along a chain -struct ForwardActor { - next: Option, -} - -impl ForwardActor { - fn new() -> Self { - Self { next: None } - } - - fn with_next(next: ActorAddress) -> Self { - Self { next: Some(next) } - } -} - -impl ActorInterface for ForwardActor { - type Incoming = Ping; - type Response = Ping; - - fn handle(&mut self, ctx: &Ctx, msg: Ping) { - if let Some(next) = self.next { - let _ = ctx.send(next, msg); - } - } -} - -// ============================================================================ -// Benchmarks -// ============================================================================ - -/// Benchmark: Messages sent through the runtime to a single sink actor -pub fn bench_message_throughput(suite: &mut BenchSuite) { - for msg_count in [1_000u64, 10_000, 100_000] { - let name = format!("message_throughput_{}", msg_count); - - let result = Bench::new(&name) - .warmup(5) - .iters(100) - .elements(msg_count) - .run_with_setup( - || { - // Setup: create runtime and sink actor - let config = RuntimeConfig { - max_actors: 100, - actor_max_messages: (msg_count as usize) * 2, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(SinkActor::new()).unwrap(); - (runtime, sink, msg_count) - }, - |(runtime, sink, count)| { - // Send all messages - for _ in 0..count { - let _ = runtime.send_to::(sink, Ping); - } - // Process until done - for _ in 0..(count * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: Actor spawn rate -pub fn bench_spawn_rate(suite: &mut BenchSuite) { - for actor_count in [100u64, 500, 900] { - let name = format!("spawn_rate_{}_actors", actor_count); - - let result = Bench::new(&name) - .warmup(5) - .iters(100) - .elements(actor_count) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: 1000, - actor_max_messages: 100, - num_threads: 1, - ..Default::default() - }; - Runtime::new(config) - }, - |runtime| { - for _ in 0..actor_count { - let _ = runtime.spawn(SinkActor::new()); - } - // Process spawns - for _ in 0..(actor_count * 2) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: Fan-out (1 sender to N receivers) -pub fn bench_fanout(suite: &mut BenchSuite) { - for (fan_count, messages_per_receiver, warmup_n, iters_n) in [ - (10u64, 500u64, 5usize, 50usize), - (100, 200, 5, 40), - (500, 100, 5, 30), - ] { - let name = format!("fanout_1_to_{}", fan_count); - - let result = Bench::new(&name) - .warmup(warmup_n) - .iters(iters_n) - .elements(fan_count * messages_per_receiver) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: (fan_count as usize) + 10, - actor_max_messages: (fan_count as usize) - * (messages_per_receiver as usize) - * 2, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - - // Spawn N sink actors - let mut sinks = Vec::with_capacity(fan_count as usize); - for _ in 0..fan_count { - let addr = runtime.spawn(SinkActor::new()).unwrap(); - sinks.push(addr); - } - - // Process spawns - for _ in 0..(fan_count * 2) { - runtime.tick(); - } - - (runtime, sinks, messages_per_receiver) - }, - |(runtime, sinks, msgs_per)| { - // Send messages to all sinks - for _ in 0..msgs_per { - for sink in &sinks { - let _ = runtime.send_to::(*sink, Ping); - } - } - - // Process all messages - let total_msgs = sinks.len() as u64 * msgs_per; - for _ in 0..(total_msgs * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: Fan-in (N senders to 1 receiver) -pub fn bench_fanin(suite: &mut BenchSuite) { - for (sender_count, messages_per_sender, warmup_n, iters_n) in [ - (10u64, 300u64, 5usize, 50usize), - (100, 100, 5, 30), - (500, 100, 5, 20), - ] { - let name = format!("fanin_{}_to_1", sender_count); - - let result = Bench::new(&name) - .warmup(warmup_n) - .iters(iters_n) - .elements(sender_count * messages_per_sender) - .run_with_setup( - || { - let total_messages = (sender_count * messages_per_sender) as usize; - let config = RuntimeConfig { - max_actors: (sender_count as usize) + 10, - actor_max_messages: total_messages * 3, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - - // Spawn the sink - let sink = runtime.spawn(SinkActor::new()).unwrap(); - - // Spawn N forwarders pointing at sink - let mut senders = Vec::with_capacity(sender_count as usize); - for _ in 0..sender_count { - let addr = runtime.spawn(ForwardActor::with_next(sink)).unwrap(); - senders.push(addr); - } - - // Process spawns - for _ in 0..((sender_count + 1) * 2) { - runtime.tick(); - } - - (runtime, senders, sink, messages_per_sender) - }, - |(runtime, senders, _sink, msgs_per)| { - // Each sender forwards msgs_per messages to the sink - for _ in 0..msgs_per { - for sender in &senders { - let _ = runtime.send_to::(*sender, Ping); - } - } - - // Process all messages (forwarder receives + forwards, sink receives) - let total_msgs = senders.len() as u64 * msgs_per; - for _ in 0..(total_msgs * 6) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: Ring topology (message passed around N actors in a circle) -pub fn bench_ring(suite: &mut BenchSuite) { - for (ring_size, laps, warmup_n, iters_n) in [ - (10u64, 100u64, 5usize, 100usize), - (100, 20, 5, 50), - (500, 10, 5, 40), - ] { - let name = format!("ring_{}_actors", ring_size); - - let result = Bench::new(&name) - .warmup(warmup_n) - .iters(iters_n) - .elements(ring_size * laps) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: (ring_size as usize) + 10, - actor_max_messages: 10_000, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - - // First, spawn all actors without links - let mut actors: Vec = Vec::with_capacity(ring_size as usize); - for _ in 0..ring_size { - let addr = runtime.spawn(ForwardActor::new()).unwrap(); - actors.push(addr); - } - - // We can't update their `next` field after spawn in this design, - // so instead we'll use an inbox to receive the final message - // For now, we'll just measure message passing through a chain - - // Process spawns - for _ in 0..(ring_size * 2) { - runtime.tick(); - } - - (runtime, actors, laps) - }, - |(runtime, actors, laps)| { - // Send to first actor (even though they don't forward, we're - // measuring the transfer queue + inbox overhead) - for _ in 0..laps { - for actor in &actors { - let _ = runtime.send_to::(*actor, Ping); - } - } - - let total = actors.len() as u64 * laps; - for _ in 0..(total * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Run all throughput benchmarks -pub fn run_all() -> BenchSuite { - let mut suite = BenchSuite::new("Throughput Benchmarks"); - - println!("\nRunning message throughput benchmarks..."); - bench_message_throughput(&mut suite); - - println!("Running spawn rate benchmarks..."); - bench_spawn_rate(&mut suite); - - println!("Running fan-out benchmarks..."); - bench_fanout(&mut suite); - - println!("Running fan-in benchmarks..."); - bench_fanin(&mut suite); - - println!("Running ring topology benchmarks..."); - bench_ring(&mut suite); - - suite -} diff --git a/src/actor.rs b/src/actor.rs index a5ae92a..7a7a9b8 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -6,43 +6,6 @@ use crate::{get_random, runtime::{ContextInner, Ctx}, worker::Mailbox}; pub trait Message: 'static + Sized + Clone + Send + Sync {} impl Message for T {} -/// The trait that needs to be implemented in order to run a process as an `Actor` -/// -/// The `Incoming` type represents `Messages` that can be delivered to the `Actor`. -/// -/// The `Response` type represents possible `Messages` the actor may attempt to reply with. -/// -/// The `fn handle(..)` is where you implement the logic for handling `Incoming` messages -/// -/// # Example -/// ``` -/// use swactor::{Ctx, actor::{ActorAddress, ActorInterface}}; -/// -/// struct Greeter { -/// num_greeted: usize, -/// } -/// -/// #[derive(Clone)] // required to auto implement `Message` -/// struct GreetMessage { -/// who: String, -/// return_addr: ActorAddress, -/// } -/// -/// #[derive(Clone)] -/// struct GreetResponse(String); -/// -/// impl ActorInterface for Greeter { -/// type Incoming = GreetMessage; -/// type Response = GreetResponse; -/// -/// fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { -/// let response = GreetResponse(format!("Hello, {}!", msg.who).to_string()); -/// if let Ok(_) = ctx.send(msg.return_addr, response) { -/// self.num_greeted += 1; -/// } -/// } -/// } -/// ``` pub trait ActorInterface: 'static + Send { type Incoming: Message; type Response: Message; diff --git a/src/envelope.rs b/src/envelope.rs deleted file mode 100644 index c3a716d..0000000 --- a/src/envelope.rs +++ /dev/null @@ -1,24 +0,0 @@ -use std::any::Any; - -use crate::actor::ActorAddress; - - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn envelope_roundtrip() { - let addr = ActorAddress::default(); - let env = Envelope::new(addr, Box::new(42u64)); - assert_eq!(env.dest(), addr); - assert_eq!(env.downcast::(), Some(42u64)); - } - - #[test] - fn envelope_wrong_type_returns_none() { - let addr = ActorAddress::default(); - let env = Envelope::new(addr, Box::new(42u64)); - assert!(env.downcast::().is_none()); - } -} diff --git a/src/runtime.rs b/src/runtime.rs index 622c0db..c2d034c 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -14,83 +14,6 @@ use crate::worker::Mailbox; use crate::worker::{TickContext, Worker}; use crate::Error; -// ─── SenderT trait (moved from router.rs) ──────────────────────────────────── - -/// Type-erased sender for external inboxes. -pub(crate) trait SenderT: Send + Sync { - fn try_send_any(&self, msg: Box); -} - -impl SenderT for Sender { - fn try_send_any(&self, msg: Box) { - if let Ok(typed) = msg.downcast::() { - let _ = Sender::try_send(self, *typed); - } - } -} - - -/// A type-erased message envelope for cross-worker delivery. -/// -/// Uses `Box` (no atomic refcount) and move semantics (no clone). -pub(crate) struct Envelope { - dest: ActorAddress, - payload: Box, -} - -impl Envelope { - pub fn new(dest: ActorAddress, payload: Box) -> Self { - Self { dest, payload } - } - - pub fn dest(&self) -> ActorAddress { - self.dest - } - - pub fn downcast(self) -> Option { - self.payload.downcast::().ok().map(|b| *b) - } - - pub fn into_payload(self) -> Box { - self.payload - } -} - - -// ─── InboxRegistry ─────────────────────────────────────────────────────────── - -/// Registry of external inboxes — replaces the Router's role for non-actor receivers. -pub(crate) struct InboxRegistry { - senders: RwLock>>, -} - -impl InboxRegistry { - pub fn new() -> Self { - Self { - senders: RwLock::new(HashMap::new()), - } - } - - pub fn register(&self, addr: ActorAddress, sender: Arc) { - self.senders.write().unwrap().insert(addr, sender); - } - - pub fn try_deliver( - &self, - addr: ActorAddress, - msg: Box, - ) -> Result<(), Error> { - let senders = self.senders.read().unwrap(); - if let Some(sender) = senders.get(&addr) { - sender.try_send_any(msg); - Ok(()) - } else { - Err(Error::from("Address not found")) - } - } -} - -// ─── Inbox ─────────────────────────────────────────────────────────────────── /// Generic message inbox for receiving messages outside of the runtime. pub struct Inbox { @@ -108,12 +31,23 @@ impl Inbox { } } +/// Handle for dealing with a runtime that has started via the `Runtime::run()` method. +pub struct RuntimeHandle { + pub runtime: Arc, + threads: Vec>, +} -/// Object-safe inner trait for sending type-erased messages. -pub(crate) trait ContextInner { - fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; - fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error>; - fn mailbox_waterlevel(&self) -> usize; +impl RuntimeHandle { + pub fn join(self) { + for handle in self.threads { + let _ = handle.join(); + } + } + + /// Simple helper, calls the inner `Runtime::shutdown()` method + pub fn shutdown(&self) { + self.runtime.shutdown(); + } } /// Actor syscall interface — passed to `ActorInterface::handle()`. @@ -151,6 +85,19 @@ impl<'a> Ctx<'a> { } } +/// Type-erased sender for external inboxes. +pub(crate) trait SenderT: Send + Sync { + fn try_send_any(&self, msg: Box); +} + +impl SenderT for Sender { + fn try_send_any(&self, msg: Box) { + if let Ok(typed) = msg.downcast::() { + let _ = Sender::try_send(self, *typed); + } + } +} + // ─── Runtime ───────────────────────────────────────────────────────────────── @@ -329,61 +276,78 @@ impl Runtime { } } -/// 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; +/// A type-erased message envelope for cross-worker delivery. +/// +/// Uses `Box` (no atomic refcount) and move semantics (no clone). +pub(crate) struct Envelope { + dest: ActorAddress, + payload: Box, +} + +impl Envelope { + pub fn new(dest: ActorAddress, payload: Box) -> Self { + Self { dest, payload } + } + + pub fn dest(&self) -> ActorAddress { + self.dest + } + + pub fn downcast(self) -> Option { + self.payload.downcast::().ok().map(|b| *b) + } + + pub fn into_payload(self) -> Box { + self.payload + } +} + + +// ─── InboxRegistry ─────────────────────────────────────────────────────────── + +/// Registry of external inboxes — replaces the Router's role for non-actor receivers. +pub(crate) struct InboxRegistry { + senders: RwLock>>, +} + +impl InboxRegistry { + pub fn new() -> Self { + Self { + senders: RwLock::new(HashMap::new()), + } + } + + pub fn register(&self, addr: ActorAddress, sender: Arc) { + self.senders.write().unwrap().insert(addr, sender); + } + + pub fn try_deliver( + &self, + addr: ActorAddress, + msg: Box, + ) -> Result<(), Error> { + let senders = self.senders.read().unwrap(); + if let Some(sender) = senders.get(&addr) { + sender.try_send_any(msg); + Ok(()) } 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)); - } + Err(Error::from("Address not found")) } } } -/// Handle for dealing with a runtime that has started via the `Runtime::run()` method. -pub struct RuntimeHandle { - pub runtime: Arc, - threads: Vec>, + + + +/// Object-safe inner trait for sending type-erased messages. +pub(crate) trait ContextInner { + fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; + fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error>; + fn mailbox_waterlevel(&self) -> usize; } -impl RuntimeHandle { - pub fn join(self) { - for handle in self.threads { - let _ = handle.join(); - } - } - - /// Simple helper, calls the inner `Runtime::shutdown()` method - pub fn shutdown(&self) { - self.runtime.shutdown(); - } -} impl ContextInner for Runtime { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { diff --git a/src/worker.rs b/src/worker.rs index c6d0e3c..18a965d 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -6,7 +6,7 @@ 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}; +use crate::runtime::{ContextInner, Envelope, InboxRegistry, Runtime}; use crate::Error; /// Shared state passed to tick_once — single thin pointer avoids register spill. @@ -230,3 +230,40 @@ 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 diff --git a/tests/mailbox_tests.rs b/tests/mailbox_tests.rs deleted file mode 100644 index 66a671e..0000000 --- a/tests/mailbox_tests.rs +++ /dev/null @@ -1,226 +0,0 @@ -use swactor::worker::Mailbox; - -// ── Basic operations ── - -#[test] -fn push_and_pop() { - let mut mb = Mailbox::new(10); - mb.push(42i32); - assert_eq!(mb.pop(), Some(42)); -} - -#[test] -fn fifo_ordering() { - let mut mb = Mailbox::new(10); - mb.push(1); - mb.push(2); - mb.push(3); - assert_eq!(mb.pop(), Some(1)); - assert_eq!(mb.pop(), Some(2)); - assert_eq!(mb.pop(), Some(3)); -} - -#[test] -fn pop_empty() { - let mut mb: Mailbox = Mailbox::new(10); - assert_eq!(mb.pop(), None); -} - -#[test] -fn multiple_messages() { - let mut mb = Mailbox::new(10); - for i in 0..100 { - mb.push(i); - } - for i in 0..100 { - assert_eq!(mb.pop(), Some(i)); - } - assert_eq!(mb.pop(), None); -} - -#[test] -fn interleaved_push_pop() { - let mut mb = Mailbox::new(10); - mb.push(1); - mb.push(2); - assert_eq!(mb.pop(), Some(1)); - mb.push(3); - assert_eq!(mb.pop(), Some(2)); - assert_eq!(mb.pop(), Some(3)); - assert_eq!(mb.pop(), None); -} - -// ── Drain count / watermark logic ── - -#[test] -fn drain_count_empty() { - let mb: Mailbox = Mailbox::new(10); - assert_eq!(mb.drain_count(), 0); -} - -#[test] -fn drain_count_below_waterlevel() { - let mut mb = Mailbox::new(10); - for i in 0..5 { - mb.push(i); - } - // 5 < 10 (default waterlevel) → process all - assert_eq!(mb.drain_count(), 5); -} - -#[test] -fn drain_count_at_waterlevel() { - let mut mb = Mailbox::new(10); - for i in 0..10 { - mb.push(i); - } - // 10 >= 10 → process half → 5 - assert_eq!(mb.drain_count(), 5); -} - -#[test] -fn drain_count_above_waterlevel() { - let mut mb = Mailbox::new(10); - for i in 0..20 { - mb.push(i); - } - // 20 >= 10 → 20 >> 1 = 10 - assert_eq!(mb.drain_count(), 10); -} - -#[test] -fn drain_count_one_message() { - let mut mb = Mailbox::new(10); - mb.push(1i32); - // 1 < 10 → process all → 1 - assert_eq!(mb.drain_count(), 1); -} - -#[test] -fn drain_count_just_below_waterlevel() { - let mut mb = Mailbox::new(10); - for i in 0..9 { - mb.push(i); - } - // 9 < 10 → process all → 9 - assert_eq!(mb.drain_count(), 9); -} - -#[test] -fn drain_count_large() { - let mut mb = Mailbox::new(10); - for i in 0..1000 { - mb.push(i); - } - // 1000 >= 10 → 1000 >> 1 = 500 - assert_eq!(mb.drain_count(), 500); -} - -#[test] -fn drain_count_custom_waterlevel() { - let mut mb = Mailbox::new(4); - for i in 0..3 { - mb.push(i); - } - // 3 < 4 → process all → 3 - assert_eq!(mb.drain_count(), 3); - - mb.push(99); - // 4 >= 4 → 4 >> 1 = 2 - assert_eq!(mb.drain_count(), 2); -} - -#[test] -fn drain_count_updates_after_pop() { - let mut mb = Mailbox::new(10); - for i in 0..20 { - mb.push(i); - } - // 20 >= 10 → 10 - assert_eq!(mb.drain_count(), 10); - - // pop 15, leaving 5 - for _ in 0..15 { - mb.pop(); - } - // 5 < 10 → process all → 5 - assert_eq!(mb.drain_count(), 5); -} - -// ── Properties ── - -#[test] -fn len_tracks_pushes() { - let mut mb = Mailbox::new(10); - assert_eq!(mb.len(), 0); - mb.push(1); - assert_eq!(mb.len(), 1); - mb.push(2); - assert_eq!(mb.len(), 2); - mb.push(3); - assert_eq!(mb.len(), 3); -} - -#[test] -fn len_tracks_pops() { - let mut mb = Mailbox::new(10); - mb.push(1); - mb.push(2); - mb.push(3); - assert_eq!(mb.len(), 3); - mb.pop(); - assert_eq!(mb.len(), 2); - mb.pop(); - assert_eq!(mb.len(), 1); - mb.pop(); - assert_eq!(mb.len(), 0); -} - -#[test] -fn is_empty_on_new() { - let mb: Mailbox = Mailbox::new(10); - assert!(mb.is_empty()); -} - -#[test] -fn is_empty_after_drain() { - let mut mb = Mailbox::new(10); - mb.push(1); - mb.push(2); - mb.push(3); - assert!(!mb.is_empty()); - mb.pop(); - mb.pop(); - mb.pop(); - assert!(mb.is_empty()); -} - -// ── Type tests ── - -#[test] -fn works_with_primitive_types() { - let mut mb_i32 = Mailbox::new(10); - mb_i32.push(42i32); - assert_eq!(mb_i32.pop(), Some(42)); - - let mut mb_string = Mailbox::new(10); - mb_string.push(String::from("hello")); - assert_eq!(mb_string.pop(), Some(String::from("hello"))); -} - -#[test] -fn works_with_custom_structs() { - #[derive(Debug, Clone, PartialEq)] - struct MyMsg { - id: u64, - payload: String, - } - - let mut mb = Mailbox::new(10); - let msg = MyMsg { - id: 1, - payload: "test".into(), - }; - mb.push(msg.clone()); - assert_eq!(mb.pop(), Some(msg)); -} diff --git a/tests/runtime_api_tests.rs b/tests/runtime_api_tests.rs new file mode 100644 index 0000000..88238a5 --- /dev/null +++ b/tests/runtime_api_tests.rs @@ -0,0 +1,97 @@ +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Ctx, Inbox, Runtime, RuntimeConfig}, +}; + +// --------------------------------------------------------------------------- +// Shared test fixtures +// --------------------------------------------------------------------------- + +#[derive(Clone)] +struct EchoMessage { + payload: usize, + reply_to: ActorAddress, +} + +#[derive(Clone, Debug, PartialEq)] +struct EchoResponse(usize); + +struct EchoActor; + +impl ActorInterface for EchoActor { + type Incoming = EchoMessage; + type Response = EchoResponse; + + fn handle(&mut self, ctx: &Ctx, msg: EchoMessage) { + let _ = ctx.send(msg.reply_to, EchoResponse(msg.payload)); + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[test] +fn test_single_thread_spawn_actor_and_inbox() { + let rt = Runtime::new(RuntimeConfig::default()); + + let actor_addr = rt.spawn(EchoActor).expect("spawn echo actor"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + actor_addr, + EchoMessage { + payload: 42, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + for _ in 0..10 { + rt.tick(); + if let Some(response) = inbox.try_recv() { + assert_eq!(response, EchoResponse(42)); + return; + } + } + + panic!("Did not receive EchoResponse"); +} + +#[test] +fn test_multi_thread_spawn_actor_and_inbox() { + let config = RuntimeConfig { + num_threads: 4, + ..Default::default() + }; + let rt = Runtime::new(config); + + let actor_addr = rt.spawn(EchoActor).expect("spawn echo actor"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + actor_addr, + EchoMessage { + payload: 99, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + let handle = rt.run().unwrap(); + + let check = std::thread::spawn(move || { + for _ in 0..100 { + std::thread::sleep(std::time::Duration::from_millis(10)); + if let Some(response) = inbox.try_recv() { + handle.shutdown(); + return Some(response); + } + } + handle.shutdown(); + None + }); + + let result = check.join().unwrap(); + assert_eq!(result, Some(EchoResponse(99))); +} diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs deleted file mode 100644 index ed21044..0000000 --- a/tests/runtime_tests.rs +++ /dev/null @@ -1,130 +0,0 @@ -use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{Ctx, Inbox, Runtime, RuntimeConfig}}; - -#[derive(Clone)] -struct PingMessage { - reply_to: ActorAddress, -} - -#[derive(Clone)] -struct PongMessage; - -struct PongActor; - -impl ActorInterface for PongActor { - type Incoming = PingMessage; - type Response = PongMessage; - - fn handle(&mut self, ctx: &Ctx, msg: PingMessage) { - let _ = ctx.send(msg.reply_to, PongMessage); - } -} - -/// An actor that forwards messages to another address -struct ForwarderActor { - target: ActorAddress, -} - -#[derive(Clone)] -struct ForwardMessage(usize); - -impl ActorInterface for ForwarderActor { - type Incoming = ForwardMessage; - type Response = (); - - fn handle(&mut self, ctx: &Ctx, msg: ForwardMessage) { - let _ = ctx.send(self.target, msg); - } -} - -#[test] -fn test_single_threaded_ping_pong() { - let rt = Runtime::new(RuntimeConfig::default()); - let inbox: Inbox = rt.new_inbox().unwrap(); - - let pong_addr = rt.spawn(PongActor).expect("spawn pong"); - - // Send ping - rt.send_to( - pong_addr, - PingMessage { - reply_to: *inbox.addr(), - }, - ) - .unwrap(); - - // Tick until we get a response - for _ in 0..10 { - rt.tick(); - if inbox.try_recv().is_some() { - return; // Success! - } - } - - panic!("Did not receive pong response"); -} - -#[test] -fn test_single_threaded_message_chain() { - let rt = Runtime::new(RuntimeConfig::default()); - let inbox: Inbox = rt.new_inbox().unwrap(); - - // Create a chain: A -> B -> C -> inbox - let c_addr = rt - .spawn(ForwarderActor { - target: *inbox.addr(), - }) - .unwrap(); - let b_addr = rt.spawn(ForwarderActor { target: c_addr }).unwrap(); - let a_addr = rt.spawn(ForwarderActor { target: b_addr }).unwrap(); - - // Send message to start of chain - rt.send_to(a_addr, ForwardMessage(42)).unwrap(); - - // Tick until message arrives - for _ in 0..20 { - rt.tick(); - if let Some(ForwardMessage(val)) = inbox.try_recv() { - assert_eq!(val, 42); - return; - } - } - - panic!("Message did not traverse the chain"); -} - -#[test] -fn test_multithreaded_message_passing() { - let config = RuntimeConfig { - num_threads: 4, - ..Default::default() - }; - let rt = Runtime::new(config); - let inbox: Inbox = rt.new_inbox().unwrap(); - - // Create a longer chain to exercise multi-threading - let mut target = *inbox.addr(); - for _ in 0..20 { - target = rt.spawn(ForwarderActor { target }).unwrap(); - } - let start_addr = target; - - // Send message - rt.send_to(start_addr, ForwardMessage(999)).unwrap(); - - // Spawn thread to check for result and shutdown - let ctx = rt.run().unwrap(); - let inbox_check = std::thread::spawn(move || { - for _ in 0..100 { - std::thread::sleep(std::time::Duration::from_millis(10)); - if let Some(ForwardMessage(val)) = inbox.try_recv() { - ctx.shutdown(); - return Some(val); - } - } - ctx.shutdown(); - None - }); - - let result = inbox_check.join().unwrap(); - assert_eq!(result, Some(999)); -} diff --git a/tests/stress/concurrency.rs b/tests/stress/concurrency.rs deleted file mode 100644 index 0656652..0000000 --- a/tests/stress/concurrency.rs +++ /dev/null @@ -1,330 +0,0 @@ -//! Concurrency stress tests - hunt for race conditions. -//! -//! These tests target the shutdown races and concurrent access patterns -//! that are most likely to expose bugs. - -use super::{BlackHole, Msg}; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; -use std::thread; -use std::time::Duration; -use swactor::runtime::{Runtime, RuntimeConfig}; - -/// Shutdown while messages are in flight. -/// Target: AtomicBool ordering bugs, use-after-shutdown. -#[test] -#[cfg(feature = "stress")] -fn shutdown_under_load() { - println!("\n>>> STRESS: Shutdown Under Load"); - - let mut panics = 0; - let mut successes = 0; - - // Run many iterations to catch rare races - for iteration in 0..100 { - let result = std::panic::catch_unwind(|| { - let config = RuntimeConfig { - max_actors: 100, - actor_max_messages: 1000, - num_threads: 4, - ..Default::default() - }; - let runtime = Runtime::new(config); - - // Spawn actors - let mut actors = Vec::new(); - for _ in 0..50 { - if let Ok(addr) = runtime.spawn(BlackHole) { - actors.push(addr); - } - } - - let handle = runtime.run().unwrap(); - let rt = handle.runtime.clone(); - - // Sender thread - blast messages - let actors_clone = actors.clone(); - let rt_send = rt.clone(); - let sender = thread::spawn(move || { - for _ in 0..1000 { - for actor in &actors_clone { - let _ = rt_send.send_to::(*actor, Msg); - } - } - }); - - // Random delay before shutdown - let delay = Duration::from_micros((iteration * 17) % 500); - thread::sleep(delay); - - // Shutdown while sender is still going - handle.shutdown(); - - // Wait for sender (it should not panic) - let _ = sender.join(); - - // Join should complete (not hang) - handle.join(); - }); - - match result { - Ok(_) => successes += 1, - Err(_) => panics += 1, - } - } - - println!(" Iterations: 100"); - println!(" Successes: {}", successes); - println!(" Panics: {}", panics); - - if panics > 0 { - println!(">>> FAIL: {} panics detected during shutdown\n", panics); - } else { - println!(">>> PASS: No panics during shutdown under load\n"); - } - - assert_eq!(panics, 0, "Shutdown under load caused panics"); -} - -/// Send to actor immediately after spawn. -/// Target: Race between spawn registration and first message. -#[test] -#[cfg(feature = "stress")] -fn send_to_newborn() { - println!("\n>>> STRESS: Send to Newborn Actor"); - - let mut total_spawned = 0; - let mut total_send_ok = 0; - let mut total_send_fail = 0; - - for _ in 0..100 { - let config = RuntimeConfig { - max_actors: 1000, - actor_max_messages: 100, - num_threads: 4, - ..Default::default() - }; - let runtime = Runtime::new(config); - let handle = runtime.run().unwrap(); - - // Immediately spawn and send - for _ in 0..50 { - if let Ok(addr) = handle.runtime.spawn(BlackHole) { - total_spawned += 1; - // Send immediately - actor may not be registered yet - if handle.runtime.send_to::(addr, Msg).is_ok() { - total_send_ok += 1; - } else { - total_send_fail += 1; - } - } - } - - handle.shutdown(); - handle.join(); - } - - println!(" Total spawned: {}", total_spawned); - println!(" Sends succeeded: {}", total_send_ok); - println!(" Sends failed: {}", total_send_fail); - - if total_send_fail > 0 { - println!(">>> FAIL: {} messages failed to send\n", total_send_fail); - } else { - println!(">>> PASS: All messages succeeded\n"); - } - - assert_eq!(total_send_fail, 0, "Race condition caused failed message delivery"); - - println!(">>> Test complete\n"); -} - -/// Rapid spawn/despawn cycles. -/// Target: Queue management under churn. -#[test] -#[cfg(feature = "stress")] -fn rapid_spawn_churn() { - println!("\n>>> STRESS: Rapid Spawn Churn"); - - let config = RuntimeConfig { - max_actors: 100, - actor_max_messages: 100, - num_threads: 4, - ..Default::default() - }; - let runtime = Runtime::new(config); - let handle = runtime.run().unwrap(); - - let spawn_count = Arc::new(AtomicUsize::new(0)); - let fail_count = Arc::new(AtomicUsize::new(0)); - - // Multiple threads spawning actors - let mut threads = Vec::new(); - for _ in 0..4 { - let rt = handle.runtime.clone(); - let spawns = spawn_count.clone(); - let fails = fail_count.clone(); - - threads.push(thread::spawn(move || { - for _ in 0..500 { - match rt.spawn(BlackHole) { - Ok(_) => { - spawns.fetch_add(1, Ordering::Relaxed); - } - Err(_) => { - fails.fetch_add(1, Ordering::Relaxed); - } - } - // Small yield to increase interleaving - thread::yield_now(); - } - })); - } - - // Let it churn - thread::sleep(Duration::from_millis(100)); - - handle.shutdown(); - - for t in threads { - let _ = t.join(); - } - handle.join(); - - let total_spawns = spawn_count.load(Ordering::Relaxed); - let total_fails = fail_count.load(Ordering::Relaxed); - - println!(" Spawn attempts: {}", total_spawns + total_fails); - println!(" Successes: {}", total_spawns); - println!(" Failures: {} (expected - queue fills)", total_fails); - println!(">>> Test complete - no panics\n"); -} - -/// Multiple threads sending to same actor. -/// Target: Inbox contention, message ordering. -#[test] -#[cfg(feature = "stress")] -fn inbox_contention() { - println!("\n>>> STRESS: Inbox Contention"); - - let config = RuntimeConfig { - max_actors: 10, - actor_max_messages: 100_000, - num_threads: 4, - ..Default::default() - }; - let runtime = Runtime::new(config); - let target = runtime.spawn(BlackHole).unwrap(); - let handle = runtime.run().unwrap(); - - // Wait for registration - thread::sleep(Duration::from_millis(10)); - - let send_count = Arc::new(AtomicUsize::new(0)); - let fail_count = Arc::new(AtomicUsize::new(0)); - - // 8 threads all sending to same actor - let mut threads = Vec::new(); - for _ in 0..8 { - let rt = handle.runtime.clone(); - let sends = send_count.clone(); - let fails = fail_count.clone(); - - threads.push(thread::spawn(move || { - for _ in 0..10_000 { - if rt.send_to::(target, Msg).is_ok() { - sends.fetch_add(1, Ordering::Relaxed); - } else { - fails.fetch_add(1, Ordering::Relaxed); - } - } - })); - } - - for t in threads { - let _ = t.join(); - } - - // Let messages process - thread::sleep(Duration::from_millis(50)); - - handle.shutdown(); - handle.join(); - - let total_sends = send_count.load(Ordering::Relaxed); - let total_fails = fail_count.load(Ordering::Relaxed); - - println!(" Threads: 8"); - println!(" Msgs per thread: 10,000"); - println!(" Total sent: {}", total_sends); - println!(" Total failed: {}", total_fails); - println!( - " Success rate: {:.1}%", - (total_sends as f64 / (total_sends + total_fails) as f64) * 100.0 - ); - println!(">>> Test complete - no panics\n"); -} - -/// Shutdown timing fuzz - randomize when shutdown is called. -/// Target: Edge cases in shutdown state machine. -#[test] -#[cfg(feature = "stress")] -fn shutdown_timing_fuzz() { - println!("\n>>> STRESS: Shutdown Timing Fuzz"); - - let mut results = Vec::new(); - - for delay_us in [0, 1, 10, 100, 1000, 5000] { - let mut ok = 0; - let mut fail = 0; - - for _ in 0..20 { - let result = std::panic::catch_unwind(|| { - let config = RuntimeConfig { - max_actors: 50, - actor_max_messages: 100, - num_threads: 4, - ..Default::default() - }; - let runtime = Runtime::new(config); - - for _ in 0..20 { - let _ = runtime.spawn(BlackHole); - } - - let handle = runtime.run().unwrap(); - - // Specific delay - if delay_us > 0 { - thread::sleep(Duration::from_micros(delay_us)); - } - - handle.shutdown(); - handle.join(); - }); - - match result { - Ok(_) => ok += 1, - Err(_) => fail += 1, - } - } - - results.push((delay_us, ok, fail)); - } - - println!(" delay_us ok fail"); - println!(" -------- -- ----"); - for (delay, ok, fail) in &results { - println!(" {:>8} {:>2} {:>4}", delay, ok, fail); - } - - let total_fails: i32 = results.iter().map(|(_, _, f)| *f).sum(); - if total_fails > 0 { - println!( - "\n>>> FAIL: {} panics across timing variations", - total_fails - ); - } else { - println!("\n>>> PASS: All timing variations succeeded"); - } -} diff --git a/tests/stress/mod.rs b/tests/stress/mod.rs deleted file mode 100644 index 2839ac9..0000000 --- a/tests/stress/mod.rs +++ /dev/null @@ -1,199 +0,0 @@ -//! Stress test utilities and result reporting. -//! -//! Provides a simple framework for stress tests with JSON + pretty output. - -#![allow(dead_code)] // Utilities may not all be used in every test - -pub mod concurrency; -pub mod saturation; - -use std::time::{Duration, Instant}; - -/// Results from a stress test -#[derive(Debug)] -pub struct StressResult { - pub name: String, - pub duration: Duration, - pub operations: u64, - pub successes: u64, - pub failures: u64, - pub notes: Vec, -} - -impl StressResult { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - duration: Duration::ZERO, - operations: 0, - successes: 0, - failures: 0, - notes: Vec::new(), - } - } - - pub fn failure_rate(&self) -> f64 { - if self.operations == 0 { - 0.0 - } else { - (self.failures as f64 / self.operations as f64) * 100.0 - } - } - - pub fn throughput(&self) -> f64 { - let secs = self.duration.as_secs_f64(); - if secs > 0.0 { - self.operations as f64 / secs - } else { - 0.0 - } - } - - pub fn note(&mut self, msg: impl Into) { - self.notes.push(msg.into()); - } - - pub fn print(&self) { - println!("\n{}", "=".repeat(60)); - println!(" STRESS: {}", self.name); - println!("{}", "=".repeat(60)); - println!(" Duration: {:?}", self.duration); - println!(" Operations: {}", self.operations); - println!(" Successes: {}", self.successes); - println!(" Failures: {}", self.failures); - println!(" Failure Rate: {:.2}%", self.failure_rate()); - println!(" Throughput: {:.2} ops/sec", self.throughput()); - - if !self.notes.is_empty() { - println!(); - println!(" Notes:"); - for note in &self.notes { - println!(" - {}", note); - } - } - println!("{}", "=".repeat(60)); - } - - pub fn to_json(&self) -> String { - format!( - r#"{{"name":"{}","duration_ms":{},"operations":{},"successes":{},"failures":{},"failure_rate_pct":{:.2},"throughput":{:.2},"notes":{:?}}}"#, - self.name, - self.duration.as_millis(), - self.operations, - self.successes, - self.failures, - self.failure_rate(), - self.throughput(), - self.notes - ) - } -} - -/// A simple stress test runner -pub struct Stress { - name: String, - duration: Option, - iterations: Option, -} - -impl Stress { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - duration: None, - iterations: None, - } - } - - /// Run for a fixed duration - pub fn for_duration(mut self, d: Duration) -> Self { - self.duration = Some(d); - self - } - - /// Run for a fixed number of iterations - pub fn for_iterations(mut self, n: u64) -> Self { - self.iterations = Some(n); - self - } - - /// Run the stress test, counting successes and failures - pub fn run(self, mut f: F) -> StressResult - where - F: FnMut() -> bool, // returns true on success, false on failure - { - let mut result = StressResult::new(&self.name); - let start = Instant::now(); - - match (self.duration, self.iterations) { - (Some(duration), _) => { - while start.elapsed() < duration { - if f() { - result.successes += 1; - } else { - result.failures += 1; - } - result.operations += 1; - } - } - (None, Some(iterations)) => { - for _ in 0..iterations { - if f() { - result.successes += 1; - } else { - result.failures += 1; - } - result.operations += 1; - } - } - (None, None) => { - // Default: 1000 iterations - for _ in 0..1000 { - if f() { - result.successes += 1; - } else { - result.failures += 1; - } - result.operations += 1; - } - } - } - - result.duration = start.elapsed(); - result - } -} - -// Test actors used across stress tests -use swactor::{actor::ActorInterface, runtime::Ctx}; - -/// An actor that just absorbs messages -pub struct BlackHole; - -#[derive(Clone)] -pub struct Msg; - -impl ActorInterface for BlackHole { - type Incoming = Msg; - type Response = (); - fn handle(&mut self, _ctx: &Ctx, _msg: Msg) {} -} - -/// An actor that counts messages received -pub struct Counter { - pub count: usize, -} - -impl Counter { - pub fn new() -> Self { - Self { count: 0 } - } -} - -impl ActorInterface for Counter { - type Incoming = Msg; - type Response = (); - fn handle(&mut self, _ctx: &Ctx, _msg: Msg) { - self.count += 1; - } -} diff --git a/tests/stress/saturation.rs b/tests/stress/saturation.rs deleted file mode 100644 index dc2aeb2..0000000 --- a/tests/stress/saturation.rs +++ /dev/null @@ -1,175 +0,0 @@ -//! Saturation stress tests - find where the runtime breaks. -//! -//! These tests intentionally push past limits to document failure modes. - -use super::{BlackHole, Counter, Msg, Stress, StressResult}; -use std::time::Duration; -use swactor::runtime::{Runtime, RuntimeConfig}; - -/// Blast the transfer queue (replaces router_inbox_overflow) -#[test] -#[cfg(feature = "stress")] -fn transfer_queue_overflow() { - println!("\n>>> STRESS: Transfer Queue Overflow"); - - let config = RuntimeConfig { - max_actors: 10, - actor_max_messages: 100, // Tiny buffer - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(BlackHole).unwrap(); - - // Process spawn - runtime.tick(); - - // Blast messages without processing - let mut result = StressResult::new("transfer_queue_overflow"); - let start = std::time::Instant::now(); - - for _ in 0..10_000 { - result.operations += 1; - if runtime.send_to::(sink, Msg).is_ok() { - result.successes += 1; - } else { - result.failures += 1; - } - } - - result.duration = start.elapsed(); - result.note(format!("Transfer buffer: 100, Messages sent: 10,000")); - - // With hybrid channel, no failures expected - assert_eq!(result.failures, 0, "Hybrid channel should not reject"); - result.print(); - println!(">>> PASS: Hybrid channel prevented transfer queue overflow\n"); -} - -/// Blast a single actor's mailbox via transfer queue -#[test] -#[cfg(feature = "stress")] -fn actor_inbox_overflow() { - println!("\n>>> STRESS: Actor Inbox Overflow"); - - let config = RuntimeConfig { - max_actors: 10, - actor_max_messages: 100_000, // Large transfer buffer - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(Counter::new()).unwrap(); - - // Process spawn - runtime.tick(); - - // Now blast messages - let mut sent = 0u64; - let mut failed = 0u64; - for _ in 0..10_000 { - if runtime.send_to::(sink, Msg).is_ok() { - sent += 1; - } else { - failed += 1; - } - // Tick occasionally to let worker deliver - if sent % 100 == 0 { - runtime.tick(); - } - } - - // Process all remaining messages - for _ in 0..5000 { - runtime.tick(); - } - - println!(" Sent: {}", sent); - println!(" Failed: {}", failed); - - assert_eq!(failed, 0, "Transfer queue rejected message under load"); - println!(">>> PASS: No message loss with hybrid channel\n"); -} - -/// Blast the runtime with actor spawns -#[test] -#[cfg(feature = "stress")] -fn actor_queue_overflow() { - println!("\n>>> STRESS: Actor Queue Overflow"); - - let config = RuntimeConfig { - max_actors: 100, // Small actor queue - actor_max_messages: 100, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - - let mut result = StressResult::new("actor_queue_overflow"); - let start = std::time::Instant::now(); - - // Try to spawn 500 actors into 100-slot queue - for _ in 0..500 { - result.operations += 1; - match runtime.spawn(BlackHole) { - Ok(_) => result.successes += 1, - Err(_) => result.failures += 1, - } - } - - result.duration = start.elapsed(); - result.note(format!("Queue capacity: 100, Spawn attempts: 500")); - result.print(); - - // Note: With hybrid channel (overflow to SegQueue), we expect no failures - assert_eq!( - result.failures, 0, - "Spawned more actors than queue capacity" - ); - println!(">>> PASS: Actor queue correctly handles overflow\n"); -} - -/// Sustained overload - run at 2x capacity for extended period. -/// Documents: Does the system degrade gracefully or crash? -#[test] -#[cfg(feature = "stress")] -fn sustained_overload() { - println!("\n>>> STRESS: Sustained Overload"); - - let config = RuntimeConfig { - max_actors: 100, - actor_max_messages: 1000, - num_threads: 1, - ..Default::default() - }; - let runtime = Runtime::new(config); - - // Spawn some actors - let mut actors = Vec::new(); - for _ in 0..50 { - if let Ok(addr) = runtime.spawn(Counter::new()) { - actors.push(addr); - } - } - - // Process spawn registrations - for _ in 0..200 { - runtime.tick(); - } - - let result = Stress::new("sustained_overload") - .for_duration(Duration::from_secs(2)) - .run(|| { - // Send to random actor - let idx = (std::time::Instant::now().elapsed().as_nanos() as usize) % actors.len(); - let success = runtime.send_to::(actors[idx], Msg).is_ok(); - - // Process some (but not all) - simulating overload - runtime.tick(); - - success - }); - - result.print(); - println!(">>> System survived sustained overload without panic\n"); -} diff --git a/tests/stress_tests.rs b/tests/stress_tests.rs deleted file mode 100644 index 216aea2..0000000 --- a/tests/stress_tests.rs +++ /dev/null @@ -1,11 +0,0 @@ -//! Stress test suite for swactor runtime. -//! -//! Run with: cargo test --features stress stress_ -- --nocapture -//! -//! These tests are hidden behind the `stress` feature flag because they: -//! - Take longer to run -//! - Intentionally push the system to failure -//! - May produce different results on different machines - -#[cfg(feature = "stress")] -mod stress;