WebSocket sends: no copy into a frame, encoded buffers taken rather than copied

components/RaftWebServer/RaftWebConnDefs.h

@@ -10,6 +10,7 @@
 
 #include <stdint.h>
 #include <functional>
+#include "SpiramAwareAllocator.h"
 
 // Callback function for any endpoint
 enum RaftWebConnSendRetVal
@@ -44,3 +45,7 @@ typedef std::function<RaftWebConnSendRetVal()> RaftWebConnReadyToSendFn;
 // Function to send on a connection
 typedef std::function<RaftWebConnSendRetVal(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxSendRetryMs)> RaftWebConnSendFn;
 
+// Function to send a buffer on a connection, which may take the buffer's contents (leaving it empty) rather than
+// copy what it can't send at once
+typedef std::function<RaftWebConnSendRetVal(SpiramAwareUint8Vector& buf, uint32_t maxSendRetryMs)> RaftWebConnSendTakeFn;
+

components/RaftWebServer/RaftWebConnManager.cpp

@@ -522,6 +522,34 @@ bool RaftWebConnManager::sendBufOnChannel(const uint8_t* pBuf, uint32_t bufLen,
     return sendOk;
 }
 
+/////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+// Send a buffer on a channel, which may be taken rather than copied
+/////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+bool RaftWebConnManager::sendBufOnChannel(SpiramAwareUint8Vector& buf, uint32_t channelID)
+{
+    // Main task only (the send path is not protected by locks)
+    RAFT_CHECK_MAIN_TASK(MODULE_PREFIX, "sendBufOnChannel");
+
+    // The buffer can only be taken by the responder if it is the only one on the channel
+    RaftWebResponder* pChannelResponder = nullptr;
+    uint32_t numOnChannel = 0;
+    for (uint32_t i = 0; i < _webConnections.size(); i++)
+    {
+        if (!_webConnections[i].isActive())
+            continue;
+        RaftWebResponder *pResponder = _webConnections[i].getResponder();
+        uint32_t usedChannelID = 0;
+        if (!pResponder || !pResponder->getChannelID(usedChannelID) || (usedChannelID != channelID))
+            continue;
+        pChannelResponder = pResponder;
+        numOnChannel++;
+    }
+    if (numOnChannel != 1)
+        return sendBufOnChannel(buf.data(), buf.size(), channelID);
+    return pChannelResponder->encodeAndSendData(buf);
+}
+
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 // Send to all server-side events
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////

components/RaftWebServer/RaftWebConnManager.h

@@ -76,6 +76,9 @@ class RaftWebConnManager
     // Send a buffer on a channel (main task only)
     bool sendBufOnChannel(const uint8_t* pBuf, uint32_t bufLen, uint32_t channelID);
 
+    // Send a buffer on a channel, which may be taken (leaving it empty) rather than copied
+    bool sendBufOnChannel(SpiramAwareUint8Vector& buf, uint32_t channelID);
+
     // Send to all server-side events (main task only)
     void serverSideEventsSendMsg(const char* eventContent, const char* eventGroup);
 

components/RaftWebServer/RaftWebConnection.cpp

@@ -209,12 +209,12 @@ void RaftWebConnection::clear()
     _debugDataRxCount = 0;
     _maxSendBufferBytes = 0;
 #ifdef DEBUG_RESPONDER_CLEAR
-    if (_socketTxQueuedBuffer.size() > 0)
+    if (txQueuedLen() > 0)
     {
-        LOG_I(MODULE_PREFIX, "clear() clearing _socketTxQueuedBuffer with %d bytes", _socketTxQueuedBuffer.size());
+        LOG_I(MODULE_PREFIX, "clear() clearing _socketTxQueuedBuffer with %d bytes", txQueuedLen());
     }
 #endif
-    _socketTxQueuedBuffer.clear();
+    txQueueClear();
     _header.clear();
 }
 
