Implementing task primitives based on asio::awaitable

Question

I'm looking for a way to implement task primitives like whenAll, whenAny, taskFromResult on top of (boost) asios awaitable<T> coroutine type.

What I've got so far is a function that creates an awaitable<T> from a completion callback. However I'm unsure how I'm supposed to run multiple tasks in parallel on the specified io_context and await all of them or until any one of them is finished.

In .NET there are primitives like Task.WhenAny, Task.WhenAll and types like TaskCompletionSource that make it easy to work with tasks. Did anyone do this for asio based coroutines?

Answer 1

You can use the experimental operator overloads to combine awaitables.

E.g.

Live On Coliru

#include <boost/asio.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <iostream>
using namespace std::chrono_literals;
auto now = std::chrono::steady_clock::now;
static auto start = now();

using namespace boost::asio::experimental::awaitable_operators;
using boost::asio::awaitable;
using boost::asio::use_awaitable;
using boost::system::error_code;


awaitable<void> foo_and() {
    boost::asio::steady_timer tim1(co_await boost::asio::this_coro::executor, 1s);
    boost::asio::steady_timer tim2(co_await boost::asio::this_coro::executor, 2s);

    co_await (tim1.async_wait(use_awaitable) && tim2.async_wait(use_awaitable));
}

awaitable<void> foo_or() {
    boost::asio::steady_timer tim1(co_await boost::asio::this_coro::executor, 1s);
    boost::asio::steady_timer tim2(co_await boost::asio::this_coro::executor, 2s);

    co_await (tim1.async_wait(use_awaitable) || tim2.async_wait(use_awaitable));
}

int main() {
    boost::asio::io_context ioc;

    auto handler = [](auto caption) {
        return [=](std::exception_ptr e) {
            try {
                if (e)
                    std::rethrow_exception(e);
                std::cout << caption << " succeeded at ";
            } catch (std::exception const& e) {
                std::cout << caption << " failed at ";
            }
                std::cout  << (now() - start) / 1.0s << "s" << std::endl;
        };
    };

    co_spawn(ioc.get_executor(), foo_and(), handler("foo_and"));
    co_spawn(ioc.get_executor(), foo_or(), handler("foo_or"));

    ioc.run();
}

Prints e.g.

foo_or succeeded at 1.00153s
foo_and succeeded at 2.00106s

BONUS

With some more C++17 and default completion tokens:

awaitable<void> foo_and(auto... delays) {
    auto ex = co_await boost::asio::this_coro::executor;

    co_await(Timer(ex, delays).async_wait() && ...);
}

awaitable<void> foo_or(auto... delays) {
    auto ex = co_await boost::asio::this_coro::executor;

    co_await(Timer(ex, delays).async_wait() || ...);
}

Now you can supply variadic lists of delays:

co_spawn(ioc.get_executor(), foo_and(100ms, 1500ms, 75ms), handler("foo_and"));
co_spawn(ioc.get_executor(), foo_or(1s, 5min, 65ms), handler("foo_or"));

See it Live On Coliru:

#include <boost/asio.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <iostream>
using namespace std::chrono_literals;
auto now = std::chrono::steady_clock::now;
static auto start = now();

using namespace boost::asio::experimental::awaitable_operators;
using boost::asio::awaitable;
using boost::asio::use_awaitable;
using boost::system::error_code;

using Timer = boost::asio::use_awaitable_t<>::as_default_on_t<boost::asio::steady_timer>;

awaitable<void> foo_and(auto... delays) {
    auto ex = co_await boost::asio::this_coro::executor;

    co_await(Timer(ex, delays).async_wait() && ...);
}

awaitable<void> foo_or(auto... delays) {
    auto ex = co_await boost::asio::this_coro::executor;

    co_await(Timer(ex, delays).async_wait() || ...);
}

int main() {
    boost::asio::io_context ioc;

    auto handler = [](auto caption) {
        return [=](std::exception_ptr e) {
            try {
                if (e)
                    std::rethrow_exception(e);
                std::cout << caption << " succeeded at ";
            } catch (std::exception const& e) {
                std::cout << caption << " failed at ";
            }
                std::cout  << (now() - start) / 1ms << "ms" << std::endl;
        };
    };

    co_spawn(ioc.get_executor(), foo_and(100ms, 1500ms, 75ms), handler("foo_and"));
    co_spawn(ioc.get_executor(), foo_or(1s, 5min, 65ms), handler("foo_or"));

    ioc.run();
}

Prints e.g.

foo_or succeeded at 65ms
foo_and succeeded at 1500ms