From explicit coroutine schedulers to implicit?

[eao197]

When implementing support for C++20 coroutines, I proceeded from the fact that the concept of a scheduler is required - an object responsible for resuming the coroutines in the working context desired by a user.

Now this scheduler is explicitly passed, for example, to the asynchronous version of select:

so_5::cpp_coro::this_thread_scheduler_t scheduler{ ... };

co_await so_5::cpp_coro::select( scheduler, ... );

If the asynchronous select is called inside the agent event, then the scheduler for the coroutine can be obtained from the agent itself through the new agent_t::so_this_agent_scheduler method:

so_5::cpp_coro::task_t<void>
my_agent::some_event( mhood_t<some_msg> cmd )
{
  co_await so_5::cpp_coro::select( this->so_this_agent_scheduler(), ... );
  ...
}

So far, this approach has not caused any problems. However, now in the implementation of support for coroutines in SO-5, I have reached the point where I need to allow using any coroutines inside the event handler, not just those provided by SObjectizer. That is, I need to make sure that the event handler can be written like this:

so_5::cpp_coro::task_t<void>
my_agent::event_with_foreign_coro( mhood_t<some_msg> cmd )
{
  ...
  auto r = co_await third_party::func( ... );
  ...
}

where func from the third-party library third_party creates a coroutine that will be executed not in the context of SObjectizer, but in the context owned by the library `third_party'.

Accordingly, the event handler event_with_foreign_coro starts working in the context of SObjectizer, then pauses on co_await, and then resumes, but this time in the context of the library `third_party'. Which is fundamentally wrong, because SObjectizer must ensure that agent handlers work only on those threads to which the agent is bound.

To cope with this situation, I planned to implement in so_5::cpp_coro::task_t::promise_type the await_transform method, which will cunningly replace the awaitable and awaiter from third_party with special awaitable and awaiter from SObjectizer. This special awaitable will be woken up by the third_party library on someone else's context, but awaitable will not wake up the source coroutine on this context, but instead will send a special message to SObjectizer so that SObjectizer wakes up the source coroutine on the correct working thread.

I already have a prototype of such a solution. But there were several ambiguous points.

---

First, when so_5::cpp_coro::task_t::promise_type is created, the current scheduler must be saved in the promise instance, which should be used to resume the coroutine. And C++ does not provide an explicit way to do this.

For example, if the event handler is a lambda function like:

so_subscribe_self().event( [](mhood_t<msg> cmd) -> so_5::cpp_coro::task_t<void> {...} );

When the compiler creates the promise object in this lambda function, either the promise_type constructor without parameters will be used, or the constructor with the only parameter mhood_t<msg>. There is no way to pass the scheduler explicitly.

Probably the easiest way to achieve the desired result is to implicitly save a pointer to the current scheduler in a thread-local variable. SObjectizer will do this automatically before calling the agent's event handler.

---

Secondly, initially I thought that await_transform would divide awaitable into ours and aliens. That is, if await_transform is called for so_5::cpp_coro::task_t<T>, then nothing is done, the original awaitable is returned. But if await_transform is called for awaitable from a third-party library, then the third-party awaitable is wrapped in our special auxiliary awaitable.

However, it turns out that so_5::cpp_coro::task_t should be bound to scheduler immediately, because this binding is necessary for the proper operation of await_transform. And if task_t is linked to some kind of scheduler, then where is the guarantee that this is the correct scheduler?

In fact, there are no guarantees. And in the current experimental version, you can step on such an unpleasant rake:

struct msg_take_ctx final : public so_5::message_t
{
	so_5::cpp_coro::scheduler_t & m_scheduler;

	msg_take_ctx( so_5::cpp_coro::scheduler_t & scheduler )
		: m_scheduler{ scheduler }
	{}
};

class a_ctx_provider_t final : public so_5::agent_t
{
	const so_5::mbox_t m_dest;

public:
	a_ctx_provider_t( context_t ctx, so_5::mbox_t dest )
		: so_5::agent_t{ ctx + so_5::enable_cpp_coro() }
		, m_dest{ std::move(dest) }
	{}