@@ -657,6 +657,8 @@ bool RaftWebConnection::serviceConnHeader(const uint8_t* pRxData, uint32_t dataL
                 std::bind(&RaftWebConnection::canSendOnConn, this),
                 std::bind(&RaftWebConnection::rawSendOnConn, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3),
                 _pClientConn->getClientId());
+    params.setWebConnRawSendTake(
+                std::bind(&RaftWebConnection::rawSendOnConnTake, this, std::placeholders::_1, std::placeholders::_2));
     _pResponder = _pConnManager->getNewResponder(_header, params, statusCode);
 #ifdef DEBUG_RESPONDER_CREATE_DELETE
     if (_pResponder) 
@@ -811,7 +813,7 @@ bool RaftWebConnection::responderHandleData(const uint8_t* pRxData, uint32_t dat
 
     // Return indication of more to come - keep processing if not inactive
     connStatus = _pResponder->getConnStatus();
-    if ((connStatus == CONN_INACTIVE) && (_socketTxQueuedBuffer.size() != 0))
+    if ((connStatus == CONN_INACTIVE) && (txQueuedLen() != 0))
     {
         return true;
     }
@@ -1113,7 +1115,7 @@ RaftWebConnSendRetVal RaftWebConnection::canSendOnConn()
 #endif
 
     // Don't accept any more data while the buffer is not empty
-    if (_socketTxQueuedBuffer.size() != 0)
+    if (txQueuedLen() != 0)
     {
         return WEB_CONN_SEND_EAGAIN;
     }
@@ -1135,7 +1137,7 @@ RaftWebConnSendRetVal RaftWebConnection::canSendOnConn()
     {
         LOG_I(MODULE_PREFIX, "canSendOnConn connId %d totalUs %d clientCanSendUs %d bufSize %d result %d",
                     _pClientConn->getClientId(), totalUs, clientCanSendUs, 
-                    (int)_socketTxQueuedBuffer.size(), result);
+                    (int)txQueuedLen(), result);
     }
 #endif
 
