From 07e60d9bbafd4e11ed0ec923f85370b0c594fdde Mon Sep 17 00:00:00 2001 From: ellensp <530024+ellensp@users.noreply.github.com> Date: Sun, 25 Oct 2026 00:54:45 +1300 Subject: [PATCH] Don't send to network clients from the caller's task Serial_2_Socket::write runs in Marlin's loop task. On each newline it processed the line there, so dispatching to the websocket and other clients ran in Marlin's task. Sending to a websocket client that has stopped reading waits up to about 15 s (WiFiClient write retries plus the WebSockets TCP timeout), long enough to trip Marlin's task watchdog. write() and flush() now put each message on a queue (dropping it if the queue is full), and handle(), in the ESP3D task, processes the queue. WebSocket_Server sends to each client in turn and disconnects a client whose send fails, so it can't stall later sends. --- src/modules/serial2socket/serial2socket.cpp | 59 +++++++++++++-------- src/modules/serial2socket/serial2socket.h | 3 ++ src/modules/websocket/websocket_server.cpp | 9 +++- 3 files changed, 47 insertions(+), 24 deletions(-) diff --git a/src/modules/serial2socket/serial2socket.cpp b/src/modules/serial2socket/serial2socket.cpp index 506a070..52e1f15 100644 --- a/src/modules/serial2socket/serial2socket.cpp +++ b/src/modules/serial2socket/serial2socket.cpp @@ -25,6 +25,7 @@ #include #include "freertos/FreeRTOS.h" #include "freertos/semphr.h" +#include "freertos/queue.h" #include "../../core/esp3d_commands.h" #include "../../core/esp3d_message.h" @@ -42,7 +43,11 @@ Serial_2_Socket::Serial_2_Socket() { if (_txBufferMutex == NULL) { esp3d_log_e("Serial2Socket: Failed to create TX mutex"); } - + _txQueue = (void*)xQueueCreate(S2S_TXQUEUESIZE, sizeof(ESP3DMessage *)); + if (_txQueue == NULL) { + esp3d_log_e("Serial2Socket: Failed to create TX queue"); + } + end(); } @@ -175,7 +180,7 @@ size_t Serial_2_Socket::write(const uint8_t *buffer, size_t size) { // Check if buffer is full and needs flushing if (_TXbufferSize + size > S2S_TXBUFFERSIZE) { esp3d_log_d("Serial2Socket: buffer full, flush it"); - flush(false); // Use flush without mutex since we already have it + queueTXbuffer(); // We already have the mutex } // Add data to buffer and flush on newline @@ -184,13 +189,13 @@ size_t Serial_2_Socket::write(const uint8_t *buffer, size_t size) { _TXbufferSize++; if (buffer[i] == (const uint8_t)'\n') { esp3d_log_d("S2S: %s TXSize: %d", (const char *)_TXbuffer, _TXbufferSize); - flush(false); // Use flush without mutex since we already have it + queueTXbuffer(); // We already have the mutex } } // Check if we need to flush based on time if (_TXbufferSize > 0 && ((millis() - _lastflush) > S2S_FLUSHTIMEOUT)) { - flush(false); // Use flush without mutex since we already have it + queueTXbuffer(); // We already have the mutex } // Release the mutex at the end @@ -286,7 +291,16 @@ int Serial_2_Socket::read(void) { return v; } -void Serial_2_Socket::handle() { handle_flush(); } +// Called from the ESP3D task. write() runs in the caller's task (Marlin's +// loop), which must not wait on network clients, so it only queues messages. +void Serial_2_Socket::handle() { + handle_flush(); + ESP3DMessage *msg; + while (_txQueue != NULL && + xQueueReceive((QueueHandle_t)_txQueue, &msg, 0) == pdTRUE) { + esp3d_commands.process(msg); + } +} void Serial_2_Socket::handle_flush() { if (_TXbufferSize > 0 && _started && !_paused) { @@ -299,33 +313,32 @@ void Serial_2_Socket::handle_flush() { } void Serial_2_Socket::flush(bool useMutex) { + if (useMutex && _txBufferMutex != NULL && + xSemaphoreTake((SemaphoreHandle_t)_txBufferMutex, pdMS_TO_TICKS(100)) != pdTRUE) { + esp3d_log_e("Serial2Socket: Failed to take mutex for flush"); + return; + } + queueTXbuffer(); + if (useMutex && _txBufferMutex != NULL) { + xSemaphoreGive((SemaphoreHandle_t)_txBufferMutex); + } +} - // Process buffer if there's data and we're not paused +// Move the TX buffer into a message for handle() to process. +// The caller holds the TX mutex. If the queue is full, the output is dropped. +void Serial_2_Socket::queueTXbuffer() { if (_TXbufferSize > 0 && _started && !_paused) { - // Protect TX buffer access with mutex if requested - if (useMutex && _txBufferMutex != NULL) { - if (xSemaphoreTake((SemaphoreHandle_t)_txBufferMutex, pdMS_TO_TICKS(100)) != pdTRUE) { - esp3d_log_e("Serial2Socket: Failed to take mutex for flush"); - return; - } - } ESP3DMessage *msg = esp3d_message_manager.newMsg( ESP3DClientType::socket_serial, ESP3DClientType::all_clients, _TXbuffer, _TXbufferSize, _auth); - - // Reset buffer before processing _lastflush = millis(); _TXbufferSize = 0; - - // Release mutex if we took it - if (useMutex && _txBufferMutex != NULL) { - xSemaphoreGive((SemaphoreHandle_t)_txBufferMutex); - } - if (msg) { - // process command msg->type = ESP3DMessageType::unique; - esp3d_commands.process(msg); + if (_txQueue == NULL || + xQueueSend((QueueHandle_t)_txQueue, &msg, 0) != pdTRUE) { + esp3d_message_manager.deleteMsg(msg); + } } else { esp3d_log_e("Cannot create message"); } diff --git a/src/modules/serial2socket/serial2socket.h b/src/modules/serial2socket/serial2socket.h index 45d654a..fbfc285 100644 --- a/src/modules/serial2socket/serial2socket.h +++ b/src/modules/serial2socket/serial2socket.h @@ -27,6 +27,7 @@ #define S2S_TXBUFFERSIZE 512 #define S2S_RXBUFFERSIZE 256 #define S2S_FLUSHTIMEOUT 500 +#define S2S_TXQUEUESIZE 32 class ESP3DMessage; // forward declaration @@ -65,6 +66,8 @@ class Serial_2_Socket : public Stream { ESP3DAuthenticationLevel _auth; void* _rxBufferMutex; // Opaque pointer for RX buffer mutex void* _txBufferMutex; // Opaque pointer for TX buffer mutex + void* _txQueue; // Opaque pointer for queue of TX messages + void queueTXbuffer(); uint32_t _lastflush; uint8_t _TXbuffer[S2S_TXBUFFERSIZE+1]; uint16_t _TXbufferSize; diff --git a/src/modules/websocket/websocket_server.cpp b/src/modules/websocket/websocket_server.cpp index a709e97..1866e1f 100644 --- a/src/modules/websocket/websocket_server.cpp +++ b/src/modules/websocket/websocket_server.cpp @@ -367,7 +367,14 @@ void WebSocket_Server::flushTXbuffer(void) { if (_started) { if ((_TXbufferSize > 0) && (_websocket_server->connectedClients() > 0)) { if (_websocket_server) { - _websocket_server->broadcastBIN(_TXbuffer, _TXbufferSize); + for (uint8_t num = 0; num < WEBSOCKETS_SERVER_CLIENT_MAX; num++) { + // A client that stops reading would stall every later send, so drop it + if (_websocket_server->clientIsConnected(num) && + !_websocket_server->sendBIN(num, _TXbuffer, _TXbufferSize)) { + esp3d_log_e("WS send failed, disconnect %d", num); + _websocket_server->disconnect(num); + } + } esp3d_log("WS Broadcast bin port %d: %d bytes", port(), _TXbufferSize); } // refresh timout