For single-actors I think the unbounded channels work fine, because implicitly they're sequential, conceptually (of course you can do more complex things).

However for actor pools, I want the ability to fill up all the actors in a given pool, and not keep spamming messages until they're ready to accept some more. I can't use .send().await like I would in a normal actor, because that's effectively just making the pool meaningless. I can't send_async unless it's trivially small data, because it's not going to have any backpressure, and will effectively memory leak.

In this case, I think it's fine to have a model of something like...

pool.send_async()
pool.wait_until_available().await

This does of course have an implicit race condition, which probably doesn't matter for a lot of use cases, but it could be baked into another send method like...

pool.send_when_available().await

For users who want more backpressure than the pool actor count provides, they can easily add a buffered channel on top of this (or another actor itself that buffers).

I have a use case where I have an actor managing some GPU state and I want to interact with it from the synchronous context of a winit application. The way I can currently do this is by essentially using a tokio::sync::oneshot channel and spawning a task with tokio which populates the channel with the result of sending a message to the actor. I suspect that internally kameo is probably using a oneshot channel or similar already, and it might be possible to interact with it to receive the response in a sync context.

I would propose a similar API to https://docs.rs/kameo/latest/kameo/actor/struct.ActorRef.html#method.send, which could be called ActorRef::send_sync. This would return an object that could be polled to see if it is complete. Polling the future directly in a sync context is not possible due to the lack of the async runtime's context that must be passed in.

Hello,

thanks for sharing the project. I'm tired of typing my own message passing with channels, an enum and tasks over and over again ;-)

What's the rational behind using unbounded channels? I debugged channel usage a lot and always came to the conclusion that a stuck task/actor that is backpressured from somewhere is easier to find than something like a memory leak in an unbounded channel.

thanks

@flxo

The derive Reply include Sync trait, but BoxDebug doesn't include it.

#[inline]
fn into_boxed_err(self) -> ::std::option::Option<::std::boxed::Box<dyn ::std::fmt::Debug + ::std::marker::Send + ::std::marker::Sync + 'static>> {
    ::std::option::Option::None
}

To process a Stream in a Kameo actor I think currently the only solution is to spawn a (Tokio) task and forward each item to the actor. Each item must be sent through the actors queue. Spawning the task and moving a ActorRef inside etc... is boilerplate and boring.

I remember that at least actix has a feature to attach streams to actors and wondered if that could be considered a feature for kameo as well?

If yes:

I hacked a prototype and noticed that every message in kameo get's boxed and was very surprised because this is generally considered as a possible performance hit. Is there no way around that? Sadly I had to do the same for the stream feature.

In the prototype an actor is not notified if a stream completes. This could be easily achieved by using StreamNotifyClose. This would be a API decision. I think it's a nice feature but adds noise and possible confusion because the handler need to have an Option around each message. This implicity must be know and may be hard to get for beginners.

Thanks for your thoughts!

#[derive(Actor)]
pub struct MyActor<T> {
    arena: Vec<T>,
}

#[messages]
impl<T: Send + 'static> MyActor<T> {
    #[message]
    pub fn add(&mut self, t: T) {}
}

and i get the compile error:

error[E0412]: cannot find type `T` in this scope
   |
24 |     pub fn add(&mut self, t: T) {}
   |                              ^ not found in this scope

If I declare an actor with the derive macro it's not possible to have an additional impl block that override the default implementations from here.

Am I forced to not use the derive macro if I need a custom impl for anything in Actor?

The macro https://github.com/tqwewe/kameo/blob/bb4c8ffdbda8bd4f56cde00ffe8bc38e7d557197/crates/kameo/examples/macro.rs#L19

example shows how to define a handler fn which is not async. Is that on purpose and what's the reason for that? I would have expected this to be async or return something impl Future<...>

ActorRef's send methods which receive replies now return SendResult, allowing message replies to be either received asyncronously, or blocking.

actor_ref.send(msg).await?; // Async recv
actor_ref.send(msg).blocking_recv()?; // Blocking recv

Closes https://github.com/tqwewe/kameo/issues/25

Closes https://github.com/tqwewe/kameo/issues/15

This PR removes stateless actors, as the implementation for it conflicts with features required by this PR. This is a breaking change.

A new trait StreamMessage is provided, which can be implemented by actors to allow them to be attached to streams using the ActorRef::attach_stream method.

ping @flxo

Hello,

Just want to share this. I was curious about the overhead of kameo compared to "plain" channels and a task.

1. Profiling

Lets see where kameo spends it's cpu time. The following minimal program built in release mode (+ debug=true in the profile) and executed on a macOS on 2,4 GHz 8-Core Intel Core i9.