@@ -1147,6 +1149,19 @@ RaftWebConnSendRetVal RaftWebConnection::canSendOnConn()
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs)
+{
+    return rawSendOnConnImpl(pBuf, bufLen, maxRetryMs, nullptr);
+}
+
+// As rawSendOnConn, but if what can't be sent at once is to be queued and the queue is empty, the queue takes the
+// buffer (leaving it empty) rather than copying it
+RaftWebConnSendRetVal RaftWebConnection::rawSendOnConnTake(SpiramAwareUint8Vector& buf, uint32_t maxRetryMs)
+{
+    return rawSendOnConnImpl(buf.data(), buf.size(), maxRetryMs, &buf);
+}
+
+RaftWebConnSendRetVal RaftWebConnection::rawSendOnConnImpl(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs,
+            SpiramAwareUint8Vector* pTakeBuf)
 {
 #if defined(ESP_PLATFORM) && defined(DEBUG_RAW_SEND_ON_CONN_TIMING)
     static uint64_t totalCycles = 0;
@@ -1223,7 +1238,7 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint
 #endif
 
     // Check if data can be sent immediately
-        if (_socketTxQueuedBuffer.size() == 0)
+        if (txQueuedLen() == 0)
         {
             // Queue is currently empty so try to send
 #if defined(ESP_PLATFORM) && defined(DEBUG_RAW_SEND_ON_CONN_TIMING)
@@ -1266,7 +1281,7 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint
             retFinal = WEB_CONN_SEND_FAIL;
             break;
         }
-    uint32_t curSize = _socketTxQueuedBuffer.size();
+    uint32_t curSize = txQueuedLen();
         if (curSize + bytesToAddToQueue > _maxSendBufferBytes)
         {
 #ifdef DEBUG_WEB_CONNECTION_DATA_PACKETS
@@ -1281,6 +1296,23 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint
 #if defined(ESP_PLATFORM) && defined(DEBUG_RAW_SEND_ON_CONN_TIMING)
     uint32_t queueStartCycles = esp_cpu_get_cycle_count();
 #endif
+        // If the queue is empty and the buffer may be taken, the queue takes it rather than copying what is left of it
+        if (pTakeBuf && (curSize == 0))
+        {
+            _socketTxQueuedBuffer.swap(*pTakeBuf);
+            _socketTxQueuedPos = bytesWritten;
+            pTakeBuf->clear();
+            retFinal = WEB_CONN_SEND_OK;
+            break;
+        }
+
+        // Remove what has been sent from the queue before adding to it
+        if (_socketTxQueuedPos > 0)
+        {
+            _socketTxQueuedBuffer.erase(_socketTxQueuedBuffer.begin(), _socketTxQueuedBuffer.begin() + _socketTxQueuedPos);
+            _socketTxQueuedPos = 0;
+        }
+
         // Check the larger buffer can be allocated - an allocation failure can't be detected afterwards (exceptions
         // are disabled) and would crash. This happens when several connections queue data at once while the link 
         // drains slowly (e.g. weak WiFi) and the heap is fragmented. The buffer is grown to the exact size needed
@@ -1309,7 +1341,7 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint
 
 #ifdef DEBUG_WEB_CONNECTION_DATA_PACKETS
     LOG_I(MODULE_PREFIX, "rawSendOnConn connId %d added %d bytes to send buffer newLen %d", 
-                _pClientConn->getClientId(), bytesToAddToQueue, _socketTxQueuedBuffer.size());
+                _pClientConn->getClientId(), bytesToAddToQueue, txQueuedLen());
 #endif
 
         // The data has been queued for sending (WEB_CONN_SEND_OK)
@@ -1544,7 +1576,7 @@ bool RaftWebConnection::handleResponseChunk()
 #endif
 
     // Check if data waiting to be sent
-    if (_socketTxQueuedBuffer.size() == 0)
+    if (txQueuedLen() == 0)
     {
         // Get next chunk of response
         uint8_t* pRespBuffer = nullptr;
@@ -1622,33 +1654,35 @@ bool RaftWebConnection::handleResponseChunk()
 bool RaftWebConnection::handleTxQueuedData()
 {
     // Check if data in the queue to send
-    if (_socketTxQueuedBuffer.size() > 0)
+    if (txQueuedLen() > 0)
     {
 #ifdef DEBUG_WEB_CONN_OPEN_CLOSE
         LOG_I(MODULE_PREFIX, "handleTxQueuedData connId %d HAS DATA: %d bytes queued", 
-                        _pClientConn ? _pClientConn->getClientId() : -1, _socketTxQueuedBuffer.size());
+                        _pClientConn ? _pClientConn->getClientId() : -1, txQueuedLen());
 #endif
         // Try to send
         uint32_t bytesWritten = 0;
-        uint32_t bytesToSend = _socketTxQueuedBuffer.size();
-        RaftWebConnSendRetVal retVal = _pClientConn->sendDataBuffer(_socketTxQueuedBuffer.data(), bytesToSend, 
+        uint32_t bytesToSend = txQueuedLen();
+        RaftWebConnSendRetVal retVal = _pClientConn->sendDataBuffer(_socketTxQueuedBuffer.data() + _socketTxQueuedPos, bytesToSend, 
                         MAX_CONTENT_SEND_RETRY_MS, bytesWritten);
         if (retVal == WEB_CONN_SEND_EAGAIN)
             return true;
 #ifdef DEBUG_WEB_CONNECTION_DATA_PACKETS
         LOG_I(MODULE_PREFIX, "handleTxQueuedData connId %d result %s bytesWritten %d remaining %d", 
                         _pClientConn->getClientId(), RaftWebConnDefs::getSendRetValStr(retVal), bytesWritten, 
-                        _socketTxQueuedBuffer.size()-bytesWritten);
+                        txQueuedLen()-bytesWritten);
 #endif
         if (retVal == WEB_CONN_SEND_FAIL)
         {
             // Clear the send buffer
-            _socketTxQueuedBuffer.clear();
+            txQueueClear();
             return false;
         }
 
-        // Sent ok so clear items from the queue that were sent
-        _socketTxQueuedBuffer.erase(_socketTxQueuedBuffer.begin(), _socketTxQueuedBuffer.begin() + bytesWritten);
+        // Sent ok so move past what was sent (emptying the queue when all is sent)
+        _socketTxQueuedPos += bytesWritten;
+        if (_socketTxQueuedPos >= _socketTxQueuedBuffer.size())
+            txQueueClear();
     }
     return true;
 }

components/RaftWebServer/RaftWebConnection.h

@@ -108,8 +108,19 @@ class RaftWebConnection
     // Max send buffer size
     uint32_t _maxSendBufferBytes;
 
-    // Queued data to send
+    // Queued data to send, from _socketTxQueuedPos (data before it has been sent - it is only removed when more is
+    // added, so sending from the queue doesn't move what is left each time)
     std::vector<uint8_t, SpiramAwareAllocator<uint8_t>> _socketTxQueuedBuffer;
+    uint32_t _socketTxQueuedPos = 0;
+    uint32_t txQueuedLen() const
+    {
+        return _socketTxQueuedBuffer.size() - _socketTxQueuedPos;
+    }
+    void txQueueClear()
+    {
+        _socketTxQueuedBuffer.clear();
+        _socketTxQueuedPos = 0;
+    }
 
     // Heap to leave free (for the network stack) when growing the send buffer
     static const uint32_t SEND_BUFFER_HEAP_MARGIN_BYTES = 8192;
@@ -147,7 +158,10 @@ class RaftWebConnection
     void setHTTPResponseStatus(RaftHttpStatusCode reponseCode);
 
     // Raw send on connection - used by websockets, etc
-    RaftWebConnSendRetVal rawSendOnConn(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs);    
+    RaftWebConnSendRetVal rawSendOnConn(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs);
+    RaftWebConnSendRetVal rawSendOnConnTake(SpiramAwareUint8Vector& buf, uint32_t maxRetryMs);
+    RaftWebConnSendRetVal rawSendOnConnImpl(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs,
+                SpiramAwareUint8Vector* pTakeBuf);    
 
     // Header handling
     bool getStandardHeaders(String& headerStr);

components/RaftWebServer/RaftWebRequestParams.h

@@ -31,10 +31,21 @@ class RaftWebRequestParams
         return _webConnReadyToSend;
     }
 
