From 5a9af73de6cf9e3505ed5f0b0ebe0d3dd7a91c4c Mon Sep 17 00:00:00 2001 From: zacheryasc Date: Fri, 6 Feb 2026 11:25:37 +0000 Subject: [PATCH] refactor: major library changes (#5) Refactoring to logically separate component modules in order to make it easier to develop tests, metrics, tracing, etc. --- .gitignore | 1 + Cargo.lock | 535 ++++++++++++++++++++++++++++++++++ Cargo.toml | 10 +- DESIGN.md | 77 ----- README.md | 59 +++- benches/harness.rs | 270 ----------------- benches/main.rs | 46 --- benches/runtime_benchmarks.rs | 276 ++++++++++++++++++ benches/scaling.rs | 280 ------------------ benches/throughput.rs | 343 ---------------------- examples/hello.rs | 6 +- examples/ring.rs | 6 +- src/actor.rs | 96 +++--- src/address_map.rs | 120 ++++++++ src/channel.rs | 26 +- src/config.rs | 55 ++++ src/lib.rs | 19 +- src/router.rs | 75 ----- src/runtime.rs | 434 +++++++++++++++++---------- src/worker.rs | 258 ++++++++++++++++ tests/runtime_api_tests.rs | 97 ++++++ tests/runtime_tests.rs | 130 --------- tests/stress/concurrency.rs | 332 --------------------- tests/stress/mod.rs | 199 ------------- tests/stress/saturation.rs | 173 ----------- tests/stress_tests.rs | 11 - 26 files changed, 1745 insertions(+), 2189 deletions(-) delete mode 100644 DESIGN.md delete mode 100644 benches/harness.rs delete mode 100644 benches/main.rs create mode 100644 benches/runtime_benchmarks.rs delete mode 100644 benches/scaling.rs delete mode 100644 benches/throughput.rs create mode 100644 src/address_map.rs create mode 100644 src/config.rs delete mode 100644 src/router.rs create mode 100644 src/worker.rs create mode 100644 tests/runtime_api_tests.rs delete mode 100644 tests/runtime_tests.rs delete mode 100644 tests/stress/concurrency.rs delete mode 100644 tests/stress/mod.rs delete mode 100644 tests/stress/saturation.rs delete mode 100644 tests/stress_tests.rs diff --git a/.gitignore b/.gitignore index ea8c4bf..5f9ff5d 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ /target +.vscode/ \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index bb8e0dc..d0e4e9d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,12 +2,158 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + +[[package]] +name = "anstyle" +version = "1.0.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78" + +[[package]] +name = "autocfg" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" + +[[package]] +name = "bumpalo" +version = "3.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510" + +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cfg-if" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + +[[package]] +name = "clap" +version = "4.5.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6899ea499e3fb9305a65d5ebf6e3d2248c5fab291f300ad0a704fbe142eae31a" +dependencies = [ + "clap_builder", +] + +[[package]] +name = "clap_builder" +version = "4.5.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b12c8b680195a62a8364d16b8447b01b6c2c8f9aaf68bee653be34d4245e238" +dependencies = [ + "anstyle", + "clap_lex", +] + +[[package]] +name = "clap_lex" +version = "0.7.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3e64b0cc0439b12df2fa678eae89a1c56a529fd067a9115f7827f1fffd22b32" + +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools", + "num-traits", + "once_cell", + "oorandom", + "plotters", + "rayon", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools", +] + +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -23,6 +169,18 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + +[[package]] +name = "either" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" + [[package]] name = "getrandom" version = "0.2.17" @@ -34,22 +192,399 @@ 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" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc" +[[package]] +name = "memchr" +version = "2.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +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", ] +[[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]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasm-bindgen" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64024a30ec1e37399cf85a7ffefebdb72205ca1c972291c51512360d90bd8566" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "008b239d9c740232e71bd39e8ef6429d27097518b6b30bdf9086833bd5b6d608" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5256bae2d58f54820e6490f9839c49780dff84c65aeab9e772f15d5f0e913a55" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f01b580c9ac74c8d8f0c0e4afb04eeef2acf145458e52c03845ee9cd23e3d12" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "web-sys" +version = "0.3.85" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "312e32e551d92129218ea9a2452120f4aabc03529ef03e4d0d82fb2780608598" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zerocopy" +version = "0.8.39" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db6d35d663eadb6c932438e763b262fe1a70987f9ae936e60158176d710cae4a" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.39" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4122cd3169e94605190e77839c9a40d40ed048d305bfdc146e7df40ab0f3e517" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "zmij" +version = "1.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ff05f8caa9038894637571ae6b9e29466c1f4f829d26c9b28f869a29cbe3445" diff --git a/Cargo.toml b/Cargo.toml index c8963bf..a348085 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,11 @@ stress = [] # Enable stress tests [dependencies] getrandom = { version = "0.2", optional = true } crossbeam-queue = "0.3.12" +crossbeam-utils = "0.8.21" -[[bin]] -name = "bench" -path = "benches/main.rs" +[dev-dependencies] +criterion = { version = "0.5", features = ["html_reports"] } + +[[bench]] +name = "runtime_benchmarks" +harness = false diff --git a/DESIGN.md b/DESIGN.md deleted file mode 100644 index c43a407..0000000 --- a/DESIGN.md +++ /dev/null @@ -1,77 +0,0 @@ -# Design goals -Get as much usability and speed as possible while keeping line count low. Aim for no footguns, ability to plug in -logic easily, and run near anywhere. We may make this a `![no_std]` library, but the MVP will use the -memory allocator and threading provided by the rust standard library. - -We are not building a new erlang/BEAM. Minimal feature set means spawning actor processes, not having supervisiors, lots of process -monitoring tools, prempting, etc. - -## Actor model - -An actor has: - - - An inbox: - this is a mpsc channel that the runtime/router dumps messages into and the actor consumes when the runtime loads it - Implemented as a barebones atomic ring buffer. The router is responsible for inserting messages. - - - an outbox channel connection: - this is a mpmc channel that is implemented by the runtime and router. Actors on this specific channel put responses and outgoing messages into this channel, to be routed to the given address. - - - a growable and mutable state: - An actor owns some, from the runtime perspective, type erased bytes. The actor when processing messages can access its own state, but no other task can. This includes viewing. - - - a set of functions for processing messages: - When the runtime loads the actor, it locks the inbox and attempts to process the messages therein. - -## Runtime - -In order for an actor to consume and send messages, it is processed by a runtime. The runtime, in order to negotiate messages between -actors, possesses a router. - -A runtime has: - - An actor processing thread(s): - the processor will mark an actor as busy, load its state and inbox, and begin consuming messages from the inbox. The number of messages consumed is determined by the runtime. A good start is a backpressure strategy: after loading, process messages until mailbox is empty or size drops below a threshold (e.g., "drain to 50%"). - - - A message router: - the router is responsible for ensuring messages posted by actors get delivered to the appropriate inbox. - - - An atomic ring buffer containing thread-safe references to actors that are not currently loaded. Actors are popped off the buffer, messages are - processed, and the reference is returned to the buffer/queue before the next actor is loaded. - -## Router - -The router is the engine for message delivery. It posesses: - - - An actor address book: - The address book maps actor ids to `Sender` references that can be used to deliver messages to the actor inbox. - - - Its own inbox: - The router possesses its own mpsc queue where references to messages are stored. The router will process this queue by dereferencing and writing directly into the recipient's inbox buffer. - -### Misc - -A means of providing an emergency overflow without adding much more code complexity. The mutex means -this will not be `no_std` however. - -```rust -struct HybridChannel { - // Start with lock-free ring buffer - ring: AtomicRingBuffer, - - // When full, spill into a Mutex> - overflow: parking_lot::Mutex>, - - // Track overflow frequency to resize ring proactively - overflow_count: AtomicUsize, -} - -impl HybridChannel { - fn push(&self, value: T) { - if self.ring.push(value).is_err() { - self.overflow.lock().push_back(value); - self.overflow_count.fetch_add(1, Relaxed); - // Optionally: if overflow_count > threshold, grow ring - } - } -} -``` diff --git a/README.md b/README.md index bc88d5a..bcce6fe 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,59 @@ # swactor -Small wasm-compatible actor library +(S)mall (W)ASM-compatible (actor) library + +## Quick example + +```rust +use swactor::{ + Ctx, + actor::{ActorAddress, ActorInterface}, + runtime::{Runtime, RuntimeConfig}, +}; + +#[derive(Debug, Default)] +struct Greeter { num_greeted: usize } + +#[derive(Debug, Default, Clone)] +struct GreetMessage { who: String, return_addr: ActorAddress } + +#[derive(Debug, Default, Clone)] +struct GreetResponse(String); + +impl ActorInterface for Greeter { + type Incoming = GreetMessage; + type Response = GreetResponse; + + fn handle(&mut self, ctx: &Ctx, msg: GreetMessage) { + let res = GreetResponse(format!("Hello, {}!", msg.who)); + self.num_greeted += 1; + if let Err(_) = ctx.send(msg.return_addr, res) { + self.num_greeted -= 1; + } + } +} + +fn main() { + let rt = Runtime::new(RuntimeConfig::default()); + let addr = rt.spawn(Greeter::default()).expect("failed to spawn"); + + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(addr, GreetMessage { + who: "world".into(), + return_addr: *inbox.addr(), + }).unwrap(); + + for _ in 0..3 { rt.tick(); } + let resp = inbox.try_recv().expect("should have response"); + println!("{}", resp.0); // "Hello, world!" +} +``` + +## Build & test + +```sh +cargo build +cargo test +cargo test --features stress # stress tests +cargo run --bin bench --release # benchmarks +cargo run --example hello +``` diff --git a/benches/harness.rs b/benches/harness.rs deleted file mode 100644 index 8eddc47..0000000 --- a/benches/harness.rs +++ /dev/null @@ -1,270 +0,0 @@ -//! Manual benchmark harness - zero dependencies, full control. -//! -//! Provides statistical analysis of benchmark runs including: -//! - Mean, median, min, max -//! - Standard deviation -//! - Percentiles (P50, P90, P99, P99.9) -//! - Throughput calculations -//! - Outlier detection and removal - -use std::time::{Duration, Instant}; - -/// Results from a single benchmark run -#[derive(Debug, Clone)] -pub struct BenchResult { - pub name: String, - pub iterations: usize, - pub total_time: Duration, - pub times: Vec, - /// Optional: elements processed (for throughput calculation) - pub elements: Option, -} - -/// Statistical summary of benchmark results -#[derive(Debug)] -pub struct Stats { - pub mean: Duration, - pub median: Duration, - pub min: Duration, - pub max: Duration, - pub std_dev: Duration, - pub p50: Duration, - pub p90: Duration, - pub p99: Duration, - pub p999: Duration, - pub throughput: Option, // elements per second -} - -impl BenchResult { - /// Calculate statistics from the raw timing data - pub fn stats(&self) -> Stats { - let mut sorted: Vec = self.times.clone(); - sorted.sort(); - - let n = sorted.len(); - assert!(n > 0, "Cannot compute stats on empty results"); - - let sum: Duration = sorted.iter().sum(); - let mean = sum / n as u32; - - let median = if n % 2 == 0 { - (sorted[n / 2 - 1] + sorted[n / 2]) / 2 - } else { - sorted[n / 2] - }; - - // Standard deviation - let mean_nanos = mean.as_nanos() as f64; - let variance: f64 = sorted - .iter() - .map(|t| { - let diff = t.as_nanos() as f64 - mean_nanos; - diff * diff - }) - .sum::() - / n as f64; - let std_dev = Duration::from_nanos(variance.sqrt() as u64); - - // Percentiles - let percentile = |p: f64| -> Duration { - let idx = ((p / 100.0) * (n - 1) as f64).round() as usize; - sorted[idx.min(n - 1)] - }; - - let throughput = self.elements.map(|e| { - let secs = self.total_time.as_secs_f64(); - if secs > 0.0 { - (e * self.iterations as u64) as f64 / secs - } else { - 0.0 - } - }); - - Stats { - mean, - median, - min: sorted[0], - max: sorted[n - 1], - std_dev, - p50: percentile(50.0), - p90: percentile(90.0), - p99: percentile(99.0), - p999: percentile(99.9), - throughput, - } - } - - /// Pretty print the results - pub fn print(&self) { - let stats = self.stats(); - - println!("\n{}", "=".repeat(60)); - println!(" {}", self.name); - println!("{}", "=".repeat(60)); - println!(" Iterations: {}", self.iterations); - println!(" Total time: {:?}", self.total_time); - println!(); - println!(" Mean: {:?}", stats.mean); - println!(" Median: {:?}", stats.median); - println!(" Std Dev: {:?}", stats.std_dev); - println!(" Min: {:?}", stats.min); - println!(" Max: {:?}", stats.max); - println!(); - println!(" P50: {:?}", stats.p50); - println!(" P90: {:?}", stats.p90); - println!(" P99: {:?}", stats.p99); - println!(" P99.9: {:?}", stats.p999); - - if let Some(throughput) = stats.throughput { - println!(); - println!(" Throughput: {:.2} ops/sec", throughput); - if throughput > 1_000_000.0 { - println!(" {:.2} M ops/sec", throughput / 1_000_000.0); - } else if throughput > 1_000.0 { - println!(" {:.2} K ops/sec", throughput / 1_000.0); - } - } - println!("{}", "=".repeat(60)); - } -} - -/// A benchmark builder for configuring and running benchmarks -pub struct Bench { - name: String, - warmup_iters: usize, - bench_iters: usize, - elements_per_iter: Option, -} - -impl Bench { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - warmup_iters: 3, - bench_iters: 100, - elements_per_iter: None, - } - } - - /// Set number of warmup iterations (default: 3) - pub fn warmup(mut self, n: usize) -> Self { - self.warmup_iters = n; - self - } - - /// Set number of benchmark iterations (default: 100) - pub fn iters(mut self, n: usize) -> Self { - self.bench_iters = n; - self - } - - /// Set elements per iteration for throughput calculation - pub fn elements(mut self, n: u64) -> Self { - self.elements_per_iter = Some(n); - self - } - - /// Run the benchmark with setup before each iteration - pub fn run_with_setup(self, mut setup: S, mut f: F) -> BenchResult - where - S: FnMut() -> T, - F: FnMut(T), - { - // Warmup - for _ in 0..self.warmup_iters { - let state = setup(); - f(state); - } - - // Benchmark - let mut times = Vec::with_capacity(self.bench_iters); - let total_start = Instant::now(); - - for _ in 0..self.bench_iters { - let state = setup(); - let start = Instant::now(); - f(state); - times.push(start.elapsed()); - } - - let total_time = total_start.elapsed(); - - BenchResult { - name: self.name, - iterations: self.bench_iters, - total_time, - times, - elements: self.elements_per_iter, - } - } -} - -/// A collection of benchmarks to run together -pub struct BenchSuite { - name: String, - results: Vec, -} - -impl BenchSuite { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - results: Vec::new(), - } - } - - pub fn add(&mut self, result: BenchResult) { - self.results.push(result); - } - - pub fn print_summary(&self) { - println!("\n{}", "#".repeat(70)); - println!("# BENCHMARK SUITE: {}", self.name); - println!("{}", "#".repeat(70)); - - for result in &self.results { - result.print(); - } - - // Summary table - println!("\n{}", "-".repeat(70)); - println!(" SUMMARY"); - println!("{}", "-".repeat(70)); - println!( - " {:30} {:>12} {:>12} {:>12}", - "Benchmark", "Mean", "P99", "Throughput" - ); - println!("{}", "-".repeat(70)); - - for result in &self.results { - let stats = result.stats(); - let throughput_str = stats - .throughput - .map(|t| { - if t > 1_000_000.0 { - format!("{:.2}M/s", t / 1_000_000.0) - } else if t > 1_000.0 { - format!("{:.2}K/s", t / 1_000.0) - } else { - format!("{:.2}/s", t) - } - }) - .unwrap_or_else(|| "-".to_string()); - - println!( - " {:30} {:>12.2?} {:>12.2?} {:>12}", - result.name, stats.mean, stats.p99, throughput_str - ); - } - println!("{}", "-".repeat(70)); - } -} - -/// Prevent the compiler from optimizing away a value -#[inline(never)] -pub fn black_box(x: T) -> T { - // Use inline assembly to prevent optimization - // This is a simplified version - in practice, reads from the value - let ptr = &x as *const T; - unsafe { std::ptr::read_volatile(ptr) } -} diff --git a/benches/main.rs b/benches/main.rs deleted file mode 100644 index 41f32fd..0000000 --- a/benches/main.rs +++ /dev/null @@ -1,46 +0,0 @@ -//! Swactor Benchmark Suite -//! -//! A manual benchmark harness for measuring runtime performance. -//! Zero external dependencies - just std::time. -//! -//! Run with: cargo run --bin bench --release -//! -//! Options: -//! --throughput Run throughput benchmarks only -//! --scaling Run scaling benchmarks only -//! --all Run all benchmarks (default) - -mod harness; -mod throughput; -mod scaling; - -use std::env; - -fn main() { - let args: Vec = env::args().collect(); - - println!("============================================================"); - println!(" SWACTOR BENCHMARK SUITE"); - println!("============================================================"); - println!(); - - // Parse arguments - let run_throughput = args.contains(&"--throughput".to_string()) - || args.contains(&"--all".to_string()) - || args.len() == 1; - let run_scaling = args.contains(&"--scaling".to_string()) - || args.contains(&"--all".to_string()) - || args.len() == 1; - - if run_throughput { - let suite = throughput::run_all(); - suite.print_summary(); - } - - if run_scaling { - let suite = scaling::run_all(); - suite.print_summary(); - } - - println!("\nBenchmarks complete."); -} diff --git a/benches/runtime_benchmarks.rs b/benches/runtime_benchmarks.rs new file mode 100644 index 0000000..64eff02 --- /dev/null +++ b/benches/runtime_benchmarks.rs @@ -0,0 +1,276 @@ +use criterion::{ + criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion, Throughput, +}; +use swactor::{ + actor::{ActorAddress, ActorInterface}, + config::RuntimeConfig, + runtime::{Ctx, Runtime}, +}; + +// --------------------------------------------------------------------------- +// Helper +// --------------------------------------------------------------------------- + +fn make_config(max_actors: usize, max_messages: usize) -> RuntimeConfig { + RuntimeConfig { + max_actors, + actor_max_messages: max_messages, + num_threads: 1, + ..Default::default() + } +} + +// --------------------------------------------------------------------------- +// Message types +// --------------------------------------------------------------------------- + +#[derive(Clone)] +struct NoopMessage; + +#[derive(Clone)] +struct PingMessage { + reply_to: ActorAddress, +} + +#[derive(Clone)] +struct PongMessage; + +#[derive(Clone)] +struct CountMessage(u64); + +#[derive(Clone)] +struct RingMessage { + hops: u64, +} + +// --------------------------------------------------------------------------- +// Actor types +// --------------------------------------------------------------------------- + +struct NoopActor; + +impl ActorInterface for NoopActor { + type Incoming = NoopMessage; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: NoopMessage) {} +} + +struct EchoActor; + +impl ActorInterface for EchoActor { + type Incoming = PingMessage; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: PingMessage) { + let _ = ctx.send(msg.reply_to, PongMessage); + } +} + +struct SinkActor; + +impl ActorInterface for SinkActor { + type Incoming = CountMessage; + type Response = (); + fn handle(&mut self, _ctx: &Ctx, _msg: CountMessage) {} +} + +struct RingActor { + next: ActorAddress, +} + +impl ActorInterface for RingActor { + type Incoming = RingMessage; + type Response = (); + fn handle(&mut self, ctx: &Ctx, msg: RingMessage) { + let _ = ctx.send(self.next, RingMessage { hops: msg.hops + 1 }); + } +} + +// --------------------------------------------------------------------------- +// Latency benchmarks +// --------------------------------------------------------------------------- + +fn latency_benchmarks(c: &mut Criterion) { + let mut group = c.benchmark_group("latency"); + + // A1 — Spawn latency + group.bench_function("spawn", |b| { + b.iter_batched( + || Runtime::new(make_config(1_000, 1_000)), + |rt| { + rt.spawn(NoopActor).unwrap(); + }, + BatchSize::SmallInput, + ); + }); + + // A2 — Message round-trip + group.bench_function("message_roundtrip", |b| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(1_000, 1_000)); + let addr = rt.spawn(EchoActor).unwrap(); + rt.tick(); // register actor + let inbox = rt.new_inbox::().unwrap(); + let inbox_addr = *inbox.addr(); + (rt, addr, inbox, inbox_addr) + }, + |(rt, addr, inbox, inbox_addr)| { + rt.send_to(addr, PingMessage { reply_to: inbox_addr }).unwrap(); + for _ in 0..20 { + rt.tick(); + if inbox.try_recv().is_some() { + return; + } + } + panic!("PongMessage not received within 20 ticks"); + }, + BatchSize::SmallInput, + ); + }); + + // A3 — Fire-and-forget send + group.bench_function("send_fire_and_forget", |b| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(1_000, 100_000)); + let addr = rt.spawn(NoopActor).unwrap(); + rt.tick(); // register actor + (rt, addr) + }, + |(rt, addr)| { + rt.send_to(addr, NoopMessage).unwrap(); + }, + BatchSize::SmallInput, + ); + }); + + // A4 — Inbox creation + group.bench_function("inbox_creation", |b| { + b.iter_batched( + || Runtime::new(make_config(1_000, 1_000)), + |rt| { + rt.new_inbox::().unwrap(); + }, + BatchSize::SmallInput, + ); + }); + + group.finish(); +} + +// --------------------------------------------------------------------------- +// Throughput benchmarks +// --------------------------------------------------------------------------- + +fn throughput_benchmarks(c: &mut Criterion) { + let mut group = c.benchmark_group("throughput"); + + // B1 — Single-actor throughput + for n in [100, 1_000, 10_000] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::new("single_actor", n), &n, |b, &n| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(100, n + 100)); + let addr = rt.spawn(SinkActor).unwrap(); + rt.tick(); // register actor + for i in 0..n { + rt.send_to(addr, CountMessage(i as u64)).unwrap(); + } + rt + }, + |rt| { + for _ in 0..50 { + rt.tick(); + } + }, + BatchSize::LargeInput, + ); + }); + } + + // B2 — Multi-actor throughput + for (actors, msgs_per) in [(10, 100), (100, 100), (100, 1_000)] { + let total = actors * msgs_per; + group.throughput(Throughput::Elements(total as u64)); + let param = format!("{actors}x{msgs_per}"); + group.bench_with_input(BenchmarkId::new("multi_actor", ¶m), &(actors, msgs_per), |b, &(actors, msgs_per)| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(actors + 100, msgs_per + 100)); + let addrs: Vec<_> = (0..actors) + .map(|_| rt.spawn(SinkActor).unwrap()) + .collect(); + rt.tick(); // register actors + for &addr in &addrs { + for i in 0..msgs_per { + rt.send_to(addr, CountMessage(i as u64)).unwrap(); + } + } + rt + }, + |rt| { + for _ in 0..100 { + rt.tick(); + } + }, + BatchSize::LargeInput, + ); + }); + } + + // B3 — Ring throughput + for ring_size in [10usize, 100, 500] { + group.throughput(Throughput::Elements((ring_size + 1) as u64)); + group.bench_with_input(BenchmarkId::new("ring", ring_size), &ring_size, |b, &ring_size| { + b.iter_batched( + || { + let rt = Runtime::new(make_config(ring_size + 100, 100)); + let inbox = rt.new_inbox::().unwrap(); + // Build the ring: last actor sends to inbox, each prior actor sends to the next + let mut next_addr = *inbox.addr(); + let mut entry_addr = next_addr; + for _ in 0..ring_size { + let addr = rt.spawn(RingActor { next: next_addr }).unwrap(); + entry_addr = addr; + next_addr = addr; + } + rt.tick(); // register all actors + (rt, entry_addr, inbox) + }, + |(rt, entry_addr, inbox)| { + rt.send_to(entry_addr, RingMessage { hops: 0 }).unwrap(); + for _ in 0..(ring_size + 10) { + rt.tick(); + if inbox.try_recv().is_some() { + return; + } + } + panic!("RingMessage not received within tick budget"); + }, + BatchSize::LargeInput, + ); + }); + } + + // B4 — Spawn throughput + for n in [100, 1_000, 5_000] { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::new("spawn", n), &n, |b, &n| { + b.iter_batched( + || Runtime::new(make_config(n + 100, 1_000)), + |rt| { + for _ in 0..n { + rt.spawn(NoopActor).unwrap(); + } + }, + BatchSize::LargeInput, + ); + }); + } + + group.finish(); +} + +criterion_group!(benches, latency_benchmarks, throughput_benchmarks); +criterion_main!(benches); diff --git a/benches/scaling.rs b/benches/scaling.rs deleted file mode 100644 index e54bf64..0000000 --- a/benches/scaling.rs +++ /dev/null @@ -1,280 +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::{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: &Runtime, _msg: Increment) { - self.count += 1; - } -} - -struct SharedCounter { - count: std::sync::Arc, -} - -impl ActorInterface for SharedCounter { - type Incoming = Increment; - type Response = (); - - fn handle(&mut self, _ctx: &Runtime, _msg: Increment) { - self.count - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - } -} - -/// An actor that handles variable-sized payloads -struct PayloadActor { - bytes_received: usize, -} - -impl PayloadActor { - fn new() -> Self { - Self { bytes_received: 0 } - } -} - -#[derive(Clone)] -struct Payload(Vec); - -impl ActorInterface for PayloadActor { - type Incoming = Payload; - type Response = (); - - fn handle(&mut self, _ctx: &Runtime, 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) { - let messages_per_actor = 100u64; - - for actor_count in [10u64, 100, 500, 1000] { - let name = format!("scaling_{}_actors", actor_count); - let total_messages = actor_count * messages_per_actor; - - let result = Bench::new(&name) - .warmup(2) - .iters(10) - .elements(total_messages) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: (actor_count as usize) + 100, - router_max_messages: (total_messages as usize) * 3, - actor_max_messages: (messages_per_actor as usize) * 2, - num_threads: 1, - }; - 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 registrations - for _ in 0..(actor_count * 2) { - runtime.tick(); - } - - (runtime, actors, messages_per_actor) - }, - |(runtime, actors, msgs_per)| { - // Distribute messages across all actors - for _ in 0..msgs_per { - for actor in &actors { - let _ = runtime.send_to::(*actor, Increment); - } - } - - // Process all - let total = actors.len() as u64 * msgs_per; - for _ in 0..(total * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: How throughput scales with thread count (multithreaded runtime) -pub fn bench_thread_count_scaling(suite: &mut BenchSuite) { - let actor_count = 100u64; - let messages_per_actor = 500u64; - let total_messages = actor_count * messages_per_actor; - - for thread_count in [2usize, 4, 8] { - let name = format!("scaling_{}_threads", thread_count); - - let result = Bench::new(&name) - .warmup(1) - .iters(5) - .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, - router_max_messages: (total_messages as usize) * 3, - actor_max_messages: (messages_per_actor as usize) * 2, - num_threads: thread_count, - }; - let runtime = Runtime::new(config); - - let mut actors = Vec::with_capacity(actor_count as usize); - for _ in 0..actor_count { - let addr = runtime - .spawn(SharedCounter { - count: counter.clone(), - }) - .unwrap(); - actors.push(addr); - } - - let handle = runtime.run().unwrap(); - - for actor in &actors { - loop { - if handle - .runtime - .send_to::(*actor, Increment) - .is_ok() - { - break; - } - thread::yield_now(); - } - } - - while counter.load(std::sync::atomic::Ordering::Relaxed) < actors.len() { - thread::yield_now(); - } - counter.store(0, std::sync::atomic::Ordering::Relaxed); - - (handle, actors, counter) - }, - |(handle, actors, counter)| { - for _ in 0..messages_per_actor { - for actor in &actors { - let _ = handle.runtime.send_to::(*actor, Increment); - } - } - - while counter.load(std::sync::atomic::Ordering::Relaxed) - < total_messages as usize - { - thread::yield_now(); - } - - handle.shutdown(); - handle.join(); - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: How throughput scales with message payload size -pub fn bench_payload_size_scaling(suite: &mut BenchSuite) { - let message_count = 1_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(2) - .iters(20) - .elements(message_count) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: 10, - router_max_messages: (message_count as usize) * 2, - actor_max_messages: (message_count as usize) * 2, - num_threads: 1, - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(PayloadActor::new()).unwrap(); - - // Process registration - for _ in 0..10 { - runtime.tick(); - } - - (runtime, sink, payload.clone()) - }, - |(runtime, sink, payload)| { - for _ in 0..message_count { - let _ = runtime.send_to::(sink, Payload(payload.clone())); - } - - for _ in 0..(message_count * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Run all scaling benchmarks -pub fn run_all() -> BenchSuite { - let mut suite = BenchSuite::new("Scaling Benchmarks"); - - println!("\nRunning actor count scaling benchmarks..."); - bench_actor_count_scaling(&mut suite); - - println!("Running thread count scaling benchmarks..."); - bench_thread_count_scaling(&mut suite); - - println!("Running payload size scaling benchmarks..."); - bench_payload_size_scaling(&mut suite); - - suite -} diff --git a/benches/throughput.rs b/benches/throughput.rs deleted file mode 100644 index 58c67b3..0000000 --- a/benches/throughput.rs +++ /dev/null @@ -1,343 +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::{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: &Runtime, _msg: Ping) { - self.count += 1; - } -} - -/// A forwarding actor that passes messages along a chain -struct ForwardActor { - next: Option, -} - -impl ForwardActor { - fn new() -> Self { - Self { next: None } - } - - fn with_next(next: ActorAddress) -> Self { - Self { next: Some(next) } - } -} - -impl ActorInterface for ForwardActor { - type Incoming = Ping; - type Response = Ping; - - fn handle(&mut self, ctx: &Runtime, msg: Ping) { - if let Some(next) = self.next { - let _ = ctx.send_to(next, msg); - } - } -} - -// ============================================================================ -// Benchmarks -// ============================================================================ - -/// Benchmark: Messages sent through the router 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(3) - .iters(20) - .elements(msg_count) - .run_with_setup( - || { - // Setup: create runtime and sink actor - let config = RuntimeConfig { - max_actors: 100, - router_max_messages: (msg_count as usize) * 2, - actor_max_messages: (msg_count as usize) * 2, - num_threads: 1, - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(SinkActor::new()).unwrap(); - (runtime, sink, msg_count) - }, - |(runtime, sink, count)| { - // Send all messages - for _ in 0..count { - let _ = runtime.send_to::(sink, Ping); - } - // Process until done - // Tick enough times to process all messages - // (router tick + actor tick) * messages / WATERLEVEL - 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(3) - .iters(50) - .elements(actor_count) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: 1000, - router_max_messages: 10_000, - actor_max_messages: 100, - num_threads: 1, - }; - Runtime::new(config) - }, - |runtime| { - for _ in 0..actor_count { - let _ = runtime.spawn(SinkActor::new()); - } - // Process router messages to register all actors - 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 in [10u64, 100, 500] { - let name = format!("fanout_1_to_{}", fan_count); - let messages_per_receiver = 100u64; - - let result = Bench::new(&name) - .warmup(2) - .iters(20) - .elements(fan_count * messages_per_receiver) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: (fan_count as usize) + 10, - router_max_messages: (fan_count as usize) - * (messages_per_receiver as usize) - * 2, - actor_max_messages: (messages_per_receiver as usize) * 2, - num_threads: 1, - }; - 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 router registrations - for _ in 0..(fan_count * 2) { - runtime.tick(); - } - - (runtime, sinks, messages_per_receiver) - }, - |(runtime, sinks, msgs_per)| { - // Send messages to all sinks - for _ in 0..msgs_per { - for sink in &sinks { - let _ = runtime.send_to::(*sink, Ping); - } - } - - // Process all messages - let total_msgs = sinks.len() as u64 * msgs_per; - for _ in 0..(total_msgs * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: Fan-in (N senders to 1 receiver) -pub fn bench_fanin(suite: &mut BenchSuite) { - for sender_count in [10u64, 100, 500] { - let name = format!("fanin_{}_to_1", sender_count); - let messages_per_sender = 100u64; - - let result = Bench::new(&name) - .warmup(2) - .iters(20) - .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, - router_max_messages: total_messages * 3, - actor_max_messages: total_messages * 2, - num_threads: 1, - }; - 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 router registrations - for _ in 0..((sender_count + 1) * 2) { - runtime.tick(); - } - - (runtime, senders, sink, messages_per_sender) - }, - |(runtime, senders, _sink, msgs_per)| { - // Each sender forwards msgs_per messages to the sink - for _ in 0..msgs_per { - for sender in &senders { - let _ = runtime.send_to::(*sender, Ping); - } - } - - // Process all messages (forwarder receives + forwards, sink receives) - let total_msgs = senders.len() as u64 * msgs_per; - for _ in 0..(total_msgs * 6) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Benchmark: Ring topology (message passed around N actors in a circle) -pub fn bench_ring(suite: &mut BenchSuite) { - for ring_size in [10u64, 100, 500] { - let name = format!("ring_{}_actors", ring_size); - let laps = 10u64; // How many times around the ring - - let result = Bench::new(&name) - .warmup(2) - .iters(20) - .elements(ring_size * laps) - .run_with_setup( - || { - let config = RuntimeConfig { - max_actors: (ring_size as usize) + 10, - router_max_messages: 10_000, - actor_max_messages: 1_000, - num_threads: 1, - }; - let runtime = Runtime::new(config); - - // First, spawn all actors without links - let mut actors: Vec = Vec::with_capacity(ring_size as usize); - for _ in 0..ring_size { - let addr = runtime.spawn(ForwardActor::new()).unwrap(); - actors.push(addr); - } - - // We can't update their `next` field after spawn in this design, - // so instead we'll use an inbox to receive the final message - // For now, we'll just measure message passing through a chain - - // Process registrations - 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 router + inbox overhead) - for _ in 0..laps { - for actor in &actors { - let _ = runtime.send_to::(*actor, Ping); - } - } - - let total = actors.len() as u64 * laps; - for _ in 0..(total * 3) { - runtime.tick(); - } - black_box(()); - }, - ); - - suite.add(result); - } -} - -/// Run all throughput benchmarks -pub fn run_all() -> BenchSuite { - let mut suite = BenchSuite::new("Throughput Benchmarks"); - - println!("\nRunning message throughput benchmarks..."); - bench_message_throughput(&mut suite); - - println!("Running spawn rate benchmarks..."); - bench_spawn_rate(&mut suite); - - println!("Running fan-out benchmarks..."); - bench_fanout(&mut suite); - - println!("Running fan-in benchmarks..."); - bench_fanin(&mut suite); - - println!("Running ring topology benchmarks..."); - bench_ring(&mut suite); - - suite -} diff --git a/examples/hello.rs b/examples/hello.rs index d67c439..7f9b483 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Runtime, RuntimeConfig}, + runtime::{Ctx, Runtime, RuntimeConfig}, }; #[derive(Debug, Default)] @@ -24,10 +24,10 @@ impl ActorInterface for Greeter { type Incoming = GreetMessage; type Response = GreetResponse; - fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) { + fn handle(&mut self, ctx: &Ctx, msg: GreetMessage) { let res = GreetResponse(format!("Hello, {}!", msg.who)); self.num_greeted += 1; - if let Err(_) = ctx.send_to(msg.return_addr, res) { + if let Err(_) = ctx.send(msg.return_addr, res) { // no error handling self.num_greeted -= 1; } diff --git a/examples/ring.rs b/examples/ring.rs index d3c6031..dd4e87b 100644 --- a/examples/ring.rs +++ b/examples/ring.rs @@ -1,6 +1,6 @@ use swactor::{ actor::{ActorAddress, ActorInterface}, - runtime::{Inbox, Runtime, RuntimeConfig}, + runtime::{Ctx, Inbox, Runtime, RuntimeConfig}, }; #[derive(Debug, Default, Clone)] @@ -30,8 +30,8 @@ impl RingActor { impl ActorInterface for RingActor { type Incoming = RingMessage; type Response = (); - fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming) { - if let Err(_) = ctx.send_to(self.next, msg.next()) { + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { + if let Err(_) = ctx.send(self.next, msg.next()) { // do nothing } } diff --git a/src/actor.rs b/src/actor.rs index 3450bee..7a7a9b8 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -1,50 +1,15 @@ -use crate::{WATERLEVEL, channel::Receiver, get_random, runtime::Runtime}; +use std::any::Any; + +use crate::{get_random, runtime::{ContextInner, Ctx}, worker::Mailbox}; /// The primary trait defining data that can be passed to and from actor processes pub trait Message: 'static + Sized + Clone + Send + Sync {} impl Message for T {} -/// The trait that needs to be implemented in order to run a process as an `Actor` -/// -/// The `Incoming` type represents `Messages` that can be delivered to the `Actor`. -/// -/// The `Response` type represents possible `Messages` the actor may attempt to reply with. -/// -/// The `fn handle(..)` is where you implement the logic for handling `Incoming` messages -/// -/// # Example -/// ``` -/// use swactor::{actor::{ActorAddress, ActorInterface}, runtime::Runtime}; -/// -/// 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: &Runtime, msg: Self::Incoming) { -/// let response = GreetResponse(format!("Hello, {}!", msg.who).to_string()); -/// if let Ok(_) = ctx.send_to(msg.return_addr, response) { -/// self.num_greeted += 1; -/// } -/// } -/// } -/// ``` pub trait ActorInterface: 'static + Send { type Incoming: Message; type Response: Message; - fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming); } /// A unique address for this actor. 32 bytes is overkill for a small application, @@ -60,47 +25,58 @@ impl ActorAddress { } } -/// The actor process as represented in the Runtime, with the actor state stored with it's inbox. +/// The actor process as represented in the Runtime, with the actor state stored with its mailbox. pub(crate) struct Actor where A: ActorInterface, { - inbox: Receiver, + addr: ActorAddress, + mailbox: Mailbox, inner: A, } impl Actor { - pub(crate) fn new(inbox: Receiver, inner: A) -> Self { - Self { inbox, inner } + pub(crate) fn new(addr: ActorAddress, mailbox: Mailbox, inner: A) -> Self { + Self { + addr, + mailbox, + inner, + } } } /// Trait for type-erased actors pub(crate) trait AnyActor: Send { - fn tick(&mut self, ctx: &Runtime); + /// Tick the actor, processing pending messages. Returns `true` if any work was done. + fn tick(&mut self, inner: &dyn ContextInner) -> bool; + /// Deliver a type-erased message into this actor's mailbox. + /// Returns `true` if the downcast succeeded. + fn deliver(&mut self, msg: Box) -> bool; } impl AnyActor for Actor where A: ActorInterface, { - fn tick(&mut self, ctx: &Runtime) { - // TODO: WATERLEVEL is hard coded, and so is this message handling scheme. We should - // make it so both are more flexible, with sane defaults. - let total_messages = self.inbox.len(); - let messages_to_process = if total_messages < WATERLEVEL { - total_messages - } else { - total_messages >> 1 - }; - - for _ in 0..messages_to_process { - match self.inbox.try_recv() { - Some(msg) => self.inner.handle(ctx, msg), - None => unreachable!( - "We checked number of unprocessed messages in the queue ahead of processing" - ), + fn tick(&mut self, inner: &dyn ContextInner) -> bool { + let n = self.mailbox.drain_count(); + if n > 0 { + let ctx = Ctx::new(inner, self.addr); + for _ in 0..n { + if let Some(msg) = self.mailbox.pop() { + self.inner.handle(&ctx, msg); + } } } + n > 0 + } + + fn deliver(&mut self, msg: Box) -> bool { + if let Ok(typed) = msg.downcast::() { + self.mailbox.push(*typed); + true + } else { + false + } } } diff --git a/src/address_map.rs b/src/address_map.rs new file mode 100644 index 0000000..48ac949 --- /dev/null +++ b/src/address_map.rs @@ -0,0 +1,120 @@ +use std::collections::HashMap; +use std::sync::RwLock; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use crate::actor::ActorAddress; + +/// Identifies a worker thread. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) struct WorkerId(pub(crate) usize); + +impl WorkerId { + pub fn as_usize(self) -> usize { + self.0 + } +} + +/// Maps actor addresses to the worker that owns them. +/// +/// `RwLock` — zero contention for parallel reads, write-rare (only on spawn). +pub(crate) struct AddressMap { + inner: RwLock>, +} + +impl AddressMap { + pub fn new() -> Self { + Self { + inner: RwLock::new(HashMap::new()), + } + } + + pub fn with_capacity(cap: usize) -> Self { + Self { + inner: RwLock::new(HashMap::with_capacity(cap)), + } + } + + pub fn insert(&self, addr: ActorAddress, worker: WorkerId) { + self.inner.write().unwrap().insert(addr, worker); + } + + pub fn remove(&self, addr: &ActorAddress) { + self.inner.write().unwrap().remove(addr); + } + + pub fn lookup(&self, addr: &ActorAddress) -> Option { + self.inner.read().unwrap().get(addr).copied() + } + + pub fn len(&self) -> usize { + self.inner.read().unwrap().len() + } +} + +/// Round-robin actor placement strategy. +pub(crate) struct Placement { + next: AtomicUsize, + num_workers: usize, +} + +impl Placement { + pub fn new(num_workers: usize) -> Self { + Self { + next: AtomicUsize::new(0), + num_workers, + } + } + + pub fn next_worker(&self) -> WorkerId { + let id = self.next.fetch_add(1, Ordering::Relaxed) % self.num_workers; + WorkerId(id) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn insert_and_lookup() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + let wid = WorkerId(3); + map.insert(addr, wid); + assert_eq!(map.lookup(&addr), Some(wid)); + } + + #[test] + fn lookup_missing_returns_none() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + assert_eq!(map.lookup(&addr), None); + } + + #[test] + fn remove_works() { + let map = AddressMap::new(); + let addr = ActorAddress::default(); + map.insert(addr, WorkerId(0)); + map.remove(&addr); + assert_eq!(map.lookup(&addr), None); + } + + #[test] + fn len_tracks_entries() { + let map = AddressMap::with_capacity(10); + assert_eq!(map.len(), 0); + let addr1 = ActorAddress::default(); + map.insert(addr1, WorkerId(0)); + assert_eq!(map.len(), 1); + } + + #[test] + fn round_robin() { + let p = Placement::new(3); + assert_eq!(p.next_worker(), WorkerId(0)); + assert_eq!(p.next_worker(), WorkerId(1)); + assert_eq!(p.next_worker(), WorkerId(2)); + assert_eq!(p.next_worker(), WorkerId(0)); + } +} diff --git a/src/channel.rs b/src/channel.rs index a210a93..c487ec4 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -1,17 +1,18 @@ -use std::{collections::VecDeque, sync::{Arc, Mutex}}; -use crossbeam_queue::ArrayQueue; +use std::sync::Arc; + +use crossbeam_queue::{ArrayQueue, SegQueue}; pub struct HybridChannel { ring: ArrayQueue, - overflow: Mutex>, + overflow: SegQueue, } impl HybridChannel { pub fn new(capacity: usize) -> Self { Self { ring: ArrayQueue::new(capacity), - overflow: Mutex::new(VecDeque::new()), + overflow: SegQueue::new(), } } @@ -19,7 +20,7 @@ impl HybridChannel { match self.ring.push(value) { Ok(()) => Ok(()), Err(v) => { - self.overflow.lock().unwrap().push_back(v); + self.overflow.push(v); Ok(()) } } @@ -30,12 +31,14 @@ impl HybridChannel { return Some(value); } - self.overflow.lock().unwrap().pop_front() + match self.overflow.pop() { + Some(value) => { + Some(value) + } + None => None, + } } - pub fn len(&self) -> usize { - self.ring.len() + self.overflow.lock().unwrap().len() - } } pub(crate) struct Receiver { @@ -48,11 +51,6 @@ impl Receiver { Self { queue } } - - pub fn len(&self) -> usize { - self.queue.len() - } - pub fn try_recv(&self) -> Option { return self.queue.pop(); } diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..95eab7b --- /dev/null +++ b/src/config.rs @@ -0,0 +1,55 @@ +/// Backoff policy for worker threads when idle. +/// +/// Workers spin → yield → sleep with increasing delay when no work is available. +pub struct BackoffPolicy { + /// Number of idle ticks before switching from spin to yield. + pub spin_threshold: u32, + /// Number of idle ticks before switching from yield to sleep. + pub yield_threshold: u32, + /// Microseconds added per tick beyond the yield threshold. + pub sleep_increment_us: u64, + /// Maximum sleep duration in microseconds. + pub sleep_max_us: u64, +} + +impl Default for BackoffPolicy { + fn default() -> Self { + Self { + spin_threshold: 64, + yield_threshold: 256, + sleep_increment_us: 50, + sleep_max_us: 1000, + } + } +} + +/// The tunable settings for the runtime. +pub struct RuntimeConfig { + pub max_actors: usize, + pub actor_max_messages: usize, + pub num_threads: usize, + pub mailbox_waterlevel: usize, + pub backoff_policy: BackoffPolicy, +} + +/// 8kB for the `Box<..>` before counting the rest of the memory +const DEFAULT_MAX_ACTORS: usize = 1_000; + +/// 16kB PER ACTOR to alloc space for storing messages. +/// With default setting of [DEFAULT_MAX_ACTORS] this is: +/// 1_000 * 16kB = 16MB +const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000; + +const DEFAULT_MAILBOX_WATERLEVEL: usize = 10; + +impl Default for RuntimeConfig { + fn default() -> Self { + Self { + max_actors: DEFAULT_MAX_ACTORS, + actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, + num_threads: 1, + mailbox_waterlevel: DEFAULT_MAILBOX_WATERLEVEL, + backoff_policy: BackoffPolicy::default(), + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 1d67009..fc57124 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,10 +1,14 @@ pub mod actor; +pub mod worker; -mod channel; +pub(crate) mod channel; pub(crate) mod error; pub use error::Error; -mod router; + +pub(crate) mod address_map; +pub mod config; + pub mod runtime; #[cfg(feature = "getrandom")] @@ -25,14 +29,3 @@ pub(crate) fn get_random(buf: &mut [u8]) { *byte = bytes[i % core::mem::size_of::()]; } } - -/// FIXME: remove hard coded defaults -/// The strategy for message processing is such: -/// -/// ```ignore -/// if total_messages < WATERLEVEL: -/// process all -/// else -/// process total_messages >> 1 -/// ``` -const WATERLEVEL: usize = 10; diff --git a/src/router.rs b/src/router.rs deleted file mode 100644 index 55296f0..0000000 --- a/src/router.rs +++ /dev/null @@ -1,75 +0,0 @@ -use std::{collections::HashMap, sync::Arc}; - -use crate::{ - actor::{ActorAddress, ActorInterface, Message}, - channel::Sender, - runtime::Runtime, -}; - -/// FIXME: Go over with a fine-toothed comb and reassure yourself this typing -/// makes sense, that we are not doing loads of indirection on a hot path. -/// -/// A type erased `Message` to be routed between actor processes. -pub(crate) type Envelope = Arc; - -pub(crate) trait SenderT: Send + Sync { - fn try_send(&self, envelope: Envelope); -} - -impl SenderT for Sender { - fn try_send(&self, envelope: Envelope) { - if let Some(msg) = envelope.downcast_ref::() { - let _ = Sender::try_send(self, msg.clone()); - } - } -} - -/// Internal messages for the Router's own inbox -#[derive(Clone)] -pub(crate) enum RouterMessage { - /// register addrs with sender - AddAddr(ActorAddress, Arc), - - /// FIXME: this will be active when we allow actors to shut themselves - /// down. For now, disable the warning. - #[allow(dead_code)] - /// remove an actor from the address book - RemoveAddr(ActorAddress), - - /// send to - SendToAddr { addr: ActorAddress, msg: Envelope }, -} - -/// The `Router` is responsible for taking in and delivering all messages in the runtime. -pub(crate) struct Router { - directory: HashMap>, -} - -impl Router { - pub fn new() -> Self { - Self { - directory: HashMap::new(), - } - } -} - -impl ActorInterface for Router { - type Incoming = RouterMessage; - type Response = (); - - fn handle(&mut self, _ctx: &Runtime, msg: Self::Incoming) { - match msg { - RouterMessage::AddAddr(addr, sender) => { - self.directory.insert(addr, sender); - } - RouterMessage::RemoveAddr(addr) => { - self.directory.remove(&addr); - } - RouterMessage::SendToAddr { addr, msg } => { - if let Some(sender) = self.directory.get(&addr) { - sender.try_send(msg); - } - } - } - } -} diff --git a/src/runtime.rs b/src/runtime.rs index 7a9f768..d667df6 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,14 +1,19 @@ +use std::any::Any; +use std::cell::RefCell; +use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, RwLock}; use std::thread::{self, JoinHandle}; -use crate::channel::HybridChannel; -use crate::{ - actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}, - channel::{Receiver, Sender}, - router::{Router, RouterMessage}, - Error, -}; +use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message}; +use crate::address_map::{AddressMap, Placement, WorkerId}; +use crate::channel::{Receiver, Sender}; +// Re-export config types so existing code using `runtime::RuntimeConfig` still works +pub use crate::config::{BackoffPolicy, RuntimeConfig}; +use crate::worker::Mailbox; +use crate::worker::{TickContext, Worker}; +use crate::Error; + /// Generic message inbox for receiving messages outside of the runtime. pub struct Inbox { @@ -26,47 +31,6 @@ impl Inbox { } } -/// The tunable settings for the runtime. -pub struct RuntimeConfig { - pub max_actors: usize, - pub router_max_messages: usize, - pub actor_max_messages: usize, - pub num_threads: usize, -} - -/// 8kB for the `Box<..>` before counting the rest of the memory -const DEFAULT_MAX_ACTORS: usize = 1_000; - -/// 160kB for the `Arc<..>` before counting the rest of the memory -const DEFAULT_ROUTER_MAX_MESSAGES: usize = 10_000; - -/// 16kB PER ACTOR to alloc space for storing the `Arc<..>` pointers -/// With default setting of [DEFAULT_MAX_ACTORS] this is: -/// 1_000 * 16kB = 16MB -const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000; - -impl Default for RuntimeConfig { - fn default() -> Self { - Self { - max_actors: DEFAULT_MAX_ACTORS, - router_max_messages: DEFAULT_ROUTER_MAX_MESSAGES, - actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES, - num_threads: 1, - } - } -} - -/// The `Runtime` struct is the primary gateway for interacting with the framework. -pub struct Runtime { - config: RuntimeConfig, - actor_queue: HybridChannel>, - router_interface: Sender, - router: Option>, // `None` if single-threaded - - // for multithreaded contexts - is_running: AtomicBool, -} - /// Handle for dealing with a runtime that has started via the `Runtime::run()` method. pub struct RuntimeHandle { pub runtime: Arc, @@ -86,97 +50,194 @@ impl RuntimeHandle { } } +/// Actor syscall interface — passed to `ActorInterface::handle()`. +/// +/// Wraps a `&dyn ContextInner` to solve the object-safety problem while +/// providing a typed public API. +pub struct Ctx<'a> { + inner: &'a dyn ContextInner, + self_addr: ActorAddress, +} + +impl<'a> Ctx<'a> { + pub(crate) fn new(inner: &'a dyn ContextInner, self_addr: ActorAddress) -> Self { + Self { inner, self_addr } + } + + /// Returns the address of the actor currently being ticked. + pub fn self_addr(&self) -> ActorAddress { + self.self_addr + } + + /// Send a typed message to an actor address. + pub fn send(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { + self.inner.send_any(addr, Box::new(msg)) + } + + /// Spawn a new actor, returning its address. + pub fn spawn(&self, actor: A) -> Result { + let addr = ActorAddress::new_random(); + let waterlevel = self.inner.mailbox_waterlevel(); + let actor = Actor::new(addr, Mailbox::new(waterlevel), actor); + let boxed: Box = Box::new(actor); + self.inner.spawn_any(addr, boxed)?; + Ok(addr) + } +} + +/// Type-erased sender for external inboxes. +pub(crate) trait SenderT: Send + Sync { + fn try_send_any(&self, msg: Box); +} + +impl SenderT for Sender { + fn try_send_any(&self, msg: Box) { + if let Ok(typed) = msg.downcast::() { + let _ = Sender::try_send(self, *typed); + } + } +} + + +// ─── Runtime ───────────────────────────────────────────────────────────────── + +/// The `Runtime` struct is the primary gateway for interacting with the framework. +pub struct Runtime { + config: RuntimeConfig, + address_map: Arc, + inbox_registry: Arc, + transfer_txs: Vec>, + spawn_txs: Vec)>>, + placement: Placement, + is_running: AtomicBool, + /// Single-threaded mode: worker stored inline + single_worker: Option>, + /// Multi-threaded mode: workers waiting to be assigned to threads by run() + pending_workers: Option>, +} + +// Safety: RefCell is only accessed from the thread that owns the Runtime +// in single-threaded mode. In multi-threaded mode, single_worker is None and +// pending_workers is consumed by run() before Arc sharing. +unsafe impl Sync for Runtime {} + impl Runtime { /// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call /// `run()`, if single threaded, needs to be driven by calls to the `tick()` method. pub fn new(config: RuntimeConfig) -> Self { - let actor_queue = HybridChannel::new(config.max_actors); - - // router is a unique actor in that the runtime needs access to it's `Sender` handle - let router_inner = Router::new(); - let router_inbox: Receiver = - Receiver::<::Incoming>::new(config.router_max_messages); - let router_sender = router_inbox.new_sender(); - let router = Actor::new(router_inbox, router_inner); - - // Single-threaded: router goes in queue. Multi-threaded: stays in Option - let router_option = if config.num_threads < 2 { - actor_queue - .push(Box::new(router) as Box) - .map_err(|_| "failed to add router to actor queue") - .expect("failed to spawn router at runtime initialization."); - None + let num_workers = if config.num_threads < 2 { + 1 } else { - Some(router) + config.num_threads }; - Self { - config, - actor_queue, - router_interface: router_sender, - is_running: AtomicBool::new(false), - router: router_option, + let address_map = Arc::new(AddressMap::with_capacity(config.max_actors)); + let inbox_registry = Arc::new(InboxRegistry::new()); + let placement = Placement::new(num_workers); + + let mut transfer_txs = Vec::with_capacity(num_workers); + let mut spawn_txs = Vec::with_capacity(num_workers); + let mut workers = Vec::with_capacity(num_workers); + + for i in 0..num_workers { + let transfer_rx = Receiver::::new(config.actor_max_messages); + let transfer_tx = transfer_rx.new_sender(); + transfer_txs.push(transfer_tx); + + let spawn_rx = + Receiver::<(ActorAddress, Box)>::new(config.max_actors); + let spawn_tx = spawn_rx.new_sender(); + spawn_txs.push(spawn_tx); + + workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx)); + } + + if config.num_threads < 2 { + // Single-threaded: store one worker inline + let worker = workers.remove(0); + Self { + config, + address_map, + inbox_registry, + transfer_txs, + spawn_txs, + placement, + is_running: AtomicBool::new(false), + single_worker: Some(RefCell::new(worker)), + pending_workers: None, + } + } else { + // Multi-threaded: stash workers for run() + Self { + config, + address_map, + inbox_registry, + transfer_txs, + spawn_txs, + placement, + is_running: AtomicBool::new(false), + single_worker: None, + pending_workers: Some(workers), + } } } /// Spawn an actor, returns its address pub fn spawn(&self, actor: A) -> Result { - // assign a stochastic let addr = ActorAddress::new_random(); - let inbox = Receiver::::new(self.config.actor_max_messages); - let sender = inbox.new_sender(); - - // Register the sender with the router - self.router_interface - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))) - .map_err(|_| { - Error::from("Runtime error: failed to add actor to router. Router inbox full") - })?; - - self.actor_queue - .push(Box::new(Actor::new(inbox, actor))) - .map_err(|_| Error::from("Runtime error: Failed to spawn actor. Queue full."))?; - + let worker_id = self.placement.next_worker(); + self.address_map.insert(addr, worker_id); + let actor = Actor::new(addr, Mailbox::new(self.config.mailbox_waterlevel), actor); + let boxed: Box = Box::new(actor); + self.spawn_txs[worker_id.as_usize()] + .try_send((addr, boxed)) + .map_err(|_| Error::from("Runtime error: spawn queue full"))?; Ok(addr) } /// Send a message to an actor address pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), Error> { - self.router_interface - .try_send(RouterMessage::SendToAddr { - addr, - msg: Arc::new(msg), - }) - .map_err(|_| Error::from("Failed to send message to router.")) + let msg_box: Box = Box::new(msg); + match self.address_map.lookup(&addr) { + Some(wid) => self.transfer_txs[wid.as_usize()] + .try_send(Envelope::new(addr, msg_box)) + .map_err(|_| Error::from("Transfer queue full")), + None => self.inbox_registry.try_deliver(addr, msg_box), + } } /// Create an external inbox for receiving messages in the outer process containing the runtime pub fn new_inbox(&self) -> Result, Error> { let addr = ActorAddress::new_random(); - let receiver = Receiver::::new(self.config.actor_max_messages); let sender = receiver.new_sender(); - - // Register the sender with the router - self.router_interface - .try_send(RouterMessage::AddAddr(addr, Arc::new(sender))) - .map_err(|_| { - Error::from( - "Runtime error: failed to add a new inbox channel. Router inbox is full.", - ) - })?; - + self.inbox_registry.register(addr, Arc::new(sender)); Ok(Inbox { addr, inner: receiver, }) } + /// Drive one tick of the single-threaded worker. + pub fn tick(&self) { + if let Some(ref worker) = self.single_worker { + let tc = TickContext { + address_map: &self.address_map, + transfer_txs: &self.transfer_txs, + spawn_txs: &self.spawn_txs, + placement: &self.placement, + inbox_registry: &self.inbox_registry, + config: &self.config, + }; + worker.borrow_mut().tick_once(&tc); + } + } + /// Spawn worker threads and start processing, returning a set of handles and /// a Runtime object to interface with. /// /// ### WARN: - /// ##### This function panics if the configuration is set as single threaded + /// ##### Returns an error if the configuration is set as single threaded /// `config.num_threads == 1` pub fn run(mut self) -> Result { if self.config.num_threads < 2 { @@ -187,53 +248,26 @@ impl Runtime { self.is_running.store(true, Ordering::Release); - // Take router out before wrapping in Arc - it will be owned by router thread - let mut router = self - .router + let workers = self + .pending_workers .take() - .expect("Router must be present for multi-threaded runtime"); + .expect("Workers must be present for multi-threaded runtime"); let rt = Arc::new(self); - let mut handles: Vec> = vec![]; + let mut handles: Vec> = Vec::with_capacity(workers.len()); - // Router thread owns the router directly - no synchronization needed - let router_handle = { - let ctx = rt.clone(); - thread::spawn(move || { - while ctx.is_running.load(Ordering::Acquire) { - router.tick(&ctx); - thread::yield_now(); - } - }) - }; - handles.push(router_handle); - - // Spawn worker threads - let num_workers = rt.config.num_threads - 1; - for _ in 0..num_workers { - let ctx = rt.clone(); + for mut worker in workers { + let rt_clone = rt.clone(); let handle = thread::spawn(move || { - while ctx.is_running.load(Ordering::Acquire) { - if let Some(mut actor) = ctx.actor_queue.pop() { - actor.tick(&ctx); - // FIXME: Justify this loop. It is here to prevent panics when the - // actor queue is full, but results in a spinlock. - loop { - match ctx.actor_queue.push(actor) { - Ok(()) => break, - Err(a) => { - actor = a; - if !ctx.is_running.load(Ordering::Acquire) { - break; - } - thread::yield_now(); - } - } - } - } else { - thread::yield_now(); - } - } + let tc = TickContext { + address_map: &rt_clone.address_map, + transfer_txs: &rt_clone.transfer_txs, + spawn_txs: &rt_clone.spawn_txs, + placement: &rt_clone.placement, + inbox_registry: &rt_clone.inbox_registry, + config: &rt_clone.config, + }; + worker.run(&tc, &rt_clone.is_running, &rt_clone.config.backoff_policy); }); handles.push(handle); } @@ -244,17 +278,105 @@ impl Runtime { }) } - /// Pop the actor off the top of the queue and process it's messages, returning it to the back of - /// the queue upon completion. - pub fn tick(&self) { - if let Some(mut actor) = self.actor_queue.pop() { - actor.tick(&self); - let _ = self.actor_queue.push(actor); - } - } - /// Signal all workers to stop pub fn shutdown(&self) { self.is_running.store(false, Ordering::Release); } } + + + +/// A type-erased message envelope for cross-worker delivery. +/// +/// Uses `Box` (no atomic refcount) and move semantics (no clone). +pub(crate) struct Envelope { + dest: ActorAddress, + payload: Box, +} + +impl Envelope { + pub fn new(dest: ActorAddress, payload: Box) -> Self { + Self { dest, payload } + } + + pub fn dest(&self) -> ActorAddress { + self.dest + } + + pub fn downcast(self) -> Option { + self.payload.downcast::().ok().map(|b| *b) + } + + pub fn into_payload(self) -> Box { + self.payload + } +} + + +// ─── InboxRegistry ─────────────────────────────────────────────────────────── + +/// Registry of external inboxes — replaces the Router's role for non-actor receivers. +pub(crate) struct InboxRegistry { + senders: RwLock>>, +} + +impl InboxRegistry { + pub fn new() -> Self { + Self { + senders: RwLock::new(HashMap::new()), + } + } + + pub fn register(&self, addr: ActorAddress, sender: Arc) { + self.senders.write().unwrap().insert(addr, sender); + } + + pub fn try_deliver( + &self, + addr: ActorAddress, + msg: Box, + ) -> Result<(), Error> { + let senders = self.senders.read().unwrap(); + if let Some(sender) = senders.get(&addr) { + sender.try_send_any(msg); + Ok(()) + } else { + Err(Error::from("Address not found")) + } + } +} + + + + +/// Object-safe inner trait for sending type-erased messages. +pub(crate) trait ContextInner { + fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; + fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error>; + fn mailbox_waterlevel(&self) -> usize; +} + + +impl ContextInner for Runtime { + fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { + match self.address_map.lookup(&addr) { + Some(wid) => { + let _ = self.transfer_txs[wid.as_usize()].try_send(Envelope::new(addr, msg)); + Ok(()) + } + None => self.inbox_registry.try_deliver(addr, msg), + } + } + + fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error> { + let worker_id = self.placement.next_worker(); + self.address_map.insert(addr, worker_id); + self.spawn_txs[worker_id.as_usize()] + .try_send((addr, actor)) + .map_err(|_| Error::from("Spawn queue full")) + } + + fn mailbox_waterlevel(&self) -> usize { + self.config.mailbox_waterlevel + } +} diff --git a/src/worker.rs b/src/worker.rs new file mode 100644 index 0000000..1ca7c0f --- /dev/null +++ b/src/worker.rs @@ -0,0 +1,258 @@ +use std::any::Any; +use std::cell::RefCell; +use std::collections::{HashMap, VecDeque}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread; + +use crate::actor::{ActorAddress, AnyActor, Message}; +use crate::address_map::{AddressMap, Placement, WorkerId}; +use crate::channel::{Receiver, Sender}; +use crate::config::{BackoffPolicy, RuntimeConfig}; +use crate::runtime::{ContextInner, Envelope, InboxRegistry}; +use crate::Error; + +/// Shared state passed to tick_once — single thin pointer avoids register spill. +pub(crate) struct TickContext<'a> { + pub address_map: &'a AddressMap, + pub transfer_txs: &'a [Sender], + pub spawn_txs: &'a [Sender<(ActorAddress, Box)>], + pub placement: &'a Placement, + pub inbox_registry: &'a InboxRegistry, + pub config: &'a RuntimeConfig, +} + +/// A worker owns a set of actors and runs them in a loop. +pub(crate) struct Worker { + id: WorkerId, + pool: ActorPool, + transfer_rx: Receiver, + spawn_rx: Receiver<(ActorAddress, Box)>, +} + +impl Worker { + pub fn new( + id: WorkerId, + transfer_rx: Receiver, + spawn_rx: Receiver<(ActorAddress, Box)>, + ) -> Self { + Self { + id, + pool: ActorPool::new(), + transfer_rx, + spawn_rx, + } + } + + /// Run one iteration of the worker loop. Returns `true` if any work was done. + pub fn tick_once(&mut self, tc: &TickContext) -> bool { + let mut did_work = false; + + // 1. Drain spawn queue → add actors to pool + while let Some((addr, actor)) = self.spawn_rx.try_recv() { + self.pool.insert(addr, actor); + did_work = true; + } + + // 2. Drain transfer queue → deliver envelopes to actors + while let Some(envelope) = self.transfer_rx.try_recv() { + let dest = envelope.dest(); + let payload = envelope.into_payload(); + self.pool.deliver(&dest, payload); + did_work = true; + } + + // 3. Tick all actors with WorkerContext + let pending_local: RefCell)>> = + RefCell::new(Vec::new()); + + { + let worker_ctx = WorkerContext { + worker_id: self.id, + address_map: tc.address_map, + transfer_txs: tc.transfer_txs, + spawn_txs: tc.spawn_txs, + placement: tc.placement, + inbox_registry: tc.inbox_registry, + config: tc.config, + pending_local: &pending_local, + }; + if self.pool.tick_all(&worker_ctx) { + did_work = true; + } + } + + // 4. Drain pending_local buffer → deliver to local actors + let pending = pending_local.into_inner(); + if !pending.is_empty() { + did_work = true; + } + for (addr, msg) in pending { + self.pool.deliver(&addr, msg); + } + + did_work + } + + pub(crate) fn run(&mut self, tc: &TickContext, is_running: &AtomicBool, backoff: &BackoffPolicy) { + let mut idle_count: u32 = 0; + while is_running.load(Ordering::Acquire) { + let did_work = self.tick_once(tc); + if did_work { + idle_count = 0; + } else { + idle_count = idle_count.saturating_add(1); + if idle_count < backoff.spin_threshold { + // Hot spin + } else if idle_count < backoff.yield_threshold { + thread::yield_now(); + } else { + let micros = std::cmp::min( + (idle_count - backoff.yield_threshold) as u64 * backoff.sleep_increment_us, + backoff.sleep_max_us, + ); + thread::sleep(std::time::Duration::from_micros(micros)); + } + } + } + } +} + +/// The `ContextInner` impl for worker threads. +/// +/// Same-worker sends are buffered in `pending_local` (delivered after current tick round). +/// Cross-worker sends go through the transfer queue. +struct WorkerContext<'a> { + worker_id: WorkerId, + address_map: &'a AddressMap, + transfer_txs: &'a [Sender], + spawn_txs: &'a [Sender<(ActorAddress, Box)>], + placement: &'a Placement, + inbox_registry: &'a InboxRegistry, + config: &'a RuntimeConfig, + pending_local: &'a RefCell)>>, +} + +impl ContextInner for WorkerContext<'_> { + fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error> { + match self.address_map.lookup(&addr) { + Some(wid) if wid == self.worker_id => { + // Same worker: buffer for local delivery (after current tick round) + self.pending_local.borrow_mut().push((addr, msg)); + Ok(()) + } + Some(wid) => { + // Cross worker: envelope through transfer queue + let envelope = Envelope::new(addr, msg); + let _ = self.transfer_txs[wid.as_usize()].try_send(envelope); + Ok(()) + } + None => { + // Try inbox registry (external inboxes) + self.inbox_registry.try_deliver(addr, msg) + } + } + } + + fn spawn_any(&self, addr: ActorAddress, actor: Box) -> Result<(), Error> { + let worker_id = self.placement.next_worker(); + self.address_map.insert(addr, worker_id); + self.spawn_txs[worker_id.as_usize()] + .try_send((addr, actor)) + .map_err(|_| Error::from("Spawn queue full")) + } + + fn mailbox_waterlevel(&self) -> usize { + self.config.mailbox_waterlevel + } +} + +/// Per-worker actor storage. +pub(crate) struct ActorPool { + actors: HashMap>, +} + +impl ActorPool { + pub fn new() -> Self { + Self { + actors: HashMap::new(), + } + } + + pub fn insert(&mut self, addr: ActorAddress, actor: Box) { + self.actors.insert(addr, actor); + } + + pub fn remove(&mut self, addr: &ActorAddress) -> Option> { + self.actors.remove(addr) + } + + /// Deliver a type-erased message to the actor at `addr`. + /// Returns `true` if the actor was found and the message type matched. + pub fn deliver(&mut self, addr: &ActorAddress, msg: Box) -> bool { + if let Some(actor) = self.actors.get_mut(addr) { + actor.deliver(msg) + } else { + false + } + } + + /// Tick all actors in the pool. Returns `true` if any actor processed messages. + pub fn tick_all(&mut self, inner: &dyn ContextInner) -> bool { + let mut did_work = false; + for actor in self.actors.values_mut() { + if actor.tick(inner) { + did_work = true; + } + } + did_work + } + + pub fn len(&self) -> usize { + self.actors.len() + } +} + + + +pub struct Mailbox { + queue: VecDeque, + waterlevel: usize, +} + +impl Mailbox { + pub fn new(waterlevel: usize) -> Self { + Self { + queue: VecDeque::new(), + waterlevel, + } + } + + pub fn push(&mut self, msg: M) { + self.queue.push_back(msg); + } + + pub fn pop(&mut self) -> Option { + self.queue.pop_front() + } + + pub fn len(&self) -> usize { + self.queue.len() + } + + pub fn is_empty(&self) -> bool { + self.queue.is_empty() + } + + /// How many messages to process this tick: + /// - `len < waterlevel` → process all (`len`) + /// - `len >= waterlevel` → process half (`len >> 1`) + pub fn drain_count(&self) -> usize { + let len = self.queue.len(); + if len < self.waterlevel { + len + } else { + len >> 1 + } + } +} + diff --git a/tests/runtime_api_tests.rs b/tests/runtime_api_tests.rs new file mode 100644 index 0000000..88238a5 --- /dev/null +++ b/tests/runtime_api_tests.rs @@ -0,0 +1,97 @@ +use swactor::{ + actor::{ActorAddress, ActorInterface}, + runtime::{Ctx, Inbox, Runtime, RuntimeConfig}, +}; + +// --------------------------------------------------------------------------- +// Shared test fixtures +// --------------------------------------------------------------------------- + +#[derive(Clone)] +struct EchoMessage { + payload: usize, + reply_to: ActorAddress, +} + +#[derive(Clone, Debug, PartialEq)] +struct EchoResponse(usize); + +struct EchoActor; + +impl ActorInterface for EchoActor { + type Incoming = EchoMessage; + type Response = EchoResponse; + + fn handle(&mut self, ctx: &Ctx, msg: EchoMessage) { + let _ = ctx.send(msg.reply_to, EchoResponse(msg.payload)); + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[test] +fn test_single_thread_spawn_actor_and_inbox() { + let rt = Runtime::new(RuntimeConfig::default()); + + let actor_addr = rt.spawn(EchoActor).expect("spawn echo actor"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + actor_addr, + EchoMessage { + payload: 42, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + for _ in 0..10 { + rt.tick(); + if let Some(response) = inbox.try_recv() { + assert_eq!(response, EchoResponse(42)); + return; + } + } + + panic!("Did not receive EchoResponse"); +} + +#[test] +fn test_multi_thread_spawn_actor_and_inbox() { + let config = RuntimeConfig { + num_threads: 4, + ..Default::default() + }; + let rt = Runtime::new(config); + + let actor_addr = rt.spawn(EchoActor).expect("spawn echo actor"); + let inbox: Inbox = rt.new_inbox().unwrap(); + + rt.send_to( + actor_addr, + EchoMessage { + payload: 99, + reply_to: *inbox.addr(), + }, + ) + .unwrap(); + + let handle = rt.run().unwrap(); + + let check = std::thread::spawn(move || { + for _ in 0..100 { + std::thread::sleep(std::time::Duration::from_millis(10)); + if let Some(response) = inbox.try_recv() { + handle.shutdown(); + return Some(response); + } + } + handle.shutdown(); + None + }); + + let result = check.join().unwrap(); + assert_eq!(result, Some(EchoResponse(99))); +} diff --git a/tests/runtime_tests.rs b/tests/runtime_tests.rs deleted file mode 100644 index 56ca321..0000000 --- a/tests/runtime_tests.rs +++ /dev/null @@ -1,130 +0,0 @@ -use swactor::{actor::{ActorAddress, ActorInterface}, runtime::{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: &Runtime, msg: PingMessage) { - let _ = ctx.send_to(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: &Runtime, msg: ForwardMessage) { - let _ = ctx.send_to(self.target, msg); - } -} - -#[test] -fn test_single_threaded_ping_pong() { - let rt = Runtime::new(RuntimeConfig::default()); - let inbox: Inbox = rt.new_inbox().unwrap(); - - let pong_addr = rt.spawn(PongActor).expect("spawn pong"); - - // Send ping - rt.send_to( - pong_addr, - PingMessage { - reply_to: *inbox.addr(), - }, - ) - .unwrap(); - - // Tick until we get a response - for _ in 0..10 { - rt.tick(); - if inbox.try_recv().is_some() { - return; // Success! - } - } - - panic!("Did not receive pong response"); -} - -#[test] -fn test_single_threaded_message_chain() { - let rt = Runtime::new(RuntimeConfig::default()); - let inbox: Inbox = rt.new_inbox().unwrap(); - - // Create a chain: A -> B -> C -> inbox - let c_addr = rt - .spawn(ForwarderActor { - target: *inbox.addr(), - }) - .unwrap(); - let b_addr = rt.spawn(ForwarderActor { target: c_addr }).unwrap(); - let a_addr = rt.spawn(ForwarderActor { target: b_addr }).unwrap(); - - // Send message to start of chain - rt.send_to(a_addr, ForwardMessage(42)).unwrap(); - - // Tick until message arrives - for _ in 0..20 { - rt.tick(); - if let Some(ForwardMessage(val)) = inbox.try_recv() { - assert_eq!(val, 42); - return; - } - } - - panic!("Message did not traverse the chain"); -} - -#[test] -fn test_multithreaded_message_passing() { - let config = RuntimeConfig { - num_threads: 4, - ..Default::default() - }; - let rt = Runtime::new(config); - let inbox: Inbox = rt.new_inbox().unwrap(); - - // Create a longer chain to exercise multi-threading - let mut target = *inbox.addr(); - for _ in 0..20 { - target = rt.spawn(ForwarderActor { target }).unwrap(); - } - let start_addr = target; - - // Send message - rt.send_to(start_addr, ForwardMessage(999)).unwrap(); - - // Spawn thread to check for result and shutdown - let ctx = rt.run().unwrap(); - let inbox_check = std::thread::spawn(move || { - for _ in 0..100 { - std::thread::sleep(std::time::Duration::from_millis(10)); - if let Some(ForwardMessage(val)) = inbox.try_recv() { - ctx.shutdown(); - return Some(val); - } - } - ctx.shutdown(); - None - }); - - let result = inbox_check.join().unwrap(); - assert_eq!(result, Some(999)); -} diff --git a/tests/stress/concurrency.rs b/tests/stress/concurrency.rs deleted file mode 100644 index 2511745..0000000 --- a/tests/stress/concurrency.rs +++ /dev/null @@ -1,332 +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, - router_max_messages: 10_000, - actor_max_messages: 1000, - num_threads: 4, - }; - let runtime = Runtime::new(config); - - // Spawn actors - let mut actors = Vec::new(); - for _ in 0..50 { - if let Ok(addr) = runtime.spawn(BlackHole) { - actors.push(addr); - } - } - - let handle = runtime.run().unwrap(); - let rt = handle.runtime.clone(); - - // Sender thread - blast messages - let actors_clone = actors.clone(); - let rt_send = rt.clone(); - let sender = thread::spawn(move || { - for _ in 0..1000 { - for actor in &actors_clone { - let _ = rt_send.send_to::(*actor, Msg); - } - } - }); - - // Random delay before shutdown - let delay = Duration::from_micros((iteration * 17) % 500); - thread::sleep(delay); - - // Shutdown while sender is still going - handle.shutdown(); - - // Wait for sender (it should not panic) - let _ = sender.join(); - - // Join should complete (not hang) - handle.join(); - }); - - match result { - Ok(_) => successes += 1, - Err(_) => panics += 1, - } - } - - println!(" Iterations: 100"); - println!(" Successes: {}", successes); - println!(" Panics: {}", panics); - - if panics > 0 { - println!(">>> FAIL: {} panics detected during shutdown\n", panics); - } else { - println!(">>> PASS: No panics during shutdown under load\n"); - } - - assert_eq!(panics, 0, "Shutdown under load caused panics"); -} - -/// Send to actor immediately after spawn. -/// Target: Race between spawn registration and first message. -#[test] -#[cfg(feature = "stress")] -fn send_to_newborn() { - println!("\n>>> STRESS: Send to Newborn Actor"); - - let mut total_spawned = 0; - let mut total_send_ok = 0; - let mut total_send_fail = 0; - - for _ in 0..100 { - let config = RuntimeConfig { - max_actors: 1000, - router_max_messages: 10_000, - actor_max_messages: 100, - num_threads: 4, - }; - let runtime = Runtime::new(config); - let handle = runtime.run().unwrap(); - - // Immediately spawn and send - for _ in 0..50 { - if let Ok(addr) = handle.runtime.spawn(BlackHole) { - total_spawned += 1; - // Send immediately - actor may not be registered yet - if handle.runtime.send_to::(addr, Msg).is_ok() { - total_send_ok += 1; - } else { - total_send_fail += 1; - } - } - } - - handle.shutdown(); - handle.join(); - } - - println!(" Total spawned: {}", total_spawned); - println!(" Sends succeeded: {}", total_send_ok); - println!(" Sends failed: {}", total_send_fail); - - if total_send_fail > 0 { - println!(">>> FAIL: {} messages failed to send\n", total_send_fail); - } else { - println!(">>> PASS: All messages succeeded\n"); - } - - assert_eq!(total_send_fail, 0, "Race condition caused failed message delivery"); - - println!(">>> Test complete\n"); -} - -/// FIXME: This test means nothing until we allow killing off actor processes -/// 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, - router_max_messages: 10_000, - actor_max_messages: 100, - num_threads: 4, - }; - 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, - router_max_messages: 100_000, - actor_max_messages: 10_000, - num_threads: 4, - }; - let runtime = Runtime::new(config); - let target = runtime.spawn(BlackHole).unwrap(); - let handle = runtime.run().unwrap(); - - // Wait for registration - thread::sleep(Duration::from_millis(10)); - - let send_count = Arc::new(AtomicUsize::new(0)); - let fail_count = Arc::new(AtomicUsize::new(0)); - - // 8 threads all sending to same actor - let mut threads = Vec::new(); - for _ in 0..8 { - let rt = handle.runtime.clone(); - let sends = send_count.clone(); - let fails = fail_count.clone(); - - threads.push(thread::spawn(move || { - for _ in 0..10_000 { - if rt.send_to::(target, Msg).is_ok() { - sends.fetch_add(1, Ordering::Relaxed); - } else { - fails.fetch_add(1, Ordering::Relaxed); - } - } - })); - } - - for t in threads { - let _ = t.join(); - } - - // Let messages process - thread::sleep(Duration::from_millis(50)); - - handle.shutdown(); - handle.join(); - - let total_sends = send_count.load(Ordering::Relaxed); - let total_fails = fail_count.load(Ordering::Relaxed); - - println!(" Threads: 8"); - println!(" Msgs per thread: 10,000"); - println!(" Total sent: {}", total_sends); - println!(" Total failed: {}", total_fails); - println!( - " Success rate: {:.1}%", - (total_sends as f64 / (total_sends + total_fails) as f64) * 100.0 - ); - println!(">>> Test complete - no panics\n"); -} - -/// FIXME: Not sure this test is meaningful. -/// 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, - router_max_messages: 1000, - actor_max_messages: 100, - num_threads: 4, - }; - let runtime = Runtime::new(config); - - for _ in 0..20 { - let _ = runtime.spawn(BlackHole); - } - - let handle = runtime.run().unwrap(); - - // Specific delay - if delay_us > 0 { - thread::sleep(Duration::from_micros(delay_us)); - } - - handle.shutdown(); - handle.join(); - }); - - match result { - Ok(_) => ok += 1, - Err(_) => fail += 1, - } - } - - results.push((delay_us, ok, fail)); - } - - println!(" delay_us ok fail"); - println!(" -------- -- ----"); - for (delay, ok, fail) in &results { - println!(" {:>8} {:>2} {:>4}", delay, ok, fail); - } - - let total_fails: i32 = results.iter().map(|(_, _, f)| *f).sum(); - if total_fails > 0 { - println!( - "\n>>> FAIL: {} panics across timing variations", - total_fails - ); - } else { - println!("\n>>> PASS: All timing variations succeeded"); - } -} diff --git a/tests/stress/mod.rs b/tests/stress/mod.rs deleted file mode 100644 index be625c4..0000000 --- a/tests/stress/mod.rs +++ /dev/null @@ -1,199 +0,0 @@ -//! Stress test utilities and result reporting. -//! -//! Provides a simple framework for stress tests with JSON + pretty output. - -#![allow(dead_code)] // Utilities may not all be used in every test - -pub mod concurrency; -pub mod saturation; - -use std::time::{Duration, Instant}; - -/// Results from a stress test -#[derive(Debug)] -pub struct StressResult { - pub name: String, - pub duration: Duration, - pub operations: u64, - pub successes: u64, - pub failures: u64, - pub notes: Vec, -} - -impl StressResult { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - duration: Duration::ZERO, - operations: 0, - successes: 0, - failures: 0, - notes: Vec::new(), - } - } - - pub fn failure_rate(&self) -> f64 { - if self.operations == 0 { - 0.0 - } else { - (self.failures as f64 / self.operations as f64) * 100.0 - } - } - - pub fn throughput(&self) -> f64 { - let secs = self.duration.as_secs_f64(); - if secs > 0.0 { - self.operations as f64 / secs - } else { - 0.0 - } - } - - pub fn note(&mut self, msg: impl Into) { - self.notes.push(msg.into()); - } - - pub fn print(&self) { - println!("\n{}", "=".repeat(60)); - println!(" STRESS: {}", self.name); - println!("{}", "=".repeat(60)); - println!(" Duration: {:?}", self.duration); - println!(" Operations: {}", self.operations); - println!(" Successes: {}", self.successes); - println!(" Failures: {}", self.failures); - println!(" Failure Rate: {:.2}%", self.failure_rate()); - println!(" Throughput: {:.2} ops/sec", self.throughput()); - - if !self.notes.is_empty() { - println!(); - println!(" Notes:"); - for note in &self.notes { - println!(" - {}", note); - } - } - println!("{}", "=".repeat(60)); - } - - pub fn to_json(&self) -> String { - format!( - r#"{{"name":"{}","duration_ms":{},"operations":{},"successes":{},"failures":{},"failure_rate_pct":{:.2},"throughput":{:.2},"notes":{:?}}}"#, - self.name, - self.duration.as_millis(), - self.operations, - self.successes, - self.failures, - self.failure_rate(), - self.throughput(), - self.notes - ) - } -} - -/// A simple stress test runner -pub struct Stress { - name: String, - duration: Option, - iterations: Option, -} - -impl Stress { - pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - duration: None, - iterations: None, - } - } - - /// Run for a fixed duration - pub fn for_duration(mut self, d: Duration) -> Self { - self.duration = Some(d); - self - } - - /// Run for a fixed number of iterations - pub fn for_iterations(mut self, n: u64) -> Self { - self.iterations = Some(n); - self - } - - /// Run the stress test, counting successes and failures - pub fn run(self, mut f: F) -> StressResult - where - F: FnMut() -> bool, // returns true on success, false on failure - { - let mut result = StressResult::new(&self.name); - let start = Instant::now(); - - match (self.duration, self.iterations) { - (Some(duration), _) => { - while start.elapsed() < duration { - if f() { - result.successes += 1; - } else { - result.failures += 1; - } - result.operations += 1; - } - } - (None, Some(iterations)) => { - for _ in 0..iterations { - if f() { - result.successes += 1; - } else { - result.failures += 1; - } - result.operations += 1; - } - } - (None, None) => { - // Default: 1000 iterations - for _ in 0..1000 { - if f() { - result.successes += 1; - } else { - result.failures += 1; - } - result.operations += 1; - } - } - } - - result.duration = start.elapsed(); - result - } -} - -// Test actors used across stress tests -use swactor::{actor::ActorInterface, runtime::Runtime}; - -/// 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: &Runtime, _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: &Runtime, _msg: Msg) { - self.count += 1; - } -} diff --git a/tests/stress/saturation.rs b/tests/stress/saturation.rs deleted file mode 100644 index a8fc97e..0000000 --- a/tests/stress/saturation.rs +++ /dev/null @@ -1,173 +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 router inbox -#[test] -#[cfg(feature = "stress")] -fn router_inbox_overflow() { - println!("\n>>> STRESS: Router Inbox Overflow"); - - let config = RuntimeConfig { - max_actors: 10, - router_max_messages: 100, // Tiny buffer - actor_max_messages: 1000, - num_threads: 1, - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(BlackHole).unwrap(); - - // Blast messages without processing - let mut result = StressResult::new("router_inbox_overflow"); - let start = std::time::Instant::now(); - - for _ in 0..10_000 { - result.operations += 1; - if runtime.send_to::(sink, Msg).is_ok() { - result.successes += 1; - } else { - result.failures += 1; - } - } - - result.duration = start.elapsed(); - result.note(format!("Router buffer: 100, Messages sent: 10,000")); - - // With hybrid, no failures expected - assert_eq!(result.failures, 0, "Hybrid channel should not reject"); - result.print(); - println!(">>> PASS: Hybrid channel prevented router overflow\n"); -} - -/// Blast a single actor's inbox -#[test] -#[cfg(feature = "stress")] -fn actor_inbox_overflow() { - println!("\n>>> STRESS: Actor Inbox Overflow"); - - let config = RuntimeConfig { - max_actors: 10, - router_max_messages: 100_000, // Large router buffer - actor_max_messages: 100, // Tiny actor inbox - num_threads: 1, - }; - let runtime = Runtime::new(config); - let sink = runtime.spawn(Counter::new()).unwrap(); - - // Process router registration - runtime.tick(); - - // Now blast messages - router will accept them but actor inbox will fill - let mut sent = 0u64; - let mut router_failed = 0u64; - for _ in 0..10_000 { - if runtime.send_to::(sink, Msg).is_ok() { - sent += 1; - } else { - router_failed += 1; - } - // Tick occasionally to let router deliver - if sent % 100 == 0 { - runtime.tick(); - } - } - - // Process all remaining messages - for _ in 0..5000 { - runtime.tick(); - } - - println!(" Router accepted: {}", sent); - println!(" Router rejected: {}", router_failed); - - assert_eq!(router_failed, 0, "Router 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 - router_max_messages: 10_000, - actor_max_messages: 100, - num_threads: 1, - }; - 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: Router also takes a slot, so we expect ~99 actors max - assert_eq!( - result.failures, 0, - "Spawned more actors than queue capacity" - ); - println!(">>> PASS: Actor queue correctly rejects when full\n"); -} - -/// FIXME: IS this actually testing what it should be? -/// 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, - router_max_messages: 1000, - actor_max_messages: 100, - num_threads: 1, - }; - 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 registrations - for _ in 0..200 { - runtime.tick(); - } - - let result = Stress::new("sustained_overload") - .for_duration(Duration::from_secs(2)) - .run(|| { - // Send to random actor - let idx = (std::time::Instant::now().elapsed().as_nanos() as usize) % actors.len(); - let success = runtime.send_to::(actors[idx], Msg).is_ok(); - - // Process some (but not all) - simulating overload - runtime.tick(); - - success - }); - - result.print(); - println!(">>> System survived sustained overload without panic\n"); -} diff --git a/tests/stress_tests.rs b/tests/stress_tests.rs deleted file mode 100644 index 216aea2..0000000 --- a/tests/stress_tests.rs +++ /dev/null @@ -1,11 +0,0 @@ -//! Stress test suite for swactor runtime. -//! -//! Run with: cargo test --features stress stress_ -- --nocapture -//! -//! These tests are hidden behind the `stress` feature flag because they: -//! - Take longer to run -//! - Intentionally push the system to failure -//! - May produce different results on different machines - -#[cfg(feature = "stress")] -mod stress;