diff --git a/DESIGN.md b/DESIGN.md index bb03be1..1377243 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -1,6 +1,7 @@ # 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. +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. @@ -11,6 +12,7 @@ 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. @@ -21,15 +23,28 @@ An actor has: - 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 and Router +## 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 + 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. \ No newline at end of file + 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 runs on its own thread and 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. \ No newline at end of file diff --git a/src/kimi.rs b/src/kimi.rs index e69de29..0ea1ab4 100644 --- a/src/kimi.rs +++ 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.rs b/src/lib.rs index dbe780e..2120990 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,13 +1,24 @@ +// mod kimi; + pub mod error; use std::{ - marker::PhantomData, - sync::{Mutex, mpsc::TryRecvError}, + marker::PhantomData, mem::MaybeUninit, sync::{Mutex, atomic::AtomicUsize, mpsc::TryRecvError} }; use crate::error::{Error, Result, convert_err}; + use std::sync::mpsc; +pub trait Message: 'static + Sized + Copy + Default {} + +pub trait Handler { + fn handle(&mut self, msg: M); +} + + + + use bytemuck::{Pod, Zeroable}; @@ -26,139 +37,3 @@ enum GreetResponse { } type GreeterId = u64; - -struct Greeter { - id: GreeterId, - inbox: mpsc::Receiver, - outbox: mpsc::Sender, - state: GreeterState, -} - -impl Greeter { - pub fn new( - id: GreeterId, - inbox: mpsc::Receiver, - outbox: mpsc::Sender, - ) -> Self { - Self { - id, - inbox, - outbox, - state: GreeterState { num_greeted: 0 }, - } - } - - pub fn id(&self) -> GreeterId { - self.id - } - - pub fn process_message(&mut self) -> Result<()> { - match self.inbox.try_recv() { - Ok(m) => { - let GreetMessage::Name(n) = m; - self.outbox - .send(GreetResponse::Greeting(format!("Hello, {n}!"))) - .map_err(convert_err)?; - self.state.num_greeted += 1; - } - Err(e) => match e { - TryRecvError::Empty => return Ok(()), - TryRecvError::Disconnected => { - return Err("Outbox has been disconnected, actor in an improper state".into()); - } - }, - } - - Ok(()) - } -} - -use std::collections::HashMap; - -struct Router { - address_book: HashMap>, - next_id: GreeterId, -} - -impl Router { - pub fn new() -> Self { - Self { - address_book: HashMap::new(), - next_id: 0, - } - } - - pub fn register(&mut self, sender: mpsc::Sender) -> GreeterId { - let id = self.next_id; - self.next_id += 1; - self.address_book.insert(id, sender); - id - } - - pub fn unregister(&mut self, id: GreeterId) -> Option> { - self.address_book.remove(&id) - } - - pub fn get_sender(&self, id: GreeterId) -> Option<&mpsc::Sender> { - self.address_book.get(&id) - } - - pub fn send(&self, id: GreeterId, message: GreetMessage) -> Result<()> { - match self.address_book.get(&id) { - Some(sender) => sender.send(message).map_err(convert_err), - None => Err(format!("No sender found for id {}", id).into()), - } - } -} - -struct Runtime { - router: Router, - greeters: Vec, - response_rx: mpsc::Receiver, - response_tx: mpsc::Sender, -} - -impl Runtime { - pub fn new() -> Self { - let (response_tx, response_rx) = mpsc::channel(); - Self { - router: Router::new(), - greeters: Vec::new(), - response_rx, - response_tx, - } - } - - pub fn spawn_greeter(&mut self) -> GreeterId { - let (inbox_tx, inbox_rx) = mpsc::channel(); - let id = self.router.register(inbox_tx); - let greeter = Greeter::new(id, inbox_rx, self.response_tx.clone()); - self.greeters.push(greeter); - id - } - - pub fn send_message(&self, id: GreeterId, message: GreetMessage) -> Result<()> { - self.router.send(id, message) - } - - pub fn tick(&mut self) -> Result<()> { - for greeter in &mut self.greeters { - greeter.process_message()?; - } - Ok(()) - } - - pub fn try_recv_response(&self) -> Option { - self.response_rx.try_recv().ok() - } - - pub fn run_until_idle(&mut self) -> Result<()> { - loop { - self.tick()?; - if self.response_rx.try_recv().is_err() { - break; - } - } - Ok(()) - } -} \ No newline at end of file