sockets - 我可以使用 Boost.Asio 在多线程 I/O 服务超时的情况下同步读取套接字吗？

Question

我有一个使用 Boost.Asio 进行 TCP 和 UDP 套接字通信的应用程序。我知道“Asio”中的“A”代表异步，因此该库倾向于鼓励您尽可能使用异步 I/O。在一些情况下，同步套接字读取更可取。但是，与此同时，我想为上述接收调用设置超时，因此不可能无限期地阻塞读取。

这在 Boost.Asio 用户中似乎是一个非常普遍的问题，关于该主题的过去 Stack Overflow 问题如下:

C++ Boost ASIO: how to read/write with a timeout?

asio::read with timeout

boost asio timeout

How to set a timeout on blocking sockets in boost asio?

甚至可能还有更多。甚至还有 examples in the documentation 用于如何实现带超时的同步操作。它们归结为将同步操作转换为异步操作，然后以 asio::deadline_timer 并行启动它。定时器的到期处理程序可以在超时到期的情况下取消异步读取。这看起来像这样(摘自上述链接示例的片段):

std::size_t receive(const boost::asio::mutable_buffer& buffer,
      boost::posix_time::time_duration timeout, boost::system::error_code& ec)
  {
    // Set a deadline for the asynchronous operation.
    deadline_.expires_from_now(timeout);

    // Set up the variables that receive the result of the asynchronous
    // operation. The error code is set to would_block to signal that the
    // operation is incomplete. Asio guarantees that its asynchronous
    // operations will never fail with would_block, so any other value in
    // ec indicates completion.
    ec = boost::asio::error::would_block;
    std::size_t length = 0;

    // Start the asynchronous operation itself. The handle_receive function
    // used as a callback will update the ec and length variables.
    socket_.async_receive(boost::asio::buffer(buffer),
        boost::bind(&client::handle_receive, _1, _2, &ec, &length));

    // Block until the asynchronous operation has completed.
    do io_service_.run_one(); while (ec == boost::asio::error::would_block);

    return length;
  }

这实际上是一个相对干净的解决方案:启动异步操作，然后手动轮询 asio::io_service 以一次执行一个异步处理程序，直到 async_receive() 完成或计时器到期。

但是，如果套接字的底层 I/O 服务已经在一个或多个后台线程中运行呢？在这种情况下，无法保证异步操作的处理程序将由上述代码段中的前台线程运行，因此 run_one() 直到稍后可能不相关的处理程序执行才会返回。这将使套接字读取相当无响应。
asio::io_service 有一个 poll_one() 函数，它会在不阻塞的情况下检查服务的队列，但我没有看到在处理程序执行之前阻塞前台线程(模拟同步调用行为)的好方法，除了没有后台线程的情况已经在执行 asio::io_service::run() 。

我看到了两种可能的解决方案，但我都不喜欢:

在启动异步操作后，使用条件变量或类似结构使前台线程阻塞。在 async_receive() 调用的处理程序中，向条件变量发出信号以解除线程阻塞。这会导致每次读取的一些锁定，我想避免这种情况，因为我想在 UDP 套接字读取上实现最大可能的吞吐量。否则，它是可行的，除非有更好的方法出现，否则我可能会这样做。

确保套接字有自己的 asio::io_service 没有被任何后台线程运行。这使得在需要的情况下更难将异步 I/O 与套接字一起使用。

关于以安全方式完成此任务的其他方法的任何想法？

旁白: 之前的 SO 问题有一些答案主张使用 SO_RCVTIMEO 套接字选项来实现套接字读取超时。这在理论上听起来不错，但至少在我的平台上似乎不起作用(Ubuntu 12.04，Boost v1.55)。我可以设置套接字超时，但它不会给 Asio 带来预期的效果。相关代码在 /boost/asio/detail/impl/socket_ops.ipp 中:

size_t sync_recvfrom(socket_type s, state_type state, buf* bufs,
    size_t count, int flags, socket_addr_type* addr,
    std::size_t* addrlen, boost::system::error_code& ec)
{
  if (s == invalid_socket)
  {
    ec = boost::asio::error::bad_descriptor;
    return 0;
  }

  // Read some data.
  for (;;)
  {
    // Try to complete the operation without blocking.
    signed_size_type bytes = socket_ops::recvfrom(
        s, bufs, count, flags, addr, addrlen, ec);

    // Check if operation succeeded.
    if (bytes >= 0)
      return bytes;

    // Operation failed.
    if ((state & user_set_non_blocking)
        || (ec != boost::asio::error::would_block
          && ec != boost::asio::error::try_again))
      return 0;

    // Wait for socket to become ready.
    if (socket_ops::poll_read(s, 0, ec) < 0)
      return 0;
  }
}

如果套接字读取超时，上面对 recvfrom() 的调用将返回 EAGAIN 或 EWOULDBLOCK ，它们被转换为 boost::asio::error::try_again 或 boost::asio::error::would_block 。在这种情况下，上面的代码将调用 poll_read() 函数，对于我的平台，它看起来像:

int poll_read(socket_type s, state_type state, boost::system::error_code& ec)
{
  if (s == invalid_socket)
  {
    ec = boost::asio::error::bad_descriptor;
    return socket_error_retval;
  }

  pollfd fds;
  fds.fd = s;
  fds.events = POLLIN;
  fds.revents = 0;
  int timeout = (state & user_set_non_blocking) ? 0 : -1;
  clear_last_error();
  int result = error_wrapper(::poll(&fds, 1, timeout), ec);
  if (result == 0)
    ec = (state & user_set_non_blocking)
      ? boost::asio::error::would_block : boost::system::error_code();
  else if (result > 0)
    ec = boost::system::error_code();
  return result;
}

我剪掉了有条件地为其他平台编译的代码。如您所见，如果套接字不是非阻塞套接字，它最终会在无限超时的情况下调用 poll()，因此阻塞直到套接字有数据要读取(并在超时时阻止尝试)。因此， SO_RCVTIMEO 选项无效。

Answer 1

Boost.Asio 对 futures 的支持可能会提供一个优雅的解决方案。当提供异步操作时， boost::asio::use_future value 作为其完成处理程序，启动函数将返回 std::future将接收操作结果的对象。此外，如果操作以失败告终，error_code转换为 system_error并通过 future 传递给调用者.

在 Boost.Asio C++11 Futures datytime client example ，一个专用线程运行 io_service ，主线程发起异步操作，然后同步等待操作完成，如下:

std::array<char, 128> recv_buf;
udp::endpoint sender_endpoint;
std::future<std::size_t> recv_length =
  socket.async_receive_from(
      boost::asio::buffer(recv_buf),
      sender_endpoint,
      boost::asio::use_future);

// Do other things here while the receive completes.

std::cout.write(
    recv_buf.data(),
    recv_length.get()); // Blocks until receive is complete.

使用时 future s，实现带超时的同步读取的总体方法与以前相同。不使用同步读取，而是使用异步读取并异步等待计时器。唯一的小变化是，而不是在 io_service 上阻塞或者定期检查谓词，可以调用 future::get()阻塞直到操作以成功或失败(例如超时)完成。

如果 C++11 不可用，则可以为 Boost.Thread 的 future 自定义异步操作的返回类型，如 this 中所示回答。