嵌套 asio 调用以 __throw_bad_function_call() 结束

Question

我围绕 Boostless asio 编写了一个包装器来处理所有网络通信。大多数交换涉及在客户端和服务器之间来回发送的多个数据包。我现在有一个问题，如果我嵌套调用 asio 方法，我最终会进入方法 __throw_bad_function_call();

这是我的听众：

.hpp

/// \brief Socket object used to manage connections to the remote server.
    struct listener : public std::enable_shared_from_this<listener> {
      /// \brief Creates an instance of the object and opens a connection to the remote host.
      /// \param[in] params Connection parameters.
      /// \param[out] ec Errors returned by the OS.
      explicit listener(const parameters &params, std::error_code &ec) noexcept;

      /// \brief Executes all queued handlers and resets the io_context for another run.
      auto execute_handlers() noexcept -> void;

      /// \brief Writes data to the socket.
      /// \details Writes data to the socket and passes any errors to the callback.
      /// \param[in] barray Byte array to send to the server.
      /// \param[in] callback Method called when data has been written to the server.
      auto write(packet barray, std::function<void(std::error_code ec, std::size_t length)> callback) noexcept -> void;

      /// \brief Reads data from the socket.
      /// \details Reads that present in the socket and passes the data as a single byte array to the callback.
      /// \param[in] callback Method called when data has been written to the server.
      auto read(std::function<void(std::error_code ec, packet barray)> callback) noexcept -> void;

     private:
      /// \brief Underlying OS socket.
      std::unique_ptr<asio::generic::stream_protocol::socket> m_socket;

      asio::io_context m_context;
      asio::local::stream_protocol::endpoint m_endpoint;

      /// \brief Connection parameters
      parameters m_params;

      /// \brief DNS resolver
      asio::ip::tcp::resolver m_resolver;

      /// \brief Buffer that holds data to be received from the socket. Reset and resized before each call to read().
      packet m_buffer;
    };

.cpp

    listener::listener(const parameters &params, std::error_code &ec) noexcept
        : m_endpoint{params.domain_socket}, m_params{params}, m_resolver{m_context} {
#if !defined(_WIN32) && !defined(_WIN64)
      if (!m_params.domain_socket.empty()) {
        m_socket.reset(new asio::generic::stream_protocol::socket(m_context));
        m_socket->connect(m_endpoint, ec);
        if (ec) {
          return;
        }
        m_socket->non_blocking(true);
        return;
      }
#endif

      m_resolver.async_resolve(m_params.host, std::to_string(m_params.port),
                               [&](const std::error_code &error, asio::ip::tcp::resolver::results_type results) {
                                 if (error) {
                                   ec = error;
                                   return;
                                 }
                                 //! \todo Add timeout to async_connect
                                 for (const auto &endpoint : results) {
                                   m_socket.reset(new asio::generic::stream_protocol::socket(m_context));
                                   m_socket->async_connect(endpoint.endpoint(), [&](const std::error_code &err_c) {
                                     if (err_c) {
                                       ec = err_c;
                                       return;
                                     }
                                   });
                                 }
                               });
    }

    auto listener::execute_handlers() noexcept -> void {
      const auto handlers = m_context.run();
      m_context.restart();
    }

    auto listener::write(packet barray, std::function<void(std::error_code ec, std::size_t length)> callback) noexcept
        -> void {
      asio::async_write(*m_socket, asio::buffer(barray), callback);
    }

    auto listener::read(std::function<void(std::error_code ec, packet barray)> callback) noexcept -> void {
      packet tmp;
      tmp.resize(1);
      asio::async_read(*m_socket, asio::buffer(tmp), asio::transfer_exactly(1),
                       [&](std::error_code ec, std::size_t size) {
                         if (!ec) {
                           const auto available = m_socket->available();
                           m_buffer.resize(available);
                           m_buffer.shrink_to_fit();
                           asio::async_read(*m_socket, asio::buffer(m_buffer), asio::transfer_all());
                         }
                         callback(ec, std::move(m_buffer));
                       });
    }

然后我开始与远程服务器进行交换，但 lldb 告诉我我们失败了：

auto connection::do_connect(std::function<void(std::error_code ec, sql_state state)> callback) noexcept -> void {
m_listener = std::make_shared<listener>(m_params, ec);
      m_listener->execute_handlers();
      if (ec) {
        callback(ec, state);
      }
        sql_state state;
        const auto startup = write::startup(m_params);
        m_listener->write(startup, [&](std::error_code ec, std::size_t length) {
          if (ec) {
            callback(ec, state);
          }
          m_listener->read([&](std::error_code ec, packet packet){});
        });
        m_listener->execute_handlers();
      }

