[Idea] C++20 coroutines inside agent's event_handler? The first take

[eao197]

These are my first thoughts about supporting C++20 coroutines inside an agent's event_handler.

I'm not going to try to fantasize about the purposes that coroutines in event_handler can be used for. It doesn't matter for now. Just assume that we want to have a possibility to write something like that:

class my_agent : public so_5::agent_t
{
  void so_define_agent() override
  {
    so_subscribe_self().event(
      [](mhood_t<some_msg> cmd) -> so_5::resumable_t {
        ...
        co_await some_expression();
        ...
      });
  }
};

The first key moment is a special return value type -- so_5::resumable_t (the name can be changed if something more appropriate will be found).

The next key moment is a special handling of event_handlers that return so_5::resumable_t. When an event_handler uses so_5::resumable_t as the return type then SO-5 will understand that event_handler is a coroutine and will do some special transformation. Let's suppose that we have something like that:

class my_agent : public so_5::agent_t
{
  so_5::resumable_t evt_some_handler(mhood_t<some_msg> cmd);
  ...
  void so_define_agent() override
  {
    so_subscribe_self().event(&my_agent::evt_some_handler);
  }
};

This code will be "transformed" into something like this:

class my_agent : public so_5::agent_t
{
  // The special hidden signal to be used for resumption.
  struct _msg_resume_current_coroutine final : public so_5::signal_t {};

  // The resumable object returned by a coroutine.
  so_5::resumable_t _m_current_resumable{ nullptr };

  // The special and hidden handler that will be called when
  // _msg_resume_current_coroutine arrives.
  void _evt_resume(mhood_t<_msg_resume_current_coroutine> cmd) {
    _m_current_resumable.resume();
    if(_m_current_resumable.done()) {
      // Coroutine finished, we have to return to normal work.
      _m_current_resumable.destroy();
      so_drop_subscription(
        so_direct_mbox(),
        so_current_state(),
        &my_agent::_evt_resume);

      _so_reactivate_normal_processing();
    }
  }

  // The special and hidden handler that will be subscribed to
  // some_msg instead of evt_some_handler.
  void _evt_auto_generated_for_some_msg(mhood_t<some_msg> cmd) {
      // We need a subscription for the hidden signal.
      so_subscribe_self().event(&my_agent::_evt_resume);

      // We need to pause normal delivery of pending demands
      // to this agent.
      _so_deactivate_normal_processing();

      // Get a resumable object for the coroutine.
      _m_current_resumable = this->evt_some_handler(std::move(cmd));

      // Just return the control back to the dispatcher.
  }

  // Coroutine to be used as an event_handler.
  so_5::resumable_t evt_some_handler(mhood_t<some_msg> cmd);

  void so_define_agent() override
  {
    so_subscribe_self().event(
      &my_agent::_evt_auto_generated_for_some_msg);
  }
}

The next key moment is the presence of two special methods _so_deactivate_normal_processing and _so_reactivate_normal_processing.

In the normal processing mode there is an event_queue with pending demands (at the moment such a queue belongs to dispatchers, agents do not own these queues).

A demand in the queue holds information about one incoming message and how it should be processed (as normal message or as evt_start/evt_finish special signals). A dispatcher takes the first pending demand, calls an event_handler and then goes to the next pending demand if any.

It could be a case where the demand queue for my_agent holds several pending demands:

some_msg
another_msg
and_yet_another_msg

The dispatcher gets the first one (for some_msg) and calls the event_handler for it. It will be the special _evt_auto_generated_for_some_msg handler that just stores the resumable object into _m_current_resumable and returns.

The dispatcher sees that event_handled completed and takes the next demand from the queue.

But this normal processing is not appropriate when the event_handler for some_msg is a coroutine. In this case the normal processing has to be paused and the agent should react to _msg_resume_current_coroutine signals only.

The new hypothetical methods _so_deactivate_normal_processing and _so_reactivate_normal_processing are intended for switching from normal processing mode to the special one, and for returning back to the normal mode.

When the agent switches to this special mode it will look like switching to a different and empty demand queue, the old demand queue will be stored somewhere. This new queue will be used for _msg_resume_current_coroutine signals only. All other messages will be redirected to the old queue (they will be stored to the old queue, but the processing will be delayed).

When _msg_resume_current_coroutine arrives it will be extracted and delivered to _evt_auto_generated_for_some_msg handler the usual way. The SO-5 will see such a signal just as usual without a need for a special handling.

And the next key moment is to use _msg_resume_current_coroutine signal for resumption of a suspended coroutine. The promise and awaitables for SO-5 have to be written in such a way to send _msg_resume_current_coroutine when coroutine has to be resumed.