	void
	so_evt_start() override
	{
		so_5::send< msg_take_ctx >( m_dest, so_this_agent_scheduler() );
	}
};

class a_consumer_t final : public so_5::agent_t
{
	struct msg_handle_it final : public so_5::signal_t {};

public:
	a_consumer_t( context_t ctx )
		: so_5::agent_t{ ctx + so_5::enable_cpp_coro() }
	{}

	void
	so_define_agent() override
	{
		so_subscribe_self().event( &a_consumer_t::evt_take_ctx );
	}

private:
	so_5::cpp_coro::task_t< void >
	evt_take_ctx( mhood_t<msg_take_ctx> cmd )
	{
		const auto this_thread_id = std::this_thread::get_id();
		std::osyncstream(std::cout) << this_thread_id << " | start" << std::endl;

		auto ch = so_5::create_mchain( so_environment() );

		so_5::send_delayed< msg_handle_it >( ch, std::chrono::milliseconds{ 50 } );

		co_await so_5::cpp_coro::select(
				// NOTE: this is not so_this_agent_scheduler!
				cmd->m_scheduler,
				so_5::from_all().handle_n(1),
				so_5::receive_case( ch,
					[]( mhood_t<msg_handle_it> ) {
						std::osyncstream(std::cout) << std::this_thread::get_id()
								<< " | msg_handle_it received" << std::endl;
					} ) );

		const auto this_thread_id_2 = std::this_thread::get_id();
		std::osyncstream(std::cout) << this_thread_id_2 << " | resume after select"
				<< std::endl;

		if( this_thread_id != this_thread_id_2 )
		{
			std::osyncstream(std::cout) << this_thread_id_2 << " | throwing"
					<< std::endl;
			throw std::runtime_error{ "work resumed on a different thread" };
		}
	}
};

void
run_demo()
{
	so_5::launch( []( so_5::environment_t & env ) {
			env.introduce_coop(
					so_5::disp::active_obj::make_dispatcher( env ).binder(),
					[]( so_5::coop_t & coop ) {
						auto consumer = coop.make_agent< a_consumer_t >();
						coop.make_agent< a_ctx_provider_t >(
								consumer->so_direct_mbox() );
					} );
		} );
}

There are two agents tied to different work threads.

The agent a_consumer_t launches the select coroutine, but binds this coroutine to foreign scheduler'. Therefore, when the coroutine resumes, the agent a_consumer_t` wakes up, but on a different working thread. Which should not be allowed, because then SObjectizer loses one of its main guarantees!

Which means that the await_transform for the case of so_5::cpp_coro::task_t cannot have such a trivial implementation:

template< typename T >
so_5::cpp_coro::task_t<T>
await_transform( so_5::cpp_coro::task_t<T> what ) { return what; }

Even for so_5::cpp_coro::task_t, you will have to return a new special awaiter. Which is not a good thing in terms of the number of actions performed by the coroutines.

---

Based on the above, I'm starting to think that it might be more correct to have implicit scheduler objects.

Let's consider working with asynchronous select in two situations.

Situation one: select is used outside of agents. Now it requires writing like this:

so_5::cpp_coro::this_thread_scheduler_t scheduler{ ... };

co_await so_5::cpp_coro::select( scheduler, ... );

It could be done like this:

so_5::cpp_coro::this_thread_scheduler_t scheduler{ ... };

// NOTE: this is a very important line!
so_5::cpp_coro::current_scheduler_setter_t scheduler_setter{ scheduler };

// NOTE: there is no more need to pass a scheduler to `select`.
co_await so_5::cpp_coro::select( ... );

The instance of current_scheduler_setter will save a reference to the current scheduler in the thread-local variable. And select will take a reference from this variable. If there is no link, then select will throw an exception.

Situation two: select is used inside the agent. Now you have to write like this:

so_5::cpp_coro::task_t<void>
my_agent::some_event( mhood_t<some_msg> cmd )
{
  co_await so_5::cpp_coro::select( this->so_this_agent_scheduler(), ... );
  ...
}

