stash: rewrite wip

Need to implement our own atomic ring buffer, as I want full control
over the feature set, to enable fine tuning optimization. Following that,
we have an outline of all the necessary components listed out in the
design doc, so implementation should happen relatively quickly.

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.
+
+    - 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.

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<M: Message> {
+    fn handle(&mut self, msg: M);
+}
+
+// -----------------------------------------------------------------------------
+// Address
+// -----------------------------------------------------------------------------
+
+#[derive(Clone, Copy)]
+pub struct Addr<M: Message> {
+    id: u8,
+    _phantom: PhantomData<M>,
+}
+
+// -----------------------------------------------------------------------------
+// Mailbox
+// -----------------------------------------------------------------------------
+
+/// FIXME: This is a failed ring buffer, but can and should be fixed
+/// 
+struct Mailbox<M: Message, const Q: usize> {
+    buffer: [MaybeUninit<M>; Q],
+    head: UnsafeCell<usize>,
+    tail: UnsafeCell<usize>,
+    len: UnsafeCell<usize>,
+}
+
+impl<M: Message, const Q: usize> Mailbox<M, Q> {
+    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<M, Error> {
+        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<A, M, const N: usize, const Q: usize>
+where
+    A: Handler<M>,
+    M: Message,
+{
+    actors: [MaybeUninit<A>; N],
+    mailboxes: [Mailbox<M, Q>; N],
+    used: [bool; N],
+}
+
+impl<A, M, const N: usize, const Q: usize> System<A, M, N, Q>
+where
+    A: Handler<M>,
+    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<Addr<M>, 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<M>, 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<M>) -> 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<M>) -> 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<CounterMsg> 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,
+        }
+    }
+}

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<M: Message> {
+    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<GreetMessage>,
-    outbox: mpsc::Sender<GreetResponse>,
-    state: GreeterState,
-}
-
-impl Greeter {
-    pub fn new(
-        id: GreeterId,
-        inbox: mpsc::Receiver<GreetMessage>,
-        outbox: mpsc::Sender<GreetResponse>,
-    ) -> 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<GreeterId, mpsc::Sender<GreetMessage>>,
-    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<GreetMessage>) -> 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<mpsc::Sender<GreetMessage>> {
-        self.address_book.remove(&id)
-    }
-
-    pub fn get_sender(&self, id: GreeterId) -> Option<&mpsc::Sender<GreetMessage>> {
-        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<Greeter>,
-    response_rx: mpsc::Receiver<GreetResponse>,
-    response_tx: mpsc::Sender<GreetResponse>,
-}
-
-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<GreetResponse> {
-        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(())
-    }
-}