diff --git a/Cargo.lock b/Cargo.lock index fd2fd7a..bb8e0dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3,102 +3,53 @@ version = 4 [[package]] -name = "bytes" -version = "1.11.0" +name = "cfg-if" +version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] -name = "futures-core" -version = "0.3.31" +name = "crossbeam-queue" +version = "0.3.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" - -[[package]] -name = "futures-sink" -version = "0.3.31" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" - -[[package]] -name = "pin-project-lite" -version = "0.2.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" - -[[package]] -name = "proc-macro2" -version = "1.0.103" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ee95bc4ef87b8d5ba32e8b7714ccc834865276eab0aed5c9958d00ec45f49e8" +checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115" dependencies = [ - "unicode-ident", + "crossbeam-utils", ] [[package]] -name = "quote" -version = "1.0.42" +name = "crossbeam-utils" +version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a338cc41d27e6cc6dce6cefc13a0729dfbb81c262b1f519331575dd80ef3067f" +checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ - "proc-macro2", + "cfg-if", + "libc", + "wasi", ] +[[package]] +name = "libc" +version = "0.2.180" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc" + [[package]] name = "swactor" version = "0.1.0" dependencies = [ - "tokio", - "tokio-util", + "crossbeam-queue", + "getrandom", ] [[package]] -name = "syn" -version = "2.0.111" +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "390cc9a294ab71bdb1aa2e99d13be9c753cd2d7bd6560c77118597410c4d2e87" -dependencies = [ - "proc-macro2", - "quote", - "unicode-ident", -] - -[[package]] -name = "tokio" -version = "1.48.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408" -dependencies = [ - "pin-project-lite", - "tokio-macros", -] - -[[package]] -name = "tokio-macros" -version = "2.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "tokio-util" -version = "0.7.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2efa149fe76073d6e8fd97ef4f4eca7b67f599660115591483572e406e165594" -dependencies = [ - "bytes", - "futures-core", - "futures-sink", - "pin-project-lite", - "tokio", -] - -[[package]] -name = "unicode-ident" -version = "1.0.22" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9312f7c4f6ff9069b165498234ce8be658059c6728633667c526e27dc2cf1df5" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" diff --git a/Cargo.toml b/Cargo.toml index ba24c26..8b21fb6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,10 @@ edition = "2024" [lib] crate-type = ["cdylib", "rlib"] +[features] +default = ["getrandom"] +getrandom = ["dep:getrandom"] + [dependencies] -tokio = { version = "1.48.0", features = ["rt", "macros"] } -tokio-util = "0.7.17" +getrandom = { version = "0.2", optional = true } +crossbeam-queue = "0.3.12" diff --git a/DESIGN.md b/DESIGN.md new file mode 100644 index 0000000..1a50043 --- /dev/null +++ b/DESIGN.md @@ -0,0 +1,77 @@ +# 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 + } + } +} +``` \ No newline at end of file diff --git a/examples/hello.rs b/examples/hello.rs index 40440e1..3cf176a 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -1,46 +1,58 @@ -use swactor::Actor; -use tokio::sync::oneshot; +use swactor::{ActorAddress, ActorInterface, Message, Runtime, RuntimeFlavor}; -pub enum GreeterMessage { - Name(String), +#[derive(Debug, Default)] +struct Greeter { + pub num_greeted: usize, } -pub enum GreeterResponse { - Hello(String), +#[derive(Debug, Default, Clone)] +struct GreetMessage { + /// who do we greet? + who: String, + + /// who do we send out greeting back to? + return_addr: ActorAddress, } +impl Message for GreetMessage {} -pub struct Greeter; +impl ActorInterface for Greeter { + type Incoming = GreetMessage; + type Response = GreetResponse; -impl Actor for Greeter { - type Message = GreeterMessage; - type Response = GreeterResponse; - - fn handle_message(&self, msg: Self::Message, tx: oneshot::Sender) { - let rep = match msg { - GreeterMessage::Name(name) => GreeterResponse::Hello(format!("Hello, {name}!")), - }; - - if let Err(_) = tx.send(rep) { - // Greeter is not responsible for a dropped Receiver + fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) { + let res = GreetResponse(format!("Hello, {}!", msg.who)); + self.num_greeted += 1; + if let Err(_) = ctx.send_to(msg.return_addr, res) { + // no error handling + self.num_greeted -= 1; } } } +#[derive(Debug, Default, Clone)] +struct GreetResponse(String); +impl Message for GreetResponse {} + fn main() { - let rt = tokio::runtime::Builder::new_current_thread() - .build() - .expect("failed to build runtime"); - let greeter = Greeter.spawn(&rt); + let mut rt = Runtime::new(100, Some(RuntimeFlavor::SingleThreaded)); + let addr = rt + .spawn(Greeter::default()) + .expect("failed to spawn greeter"); + let inbox = rt.new_inbox::(); - let response = rt - .block_on(async move { - greeter - .send(GreeterMessage::Name("world".to_string())) - .await - }) - .expect("failed to get respose"); - - match response { - GreeterResponse::Hello(hello) => println!("{hello}"), + 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("greeter should have said hello"); + + println!("{}", resp.0); } diff --git a/src/kimi.rs b/src/kimi.rs new file mode 100644 index 0000000..0ea1ab4 --- /dev/null +++ b/src/kimi.rs @@ -0,0 +1,200 @@ +#![no_std] + +use core::cell::UnsafeCell; +use core::marker::PhantomData; +use core::mem::MaybeUninit; + +/// Actor framework that compiles on thumbv6m-none-eabi +/// No heap, no async, no dyn Trait, no hidden allocations + +// ----------------------------------------------------------------------------- +// Errors +// ----------------------------------------------------------------------------- + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Error { + MailboxFull, + ActorNotFound, + MailboxEmpty, +} + +// ----------------------------------------------------------------------------- +// Core Traits +// ----------------------------------------------------------------------------- + +pub trait Message: 'static + Sized + Copy {} + +pub trait Handler { + fn handle(&mut self, msg: M); +} + +// ----------------------------------------------------------------------------- +// Address +// ----------------------------------------------------------------------------- + +#[derive(Clone, Copy)] +pub struct Addr { + id: u8, + _phantom: PhantomData, +} + +// ----------------------------------------------------------------------------- +// Mailbox +// ----------------------------------------------------------------------------- + +/// FIXME: This is a failed ring buffer, but can and should be fixed +/// +struct Mailbox { + buffer: [MaybeUninit; Q], + head: UnsafeCell, + tail: UnsafeCell, + len: UnsafeCell, +} + +impl Mailbox { + const fn new() -> Self { + Self { + buffer: [MaybeUninit::uninit(); Q], + head: UnsafeCell::new(0), + tail: UnsafeCell::new(0), + len: UnsafeCell::new(0), + } + } + + fn send(&self, msg: M) -> Result<(), Error> { + unsafe { + let len = &mut *self.len.get(); + if *len == Q { + return Err(Error::MailboxFull); + } + let tail = *self.tail.get(); + self.buffer[tail].write(msg); + *self.tail.get() = (tail + 1) % Q; + *len += 1; + Ok(()) + } + } + + fn recv(&self) -> Result { + unsafe { + let len = &mut *self.len.get(); + if *len == 0 { + return Err(Error::MailboxEmpty); + } + let head = *self.head.get(); + let msg = self.buffer[head].assume_init_read(); + *self.head.get() = (head + 1) % Q; + *len -= 1; + Ok(msg) + } + } +} + +// ----------------------------------------------------------------------------- +// System +// ----------------------------------------------------------------------------- + +pub struct System +where + A: Handler, + M: Message, +{ + actors: [MaybeUninit; N], + mailboxes: [Mailbox; N], + used: [bool; N], +} + +impl System +where + A: Handler, + M: Message, +{ + pub const fn new() -> Self { + Self { + actors: [MaybeUninit::uninit(); N], + mailboxes: [Mailbox::new(); N], + used: [false; N], + } + } + + pub fn spawn(&mut self, actor: A) -> Result, Error> { + for i in 0..N { + if !self.used[i] { + self.actors[i].write(actor); + self.used[i] = true; + return Ok(Addr { + id: i as u8, + _phantom: PhantomData, + }); + } + } + Err(Error::MailboxFull) + } + + pub fn send(&mut self, addr: &Addr, msg: M) -> Result<(), Error> { + let idx = addr.id as usize; + if idx >= N || !self.used[idx] { + return Err(Error::ActorNotFound); + } + self.mailboxes[idx].send(msg) + } + + pub fn process_one(&mut self, addr: &Addr) -> Result<(), Error> { + let idx = addr.id as usize; + if idx >= N || !self.used[idx] { + return Err(Error::ActorNotFound); + } + + let msg = self.mailboxes[idx].recv()?; + let actor = unsafe { self.actors[idx].assume_init_mut() }; + actor.handle(msg); + Ok(()) + } + + pub fn process_all(&mut self) { + for i in 0..N { + if self.used[i] { + let addr = Addr { + id: i as u8, + _phantom: PhantomData, + }; + while self.process_one(&addr).is_ok() {} + } + } + } + + pub fn get_mut(&mut self, addr: &Addr) -> Result<&mut A, Error> { + let idx = addr.id as usize; + if idx >= N || !self.used[idx] { + return Err(Error::ActorNotFound); + } + Ok(unsafe { self.actors[idx].assume_init_mut() }) + } +} + +// ----------------------------------------------------------------------------- +// Example +// ----------------------------------------------------------------------------- + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CounterMsg { + Increment(u32), + Decrement(u32), + Reset, +} + +impl Message for CounterMsg {} + +pub struct Counter { + pub value: u32, +} + +impl Handler for Counter { + fn handle(&mut self, msg: CounterMsg) { + match msg { + CounterMsg::Increment(n) => self.value = self.value.wrapping_add(n), + CounterMsg::Decrement(n) => self.value = self.value.wrapping_sub(n), + CounterMsg::Reset => self.value = 0, + } + } +} \ No newline at end of file diff --git a/src/lib.bak.rs b/src/lib.bak.rs new file mode 100644 index 0000000..1641435 --- /dev/null +++ b/src/lib.bak.rs @@ -0,0 +1,118 @@ +pub mod error; + +/// Public export as the oneshot channel is in the `Actor` trait signature +pub use tokio::sync::oneshot; + +use tokio::{sync::mpsc, task::JoinHandle}; +use tokio_util::sync::CancellationToken; + +use crate::error::{Result, convert_err}; + +const DEFAULT_CHANNEL_SIZE: usize = 100; + +type ActorRequest = ( + ::Message, + oneshot::Sender<::Response>, +); + +/// Wrapper defining the transmission end of a Request/Response channel with an `Actor` +pub struct ActorRequestSender(mpsc::Sender>); + +impl ActorRequestSender { + pub async fn send(&self, request: A::Message) -> Result { + let (tx, rx) = oneshot::channel::(); + self.0.send((request, tx)).await.map_err(convert_err)?; + + rx.await.map_err(convert_err) + } +} + +impl From>> for ActorRequestSender { + fn from(value: mpsc::Sender>) -> Self { + Self(value) + } +} + +impl Clone for ActorRequestSender { + fn clone(&self) -> Self { + Self(self.0.clone()) + } +} + +/// Combined 'JoinHandle' to await the actor process and 'Sender' for communication +pub struct Handle +where + A: Actor, +{ + cancel_token: CancellationToken, + tx: ActorRequestSender, + /// the task drops when the `JoinHandle` does, so be careful with the `Handle` + _handle: JoinHandle>, + // to prevent accidental swaps, strongly type the handle + _type: std::marker::PhantomData, +} + +impl Handle { + /// Send a message to the spawned `Actor` task and get a response corresponding to the `Actor::Response` type + pub async fn send(&self, msg: A::Message) -> Result { + self.tx.send(msg).await + } + + /// Get a cloned sender for messaging the `Actor` this handle is for + pub fn get_connection(&self) -> ActorRequestSender { + self.tx.clone() + } +} + +impl Drop for Handle { + fn drop(&mut self) { + self.cancel_token.cancel(); + } +} + +/// Primary trait defining an `Actor` capable of receiving, processing, and transmitting messages +pub trait Actor: Send + Sized + 'static { + /// The type for messages received by this `Actor` + type Message: Send; + /// The type for responses given by this actor when called from `Handle::send(..)` + type Response: Send; + + /// Inner method that defines actor behavior + fn handle_message(&self, msg: Self::Message, tx: oneshot::Sender); + + /// Spawns the `Actor` utilizing the given runtime context + /// Only `tokio` runtime is accepted for now + fn spawn(self, ctx: &tokio::runtime::Runtime) -> Handle { + let cancel_token = CancellationToken::new(); + let cancel = cancel_token.clone(); + + let (tx, mut rx) = + mpsc::channel::<(Self::Message, oneshot::Sender)>(DEFAULT_CHANNEL_SIZE); + let handle = ctx.spawn(async move { + let mut res = Ok(()); + loop { + tokio::select! { + _ = cancel.cancelled() => { + break; + }, + + msg = rx.recv() => { + match msg { + Some(m) => { self.handle_message(m.0, m.1); }, + None => {res = Err(format!("Sender handle was dropped without calling cancel!").into()); break; }, + } + } + }; + } + + res + }); + + Handle { + cancel_token, + _handle: handle, + tx: tx.into(), + _type: std::marker::PhantomData::, + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 1641435..bd6913b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,118 +1,251 @@ +mod ring_buffer; + +use std::collections::HashMap; + +use crossbeam_queue::ArrayQueue; +use ring_buffer::{Receiver, Sender}; + pub mod error; +use error::Error; -/// Public export as the oneshot channel is in the `Actor` trait signature -pub use tokio::sync::oneshot; - -use tokio::{sync::mpsc, task::JoinHandle}; -use tokio_util::sync::CancellationToken; - -use crate::error::{Result, convert_err}; - -const DEFAULT_CHANNEL_SIZE: usize = 100; - -type ActorRequest = ( - ::Message, - oneshot::Sender<::Response>, -); - -/// Wrapper defining the transmission end of a Request/Response channel with an `Actor` -pub struct ActorRequestSender(mpsc::Sender>); - -impl ActorRequestSender { - pub async fn send(&self, request: A::Message) -> Result { - let (tx, rx) = oneshot::channel::(); - self.0.send((request, tx)).await.map_err(convert_err)?; - - rx.await.map_err(convert_err) - } +#[cfg(feature = "getrandom")] +pub fn get_random(buf: &mut [u8]) { + getrandom::getrandom(buf).unwrap() } -impl From>> for ActorRequestSender { - fn from(value: mpsc::Sender>) -> Self { - Self(value) - } +/// The strategy for message processing is such: +/// if total_messages < WATERLEVEL: +/// process all +/// else +/// process total_messages // 2 +const WATERLEVEL: usize = 10; + +const DEFAULT_INBOX_CAPACITY: usize = 100; + +pub trait Message: 'static + Sized + Clone + Send {} +pub type Envelope = Box; + +pub trait ActorInterface: 'static + Send { + type Incoming: Message; + type Response: Message; + fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming); } -impl Clone for ActorRequestSender { - fn clone(&self) -> Self { - Self(self.0.clone()) - } -} +pub type ActorAddress = u64; -/// Combined 'JoinHandle' to await the actor process and 'Sender' for communication -pub struct Handle +pub struct Actor where - A: Actor, + A: ActorInterface, { - cancel_token: CancellationToken, - tx: ActorRequestSender, - /// the task drops when the `JoinHandle` does, so be careful with the `Handle` - _handle: JoinHandle>, - // to prevent accidental swaps, strongly type the handle - _type: std::marker::PhantomData, + _addr: ActorAddress, + inbox: Receiver, + inner: A, } -impl Handle { - /// Send a message to the spawned `Actor` task and get a response corresponding to the `Actor::Response` type - pub async fn send(&self, msg: A::Message) -> Result { - self.tx.send(msg).await - } - - /// Get a cloned sender for messaging the `Actor` this handle is for - pub fn get_connection(&self) -> ActorRequestSender { - self.tx.clone() - } +/// Trait for type-erased actors +trait AnyActor: Send { + fn tick(&mut self, ctx: &Runtime); } -impl Drop for Handle { - fn drop(&mut self) { - self.cancel_token.cancel(); - } -} +impl AnyActor for Actor +where + A: ActorInterface, +{ + fn tick(&mut self, ctx: &Runtime) { + let total_messages = self.inbox.len(); + let messages_to_process = if total_messages < WATERLEVEL { + total_messages + } else { + total_messages >> 1 + }; -/// Primary trait defining an `Actor` capable of receiving, processing, and transmitting messages -pub trait Actor: Send + Sized + 'static { - /// The type for messages received by this `Actor` - type Message: Send; - /// The type for responses given by this actor when called from `Handle::send(..)` - type Response: Send; - - /// Inner method that defines actor behavior - fn handle_message(&self, msg: Self::Message, tx: oneshot::Sender); - - /// Spawns the `Actor` utilizing the given runtime context - /// Only `tokio` runtime is accepted for now - fn spawn(self, ctx: &tokio::runtime::Runtime) -> Handle { - let cancel_token = CancellationToken::new(); - let cancel = cancel_token.clone(); - - let (tx, mut rx) = - mpsc::channel::<(Self::Message, oneshot::Sender)>(DEFAULT_CHANNEL_SIZE); - let handle = ctx.spawn(async move { - let mut res = Ok(()); - loop { - tokio::select! { - _ = cancel.cancelled() => { - break; - }, - - msg = rx.recv() => { - match msg { - Some(m) => { self.handle_message(m.0, m.1); }, - None => {res = Err(format!("Sender handle was dropped without calling cancel!").into()); break; }, - } - } - }; + 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" + ), + } + } + } +} + +pub struct Inbox { + addr: ActorAddress, + inner: Receiver, +} + +impl Inbox { + pub fn addr(&self) -> &ActorAddress { + &self.addr + } + + pub fn try_recv(&self) -> Option { + self.inner.try_recv() + } +} + +#[derive(Debug, Default)] +pub enum RuntimeFlavor { + #[default] + SingleThreaded, + Multithreaded(usize), +} + +pub struct Runtime { + flavor: RuntimeFlavor, + router: Router, + router_inbox: Sender, + actor_queue: ArrayQueue>, +} + +impl Runtime { + pub fn new(capacity: usize, flavor: Option) -> Self { + let router = Router::new(DEFAULT_INBOX_CAPACITY); + let router_inbox = router.new_sender(); + Self { + flavor: flavor.unwrap_or_default(), + router, + router_inbox, + actor_queue: ArrayQueue::new(capacity), + } + } + + pub fn spawn(&self, actor: A) -> Result { + let addr = { + let mut bytes = u64::to_le_bytes(0); + get_random(&mut bytes); + u64::from_le_bytes(bytes) + }; + let inbox = Receiver::::new(DEFAULT_INBOX_CAPACITY); + let sender = inbox.new_sender(); + + // Register the sender with the router + let _ = self + .router_inbox + .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); + + self.actor_queue + .push(Box::new(Actor { + _addr: addr, + inbox, + inner: actor, + })) + .map_err(|_| Error::from("Runtime error: Failed to spawn actor."))?; + + Ok(addr) + } + + pub fn send_to(&self, addr: ActorAddress, msg: M) -> Result<(), ()> { + let envelope: Envelope = Box::new(msg); + self.router_inbox + .try_send(RouterMessage::SendToAddr { + addr, + msg: envelope, + }) + .map_err(|_| ()) + } + + pub fn tick(&mut self) { + // Pop actor, tick it, push it back + if let Some(mut actor) = self.actor_queue.pop() { + actor.tick(self); + let _ = self.actor_queue.push(actor); + } + + match self.flavor { + RuntimeFlavor::Multithreaded(_) => (), // router has its own thread + RuntimeFlavor::SingleThreaded => self.router.tick(), + } + } + + pub fn new_inbox(&self) -> Inbox { + let addr = { + let mut bytes = u64::to_le_bytes(0); + get_random(&mut bytes); + u64::from_le_bytes(bytes) + }; + let receiver = Receiver::::new(DEFAULT_INBOX_CAPACITY); + let sender = receiver.new_sender(); + // Register the sender with the router + let _ = self + .router_inbox + .try_send(RouterMessage::AddAddr(addr, Box::new(sender))); + Inbox { + addr, + inner: receiver, + } + } +} + +pub trait SenderT: Send { + fn try_send(&self, envelope: Envelope); +} + +impl SenderT for Sender { + fn try_send(&self, envelope: Envelope) { + if let Ok(msg) = envelope.downcast::() { + let _ = Sender::try_send(self, *msg); + } + } +} + +/// Internal messages for the Router's own inbox +pub enum RouterMessage { + /// register addrs with sender + AddAddr(ActorAddress, Box), + /// remove an actor from the address book + RemoveAddr(ActorAddress), + /// send to + SendToAddr { addr: ActorAddress, msg: Envelope }, +} + +struct Router { + directory: HashMap>, + inbox: Receiver, +} + +impl Router { + pub fn new(cap: usize) -> Self { + Self { + directory: HashMap::new(), + inbox: Receiver::new(cap), + } + } + + pub fn tick(&mut self) { + 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.handle(msg), + None => unreachable!("We ran checks on total messages before processing."), + } + } + } + + pub fn new_sender(&self) -> Sender { + self.inbox.new_sender() + } + + fn handle(&mut self, msg: RouterMessage) { + 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); + } } - - res - }); - - Handle { - cancel_token, - _handle: handle, - tx: tx.into(), - _type: std::marker::PhantomData::, } } } diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs new file mode 100644 index 0000000..fea5113 --- /dev/null +++ b/src/ring_buffer.rs @@ -0,0 +1,56 @@ +pub use crossbeam_queue::ArrayQueue; +use std::sync::Arc; + +/// The receiving end of a `crossbeam_queue::ArrayQueue`, a lock-free mpsc queue. +/// The queue is constructed by the `Receiver::new()` method. +/// Responsible for creating the `Sender` ends of itself. +/// +/// Notably: The `Receiver` provides no guarentees that a sending end of the channel exists. +pub(crate) struct Receiver { + queue: Arc>, +} + +impl Receiver { + /// Constructs a new `ArrayQueue` with given capacity. + /// + /// # Panics + /// Will panic if capacity is passed as 0 + pub fn new(capacity: usize) -> Self { + Self { + queue: Arc::new(ArrayQueue::new(capacity)), + } + } + + /// Returns the number of elements in the inner queue + pub fn len(&self) -> usize { + self.queue.len() + } + + /// Attempt to retrieve a value from the queue. Returns `None` if empty + pub fn try_recv(&self) -> Option { + self.queue.pop() + } + + /// Construct a new `Sender` assosciated with this queue. + pub fn new_sender(&self) -> Sender { + Sender { + queue: self.queue.clone(), + } + } +} + +/// The sending end of a `crossbeam_queue::ArrayQueue`, a lock free mpsc queue. +/// The queue is initialized via calling the corresponding `Receiver::::new()` method, +/// and the sending end of the queue is constructed via calling `receiver.new_sender()`. +/// +/// Notably: The `Sender` provides no guarentees that a receiving end of the channel exists. +pub(crate) struct Sender { + queue: Arc>, +} + +impl Sender { + /// Attempt to push a value to the queue. Returns Err(value) if the queue is full. + pub fn try_send(&self, value: T) -> Result<(), T> { + self.queue.push(value) + } +}