如何并行运行N个boost::asio::awaitable协程并添加超时机制？

Deadline Timer 思路的可行性 #

你的Deadline Timer方案完全可行，这是Boost.Asio生态中实现批量协程超时取消的常规思路之一。核心逻辑是通过统一的取消信号绑定所有动态生成的协程，利用Deadline Timer触发全局取消，完美契合你“并行执行+超时统一取消”的需求。

该方案的优势：

• 贴合Boost.Asio原生设计：借助cancellation_signal和协程的取消机制，无需手动维护协程生命周期，避免内存泄漏或逻辑混乱
• 逻辑清晰：超时触发与协程执行解耦，代码可读性高
• 扩展性强：动态添加协程时只需绑定取消信号，无需修改超时逻辑

具体实现步骤与代码示例 #

核心组件说明 #

1. cancellation_signal：全局取消信号，所有动态协程绑定该信号的取消槽，超时后触发emit()即可取消所有关联协程
2. when_all：等待所有动态协程完成，结合race操作符与Timer的awaitable，实现“任务全部完成/超时”的二选一逻辑
3. 协程内取消响应：协程需通过cancellation_state绑定全局信号，并在异步操作中传递取消令牌，确保取消能被正确捕获

完整代码示例 #

#include <boost/asio.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <vector>
#include <iostream>

using namespace boost::asio;
using namespace boost::asio::experimental::awaitable_operators;

// 单个任务协程：模拟动态生成的业务逻辑，支持取消
awaitable<void> dynamic_task(int task_id, cancellation_signal& global_cancel) {
    try {
        // 将当前协程的取消状态绑定到全局取消信号
        auto local_cancel = co_await this_coro::cancellation_state;
        local_cancel.link_cancellation_slot(global_cancel.slot());

        // 模拟耗时异步操作（示例：带取消的sleep）
        steady_timer timer(co_await this_coro::executor);
        timer.expires_after(std::chrono::seconds(task_id + 2)); // 不同任务耗时不同
        // 使用as_tuple避免取消抛出异常，也可以直接捕获operation_aborted错误
        auto [ec] = co_await timer.async_wait(as_tuple(cancel_token));
        if (ec) {
            std::cout << "Task " << task_id << " cancelled: " << ec.message() << "\n";
            co_return;
        }

        std::cout << "Task " << task_id << " executed successfully\n";
    } catch (const system_error& e) {
        if (e.code() == error::operation_aborted) {
            std::cout << "Task " << task_id << " cancelled due to timeout\n";
        } else {
            std::cout << "Task " << task_id << " failed: " << e.what() << "\n";
        }
    }
}

// 主协程：管理动态任务、超时逻辑
awaitable<void> batch_task_manager(io_context& io_ctx, std::chrono::seconds timeout) {
    cancellation_signal global_cancel;
    std::vector<awaitable<void>> task_list;

    // 模拟运行时动态生成任务（数量可根据业务条件调整）
    for (int i = 0; i < 3; ++i) {
        task_list.emplace_back(dynamic_task(i, global_cancel));
    }

    // 超时触发逻辑：Timer到期后发出全局取消信号
    auto timeout_trigger = [&]() -> awaitable<void> {
        steady_timer timer(io_ctx);
        timer.expires_after(timeout);
        co_await timer.async_wait();
        global_cancel.emit(cancellation_type::all);
        std::cout << "Batch operation timed out, cancelling all tasks\n";
    }();

    // 等待所有任务完成 或 超时触发，谁先结束就执行对应逻辑
    co_await (when_all(task_list.begin(), task_list.end()) || timeout_trigger);
}

int main() {
    io_context io_ctx;
    // 启动主协程，超时设置为3秒
    co_spawn(io_ctx, batch_task_manager(io_ctx, std::chrono::seconds(3)), detached);
    io_ctx.run();
    return 0;
}

关键注意事项 #

• 所有动态生成的协程必须绑定global_cancel的取消槽，否则无法被超时取消
• 协程内的异步操作需支持取消：要么传递cancel_token，要么使用as_tuple处理取消错误，避免未捕获的异常导致程序崩溃
• 如果任务中存在阻塞操作（非异步），需要手动检查取消状态（比如调用cancellation_state::cancelled()），否则无法及时响应取消

内容的提问来源于stack exchange，提问作者Mikey A. Leonetti