使用协程时 boost::asio 超时的最佳实践

Question

下面的连接尝试会（在我的网络配置上）造成 2 分钟的延迟，因为目标不存在并且地址与我的机器位于不同的子网上。所以我添加了超时逻辑以将尝试限制为 5 秒：

#define BOOST_ASIO_HAS_CO_AWAIT
#define BOOST_ASIO_HAS_STD_COROUTINE

#include <iostream>
#include <chrono>
#include <thread>

#include <boost/asio/awaitable.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/address_v4.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/redirect_error.hpp>

namespace this_coro = boost::asio::this_coro;
using namespace std::chrono_literals;

boost::asio::awaitable<void> fail_to_connect()
{
    auto executor = co_await this_coro::executor;
    auto socket = boost::asio::ip::tcp::socket{executor};
    auto ep = boost::asio::ip::tcp::endpoint{
                    boost::asio::ip::make_address_v4("192.168.1.52"),
                    80};

    auto timer = boost::asio::steady_timer{executor};
    timer.expires_after(5s);
    boost::asio::co_spawn(
        executor,
        [&]() -> boost::asio::awaitable<void> {
            auto ec = boost::system::error_code{};
            co_await timer.async_wait(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
            std::cout << "Thread ID: " << std::this_thread::get_id()
                      << " Timer: " << ec.message() << std::endl;
            if (!ec) {
                socket.close();
            }
        },
        boost::asio::detached
    );

    std::cout << "Thread ID: " << std::this_thread::get_id()
              << " Starting connection" << std::endl;
    co_await boost::asio::async_connect(socket,
                                        std::array{std::move(ep)},
                                        boost::asio::use_awaitable);
    timer.cancel();
}

int main()
{
    auto ctx = boost::asio::io_context{};
    auto guard = boost::asio::make_work_guard(ctx.get_executor());

    auto exception_handler = [&](auto e_ptr) {
        if (e_ptr) {
            std::rethrow_exception(e_ptr);
        }
    };

    boost::asio::co_spawn(ctx, fail_to_connect, std::move(exception_handler));
    ctx.run();
}

这按预期工作。从线程 ID 可以看出，在两个协程之间使用相同的执行上下文意味着我也不会同时访问套接字。

14:58:41: Starting /home/cmannett85/workspace/build-scratch-Desktop-Debug/scratch ...
Thread ID: 140171855615808 Starting connection
Thread ID: 140171855615808 Timer: Success
terminate called after throwing an instance of 'boost::system::system_error'
  what():  Operation canceled
14:58:46: The program has unexpectedly finished.

然而，这感觉很笨拙，尤其是与非协程回调样式相比。使用协程是否有更好的超时方法？我正在努力寻找示例。

Answer 1

一个建议是将样板提升到一个类中。

基本上只是一个定时器的包装器，可以通过更简单的界面设置和取消回调：

#include "asio.hpp"

class timeout {
  asio::high_resolution_timer timer_;
  using duration = asio::high_resolution_timer::duration;

 public:
  template <typename Executor>
  timeout(Executor&& executor) : timer_(executor){};

  template <typename Executor, typename Func>
  timeout(Executor&& executor, const duration& timeout, Func&& on_timeout)
      : timer_(executor) {
    set(timeout, on_timeout);
  }

  template <typename Func>
  void set(const duration& timeout, Func&& on_timeout) {
    timer_.expires_after(timeout);  // cancels outstanding timeouts
    timer_.async_wait([&](std::error_code ec) {
      if (ec) return;
      std::cout << "cancelling\n";
      on_timeout();
    });
  }

  void cancel() { timer_.cancel(); }

  ~timeout() { timer_.cancel(); }
};

可以这样使用：

tmo.set(1s, [&]() { socket.close(); });
std::size_t n = co_await asio::async_read_until(socket, incoming, '\n',
                                                asio::use_awaitable);

asio::const_buffer response(asio::buffers_begin(incoming.data()),
                            asio::buffers_begin(incoming.data()) + n);

tmo.set(1s, [&]() { socket.close(); });
co_await asio::async_write(socket, response, asio::use_awaitable);

incoming.consume(n);

它通常可以与任何支持真正取消的对象一起使用，这有点稀疏支持。例如，取消套接字上正在进行的读取或写入的唯一真正方法是调用close()。

例如，取消async_resolve 是不可能的，因为cancel() 只取消挂起的操作，而不是正在进行的操作。换句话说，这并没有按预期工作：

// does not actually cancel anything :(
tmo.set(1s, [&]() { resolver.cancel(); });
resolver.async_resolve(...);

这是一个完整的 echo 客户端示例，如果其中任何一个时间超过 1 秒，则该客户端会停止：

• async_connect
• async_read_until
• async_write

如前所述，async_resolve 无法取消，因此为其编写超时没有多大意义。也许向协程的执行上下文发布异常会起作用？

#include <chrono>
#include <iostream>
#include <string>
#include <string_view>

// clang-format off
#ifdef _WIN32
#include "sdkddkver.h"
#endif
#include "asio.hpp"
// clang-format on

using asio::ip::tcp;

class timeout {
  asio::high_resolution_timer timer_;
  using duration = asio::high_resolution_timer::duration;

 public:
  template <typename Executor>
  timeout(Executor&& executor) : timer_(executor){};

  template <typename Executor, typename Func>
  timeout(Executor&& executor, const duration& timeout, Func&& on_timeout)
      : timer_(executor) {
    set(timeout, on_timeout);
  }

  template <typename Func>
  void set(const duration& timeout, Func&& on_timeout) {
    timer_.expires_after(timeout);  // cancels outstanding timeouts
    timer_.async_wait([&](std::error_code ec) {
      if (ec) return;
      std::cout << "cancelling\n";
      on_timeout();
    });
  }

  void cancel() { timer_.cancel(); }

  ~timeout() { timer_.cancel(); }
};

asio::awaitable<void> echo_client(tcp::socket& socket, std::string_view host,
                                  std::string_view service) {
  try {
    using namespace std::chrono_literals;
    auto exec = socket.get_executor();

    tcp::resolver resolver(exec);
    timeout tmo(exec);

    std::cout << "resolving...\n";
    tmo.set(1s, [&]() { resolver.cancel(); });
    auto endpoints =
        co_await resolver.async_resolve(host, service, asio::use_awaitable);

    std::cout << "connecting...\n";
    tmo.set(1s, [&]() { socket.close(); });
    co_await asio::async_connect(socket, endpoints, asio::use_awaitable);

    asio::streambuf incoming;

    for (;;) {
      std::cout << "reading... ";

      tmo.set(1s, [&]() { socket.close(); });
      std::size_t n = co_await asio::async_read_until(socket, incoming, '\n',
                                                      asio::use_awaitable);

      std::string response(asio::buffers_begin(incoming.data()),
                           asio::buffers_begin(incoming.data()) + n);

      std::cout << response;

      tmo.set(1s, [&]() { socket.close(); });
      co_await asio::async_write(socket, asio::buffer(response),
                                 asio::use_awaitable);

      incoming.consume(n);
    }
  } catch (std::exception& e) {
    std::cerr << e.what() << "\n";
  }
}

int main(int argc, char** argv) {
  if (argc < 3) {
    std::cout << "usage: " << argv[0] << " [host] [port]\n";
    return 1;
  }

  asio::io_context io_context;
  tcp::socket socket(io_context);

  asio::co_spawn(io_context, echo_client(socket, argv[1], argv[2]),
                 asio::detached);

  io_context.run();

  return 0;
}