Boost ASIO自定义超时令牌与异步操作取消传播问题

核心问题定位 #

你的自定义async_wait_until未与超时令牌的取消机制联动——Boost ASIO的超时令牌依赖**取消槽（cancellation slot）**传递取消信号，内置操作已经适配该机制，但自定义操作需要手动绑定取消逻辑，才能在超时触发时终止等待并通知调用方。

关键修复步骤 #

1. 为等待项添加取消连接存储，用于绑定和管理取消回调
2. 在async_wait_until初始化时，获取超时令牌的取消槽并绑定取消处理逻辑
3. 取消触发时，安全移除等待项并调用完成处理器返回取消错误
4. 用智能指针避免迭代器失效问题，确保多操作场景下的安全性

代码实现示例 #

调整等待项结构 #

将等待项改为智能指针存储，避免迭代器失效：

#include <boost/asio.hpp>
#include <list>
#include <memory>
#include <functional>

// 假设你的消息类型
struct Message { /* ... */ };

struct WaitEntry {
    std::function<bool(const Message&)> predicate;
    asio::any_completion_handler<void(asio::error_code, Message)> handler;
    asio::cancellation_slot::connection cancel_connection;
};

std::list<std::shared_ptr<WaitEntry>> wait_list_;
asio::any_io_executor executor_; // 协议类持有的执行器

适配取消逻辑的async_wait_until实现 #

template <typename Predicate, typename Token>
auto async_wait_until(Predicate pred, Token&& token) {
    // 初始化异步操作，处理令牌类型
    using completion_signature = void(asio::error_code, Message);
    asio::async_completion<Token, completion_signature> init(token);

    // 创建等待项
    auto entry = std::make_shared<WaitEntry>();
    entry->predicate = std::move(pred);
    entry->handler = std::move(init.completion_handler);

    // 获取超时令牌的取消槽，绑定取消回调
    auto cancel_slot = asio::get_cancellation_slot(token);
    if (cancel_slot.is_valid()) {
        entry->cancel_connection = cancel_slot.connect(
            [this, entry](asio::cancellation_type type) {
                // 仅处理超时类型的取消（可根据需求调整）
                if (!(type & asio::cancellation_type::timeout)) return;

                // 从列表中查找并移除等待项
                auto it = std::find_if(wait_list_.begin(), wait_list_.end(),
                    [entry](const auto& ptr) { return ptr == entry; });
                if (it == wait_list_.end()) return;

                // 在执行器上下文调用完成处理器，传递取消错误
                asio::dispatch(executor_, [handler = std::move((*it)->handler)]() mutable {
                    handler(asio::error::operation_aborted, Message{});
                });
                wait_list_.erase(it);
            });
    }

    // 将等待项加入列表
    wait_list_.push_back(entry);

    return init.result.get();
}

消息匹配时的清理逻辑 #

当收到匹配消息时，需先断开取消连接再调用处理器，避免重复触发：

void on_message_received(const Message& msg) {
    auto it = wait_list_.begin();
    while (it != wait_list_.end()) {
        if ((*it)->predicate(msg)) {
            // 断开取消连接，防止超时回调重复调用
            (*it)->cancel_connection.disconnect();
            // 调用处理器
            asio::dispatch(executor_, [handler = std::move((*it)->handler), msg]() mutable {
                handler(asio::error_code{}, msg);
            });
            // 移除项
            it = wait_list_.erase(it);
        } else {
            ++it;
        }
    }
}

验证要点 #

• 触发超时后，检查wait_list_中对应项是否被移除
• 等待的协程/回调是否收到asio::error::operation_aborted错误
• 消息正常匹配时，处理器是否能正确收到消息且无重复调用

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