基于ASIO实现自定义异步队列等待函数的技术求助

我正在开发一款用于多人游戏的异步UDP服务器（个人学习项目），使用独立版ASIO与C++20协程开发。需要实现一个类似ASIO内置async_read的自定义异步函数async_wait_for_queue：当传入队列为空时将控制权交还给io_context，队列有元素时自动恢复执行。目标使用场景如下：

asio::awaitable<void> udp_server::process() {
    // 持续处理队列消息
    while (true) {
        // 暂停直到队列有消息，期间io_context可处理其他任务（如客户端读取、计算）
        co_await async_wait_for_queue(coro_queue_in, asio::use_awaitable);
        
        // 取出并处理消息
        auto& msg = coro_queue_in.front();
        coro_queue_in.pop();
        std::cout << msg << "\n";
    }
}

现有实现代码可编译运行，但协程会永远挂起，即便队列新增元素也无法恢复：

struct async_queue_implementation {
    std::queue<owned_message>& queue_;
    template<typename Self> void operator()(Self& self) {
        if (!queue_.empty()) {
            self.complete();
        }
    }
};

template<typename CompletionToken>
auto async_wait_for_queue(std::queue<owned_message>& queue,
    CompletionToken&& token) ->
    decltype(asio::async_compose<CompletionToken, void()>(
    std::declval<async_queue_implementation>(), token))
{
    return asio::async_compose<CompletionToken, void()>(
        async_queue_implementation{ queue }, token);
}

现有实现仅在async_compose初始化时检查一次队列状态：如果当时队列为空，没有任何逻辑会在后续队列新增元素时唤醒等待的协程，导致协程永久挂起。

需要给队列添加通知机制，在新元素入队时主动触发等待的异步操作，同时保证线程安全（UDP接收与队列处理可能在不同协程/线程执行）。以下是完整实现：

1. 封装带通知功能的线程安全队列 #

#include <asio.hpp>
#include <queue>
#include <mutex>
#include <vector>

template<typename T>
class notify_queue {
public:
    explicit notify_queue(asio::any_io_executor executor)
        : executor_(std::move(executor)) {}

    // 入队并触发通知
    void push(T value) {
        std::lock_guard lock(mutex_);
        queue_.push(std::move(value));
        notify_waiters();
    }

    // 尝试出队（线程安全）
    bool try_pop(T& out) {
        std::lock_guard lock(mutex_);
        if (queue_.empty()) return false;
        out = std::move(queue_.front());
        queue_.pop();
        return true;
    }

    // 异步等待队列非空
    template<typename CompletionToken>
    auto async_wait(CompletionToken&& token) {
        return asio::async_compose<CompletionToken, void()>(
            [this](auto& self) {
                std::lock_guard lock(mutex_);
                if (!queue_.empty()) {
                    self.complete();
                    return;
                }
                // 队列为空时，保存等待操作
                waiters_.push_back(std::move(self));
            },
            token,
            executor_
        );
    }

private:
    // 通知所有等待的异步操作
    void notify_waiters() {
        std::vector<asio::async_compose_handler<void()>> waiters;
        {
            std::lock_guard lock(mutex_);
            waiters.swap(waiters_);
        }
        // 在executor上触发完成，符合ASIO线程安全规则
        for (auto& waiter : waiters) {
            asio::post(executor_, [waiter = std::move(waiter)]() mutable {
                waiter.complete();
            });
        }
    }

    asio::any_io_executor executor_;
    std::mutex mutex_;
    std::queue<T> queue_;
    std::vector<asio::async_compose_handler<void()>> waiters_;
};

2. 修改process协程 #

使用封装后的队列替代原std::queue：

asio::awaitable<void> udp_server::process() {
    notify_queue<owned_message> coro_queue_in(co_await asio::this_coro::executor);

    while (true) {
        co_await coro_queue_in.async_wait(asio::use_awaitable);
        
        owned_message msg;
        // 循环取出所有可用消息（避免多次等待）
        while (coro_queue_in.try_pop(msg)) {
            std::cout << msg << "\n";
        }
    }
}

3. UDP接收协程示例 #

在接收客户端数据时，通过push方法入队并自动触发通知：

asio::awaitable<void> udp_server::receive_loop() {
    asio::ip::udp::socket socket(co_await asio::this_coro::executor, 
        asio::ip::udp::endpoint(asio::ip::udp::v4(), 1234));
    
    while (true) {
        owned_message msg;
        co_await socket.async_receive_from(asio::buffer(msg.data, msg.size), 
            msg.sender, asio::use_awaitable);
        
        coro_queue_in.push(std::move(msg));
    }
}

• 线程安全：用std::mutex保护队列与等待列表的访问，适配多协程/线程场景。
• 延迟通知：仅在队列新增元素时触发等待操作，避免无意义的轮询。
• ASIO规则兼容：通过asio::post在指定executor上完成异步操作，符合ASIO的线程安全要求。

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