stash: rewrite wip

We have a basic ring buffer for dealing with concurrent message exchange. Currently
in the middle of reifying the traits, types and structs needed for the `hello.rs`
example of the simple `Hello, World!` greeter type actor.

Cargo.lock

@@ -23,28 +23,19 @@ dependencies = [
 ]
 
 [[package]]
-name = "bytes"
-version = "1.11.0"
+name = "crossbeam-queue"
+version = "0.3.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3"
+checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
+dependencies = [
+ "crossbeam-utils",
+]
 
 [[package]]
-name = "futures-core"
-version = "0.3.31"
+name = "crossbeam-utils"
+version = "0.8.21"
 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"
+checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
 
 [[package]]
 name = "proc-macro2"
@@ -69,8 +60,7 @@ name = "swactor"
 version = "0.1.0"
 dependencies = [
  "bytemuck",
- "tokio",
- "tokio-util",
+ "crossbeam-queue",
 ]
 
 [[package]]
@@ -84,40 +74,6 @@ dependencies = [
  "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"

Cargo.toml

@@ -8,5 +8,4 @@ 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"
+crossbeam-queue = "0.3.12"

DESIGN.md

@@ -48,3 +48,31 @@ The router is the engine for message delivery. It runs on its own thread and pos
     - 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<T> {
+    // Start with lock-free ring buffer
+    ring: AtomicRingBuffer<T>,
+    
+    // When full, spill into a Mutex<VecDeque<T>>
+    overflow: parking_lot::Mutex<VecDeque<T>>,
+    
+    // Track overflow frequency to resize ring proactively
+    overflow_count: AtomicUsize,
+}
+
+impl<T> HybridChannel<T> {
+    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
+        }
+    }
+}
+```

examples/hello.rs

@@ -1,63 +1,34 @@
+use bytemuck::{Pod, Zeroable};
+use swactor::{ActorAddress, ActorInterface, Message, Runtime};
 
-// use std::sync::{Arc, atomic::AtomicBool};
+#[repr(C)]
+#[derive(Pod, Zeroable)]
+struct Greeter {
+    pub num_greeted: usize,
+}
 
-// use swactor::error::*;
-// use tokio::task::JoinHandle;
-
-// struct ActorInbox {
-//     _guard: AtomicBool
-// }
-
-// struct GenericGreeter {
-//     _guard: Arc<AtomicBool>,
-//     inbox: Vec<GreeterMessage>,
-//     outbox: Vec<GreeterResponse>,
-// }
-
-// impl GenericGreeter {
-//     pub fn new() -> Self {
-//         Self {
-//             _guard: Arc::new(AtomicBool::new(false)),
-//             inbox: Vec::new(),
-//             outbox: Vec::new(),
-//         }
-//     }
-// }
+#[derive(Clone)]
+struct GreetMessage {
+    name: String,
+    addr: ActorAddress,
+}
+impl Message for GreetMessage {}
 
 
-// pub enum GreeterMessage {
-//     Name(String),
-// }
+impl ActorInterface<GreetMessage> for Greeter {
+    fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) {
+        let res = GreetResponse(format!("Hello, {}!", msg.name));
+        if let Err(_) = ctx.send(res, msg.addr) {
+            // no error handling
+        }
+    }
+}
 
-// pub enum GreeterResponse {
-//     Hello(String),
-// }
+#[derive(Debug, Default, Clone)]
+struct GreetResponse(String);
+impl Message for GreetResponse {}
 
-// pub struct Greeter;
+fn main() {
+    let rt = Runtime::new();
 
-
-// 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.rs

@@ -1,39 +1,47 @@
-// mod kimi;
+mod ring_buffer;
+use bytemuck::{Pod, Zeroable};
+use ring_buffer::{Receiver, Sender};
 
 pub mod error;
 
-use std::{
-    marker::PhantomData, mem::MaybeUninit, sync::{Mutex, atomic::AtomicUsize, mpsc::TryRecvError}
-};
+pub trait Message: 'static + Sized + Clone {}
 
-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);
+pub trait ActorInterface<M: Message>: Pod + Zeroable {
+    fn handle(&mut self, ctx: &Runtime, msg: M);
 }
 
+pub type ActorAddress = u64;
 
-
-
-
-use bytemuck::{Pod, Zeroable};
-
-#[repr(C)]
-#[derive(Copy, Clone, Pod, Zeroable)]
-struct GreeterState {
-    pub num_greeted: usize,
+pub struct Actor<S, M, N>
+where
+    M: Message,
+    N: Message,
+    S: ActorInterface<M>,
+{
+    inbox: Receiver<M>,
+    outbox: Sender<N>,
+    state: S,
 }
 
-enum GreetMessage {
-    Name(String),
+pub struct Runtime {
+    router: (),
+    actor_queue: (),
 }
 
-enum GreetResponse {
-    Greeting(String),
+impl Runtime {
+    pub fn new() -> Self {
+        Self {
+            router: (),
+            actor_queue: (),
+        }
+    }
+
+    pub fn send<M>(&self, msg: M, addr: ActorAddress) -> Result<(), M> {
+        Err(msg)
+    }
 }
 
-type GreeterId = u64;
+pub struct MessageRouter<M> {
+    inbox: Receiver<M>,
+    address_book: (),
+}

src/ring_buffer.rs

@@ -0,0 +1,49 @@
+use std::sync::Arc;
+use crossbeam_queue::ArrayQueue;
+
+/// 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<T> {
+    queue: Arc<ArrayQueue<T>>,
+}
+
+impl<T> Receiver<T> {
+    /// 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))
+        }
+    }
+    
+    /// Attempt to retrieve a value from the queue. Returns `None` if empty
+    pub fn try_recv(&self) -> Option<T> {
+        self.queue.pop()
+    }
+
+    /// Construct a new `Sender` assosciated with this queue.
+    pub fn new_sender(&self) -> Sender<T> {
+        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::<T>::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<T> {
+    queue: Arc<ArrayQueue<T>>,
+}
+
+impl<T> Sender<T> {
+    /// 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)
+    }
+}