Skip to content

AsyncEventSource queue management #480

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
willmmiles wants to merge 3 commits into
base: main
Choose a base branch
from
+54 −172
Open

AsyncEventSource queue management #480

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c1b5566
fix(es): Fix queue handling
willmmiles Sep 14, 2026
e48f867
perf(es): Remove redundant space check
willmmiles Sep 14, 2026
880c970
perf(es): Release messages after send
willmmiles Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 38 additions & 100 deletions src/AsyncEventSource.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() || !client->canSend()) {
return 0;
}

size_t len = std::min(_data->length() - _sent, client->space());
/*
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)
Expand Down Expand Up @@ -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);

Expand All @@ -201,33 +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;
}

/*
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) {
_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) {
Expand All @@ -237,19 +166,14 @@ 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 && (_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);

Expand All @@ -260,14 +184,7 @@ 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
if (_messageQueue.size()) {
Expand All @@ -278,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();
}
}
Expand Down Expand Up @@ -317,15 +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 (bytes_written == 0 || _inflight > _max_inflight) {
// Serial.print("_");
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;
}
}

Expand Down
88 changes: 16 additions & 72 deletions src/AsyncEventSource.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,75 +55,6 @@ using ArAuthorizeConnectHandler = ArAuthorizeFunction;
// shared message object container
using AsyncEvent_SharedData_t = std::shared_ptr<String>;

/**
* @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<String>(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<String>()) {
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
*
Expand All @@ -135,9 +66,10 @@ 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
std::list<AsyncEventSourceMessage> _messageQueue;
bool _ack_pending = false;
size_t _sent{0}; // num of bytes already sent in the head message
std::list<AsyncEvent_SharedData_t> _messageQueue;
mutable asyncsrv::mutex_type _lockmq;
bool _queueMessage(const char *message, size_t len);
bool _queueMessage(AsyncEvent_SharedData_t &&msg);
void _runQueue();

Expand Down Expand Up @@ -186,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<String>(message, len));
#else
auto data = std::make_shared<String>();
if (message && len) {
data->concat(message, len);
}
return _queueMessage(std::move(data));
#endif
};

// close client's connection
Expand Down
Loading