Coroutines are not distributed over asio::thread_pool threads

Question

I wanted to try out to c++20 coroutines together with asio. In a simple test three are coroutines which would be executed on a asio::thread_pool with 4 threads. When I run the test all the coroutines are executed one-by-one after each other, and not simultaneously. This is not the behavior that I expected. I thought that the executor of asio::thread_pool would distribute the coroutines over multiple threads. Is there something that I overlooked?

using namespace asio::experimental::awaitable_operators;

asio::awaitable<std::string> one() {
    std::this_thread::sleep_for( 1s );
    co_return "egy";
}

asio::awaitable<std::string> two() {
    std::this_thread::sleep_for( 1s );
    co_return "ketto";
}

asio::awaitable<std::string> three() {
    std::this_thread::sleep_for( 1s );
    co_return "harom";
}

asio::awaitable<void> run_all() {
    auto [first, second, third] = co_await( one() &&
                                            two() &&
                                            three() );
}

int main() {
    asio::thread_pool pool( 4 );
    co_spawn( pool, run_all(), asio::detached );
    pool.join();

    return 0;
}

Running example: https://godbolt.org/z/affo4EvWb

Answer 1

This will create a coroutine and distribute it to the thread in thread pool. Thread pool's Schedular will work in here.

co_spawn( pool, run_all(), asio::detached );

Now you are in one thread of the thread pool, and the coroutine context will run in this thread, thread pool's schedular will not work here.

Instead of thread pool's schedular, The coroutine's schedular will work here. And this schedular will just run in the same thread.

asio::awaitable<void> run_all()
{
auto [first, second, third] = co_await( one() && two() && three() );
}

What is the operator &&?

It create 3 coroutine in the caller's executor, run all of them and wait them for finish.

So The coroutine function one, two, three will run in the same executor as run_all.

In coroutine function, you can get the executor by co_await asio::this_coro::executor.

That is why they are run in the same thread.

Remember, the coroutine function shouldn't be blocked. The coroutine's schedular is run in user's code, not in OS. the function std::this_thread::sleep_for will just block the thread, and coroutine's schedular can't do anything.

asio::awaitable<std::string> three() {
    std::this_thread::sleep_for( 1s );
    co_return "harom";
}

How to let coroutine schedule know that you what to yield for a second? You need call an async function.

asio::awaitable<std::string> echo(std::string_view sv) {
    asio::steady_timer timer(co_await asio::this_coro::executor);
    timer.expires_after(1s);
    co_await timer.async_wait(asio::use_awaitable);
    co_return sv;
}

you can see the result here. link

in the code co_await clk.async_wait_for(1s). you notify the schedular that you want to yield for 1s. That it is worked.

Remember, if you call an function without co_await or co_return, it must be an blocked call.

ASIO supply many async I/O function, such as timer, file, socket, ssl, pipe, channel, serial ports and signals. Those function have an async prefix.

By the way, if you want to define your own async call, see this sample:

using namespace asio;
static auto async_yield(auto&& handler)
{
    return async_initiate<decltype(handler), void(std::error_code)>(
        [](auto&& handler) {
            auto executor = get_associated_executor(handler);
            auto state = cancellation_state(get_associated_cancellation_slot(handler));
            post(executor, [handler = std::move(handler), state = std::move(state)]() mutable {
                std::error_code e;
                if (state.cancelled() != cancellation_type::none)
                    e = error::operation_aborted;
                handler(e);
            });
        }, handler
        );
}

asio::awaitable<void> foo()
{
   while(true)
   {
       co_await async_yield(asio::use_awaitable); //just like std::this_thread::yield(), but work in ASIO's coroutine context.
   }
}

another example for heavy calculate:

//a heavy calculation work 10s in another thread.
asio::thread_pool pool;
using namespace std::chrono;
using namespace asio;
auto async_calculation(auto&& handler)
{
    return async_initiate<decltype(handler), void(std::error_code e,size_t)>(
        [](auto&& handler) {
            auto slot = get_associated_cancellation_slot(handler);
            auto state = cancellation_state(get_associated_cancellation_slot(handler));
            auto executor = get_associated_executor(handler);
            post(pool, [executor = std::move(executor), handler = std::move(handler), state = std::move(state)]{
                size_t answer=0;
                auto now = steady_clock::now();
                while (state.cancelled() == cancellation_type::none && steady_clock::now() - now < 10s)
                {
                    answer+=1;
                }
                post(executor,[handler = std::move(handler), state = std::move(state), answer]() mutable {
                    std::error_code e;
                    if (state.cancelled() != cancellation_type::none)
                        e = error::operation_aborted;
                    handler(e,answer);
                });
            });
        }, handler
        );
}
asio::awaitable<void> foo()
{
   //this calculation won't block the coroutine.
   size_t answer=co_await async_calculation(asio::use_awaitable);
}

It's dirty for author, but beautiful for user.