运行这在 lldb 中失败并显示以下跟踪：

Process 82736 launched: '/Users/ruihpacheco/Desktop/databaseclient/build_ninja/tests/integration/integration' (x86_64)
Process 82736 stopped
* thread #1, queue = 'com.apple.main-thread', stop reason = EXC_BAD_ACCESS (code=EXC_I386_GPFLT)
    frame #0: 0x0000000100178118 integration`std::__1::function<void (std::__1::error_code, std::__1::vector<nonstd::byte, std::__1::allocator<nonstd::byte> >)>::operator(this=0x00007ffeefbfd780, __arg=(__val_ = 0, __cat_ = 0x00007fffa6cd6cd8), __arg=size=329)(std::__1::error_code, std::__1::vector<nonstd::byte, std::__1::allocator<nonstd::byte> >) const at functional:1913
   1910 {
   1911     if (__f_ == 0)
   1912         __throw_bad_function_call();
-> 1913     return (*__f_)(_VSTD::forward<_ArgTypes>(__arg)...);
   1914 }
   1915 
   1916 #ifndef _LIBCPP_NO_RTTI
Target 0: (integration) stopped.

这感觉应该可行...

Answer 1

您的问题似乎是启动异步操作，该操作将缓冲区用于局部变量。

auto listener::write(packet barray, std::function<void(std::error_code ec, std::size_t length)> callback) noexcept
        -> void {
      asio::async_write(*m_socket, asio::buffer(barray), callback);
    }

barray 是 write 内部的局部变量，async_write returns 立即，但 asio::buffer() 只是 returns 传递数据的包装器（指针到数据及其大小），没有数据被复制。 write 结束，barray 被销毁并且 async_write 获得了删除数据的缓冲区..

同

auto listener::read(std::function<void(std::error_code ec, packet barray)> callback) noexcept -> void {
      packet tmp;
      tmp.resize(1);
      asio::async_read(*m_socket, asio::buffer(tmp), asio::transfer_exactly(1),
                       [&](std::error_code ec, std::size_t size) {
                         if (!ec) {
                           const auto available = m_socket->available();
                           m_buffer.resize(available);
                           m_buffer.shrink_to_fit();
                           asio::async_read(*m_socket, asio::buffer(m_buffer), asio::transfer_all());
                         }
                         callback(ec, std::move(m_buffer));
                       });

async_read 采用 buffer(tmp)，其中 tmp 是本地的。 async_read 也立即 returns，所以当调用处理程序时，tmp 不存在。

编辑

关于listener::read方法的一些话：

auto listener::read(
   std::function<void(std::error_code ec, packet barray)> callback) noexcept -> void  // [1]
{
  packet tmp;
  tmp.resize(1);
  asio::async_read(*m_socket, asio::buffer(tmp), asio::transfer_exactly(1),
           [&](std::error_code ec, std::size_t size) { // [2]
              if (!ec) {
                 const auto available = m_socket->available();
                 m_buffer.resize(available);
                 m_buffer.shrink_to_fit();
                 asio::async_read(*m_socket, asio::buffer(m_buffer), asio::transfer_all()); // [3]
               }
               callback(ec, std::move(m_buffer)); // [4]
         });
}

[1]回调传值

[2] lambda 通过引用捕获 all，因此没有制作 callback 的副本，闭包保留对局部变量的引用，在 [4] 中你正在调用闭包 callback(ec,std::move(m_buffer)) 是悬空的，因为 async_read returns 立即并且 listener::read 结束

[3] 和 [4] 很奇怪，async_read returns 马上，你在调用 ayns_read 时把 buffer 传给了 m_buffer, async_read returns（正在执行读取异步操作）并调用移动 m_buffer 的回调（它更改异步操作使用的对象！），您应该等到 [=21 的处理程序=] 被调用，然后你可以将 m_buffer 移动到回调

主要问题：在我看来，问题在于通过 callback by reference in listener::read方法。调用处理程序时，它会在悬空引用上调用 callback(..)。尝试按值传递 callback：

[&,callback /*pass by value*/](std::error_code ec, std::size_t size) { if (!ec) { const auto available = m_socket->available(); m_buffer.resize(available); m_buffer.shrink_to_fit(); asio::async_read(*m_socket, asio::buffer(m_buffer), asio::transfer_all()); } callback(ec, std::move(m_buffer)); });

嵌套 asio 调用以 __throw_bad_function_call() 结束

Nesting asio call ends in __throw_bad_function_call()

c++

callback

boost-asio

c++11