[FEATURE] - more flexible concurrency

[blueforesticarus]

Motivation

Lets say my actor represents a client to an API. Among the things this actor does is request data on Items. The requests are batched using an internal queue and a separate task, but we want the actor to be able for the individual items. The initial handling of the request does need &mut, (to add items to the queue), but waiting on the result does not need (and cannot hold up) the &mut Self.

ie. it is a class of problem which has a sequential mutable part, and a concurrent part
(and really it might need the &mut after the current part as well.)

Proposed Solution

None. Want to open this discussion.

Alternatives Considered

Returning a one-shot channel.

[tqwewe]

If I understand correctly, could this be solved by using only tell requests between the actors instead of ask? That way, your actor which needs to mutate the state can handle the initial request, and then tell a message to some other part of your app (actor pool, tokio task, etc), and then when that other part finished processing, it sends a different tell message to the original actor which can finalize whatever it needs to using its &mut self access.

Would this approach possibly solve your issue? Its the more "pure" way of messaging between actors and can be pretty powerful.

[blueforesticarus]

Yes, that is an option.

Here is my counterpoint: there is a reason kameo supports ask requests.

It is common and useful to have that bidirectional request-style communication. In my mind the greatest benefit is that it makes actors kinda like async objects (in the java/oop sense), where the message passing looks like functions. I think this is an easier model to think-in than channels (and also the reason why alot of net protocols are also request response, rather than bidirectional pubsub); it's hard to think about control flow when the response is disconnected from the request ("ie. the data/event goes here, then here, then eventually works its way back around to the caller, hopefully...").

Here is my stab at a current way to do it.

1. Create a one-shot channel in the handler, pass the Tx to the inner handler (spawn, queue, another actor, whatever), return the Rx to the caller (asker).
2. The inner task does whatever, then send a message (tell) back to the actor with the Tx
3. The actor updates its state and then sends the result through the Tx
4. The caller asks the actor and then awaits the returned channel. (A side benefit of this method is the caller can decide whether to await the completion or do something in the meantime).

In the course of writing this, this feels more solid. I'll try it out and post some code.

A side note: one of the things making my head hurt here is the difference between fast and slow async. Ie. putting something in an async queue is fast, network requests are slow. When designing things, one kind has to know how long they expect any given .await to block for and when and why.

[tqwewe]

Kameo does let you get control of the reply oneshot channel on the Context struct using .reply_sender(). You can store this and reply at any later time which might give you some of the flexibility you're looking for

[blueforesticarus]

A random thought, but async tasks could probably be written to take a RefCell (where Self is the actor) and then as long as borrows are not held across an await point (which I believe the borrow checker would catch). It ought to be possible to write long running tasks as regular async functions, without having to spawn tasks, or deal with the shared state spawning creates. (ie. no members of the Actor struct need Arc/Mutex)

( Then the question is what polls the tasks, and how would tasks be defined and queued )

[blueforesticarus]

I would say it is more like the opposite.
The idea here is to allow simultaneous tasks with mutable access to the actor (via the RefCell)
It's important they don't starve out the ordinary messages.

[tqwewe]

Hmm that doesn't really align with the actor model. I'm not quite sure what a good example use case for this would be

[hanekoo]

The actix::prelude::AsyncContext provides asynchronous support, allowing for spawn tasks, wait, and cancel_future.

• There’s a scenario where the attach_stream API is either unavailable or insufficient for my needs. For example, if there’s a stream that contains metadata and subsequent normal data , but start_value must be the metadata, then start_value: T should become start_value: FnOnce(&stream<T>) -> T.Of course, I could extract the metadata before using attach_stream.

• Or consider more complex stream processing logic. For instance, in FFmpeg, both audio and video streams are represented as Packet. However, Message<Packet> can only be declared once and cannot distinguish between video packet and audio packet. While wrapping them as VideoPacketStream and AudioPacketStream can solve this, it might not always be ideal.

Based on the above practical scenarios, I would start an asynchronous task in the actor, with the task being strongly associated with the actor, and I also have some monitoring requirements.

[tqwewe]

It sounds like there's a couple solutions I can think of, mainly for your second point:

