diff --git a/.gitignore b/.gitignore index 5f9ff5d..42840ac 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ /target -.vscode/ \ No newline at end of file +.vscode/ +.venv \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index d0e4e9d..4f55210 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -203,12 +203,27 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + [[package]] name = "hermit-abi" version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +[[package]] +name = "indoc" +version = "2.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79cf5c93f93228cf8efb3ba362535fb11199ac548a09ce117c9b1adc3030d706" +dependencies = [ + "rustversion", +] + [[package]] name = "is-terminal" version = "0.4.17" @@ -257,6 +272,15 @@ version = "2.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -306,6 +330,12 @@ dependencies = [ "plotters-backend", ] +[[package]] +name = "portable-atomic" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" + [[package]] name = "proc-macro2" version = "1.0.106" @@ -315,6 +345,69 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "pyo3" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7778bffd85cf38175ac1f545509665d0b9b92a198ca7941f131f85f7a4f9a872" +dependencies = [ + "cfg-if", + "indoc", + "libc", + "memoffset", + "once_cell", + "portable-atomic", + "pyo3-build-config", + "pyo3-ffi", + "pyo3-macros", + "unindent", +] + +[[package]] +name = "pyo3-build-config" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94f6cbe86ef3bf18998d9df6e0f3fc1050a8c5efa409bf712e661a4366e010fb" +dependencies = [ + "once_cell", + "target-lexicon", +] + +[[package]] +name = "pyo3-ffi" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f1b4c431c0bb1c8fb0a338709859eed0d030ff6daa34368d3b152a63dfdd8d" +dependencies = [ + "libc", + "pyo3-build-config", +] + +[[package]] +name = "pyo3-macros" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fbc2201328f63c4710f68abdf653c89d8dbc2858b88c5d88b0ff38a75288a9da" +dependencies = [ + "proc-macro2", + "pyo3-macros-backend", + "quote", + "syn", +] + +[[package]] +name = "pyo3-macros-backend" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fca6726ad0f3da9c9de093d6f116a93c1a38e417ed73bf138472cf4064f72028" +dependencies = [ + "heck", + "proc-macro2", + "pyo3-build-config", + "quote", + "syn", +] + [[package]] name = "quote" version = "1.0.44" @@ -439,6 +532,7 @@ dependencies = [ "crossbeam-queue", "crossbeam-utils", "getrandom", + "pyo3", ] [[package]] @@ -452,6 +546,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "target-lexicon" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1" + [[package]] name = "tinytemplate" version = "1.2.1" @@ -468,6 +568,12 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9312f7c4f6ff9069b165498234ce8be658059c6728633667c526e27dc2cf1df5" +[[package]] +name = "unindent" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3" + [[package]] name = "walkdir" version = "2.5.0" diff --git a/Cargo.toml b/Cargo.toml index a348085..c223be1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,12 +11,13 @@ crate-type = ["cdylib", "rlib"] default = ["getrandom"] getrandom = ["dep:getrandom"] no_random = [] # compile without access to a source of randomness -stress = [] # Enable stress tests +python = ["dep:pyo3"] [dependencies] getrandom = { version = "0.2", optional = true } crossbeam-queue = "0.3.12" crossbeam-utils = "0.8.21" +pyo3 = { version = "0.23", features = ["extension-module"], optional = true } [dev-dependencies] criterion = { version = "0.5", features = ["html_reports"] } diff --git a/examples/hello_async.py b/examples/hello_async.py new file mode 100644 index 0000000..b56aa5e --- /dev/null +++ b/examples/hello_async.py @@ -0,0 +1,37 @@ +"""Async hello world: runtime runs in background threads, driven from asyncio.""" + +import asyncio +from swactor import Runtime, RuntimeConfig + + +async def recv(inbox, timeout=1.0): + """Poll an inbox until a message arrives.""" + while timeout > 0: + msg = inbox.try_recv() + if msg is not None: + return msg + await asyncio.sleep(0.01) + timeout -= 0.01 + return None + + +async def main(): + rt = Runtime(RuntimeConfig(num_threads=2)) + + def echo(ctx, msg): + ctx.send(msg["reply_to"], f"hello, {msg['name']}!") + + addr = rt.spawn(echo) + inbox = rt.inbox() + handle = rt.run() + + for name in ["alice", "bob", "charlie"]: + handle.send(addr, {"name": name, "reply_to": inbox.addr}) + reply = await recv(inbox) + print(reply) + + handle.shutdown() + handle.join() + + +asyncio.run(main()) diff --git a/examples/hello_single_thread.py b/examples/hello_single_thread.py new file mode 100644 index 0000000..6c27c8e --- /dev/null +++ b/examples/hello_single_thread.py @@ -0,0 +1,13 @@ +"""Hello world: spawn an echo actor, send a message, get it back.""" + +from swactor import Runtime + +def echo(ctx, msg): + ctx.send(msg["reply_to"], f"hello, {msg['name']}!") + +rt = Runtime() +addr = rt.spawn(echo) +inbox = rt.inbox() +rt.send(addr, {"name": "world", "reply_to": inbox.addr}) +rt.tick() +print(inbox.try_recv()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..43b7833 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,11 @@ +[build-system] +requires = ["maturin>=1.7,<2"] +build-backend = "maturin" + +[project] +name = "swactor" +version = "0.1.0" +requires-python = ">=3.9" + +[tool.maturin] +features = ["python"] diff --git a/src/lib.rs b/src/lib.rs index fc57124..715af33 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -11,6 +11,15 @@ pub mod config; pub mod runtime; +#[cfg(feature = "python")] +mod python; + +#[cfg(feature = "python")] +#[pyo3::pymodule] +fn swactor(m: &pyo3::Bound<'_, pyo3::types::PyModule>) -> pyo3::PyResult<()> { + python::register(m) +} + #[cfg(feature = "getrandom")] pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() diff --git a/src/python.rs b/src/python.rs new file mode 100644 index 0000000..fc18560 --- /dev/null +++ b/src/python.rs @@ -0,0 +1,455 @@ +use std::any::Any; +use std::cell::RefCell; + +use pyo3::prelude::*; +use pyo3::types::PyModule; + +use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor}; +use crate::config::{BackoffPolicy, RuntimeConfig}; +use crate::runtime::{Ctx, Inbox, Runtime, RuntimeHandle}; +use crate::worker::Mailbox; +use crate::Error; + +// ─── PyMsg newtype ─────────────────────────────────────────────────────────── + +/// Newtype around `PyObject` that implements `Clone + Send + Sync`. +/// +/// `Py` in pyo3 0.23 doesn't implement `Clone` by default. +/// We implement it by acquiring the GIL to bump the refcount. +/// `Send + Sync` are safe because `Py` is a reference-counted +/// pointer to a Python object protected by the GIL. +#[derive(Debug)] +struct PyMsg(PyObject); + +impl Clone for PyMsg { + fn clone(&self) -> Self { + Python::with_gil(|py| PyMsg(self.0.clone_ref(py))) + } +} + +// Safety: Py is Send + Sync — access is serialized by the GIL. +unsafe impl Send for PyMsg {} +unsafe impl Sync for PyMsg {} + +impl PyMsg { + fn into_inner(self) -> PyObject { + self.0 + } +} + +// ─── Helpers ───────────────────────────────────────────────────────────────── + +fn to_py_err(e: Error) -> PyErr { + pyo3::exceptions::PyRuntimeError::new_err(e.to_string()) +} + +// ─── PyActorAddress ────────────────────────────────────────────────────────── + +#[pyclass(name = "ActorAddress")] +#[derive(Clone)] +pub struct PyActorAddress { + inner: ActorAddress, +} + +#[pymethods] +impl PyActorAddress { + fn hex(&self) -> String { + self.inner + .0 + .iter() + .map(|b| format!("{b:02x}")) + .collect() + } + + fn to_bytes(&self) -> Vec { + self.inner.0.to_vec() + } + + fn __repr__(&self) -> String { + let hex = self.hex(); + format!("ActorAddress({hex})") + } + + fn __eq__(&self, other: &PyActorAddress) -> bool { + self.inner == other.inner + } + + fn __hash__(&self) -> u64 { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + self.inner.hash(&mut hasher); + hasher.finish() + } +} + +impl From for PyActorAddress { + fn from(inner: ActorAddress) -> Self { + Self { inner } + } +} + +// ─── Effects / PyCtx ───────────────────────────────────────────────────────── + +enum Effect { + Send { + addr: ActorAddress, + msg: PyObject, + }, + Spawn { + addr: ActorAddress, + handler: PyObject, + }, +} + +#[pyclass(name = "Ctx", unsendable)] +pub struct PyCtx { + self_addr: ActorAddress, + effects: RefCell>, +} + +impl PyCtx { + fn new(self_addr: ActorAddress) -> Self { + Self { + self_addr, + effects: RefCell::new(Vec::new()), + } + } + + fn take_effects(&self) -> Vec { + self.effects.borrow_mut().drain(..).collect() + } +} + +#[pymethods] +impl PyCtx { + #[getter] + fn self_addr(&self) -> PyActorAddress { + PyActorAddress::from(self.self_addr) + } + + fn send(&self, addr: &PyActorAddress, msg: PyObject) { + self.effects.borrow_mut().push(Effect::Send { + addr: addr.inner, + msg, + }); + } + + fn spawn(&self, handler: PyObject) -> PyActorAddress { + let addr = ActorAddress::new_random(); + self.effects.borrow_mut().push(Effect::Spawn { + addr, + handler, + }); + PyActorAddress::from(addr) + } +} + +// ─── PyActor ───────────────────────────────────────────────────────────────── + +struct PyActor { + handler: PyObject, +} + +impl PyActor { + fn new(handler: PyObject) -> Self { + Self { handler } + } +} + +impl ActorInterface for PyActor { + type Incoming = PyMsg; + type Response = PyMsg; + + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { + let py_ctx = PyCtx::new(ctx.self_addr()); + + let call_result = Python::with_gil(|py| { + let ctx_bound = Bound::new(py, py_ctx)?; + self.handler + .call1(py, (&ctx_bound, msg.into_inner()))?; + let ctx_ref = ctx_bound.borrow(); + Ok::, PyErr>(ctx_ref.take_effects()) + }); + + match call_result { + Ok(effects) => { + for effect in effects { + match effect { + Effect::Send { addr, msg } => { + let _ = ctx.raw_inner().send_via_queue( + addr, + Box::new(PyMsg(msg)) as Box, + ); + } + Effect::Spawn { addr, handler } => { + let waterlevel = ctx.raw_inner().mailbox_waterlevel(); + let actor = PyActor::new(handler); + let actor = Actor::new(addr, Mailbox::new(waterlevel), actor); + let boxed: Box = Box::new(actor); + let _ = ctx.raw_inner().spawn_any(addr, boxed); + } + } + } + } + Err(e) => { + Python::with_gil(|py| { + e.print(py); + }); + } + } + } +} + +// ─── PyInbox ───────────────────────────────────────────────────────────────── + +#[pyclass(name = "Inbox")] +pub struct PyInbox { + inner: Inbox, +} + +#[pymethods] +impl PyInbox { + #[getter] + fn addr(&self) -> PyActorAddress { + PyActorAddress::from(*self.inner.addr()) + } + + fn try_recv(&self) -> Option { + self.inner.try_recv().map(|m| m.into_inner()) + } +} + +// ─── PyRuntimeConfig ───────────────────────────────────────────────────────── + +#[pyclass(name = "RuntimeConfig")] +#[derive(Clone)] +pub struct PyRuntimeConfig { + #[pyo3(get, set)] + num_threads: usize, + #[pyo3(get, set)] + max_actors: usize, + #[pyo3(get, set)] + actor_max_messages: usize, + #[pyo3(get, set)] + mailbox_waterlevel: usize, + #[pyo3(get, set)] + spin_threshold: u32, + #[pyo3(get, set)] + yield_threshold: u32, + #[pyo3(get, set)] + sleep_increment_us: u64, + #[pyo3(get, set)] + sleep_max_us: u64, +} + +#[pymethods] +impl PyRuntimeConfig { + #[new] + #[pyo3(signature = ( + *, + num_threads = 1, + max_actors = 1_000, + actor_max_messages = 1_000, + mailbox_waterlevel = 10, + spin_threshold = 64, + yield_threshold = 256, + sleep_increment_us = 50, + sleep_max_us = 1_000, + ))] + fn new( + num_threads: usize, + max_actors: usize, + actor_max_messages: usize, + mailbox_waterlevel: usize, + spin_threshold: u32, + yield_threshold: u32, + sleep_increment_us: u64, + sleep_max_us: u64, + ) -> Self { + Self { + num_threads, + max_actors, + actor_max_messages, + mailbox_waterlevel, + spin_threshold, + yield_threshold, + sleep_increment_us, + sleep_max_us, + } + } +} + +impl From for RuntimeConfig { + fn from(py: PyRuntimeConfig) -> Self { + RuntimeConfig { + num_threads: py.num_threads, + max_actors: py.max_actors, + actor_max_messages: py.actor_max_messages, + mailbox_waterlevel: py.mailbox_waterlevel, + backoff_policy: BackoffPolicy { + spin_threshold: py.spin_threshold, + yield_threshold: py.yield_threshold, + sleep_increment_us: py.sleep_increment_us, + sleep_max_us: py.sleep_max_us, + }, + } + } +} + +// ─── PyRuntime ─────────────────────────────────────────────────────────────── + +#[pyclass(name = "Runtime")] +pub struct PyRuntime { + inner: Option, +} + +#[pymethods] +impl PyRuntime { + #[new] + #[pyo3(signature = (config=None))] + fn new(config: Option) -> Self { + let config: RuntimeConfig = match config { + Some(c) => c.into(), + None => RuntimeConfig::default(), + }; + Self { + inner: Some(Runtime::new(config)), + } + } + + fn spawn(&self, handler: PyObject) -> PyResult { + let rt = self + .inner + .as_ref() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?; + let actor = PyActor::new(handler); + let addr = rt.spawn(actor).map_err(to_py_err)?; + Ok(PyActorAddress::from(addr)) + } + + fn send(&self, addr: &PyActorAddress, msg: PyObject) -> PyResult<()> { + let rt = self + .inner + .as_ref() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?; + rt.send_to(addr.inner, PyMsg(msg)).map_err(to_py_err) + } + + fn inbox(&self) -> PyResult { + let rt = self + .inner + .as_ref() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?; + let inbox: Inbox = rt.new_inbox().map_err(to_py_err)?; + Ok(PyInbox { inner: inbox }) + } + + fn tick(&self) -> PyResult<()> { + let rt = self + .inner + .as_ref() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?; + rt.tick(); + Ok(()) + } + + fn run(&mut self, py: Python<'_>) -> PyResult { + let rt = self + .inner + .take() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?; + let handle = py.allow_threads(|| rt.run().map_err(to_py_err))?; + Ok(PyRuntimeHandle { + inner: Some(handle), + }) + } + + fn shutdown(&self) -> PyResult<()> { + let rt = self + .inner + .as_ref() + .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?; + rt.shutdown(); + Ok(()) + } +} + +// ─── PyRuntimeHandle ───────────────────────────────────────────────────────── + +#[pyclass(name = "RuntimeHandle")] +pub struct PyRuntimeHandle { + inner: Option, +} + +#[pymethods] +impl PyRuntimeHandle { + fn spawn(&self, handler: PyObject) -> PyResult { + let handle = self + .inner + .as_ref() + .ok_or_else(|| { + pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") + })?; + let actor = PyActor::new(handler); + let addr = handle.runtime.spawn(actor).map_err(to_py_err)?; + Ok(PyActorAddress::from(addr)) + } + + fn send(&self, addr: &PyActorAddress, msg: PyObject) -> PyResult<()> { + let handle = self + .inner + .as_ref() + .ok_or_else(|| { + pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") + })?; + handle + .runtime + .send_to(addr.inner, PyMsg(msg)) + .map_err(to_py_err) + } + + fn inbox(&self) -> PyResult { + let handle = self + .inner + .as_ref() + .ok_or_else(|| { + pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") + })?; + let inbox: Inbox = handle.runtime.new_inbox().map_err(to_py_err)?; + Ok(PyInbox { inner: inbox }) + } + + fn shutdown(&self) -> PyResult<()> { + let handle = self + .inner + .as_ref() + .ok_or_else(|| { + pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") + })?; + handle.shutdown(); + Ok(()) + } + + fn join(&mut self, py: Python<'_>) -> PyResult<()> { + let handle = self + .inner + .take() + .ok_or_else(|| { + pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()") + })?; + py.allow_threads(|| handle.join()); + Ok(()) + } +} + +// ─── Module registration ───────────────────────────────────────────────────── + +pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + Ok(()) +} diff --git a/src/runtime.rs b/src/runtime.rs index d667df6..d8e1368 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -64,6 +64,11 @@ impl<'a> Ctx<'a> { Self { inner, self_addr } } + #[cfg(feature = "python")] + pub(crate) fn raw_inner(&self) -> &dyn ContextInner { + self.inner + } + /// Returns the address of the actor currently being ticked. pub fn self_addr(&self) -> ActorAddress { self.self_addr @@ -233,25 +238,22 @@ impl Runtime { } } - /// Spawn worker threads and start processing, returning a set of handles and - /// a Runtime object to interface with. + /// Spawn worker threads and start processing, returning a handle + /// to interact with the runtime and join the threads later. /// - /// ### WARN: - /// ##### Returns an error if the configuration is set as single threaded - /// `config.num_threads == 1` + /// Works in both single-threaded and multi-threaded configurations. + /// In single-threaded mode, one background thread is spawned. pub fn run(mut self) -> Result { - if self.config.num_threads < 2 { - return Err(Error::from( - "Runtime error: cannot call `Runtime::run()` from a single-threaded context.", - )); - } - self.is_running.store(true, Ordering::Release); - let workers = self - .pending_workers - .take() - .expect("Workers must be present for multi-threaded runtime"); + let mut workers: Vec = Vec::new(); + + if let Some(w) = self.single_worker.take() { + workers.push(w.into_inner()); + } + if let Some(ws) = self.pending_workers.take() { + workers.extend(ws); + } let rt = Arc::new(self); let mut handles: Vec> = Vec::with_capacity(workers.len()); diff --git a/tests/test_python.py b/tests/test_python.py new file mode 100644 index 0000000..07b4bb3 --- /dev/null +++ b/tests/test_python.py @@ -0,0 +1,162 @@ +"""Tests for swactor Python bindings.""" + +import unittest +from swactor import Runtime, RuntimeConfig, ActorAddress + + +class TestActorAddress(unittest.TestCase): + def test_repr(self): + rt = Runtime() + addr = rt.spawn(lambda ctx, msg: None) + r = repr(addr) + self.assertTrue(r.startswith("ActorAddress(")) + self.assertTrue(r.endswith(")")) + # hex string should be 64 chars (32 bytes) + hex_part = r[len("ActorAddress("):-1] + self.assertEqual(len(hex_part), 64) + + def test_hex(self): + rt = Runtime() + addr = rt.spawn(lambda ctx, msg: None) + self.assertEqual(len(addr.hex()), 64) + + def test_to_bytes(self): + rt = Runtime() + addr = rt.spawn(lambda ctx, msg: None) + self.assertEqual(len(addr.to_bytes()), 32) + + def test_equality(self): + rt = Runtime() + addr = rt.spawn(lambda ctx, msg: None) + # Same address object should be equal to itself + self.assertEqual(addr, addr) + + def test_hashable(self): + rt = Runtime() + addr1 = rt.spawn(lambda ctx, msg: None) + addr2 = rt.spawn(lambda ctx, msg: None) + s = {addr1, addr2} + self.assertEqual(len(s), 2) + s.add(addr1) + self.assertEqual(len(s), 2) + + +class TestRuntimeConfig(unittest.TestCase): + def test_defaults(self): + cfg = RuntimeConfig() + self.assertEqual(cfg.num_threads, 1) + self.assertEqual(cfg.max_actors, 1000) + self.assertEqual(cfg.actor_max_messages, 1000) + self.assertEqual(cfg.mailbox_waterlevel, 10) + self.assertEqual(cfg.spin_threshold, 64) + self.assertEqual(cfg.yield_threshold, 256) + self.assertEqual(cfg.sleep_increment_us, 50) + self.assertEqual(cfg.sleep_max_us, 1000) + + def test_custom(self): + cfg = RuntimeConfig(num_threads=4, max_actors=500) + self.assertEqual(cfg.num_threads, 4) + self.assertEqual(cfg.max_actors, 500) + + +class TestSingleThreaded(unittest.TestCase): + def test_echo(self): + """Spawn an echo actor, send a message, tick, and recv.""" + rt = Runtime() + + def echo(ctx, msg): + ctx.send(msg["reply_to"], msg["payload"]) + + addr = rt.spawn(echo) + inbox = rt.inbox() + rt.send(addr, {"payload": "hello", "reply_to": inbox.addr}) + rt.tick() + result = inbox.try_recv() + self.assertEqual(result, "hello") + + def test_spawn_from_handler(self): + """Actor spawns a child and forwards work to it.""" + rt = Runtime() + + def child(ctx, msg): + ctx.send(msg["reply_to"], "from_child") + + def parent(ctx, msg): + c = ctx.spawn(child) + ctx.send(c, {"reply_to": msg["reply_to"]}) + + addr = rt.spawn(parent) + inbox = rt.inbox() + rt.send(addr, {"reply_to": inbox.addr}) + # First tick: parent runs, spawns child, sends to child + rt.tick() + # Second tick: child runs, sends to inbox + rt.tick() + result = inbox.try_recv() + self.assertEqual(result, "from_child") + + def test_stateful_actor(self): + """Callable class maintains state across messages.""" + rt = Runtime() + + class Counter: + def __init__(self): + self.n = 0 + + def __call__(self, ctx, msg): + self.n += 1 + ctx.send(msg["reply_to"], self.n) + + addr = rt.spawn(Counter()) + inbox = rt.inbox() + rt.send(addr, {"reply_to": inbox.addr}) + rt.send(addr, {"reply_to": inbox.addr}) + rt.tick() + self.assertEqual(inbox.try_recv(), 1) + self.assertEqual(inbox.try_recv(), 2) + + def test_no_message_returns_none(self): + rt = Runtime() + inbox = rt.inbox() + self.assertIsNone(inbox.try_recv()) + + +class TestMultiThreaded(unittest.TestCase): + def test_run_shutdown_join(self): + """Multi-threaded runtime can spawn, send, and receive.""" + import time + + rt = Runtime(RuntimeConfig(num_threads=2)) + + def echo(ctx, msg): + ctx.send(msg["reply_to"], msg["payload"]) + + addr = rt.spawn(echo) + inbox = rt.inbox() + handle = rt.run() + handle.send(addr, {"payload": "mt_hello", "reply_to": inbox.addr}) + + # Poll for result + result = None + for _ in range(100): + result = inbox.try_recv() + if result is not None: + break + time.sleep(0.01) + self.assertEqual(result, "mt_hello") + + handle.shutdown() + handle.join() + + def test_run_consumes_runtime(self): + """After run(), tick() should raise.""" + rt = Runtime(RuntimeConfig(num_threads=2)) + handle = rt.run() + with self.assertRaises(RuntimeError): + rt.tick() + handle.shutdown() + handle.join() + + +if __name__ == "__main__": + unittest.main() diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..0bcbb24 --- /dev/null +++ b/uv.lock @@ -0,0 +1,8 @@ +version = 1 +revision = 3 +requires-python = ">=3.9" + +[[package]] +name = "swactor" +version = "0.1.0" +source = { editable = "." }