---

This is a very basic idea with a lot of white spots.

The main of them is how to implement this changing of demand queues for an agent. It seems to me that the work on this old idea has to be resumed. If an agent has its custom event_queue then it could be changed to any other queue when it's appropriate and necessary.

Another white spot is support coroutine-based thread safe event handlers on adv_thread_pool dispatcher. The example shown above requires changing of the agent internal members for processing coroutine, but those changes are prohibited in thread safe event handlers. I hope the solution will be found. Even if this won't happen in the near future the first implementation for support of coroutines could be made for non-thread safe event handlers. With the hope that the solution will be found with time.

---

I think that one way to add support for C++20 coroutines into SO-5 is to use a special mark that enables coroutine-based event_handlers for an agent. Something like that:

class my_agent : public so_5::agent_t
{
  my_agent(context_t ctx)
    : so_5::agent_t{ctx + so_5::enable_courotines}
  {}
};

An attempt to subscribe an event_handler that returns so_5::resumable_t without addition of so_5::enable_coroutines to the agent context will lead to an exception during subscription.

I also think that SO-5.8 will still require C++17 as the minimum, the support for coroutines will be available only if SO-5 is being compiled in C++20/23 mode.

[ilpropheta]

Hey @eao197, thanks for sharing! That's an interesting topic. I would be interested in the (experimental) code, if you have some that can be disclosed.

I think the main point worth highlighting is the purpose of coroutines in agents (I know you were a bit reluctant at the beginning of the discussion): they (can) enable better concurrency without breaking the actor model.

Coroutines give dispatchers more flexibility in thread usage. For example, if two agents share a one_thread dispatcher, they run on the same context thread. If one handler can suspend, the dispatcher can resume another agent, then return to the first when ready, improving responsiveness. The same principle applies to thread pools, where suspension points help avoid long tasks monopolizing threads.

However, for agents bound to dedicated threads, the benefit is less evident, but coroutines can still simplify asynchronous workflows and integrate with modern async APIs.

What do you think?

[eao197]

I would be interested in the (experimental) code, if you have some that can be disclosed.

No, there is nothing to share yet. I hope to do some first experiments before the New Year, but nothing working will be available until the end of January 2026 (at best).

If one handler can suspend, the dispatcher can resume another agent, then return to the first when ready, improving responsiveness.

Yes, this is what some users asked us. Especially in relation to synchronous interaction between agents.

However, for agents bound to dedicated threads, the benefit is less evident, but coroutines can still simplify asynchronous workflows and integrate with modern async APIs.

An interesting topic is integration with Asio on asio-based dispatchers from so5extra.
It also can open new possibilities for integrating SO-5 into other frameworks.

[eao197]

Another white spot is support coroutine-based thread safe event handlers on adv_thread_pool dispatcher.

Maybe this case has a natural and simple solution: when an agent uses own custom event queue, then it pushes a special execution_demand into the dispatcher queue. This execution_demand has to be treated as not-thread-safe one, because it will modify internal agent's state (extraction of the actual pending message from the custom queue).

It means that agents with custom event queues simply can't have thread-safe event handlers.

"Can't" means that an agent can make a subscription with the thread_safe flag, but this flag will be ignored and all event handlers will be treated as if they are not thread-safe.

[eao197]

If a coroutine-like event-handler like this:

so_5::resumable_t evt_sample(const msg_sample & msg) {
  ...
  co_await some_action(msg.field);
  ...
}

will be transformed to something like that:

void _initial_point(const msg_sample & msg) {
   // We need a subscription for the hidden signal.
   so_subscribe_self().event(&my_agent::_completion_point);

   // We need to pause normal delivery of pending demands
   // to this agent.
   _so_deactivate_normal_processing();

   // Get a resumable object for the coroutine.
   _m_current_resumable = this->evt_sample(msg);

   // Just return the control back to the dispatcher.
}

Then it's possible to get a dangling reference to msg. It's because SO-5 sees only _initial_point as the event-handler for msg_sample instance and prolongs livetime of msg_sample instance for duration of initial_point call only. After the initial_point returns the control back to SO-5 the msg_sample instance could be destroyed.

It means that the lifetime of an incident has to be prolonged until completion of the coroutine. So the message_ref to the incident has to be stored along with a resumable object.

Maybe something like that:

namespace so_5
{

struct current_coro_info_t {
  message_ref_t m_incident;
  resumable_t m_resumable;
  ...
};

} /* namespace so_5 */