1. Attach two separate streams, one for the video stream and another for the audio stream.
2. Create an enum, and use futures StreamExt::zip and StreamExt::map to join them into a single stream which can then be attached to the actor.

I'm not sure I follow your first point exactly. The start_value and finish_value's are just arbitrary data that can be added to the attached stream. Actix historically is mostly its own runtime, and thus provides many features such as spawn, etc. But since kameo is based entirely on tokio, you can utilize a lot of tokios features along side kameo, such as spawning arbitrary tasks, etc.

[hanekoo]

Thank you for your suggestion! I’ll give it a try later. (By the way, I looked back at my code earlier, and noticed that PubSub<M> can also be implemented using an enum.)

[blueforesticarus]

Another sticking point I bumped into is that it would be nice if there was a way to spawn an Actor defered, where the Actor is returned by a future, but you get the ActorRef right away. This would help to avoid having to Option wrap Actor fields which are initialized in on_startup.

[blueforesticarus]

I can't tell for sure, but reading the code, this does not seem like it will actually defer creating the Actor. Ie. by the time spawn_with().await returns an ActorRef, the factory has finished.

[tqwewe]

I think I should make prepare_with public. I'm not sure why I've chosen to make this private initially.

kameo/src/actor/spawn.rs

Lines 215 to 222 in fbc7826

	async fn prepare_with<A, F, Fu>(f: F) -> PreparedActor<A>
	where
	A: Actor,
	F: FnOnce(ActorRef<A>) -> Fu,
	Fu: Future<Output = A>,
	{
	PreparedActor::new_with(f).await
	}

But it would give you a prepared actor, which can be initialized with the actor ref, and only ran later when you need.

[blueforesticarus]

https://github.com/tqwewe/kameo/blob/fbc7826a38ea579aa2fae253bb86cfc8b28e0160/src/actor/spawn.rs#L264C1-L273C11

Is there a reason we can't get these lines without let actor = f(actor_ref.clone()).await; on the next line.

Instead I want the order of operations to be.

1. ActorRef is created.
2. f(actor_ref.clone()).await get's tokio::spawn'd to the background
3. spawn_defer(f) immediately returns the ActorRef

[tqwewe]

The prepare_with could be used like so:

// ActorRef is created.
let prepared_actor = prepare_with(|actor_ref| async move {
     MyActor { actor_ref }
});

// 2. Spawn tokio task
prepared_actor.spawn();

Does this not match what you're describing? You have access to the actor ref, and your actor is only spawned when you call .spawn().

[blueforesticarus]

// prepare_with is **async**
let  fut = prepare_with(|actor_ref| async move {
     // <imagine slow async task HERE >
     MyActor { actor_ref }
});

let prepared_actor : PreparedActor = fut.await; // this will be slow

// HERE THE CLOSURE HAS COMPLETED RUNNING

// This is only spawning the task's event loop. We were still forced to wait for the closure.
prepared_actor.spawn();

[blueforesticarus]

Thoughts on latching messages for state machine.

• Using messages as the crank in a state-machine (with &mut access)
• With other parts of the code (who know when something changes) sending the signal to wake up the crank
• but this way we avoid filling the queue (the crank resets the trigger when it runs.)

EDIT:
This works. https://github.com/blueforesticarus/goontunes2/blob/4ad722909edc2a7146df5d958b1a6231db55e610/src/service/spotify/mod.rs#L750-L790

[blueforesticarus]

@tqwewe ^ my LatchedTask as mentioned in the other comment.

For some more motivation on what this is useful for. My Actor maintains a pool of concurrent connections, and an internal queues (separate from mailbox) of requests. Whenever a new request comes in, or a previous one finishes, Task must rerun to decide what new request to dispatch.

It is important this is a Message and not just a function called from other Messages because I want it to run after all other messages process. Otherwise, I'd end up dispatching a request for a single Track, even if there are 100 in the Mailbox. However, I don't want to fill up my actor's mailbox with 100 Task requests, and run Task 100 times, when it only needs to be run once.

Some thoughts:

