Boost Asio 1.84.0：如何并行等待数量动态的可等待异步函数？

场景说明 #

使用Boost 1.84.0时，Boost Asio的并行co_wait功能对应两种场景：

• 静态固定数量：例如同时等待async_read与超时操作（共2个异步函数）
• 动态变化数量：例如向所有在线聊天客户端执行async_write，需等待全部完成后返回"消息已送达"提示

静态数量异步函数的等待方案 #

针对编译期确定操作数的场景，有两种适配方案：

1. 可变参数版experimental::make_parallel_group：仅支持使用deferred或experimental::use_promise的异步操作，无法与use_awaitable或自定义awaitable<T>函数配合
2. experimental::awaitable_operator：适配use_awaitable或自定义awaitable<T>函数，但不支持deferred或experimental::use_promise

动态数量异步函数的现有方案 #

range版experimental::make_parallel_group #

该方案仅支持deferred或experimental::use_promise风格的异步操作，无法直接适配use_awaitable或自定义awaitable<T>函数，示例代码：

#include <iostream>
#include <vector>
#include <boost/asio.hpp>
#include <boost/asio/experimental/co_composed.hpp>
#include <boost/asio/experimental/promise.hpp>
#include <boost/asio/experimental/use_promise.hpp>
#include <boost/asio/experimental/parallel_group.hpp>
namespace as = boost::asio;

as::awaitable<void> test() {
    auto exe = co_await as::this_coro::executor;
    using op_type = decltype(as::post(exe, as::append(as::deferred, int{})));
    std::vector<op_type> ops;
    for (int i = 0; i != 5; ++i) {
        ops.push_back(as::post(exe, as::append(as::deferred, i*10)));
    }
    auto [orders, values] = 
        co_await as::experimental::make_parallel_group(
            ops
        ).async_wait(
            as::experimental::wait_for_all(),
            as::deferred
        );
    for (auto order : orders) {
        std::cout << order << " " << values[order] << std::endl;
    }
    co_return;
}

int main() {
    as::io_context ioc;
    as::co_spawn(
        ioc.get_executor(),
        test,
        [](auto...){ std::cout << "async_func() finished" << std::endl; }
    );
    ioc.run();
    std::cout << "main() finished" << std::endl;
}

当前采用的CompletionToken风格方案 #

为适配动态等待需求，放弃awaitable<T>函数，改用CompletionToken风格实现，但需要将所有关联的异步函数都改为同风格，示例代码：

template <typename CompletionToken>
auto completion_token_based_async_func(as::any_io_executor exe, CompletionToken&& token) {
    return as::async_initiate<
        CompletionToken,
        void()
    >(
        as::experimental::co_composed<
            void()
        >(
            [](
                auto /* state */,
                as::any_io_executor exe
            ) -> void {
                // co_await awaitable_function() // 此处报错
                co_await other_async_func(as::deferred);
                co_return {};
            }
        ),
        token
    );
}

更优雅的实现方案 #

无需重构所有函数为CompletionToken风格，只需将awaitable<T>函数转换为支持deferred的操作，即可兼容range版make_parallel_group，核心思路如下：

方法1：为awaitable函数添加CompletionToken重载 #

如果自定义awaitable函数支持CompletionToken参数，直接传入as::deferred即可生成可加入并行组的操作：

// 自定义支持CompletionToken的awaitable函数
template <typename CompletionToken = as::use_awaitable_t<>>
auto custom_awaitable_func(as::any_io_executor exe, int value, CompletionToken&& token = {}) {
    return as::async_initiate<CompletionToken, void(int)>(
        [exe, value](auto handler) {
            as::co_spawn(exe, [value]() -> as::awaitable<int> {
                co_await as::post(exe, as::use_awaitable);
                co_return value * 2;
            }, std::move(handler));
        },
        token
    );
}

// 动态并行等待示例
as::awaitable<void> dynamic_awaitable_test() {
    auto exe = co_await as::this_coro::executor;
    std::vector<decltype(custom_awaitable_func(exe, 0, as::deferred))> ops;

    for (int i = 0; i < 5; ++i) {
        ops.push_back(custom_awaitable_func(exe, i, as::deferred));
    }

    auto [orders, values] = co_await as::experimental::make_parallel_group(ops)
        .async_wait(as::experimental::wait_for_all(), as::deferred);

    for (auto order : orders) {
        std::cout << "操作" << order << "返回值：" << values[order] << std::endl;
    }
    co_return;
}

方法2：用co_spawn包装现有awaitable函数 #

如果自定义awaitable函数不支持CompletionToken重载，可通过co_spawn结合as::deferred将其转换为兼容操作：

// 现有自定义awaitable函数
as::awaitable<int> custom_awaitable_func(as::any_io_executor exe, int value) {
    co_await as::post(exe, as::use_awaitable);
    co_return value * 2;
}

// 动态并行等待示例
as::awaitable<void> dynamic_awaitable_test() {
    auto exe = co_await as::this_coro::executor;
    std::vector<decltype(as::co_spawn(exe, custom_awaitable_func(exe, 0), as::deferred))> ops;

    for (int i = 0; i < 5; ++i) {
        ops.push_back(as::co_spawn(exe, custom_awaitable_func(exe, i), as::deferred));
    }

    auto [orders, values] = co_await as::experimental::make_parallel_group(ops)
        .async_wait(as::experimental::wait_for_all(), as::deferred);

    for (auto order : orders) {
        std::cout << "操作" << order << "返回值：" << values[order] << std::endl;
    }
    co_return;
}

通过上述方式，既能保留awaitable函数的协程语法便利性，又能实现动态数量异步函数的并行等待，无需放弃协程风格重构代码。

内容的提问来源于stack exchange，提问作者Takatoshi Kondo