+    // Send that may take the buffer (see RaftWebConnSendTakeFn) - not set for every connection
+    void setWebConnRawSendTake(RaftWebConnSendTakeFn webConnRawSendTake)
+    {
+        _webConnRawSendTake = webConnRawSendTake;
+    }
+    RaftWebConnSendTakeFn getWebConnRawSendTake() const
+    {
+        return _webConnRawSendTake;
+    }
+
     // Connection ID (for debugging)
     uint32_t connId = 0;
 
 private:
     RaftWebConnSendFn _webConnRawSend;
+    RaftWebConnSendTakeFn _webConnRawSendTake = nullptr;
     RaftWebConnReadyToSendFn _webConnReadyToSend;
 };

components/RaftWebServer/RaftWebResponder.h

@@ -114,6 +114,13 @@ class RaftWebResponder
         return false;
     }
 
+    // Encode and send data whose buffer may be taken (leaving it empty) rather than copied - by default it is sent
+    // as above
+    virtual bool encodeAndSendData(SpiramAwareUint8Vector& buf)
+    {
+        return encodeAndSendData(buf.data(), buf.size());
+    }
+
     // Send event content and group
     virtual void sendEvent(const char* eventContent, const char* eventGroup)
     {

components/RaftWebServer/RaftWebResponderWS.cpp

@@ -67,6 +67,7 @@ RaftWebResponderWS::RaftWebResponderWS(RaftWebHandlerWS* pWebHandler, const Raft
     _webSocketLink.setup(std::bind(&RaftWebResponderWS::onWebSocketEvent, this, 
                             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3),
                 params.getWebConnRawSend(), pingIntervalMs, true, disconnIfNoPongMs, isBinary);
+    _webSocketLink.setRawConnSendTakeFn(params.getWebConnRawSendTake());
     _webSocketLink.setMaxRxMsgLen(_packetMaxBytes);
 }
 
@@ -410,6 +411,16 @@ bool RaftWebResponderWS::isReadyToSend()
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 bool RaftWebResponderWS::encodeAndSendData(const uint8_t* pBuf, uint32_t bufLen)
+{
+    return encodeAndSendDataImpl(pBuf, bufLen, nullptr);
+}
+
+bool RaftWebResponderWS::encodeAndSendData(SpiramAwareUint8Vector& buf)
+{
+    return encodeAndSendDataImpl(buf.data(), buf.size(), &buf);
+}
+
+bool RaftWebResponderWS::encodeAndSendDataImpl(const uint8_t* pBuf, uint32_t bufLen, SpiramAwareUint8Vector* pTakeBuf)
 {
 #ifdef DEBUG_WS_SEND_APP_DATA
     LOG_I(MODULE_PREFIX, "encodeAndSendData connId %d len=%d status=%d", 
@@ -467,7 +478,9 @@ bool RaftWebResponderWS::encodeAndSendData(const uint8_t* pBuf, uint32_t bufLen)
     return putRslt;
 #else
     // Send
-    RaftWebConnSendRetVal retVal = _webSocketLink.sendMsg(_webSocketLink.msgOpCodeDefault(), pBuf, bufLen);
+    RaftWebConnSendRetVal retVal = pTakeBuf ?
+                _webSocketLink.sendMsg(_webSocketLink.msgOpCodeDefault(), *pTakeBuf) :
+                _webSocketLink.sendMsg(_webSocketLink.msgOpCodeDefault(), pBuf, bufLen);
 
 #ifdef DEBUG_WS_SEND_APP_DATA
     LOG_W(MODULE_PREFIX, "encodeAndSendData connId %d sent len %d retc %d",

components/RaftWebServer/RaftWebResponderWS.h

@@ -70,6 +70,7 @@ class RaftWebResponderWS : public RaftWebResponder
 
     // Send a frame of data
     virtual bool encodeAndSendData(const uint8_t* pBuf, uint32_t bufLen) override final;
+    virtual bool encodeAndSendData(SpiramAwareUint8Vector& buf) override final;
 
     // Get responder type
     virtual const char* getResponderType() override final
@@ -91,6 +92,9 @@ class RaftWebResponderWS : public RaftWebResponder
     }
 
 private:
+    // Encode and send data (pTakeBuf, if not null, is the buffer holding it, which may be taken)
+    bool encodeAndSendDataImpl(const uint8_t* pBuf, uint32_t bufLen, SpiramAwareUint8Vector* pTakeBuf);
+
     // Handler
     RaftWebHandlerWS* _pWebHandler;
 

components/RaftWebServer/RaftWebServer.h

@@ -50,6 +50,12 @@ class RaftWebServer
         return _connManager.sendBufOnChannel(pBuf, bufLen, channelID);
     }
 
+    // Send a buffer on a channel, which may be taken (leaving it empty) rather than copied (main task only)
+    bool sendBufferOnChannel(SpiramAwareUint8Vector& buf, uint32_t channelID)
+    {
+        return _connManager.sendBufOnChannel(buf, channelID);
+    }
+
     // Send to all server-side events (main task only)
     void serverSideEventsSendMsg(const char* eventContent, const char* eventGroup);