1. It would be nice to be able to defer processing even more. Instead of putting it on the end of the mailbox, have it only run if the mailbox is empty. Either an IdleTask or actor.trigger_idle(Task)
2. For the above to be robust there probably needs to be some way to give it a timeout after which the task is prioritized over the Messages actor.trigger_idle_with_timeout(). Maybe also a max frequency for tasks.
3. Likewise, a way to tell the system run this task immediately would be good. Note that it should still clear any pending run of the task.

[blueforesticarus]

Do you have any thoughts on making a PubSub type which works for all message types which implement a trait (probably should be called Topic).

I am trying to make something now which lazily spins up a global PubSub when something tries to publish or subscribe to M. I think this makes ergonomic sense for UseCases where things are very decoupled. Ex) I want to put publish hooks into my discord client for when messages are received, but only in some configurations will anyone be listening.

I think there is an argument that the pubsub broker should be explicitly instantiated, and passed to the actor, but I definitely think ad-hoc muxing over the Type of message is a good idea (similar to how we mux over actor messages).

[blueforesticarus]

https://github.com/blueforesticarus/goontunes2/blob/068ad6d77c81551d83d29a2645043ed56b642c82/src/utils/pubsub.rs

Current attempt at implementing something.

• I made the message require a marker trait Topic, this was to avoid accidentally subscribing to a slightly different struct and missing messages without failing. (ie. sub to MessageBundle instead of Vec)
• At first I was going to do this with mutex's, but spawning the PubSub turned out to be easier. The typemap still requires an async RwLock, and I feel there is likely a better way.
• I am trying to figure out if there is something less adhoc (without a typemap) which is still ergonomic for flexible usecases. I want to avoid refactor burden when I realize "hey, there should be an event hook here".

[tqwewe]

Yeah I think something like this could definitely be a nice feature, and as you said it doesn't stray from kameos internal implementation of downcasting messages internally already.

I think the way I'd approach this would pretty much exactly as you did in your linked code. I haven't heard of typemap before, but looking at the source code of it, its as I'd implement this, with the HashMap<TypeId, Box<dyn Any>> approach.

Initially I was thinking the actors would be stored as the values of the typemap, but then it can't really be downcasted into a solid type, but what you've done with making the values be PubSub actors themselves is awesome.

I am trying to figure out if there is something less adhoc (without a typemap) which is still ergonomic for flexible usecases.

Unfortunately the only other approach I can think of if not downcasting is using an enum, which can be a little restricting. I'm quite happy with how boxing of messages and downcasting has worked with kameo so far. If you can find a better solution for your generic pubsub implementation, I'd love to see it, but I'm honestly happy to include your implementation into Kameo as is. Maybe as a Topic actor as you suggested, or improved PubSub actor.

Alternatively, there could be a new crate for things like this, such as kameo_actors which I've published for now as an empty crate. This would keep the core of kameo more simple, but allow for additional actors like this to be added.

[blueforesticarus]

Re: kameo_actor crate.

I'll point out one con to this. It might be advantageous to keep them in the same crate, where they have access to pub(crate) internals. More tight coupling might make certain ergonomic / performance improvements possible down the line.

One example is my TriggerTask, which might be nicer to work with like this: impl Task<MyTask> for MyActor + myactor.trigger<MyTask>() . There are some under the hood improvements that could be made as well, perhaps bypassing the MailBox to avoid issues on a full mailbox.

Actually, in general, what are you supposed to do about a Bounded actor messaging itself. Isn't that always a risk of deadlock?

[tqwewe]

Ideally I'd like kameo's API flexible enough such that there wouldn't be any benefits to having it all within the core kameo crate. If there's something that provides improved performance etc, then the API for that kind of thing ideally would be public.

Actually, in general, what are you supposed to do about a Bounded actor messaging itself. Isn't that always a risk of deadlock?

Deadlocks can happen when an actor messages itself either:

• If sending an ask request
• If sending a tell request on a bounded mailbox

It's only safe for the actor to message itself using tell requests using: send_sync, try_send, try_send_sync, try_blocking_send, since these won't wait for mailbox capacity.

[blueforesticarus]

I suppose if the actor is already processing a message there is a good chance there's at least one free slot, so try_send should only fail rarely.

Actually, come to think of it probably better to just use tell with a timeout.