feat: rewrite
Did not like the requirement of a tokio runtime and async/await. Rewrote to use lock free queues (`crossbeam_queue::ArrayQueue`) as the basic primitive to enable a runtime consisting of many psuedo-processes that can pass messages to each other.
This commit is contained in:
Zachery Aaron Shores-Chmielewski
parent
b8b5b839df
commit
4edca69d7c
6 changed files with 446 additions and 213 deletions
109
Cargo.lock
generated
109
Cargo.lock
generated
|
|
@ -3,102 +3,53 @@
|
|||
version = 4
|
||||
|
||||
[[package]]
|
||||
name = "bytes"
|
||||
version = "1.11.0"
|
||||
name = "cfg-if"
|
||||
version = "1.0.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3"
|
||||
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
|
||||
|
||||
[[package]]
|
||||
name = "futures-core"
|
||||
version = "0.3.31"
|
||||
name = "crossbeam-queue"
|
||||
version = "0.3.12"
|
||||
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"
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro2"
|
||||
version = "1.0.103"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5ee95bc4ef87b8d5ba32e8b7714ccc834865276eab0aed5c9958d00ec45f49e8"
|
||||
checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
|
||||
dependencies = [
|
||||
"unicode-ident",
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quote"
|
||||
version = "1.0.42"
|
||||
name = "crossbeam-utils"
|
||||
version = "0.8.21"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a338cc41d27e6cc6dce6cefc13a0729dfbb81c262b1f519331575dd80ef3067f"
|
||||
checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
|
||||
|
||||
[[package]]
|
||||
name = "getrandom"
|
||||
version = "0.2.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"cfg-if",
|
||||
"libc",
|
||||
"wasi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.180"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc"
|
||||
|
||||
[[package]]
|
||||
name = "swactor"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"crossbeam-queue",
|
||||
"getrandom",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "2.0.111"
|
||||
name = "wasi"
|
||||
version = "0.11.1+wasi-snapshot-preview1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "390cc9a294ab71bdb1aa2e99d13be9c753cd2d7bd6560c77118597410c4d2e87"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"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"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9312f7c4f6ff9069b165498234ce8be658059c6728633667c526e27dc2cf1df5"
|
||||
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
|
||||
|
|
|
|||
|
|
@ -6,6 +6,10 @@ edition = "2024"
|
|||
[lib]
|
||||
crate-type = ["cdylib", "rlib"]
|
||||
|
||||
[features]
|
||||
default = ["getrandom"]
|
||||
getrandom = ["dep:getrandom"]
|
||||
|
||||
[dependencies]
|
||||
tokio = { version = "1.48.0", features = ["rt", "macros"] }
|
||||
tokio-util = "0.7.17"
|
||||
getrandom = { version = "0.2", optional = true }
|
||||
crossbeam-queue = "0.3.12"
|
||||
|
|
|
|||
77
DESIGN.md
Normal file
77
DESIGN.md
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
# 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 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.
|
||||
|
||||
## 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
|
||||
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.
|
||||
|
||||
- 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
|
||||
|
||||
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 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 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.
|
||||
|
||||
### 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
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
|
@ -1,46 +1,58 @@
|
|||
use swactor::Actor;
|
||||
use tokio::sync::oneshot;
|
||||
use swactor::{ActorAddress, ActorInterface, Message, Runtime, RuntimeFlavor};
|
||||
|
||||
pub enum GreeterMessage {
|
||||
Name(String),
|
||||
#[derive(Debug, Default)]
|
||||
struct Greeter {
|
||||
pub num_greeted: usize,
|
||||
}
|
||||
|
||||
pub enum GreeterResponse {
|
||||
Hello(String),
|
||||
#[derive(Debug, Default, Clone)]
|
||||
struct GreetMessage {
|
||||
/// who do we greet?
|
||||
who: String,
|
||||
|
||||
/// who do we send out greeting back to?
|
||||
return_addr: ActorAddress,
|
||||
}
|
||||
impl Message for GreetMessage {}
|
||||
|
||||
pub struct Greeter;
|
||||
impl ActorInterface for Greeter {
|
||||
type Incoming = GreetMessage;
|
||||
type Response = GreetResponse;
|
||||
|
||||
impl Actor for Greeter {
|
||||
type Message = GreeterMessage;
|
||||
type Response = GreeterResponse;
|
||||
|
||||
fn handle_message(&self, msg: Self::Message, tx: oneshot::Sender<Self::Response>) {
|
||||
let rep = match msg {
|
||||
GreeterMessage::Name(name) => GreeterResponse::Hello(format!("Hello, {name}!")),
|
||||
};
|
||||
|
||||
if let Err(_) = tx.send(rep) {
|
||||
// Greeter is not responsible for a dropped Receiver
|
||||
fn handle(&mut self, ctx: &Runtime, msg: GreetMessage) {
|
||||
let res = GreetResponse(format!("Hello, {}!", msg.who));
|
||||
self.num_greeted += 1;
|
||||
if let Err(_) = ctx.send_to(msg.return_addr, res) {
|
||||
// no error handling
|
||||
self.num_greeted -= 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone)]
|
||||
struct GreetResponse(String);
|
||||
impl Message for GreetResponse {}
|
||||
|
||||
fn main() {
|
||||
let rt = tokio::runtime::Builder::new_current_thread()
|
||||
.build()
|
||||
.expect("failed to build runtime");
|
||||
let greeter = Greeter.spawn(&rt);
|
||||
let mut rt = Runtime::new(100, Some(RuntimeFlavor::SingleThreaded));
|
||||
let addr = rt
|
||||
.spawn(Greeter::default())
|
||||
.expect("failed to spawn greeter");
|
||||
let inbox = rt.new_inbox::<GreetResponse>();
|
||||
|
||||
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}"),
|
||||
rt.send_to(
|
||||
addr,
|
||||
GreetMessage {
|
||||
who: "world".into(),
|
||||
return_addr: *inbox.addr(),
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
for _ in 0..3 {
|
||||
rt.tick();
|
||||
}
|
||||
|
||||
let resp = inbox.try_recv().expect("greeter should have said hello");
|
||||
|
||||
println!("{}", resp.0);
|
||||
}
|
||||
|
|
|
|||
333
src/lib.rs
333
src/lib.rs
|
|
@ -1,118 +1,251 @@
|
|||
mod ring_buffer;
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use crossbeam_queue::ArrayQueue;
|
||||
use ring_buffer::{Receiver, Sender};
|
||||
|
||||
pub mod error;
|
||||
use error::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)
|
||||
}
|
||||
#[cfg(feature = "getrandom")]
|
||||
pub fn get_random(buf: &mut [u8]) {
|
||||
getrandom::getrandom(buf).unwrap()
|
||||
}
|
||||
|
||||
impl<A: Actor> From<mpsc::Sender<ActorRequest<A>>> for ActorRequestSender<A> {
|
||||
fn from(value: mpsc::Sender<ActorRequest<A>>) -> Self {
|
||||
Self(value)
|
||||
}
|
||||
/// The strategy for message processing is such:
|
||||
/// if total_messages < WATERLEVEL:
|
||||
/// process all
|
||||
/// else
|
||||
/// process total_messages // 2
|
||||
const WATERLEVEL: usize = 10;
|
||||
|
||||
const DEFAULT_INBOX_CAPACITY: usize = 100;
|
||||
|
||||
pub trait Message: 'static + Sized + Clone + Send {}
|
||||
pub type Envelope = Box<dyn std::any::Any + Send>;
|
||||
|
||||
pub trait ActorInterface: 'static + Send {
|
||||
type Incoming: Message;
|
||||
type Response: Message;
|
||||
fn handle(&mut self, ctx: &Runtime, msg: Self::Incoming);
|
||||
}
|
||||
|
||||
impl<A: Actor> Clone for ActorRequestSender<A> {
|
||||
fn clone(&self) -> Self {
|
||||
Self(self.0.clone())
|
||||
}
|
||||
}
|
||||
pub type ActorAddress = u64;
|
||||
|
||||
/// Combined 'JoinHandle' to await the actor process and 'Sender' for communication
|
||||
pub struct Handle<A>
|
||||
pub struct Actor<A>
|
||||
where
|
||||
A: Actor,
|
||||
A: ActorInterface,
|
||||
{
|
||||
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>,
|
||||
_addr: ActorAddress,
|
||||
inbox: Receiver<A::Incoming>,
|
||||
inner: 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()
|
||||
}
|
||||
/// Trait for type-erased actors
|
||||
trait AnyActor: Send {
|
||||
fn tick(&mut self, ctx: &Runtime);
|
||||
}
|
||||
|
||||
impl<A: Actor> Drop for Handle<A> {
|
||||
fn drop(&mut self) {
|
||||
self.cancel_token.cancel();
|
||||
}
|
||||
}
|
||||
impl<A> AnyActor for Actor<A>
|
||||
where
|
||||
A: ActorInterface,
|
||||
{
|
||||
fn tick(&mut self, ctx: &Runtime) {
|
||||
let total_messages = self.inbox.len();
|
||||
let messages_to_process = if total_messages < WATERLEVEL {
|
||||
total_messages
|
||||
} else {
|
||||
total_messages >> 1
|
||||
};
|
||||
|
||||
/// 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; },
|
||||
}
|
||||
}
|
||||
};
|
||||
for _ in 0..messages_to_process {
|
||||
match self.inbox.try_recv() {
|
||||
Some(msg) => self.inner.handle(ctx, msg),
|
||||
None => unreachable!(
|
||||
"We checked number of unprocessed messages in the queue ahead of processing"
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct Inbox<M: Message> {
|
||||
addr: ActorAddress,
|
||||
inner: Receiver<M>,
|
||||
}
|
||||
|
||||
impl<M: Message> Inbox<M> {
|
||||
pub fn addr(&self) -> &ActorAddress {
|
||||
&self.addr
|
||||
}
|
||||
|
||||
pub fn try_recv(&self) -> Option<M> {
|
||||
self.inner.try_recv()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub enum RuntimeFlavor {
|
||||
#[default]
|
||||
SingleThreaded,
|
||||
Multithreaded(usize),
|
||||
}
|
||||
|
||||
pub struct Runtime {
|
||||
flavor: RuntimeFlavor,
|
||||
router: Router,
|
||||
router_inbox: Sender<RouterMessage>,
|
||||
actor_queue: ArrayQueue<Box<dyn AnyActor>>,
|
||||
}
|
||||
|
||||
impl Runtime {
|
||||
pub fn new(capacity: usize, flavor: Option<RuntimeFlavor>) -> Self {
|
||||
let router = Router::new(DEFAULT_INBOX_CAPACITY);
|
||||
let router_inbox = router.new_sender();
|
||||
Self {
|
||||
flavor: flavor.unwrap_or_default(),
|
||||
router,
|
||||
router_inbox,
|
||||
actor_queue: ArrayQueue::new(capacity),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
|
||||
let addr = {
|
||||
let mut bytes = u64::to_le_bytes(0);
|
||||
get_random(&mut bytes);
|
||||
u64::from_le_bytes(bytes)
|
||||
};
|
||||
let inbox = Receiver::<A::Incoming>::new(DEFAULT_INBOX_CAPACITY);
|
||||
let sender = inbox.new_sender();
|
||||
|
||||
// Register the sender with the router
|
||||
let _ = self
|
||||
.router_inbox
|
||||
.try_send(RouterMessage::AddAddr(addr, Box::new(sender)));
|
||||
|
||||
self.actor_queue
|
||||
.push(Box::new(Actor {
|
||||
_addr: addr,
|
||||
inbox,
|
||||
inner: actor,
|
||||
}))
|
||||
.map_err(|_| Error::from("Runtime error: Failed to spawn actor."))?;
|
||||
|
||||
Ok(addr)
|
||||
}
|
||||
|
||||
pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), ()> {
|
||||
let envelope: Envelope = Box::new(msg);
|
||||
self.router_inbox
|
||||
.try_send(RouterMessage::SendToAddr {
|
||||
addr,
|
||||
msg: envelope,
|
||||
})
|
||||
.map_err(|_| ())
|
||||
}
|
||||
|
||||
pub fn tick(&mut self) {
|
||||
// Pop actor, tick it, push it back
|
||||
if let Some(mut actor) = self.actor_queue.pop() {
|
||||
actor.tick(self);
|
||||
let _ = self.actor_queue.push(actor);
|
||||
}
|
||||
|
||||
match self.flavor {
|
||||
RuntimeFlavor::Multithreaded(_) => (), // router has its own thread
|
||||
RuntimeFlavor::SingleThreaded => self.router.tick(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_inbox<M: Message>(&self) -> Inbox<M> {
|
||||
let addr = {
|
||||
let mut bytes = u64::to_le_bytes(0);
|
||||
get_random(&mut bytes);
|
||||
u64::from_le_bytes(bytes)
|
||||
};
|
||||
let receiver = Receiver::<M>::new(DEFAULT_INBOX_CAPACITY);
|
||||
let sender = receiver.new_sender();
|
||||
// Register the sender with the router
|
||||
let _ = self
|
||||
.router_inbox
|
||||
.try_send(RouterMessage::AddAddr(addr, Box::new(sender)));
|
||||
Inbox {
|
||||
addr,
|
||||
inner: receiver,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub trait SenderT: Send {
|
||||
fn try_send(&self, envelope: Envelope);
|
||||
}
|
||||
|
||||
impl<M: Message> SenderT for Sender<M> {
|
||||
fn try_send(&self, envelope: Envelope) {
|
||||
if let Ok(msg) = envelope.downcast::<M>() {
|
||||
let _ = Sender::try_send(self, *msg);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Internal messages for the Router's own inbox
|
||||
pub enum RouterMessage {
|
||||
/// register addrs <addr> with sender <sender>
|
||||
AddAddr(ActorAddress, Box<dyn SenderT>),
|
||||
/// remove an actor from the address book
|
||||
RemoveAddr(ActorAddress),
|
||||
/// send <msg> to <addr>
|
||||
SendToAddr { addr: ActorAddress, msg: Envelope },
|
||||
}
|
||||
|
||||
struct Router {
|
||||
directory: HashMap<ActorAddress, Box<dyn SenderT>>,
|
||||
inbox: Receiver<RouterMessage>,
|
||||
}
|
||||
|
||||
impl Router {
|
||||
pub fn new(cap: usize) -> Self {
|
||||
Self {
|
||||
directory: HashMap::new(),
|
||||
inbox: Receiver::new(cap),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn tick(&mut self) {
|
||||
let total_messages = self.inbox.len();
|
||||
let messages_to_process = if total_messages < WATERLEVEL {
|
||||
total_messages
|
||||
} else {
|
||||
total_messages >> 1
|
||||
};
|
||||
|
||||
for _ in 0..messages_to_process {
|
||||
match self.inbox.try_recv() {
|
||||
Some(msg) => self.handle(msg),
|
||||
None => unreachable!("We ran checks on total messages before processing."),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_sender(&self) -> Sender<RouterMessage> {
|
||||
self.inbox.new_sender()
|
||||
}
|
||||
|
||||
fn handle(&mut self, msg: RouterMessage) {
|
||||
match msg {
|
||||
RouterMessage::AddAddr(addr, sender) => {
|
||||
self.directory.insert(addr, sender);
|
||||
}
|
||||
RouterMessage::RemoveAddr(addr) => {
|
||||
self.directory.remove(&addr);
|
||||
}
|
||||
RouterMessage::SendToAddr { addr, msg } => {
|
||||
if let Some(sender) = self.directory.get(&addr) {
|
||||
sender.try_send(msg);
|
||||
}
|
||||
}
|
||||
|
||||
res
|
||||
});
|
||||
|
||||
Handle {
|
||||
cancel_token,
|
||||
_handle: handle,
|
||||
tx: tx.into(),
|
||||
_type: std::marker::PhantomData::<Self>,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
56
src/ring_buffer.rs
Normal file
56
src/ring_buffer.rs
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
pub use crossbeam_queue::ArrayQueue;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// 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)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the number of elements in the inner queue
|
||||
pub fn len(&self) -> usize {
|
||||
self.queue.len()
|
||||
}
|
||||
|
||||
/// 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)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue