c++ - 发送批量消息时 BOOST::async_wait 错误

Question

在我的应用程序中，我发送和接收大量消息并充当服务器。与此同时，我检查客户端的 session 超时。当我在我的应用程序中不使用 async_wait 时，我的程序运行良好。但是当我使用 async_wait 时，我的应用程序在发送和接收大约 100,000 - 200,000 条消息后崩溃，并出现以下错误。

"terminate called after throwing an instance of 'boost::exception_detail::
clone_impl<boost::exception_detail::error_info_injector<boost::system::system_error> >' what():  asio.ssl error" 

每当服务器收到来自客户端的任何消息时，它就会调用 start()。

   void Connection::start()
  {
   this->resetTimer();

   memset(_messageHeader.c_array(), 0, 8);
    boost::asio::async_read(_socket,
        boost::asio::buffer(_messageHeader, 8),
        _strand.wrap(
          boost::bind(
            &Connection::handleRead,
            shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred
            )
          )
        );
  }

   void Connection::handleRead(const boost::system::error_code& e,
      std::size_t bytesTransferred)
    {
     if (!e) {

      BOOST_STATIC_ASSERT( sizeof(MessageHeader) == 8 );

      Message message ;
      message.header.fromString(
          std::string(_messageHeader.data(),
            sizeof(MessageHeader)
            )
          );
      uint32_t type = message.header.type;
      uint32_t length = message.header.length;

      boost::asio::async_read(_socket,
          boost::asio::buffer(_buffer, message.header.length),
          _strand.wrap(
            boost::bind(
              &Connection::messageRead,
              shared_from_this(),
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred,
              type,
              length
              )
            )
          );
         }
    else {
      this->close();
    }
  }

void Connection::messageRead(const boost::system::error_code& e,
    std::size_t bytesTransferred, uint32_t type, uint32_t length)
{
  if (!e) {

    Message message;

    message.header.type = type;
    message.header.length = length;

    message.body = std::string(_buffer.data(), message.header.length);
    memset(_buffer.c_array(), 0, message.header.length);

    _timeoutCount = 0;

     this->start();
  }
  else {
    this->close();
  }
}


 void Connection::sendResponse(const Message& response)
 {
   std::string stringToSend(response.toString());

   boost::system::error_code error;
   boost::asio::ip::tcp::endpoint endpoint = socket().remote_endpoint(error);

   if ( error ) {
     this->close();
   }

   _strand.post(
       boost::bind(
         &CclConnection::writeImpl,
         shared_from_this(),
         stringToSend
         )
       );
 }


 void Connection::writeImpl(const std::string &message)
 {
   _outbox.push_back( message );
   if ( _outbox.size() > 1 ) {
     return;
   }
   this->write();
 }

 void Connection::write()
 {
   const std::string& message = _outbox[0];
   boost::asio::async_write(
       _socket,
       boost::asio::buffer( message.c_str(), message.size() ),
       _strand.wrap(
         boost::bind(
           &Connection::writeHandler,
           shared_from_this(),
           boost::asio::placeholders::error,
           boost::asio::placeholders::bytes_transferred
           )
         )
       );
 }

 void Connection::writeHandler(
     const boost::system::error_code& e,
     const size_t bytesTransferred
     )
 {
   _outbox.pop_front();

   if ( e ) {
     this->close();
   }
   if ( !_outbox.empty() ) {
     this->write();
   }
 }

 // close connection
 void Connection::close()
 {
   boost::system::error_code ignoredCode;
   socket().shutdown(boost::asio::ip::tcp::socket::shutdown_both,
       ignoredCode);
   }

 void Connection::resetTimer()
 {
   if (Context::getSessionTimeout() > 0)
   {
     // timer of 10 seconds
     _timer.expires_from_now(10);
     _timer.async_wait(
         _strand.wrap(
           boost::bind(&Connection::handleTimeout,
             shared_from_this(),
             boost::asio::placeholders::error)
           )
         );
   }
 }


 void Connection::handleTimeout(const boost::system::error_code &e)
 {
   if (!_loggedIn)
     return;

   if (!e)
   {
     boost::system::error_code ec;
     boost::asio::ip::tcp::endpoint endpoint = socket().remote_endpoint(ec);

     if (ec)
     {
       this->close();
       return;
     }
     else
     {
       _timeoutCount++;

       std::string ping = "ping";
       Message message(REQ_PING, ping);
       this->sendResponse(message);

       // no activity for quite some time now
       // so we just disconnect the session
       if (_timeoutCount > 30)
       {
         ERROR_1(TIMEOUT_OCCURED);
         this->close();
       }
       else
         this->start();
     }
   }
 }

Answer 1

您的 async_wait 处理程序每​​次通过调用 Connection::start()

调度 async_read

 else
 {
   _timeoutCount++;

   std::string ping = "ping";
   Message message(REQ_PING, ping);
   this->sendResponse(message);

   // no activity for quite some time now
   // so we just disconnect the session
   if (_timeoutCount > 30)
   {
     ERROR_1(TIMEOUT_OCCURED);
     this->close();
   }
   else
     this->start(); // this will perform an async_read
 }

这是错误的，我怀疑你的意图。你的程序是ill-formed当它在已经有一个未完成的套接字上调用后续 async_read 操作时。我建议从 async_wait 处理程序中重置计时器并再次调用 async_wait，而不是调用 Connection::start()。