mirror of
https://github.com/esphome/esphome.git
synced 2026-10-14 10:12:52 +00:00
[web_server_idf] Keep active SSE sessions out of the LRU purge (#17383)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: J. Nick Koston <nick@koston.org> Co-authored-by: J. Nick Koston <nick@home-assistant.io>
This commit is contained in:
co-authored by
Claude Fable 5
J. Nick Koston
J. Nick Koston
parent
7f2c3ba3f5
commit
35b416c6c0
@@ -959,10 +959,39 @@ void AsyncEventSourceResponse::process_close_() {
|
||||
}
|
||||
}
|
||||
|
||||
void AsyncEventSourceResponse::refresh_lru_() {
|
||||
// httpd only advances a session's LRU counter when a request arrives, and an event stream receives one
|
||||
// request ever, so the purge that frees a socket for a new client would evict the live streams first.
|
||||
// The HTTPD task owns the session table, so the refresh runs there.
|
||||
// A stalled stream is the socket the purge should take, and a zeroed fd has no session left to refresh.
|
||||
if (this->send_failure_started_ms_ != 0 || this->fd_.load() == 0)
|
||||
return;
|
||||
const uint32_t now = App.get_loop_component_start_time();
|
||||
if (now - this->last_lru_refresh_ms_ < LRU_REFRESH_INTERVAL_MS ||
|
||||
this->lru_work_queued_.load(std::memory_order_acquire)) {
|
||||
return;
|
||||
}
|
||||
this->last_lru_refresh_ms_ = now; // a full queue also waits for the next interval
|
||||
this->lru_work_queued_.store(true, std::memory_order_release);
|
||||
if (httpd_queue_work(this->hd_, &AsyncEventSourceResponse::refresh_lru_work, this) != ESP_OK) {
|
||||
this->lru_work_queued_.store(false, std::memory_order_release);
|
||||
}
|
||||
}
|
||||
|
||||
void AsyncEventSourceResponse::refresh_lru_work(void *arg) {
|
||||
auto *response = static_cast<AsyncEventSourceResponse *>(arg);
|
||||
const int fd = response->fd_.load();
|
||||
if (response->owns_session_(fd)) {
|
||||
httpd_sess_update_lru_counter(response->hd_, fd);
|
||||
}
|
||||
// Release self only after the HTTPD-task callback has finished every access.
|
||||
response->lru_work_queued_.store(false, std::memory_order_release);
|
||||
}
|
||||
|
||||
void AsyncEventSourceResponse::close_session_work(void *arg) {
|
||||
auto *response = static_cast<AsyncEventSourceResponse *>(arg);
|
||||
const int fd = response->fd_.load();
|
||||
if (fd != 0 && httpd_sess_get_ctx(response->hd_, fd) == response) {
|
||||
if (response->owns_session_(fd)) {
|
||||
// The HTTPD task remains the session owner. Shutting the socket down makes
|
||||
// its next select/recv path delete the session and invoke destroy().
|
||||
shutdown(fd, SHUT_RDWR);
|
||||
@@ -976,7 +1005,7 @@ ssize_t AsyncEventSourceResponse::send_(struct iovec *iov, int iovcnt) {
|
||||
// httpd frees a session before closing its socket, so the fd may already be a new client's.
|
||||
// Treated as would-block; the stall timer ends a session that never becomes ours again.
|
||||
const int fd = this->fd_.load();
|
||||
if (httpd_sess_get_ctx(this->hd_, fd) != this) {
|
||||
if (!this->owns_session_(fd)) {
|
||||
return 0;
|
||||
}
|
||||
struct msghdr msg {};
|
||||
@@ -1085,6 +1114,7 @@ void AsyncEventSourceResponse::loop() {
|
||||
this->process_close_();
|
||||
return;
|
||||
}
|
||||
this->refresh_lru_();
|
||||
drain_tail_();
|
||||
process_deferred_queue_();
|
||||
if (this->close_requested_)
|
||||
|
||||
@@ -339,10 +339,15 @@ class AsyncEventSourceResponse {
|
||||
};
|
||||
void process_close_();
|
||||
static void close_session_work(void *arg);
|
||||
void refresh_lru_();
|
||||
static void refresh_lru_work(void *arg);
|
||||
// httpd frees a session before closing its socket, so the fd may already belong to a new client
|
||||
bool owns_session_(int fd) const { return fd != 0 && httpd_sess_get_ctx(this->hd_, fd) == this; }
|
||||
|
||||
// Deletable only after destroy() zeroed fd_ and no queued HTTPD close work still references this object.
|
||||
// Deletable only after destroy() zeroed fd_ and no queued HTTPD work still references this object.
|
||||
bool safe_to_delete_() const {
|
||||
return this->fd_.load() == 0 && !this->close_work_queued_.load(std::memory_order_acquire);
|
||||
return this->fd_.load() == 0 && !this->close_work_queued_.load(std::memory_order_acquire) &&
|
||||
!this->lru_work_queued_.load(std::memory_order_acquire);
|
||||
}
|
||||
|
||||
static void destroy(void *p);
|
||||
@@ -357,11 +362,13 @@ class AsyncEventSourceResponse {
|
||||
RAMUniquePtr<uint8_t[]> tail_;
|
||||
uint32_t send_failure_started_ms_{0}; // Zero means no send stall in progress.
|
||||
uint32_t next_close_attempt_ms_{0};
|
||||
uint32_t last_lru_refresh_ms_{0};
|
||||
uint16_t tail_cap_{0};
|
||||
uint16_t tail_len_{0}; // Zero means nothing pending
|
||||
uint16_t tail_sent_{0};
|
||||
// Set on the main loop before queueing close work, cleared by the HTTPD-task callback when done.
|
||||
std::atomic<bool> close_work_queued_{false};
|
||||
std::atomic<bool> lru_work_queued_{false};
|
||||
// Main-loop only; the HTTPD task never reads or writes these flags.
|
||||
bool close_requested_{false};
|
||||
bool close_retry_warning_logged_{false};
|
||||
@@ -383,6 +390,7 @@ class AsyncEventSourceResponse {
|
||||
static constexpr uint32_t SEND_STALL_TIMEOUT_MS = 20000;
|
||||
static constexpr uint32_t CLOSE_RETRY_INTERVAL_MS = 250;
|
||||
static constexpr uint32_t CLOSE_CONFIRM_INTERVAL_MS = 1000;
|
||||
static constexpr uint32_t LRU_REFRESH_INTERVAL_MS = 1000;
|
||||
};
|
||||
|
||||
using AsyncEventSourceClient = AsyncEventSourceResponse;
|
||||
|
||||
Reference in New Issue
Block a user