class my_agent : public so_5::agent_t {
  // Information about the current active coroutine.
  so_5::current_coro_info_t _m_current_coro_info;
  ...
  void _initial_point(const so_5::message_ref_t & incident) {
    // We need a subscription for the hidden signal.
    so_subscribe_self().event(&my_agent::_completion_point);

    // We need to pause normal delivery of pending demands
    // to this agent.
    _so_deactivate_normal_processing();

    // Get a resumable object for the coroutine.
    _m_current_coro_info = so_5::current_coro_info_t{
      .m_incident = incident,
      .m_resumable = this->evt_sample(extract_reference_from(incident))
    };

    // Just return the control back to the dispatcher.
  }

  void _completion_point(mhood_t<_msg_resume_current_coro>) {
    _m_current_coro_info.m_resumable.resume();
    if(_m_current_coro_info.m_resumable.done()) {
      _m_current_coro_info.destroy();
      so_drop_subscription(
        so_direct_mbox(),
        so_current_state(),
        &my_agent::_completion_point);

      _so_reactivate_normal_processing();
    }
  }
  ...
} 

Because _evt_autogenerated_initial_point receives the incident as so_5::message_ref_t it has to be subscribed by some new subscription method because the current scheme doesn't support this event-handler format. Or there has to be some additional actions inside the initil_point to get a so_5::message_ref_t from a mhood_t or mutable_mhood_t object.

[eao197]

One of the initial white spots was delivery of msg_resume_current_coroutine signal. When and how will this signal be sent?

If I understand C++20 coroutines correctly, it's a task of the awaiter object to wake up a coroutine that sleeps on co_await. For example, let's suppose that there is a library A that performs asynchronous network operations and supports coroutines. I could write something like:

A::task_type
some_class::initiate_connect()
{
  auto client_socket = co_await A::connect_to(remote_ip, connect_timeout);
  ... // result processing.
}

The initiate_connect will be a coroutine. But it uses an awaiter object that is provided by library A. And the coroutine will be awakened somewhere from A internals by the following mechanism:

• library A provides an implementation of the Awaiter concept. Let's suppose it is A::awaiter class;
• the A::awaiter class has an await_suspend method. This method will be called inside my initiate_connect coroutine;
• the A::awaiter::await_suspend method receives an instance of std::coroutine_handle as the parameter. It's the handle of my initiate_connect coroutine;
• this handle will be stored somewhere in A internals;
• when A detects the result of the operation it finds the stored handle and calls resume method for the handle;
• my coroutine will be awakened.

This rather simple picture becomes more complex when we're speaking about event-handlers of SObjectizer's agents. Those event-handlers must be resumed on the appropriate worker thread.

The A library doesn't know about SObjectizer's threads. In the best case the A will just call resume for the handle on my coroutine and this call should lead to sending of msg_resume_current_coroutine signal.

But how to do that if A::awaiter::await_suspend will receive the ordinary std::couroutine_handle that can't be specialized for SObjectizer's promise type?

There is an idea how it could be done.

DISCLAIMER: I don't know yet if it is implementable. Just write it here as a memo because I don't know when I will get time to try to make an experiment.

The first ingredient. A helper coroutine has to be created in the agent's constructor:

enum class awakener_action_t { resume_next, quit };

so_5::coro::awakener_task_t
awakener()
{
  bool need_quit = false;
  while( !need_quit )
  {
    const auto action = co_await query_next_action();
    switch( action )
    {
    case awakener_action_t::resume_next:
      ... // sending next msg_resume_current_coroutine.
    break;

    case awakener_action_t::quit:
      need_quit = true;
    break;
    }
  }
}

The result so_5::coro::awakener_task_t instance has to be stored inside the agent (it will be used later).

The next ingredient is a special wrapper for an arbitrary Awaiter:

template< typename Awaiter >
class wrapped_awaiter_t
{
  so_5::coro::awakener_task_t m_awakener;
  Awaiter m_original;

public:
  template< typename SomeAwaiterType >
  wrapped_awaiter_t(
    so_5::coro::awakener_task_t awakener,
    SomeAwaiterType && original )
    : m_awakener{ std::move(awakener) }
    , m_original{ std::forward<SomeAwaiterType>(original) }
  {}

  [[nodiscard]]
  bool
  await_ready() noexcept(noexcept(m_original.await_ready()))
  {
    return m_original.await_ready();
  }

  auto
  await_suspend(std::coroutine_handle<so_5::coro::promise_type_t>)
  {
    return m_original.await_suspend(m_awakener);
  }

  auto
  await_ready()
  {
    return m_original.await_ready();
  }
};

The trick is to replace an instance of the std::coroutine_handle passed to await_suspend method by our helper awakener coroutine.

