stash: rewrite

Added desgin doc and goals. Got a reference implementation that kimi spat out, looks
like something from hardware/embedded guys, which is a great sign.

Cargo.lock

@@ -2,6 +2,26 @@
 # It is not intended for manual editing.
 version = 4
 
+[[package]]
+name = "bytemuck"
+version = "1.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fbdf580320f38b612e485521afda1ee26d10cc9884efaaa750d383e13e3c5f4"
+dependencies = [
+ "bytemuck_derive",
+]
+
+[[package]]
+name = "bytemuck_derive"
+version = "1.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f9abbd1bc6865053c427f7198e6af43bfdedc55ab791faed4fbd361d789575ff"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "bytes"
 version = "1.11.0"
@@ -48,6 +68,7 @@ dependencies = [
 name = "swactor"
 version = "0.1.0"
 dependencies = [
+ "bytemuck",
  "tokio",
  "tokio-util",
 ]

Cargo.toml

@@ -7,5 +7,6 @@ edition = "2024"
 crate-type = ["cdylib", "rlib"]
 
 [dependencies]
+bytemuck = { version = "1.24.0", features = ["derive"] }
 tokio = { version = "1.48.0", features = ["rt", "macros"] }
 tokio-util = "0.7.17"

DESIGN.md

@@ -0,0 +1,35 @@
+# 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 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
+
+    - 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 and Router
+
+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 message router:
+        the router is responsible for ensuring messages posted by actors get delivered to the appropriate inbox.

examples/hello.rs

@@ -1,46 +1,63 @@
-use swactor::Actor;
-use tokio::sync::oneshot;
 
-pub enum GreeterMessage {
+// use std::sync::{Arc, atomic::AtomicBool};
-    Name(String),
-}
 
-pub enum GreeterResponse {
+// use swactor::error::*;
-    Hello(String),
+// use tokio::task::JoinHandle;
-}
 
-pub struct Greeter;
+// struct ActorInbox {
+//     _guard: AtomicBool
+// }
 
-impl Actor for Greeter {
+// struct GenericGreeter {
-    type Message = GreeterMessage;
+//     _guard: Arc<AtomicBool>,
-    type Response = GreeterResponse;
+//     inbox: Vec<GreeterMessage>,
+//     outbox: Vec<GreeterResponse>,
+// }
 
-    fn handle_message(&self, msg: Self::Message, tx: oneshot::Sender<Self::Response>) {
+// impl GenericGreeter {
-        let rep = match msg {
+//     pub fn new() -> Self {
-            GreeterMessage::Name(name) => GreeterResponse::Hello(format!("Hello, {name}!")),
+//         Self {
-        };
+//             _guard: Arc::new(AtomicBool::new(false)),
+//             inbox: Vec::new(),
+//             outbox: Vec::new(),
+//         }
+//     }
+// }
 
-        if let Err(_) = tx.send(rep) {
-            // Greeter is not responsible for a dropped Receiver
-        }
-    }
-}
 
-fn main() {
+// pub enum GreeterMessage {
-    let rt = tokio::runtime::Builder::new_current_thread()
+//     Name(String),
-        .build()
+// }
-        .expect("failed to build runtime");
-    let greeter = Greeter.spawn(&rt);
 
-    let response = rt
+// pub enum GreeterResponse {
-        .block_on(async move {
+//     Hello(String),
-            greeter
+// }
-                .send(GreeterMessage::Name("world".to_string()))
-                .await
-        })
-        .expect("failed to get respose");
 
-    match response {
+// pub struct Greeter;
-        GreeterResponse::Hello(hello) => println!("{hello}"),
+
-    }
+
-}
+// fn main() {
+//     let rt = tokio::runtime::Builder::new_current_thread()
+//         .build()
+//         .expect("failed to build runtime");
+
+//     let greet = GenericGreeter::spawn(&rt);
+
+//     let res = rt.block_on(async {greet.await}).expect("runtime error").expect("greeter error");
+
+//     println!("Success!");
+
+//         // let greeter = Greeter.spawn(&rt);
+
+//     // 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}"),
+//     // }
+// }

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<A> = (
+    <A as Actor>::Message,
+    oneshot::Sender<<A as Actor>::Response>,
+);
+
+/// Wrapper defining the transmission end of a Request/Response channel with an `Actor`
+pub struct ActorRequestSender<A: Actor>(mpsc::Sender<ActorRequest<A>>);
+
+impl<A: Actor> ActorRequestSender<A> {
+    pub async fn send(&self, request: A::Message) -> Result<A::Response> {
+        let (tx, rx) = oneshot::channel::<A::Response>();
+        self.0.send((request, tx)).await.map_err(convert_err)?;
+
+        rx.await.map_err(convert_err)
+    }
+}
+
+impl<A: Actor> From<mpsc::Sender<ActorRequest<A>>> for ActorRequestSender<A> {
+    fn from(value: mpsc::Sender<ActorRequest<A>>) -> Self {
+        Self(value)
+    }
+}
+
+impl<A: Actor> Clone for ActorRequestSender<A> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
+
+/// Combined 'JoinHandle' to await the actor process and 'Sender' for communication
+pub struct Handle<A>
+where
+    A: Actor,
+{
+    cancel_token: CancellationToken,
+    tx: ActorRequestSender<A>,
+    /// the task drops when the `JoinHandle` does, so be careful with the `Handle`
+    _handle: JoinHandle<Result<()>>,
+    // to prevent accidental swaps, strongly type the handle
+    _type: std::marker::PhantomData<A>,
+}
+
+impl<A: Actor> Handle<A> {
+    /// 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<A::Response> {
+        self.tx.send(msg).await
+    }
+
+    /// Get a cloned sender for messaging the `Actor` this handle is for
+    pub fn get_connection(&self) -> ActorRequestSender<A> {
+        self.tx.clone()
+    }
+}
+
+impl<A: Actor> Drop for Handle<A> {
+    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<Self::Response>);
+
+    /// Spawns the `Actor` utilizing the given runtime context
+    /// Only `tokio` runtime is accepted for now
+    fn spawn(self, ctx: &tokio::runtime::Runtime) -> Handle<Self> {
+        let cancel_token = CancellationToken::new();
+        let cancel = cancel_token.clone();
+
+        let (tx, mut rx) =
+            mpsc::channel::<(Self::Message, oneshot::Sender<Self::Response>)>(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::<Self>,
+        }
+    }
+}

