From c1b5566fe6e491903b9012b22302054120e1fb26 Mon Sep 17 00:00:00 2001 From: Will Miles Date: Mon, 14 Sep 2026 19:25:41 -0400 Subject: [PATCH 1/3] fix(es): Fix queue handling Track when we've filled the underlying socket and stop until we receive an ack. Reduces CPU usage blindly retrying when the buffer is full. Also fixes a correctness issue where a _runQueue which partially sends a message will still attempt to send the next message, which will tear if an ack arrives in between writes. --- src/AsyncEventSource.cpp | 23 ++++++++--------------- src/AsyncEventSource.h | 1 + 2 files changed, 9 insertions(+), 15 deletions(-) diff --git a/src/AsyncEventSource.cpp b/src/AsyncEventSource.cpp index 39d94e3f..8a81a063 100644 --- a/src/AsyncEventSource.cpp +++ b/src/AsyncEventSource.cpp @@ -208,13 +208,8 @@ bool AsyncEventSourceClient::_queueMessage(const char *message, size_t len) { return false; } - /* - throttle queue run - if Q is filled for >25% then network/CPU is congested, since there is no zero-copy mode for socket buff - forcing Q run will only eat more heap ram and blow the buffer, let's just keep data in our own queue - the queue will be processed at least on each onAck()/onPoll() call from AsyncTCP - */ - if (_client && _client->canSend() && _messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2) { + // Send new content if we're not waiting on network buffer space + if (!_ack_pending) { _runQueue(); } @@ -237,13 +232,8 @@ bool AsyncEventSourceClient::_queueMessage(AsyncEvent_SharedData_t &&msg) { return false; } - /* - throttle queue run - if Q is filled for >25% then network/CPU is congested, since there is no zero-copy mode for socket buff - forcing Q run will only eat more heap ram and blow the buffer, let's just keep data in our own queue - the queue will be processed at least on each onAck()/onPoll() call from AsyncTCP - */ - if (_client && _client->canSend() && _messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2) { + // Send new content if we're not waiting on network buffer space + if (!_ack_pending) { _runQueue(); } return true; @@ -269,6 +259,8 @@ void AsyncEventSourceClient::_onAck(size_t len __attribute__((unused)), uint32_t } } + _ack_pending = false; // some space has been cleared + // try to send another batch of data if (_messageQueue.size()) { _runQueue(); @@ -322,8 +314,9 @@ void AsyncEventSourceClient::_runQueue() { const size_t bytes_written = i->write(_client); total_bytes_written += bytes_written; _inflight += bytes_written; - if (bytes_written == 0 || _inflight > _max_inflight) { + if (!i->sent() || _inflight > _max_inflight) { // Serial.print("_"); + _ack_pending = true; // Output buffer is saturated. break; } } diff --git a/src/AsyncEventSource.h b/src/AsyncEventSource.h index e6f94dff..9500beb3 100644 --- a/src/AsyncEventSource.h +++ b/src/AsyncEventSource.h @@ -135,6 +135,7 @@ class AsyncEventSourceClient { uint32_t _lastId{0}; size_t _inflight{0}; // num of unacknowledged bytes that has been written to socket buffer size_t _max_inflight{SSE_MAX_INFLIGH}; // max num of unacknowledged bytes that could be written to socket buffer + bool _ack_pending = false; std::list _messageQueue; mutable asyncsrv::mutex_type _lockmq; bool _queueMessage(const char *message, size_t len); From e48f8677504ca4bd779806fd08882e7616762102 Mon Sep 17 00:00:00 2001 From: Will Miles Date: Mon, 14 Sep 2026 19:27:47 -0400 Subject: [PATCH 2/3] perf(es): Remove redundant space check Drop redundant space check prior to attempting data send. add() has to check for available space internally, interlocked with the TCP layer; the additional check just wastes time. --- src/AsyncEventSource.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/AsyncEventSource.cpp b/src/AsyncEventSource.cpp index 8a81a063..e938bb39 100644 --- a/src/AsyncEventSource.cpp +++ b/src/AsyncEventSource.cpp @@ -119,11 +119,11 @@ size_t AsyncEventSourceMessage::write(AsyncClient *client) { return 0; } - if (_sent >= _data->length() || !client->canSend()) { + if (_sent >= _data->length()) { return 0; } - size_t len = std::min(_data->length() - _sent, client->space()); + size_t len = _data->length() - _sent; /* add() would call lwip's tcp_write() under the AsyncTCP hood with apiflags argument. By default apiflags=ASYNC_WRITE_FLAG_COPY From 880c97068abb0f0e63f803cd3c0cac84fa5cd7bb Mon Sep 17 00:00:00 2001 From: Will Miles Date: Mon, 14 Sep 2026 19:50:03 -0400 Subject: [PATCH 3/3] perf(es): Release messages after send As we are committed to ASYNC_WRITE_FLAG_COPY, we can release written event messages as soon as they are accepted by the TCP socket. Originally implemented in: - https://github.com/esphome/ESPAsyncWebServer/pull/41 - https://github.com/mathieucarbou/ESPAsyncWebServer/pull/92 Reverted by 43e0b5c5 in the hopes of supporting zero-copy TCP writes, but (as noted in that commit!) they are not supported on any Espressif platform, so the additional complexity buys no benefit. --- src/AsyncEventSource.cpp | 127 +++++++++++---------------------------- src/AsyncEventSource.h | 87 +++++---------------------- 2 files changed, 51 insertions(+), 163 deletions(-) diff --git a/src/AsyncEventSource.cpp b/src/AsyncEventSource.cpp index e938bb39..810877ed 100644 --- a/src/AsyncEventSource.cpp +++ b/src/AsyncEventSource.cpp @@ -99,52 +99,6 @@ static String generateEventMessage(const char *message, const char *event, uint3 return str; } -// Message - -size_t AsyncEventSourceMessage::ack(size_t len, __attribute__((unused)) uint32_t time) { - // If the whole message is now acked... - if (_acked + len > _data->length()) { - // Return the number of extra bytes acked (they will be carried on to the next message) - const size_t extra = _acked + len - _data->length(); - _acked = _data->length(); - return extra; - } - // Return that no extra bytes left. - _acked += len; - return 0; -} - -size_t AsyncEventSourceMessage::write(AsyncClient *client) { - if (!client) { - return 0; - } - - if (_sent >= _data->length()) { - return 0; - } - - size_t len = _data->length() - _sent; - /* - add() would call lwip's tcp_write() under the AsyncTCP hood with apiflags argument. - By default apiflags=ASYNC_WRITE_FLAG_COPY - we could have used apiflags with this flag unset to pass data by reference and avoid copy to socket buffer, - but looks like it does not work for Arduino's lwip in ESP32/IDF - it is enforced in https://github.com/espressif/esp-lwip/blob/0606eed9d8b98a797514fdf6eabb4daf1c8c8cd9/src/core/tcp_out.c#L422C5-L422C30 - if LWIP_NETIF_TX_SINGLE_PBUF is set, and it is set indeed in IDF - https://github.com/espressif/esp-idf/blob/a0f798cfc4bbd624aab52b2c194d219e242d80c1/components/lwip/port/include/lwipopts.h#L744 - - So let's just keep it enforced ASYNC_WRITE_FLAG_COPY and keep in mind that there is no zero-copy - */ - size_t written = client->add(_data->c_str() + _sent, len, ASYNC_WRITE_FLAG_COPY); // ASYNC_WRITE_FLAG_MORE - _sent += written; - return written; -} - -size_t AsyncEventSourceMessage::send(AsyncClient *client) { - size_t sent = write(client); - return sent && client->send() ? sent : 0; -} - // Client AsyncEventSourceClient::AsyncEventSourceClient(AsyncClient *client, AsyncEventSource *server, uint32_t lastId) @@ -192,7 +146,7 @@ AsyncEventSourceClient::~AsyncEventSourceClient() { close(); } -bool AsyncEventSourceClient::_queueMessage(const char *message, size_t len) { +bool AsyncEventSourceClient::_queueMessage(AsyncEvent_SharedData_t &&msg) { // Protect message queue access (size checks and modifications) which is not thread-safe. asyncsrv::lock_guard_type lock(_lockmq); @@ -201,28 +155,8 @@ bool AsyncEventSourceClient::_queueMessage(const char *message, size_t len) { return false; } - if (_client) { - _messageQueue.emplace_back(message, len); - } else { - _messageQueue.clear(); - return false; - } - - // Send new content if we're not waiting on network buffer space - if (!_ack_pending) { - _runQueue(); - } - - return true; -} - -bool AsyncEventSourceClient::_queueMessage(AsyncEvent_SharedData_t &&msg) { - // Protect message queue access (size checks and modifications) which is not thread-safe. - asyncsrv::lock_guard_type lock(_lockmq); - - if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) { - async_ws_log_w("Event message queue overflow: discard message"); - return false; + if (!msg || msg->length() == 0) { + return false; // Invalid message } if (_client) { @@ -233,13 +167,13 @@ bool AsyncEventSourceClient::_queueMessage(AsyncEvent_SharedData_t &&msg) { } // Send new content if we're not waiting on network buffer space - if (!_ack_pending) { + if (!_ack_pending && (_inflight < _max_inflight)) { _runQueue(); } return true; } -void AsyncEventSourceClient::_onAck(size_t len __attribute__((unused)), uint32_t time __attribute__((unused))) { +void AsyncEventSourceClient::_onAck(size_t len, uint32_t time __attribute__((unused))) { // Protect message queue access (size checks and modifications) which is not thread-safe. asyncsrv::lock_guard_type lock(_lockmq); @@ -250,15 +184,6 @@ void AsyncEventSourceClient::_onAck(size_t len __attribute__((unused)), uint32_t _inflight = 0; } - // acknowledge as much messages's data as we got confirmed len from a AsyncTCP - while (len && _messageQueue.size()) { - len = _messageQueue.front().ack(len); - if (_messageQueue.front().finished()) { - // now we could release full ack'ed messages, we were keeping it unless send confirmed from AsyncTCP - _messageQueue.pop_front(); - } - } - _ack_pending = false; // some space has been cleared // try to send another batch of data @@ -270,7 +195,7 @@ void AsyncEventSourceClient::_onAck(size_t len __attribute__((unused)), uint32_t void AsyncEventSourceClient::_onPoll() { // Protect message queue access (size checks and modifications) which is not thread-safe. asyncsrv::lock_guard_type lock(_lockmq); - if (_messageQueue.size()) { + if ((_messageQueue.size()) && (_inflight < _max_inflight)) { _runQueue(); } } @@ -309,16 +234,36 @@ void AsyncEventSourceClient::_runQueue() { // there is no need to lock the mutex here, 'cause all the calls to this method must be already lock'ed size_t total_bytes_written = 0; - for (auto i = _messageQueue.begin(); i != _messageQueue.end(); ++i) { - if (!i->sent()) { - const size_t bytes_written = i->write(_client); - total_bytes_written += bytes_written; - _inflight += bytes_written; - if (!i->sent() || _inflight > _max_inflight) { - // Serial.print("_"); - _ack_pending = true; // Output buffer is saturated. - break; - } + while (!_messageQueue.empty()) { + auto &data = _messageQueue.front(); + size_t len = data->length() - _sent; + /* + add() would call lwip's tcp_write() under the AsyncTCP hood with apiflags argument. + By default apiflags=ASYNC_WRITE_FLAG_COPY + we could have used apiflags with this flag unset to pass data by reference and avoid copy to socket buffer, + but looks like it does not work for Arduino's lwip in ESP32/IDF + it is enforced in https://github.com/espressif/esp-lwip/blob/0606eed9d8b98a797514fdf6eabb4daf1c8c8cd9/src/core/tcp_out.c#L422C5-L422C30 + if LWIP_NETIF_TX_SINGLE_PBUF is set, and it is set indeed in IDF + https://github.com/espressif/esp-idf/blob/a0f798cfc4bbd624aab52b2c194d219e242d80c1/components/lwip/port/include/lwipopts.h#L744 + + So let's just keep it enforced ASYNC_WRITE_FLAG_COPY and keep in mind that there is no zero-copy + */ + size_t bytes_written = _client->add(data->c_str() + _sent, len, ASYNC_WRITE_FLAG_COPY); // ASYNC_WRITE_FLAG_MORE + if (bytes_written == 0) { + break; + } + total_bytes_written += bytes_written; + _inflight += bytes_written; + if ((_sent + bytes_written) == data->length()) { + _messageQueue.pop_front(); + _sent = 0; + } else { + _sent += bytes_written; + } + if (_sent || _inflight > _max_inflight) { + // Serial.print("_"); + _ack_pending = true; // Output buffer is saturated. + break; } } diff --git a/src/AsyncEventSource.h b/src/AsyncEventSource.h index 9500beb3..0bff368c 100644 --- a/src/AsyncEventSource.h +++ b/src/AsyncEventSource.h @@ -55,75 +55,6 @@ using ArAuthorizeConnectHandler = ArAuthorizeFunction; // shared message object container using AsyncEvent_SharedData_t = std::shared_ptr; -/** - * @brief Async Event Message container with shared message content data - * - */ -class AsyncEventSourceMessage { - -private: - const AsyncEvent_SharedData_t _data; - size_t _sent{0}; // num of bytes already sent - size_t _acked{0}; // num of bytes acked - -public: - AsyncEventSourceMessage(AsyncEvent_SharedData_t data) : _data(data){}; -#if defined(ESP32) - AsyncEventSourceMessage(const char *data, size_t len) : _data(std::make_shared(data, len)){}; -#elif defined(TARGET_RP2040) || defined(TARGET_RP2350) || defined(PICO_RP2040) || defined(PICO_RP2350) - AsyncEventSourceMessage(const char *data, size_t len) : _data(std::make_shared()) { - if (data && len > 0) { - _data->concat(data, len); - } - }; -#else - // esp8266's String does not have constructor with data/length arguments. Use a concat method here - AsyncEventSourceMessage(const char *data, size_t len) { - _data->concat(data, len); - }; -#endif - - /** - * @brief acknowledge sending len bytes of data - * @note if num of bytes to ack is larger then the unacknowledged message length the number of carried over bytes are returned - * - * @param len bytes to acknowledge - * @param time - * @return size_t number of extra bytes carried over - */ - size_t ack(size_t len, uint32_t time = 0); - - /** - * @brief write message data to client's buffer - * @note this method does NOT call client's send - * - * @param client - * @return size_t number of bytes written - */ - size_t write(AsyncClient *client); - - /** - * @brief writes message data to client's buffer and calls client's send method - * - * @param client - * @return size_t returns num of bytes the clien was able to send() - */ - size_t send(AsyncClient *client); - - // returns true if full message's length were acked - bool finished() { - return _acked == _data->length(); - } - - /** - * @brief returns true if all data has been sent already - * - */ - bool sent() { - return _sent == _data->length(); - } -}; - /** * @brief class holds a sse messages queue for a particular client's connection * @@ -136,9 +67,9 @@ class AsyncEventSourceClient { size_t _inflight{0}; // num of unacknowledged bytes that has been written to socket buffer size_t _max_inflight{SSE_MAX_INFLIGH}; // max num of unacknowledged bytes that could be written to socket buffer bool _ack_pending = false; - std::list _messageQueue; + size_t _sent{0}; // num of bytes already sent in the head message + std::list _messageQueue; mutable asyncsrv::mutex_type _lockmq; - bool _queueMessage(const char *message, size_t len); bool _queueMessage(AsyncEvent_SharedData_t &&msg); void _runQueue(); @@ -187,7 +118,19 @@ class AsyncEventSourceClient { [[deprecated("Use _write(AsyncEvent_SharedData_t message) instead to share same data with multiple SSE clients")]] bool write(const char *message, size_t len) { - return connected() && _queueMessage(message, len); + if (!connected()) { + return false; + } + // Portable String construction - not all platforms have String(char*, size_t) +#if defined(ESP32) + return _queueMessage(std::make_shared(message, len)); +#else + auto data = std::make_shared(); + if (message && len) { + data->concat(message, len); + } + return _queueMessage(std::move(data)); +#endif }; // close client's connection