From cc9786c415ec237c8c1979b6eb71ab3c5d99e45f Mon Sep 17 00:00:00 2001 From: Rob Dobson Date: Sun, 25 Oct 2026 20:55:47 +0100 Subject: [PATCH 1/2] WebSocket server sends: no copy of the message into a frame A server's (unmasked) message is sent as the header then the data, rather than copied into a frame buffer first. What the connection can't send at once is queued in order as before, and if it can't be queued the send fails and the connection is closed, so a frame is never left part sent. A client's (masked) message is still built into a frame, as it has to be masked. Measured publishing camera frames from an ESP32-S3 with quad PSRAM: building the frame took up to 4-6 ms for a 29 KB frame and now takes nothing, and the longest publish went from 21-22 ms to 17-19 ms (13-16 ms to about 12 ms for 19 KB frames). Co-Authored-By: Claude Opus 5.5 --- .../RaftWebServer/RaftWebSocketLink.cpp | 60 ++++++++++++------- 1 file changed, 39 insertions(+), 21 deletions(-) diff --git a/components/RaftWebServer/RaftWebSocketLink.cpp b/components/RaftWebServer/RaftWebSocketLink.cpp index edccab0..2d88e4f 100644 --- a/components/RaftWebServer/RaftWebSocketLink.cpp +++ b/components/RaftWebServer/RaftWebSocketLink.cpp @@ -449,32 +449,50 @@ RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, co return WEB_CONN_SEND_FRAME_ERROR; } - // Frame: header then data, appended to reserved space (it isn't zero-filled first, so large messages are - // written once rather than twice) - std::vector> frameBuffer; - frameBuffer.reserve(frameLen); - frameBuffer.insert(frameBuffer.end(), header, header + pos); - frameBuffer.insert(frameBuffer.end(), pBuf, pBuf + bufLen); - - // Mask the data - if (_maskSentData) + // Unmasked data (sent by a server) is sent as it is, after the header - without copying it into a frame first. + // If the connection can't take all of it, what is left is queued in order by the connection, and if it can't be + // queued the send fails and the connection is closed, so a frame is never sent in part +#ifdef DEBUG_WEBSOCKET_LINK_SEND + uint64_t timeNowUs = micros(); +#endif + RaftWebConnSendRetVal sendRetc = RaftWebConnSendRetVal::WEB_CONN_SEND_FAIL; + if (!_maskSentData) { +#ifdef DEBUG_WEBSOCKET_TIME_SEND_MSG + framePrepareUs += (micros() - startUs); + startUs = micros(); +#endif + if (_rawConnSendFn) + { + sendRetc = _rawConnSendFn(header, pos, MAX_WS_SEND_RETRY_MS); + if ((sendRetc == RaftWebConnSendRetVal::WEB_CONN_SEND_OK) && (bufLen > 0)) + { + sendRetc = _rawConnSendFn(pBuf, bufLen, MAX_WS_SEND_RETRY_MS); + + // The header has gone, so the frame must be completed or the connection closed + if (sendRetc != RaftWebConnSendRetVal::WEB_CONN_SEND_OK) + sendRetc = RaftWebConnSendRetVal::WEB_CONN_SEND_FAIL; + } + } + } + else + { + // Masked data (sent by a client): header then data, appended to reserved space (it isn't zero-filled first, + // so large messages are written once rather than twice) + std::vector> frameBuffer; + frameBuffer.reserve(frameLen); + frameBuffer.insert(frameBuffer.end(), header, header + pos); + frameBuffer.insert(frameBuffer.end(), pBuf, pBuf + bufLen); for (uint32_t i = 0; i < bufLen; i++) frameBuffer[pos + i] ^= maskBytes[i % WSHeaderInfo::WEB_SOCKET_MASK_KEY_BYTES]; - } #ifdef DEBUG_WEBSOCKET_TIME_SEND_MSG - framePrepareUs += (micros() - startUs); - startUs = micros(); -#endif - - // Send -#ifdef DEBUG_WEBSOCKET_LINK_SEND - uint64_t timeNowUs = micros(); + framePrepareUs += (micros() - startUs); + startUs = micros(); #endif - RaftWebConnSendRetVal sendRetc = RaftWebConnSendRetVal::WEB_CONN_SEND_FAIL; - if (_rawConnSendFn) - sendRetc = _rawConnSendFn(frameBuffer.data(), frameBuffer.size(), MAX_WS_SEND_RETRY_MS); + if (_rawConnSendFn) + sendRetc = _rawConnSendFn(frameBuffer.data(), frameBuffer.size(), MAX_WS_SEND_RETRY_MS); + } #ifdef DEBUG_WEBSOCKET_TIME_SEND_MSG rawConnSendUs += (micros() - startUs); @@ -498,7 +516,7 @@ RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, co uint64_t timeTakenUs = micros() - timeNowUs; LOG_I(MODULE_PREFIX, "sendMsg result %s send %d bytes took %dus", RaftWebConnDefs::getSendRetValStr(sendRetc), - frameBuffer.size(), (int)timeTakenUs); + frameLen, (int)timeTakenUs); #endif return sendRetc; From 297ded3620de95ce14e179c8d256b7893dcac4f1 Mon Sep 17 00:00:00 2001 From: Rob Dobson Date: Sun, 25 Oct 2026 21:02:19 +0100 Subject: [PATCH 2/2] WebSocket sends: the connection takes an encoded message's buffer rather than copying it - A message whose buffer may be taken (RaftCore's CommsChannelMsg::isBufTakeable(), set on a RICREST message just encoded for sending) goes through a take path (WebServer SysMod, RaftWebServer, RaftWebConnManager, the responder, RaftWebSocketLink, RaftWebConnection): what the socket can't take at once (all but the first 5.7 KB of a camera frame) is kept by taking the buffer into the empty send queue, rather than copied into it. The buffer is only taken when one connection is on the channel; other messages and senders are unchanged. - The send queue keeps a read position: sending from it no longer erases what was sent from the front, which moved everything left each time (about 57 KB of copying in later loops for a 29 KB frame). Measured publishing camera frames from an ESP32-S3 with quad PSRAM: longest publish 13.4 ms for 28 KB frames (17-19 ms with the previous commit, 21-22 ms before), 10.7-11.2 ms for 18 KB frames, and more frames got through (1.8 a second at 2 Hz, was 1.4). Co-Authored-By: Claude Opus 5.5 --- components/RaftWebServer/RaftWebConnDefs.h | 5 ++ .../RaftWebServer/RaftWebConnManager.cpp | 28 ++++++++ components/RaftWebServer/RaftWebConnManager.h | 3 + .../RaftWebServer/RaftWebConnection.cpp | 70 ++++++++++++++----- components/RaftWebServer/RaftWebConnection.h | 18 ++++- .../RaftWebServer/RaftWebRequestParams.h | 11 +++ components/RaftWebServer/RaftWebResponder.h | 7 ++ .../RaftWebServer/RaftWebResponderWS.cpp | 15 +++- components/RaftWebServer/RaftWebResponderWS.h | 4 ++ components/RaftWebServer/RaftWebServer.h | 6 ++ .../RaftWebServer/RaftWebSocketLink.cpp | 18 ++++- components/RaftWebServer/RaftWebSocketLink.h | 17 ++++- components/WebServer/WebServer.cpp | 4 ++ 13 files changed, 183 insertions(+), 23 deletions(-) diff --git a/components/RaftWebServer/RaftWebConnDefs.h b/components/RaftWebServer/RaftWebConnDefs.h index 418b6e7..860fe9b 100644 --- a/components/RaftWebServer/RaftWebConnDefs.h +++ b/components/RaftWebServer/RaftWebConnDefs.h @@ -10,6 +10,7 @@ #include #include +#include "SpiramAwareAllocator.h" // Callback function for any endpoint enum RaftWebConnSendRetVal @@ -44,3 +45,7 @@ typedef std::function RaftWebConnReadyToSendFn; // Function to send on a connection typedef std::function 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 RaftWebConnSendTakeFn; + diff --git a/components/RaftWebServer/RaftWebConnManager.cpp b/components/RaftWebServer/RaftWebConnManager.cpp index f8aca03..94d7b01 100644 --- a/components/RaftWebServer/RaftWebConnManager.cpp +++ b/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 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////// diff --git a/components/RaftWebServer/RaftWebConnManager.h b/components/RaftWebServer/RaftWebConnManager.h index e202e63..0ed52a2 100644 --- a/components/RaftWebServer/RaftWebConnManager.h +++ b/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); diff --git a/components/RaftWebServer/RaftWebConnection.cpp b/components/RaftWebServer/RaftWebConnection.cpp index 33a3b4d..ecbbe80 100644 --- a/components/RaftWebServer/RaftWebConnection.cpp +++ b/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; } \ No newline at end of file diff --git a/components/RaftWebServer/RaftWebConnection.h b/components/RaftWebServer/RaftWebConnection.h index 6288573..31d0615 100644 --- a/components/RaftWebServer/RaftWebConnection.h +++ b/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> _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); diff --git a/components/RaftWebServer/RaftWebRequestParams.h b/components/RaftWebServer/RaftWebRequestParams.h index 7c82f5d..d75b7b5 100644 --- a/components/RaftWebServer/RaftWebRequestParams.h +++ b/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; }; diff --git a/components/RaftWebServer/RaftWebResponder.h b/components/RaftWebServer/RaftWebResponder.h index 6c4e010..6abcbd3 100644 --- a/components/RaftWebServer/RaftWebResponder.h +++ b/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) { diff --git a/components/RaftWebServer/RaftWebResponderWS.cpp b/components/RaftWebServer/RaftWebResponderWS.cpp index f490e5b..cca3b41 100644 --- a/components/RaftWebServer/RaftWebResponderWS.cpp +++ b/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", diff --git a/components/RaftWebServer/RaftWebResponderWS.h b/components/RaftWebServer/RaftWebResponderWS.h index cd010b8..e1fe1eb 100644 --- a/components/RaftWebServer/RaftWebResponderWS.h +++ b/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; diff --git a/components/RaftWebServer/RaftWebServer.h b/components/RaftWebServer/RaftWebServer.h index fee50aa..cf23034 100644 --- a/components/RaftWebServer/RaftWebServer.h +++ b/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); diff --git a/components/RaftWebServer/RaftWebSocketLink.cpp b/components/RaftWebServer/RaftWebSocketLink.cpp index 2d88e4f..03a586a 100644 --- a/components/RaftWebServer/RaftWebSocketLink.cpp +++ b/components/RaftWebServer/RaftWebSocketLink.cpp @@ -367,6 +367,17 @@ uint32_t RaftWebSocketLink::getTxData(uint8_t*& pBuf, uint32_t bufMaxLen) ///////////////////////////////////////////////////////////////////////////////////////////////////////////////// RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, const uint8_t *pBuf, uint32_t bufLen) +{ + return sendMsgImpl(opCode, pBuf, bufLen, nullptr); +} + +RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, SpiramAwareUint8Vector& buf) +{ + return sendMsgImpl(opCode, buf.data(), buf.size(), &buf); +} + +RaftWebConnSendRetVal RaftWebSocketLink::sendMsgImpl(RaftWebSocketOpCodes opCode, const uint8_t *pBuf, uint32_t bufLen, + SpiramAwareUint8Vector* pTakeBuf) { #ifdef DEBUG_WEBSOCKET_SEND // Log all sends to debug reconnection issue @@ -467,7 +478,12 @@ RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, co sendRetc = _rawConnSendFn(header, pos, MAX_WS_SEND_RETRY_MS); if ((sendRetc == RaftWebConnSendRetVal::WEB_CONN_SEND_OK) && (bufLen > 0)) { - sendRetc = _rawConnSendFn(pBuf, bufLen, MAX_WS_SEND_RETRY_MS); + // The data's buffer is taken by the connection (rather than what it can't send at once copied) if + // it may be + if (pTakeBuf && _rawConnSendTakeFn) + sendRetc = _rawConnSendTakeFn(*pTakeBuf, MAX_WS_SEND_RETRY_MS); + else + sendRetc = _rawConnSendFn(pBuf, bufLen, MAX_WS_SEND_RETRY_MS); // The header has gone, so the frame must be completed or the connection closed if (sendRetc != RaftWebConnSendRetVal::WEB_CONN_SEND_OK) diff --git a/components/RaftWebServer/RaftWebSocketLink.h b/components/RaftWebServer/RaftWebSocketLink.h index 158cf49..47cd757 100644 --- a/components/RaftWebServer/RaftWebSocketLink.h +++ b/components/RaftWebServer/RaftWebSocketLink.h @@ -51,6 +51,16 @@ class RaftWebSocketLink // Send message RaftWebConnSendRetVal sendMsg(RaftWebSocketOpCodes opCode, const uint8_t* pBuf, uint32_t bufLen); + // Send a message whose buffer may be taken (leaving it empty) rather than copied, if the connection can take it + // (see setRawConnSendTakeFn) + RaftWebConnSendRetVal sendMsg(RaftWebSocketOpCodes opCode, SpiramAwareUint8Vector& buf); + + // Raw send on the connection which may take the buffer (optional) + void setRawConnSendTakeFn(RaftWebConnSendTakeFn rawConnSendTakeFn) + { + _rawConnSendTakeFn = rawConnSendTakeFn; + } + // Check active bool isActive() { @@ -100,8 +110,13 @@ class RaftWebSocketLink // Received data not yet processed SpiramAwareUint8Vector _rxDataToProcess; - // Raw send on the connection + // Raw send on the connection, and one which may take the buffer RaftWebConnSendFn _rawConnSendFn = nullptr; + RaftWebConnSendTakeFn _rawConnSendTakeFn = nullptr; + + // Send a message (pTakeBuf, if not null, is the buffer holding it, which may be taken) + RaftWebConnSendRetVal sendMsgImpl(RaftWebSocketOpCodes opCode, const uint8_t* pBuf, uint32_t bufLen, + SpiramAwareUint8Vector* pTakeBuf); // Data to be sent String _wsUpgradeResponse; diff --git a/components/WebServer/WebServer.cpp b/components/WebServer/WebServer.cpp index 893c5f5..76b9c27 100644 --- a/components/WebServer/WebServer.cpp +++ b/components/WebServer/WebServer.cpp @@ -381,6 +381,10 @@ void WebServer::webSocketSetup() interfaceName.c_str(), wsName.c_str(), [this](CommsChannelMsg& msg) { + // A message just encoded for sending (e.g. a publish) can have its buffer taken by the + // connection, rather than copied, for what can't be sent at once + if (msg.isBufTakeable()) + return _raftWebServer.sendBufferOnChannel(msg.getCmdVector(), msg.getChannelID()); return _raftWebServer.sendBufferOnChannel(msg.getBuf(), msg.getBufLen(), msg.getChannelID()); },