The next ingredient is the await_transform in so_5::coro::promise_type_t that transforms an external Awaiter object to wrapped_awaiter_t instance:

// Helper to get Awaiter instance from Awaitable expression.
// Borrowed from:
// https://lewissbaker.github.io/2017/11/17/understanding-operator-co-await
template<typename Awaitable>
decltype(auto) get_awaiter(Awaitable&& awaitable)
{
  if constexpr (has_member_operator_co_await_v<Awaitable>)
    return static_cast<Awaitable&&>(awaitable).operator co_await();
  else if constexpr (has_non_member_operator_co_await_v<Awaitable&&>)
    return operator co_await(static_cast<Awaitable&&>(awaitable));
  else
    return static_cast<Awaitable&&>(awaitable);
}

class promise_type_t
{
  so_5::coro::awakener_task_t m_awakener;

public:
  ...
  template< typename Awaitable >
  auto
  await_transform( Awaitable && awaitable )
  {
    using awaiter_t = std::decay_t< decltype(
      get_awaiter(std::forward<Awaitable>(awaitable))) >;

    return wrapped_awaiter_t< awaiter_t >{
        m_awakener,
        get_awaiter( std::forward<Awaitable>(awaitable) )
      };
  }
};

With such ingredients we can achieve the desired behaviour. For example:

so_5::resumable_t
some_agent::evt_connect(mhood_t<msg_connect> cmd)
{
  ...
  auto client_socket = co_await A::connect(cmd->remote_ip, connect_timeout);
  ... // handling of the result.
}

In that case a call to co_await A::connect leads to the following actions:

• an instance of so_5::coro::promise_type_t will be created (I hope that the this pointer of the agent will be passed to the promise_type_t somehow);
• an Awaitable expression A::connect will be executed and the result will be passed to await_transform method of so_5::coro::promise_type_t;
• the await_transform method will receive this Awaitable and will get the actual A's Awaiter object;
• this Awaiter object will be wrapped into so_5::coro::wrapped_awaiter_t object;
• the await_suspend will be called for the new wrapped_awaiter_t object and the actual std::coroutine_handle for evt_connect coroutine will be passed to it by the C++ compiler;
• the await_suspend of the wrapped_awaiter_t will hide this handle and will call A's Awaiter's await_suspend with our helper awakener handle;
• the A will store the handle of our awakener somewhere;
• when A detects that the connect operation completes it will call the resume method of our awakener handle on some context. We don't care what this context is. The resumption of the awakener will lead to sending of the msg_resume_current_coroutine signal;
• this signal will be handled on the proper SObjectizer's thread.

[eao197]

Yet another aspect: it seems that the msg_resume_current_coroutine has to be handled via deadletter handler (see the docs). It is because the coroutine can contain so_change_state inside:

so_5::resumable_t
my_agent::evt_connect(mhood_t<msg_connect> cmd)
{
  ...
  auto socket = co_await A::connect_to(cmd->remote_ip, connection_timeout); // (1)
  if(socket)
  {
    so_change_state(st_connected);
    co_await connection_listener.register_new_connect(socket); // (2)
    ...
  }
}

Here we have to receive two msg_resume_current_coroutine signals:

• one for the point (1);
• another for the point (2).

But the second msg_resume_current_coroutine will be received when the agent is in different state.

[eao197]

It seems that the delivery of msg_resume_current_coroutine to the receiver should be guaranteed operation -- it simply shouldn't fail. If it fails then the agent with suspended coroutine can't be resumed and will be stuck forever.

But there is a possibility of losing msg_resume_current_coroutine because it will be delivered to the agent as an ordinary execution_demand_t instance. Most dispatchers use dynamic memory for holding pending execution_demands and allocation for the new demand object can throw std::bad_alloc.

It means that when an execution_demand is being passed to event_queue_t::push method an exception can be thrown (std::bad_alloc as an example).

Even if this exception is caught, the problem remains: the suspended agent won't be resumed because msg_resume_current_coroutine won't be delivered to it.

Because of that I think that sending of msg_resume_current_coroutine has to be considered as noexcept-operation.

I also think that this fact should be expressed in the code in the following way:

The event_queue_t interface has to be extended with a new virtual method:

virtual void
push_evt_resume_current_coroutine(
  execution_demand_t demand ) noexcept;

In SObjectizer-5.8 this method will have the default implementation:

void
event_queue_t::push_evt_resume_current_coroutine(
  execution_demand_t demand) noexcept
{
  this->push( demand ); // Just delegate the work.
}

This default implementation allows compatibility with existing custom event_queue implementations. The downside -- it will crash the whole application if push(demand) throws.

