如何等待函数返回 Boost:::Asio？

Question

背景

我刚开始使用 Boost::Asio 库，无法获得我想要的行为。我正在尝试为自定义硬件解决方案实施一些网络通信。我们使用的通信协议栈严重依赖 Boost::Asio 异步方法，我不认为它是完全线程安全的。

我已成功实现发送，但在尝试设置等待接收时遇到问题。我发现大多数boost::asio examples 都依赖套接字行为来使用socket_.async_read_some() 或其他类似功能实现异步等待。然而这对我们不起作用，因为我们的硬件解决方案需要直接调用驱动程序函数而不是使用套接字。

应用程序使用io_service 传递给boost::asio::generic::raw_protocol::socket 以及其他类。

使用套接字的协议栈示例代码

这是来自协议栈的示例代码。在 RawSocketLink 的构造函数中调用了do_receive()。

void RawSocketLink::do_receive()
{
    namespace sph = std::placeholders;
    socket_.async_receive_from(
            boost::asio::buffer(receive_buffer_), receive_endpoint_,
            std::bind(&RawSocketLink::on_read, this, sph::_1, sph::_2));
}

void RawSocketLink::on_read(const boost::system::error_code& ec, std::size_t read_bytes)
{
    if (!ec) {
        // Do something with received data...
        do_receive();
    }
}

我们之前没有协议栈的接收代码

在实现堆栈之前，我们一直在使用线程库为发送和接收创建单独的线程。接收方法如下所示。大多数情况下，它依赖于从硬件驱动程序调用receive_data() 函数并等待它返回。这是一个阻塞调用，但需要返回数据。

void NetworkAdapter::Receive() {

  uint8_t temp_rx_buffer[2048];
  rc_t rc;
  socket_t *socket_ptr;
  receive_params_t rx_params;
  size_t rx_buffer_size;
  char str[100];

  socket_ptr = network_if[0];

  while (1) {
    rx_buffer_size = sizeof(temp_rx_buffer);
    // Wait until receive_data returns then process
    rc = receive_data(socket_ptr,
                     temp_rx_buffer,
                     &rx_buffer_size,
                     &rx_params,
                     WAIT_FOREVER);
    if (rc_error(rc)) {
      (void)fprintf(stderr, "Receive failed");
      continue;
    }
    
    // Do something with received packet ....
    
  }

  return;
}

请注意，此代码中的 socket_t 指针不与 Boost::Asio 的 TCP/UDP 套接字相同。

异步接收的当前实现

这是我当前的代码，我需要帮助。我不确定如何使用 boost::asio 方法等待 receive_data 返回。我们正在尝试复制socket.async_read_from() 的行为。 NetworkAdapter 可以访问io_service。

void NetworkAdapter::do_receive() {
  
  rc_t rc;
  socket_t *socket_ptr;
  receive_params_t rx_params;
  size_t rx_buffer_size;

  socket_ptr = network_if[0];

  rx_buffer_size = receive_buffer_.size();
  
  // What do I put here to await for this to return asynchronously?
  rc = receive_data(socket_ptr, receive_buffer_.data(), &rx_buffer_size, &rx_params, ATLK_WAIT_FOREVER);
  on_read(rc, rx_buffer_size, rx_params);
}

void NetworkAdapter::on_read(const rc_t &rc, std::size_t read_bytes, const receive_params_t &rx_params) {
  if (!rc) {

    // Do something with received data...

  } else {
    LOG(ERROR) << "Packet receieve failure";
  }
  do_receive();
}

总结

如何使用 boost::asio async/await 函数来等待函数返回？特别是我想复制socket.async_receive_from() 的行为，但使用函数而不是套接字。

---

^{*由于数据保护要求，某些函数名称和类型已更改。}

Answer 1

用于异步操作的 N4045 库基础，修订版 2
http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2014/n4045.pdf

第 24 页上有一个示例，说明如何根据基于回调的 os API 实现 asio 异步 API。

// the async version of your operation, implementing all kinds of async paradigm in terms of callback async paradigm
template <class CompletionToken>
auto async_my_operation(/* any parameters needed by the sync version of your operation */, CompletionToken&& token) 
{
  // if CompletionToken is a callback function object, async_my_operation returns void, the callback's signature should be void(/* return type of the sync version of your operation, */error_code)
  // if CompletionToken is boost::asio::use_future, async_my_operation returns future</* return type of the sync version of your operation */>
  // if CompletionToken is ..., ...

  // you are not inventing new async paradigms so you don't have to specialize async_completion or handler_type, you should focus on implement the os_api below
  async_completion<CompletionToken, void(/* return type of the sync version of your operation, */error_code)/* signature of callback in the callback case */> completion(token); 
  typedef handler_type_t<CompletionToken, void(error_code)> Handler; 
  unique_ptr<wait_op<Handler>> op(new wait_op<Handler>(move(completion.handler))); // async_my_operation initates your async operation and exits, so you have to store completion.handler on the heap, the completion.handler will be invoked later on a thread pool (e.g. threads blocked in IOCP if you are using os api, threads in io_context::run() if you are using asio (sockets accept an io_context during construction, so they know to use which io_context to run completion.handler))
  
  // most os api accepts a void* and a void(*)(result_t, void*) as its C callback function, this is type erasure: the void* points to (some struct that at least contains) the C++ callback function object (can be any type you want), the void(*)(result_t, void*) points to a C callback function to cast the void* to a pointer to C++ callback function object and call it
  os_api(/* arguments, at least including:*/ op.get(), &wait_callback<Handler>);

  return completion.result.get();
}

// store the handler on the heap
template <class Handler>
struct wait_op {
  Handler handler_;
  explicit wait_op(Handler  handler) : handler_(move(handler)) {}
};

// os post a message into your process's message queue, you have several threads blocking in a os api (such as IOCP) or asio api (such as io_context::run()) that continuously takes message out from the queue and then call the C callback function, the C callback function calls your C++ callback function
template <class Handler> 
void wait_callback(result_t result, void* param) 
{
  unique_ptr<wait_op<Handler>> op(static_cast<wait_op<Handler>*>(param));
  op‐>handler_(/* turn raw result into C++ classes before passing it to C++ code */, error_code{});
}

//trivial implementation, you should consult the socket object to get the io_context it uses
void os_api(/* arguments needed by your operation */, void* p_callback_data, void(*p_callback_function)(result_t, void*))
{
  std::thread([](){
    get the result, blocks
    the_io_context_of_the_socket_object.post([](){ (*p_callback_function)(result, p_callback_data); });
  }).detach();
}

boost.asio 已经从async_completion 和handler_type 更改为async_result，所以上面的代码已经过时了。

对异步操作的要求 - 1.75.0 https://www.boost.org/doc/libs/1_75_0/doc/html/boost_asio/reference/asynchronous_operations.html