From 08b241481f14deaec253f57188469a9c1f05e227 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Thu, 22 Jan 2026 14:04:07 +0700 Subject: [PATCH] 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 | 64 ++++++----------------------------- Cargo.toml | 3 +- DESIGN.md | 30 ++++++++++++++++- examples/hello.rs | 83 +++++++++++++++------------------------------- src/lib.rs | 60 ++++++++++++++++++--------------- src/ring_buffer.rs | 49 +++++++++++++++++++++++++++ 6 files changed, 150 insertions(+), 139 deletions(-) create mode 100644 src/ring_buffer.rs diff --git a/Cargo.lock b/Cargo.lock index 318f003..04eda39 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 3ce83f1..95b6f23 100644 --- a/Cargo.toml +++ b/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" diff --git a/DESIGN.md b/DESIGN.md index 1377243..a2e804a 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -47,4 +47,32 @@ 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. \ No newline at end of file + 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 6fa5b1b..8ac6d71 100644 --- a/examples/hello.rs +++ b/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, -// inbox: Vec, -// outbox: Vec, -// } - -// 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 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}"), -// // } -// } +} diff --git a/src/lib.rs b/src/lib.rs index 2120990..83a600c 100644 --- a/src/lib.rs +++ b/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 { - fn handle(&mut self, msg: M); +pub trait ActorInterface: 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 +where + M: Message, + N: Message, + S: ActorInterface, +{ + inbox: Receiver, + outbox: Sender, + 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(&self, msg: M, addr: ActorAddress) -> Result<(), M> { + Err(msg) + } } -type GreeterId = u64; +pub struct MessageRouter { + inbox: Receiver, + address_book: (), +} diff --git a/src/ring_buffer.rs b/src/ring_buffer.rs new file mode 100644 index 0000000..f82f6e0 --- /dev/null +++ b/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 { + 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)) + } + } + + /// 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) + } +} \ No newline at end of file