It could be done like this:

so_5::cpp_coro::task_t<void>
my_agent::some_event( mhood_t<some_msg> cmd )
{
  co_await so_5::cpp_coro::select( ... );
  ...
}

SObjectizer itself will use current_scheduler_setter_t when calling event handlers for agents with coroutines.

---

With this approach, the await_transform for the case of so_5::cpp_coro::task_t can be made quite simple with a single check inside:

template< typename T >
so_5::cpp_coro::task_t<T>
await_transform( so_5::cpp_coro::task_t<T> what )
{
  if( not is_same_scheduler(this->scheduler(), what.scheduler()) )
    throw std::runtime_error{ "..." };

  return what;
}

If task_t is linked to another scheduler, then an exception will be thrown. But if task_t is bound to the correct scheduler, then no additional actions are performed.

Plus, in many cases, the user will not have to think about the scheduler at all: in agents, the correct scheduler will be installed by SObjectizer itself. If the agents are not used, it will be enough to create a "current_scheduler_setter" next to the "this_thread_scheduler".

On the other hand, because of the hidden control of the "scheduler" from the user's eyes, there is a feeling of some "magic" - something somehow arises from somewhere. And I personally don't like this kind of implicitness very much.

[eao197]

It seems that implicit scheduler with so_5::cpp_coro::current_scheduler_setter_t is not good in case when a coroutine with lazy start is created just before creation of a scheduler. For example, something like that:

so_5::cpp_coro::task_t<void>
some_task(...)
{
  ... // Some code with co_await/co_return.
}

int main()
{
  auto first_task = some_task(...); // Coroutine created, but not started yet.
  auto second_task = some_task(...); // Coroutine created, but not started yet.
  ...
  // Only now we can create a scheduler.
  so_5::cpp_coro::this_thread_scheduler_t scheduler{ ... };
  so_5::cpp_coro::current_scheduler_setter_t current_scheduler{ scheduler };
  ...
  scheduler.sync_wait(first_task);
  ...
  scheduler.sync_wait(second_task);
  ...
}

In this case an exception will be thrown at the first call of some_task. It's because of the creation of promise_type inside some_task. At this moment there is no current scheduler and promise_type can't get the scheduler.

But I don't know if it is a realistic scenario.

Even if it is, I don't know how to bind an instance of promise_type with the scheduler without holding a pointer to the scheduler in a thread-local variable :(

[eao197]

Another possible drawback is that it makes it impossible to create a coroutine in one thread and pass it to another thread for execution.

For example, with ordinary functions we can do something like that:

using my_functor_t = std::function<void()>;

struct msg_transfer_execution : public so_5::message_t
{
  my_functor_t m_what;
  ...
};

void task_maker_agent::evt_some_event(mhood_t<msg_create_task>) {
  // Create a function.
  my_functor_t f = []() { ... /* some code to be executed somewhere else */ };

  // Pass it for execution to another thread.
  so_5::send<msg_transfer_execution>(select_free_worker, f);
}

void worker_agent::evt_execute_task(mhood_t<msg_transfer_execution> cmd) {
  // Execution of the task created elsewhere.
  cmd->m_what();
}

But if a coroutine is being bound to a scheduler implicitly during creation then we can't do that:

using my_functor_t = std::function<so_5::cpp_coro::task_t<void>()>;

struct msg_transfer_execution : public so_5::message_t
{
  my_functor_t m_what;
  ...
};

void task_maker_agent::evt_some_event(mhood_t<msg_create_task>) {
  // Create a coroutine.
  my_functor_t f = []() -> so_5::cpp_coro::task_t<void> {
     ... /* some code to be executed somewhere else */
  };

  // Pass it for execution to another thread.
  so_5::send<msg_transfer_execution>(select_free_worker, f);
}

so_5::cpp_coro::task_t<void>
worker_agent::evt_execute_task(mhood_t<msg_transfer_execution> cmd) {
  // Execution of the task created elsewhere.
  co_await cmd->m_what();
}

And it seems to me that this is a very big drawback.