src/lib.rs

@@ -1,118 +1,164 @@
 pub mod error;
 
-/// Public export as the oneshot channel is in the `Actor` trait signature
+use std::{
-pub use tokio::sync::oneshot;
+    marker::PhantomData,
-
+    sync::{Mutex, mpsc::TryRecvError},
-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<A> = (
-    <A as Actor>::Message,
-    oneshot::Sender<<A as Actor>::Response>,
-);
-
-/// Wrapper defining the transmission end of a Request/Response channel with an `Actor`
-pub struct ActorRequestSender<A: Actor>(mpsc::Sender<ActorRequest<A>>);
-
-impl<A: Actor> ActorRequestSender<A> {
-    pub async fn send(&self, request: A::Message) -> Result<A::Response> {
-        let (tx, rx) = oneshot::channel::<A::Response>();
-        self.0.send((request, tx)).await.map_err(convert_err)?;
-
-        rx.await.map_err(convert_err)
-    }
-}
-
-impl<A: Actor> From<mpsc::Sender<ActorRequest<A>>> for ActorRequestSender<A> {
-    fn from(value: mpsc::Sender<ActorRequest<A>>) -> Self {
-        Self(value)
-    }
-}
-
-impl<A: Actor> Clone for ActorRequestSender<A> {
-    fn clone(&self) -> Self {
-        Self(self.0.clone())
-    }
-}
-
-/// Combined 'JoinHandle' to await the actor process and 'Sender' for communication
-pub struct Handle<A>
-where
-    A: Actor,
-{
-    cancel_token: CancellationToken,
-    tx: ActorRequestSender<A>,
-    /// the task drops when the `JoinHandle` does, so be careful with the `Handle`
-    _handle: JoinHandle<Result<()>>,
-    // to prevent accidental swaps, strongly type the handle
-    _type: std::marker::PhantomData<A>,
-}
-
-impl<A: Actor> Handle<A> {
-    /// 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<A::Response> {
-        self.tx.send(msg).await
-    }
-
-    /// Get a cloned sender for messaging the `Actor` this handle is for
-    pub fn get_connection(&self) -> ActorRequestSender<A> {
-        self.tx.clone()
-    }
-}
-
-impl<A: Actor> Drop for Handle<A> {
-    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<Self::Response>);
-
-    /// Spawns the `Actor` utilizing the given runtime context
-    /// Only `tokio` runtime is accepted for now
-    fn spawn(self, ctx: &tokio::runtime::Runtime) -> Handle<Self> {
-        let cancel_token = CancellationToken::new();
-        let cancel = cancel_token.clone();
-
-        let (tx, mut rx) =
-            mpsc::channel::<(Self::Message, oneshot::Sender<Self::Response>)>(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; },
-                        }
-                    }
 };
+
+use crate::error::{Error, Result, convert_err};
+
+use std::sync::mpsc;
+
+use bytemuck::{Pod, Zeroable};
+
+#[repr(C)]
+#[derive(Copy, Clone, Pod, Zeroable)]
+struct GreeterState {
+    pub num_greeted: usize,
 }
 
-            res
+enum GreetMessage {
-        });
+    Name(String),
+}
 
-        Handle {
+enum GreetResponse {
-            cancel_token,
+    Greeting(String),
-            _handle: handle,
+}
-            tx: tx.into(),
+
-            _type: std::marker::PhantomData::<Self>,
+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(())
+    }
+}