Configurable ActorPool Backpressure

[dnaut]

When creating a new ActorPool, the mailbox is a default bounded(1000) mailbox. Is there a way to set the bounded capacity of the ActorPool? The caller won't see the backpressure until the 1000 mailbox limit.

[moofone]

impl Actor for MyActor {
    type Mailbox = BoundedMailbox<Self>;

    fn new_mailbox() -> (Self::Mailbox, <Self::Mailbox as Mailbox<Self>>::Receiver) {
        BoundedMailbox::new(100)
    }

[dnaut]

Thanks for your reply. Sorry, I should have provided an example. In the ActorPool scenario, the worker mailbox can be set but not the pool mailbox. I tried creating my own ActorPool but can't see a way to make the pool mailbox configurable.

From pool.rs, this is where the ActorPool is being defined with no override on new_mailbox() trait fn.

impl<A> Actor for ActorPool<A>
where
    A: Actor,
{
    type Mailbox = BoundedMailbox<Self>;
    type Error = Infallible;
    ....

Here's an example, you can see the pool capacity is 1000 (default) and we've set the worker mailbox. Here you can see the pool requests are continuing without backpressure until we reach the default (1000). The requests are being left in the pool mailbox until the worker becomes available. Perhaps it should be pool mailbox = bounded(n) and the worker mailbox is unbounded. As a guess and API breaking, maybe Actor::new_mailbox() needs an options/capacity parameter that can be optionally passed during the prepare/spawn process.

pub struct MyWorkerMessage(usize);

#[derive(Actor)]
#[actor(name = "MyWorker", mailbox = bounded(5))]
pub struct MyWorker;

impl Message<MyWorkerMessage> for MyWorker {
    type Reply = ();
    async fn handle(&mut self, msg: MyWorkerMessage, _ctx: &mut kameo::message::Context<Self, Self::Reply>) -> Self::Reply {
        let i = msg.0;
        info!(i, "Worker start");
        tokio::spawn(async { tokio::time::sleep(Duration::from_secs(5)).await }).await.unwrap();
        info!(i, "Worker end");
        ()
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    dotenvy::dotenv().ok();
    init_tracing();
       
    let pool = kameo::spawn(ActorPool::new(5, || kameo::spawn(MyWorker)));    
    for i in 0..10 {
        info!(i, pool_capacity=pool.capacity(), "pool tell");
        pool.tell(WorkerMsg(MyWorkerMessage(i))).await?;        
    }

    info!("Waiting for shutdown...");
    tokio::signal::ctrl_c().await?;

    info!("Shutdown");
    Ok(())
}

Gives this output

{"timestamp":"2025-03-25T20:57:45.604370Z","level":"INFO","message":"pool tell","i":0,"pool_capacity":1000}
{"timestamp":"2025-03-25T20:57:45.644044Z","level":"INFO","message":"pool tell","i":1,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.644198Z","level":"INFO","message":"Worker start","i":0}
{"timestamp":"2025-03-25T20:57:45.678987Z","level":"INFO","message":"Worker start","i":1}
{"timestamp":"2025-03-25T20:57:45.679059Z","level":"INFO","message":"pool tell","i":2,"pool_capacity":1000}
{"timestamp":"2025-03-25T20:57:45.684736Z","level":"INFO","message":"pool tell","i":3,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.684799Z","level":"INFO","message":"Worker start","i":2}
{"timestamp":"2025-03-25T20:57:45.686100Z","level":"INFO","message":"pool tell","i":4,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.686213Z","level":"INFO","message":"Worker start","i":3}
{"timestamp":"2025-03-25T20:57:45.687977Z","level":"INFO","message":"pool tell","i":5,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.688045Z","level":"INFO","message":"Worker start","i":4}
{"timestamp":"2025-03-25T20:57:45.691084Z","level":"INFO","message":"pool tell","i":6,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.692562Z","level":"INFO","message":"pool tell","i":7,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.692963Z","level":"INFO","message":"pool tell","i":8,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.695327Z","level":"INFO","message":"pool tell","i":9,"pool_capacity":999}
{"timestamp":"2025-03-25T20:57:45.696201Z","level":"INFO","message":"Waiting for shutdown..."}
{"timestamp":"2025-03-25T20:57:50.695708Z","level":"INFO","message":"Worker end","i":2}
{"timestamp":"2025-03-25T20:57:50.695707Z","level":"INFO","message":"Worker end","i":0}
{"timestamp":"2025-03-25T20:57:50.695745Z","level":"INFO","message":"Worker end","i":4}
{"timestamp":"2025-03-25T20:57:50.695771Z","level":"INFO","message":"Worker end","i":3}
{"timestamp":"2025-03-25T20:57:50.696145Z","level":"INFO","message":"Worker start","i":7}
{"timestamp":"2025-03-25T20:57:50.696639Z","level":"INFO","message":"Worker start","i":5}
{"timestamp":"2025-03-25T20:57:50.697009Z","level":"INFO","message":"Worker start","i":9}
{"timestamp":"2025-03-25T20:57:50.697211Z","level":"INFO","message":"Worker start","i":8}
{"timestamp":"2025-03-25T20:57:50.697539Z","level":"INFO","message":"Worker end","i":1}
{"timestamp":"2025-03-25T20:57:50.698696Z","level":"INFO","message":"Worker start","i":6}
{"timestamp":"2025-03-25T20:57:55.708585Z","level":"INFO","message":"Worker end","i":8}
{"timestamp":"2025-03-25T20:57:55.708612Z","level":"INFO","message":"Worker end","i":6}
{"timestamp":"2025-03-25T20:57:55.708586Z","level":"INFO","message":"Worker end","i":7}
{"timestamp":"2025-03-25T20:57:55.708625Z","level":"INFO","message":"Worker end","i":9}
{"timestamp":"2025-03-25T20:57:55.709077Z","level":"INFO","message":"Worker end","i":5}
{"timestamp":"2025-03-25T20:58:01.976843Z","level":"INFO","message":"Shutdown"}

[tqwewe]

As a guess and API breaking, maybe Actor::new_mailbox() needs an options/capacity parameter that can be optionally passed during the prepare/spawn process.

Something like this could be nice, but I'm not sure the best way to design it from an API point of view.

[tqwewe]

There's no way of configuring it currently unfortunately. I actually would like to redesign the actor pool a bit, since its been giving me some issues recently in regards to scheduling, so I'll need to spend some time improving its usability, maybe even making the actor pool itself not be an actor, but not entirely sure yet.

[dnaut]

Thanks for the that. You could move the mailbox settings from the Actor trait and optionally pass these on the spawn/prepare pathway. But perhaps a bit drastic.

My two cents on the ActorPool... An ActorPool with one async mpmc channel (say flume) with a configurable number of workers (work stealing) should be pretty performant. Where each worker would receive from the mpmc channel (+control, broadcast channel). And then maybe later an ActorPool (1 worker) = Actor?

[tqwewe]

You actually encouraged me to explore making the mailboxes separate from the actor trait, and I've done some changes locally which simplifies a lot around requests and such.

The API I'm exploring would let you provide the mailbox when spawning the actor with kameo::spawn(MyActor, mailbox::unbounded()).

An ActorPool with one async mpmc channel (say flume) with a configurable number of workers (work stealing) should be pretty performant.

Yeah I did explore using the async_channel crate for the actor pool, but a bunch of the methods don't satisfy the methods on the Mailbox trait sadly.

I'll keep thinking on a nicer solution to handle it all, but the biggest blocker is that I can't really effectively know the number of messages in the mailboxes of each worker, which makes scheduling quite difficult. I wonder if a shared mailbox could work.

[tqwewe]

I've made a pretty big PR which lets you specify the mailbox for actors when spawning, which should solve this issue.

#153

This would let you do:

let pool = kameo::spawn(ActorPool::new(...), mailbox::bounded(10))

[dnaut]

Thank you, I'll try it out.