关于Boost.Asio中async_compose函数的疑问：回调重复触发

我正在学习用Boost.Asio编写自定义异步函数，尝试实现一个异步NTP查询功能。参考Boost.Asio的async_compose文档写了示例代码，但发现回调函数被调用了三次，找不到问题所在，也没找到更多相关文档或示例，求指点。

#include <cstdint>
#include <iostream>
#include <memory>
#include <string>

#include <boost/asio/buffer.hpp>
#include <boost/asio/compose.hpp>
#include <boost/asio/coroutine.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/address_v4.hpp>
#include <boost/asio/ip/udp.hpp>
#include <boost/system/error_code.hpp>

namespace asio = boost::asio;
namespace sys  = boost::system;
using asio::ip::udp;

// Address of one NTP server.
static udp::endpoint ntp_server_ep =
    udp::endpoint(asio::ip::address_v4::from_string("203.107.6.88"), 123);

// Fixed NTP request message.
static std::array<uint8_t, 48> ntp_req_message{
    0x1B, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00};

void hexdump(const void* data, std::size_t size)
{
    const uint8_t* pdata = (const uint8_t*) data;

    for (std::size_t n = 0; n < size; ++n) {
        std::printf("%02x ", pdata[n]);
    }
    std::printf("\n");
}

// #include <boost/asio/yield.hpp>
struct async_ntp_query_impl
{
    udp::socket& socket_;
    asio::coroutine coro_;
    udp::endpoint local_ep_;
    std::unique_ptr<uint8_t[]> buffer_{nullptr};

    template<typename Self>
    void operator()(Self& self,
                    sys::error_code err = sys::error_code(),
                    std::size_t n       = 0)
    {
        BOOST_ASIO_CORO_REENTER(coro_)
        {
            BOOST_ASIO_CORO_YIELD socket_.async_send_to(
                asio::buffer(ntp_req_message), ntp_server_ep, std::move(self));
            if (err) {
                std::printf("async_send_to() error: %s\n",
                            err.message().data());
                break;
            }

            BOOST_ASIO_CORO_YIELD socket_.async_receive_from(
                asio::buffer(buffer_.get(), 1024), local_ep_, std::move(self));
            if (err) {
                std::printf("async_receive_from() error: %s\n",
                            err.message().data());
            } else {
                std::printf("async_receive_from() ok: n=%ld\n", n);
                hexdump(buffer_.get(), n);
            }
            break;
        }

        self.complete(err);
    }
};
// #include <boost/asio/unyield.hpp>

template<typename CompletionToken>
auto async_ntp_query(udp::socket& socket, CompletionToken&& token)
{
    std::unique_ptr<uint8_t[]> buffer{new uint8_t[1024]};

    return asio::async_compose<CompletionToken, void(sys::error_code)>(
        async_ntp_query_impl{
            socket, asio::coroutine(), udp::endpoint(), std::move(buffer)},
        token,
        socket);
}

int main()
{
    asio::io_context ioctx;

    udp::socket socket(ioctx, udp::endpoint(udp::v4(), 0));
    async_ntp_query(socket, [](sys::error_code err) {
        if (err) {
            std::printf("async_ntp_query() error: %s\n", err.message().data());
        } else {
            // I don't known why this line print three times???
            std::printf("async_ntp_query() ok\n");
        }
    });

    ioctx.run();
    return 0;
}

问题根源 #

核心问题是self.complete(err);被放在了BOOST_ASIO_CORO_REENTER代码块的外部，导致协程每次resume后都会执行一次回调触发：

1. 第一次async_send_to完成后，协程resume并进入async_receive_from的yield分支，此时会执行外部的self.complete，第一次触发回调；
2. 第二次async_receive_from完成后，协程resume并处理结果，之后执行外部的self.complete，第二次触发回调；
3. 部分场景下，协程初始化时的额外调用也会触发一次self.complete，最终导致三次回调。

修复方案 #

把self.complete(err);移到BOOST_ASIO_CORO_REENTER块内部，确保只有当协程完全结束所有逻辑时才调用一次。修改后的operator()代码如下：

template<typename Self>
void operator()(Self& self,
                sys::error_code err = sys::error_code(),
                std::size_t n       = 0)
{
    BOOST_ASIO_CORO_REENTER(coro_)
    {
        BOOST_ASIO_CORO_YIELD socket_.async_send_to(
            asio::buffer(ntp_req_message), ntp_server_ep, std::move(self));
        if (err) {
            std::printf("async_send_to() error: %s\n",
                        err.message().data());
            self.complete(err);
            break;
        }

        BOOST_ASIO_CORO_YIELD socket_.async_receive_from(
            asio::buffer(buffer_.get(), 1024), local_ep_, std::move(self));
        if (err) {
            std::printf("async_receive_from() error: %s\n",
                        err.message().data());
        } else {
            std::printf("async_receive_from() ok: n=%ld\n", n);
            hexdump(buffer_.get(), n);
        }
        self.complete(err);
        break;
    }
}

额外优化建议 #

• 规范包含<boost/asio/yield.hpp>和<boost/asio/unyield.hpp>在协程结构的前后，避免潜在的编译或运行问题；
• 用std::vector<uint8_t>替代std::unique_ptr<uint8_t[]>作为缓冲区，代码更简洁且无需手动管理内存；
• 错误处理逻辑保持一致，确保所有错误路径都能正确传递错误码到回调。

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