use kameo::{
    message::{Context, Message},
    Actor,
};

#[derive(Default)]
pub struct MyActor;

impl Actor for MyActor {}

impl Message<()> for MyActor {
    type Reply = ();

    async fn handle(&mut self, _: (), _ctx: Context<'_, Self, Self::Reply>) -> Self::Reply {}
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let my_actor_ref = kameo::spawn(MyActor::default());
    loop {
        my_actor_ref.send(()).await?;
    }
}

produces this profile. Captured with samply(❤️).

You can see the two big blocks: kameo::actor::actor_ref::ActorRef::send::{{closure}} and <kameo::actor_kind::SyncActor as kameo::actor_kind::ActorState>::handle_message::{{closure}} in the flame graph beside the usual Tokio runtime stuff. Interesting is: ~5.4% overhead due to Box::new in the tx path. ~8.3% (+4% free) for boxing on the rx side.

2. Comparison to a plain channel + Tokio task

I used criterion to compare a simple echo actor that is processing sync calls with a plain Tokio task that sends a reply on a received on shot Sender. Think this gives an impression about the overhead of the convenience that kameo brings.

use criterion::Criterion;
use criterion::{criterion_group, criterion_main};
use kameo::{
    message::{Context, Message},
    Actor,
};
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::task;

fn actor(c: &mut Criterion) {
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    let _guard = rt.enter();

    struct BenchActor;

    impl Actor for BenchActor {}

    impl Message<u32> for BenchActor {
        type Reply = u32;

        async fn handle(&mut self, msg: u32, _ctx: Context<'_, Self, Self::Reply>) -> Self::Reply {
            msg
        }
    }
    let actor_ref = kameo::actor::spawn(BenchActor {});

    c.bench_function("actor_sync_messages", |b| {
        b.to_async(&rt).iter(|| async {
            actor_ref.send(0).await.unwrap();
        });
    });
}

fn plain(c: &mut Criterion) {
    let rt = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    let _guard = rt.enter();

    // Echo task - pendant to the actor.
    let (tx, mut rx) = mpsc::unbounded_channel::<(u32, oneshot::Sender<u32>)>();
    task::spawn(async move {
        while let Some((msg, tx)) = rx.recv().await {
            tx.send(msg).unwrap();
        }
    });

    c.bench_function("plain_sync_messages_unbounded", |b| {
        b.to_async(&rt).iter(|| async {
            let (reply_tx, reply_rx) = oneshot::channel();
            tx.send((0, reply_tx)).unwrap();
            reply_rx.await.unwrap();
        });
    });

    // Echo task bounded - pendant to the actor.
    let (tx, mut rx) = mpsc::channel::<(u32, oneshot::Sender<u32>)>(10);
    task::spawn(async move {
        while let Some((msg, tx)) = rx.recv().await {
            tx.send(msg).unwrap();
        }
    });
    
    c.bench_function("plain_sync_messages_bounded", |b| {
        b.to_async(&rt).iter(|| async {
            let (reply_tx, reply_rx) = oneshot::channel();
            tx.send((0, reply_tx)).await.unwrap();
            reply_rx.await.unwrap();
        });
    });
}

criterion_group! {
    name = benches;
    config = Criterion::default();
    targets = actor, plain
}

criterion_main!(benches);

Question: Did I do something wrong here?

Output:

actor_sync_messages     time:   [750.69 ns 756.38 ns 763.39 ns]
                        change: [+0.1536% +2.0887% +4.0079%] (p = 0.03 < 0.05)
                        Change within noise threshold.
Found 14 outliers among 100 measurements (14.00%)
  3 (3.00%) low mild
  6 (6.00%) high mild
  5 (5.00%) high severe

plain_sync_messages_unbounded
                        time:   [270.36 ns 271.76 ns 273.30 ns]
                        change: [+4.3599% +5.3530% +6.4111%] (p = 0.00 < 0.05)
                        Performance has regressed.
Found 14 outliers among 100 measurements (14.00%)
  2 (2.00%) low mild
  10 (10.00%) high mild
  2 (2.00%) high severe

plain_sync_messages_bounded
                        time:   [309.61 ns 312.05 ns 314.86 ns]
                        change: [-3.6089% -1.9707% -0.2564%] (p = 0.02 < 0.05)
                        Change within noise threshold.
Found 7 outliers among 100 measurements (7.00%)
  5 (5.00%) high mild
  2 (2.00%) high severe

The difference is quite huge (~2.5). Clearly everything in between you and the raw channel costs performance but I wouldn't have expect that. I didn't examine async messages.

I'm evaluation this all here because we're thinking about using it in some networking application and thoughput matters.

Let me know if I missed something or there's a systematic error.

cheers,

@flxo