feat: refactor tests
Refactor the tests and component APIs to be more amenable to metrics, fuzzing, understanding data flows, etc.
This commit is contained in:
Zachery Aaron Shores-Chmielewski
parent
c80060f453
commit
799b654251
19 changed files with 1040 additions and 2417 deletions
199
ARCHITECTURE.md
199
ARCHITECTURE.md
|
|
@ -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<M>` 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 │
|
||||
└───────────────────────┘
|
||||
```
|
||||
535
Cargo.lock
generated
535
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
10
Cargo.toml
10
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
|
||||
|
|
|
|||
|
|
@ -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<Duration>,
|
||||
/// Optional: elements processed (for throughput calculation)
|
||||
pub elements: Option<u64>,
|
||||
}
|
||||
|
||||
/// 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<f64>, // elements per second
|
||||
}
|
||||
|
||||
impl BenchResult {
|
||||
/// Calculate statistics from the raw timing data
|
||||
pub fn stats(&self) -> Stats {
|
||||
let mut sorted: Vec<Duration> = 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::<f64>()
|
||||
/ 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<u64>,
|
||||
}
|
||||
|
||||
impl Bench {
|
||||
pub fn new(name: impl Into<String>) -> 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<S, T, F>(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<BenchResult>,
|
||||
}
|
||||
|
||||
impl BenchSuite {
|
||||
pub fn new(name: impl Into<String>) -> 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<T>(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) }
|
||||
}
|
||||
|
|
@ -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<String> = 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.");
|
||||
}
|
||||
276
benches/runtime_benchmarks.rs
Normal file
276
benches/runtime_benchmarks.rs
Normal file
|
|
@ -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::<PongMessage>().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::<NoopMessage>().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::<RingMessage>().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);
|
||||
|
|
@ -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<std::sync::atomic::AtomicUsize>,
|
||||
}
|
||||
|
||||
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<u8>);
|
||||
|
||||
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::<Increment>(*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::<Increment>(*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::<Increment>(*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::<Payload>(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
|
||||
}
|
||||
|
|
@ -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<ActorAddress>,
|
||||
}
|
||||
|
||||
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::<Ping>(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::<Ping>(*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::<Ping>(*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<ActorAddress> = 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::<Ping>(*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
|
||||
}
|
||||
37
src/actor.rs
37
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<T: 'static + Sized + Clone + Send + Sync> 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;
|
||||
|
|
|
|||
|
|
@ -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::<u64>(), Some(42u64));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn envelope_wrong_type_returns_none() {
|
||||
let addr = ActorAddress::default();
|
||||
let env = Envelope::new(addr, Box::new(42u64));
|
||||
assert!(env.downcast::<String>().is_none());
|
||||
}
|
||||
}
|
||||
220
src/runtime.rs
220
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<dyn Any + Send>);
|
||||
}
|
||||
|
||||
impl<M: Message> SenderT for Sender<M> {
|
||||
fn try_send_any(&self, msg: Box<dyn Any + Send>) {
|
||||
if let Ok(typed) = msg.downcast::<M>() {
|
||||
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<dyn Any + Send>,
|
||||
}
|
||||
|
||||
impl Envelope {
|
||||
pub fn new(dest: ActorAddress, payload: Box<dyn Any + Send>) -> Self {
|
||||
Self { dest, payload }
|
||||
}
|
||||
|
||||
pub fn dest(&self) -> ActorAddress {
|
||||
self.dest
|
||||
}
|
||||
|
||||
pub fn downcast<M: 'static>(self) -> Option<M> {
|
||||
self.payload.downcast::<M>().ok().map(|b| *b)
|
||||
}
|
||||
|
||||
pub fn into_payload(self) -> Box<dyn Any + Send> {
|
||||
self.payload
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// ─── InboxRegistry ───────────────────────────────────────────────────────────
|
||||
|
||||
/// Registry of external inboxes — replaces the Router's role for non-actor receivers.
|
||||
pub(crate) struct InboxRegistry {
|
||||
senders: RwLock<HashMap<ActorAddress, Arc<dyn SenderT>>>,
|
||||
}
|
||||
|
||||
impl InboxRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
senders: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn register(&self, addr: ActorAddress, sender: Arc<dyn SenderT>) {
|
||||
self.senders.write().unwrap().insert(addr, sender);
|
||||
}
|
||||
|
||||
pub fn try_deliver(
|
||||
&self,
|
||||
addr: ActorAddress,
|
||||
msg: Box<dyn Any + Send>,
|
||||
) -> 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<M: Message> {
|
||||
|
|
@ -108,12 +31,23 @@ impl<M: Message> Inbox<M> {
|
|||
}
|
||||
}
|
||||
|
||||
/// Handle for dealing with a runtime that has started via the `Runtime::run()` method.
|
||||
pub struct RuntimeHandle {
|
||||
pub runtime: Arc<Runtime>,
|
||||
threads: Vec<JoinHandle<()>>,
|
||||
}
|
||||
|
||||
/// Object-safe inner trait for sending type-erased messages.
|
||||
pub(crate) trait ContextInner {
|
||||
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
|
||||
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> 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<dyn Any + Send>);
|
||||
}
|
||||
|
||||
impl<M: Message> SenderT for Sender<M> {
|
||||
fn try_send_any(&self, msg: Box<dyn Any + Send>) {
|
||||
if let Ok(typed) = msg.downcast::<M>() {
|
||||
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<dyn Any + Send>,
|
||||
}
|
||||
|
||||
impl Envelope {
|
||||
pub fn new(dest: ActorAddress, payload: Box<dyn Any + Send>) -> Self {
|
||||
Self { dest, payload }
|
||||
}
|
||||
|
||||
pub fn dest(&self) -> ActorAddress {
|
||||
self.dest
|
||||
}
|
||||
|
||||
pub fn downcast<M: 'static>(self) -> Option<M> {
|
||||
self.payload.downcast::<M>().ok().map(|b| *b)
|
||||
}
|
||||
|
||||
pub fn into_payload(self) -> Box<dyn Any + Send> {
|
||||
self.payload
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// ─── InboxRegistry ───────────────────────────────────────────────────────────
|
||||
|
||||
/// Registry of external inboxes — replaces the Router's role for non-actor receivers.
|
||||
pub(crate) struct InboxRegistry {
|
||||
senders: RwLock<HashMap<ActorAddress, Arc<dyn SenderT>>>,
|
||||
}
|
||||
|
||||
impl InboxRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
senders: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn register(&self, addr: ActorAddress, sender: Arc<dyn SenderT>) {
|
||||
self.senders.write().unwrap().insert(addr, sender);
|
||||
}
|
||||
|
||||
pub fn try_deliver(
|
||||
&self,
|
||||
addr: ActorAddress,
|
||||
msg: Box<dyn Any + Send>,
|
||||
) -> 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<Runtime>,
|
||||
threads: Vec<JoinHandle<()>>,
|
||||
|
||||
|
||||
|
||||
/// Object-safe inner trait for sending type-erased messages.
|
||||
pub(crate) trait ContextInner {
|
||||
fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
|
||||
fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> 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<dyn Any + Send>) -> Result<(), Error> {
|
||||
|
|
|
|||
|
|
@ -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<M: Message> Mailbox<M> {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Worker thread loop for multi-threaded runtime.
|
||||
/// Uses spin → yield → park backoff to reduce CPU usage when idle.
|
||||
fn worker_loop(worker: &mut Worker, rt: &Runtime) {
|
||||
let tc = TickContext {
|
||||
address_map: &rt.address_map,
|
||||
transfer_txs: &rt.transfer_txs,
|
||||
spawn_txs: &rt.spawn_txs,
|
||||
placement: &rt.placement,
|
||||
inbox_registry: &rt.inbox_registry,
|
||||
config: &rt.config,
|
||||
};
|
||||
let bp = &rt.config.backoff_policy;
|
||||
let mut idle_count: u32 = 0;
|
||||
|
||||
while rt.is_running.load(Ordering::Acquire) {
|
||||
let did_work = worker.tick_once(&tc);
|
||||
|
||||
if did_work {
|
||||
idle_count = 0;
|
||||
} else {
|
||||
idle_count = idle_count.saturating_add(1);
|
||||
if idle_count < bp.spin_threshold {
|
||||
// Hot spin — no hint, keep polling fast
|
||||
} else if idle_count < bp.yield_threshold {
|
||||
thread::yield_now();
|
||||
} else {
|
||||
// Park: sleep briefly, cap at configured max
|
||||
let micros = std::cmp::min(
|
||||
(idle_count - bp.yield_threshold) as u64 * bp.sleep_increment_us,
|
||||
bp.sleep_max_us,
|
||||
);
|
||||
thread::sleep(std::time::Duration::from_micros(micros));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<i32> = 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<i32> = 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<i32> = 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));
|
||||
}
|
||||
97
tests/runtime_api_tests.rs
Normal file
97
tests/runtime_api_tests.rs
Normal file
|
|
@ -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<EchoResponse> = 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<EchoResponse> = 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)));
|
||||
}
|
||||
|
|
@ -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<PongMessage> = 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<ForwardMessage> = 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<ForwardMessage> = 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));
|
||||
}
|
||||
|
|
@ -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::<Msg>(*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::<Msg>(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::<Msg>(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");
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String>,
|
||||
}
|
||||
|
||||
impl StressResult {
|
||||
pub fn new(name: impl Into<String>) -> 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<String>) {
|
||||
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<Duration>,
|
||||
iterations: Option<u64>,
|
||||
}
|
||||
|
||||
impl Stress {
|
||||
pub fn new(name: impl Into<String>) -> 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<F>(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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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::<Msg>(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::<Msg>(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::<Msg>(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");
|
||||
}
|
||||
|
|
@ -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;
|
||||
Loading…
Reference in a new issue