But in SObjectizer-5.9 this method will become a pure virtual method. So the custom implementations of the event_queue_t would be updated.

The nef_one_thread and nef_thread_pool dispatchers will implement the new push_evt_resume_current_coroutine method in a non-throwing way (using the same technique as for push_evt_finish).

[eao197]

It also seems that there is no need to send msg_resume_current_coroutine at all. The helper awakener coroutine can call push_evt_resume_current_coroutine directly via pointer to the custom event-queue:

so_5::coro::awakener_task_t
awakener()
{
  bool need_quit = false;
  while( !need_quit )
  {
    const auto action = co_await query_next_action();
    switch( action )
    {
    case awakener_action_t::resume_next:
      m_custom_event_queue.push_evt_resume_current_coroutine(...);
    break;

    case awakener_action_t::quit:
      need_quit = true;
    break;
    }
  }
}

[eao197]

The proposed approach is not compatible with prio_one_thread dispatcher.

The main idea of prio_one_thread dispatcher is to allow processing of low priority agents' demands only after handling demands of the higher priority agents. Let's suppose that we have an agent processor with priority p0 that handles take_job messages. And we also have a coordinator agent with priority p2 that handles reconfigure messages. When reconfigure arrives then all existing take_job demands should be paused until reconfigure is completely handled.

Let's suppose that the event-handler for reconfigure message uses co_await in its body, something like:

so_5::resumable_t
coordinator::evt_reconfigure(mhood_t<reconfigure> cmd) {
  ...
  auto params = co_await params_extractor.extract_new_params();
  ...
}

It means that actual processing of the reconfigure will be splitted into two parts:

• the first part is invoked when prio_one_thread dispatcher extracts the demand for reconfigure message for the coordinator agent;
• the second part is invoked when prio_one_thread dispatcher extracts the demand for msg_resume_current_coroutine message for the coordinator agent.

But the dispatcher doesn't know about the relation between reconfigure and msg_resume_current_coroutine messages. It just sees a "completion" of the demand-handler for reconfigure message and thinks that it can take the next pending demand and call demand-handler for it.

It would lead to the case when the reconfigure is not handled yet, but processing of pending take_job is started. This shouldn't happen for the prio_one_thread dispatcher.

[eao197]

One possible way to solve such a problem is to ask the dispatcher if it does support demand-handler suspension and resumption.

An agent has no access to the dispatcher directly. But there are a couple of things that have relation to the underlying dispatcher:

• the dispatcher binder. There is disp_binder_t interface and there is the so_this_agent_disp_binder method for an agent that return dispatcher binder used for the agent;
• the event queue. There is an event_queue_t interface and there is the so_bind_to_dispatcher method for an agent that is called when the agent is bound to the dispatcher.

So we can extend one of these interfaces with a method like that:

[[nodiscard]] virtual
const working_context_properties_t &
working_context_properties() const noexcept;

Where working_context_properties_t could be something like that:

class working_context_properties_t
{
public:
  [[nodiscard]]
  bool
  does_support_coroutine_suspension() const noexcept;
};

By the help of those facilities the body of evt_reconfigure could be transformed that way:

void
coordinator::_auto_generated_evt_reconfigure(mhood_t<reconfigure> cmd)
{
  // Get the resumable object for the coroutine.
  auto current_resumable = this->evt_reconfigure(cmd);

  // If the dispatcher supports coroutine suspension...
  if(so_this_agent_disp_binder()->working_context_propertines()
    .does_support_coroutine_suspension())
  {
    // Yes! We can store the task and wait for the resumption signal.
    _m_current_coroutine = std::move(current_coroutine);
    _so_deactivate_normal_processing();
  }
  else
  {
    // No, coroutines are not supported.
    // Will block the current thread.
    current_resumable.wait_for_resumption();
    current_resumable.resume();
  }
}

The prio_one_thread dispatcher will have working_context_properties that returns false from the does_support_coroutine_suspension method.

NOTE. The wait_for_resumption is a special method of so_5::resumable_t. It's not a part of C++20 coroutines interface. I assume that it will be implemented by using std::mutex/std::condition_variable.

[eao197]

Maybe it's better to check compatibilty of the dispatcher with resumption of event_handler.

At the moment I think that support of coroutines has to be added to an agent manually in the constructor:

class my_agent : public so_5::agent_t
{
  my_agent(context_t ctx)
    : so_5::agent_t{ctx + so_5::enable_courotines}
  {}
};

When this support is turned on the disp_binder has to check properties of the dispatcher at binding time. If the dispatcher doesn't support event_handler suspension/resumption then an exception has be thrown. It will prevent of binding agents with coroutines to dispatcher that doesn't support coroutines.