Fix internal connection issue under high transmit rate (#1091)

Author: starwarfan

source/core/owt_base/internal/TransportBase.cpp

@@ -74,6 +74,8 @@ uint32_t TransportMessage::fillData(const uint8_t* data, uint32_t length)
     }
     if (data) {
         memcpy(m_buffer.get() + m_receivedBytes, data, length);
+    } else {
+        length = 0;
     }
     m_receivedBytes += length;
     return length;
@@ -157,10 +159,19 @@ void TransportSession::sendData(TransportData data)
         return;
     }
     auto self(shared_from_this());
-    m_service->post(boost::bind(&TransportSession::sendHandler, self, data));
+    m_service->post(boost::bind(&TransportSession::prepareSend, self, data));
+}
+
+void TransportSession::prepareSend(TransportData data)
+{
+    // Only access m_sendQueue in IO service thread.
+    m_sendQueue.push(data);
+    if (m_sendQueue.size() == 1) {
+        sendHandler();
+    }
 }
 
-void TransportSession::sendHandler(TransportData data)
+void TransportSession::sendHandler()
 {
     if (m_isClosed) {
         return;
@@ -169,7 +180,11 @@ void TransportSession::sendHandler(TransportData data)
         ELOG_WARN("sendHandler: socket is not open");
         return;
     }
+    if (m_sendQueue.empty()) {
+        return;
+    }
 
+    TransportData data = m_sendQueue.front();
     TransportMessage toSend(data.buffer.get(), data.length);
     TransportData wrappedData{toSend.messageData(),
                               toSend.messageLength()};
@@ -197,13 +212,18 @@ void TransportSession::writeHandler(
     const boost::system::error_code& ec,
     std::size_t bytes)
 {
+    assert(m_sendQueue.size() > 0);
+    m_sendQueue.pop();
     if (ec) {
         ELOG_DEBUG("Error writing data: %s", ec.message().c_str());
         if (!m_isClosed) {
             m_isClosed = true;
             // Notify the listener about the socket error if the listener is not closing me.
             m_listener->onClose(m_id);
         }
+    } else {
+        ELOG_DEBUG("Wrote data: %zu", bytes);
+        sendHandler();
     }
 }
 
@@ -254,6 +274,7 @@ void TransportSession::receiveData()
 
     uint32_t bytesToRead = m_receivedMessage.missingBytes();
     assert(bytesToRead > 0);
+
     if (bytesToRead > m_receivedBufferSize) {
         // Double the received buffer size
         m_receivedBufferSize = std::max(m_receivedBufferSize * 2, bytesToRead);
@@ -286,7 +307,10 @@ void TransportSession::readHandler(
     if (!ec || ec.value() == boost::asio::error::message_size) {
         uint32_t bytesToRead = m_receivedMessage.missingBytes();
         assert(bytesToRead >= bytes);
-        m_receivedMessage.fillData(m_receivedBuffer.get(), bytes);
+        uint32_t filled = m_receivedMessage.fillData(m_receivedBuffer.get(), bytes);
+        if (filled != bytes) {
+            ELOG_WARN("Message fill length %u, %zu\n", filled, bytes);
+        }
         receiveData();
     } else {
         if (ec.value() != boost::system::errc::operation_canceled &&

source/core/owt_base/internal/TransportBase.h

@@ -15,6 +15,7 @@
 
 #include "IOService.h"
 #include <memory>
+#include <queue>
 
 namespace owt_base {
 
@@ -102,17 +103,18 @@ class TransportSession
 
 private:
     void receiveData();
-    void sendHandler(TransportData data);
-    void writeHandler(const boost::system::error_code&, std::size_t);
     void readHandler(const boost::system::error_code&, std::size_t);
-
+    void prepareSend(TransportData data);
+    void sendHandler();
+    void writeHandler(const boost::system::error_code&, std::size_t);
 
     uint32_t m_id;
     std::shared_ptr<IOService> m_service;
     boost::asio::ip::tcp::socket m_socket;
     std::shared_ptr<SSLSocket> m_sslSocket;
     TransportMessage m_receivedMessage;
     boost::shared_array<uint8_t> m_receivedBuffer;
+    std::queue<TransportData> m_sendQueue;
     uint32_t m_receivedBufferSize;
     bool m_isClosed;
     Listener* m_listener;