数据在 TCP/IP 中被 boost::asio::read_some 丢弃？

Question

我已经使用boost::asio 实现了一个 TCP 服务器。此服务器使用basic_stream_socket::read_some 函数读取数据。我知道read_some 不保证提供的缓冲区在返回之前是满的。

在我的项目中，我发送由分隔符分隔的字符串（如果重要的话）。在客户端，我使用WinSock::send() 函数发送数据。现在我的问题出在服务器端，我无法获取从客户端发送的所有字符串。我怀疑read_some 正在接收一些数据并出于某种原因丢弃剩余数据。比在下一次调用中再次接收另一个字符串。

在 TCP/IP 中真的可以吗？

我尝试使用async_receive，但这占用了我所有的 CPU，而且由于必须通过回调函数清理缓冲区，这会导致我的程序出现严重的内存泄漏。 （我使用IoService::poll() 来调用处理程序。与async_read() 的调用率相比，该处理程序的调用速度非常慢）。

我再次尝试使用免费函数read，但这并不能解决我的目的，因为它会因我提供的缓冲区大小而阻塞太多时间。

我之前的服务器实现是使用WinSock API，我可以使用WinSock::recv() 接收所有数据。 请给我一些线索，以便我可以使用boost::asio 接收完整的数据。

这是我的服务器端线程循环

void 
TCPObject::receive()
{
    if (!_asyncModeEnabled)
    {
        std::string recvString;
        if ( !_tcpSocket->receiveData( _maxBufferSize, recvString ) )
        {
            LOG_ERROR("Error Occurred while receiving data on socket.");
        }
        else
            _parseAndPopulateQueue ( recvString );
    }
    else
    {
        if ( !_tcpSocket->receiveDataAsync( _maxBufferSize ) )
        {
            LOG_ERROR("Error Occurred while receiving data on socket.");
        }
    }
}

TCPSocket 中的receiveData()

bool 
TCPSocket::receiveData( unsigned int bufferSize, std::string& dataString )
{
   boost::system::error_code error;
   char *buf = new char[bufferSize + 1];
   size_t len = _tcpSocket->read_some( boost::asio::buffer((void*)buf, bufferSize),        error);
   if(error)
   {
       LOG_ERROR("Error in receiving data.");
       LOG_ERROR( error.message() );
       _tcpSocket->close();
       delete [] buf; 
       return false;
   }
   buf[len] ='\0';
   dataString.insert( 0, buf );
   delete [] buf;
   return true;
}

TCP Socket 中的receiveDataAsync

bool 
TCPSocket::receiveDataAsync( unsigned int bufferSize )
{
    char *buf = new char[bufferSize + 1];

    try
    {
        _tcpSocket->async_read_some( boost::asio::buffer( (void*)buf, bufferSize ), 
                                     boost::bind(&TCPSocket::_handleAsyncReceive, 
                                                    this,
                                                    buf,
                                                    boost::asio::placeholders::error,
                                                    boost::asio::placeholders::bytes_transferred) );
        //! Asks io_service to execute callback
        _ioService->poll();
    }
    catch (std::exception& e)
    {
        LOG_ERROR("Error Receiving Data Asynchronously");
        LOG_ERROR( e.what() );
        delete [] buf;
        return false;
    }

    //we dont delete buf here as it will be deleted by callback _handleAsyncReceive
    return true;
}

异步接收处理程序

void 
TCPSocket::_handleAsyncReceive(char *buf, const boost::system::error_code& ec, size_t size)
{
    if(ec)
    {
        LOG_ERROR ("Error occurred while sending data Asynchronously.");
        LOG_ERROR ( ec.message() );
    }
    else if ( size > 0 )
    {
        buf[size] = '\0';
        emit _asyncDataReceivedSignal( QString::fromLocal8Bit( buf ) );
    }
    delete [] buf;
}

客户端 sendData 函数。

sendData(std::string data)
{
    if(!_connected)
    {
        return;
    }

    const char *pBuffer = data.c_str();

    int bytes = data.length() + 1;

    int i = 0,j;
    while (i < bytes)
    {
        j = send(_connectSocket, pBuffer+i, bytes-i, 0);

        if(j == SOCKET_ERROR)
        {
            _connected = false;
            if(!_bNetworkErrNotified)
            {
                _bNetworkErrNotified=true;
                emit networkErrorSignal(j);
            }
            LOG_ERROR( "Unable to send Network Packet" );
            break;
        }
        i += j;
    }
}

Answer 1

Boost.Asio 的 TCP 功能使用得很好，所以我会犹豫是否怀疑它是问题的根源。在大多数数据丢失的情况下，问题是由应用程序代码造成的。

在这种情况下，接收器代码存在问题。发件人使用\0 分隔字符串。但是，在单个读取操作中读取多个字符串的情况下，接收器无法正确处理分隔符，因为string::insert() 会导致char* 在到达第一个分隔符时被截断。

例如，发送者写入两个字符串"Test string\0" 和"Another test string\0"。在TCPSocket::receiveData() 中，接收者将"Test string\0Another test string\0" 读入buf。然后dataString 填充有dataString.insert(0, buf)。这个特定的重载将复制到分隔符，因此dataString 将包含"Test string"。要解决此问题，请考虑使用 string::insert() 重载，该重载采用要插入的字符数：dataString.insert(0, buf, len)。

Answer 2

我之前没有使用过投票功能。我所做的是创建一个工作线程，该线程专用于使用 run 函数处理 ASIO 处理程序，该函数会阻塞。 Boost 文档说每个可用于处理异步事件处理程序的线程必须首先调用 io_service:run 或 io_service:poll 方法。我不确定您对调用 poll 的线程还做了什么。

因此，我建议至少为异步 ASIO 事件处理程序分配一个工作线程，并使用 run 而不是 poll。如果您希望该工作线程继续处理所有异步消息而不返回和退出，则将工作对象添加到 io_service 对象。有关示例，请参阅此 link。