Parallelism issue of dispatch when one actor blocks thread

[svetlyak40wt]

By default shared dispatcher uses 4 threads. If some actor's RECEIVE function takes significant time, and there are many messages in it's queue, then other dispatcher thread will be locked trying to acquire actor's message box lock here. Because of this if we are having yet another actor which also uses shared dispatcher, then mesages to this second actor will not be processed untill all messages to the first actor will be processed.

To demonstrate the issue, let's create a system with two actors:

• answerer processes each message 10 seconds;
• checker returns immediately.

(uiop:define-package #:test-sento-parallelism
  (:use #:cl))
(in-package #:test-sento-parallelism)


(defparameter *system* (sento.actor-system:make-actor-system))

(defparameter *answerer*
  (sento.actor-context:actor-of *system* :name "answerer"
                                         :receive
                                         (lambda (msg)
                                           (log:debug "Thinking on ~A" msg)
                                           (sleep 10)
                                           (log:debug "Replying to ~A" msg)
                                           (let ((output (format nil "Hello ~a" msg)))
                                             (sento.actor:reply output)))))


(defparameter *checker*
  (sento.actor-context:actor-of *system* :name "checker"
                                         :receive
                                         (lambda (msg)
                                           (log:debug "Checking ~A" msg)
                                           (let ((output (format nil "Checked ~a" msg)))
                                             (sento.actor:reply output)))))


(defun run-test (&key (num-messages 4))
  (loop for i from 1 upto num-messages
        do (sento.actor:ask *answerer* i)
        finally (sento.actor:ask *checker* 100500)))

Results of this code:

TEST-SENTO-PARALLELISM> (run-test)
<DEBUG> [2025-02-01T11:18:48.985165Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 1
NIL
<DEBUG> [2025-02-01T11:18:58.985431Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 1
<DEBUG> [2025-02-01T11:18:58.985617Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 2
<DEBUG> [2025-02-01T11:19:08.985794Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 2
<DEBUG> [2025-02-01T11:19:08.985937Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Checking 100500
<DEBUG> [2025-02-01T11:19:08.986015Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 3
<DEBUG> [2025-02-01T11:19:18.986221Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 3
<DEBUG> [2025-02-01T11:19:18.986371Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 4
<DEBUG> [2025-02-01T11:19:28.986530Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 4

Here Checking 100500 message appears only after the first message to answered, 10 seconds after the test was runned.

On another run it starts 20 seconds after:

TEST-SENTO-PARALLELISM> (run-test)
NIL
<DEBUG> [2025-02-01T11:20:12.193907Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 1
<DEBUG> [2025-02-01T11:20:22.194082Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 1
<DEBUG> [2025-02-01T11:20:22.194221Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 2
<DEBUG> [2025-02-01T11:20:32.194417Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 2
<DEBUG> [2025-02-01T11:20:32.194572Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Checking 100500
<DEBUG> [2025-02-01T11:20:32.194647Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 3
<DEBUG> [2025-02-01T11:20:42.194823Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 3
<DEBUG> [2025-02-01T11:20:42.195007Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 4
<DEBUG> [2025-02-01T11:20:52.195192Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 4

If we will send more messages to the first actor, then the message to the second actor will be processed with more delay:

TEST-SENTO-PARALLELISM> (run-test :num-messages 20)
NIL
<DEBUG> [2025-02-01T11:24:15.766732Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 1
<DEBUG> [2025-02-01T11:24:25.766886Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 1
<DEBUG> [2025-02-01T11:24:25.767037Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 2
<DEBUG> [2025-02-01T11:24:35.767161Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 2
<DEBUG> [2025-02-01T11:24:35.767294Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 3
<DEBUG> [2025-02-01T11:24:45.767407Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 3
...
<DEBUG> [2025-02-01T11:27:25.771623Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 19
<DEBUG> [2025-02-01T11:27:25.771745Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Thinking on 20
<DEBUG> [2025-02-01T11:27:35.771863Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Replying to 20
<DEBUG> [2025-02-01T11:27:35.772013Z] test-sento-parallelism test-sento-parallelism.lisp (top level form) Checking 100500

Message to the second worker was processed 3 minutes from test start. And all this time three threads from four were just waiting for a lock on the first actor's message box.

Expected behaviour

Only one thread of the shared dispatcher is used to process messages of a single message-box/dp. Other threads are able to process messages of other actors immediately.

In the situation modelled in the test code above, I expect that one thread will start processing the first message to answerer actor, second thread will start processing a message to checker actor and two more threads will be available for processing of other messages for any other actors if they will be created.

[mdbergmann]

What you describe may be true. More likely when using a 'random' dispatcher strategy, because indeed message to second actor may be processed by first dispatcher which is still blocked with processing a message to first actor if random provides the same router index.
Using 'round-robin' strategy can mitigate this a bit.

However, unless we have something like virtual threads (like in Java 21 or Erlang) where each actor can be bound to one thread we have to live with the fact that different threads are processing message on different actors.
One of the pattern that was carved out is that you should never block message processing. If you have work that takes longer you should offload that work to custom dispatchers, or spawn another BT thread, or use a custom thread-pool library.

See also a section on this in readme: https://github.com/mdbergmann/cl-gserver#long-running-and-asynchronous-operations-in-receive

Or also in Akka documentation: https://doc.akka.io/libraries/akka-core/current/typed/dispatchers.html#blocking-needs-careful-management

[mdbergmann]

offload that work to custom dispatchers

In Chipi project I use custom dispatchers for various things using 'tasks' api to offload stuff, i.e. reading from serial device or some such.

[svetlyak40wt]

More likely when using a 'random' dispatcher strategy, because indeed message to second actor may be processed by first dispatcher which is still blocked with processing a message to first actor if random provides the same router index.
Using 'round-robin' strategy can mitigate this a bit.

Seems you didn't understand the situation I've describe above. I'll try to demonstrate visually. Imagine we've received the code from the initial posts with num-messages = 40. Now lets visualize the state of dispatch actors queues:

dispatch(SHARED)-worker-1: A A A A A A A A A C
dispatch(SHARED)-worker-2: A A A A A A A A A 
dispatch(SHARED)-worker-3: A A A A A A A A A
dispatch(SHARED)-worker-4: A A A A A A A A A

Here A is a message to the dispatch worker saying it should call dispatcher-exec-fun on actor "answerer", and C is that it should do the same for actor "checker".

Now, one of these worker threads is actually executing code of "answerer", but other three workers can't, because the are waiting for lock.

Because of this, code of actor "checker" will be executed only when worker-1 will execute "answerer" 10 times.

This happens because when we submit message to message-box/dp the two messages added to two queue:

• real message is added to the message-box/dp queue of the actor
• and the request for calling the dispatcher-exec-fun is added to dispatcher's message-box/bt.

Because of this, requests to the "answerer" fill queues of the "dispatch" actors preventing message to "checker" processing.

Random dispatching will not help here, just queues will be filled more randomly.

Possible solution

Add a flag processing on message-box/dp objects. It will change how messages are submitted and processed.

Submitting

When processing = NIL the submit will add message to the box's queue and ask dispatcher to call dispatcher-exec-fun. Then processing will be changed to T.

When processing = T then submit will just add a message to the box's queue.

Processing

When dispatcher-exec-fun finishes to process the message, it should see if there is more items in the actor's queue.

If there is one more item, then dispatcher-exec-fun should schedule itself into the queue of the dispatcher to process this new item. (Or probably to process it right away, however in this case some high-loaded actor might occupy the thread leaving other actors no chances to execute on this thread).

If there is no more items, then dispatcher-exec-fun should set processing to NIL.

Final note

Of cause all work with the processing flag should be guarded by a lock.

If you are ok with such approach, I can implement it.

[mdbergmann]

As far as I can tell this works as designed.

I have an example here where message to answerer and checker are processed in parallel:

USER> (run-test)
 <INFO> [00:11:06] sento.user () - Thinking on 1
 <INFO> [00:11:06] sento.user () - Checking 100500
 <INFO> [00:11:16] sento.user () - Replying to 1
 <INFO> [00:11:16] sento.user () - Thinking on 2
 <INFO> [00:11:26] sento.user () - Replying to 2
 <INFO> [00:11:26] sento.user () - Thinking on 3

It highly depends on which dispatcher are you hitting and whether it has a message that blocks processing.
The lock you mentioned is an actor mailbox dedicated lock and makes sure that messages that are sent to an actor are processed in the order they were enqueued in. Otherwise you can't guarantee this.

In Akka, the dispatcher actually has one queue where all worker threads work on. Which means that if you have one message that blocks it will block everything. In Sento each dispatcher worker has its own queue, because workers are actually also actors.

But yeah, in case I really missed something, this works as it should. The key is: don't block message processing.

[mdbergmann]

Having said this, I've planed on implementing a 'balanced' router strategy that is able to peek into the queue and choose a worker that is idle, or has the lowest number of queued items.

[svetlyak40wt]

I have an example here where message to answerer and checker are processed in parallel

Please, try (run-test :num-messages 20).

I understand that current dispatcher is designed this way. But I believe that with such design it will have poor performance even if actors are processing their messages very fast. Actually, instead of believing, I have to extend our benchmark with an example where a multiple actors are receiving the load and using the shared dispatcher.

[mdbergmann]

If I see it correctly then your proposal will still maintain the execution order of the actors mailbox/dp. This might make things faster.
My concern is that since a dispatcher worker will then continue processing messages of a particular actor (so basically this binds the worker to a particular actor) that when you have many actors and only few dispatcher worker but things are busy that processing of actors is not balanced. Some actors might see their mailboxes processed completely while other have to wait until there is a free worker that has finished processing an actors mailbox, which may or may not happen if messages continue to arrive.

[mdbergmann]

So I see the current implementation of dispatching treats each actor equally since its messages are processed roughly in the order they were submitted, across all actors.
You're suggestion focuses more on throughput but does not treat actors equally.
I see those as alternative implementations. Maybe one or the other can be chosen based on setting or so depending on what suits better for different use-cases.

[svetlyak40wt]

You're suggestion focuses more on throughput but does not treat actors equally.

Correct!

Actually in my proposal there are two possible implementations:

If there is one more item, then dispatcher-exec-fun should schedule itself into the queue of the dispatcher to process this new item. (Or probably to process it right away, however in this case some high-loaded actor might occupy the thread leaving other actors no chances to execute on this thread).

First possible solution, is to process one message after another right in the dispatcher-exec-fun until actor's queue become empty. This way we really play unfair and make preference to one actor over anothers.

Second possible solution is to throw new dispatcher-exec-fun message into the dispatcher only after the one message to the actor was processed. This new go through a router again and can be processed by another thread while giving a chance to messages to other agents be processed.

First solution is unfair but will have lowest overhead.

[svetlyak40wt]

How to create a custom type of shared dispatcher? I see implementation of message-box/dp is tied to the dispatcher implementation.

Should I implement both a new dispatcher type and a new message-box type to experiment with my proposal?

Is there a way to tell actor system about my new type of a dispatcher?

[mdbergmann]

Actually in my proposal there are two possible implementations:

If there is one more item, then dispatcher-exec-fun should schedule itself into the queue of the dispatcher to process this new item. (Or probably to process it right away, however in this case some high-loaded actor might occupy the thread leaving other actors no chances to execute on this thread).

First possible solution, is to process one message after another right in the dispatcher-exec-fun until actor's queue become empty. This way we really play unfair and make preference to one actor over anothers.

Second possible solution is to throw new dispatcher-exec-fun message into the dispatcher only after the one message to the actor was processed. This new go through a router again and can be processed by another thread while giving a chance to messages to other agents be processed.

First solution is unfair but will have lowest overhead.

I think second solution will be similar as current implementation, or at least offer only slightly different behavior.
Maybe focus on first solution.

[mdbergmann]

How to create a custom type of shared dispatcher? I see implementation of message-box/dp is tied to the dispatcher implementation.

Should I implement both a new dispatcher type and a new message-box type to experiment with my proposal?

Is there a way to tell actor system about my new type of a dispatcher?

I'll come back on this, need to think about it for a bit.

[mdbergmann]

How to create a custom type of shared dispatcher? I see implementation of message-box/dp is tied to the dispatcher implementation.
Should I implement both a new dispatcher type and a new message-box type to experiment with my proposal?
Is there a way to tell actor system about my new type of a dispatcher?

I'll come back on this, need to think about it for a bit.

Actually the dispatcher implementation doesn't need to change. What we need is a different message-box behavior.

So I would propose following:

• either add setting to message-box/dp constructor representing this setting
• or make new message-box/dp type where the new setting will be applied
• add a new global setting (actor-system-api.lisp), e.g.: :message-box-options (:behavior :throughput or :balanced) where :balanced is default. This setting is queried in actor-context.lisp %message-box-for-dispatcher-id where the message-box is created.
• since it should be possible to also allow this setting to be applied individually actor-of interface should allow the above as a custom key in rest
• the individual setting takes precedence if exists, otherwise the global setting is applied on each message-box/dp creation

[svetlyak40wt]

What if we allow to specify a message box class name instead of :pinned or :shared? This will allow anybody to extend Sento without modifying its core.

[mdbergmann]

Well, I think a framework should stay in control of how things work and how things are internally wired and allow the user control only through an abstracted set of options.

The 'dispatcher-id' as this parameter you mentioned actually specifies the type of dispatcher, not the message-box. It's actually possible to specify custom dispatchers via configuration when creating the actor-system. And distatcher key in the config is the thing the user can use here. :pinned doesn't exist in config, because it's hard-wired. But :shared is part of config. Any additional specified dispatcher can have arbitrary names (in form of global symbols) like :foobar or so.

But actually, that brings me to a better idea.
The message-box type used for the dispatcher should be configurable there as well, because those two belong together.

So if we take the current config:

'(:dispatchers
    (:shared (:workers 4 :strategy :random)))

We should allow to specify the message-box type there. So the user can override the type of message box for :shared like so:

'(:dispatchers
    (:shared (:workers 4 :strategy :random :mbox-type my-mbox-type)))

Where where mesgb:message-box/dp is the default.

Or create his own custom dispatcher (it's additional, because config is merged with default config):

'(:dispatchers
    (:my-disp (:workers 4 :strategy :round-robin :mbox-type my-mbox-type)))

The implementation of this will likely still have to happen in %message-box-for-dispatcher-id where the message-box is created.