Compare commits

..
Author SHA1 Message Date
J. Nick Koston 0ac2299482 Merge remote-tracking branch 'origin/web-server-sse-listing-per-session' into HEAD 2026-11-14 08:40:18 -06:00
J. Nick Koston b58038af63 Merge remote-tracking branch 'origin/web-server-idf-stream-response' into HEAD 2026-11-14 08:39:50 -06:00
J. Nick Koston c741de7a47 Merge remote-tracking branch 'origin/dev' into HEAD 2026-11-14 08:38:08 -06:00
J. Nick Koston 8a5e07e29f [web_server] Keep the server out of host builds, which only need sse_chunk.cpp for unit tests
tests/components/web_server now pulls web_server into the --all host
build, where another test defines USE_WEBSERVER.
2026-11-14 00:19:08 -06:00
J. Nick Koston f7b52be670 Merge remote-tracking branch 'origin/web-server-sse-listing-per-session' into web-server-event-source-shared 2026-11-13 23:15:51 -06:00
J. Nick Koston d5317669c9 Merge remote-tracking branch 'origin/web-server-idf-stream-response' into web-server-sse-listing-per-session 2026-11-13 23:15:42 -06:00
J. Nick Koston 5d0bd88bc6 Merge remote-tracking branch 'origin/dev' into web-server-idf-stream-response 2026-11-13 23:13:08 -06:00
pre-commit-ci-lite[bot] b714d8f69b apply automatic formatting fixes 2026-11-14 04:33:03 +00:00
J. Nick Koston ecabdbb24b [web_server] Keep the shared /events layer free of IDF-only types
The gather type comes from stream_response.h, the default header count is
defined on every platform, the session takes a non-const request, and the
fd in log lines goes through log_id(). Also drops a dead forward declaration
and an unused alias, and narrows the source filter.
2026-11-13 22:28:19 -06:00
J. Nick Koston ffbc4148fb [web_server] Move the ESP32 /events layer to web_server/event_source
The event source and its sessions move out of web_server_idf into
web_server/event_source.h/.cpp, sse_chunk moves to web_server and
stream_head.h to web_server_base, so the ESPAsyncWebServer platforms can
share them later. web_server_base/stream_response.h picks the stream
transport. No behaviour change; off ESP32 the build is unchanged.
2026-11-13 22:16:53 -06:00
J. Nick Koston 2169167908 [web_server] Name the iterator's session member session_ 2026-11-13 22:00:50 -06:00
J. Nick Koston 900574c03a [web_server_idf] List entities to the new /events session only
The listing iterator held the whole event source, so each new client's
detail_all listing went to every open session. Hold the session instead,
as the Arduino path already does.
2026-11-13 21:55:42 -06:00
J. Nick Koston 7b1f9c6bf1 Merge remote-tracking branch 'origin/dev' into web-server-idf-stream-response
# Conflicts:
#	.agents/skills/hardware-test/reference/recipes/web_server.md
2026-11-07 22:09:36 -06:00
J. Nick Koston 3587e29725 Merge branch 'dev' into web-server-idf-stream-response 2026-11-04 22:21:12 -06:00
J. Nick Koston d4244c5388 [web_server_idf] Correct the session close order comment and point stall docs at the stream 2026-11-04 21:47:51 -06:00
J. Nick Koston bec53baf1d Merge branch 'dev' into web-server-idf-stream-response 2026-11-04 21:29:11 -06:00
J. Nick Koston 98fd3786dd [web_server_idf] Create the stream send mutex at boot and test that a failed head batch stops the rest 2026-11-04 21:15:36 -06:00
J. Nick Koston 1f531ff022 Merge branch 'dev' into web-server-idf-stream-response 2026-11-04 20:58:44 -06:00
J. Nick Koston 65e7eb6b06 [web_server_idf] Note that sess_ctx is reserved for AsyncStreamResponse 2026-11-04 20:41:33 -06:00
J. Nick Koston 7538864e1e [web_server_idf] Keep a stream from sending on a socket httpd has closed 2026-11-04 20:36:27 -06:00
J. Nick Koston c0f97475d3 [web_server_idf] Document the stream head body as a C string 2026-11-04 20:07:37 -06:00
J. Nick Koston df6fcaa6fe Merge remote-tracking branch 'origin/dev' into web-server-idf-stream-response
# Conflicts:
#	esphome/components/web_server_idf/web_server_idf.cpp
#	esphome/components/web_server_idf/web_server_idf.h
2026-11-03 18:47:25 -06:00
J. Nick Koston eb1b793af9 Merge branch 'dev' into web-server-idf-stream-response 2026-10-30 15:46:21 -10:00
J. Nick Koston b8ce904e0c [web_server_idf] Let a stream response header replace a default header of the same name 2026-10-30 14:24:31 -10:00
J. Nick Koston 6befa73b00 [web_server_idf] Shut the socket down when the stream head is not fully sent 2026-10-30 13:09:28 -10:00
J. Nick Koston 02211e15d8 [web_server_idf] Skip the stall clock for empty writes and clear it on send errors 2026-10-30 12:52:36 -10:00
J. Nick Koston dbbe02f423 [web_server_idf] Include stream_head.h without web_server, for HttpHeader 2026-10-30 12:35:17 -10:00
J. Nick Koston 97a1aa781e [web_server_idf] Simplify the stream response stall clock and head batching
Log the no-memory warning once per stall, free the closing session's buffers once,
add is_stalled() and check that the event stream head fits one write.
2026-10-30 12:22:07 -10:00
J. Nick Koston 0b67a754cd [web_server_idf] Build the stream head tests only for host 2026-10-30 12:15:18 -10:00
J. Nick Koston 320b6a8221 [web_server_idf] Factor the stream response head, iovec advance and stall clock out for host tests
Moves the pure parts of AsyncStreamResponse into stream_head.h (no httpd or
lwIP includes) and adds gtests: the event stream head is compared byte for
byte against what httpd_resp_send_chunk() sent, plus raw heads, batching,
short writes and the stall clock including a millis() rollover.
2026-10-30 12:06:59 -10:00
J. Nick Koston 4e5a5ec32d [web_server_idf] Send the stream response head in one write and tidy the stall clock
Gather the status line, headers and optional body into one iovec list
and send it with a single blocking sendmsg, instead of one httpd_send per
piece. Split the stall clock into start_stall_(), stall_expired_() and
clear_stall_(), and use is_connected() and is_closing() in the event
stream.
2026-10-30 11:40:30 -10:00
J. Nick Koston 09ce8d7d6e [web_server_idf] Add a long-lived streaming response that components can use
Move the session handling of AsyncEventSourceResponse into a new
AsyncStreamResponse base: the response head and session context set up
in the request handler, the non-blocking send override, main loop gather
writes with the send stall clock, the identity-checked close through
httpd_queue_work and the LRU refresh. The event stream now derives from
it, so other components can stream raw or chunked bytes from the main
loop without their own tasks or raw socket calls.

The response head is written directly instead of through
httpd_resp_send_chunk(), so a stream does not have to start with a data
chunk. The event stream still sends the same bytes on the wire.
2026-10-30 11:35:09 -10:00
23 changed files with 1401 additions and 1532 deletions

No files matched your search

+17 -1
View File
@@ -614,9 +614,25 @@ async def to_code(config: ConfigType) -> None:
def FILTER_SOURCE_FILES() -> list[str]:
"""Filter out web_server_v1.cpp when version is not 1."""
"""Filter out web_server_v1.cpp when version is not 1, the ESP32 event stream elsewhere, and the server on host."""
files_to_filter: list[str] = []
# The host has no web server; it builds only sse_chunk.cpp, for its unit tests
if CORE.is_host:
files_to_filter.extend(
(
"event_source.cpp",
"list_entities.cpp",
"web_server.cpp",
"web_server_v1.cpp",
)
)
return files_to_filter
# The /events implementation is ESP32 only
if not CORE.is_esp32:
files_to_filter.extend(("event_source.cpp", "sse_chunk.cpp"))
# web_server_v1.cpp is only needed when version is 1
config = CORE.config.get(DOMAIN, {})
if config.get(CONF_VERSION, 2) != 1:
@@ -0,0 +1,491 @@
#include "event_source.h"
#if defined(USE_ESP32) && defined(USE_WEBSERVER)
#include <algorithm>
#include <cstring>
#include <iterator>
#include "esphome/core/application.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
#include "web_server.h"
namespace esphome::web_server {
ESPHOME_LOG_TAG(TAG, "web_server_idf");
// Only prepare_json_() may hold the JSON arena: every other frame in this file is capped below one
// arena. Measured at -Os on GCC 14; newer toolchains stay checked on purpose, older ones skip it.
#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
#pragma GCC diagnostic error "-Wstack-usage=2048"
#endif
AsyncEventSource::~AsyncEventSource() {
LockGuard guard{this->pending_mutex_};
for (auto *vec : {&this->sessions_, &this->pending_sessions_}) {
for (auto *ses : *vec) {
delete ses; // NOLINT(cppcoreguidelines-owning-memory)
}
}
}
void AsyncEventSource::handleRequest(AsyncWebServerRequest *request) {
// Httpd task: set up the live httpd_req_t and park the session; main loop does the rest.
// NOLINTNEXTLINE(cppcoreguidelines-owning-memory,clang-analyzer-cplusplus.NewDeleteLeaks)
auto *rsp = new AsyncEventSourceResponse(request, this, this->web_server_);
{
LockGuard guard{this->pending_mutex_};
this->pending_sessions_.push_back(rsp);
this->has_pending_sessions_.store(true, std::memory_order_release);
}
this->web_server_->enable_loop_soon_any_context();
}
// clang-analyzer traces a false-positive leak path from loop() through
// adopt_pending_sessions_main_loop_() into start_session_main_loop_() and
// finally ArduinoJson. Suppress along the entire in-our-code call chain.
// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks)
bool AsyncEventSource::loop() {
// Fast path: one atomic load per tick. Slow path is out-of-line on connect.
if (this->has_pending_sessions_.load(std::memory_order_acquire)) {
this->adopt_pending_sessions_main_loop_();
}
// Clean up dead sessions safely
// This follows the ESP-IDF pattern where free_ctx marks resources as dead
// and the main loop handles the actual cleanup to avoid race conditions
for (size_t i = 0; i < this->sessions_.size();) {
auto *ses = this->sessions_[i];
// If the session has a dead socket (marked by destroy callback)
if (ses->safe_to_delete()) {
// destroy() already logged the close with the fd; don't double-log here.
delete ses; // NOLINT(cppcoreguidelines-owning-memory)
// Remove by swapping with last element (O(1) removal, order doesn't matter for sessions)
this->sessions_[i] = this->sessions_.back();
this->sessions_.pop_back();
} else {
ses->loop();
++i;
}
}
// A session still waiting for httpd to commit its context keeps the loop alive too
return !this->sessions_.empty() || this->has_pending_sessions_.load(std::memory_order_acquire);
}
void AsyncEventSource::adopt_pending_sessions_main_loop_() {
std::vector<AsyncEventSourceResponse *> incoming;
{
LockGuard guard{this->pending_mutex_};
incoming.swap(this->pending_sessions_);
this->has_pending_sessions_.store(false, std::memory_order_relaxed);
}
for (auto *rsp : incoming) {
// Already disconnected? Drop it; skip session start on a dead session.
if (rsp->safe_to_delete()) {
delete rsp; // NOLINT(cppcoreguidelines-owning-memory)
continue;
}
// httpd commits the session context only after the creating handler returns, so stay
// pending until then; httpd_req_cleanup() always commits it or calls destroy()
if (!rsp->is_attached()) {
LockGuard guard{this->pending_mutex_};
this->pending_sessions_.push_back(rsp);
this->has_pending_sessions_.store(true, std::memory_order_release);
continue;
}
this->sessions_.push_back(rsp);
rsp->start_session_main_loop_();
}
}
// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
void AsyncEventSource::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
uint32_t reconnect) {
for (auto *ses : this->sessions_) {
if (ses->is_connected()) {
ses->try_send_nodefer(message, message_len, event, id, reconnect);
}
}
}
void AsyncEventSource::deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator) {
// Skip if no connected clients to avoid unnecessary processing
if (this->empty())
return;
for (auto *ses : this->sessions_) {
if (ses->is_connected()) {
ses->deferrable_send_state(source, detail_all, message_generator);
}
}
}
AsyncEventSourceResponse::AsyncEventSourceResponse(AsyncWebServerRequest *request, AsyncEventSource *server,
WebServer *ws)
: server_(server), web_server_(ws), entities_iterator_(ws, this) {
// Httpd task only. start_session_main_loop_() sends the greeting and starts the iterator.
this->begin(request, SSE_STATUS, SSE_CONTENT_TYPE, SSE_HEADERS, SSE_FIRST_CHUNK);
}
static_assert(web_server_base::stream_head_iov_count(std::size(SSE_HEADERS) + WEB_SERVER_DEFAULT_HEADERS_COUNT) <=
web_server_base::STREAM_HEAD_MAX_IOV,
"the event stream head must go out in one write");
// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson
void AsyncEventSourceResponse::start_session_main_loop_() {
auto *ws = this->web_server_;
// The tcp send buffer is empty on connect. A refusal is a closing session or a failed tail
// allocation; nothing retries the greeting, so close and let the client reconnect.
auto message = ws->get_config_json();
if (!this->try_send_nodefer(message.c_str(), message.size(), "ping", millis(), 30000)) {
ESP_LOGW(TAG, "Config not sent to fd %d", this->log_id());
this->close();
return;
}
#ifdef USE_WEBSERVER_SORTING
for (auto &group : ws->sorting_groups_) {
json::JsonBuilder builder;
JsonObject root = builder.root();
root["name"] = json::linked(group.second.name);
root["sorting_weight"] = group.second.weight;
message = builder.serialize();
// a (very) large number of these should be able to be queued initially without defer
// since the only thing in the send buffer at this point is the initial ping/config.
// A refusal means the socket is full or closing; the remaining groups are not sent. The
// session stays up on purpose: partial grouping beats a reconnect loop on a slow link.
if (!this->try_send_nodefer(message.c_str(), message.size(), "sorting_group")) {
ESP_LOGW(TAG, "Sorting groups not sent to fd %d", this->log_id());
break;
}
}
#endif
this->entities_iterator_.begin(ws->include_internal_);
}
// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
// helper for allowing only unique entries in the queue
void AsyncEventSourceResponse::deq_push_back_with_dedup_(void *source, message_generator_t *message_generator) {
DeferredEvent item(source, message_generator);
// Use range-based for loop instead of std::find_if to reduce template instantiation overhead and binary size
for (auto &event : this->deferred_queue_) {
if (event == item) {
return; // Already in queue, no need to update since items are equal
}
}
this->deferred_queue_.push_back(item);
}
void AsyncEventSourceResponse::process_deferred_queue_() {
if (this->is_closing()) {
return;
}
while (!deferred_queue_.empty()) {
DeferredEvent &de = deferred_queue_.front();
if (this->send_json_(de.source_, de.message_generator_)) {
if (this->is_closing() || deferred_queue_.empty()) {
return;
}
// O(n) but memory efficiency is more important than speed here which is why std::vector was chosen
deferred_queue_.erase(deferred_queue_.begin());
} else {
break;
}
}
}
void AsyncEventSourceResponse::release_buffers_() {
std::vector<DeferredEvent>().swap(this->deferred_queue_);
this->tail_.reset();
this->tail_cap_ = 0;
this->tail_len_ = 0;
this->tail_sent_ = 0;
}
void AsyncEventSourceResponse::drain_tail_() {
if (this->sending_ || this->is_closing() || this->tail_len_ == 0) {
return;
}
SendGuard guard{*this};
const size_t remaining = this->tail_len_ - this->tail_sent_;
const ssize_t sent = this->write(this->tail_.get() + this->tail_sent_, remaining);
if (sent <= 0) {
return; // write() owns the stall clock and the close
}
this->no_memory_logged_ = false;
this->tail_sent_ += sent;
if (this->tail_sent_ < this->tail_len_) {
ESP_LOGV(TAG, "Partial send: %zd/%zu bytes (total: %u/%u)", sent, remaining, this->tail_sent_, this->tail_len_);
return;
}
// Fully sent; the storage stays for the next stall
this->tail_len_ = 0;
}
bool AsyncEventSourceResponse::reserve_tail_(size_t len) {
if (this->tail_cap_ >= len) {
return true;
}
if (len > TAIL_MAX_SIZE) {
return false;
}
// Nothing is pending while the tail grows, so free the old block first. PREFER_INTERNAL keeps
// the tail where plain new put it.
this->tail_.reset();
this->tail_ = RAMAllocator<uint8_t>(RAMAllocator<uint8_t>::PREFER_INTERNAL).make_unique_array_for_overwrite(len);
this->tail_cap_ = this->tail_ ? len : 0;
return this->tail_cap_ != 0;
}
bool AsyncEventSourceResponse::stash_chunk_(const char *prefix, size_t prefix_len, const char *message,
size_t message_len, size_t total, size_t sent) {
// A log event of nothing but line breaks is the largest chunk; reserve_tail_() must never
// refuse it for size, only for memory
static_assert(SSE_SEP_LEN * LOG_EVENT_MAX_LEN + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN <= TAIL_MAX_SIZE,
"the log cut in web_server.h must keep a worst case log event inside the tail ceiling");
if (!this->reserve_tail_(total)) {
if (sent != 0) {
// Part of the chunk is on the wire, so the stream is broken and the client has to go
ESP_LOGW(TAG, "Cannot buffer a %zu byte chunk, closing", total);
this->close();
} else {
this->tail_alloc_failed_(total); // nothing on the wire, the caller retries on the stall clock
}
return false;
}
uint8_t *dst = this->tail_.get();
std::memcpy(dst, prefix, prefix_len);
dst += prefix_len;
for_each_chunk_piece(
message, message_len,
[](void *ctx, const char *piece, size_t len) {
auto &out = *static_cast<uint8_t **>(ctx);
std::memcpy(out, piece, len);
out += len;
},
&dst);
this->tail_len_ = total;
this->tail_sent_ = sent;
return true;
}
void AsyncEventSourceResponse::loop() {
AsyncStreamResponse::loop();
if (this->is_closing()) {
if (this->tail_cap_ != 0 || this->deferred_queue_.capacity() != 0) {
this->release_buffers_();
}
return;
}
drain_tail_();
process_deferred_queue_();
if (this->is_closing())
return;
// One step per loop; refusals retry next pass
this->entities_iterator_.try_advance(1);
}
#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wstack-usage=" // the one frame that holds the JSON arena
#endif
AsyncEventSourceResponse::JsonPrepareResult AsyncEventSourceResponse::prepare_json_(void *source,
message_generator_t *generator,
char *buffer, size_t *size_out) {
// Log lines emitted while the generator runs are dropped for this session: on the large path they would fill
// tail_ before it is overwritten, and on the small path they would go out ahead of the event being built.
SendGuard guard{*this};
json::JsonArena<json::JSON_ARENA_SIZE> arena;
json::JsonBuilder builder(&arena);
generator(this->web_server_, source, builder);
*size_out = builder.serialize_to(buffer, JSON_BUF_SIZE);
if (*size_out < JSON_BUF_SIZE) {
return JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER;
}
char prefix[PREFIX_BUF_SIZE];
const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), "state", 0, 0, true);
// Nothing reaches the wire until the complete document and its framing are owned by the caller.
size_t json_len = 0;
size_t cap = std::max<size_t>(JSON_BUF_SIZE * 2, this->tail_cap_);
for (;;) {
if (!this->reserve_tail_(cap)) {
*size_out = cap;
return JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY;
}
const size_t room = cap - prefix_len - SSE_SUFFIX_LEN;
json_len = builder.serialize_to(reinterpret_cast<char *>(this->tail_.get()) + prefix_len, room);
if (json_len < room) {
break;
}
if (cap >= TAIL_MAX_SIZE) {
*size_out = room;
return JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE;
}
cap = std::min<size_t>(cap * 2, TAIL_MAX_SIZE);
}
const size_t total = prefix_len + json_len + SSE_SUFFIX_LEN;
write_chunk_header(prefix, total - CHUNK_HDR_LEN - CHUNK_END_LEN);
uint8_t *dst = this->tail_.get();
std::memcpy(dst, prefix, prefix_len);
std::memcpy(dst + prefix_len + json_len, SSE_SUFFIX, SSE_SUFFIX_LEN);
this->tail_len_ = total;
this->tail_sent_ = 0;
return JsonPrepareResult::JSON_PREPARE_RESULT_TAIL;
}
#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
#pragma GCC diagnostic pop
#endif
bool AsyncEventSourceResponse::send_json_(void *source, message_generator_t *generator) {
if (!this->ready_to_send_()) {
return false;
}
char buffer[JSON_BUF_SIZE];
size_t size;
const auto result = this->prepare_json_(source, generator, buffer, &size);
switch (result) {
case JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER:
return this->try_send_nodefer(buffer, size, "state");
case JsonPrepareResult::JSON_PREPARE_RESULT_TAIL:
this->drain_tail_();
return true;
case JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY: {
SendGuard guard{*this};
this->tail_alloc_failed_(size);
return false;
}
case JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE: {
SendGuard guard{*this};
ESP_LOGW(TAG, "State event does not fit %zu bytes, dropped", size);
this->tail_.reset();
this->tail_cap_ = 0;
this->stall_.clear();
this->no_memory_logged_ = false;
return true;
}
}
return false;
}
void AsyncEventSourceResponse::tail_alloc_failed_(size_t cap) {
// Same stall clock as a socket that stops draining, so a session cannot retry forever. A refused write may
// already have started it, so the warning keeps its own flag.
if (!this->no_memory_logged_) {
this->no_memory_logged_ = true;
ESP_LOGW(TAG, "No memory for a %zu byte chunk", cap);
}
const uint32_t now = App.get_loop_component_start_time();
if (!this->stall_.is_running()) {
this->stall_.start(now);
return;
}
if (this->stall_.is_expired(now, SEND_STALL_TIMEOUT_MS)) {
ESP_LOGW(TAG, "Closing EventSource after %" PRIu32 " ms without memory", this->stall_.elapsed(now));
this->close();
}
}
bool AsyncEventSourceResponse::ready_to_send_() {
if (this->sending_ || !this->is_connected() || this->is_closing()) {
return false;
}
drain_tail_();
return !this->is_closing() && this->tail_len_ == 0;
}
bool AsyncEventSourceResponse::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
uint32_t reconnect) {
if (!this->ready_to_send_()) {
return false;
}
SendGuard guard{*this};
// Everything after the prefix goes out straight from the caller's buffer
char prefix[PREFIX_BUF_SIZE];
const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), event, id, reconnect, message != nullptr);
if (message == nullptr && prefix_len == CHUNK_HDR_LEN) {
return true; // Match ESPAsyncWebServer: nothing to send
}
if (prefix_len >= PREFIX_BUF_SIZE - 1) {
// The appenders truncate silently, which would put a malformed event on the wire
ESP_LOGW(TAG, "Event name too long, dropped");
return true;
}
// Gather list: the prefix, then the data lines and their separators from the message
struct Gather {
web_server_base::StreamIovec iov[MAX_SEND_IOV]; // left uninitialized on purpose
int iovcnt{1};
size_t total{0};
bool fits{true};
} g;
g.total = prefix_len;
for_each_chunk_piece(
message, message_len,
[](void *ctx, const char *piece, size_t len) {
auto &g = *static_cast<Gather *>(ctx);
g.total += len;
if (len == 0) {
return;
}
if (g.iovcnt == MAX_SEND_IOV) {
g.fits = false;
return;
}
g.iov[g.iovcnt++] = {const_cast<char *>(piece), len};
},
&g);
// The header and the terminator are not part of the chunk length
write_chunk_header(prefix, g.total - CHUNK_HDR_LEN - CHUNK_END_LEN);
g.iov[0] = {prefix, prefix_len};
// A message with more lines than the list holds skips straight to the tail
const ssize_t sent = g.fits ? this->write(g.iov, g.iovcnt) : 0;
if (sent < 0) {
return false;
}
if (sent > 0) {
this->no_memory_logged_ = false;
}
if (static_cast<size_t>(sent) == g.total) {
return true;
}
// The caller's buffers do not outlive this call, so keep the chunk and continue from loop()
return this->stash_chunk_(prefix, prefix_len, message, message_len, g.total, sent);
}
void AsyncEventSourceResponse::deferrable_send_state(void *source, bool detail_all,
message_generator_t *message_generator) {
// Let the detail_all listing finish before bare state events, so the web UI never shows unnamed entries and
// initial connect sends less
if (!detail_all && !this->entities_iterator_.completed())
return;
drain_tail_();
process_deferred_queue_();
if (this->is_closing()) {
return;
}
if (this->tail_len_ != 0 || !deferred_queue_.empty()) {
// outgoing event buffer or deferred queue still not empty which means downstream tcp send buffer full, no point
// trying to send first
deq_push_back_with_dedup_(source, message_generator);
} else {
// A send error closes the session and clears the queue; nothing is queued after that
if (!this->send_json_(source, message_generator) && !this->is_closing()) {
deq_push_back_with_dedup_(source, message_generator);
}
}
}
} // namespace esphome::web_server
#endif // USE_ESP32 && USE_WEBSERVER
@@ -0,0 +1,176 @@
#pragma once
#include "esphome/core/defines.h"
#if defined(USE_ESP32) && defined(USE_WEBSERVER)
#include "esphome/components/json/json_util.h"
#include "esphome/components/web_server_base/stream_response.h"
#include "esphome/core/helpers.h"
#include "esphome/core/string_ref.h"
#include "list_entities.h"
#include "sse_chunk.h"
#include <atomic>
#include <cstdint>
#include <vector>
namespace esphome::web_server {
class WebServer;
class AsyncEventSource;
class AsyncEventSourceResponse;
using message_generator_t = void(WebServer *, void *, json::JsonBuilder &);
/*
This class holds a pointer to the source component that wants to publish a state event, and a pointer to a function
that will lazily generate that event. The two pointers allow dedup in the deferred queue if multiple publishes for
the same component are backed up, and take up only two pointers of memory. The entry in the deferred queue (a
std::vector) is the DeferredEvent instance itself (not a pointer to one elsewhere in heap) so still only two pointers
per entry (and no heap fragmentation). Even 100 backed up events (you'd have to have at least 100 sensors publishing
because of dedup) would take up only 0.8 kB.
*/
struct DeferredEvent {
friend class AsyncEventSourceResponse;
protected:
void *source_;
message_generator_t *message_generator_;
public:
DeferredEvent(void *source, message_generator_t *message_generator)
: source_(source), message_generator_(message_generator) {}
bool operator==(const DeferredEvent &test) const {
return (source_ == test.source_ && message_generator_ == test.message_generator_);
}
};
static_assert(sizeof(DeferredEvent) == sizeof(void *) + sizeof(message_generator_t *),
"DeferredEvent should have no padding");
class AsyncEventSourceResponse : public web_server_base::AsyncStreamResponse {
friend class AsyncEventSource;
public:
bool try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
uint32_t reconnect = 0);
// detail_all events go out while entities are still being listed; state events wait until listing is done
void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
// The old event name argument would otherwise convert to true
void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
void loop();
protected:
AsyncEventSourceResponse(AsyncWebServerRequest *request, AsyncEventSource *server, WebServer *ws);
// Main-loop only: sends initial ping/config/sorting_groups, starts entity iterator.
void start_session_main_loop_();
void deq_push_back_with_dedup_(void *source, message_generator_t *message_generator);
void process_deferred_queue_();
// A new chunk may go out: not re-entered from a log line, session alive, tail empty
bool ready_to_send_();
// Push what is left of the chunk in tail_ to the socket
void drain_tail_();
// Grow tail_ to hold len bytes, kept at its high-water mark. False on OOM.
bool reserve_tail_(size_t len);
// Keep the whole chunk in tail_ and continue from sent; false when the tail cannot be allocated.
bool stash_chunk_(const char *prefix, size_t prefix_len, const char *message, size_t message_len, size_t total,
size_t sent);
// Send a state event; JSON too large for the stack buffer is serialized into tail_ instead
bool send_json_(void *source, message_generator_t *generator);
enum class JsonPrepareResult : uint8_t {
JSON_PREPARE_RESULT_BUFFER,
JSON_PREPARE_RESULT_TAIL,
JSON_PREPARE_RESULT_NO_MEMORY,
JSON_PREPARE_RESULT_TOO_LARGE,
};
// Keep the arena out of the caller's frame so it is gone before the state send.
// size_out: BUFFER = JSON length, NO_MEMORY = requested tail capacity, TOO_LARGE = available JSON room;
// unused for TAIL.
JsonPrepareResult __attribute__((noinline)) prepare_json_(void *source, message_generator_t *generator, char *buffer,
size_t *size_out);
// Warn once per stall, and close the session once the stall timeout passes with no memory for the tail
void tail_alloc_failed_(size_t cap);
// A log line emitted inside a send re-enters try_send_nodefer on this session; refuse it
struct SendGuard {
AsyncEventSourceResponse &owner;
explicit SendGuard(AsyncEventSourceResponse &owner) : owner(owner) { owner.sending_ = true; }
~SendGuard() { this->owner.sending_ = false; }
};
// Frees the tail and the queue's storage once the session is closing
void release_buffers_();
AsyncEventSource *server_;
std::vector<DeferredEvent> deferred_queue_;
WebServer *web_server_;
ListEntitiesIterator entities_iterator_;
// One chunk the socket did not take whole, allocated on the first stall; the only heap use
// on the send path
RAMUniquePtr<uint8_t[]> tail_;
uint16_t tail_cap_{0};
uint16_t tail_len_{0}; // Zero means nothing pending
uint16_t tail_sent_{0};
bool sending_{false};
// The no-memory warning went out during the current stall; cleared on send progress
bool no_memory_logged_{false};
// The longest multi line log message in the tree (a climate dump_config) has 22 lines; a
// longer one goes through the tail
static constexpr size_t MAX_SEND_LINES = 22;
static constexpr size_t MAX_SEND_IOV = 1 + 2 * MAX_SEND_LINES;
// Chunk header, retry/id/event lines and the first "data: "
static constexpr size_t PREFIX_BUF_SIZE = 128;
// Stack buffer for a state event's JSON; a larger document is serialized into the tail
static constexpr size_t JSON_BUF_SIZE = 1024;
// Same ceiling JsonBuilder::serialize() applies (max_heap_size in json_util.cpp); a larger
// document is dropped before anything is on the wire
static constexpr size_t JSON_MAX_SIZE = 5120;
// Most RAM a stalled session keeps: the largest state document plus any accepted framing
static constexpr size_t TAIL_MAX_SIZE = JSON_MAX_SIZE + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN;
};
class AsyncEventSource : public AsyncWebHandler {
friend class AsyncEventSourceResponse;
public:
AsyncEventSource(StringRef url, WebServer *ws) : url_(url), web_server_(ws) {}
~AsyncEventSource() override;
// NOLINTNEXTLINE(readability-identifier-naming)
bool canHandle(AsyncWebServerRequest *request) const override {
if (request->method() != HTTP_GET)
return false;
char url_buf[AsyncWebServerRequest::URL_BUF_SIZE];
return request->url_to(url_buf) == this->url_;
}
// NOLINTNEXTLINE(readability-identifier-naming)
void handleRequest(AsyncWebServerRequest *request) override;
void try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
uint32_t reconnect = 0);
void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
// The old event name argument would otherwise convert to true
void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
/// Returns true if there are sessions remaining (including pending cleanup).
bool loop();
bool empty() { return this->count() == 0; }
size_t count() const { return this->sessions_.size(); }
protected:
// Cold path: move sessions from pending_sessions_ into sessions_ and greet each one.
void __attribute__((noinline, cold)) adopt_pending_sessions_main_loop_();
StringRef url_; // Must outlive this object (string literal)
// Main-loop only. Vector: SSE sessions are 1-5 connections, linear search beats set.
std::vector<AsyncEventSourceResponse *> sessions_;
// Httpd-task intake; guarded by pending_mutex_, gated by has_pending_sessions_.
std::vector<AsyncEventSourceResponse *> pending_sessions_;
Mutex pending_mutex_;
WebServer *web_server_;
std::atomic<bool> has_pending_sessions_{false};
};
} // namespace esphome::web_server
#endif // USE_ESP32 && USE_WEBSERVER
@@ -8,14 +8,8 @@
namespace esphome::web_server {
#ifdef USE_ESP32
ListEntitiesIterator::ListEntitiesIterator(const WebServer *ws, AsyncEventSource *es) : web_server_(ws), events_(es) {}
#elif USE_ARDUINO
ListEntitiesIterator::ListEntitiesIterator(const WebServer *ws, DeferredUpdateEventSource *es)
: web_server_(ws), events_(es) {}
#endif
void ListEntitiesIterator::send_all_(void *source, message_generator_t *generator) {
this->events_->deferrable_send_state(source, true, generator);
this->session_->deferrable_send_state(source, true, generator);
}
#ifdef USE_BINARY_SENSOR
+9 -18
View File
@@ -1,33 +1,28 @@
#pragma once
#include "esphome/core/defines.h"
#ifdef USE_WEBSERVER
#if defined(USE_WEBSERVER) && (defined(USE_ESP32) || defined(USE_ARDUINO))
#include "esphome/core/component.h"
#include "esphome/core/component_iterator.h"
namespace esphome::json {
class JsonBuilder;
} // namespace esphome::json
namespace esphome::web_server_idf {
#ifdef USE_ESP32
class AsyncEventSource;
#endif
} // namespace esphome::web_server_idf
namespace esphome::web_server {
#if !defined(USE_ESP32) && defined(USE_ARDUINO)
// The /events session the iterator lists entities to
#ifdef USE_ESP32
class AsyncEventSourceResponse;
using EventSession = AsyncEventSourceResponse;
#elif defined(USE_ARDUINO)
class DeferredUpdateEventSource;
using EventSession = DeferredUpdateEventSource;
#endif
class WebServer;
class ListEntitiesIterator final : public ComponentIterator {
public:
#ifdef USE_ESP32
ListEntitiesIterator(const WebServer *ws, esphome::web_server_idf::AsyncEventSource *es);
#elif defined(USE_ARDUINO)
ListEntitiesIterator(const WebServer *ws, DeferredUpdateEventSource *es);
#endif
ListEntitiesIterator(const WebServer *ws, EventSession *session) : web_server_(ws), session_(session) {}
// Entity overrides (generated from entity_types.h).
// Implementations live in list_entities.cpp.
@@ -45,11 +40,7 @@ class ListEntitiesIterator final : public ComponentIterator {
void send_all_(void *source, void (*generator)(WebServer *, void *, json::JsonBuilder &));
const WebServer *web_server_;
#ifdef USE_ESP32
esphome::web_server_idf::AsyncEventSource *events_;
#elif USE_ARDUINO
DeferredUpdateEventSource *events_;
#endif
EventSession *session_;
};
} // namespace esphome::web_server
@@ -5,7 +5,7 @@
#include "esphome/core/helpers.h"
namespace esphome::web_server_idf {
namespace esphome::web_server {
void for_each_chunk_piece(const char *message, size_t message_len, ChunkPieceSink sink, void *ctx) {
if (message == nullptr) {
@@ -63,4 +63,4 @@ void write_chunk_header(char *buf, size_t chunk_len) {
buf[9] = '\n';
}
} // namespace esphome::web_server_idf
} // namespace esphome::web_server
@@ -3,7 +3,9 @@
#include <cstddef>
#include <cstdint>
namespace esphome::web_server_idf {
#include "esphome/components/web_server_base/stream_head.h"
namespace esphome::web_server {
// Wire framing of one Server-Sent-Events event inside the chunked /events response:
// "%08x\r\n" + fields + "data: " line ["\r\ndata: " line]... "\r\n" + "\r\n" + "\r\n"
@@ -22,6 +24,17 @@ constexpr size_t SSE_SEP_LEN = sizeof(SSE_SEP) - 1;
constexpr char SSE_SUFFIX[] = "\r\n\r\n\r\n";
constexpr size_t SSE_SUFFIX_LEN = sizeof(SSE_SUFFIX) - 1;
// Status, Content-Type and headers of the /events response head
constexpr char SSE_STATUS[] = "200 OK";
constexpr char SSE_CONTENT_TYPE[] = "text/event-stream";
constexpr web_server_base::HttpHeader SSE_HEADERS[] = {
{"Transfer-Encoding", "chunked"},
{"Cache-Control", "no-cache"},
{"Connection", "keep-alive"},
};
// Hand-framed first chunk holding an empty event (one blank line), as httpd_resp_send_chunk() sent it before
constexpr char SSE_FIRST_CHUNK[] = "2\r\n\r\n\r\n";
// Receives one piece of the chunk; ctx is whatever the caller passed to for_each_chunk_piece()
using ChunkPieceSink = void (*)(void *ctx, const char *piece, size_t len);
@@ -37,4 +50,4 @@ size_t build_chunk_prefix(char *buf, size_t size, const char *event, uint32_t id
// Fills the header placeholder at the start of buf once the chunk length is known
void write_chunk_header(char *buf, size_t chunk_len);
} // namespace esphome::web_server_idf
} // namespace esphome::web_server
+1 -1
View File
@@ -245,7 +245,7 @@ void DeferredUpdateEventSource::process_deferred_queue_() {
deferred_queue_.erase(deferred_queue_.begin());
this->consecutive_send_failures_ = 0; // Reset failure count on successful send
} else {
// NOTE: Similar logic exists in web_server_idf/web_server_idf.cpp in AsyncEventSourceResponse::drain_tail_().
// NOTE: Similar logic exists in web_server_idf/web_server_idf.cpp in AsyncStreamResponse::write().
// The close mechanisms are platform-specific (this path calls close() directly; the IDF path is time-based and
// closes through HTTPD to preserve session ownership), but both drop a client after roughly 20 seconds without
// send progress. Keep that stall policy in sync when changing either side.
+5 -1
View File
@@ -5,7 +5,11 @@
#include "esphome/components/json/json_util.h"
#include "esphome/components/web_server_base/web_server_base.h"
#include "esphome/components/web_server_base/captive_dns.h"
#ifdef USE_WEBSERVER
// The host only builds sse_chunk.cpp for unit tests, but other tests can define USE_WEBSERVER there
#if defined(USE_WEBSERVER) && (defined(USE_ESP32) || defined(USE_ARDUINO))
#ifdef USE_ESP32
#include "event_source.h"
#endif
#include "esphome/core/component.h"
#include "esphome/core/entity_base.h"
#include "esphome/core/entity_includes.h"
@@ -105,9 +105,9 @@ async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
cg.add(cg.RawExpression(f"{web_server_base_ns}::global_web_server_base = {var}"))
# Default headers added in init(): web_server_idf's StaticVector, or the library's DefaultHeaders
cg.add_define("WEB_SERVER_DEFAULT_HEADERS_COUNT", 1)
if CORE.is_esp32:
# Count for StaticVector in web_server_idf - matches headers added in init()
cg.add_define("WEB_SERVER_DEFAULT_HEADERS_COUNT", 1)
return
# ESP32 uses IDF web server (early return above), so this is for other Arduino platforms
@@ -0,0 +1,126 @@
#pragma once
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <span>
#include <strings.h>
// Pure helpers of AsyncStreamResponse, kept free of httpd and lwIP so they build on the host.
// The iovec type is a template parameter: lwIP defines its own on the device, POSIX on the host.
namespace esphome::web_server_base {
struct HttpHeader {
const char *name;
const char *value;
};
/// Drops the first sent bytes from a gather list after a write; a short write resumes inside the
/// element it stopped in. count reaches 0 once everything went out.
template<typename Iovec> void advance_iovecs(Iovec *&iov, int &count, size_t sent) {
while (count > 0 && sent >= iov->iov_len) {
sent -= iov->iov_len;
++iov;
--count;
}
if (count > 0) {
iov->iov_base = static_cast<char *>(iov->iov_base) + sent;
iov->iov_len -= sent;
}
}
/// Pieces in a head: 4 for the status line and Content-Type, 4 per header, the blank line and the body
constexpr size_t stream_head_iov_count(size_t header_count) { return 4 + 4 * header_count + 2; }
/// Pieces per batch: a head with up to four headers goes out in one write; 192 bytes on 32-bit
static constexpr int STREAM_HEAD_MAX_IOV = 24;
/// True if headers holds a header named name, compared case insensitively
inline bool has_header(std::span<const HttpHeader> headers, const char *name) {
for (const auto &header : headers) {
if (strcasecmp(header.name, name) == 0) {
return true;
}
}
return false;
}
/// Gathers the head of a stream response: "HTTP/1.1 <status>", Content-Type, headers, the default_headers that
/// headers does not replace, the blank line and body (may be null). send(Iovec *iov, int count) gets each full batch of
/// STREAM_HEAD_MAX_IOV pieces and then the rest; once it returns false the later pieces are skipped. Returns false if
/// any send failed.
template<typename Iovec, typename Send>
bool write_stream_head(Send &&send, const char *status, const char *content_type, std::span<const HttpHeader> headers,
std::span<const HttpHeader> default_headers, const char *body) {
struct Batch {
Send &send;
int count{0};
bool ok{true};
Iovec iov[STREAM_HEAD_MAX_IOV]; // left uninitialized on purpose
void flush() {
if (this->ok && this->count > 0) {
this->ok = this->send(this->iov, this->count);
}
this->count = 0;
}
void add(const char *buf, size_t len) {
if (this->count == STREAM_HEAD_MAX_IOV) {
this->flush();
}
this->iov[this->count++] = {const_cast<char *>(buf), len};
}
void add(const char *str) { this->add(str, strlen(str)); }
void add_header(const HttpHeader &header) {
this->add("\r\n", 2);
this->add(header.name);
this->add(": ", 2);
this->add(header.value);
}
} batch{send};
batch.add("HTTP/1.1 ", 9);
batch.add(status);
batch.add("\r\nContent-Type: ", 16);
batch.add(content_type);
for (const auto &header : headers) {
batch.add_header(header);
}
for (const auto &header : default_headers) {
if (!has_header(headers, header.name)) {
batch.add_header(header);
}
}
batch.add("\r\n\r\n", 4);
if (body != nullptr) {
batch.add(body);
}
batch.flush();
return batch.ok;
}
/// How long sends have made no progress. Zero means not running, so a start at millis() == 0 records 1.
/// The comparison is wraparound safe across a millis() rollover.
class StallClock {
public:
/// Starts the clock; a running clock keeps its start time.
void start(uint32_t now) {
if (this->started_ms_ == 0) {
this->started_ms_ = now != 0 ? now : 1;
}
}
void clear() { this->started_ms_ = 0; }
bool is_running() const { return this->started_ms_ != 0; }
/// True once the clock has run for timeout_ms
bool is_expired(uint32_t now, uint32_t timeout_ms) const {
return this->started_ms_ != 0 && static_cast<int32_t>(now - (this->started_ms_ + timeout_ms)) >= 0;
}
uint32_t elapsed(uint32_t now) const { return now - this->started_ms_; }
protected:
uint32_t started_ms_{0};
};
} // namespace esphome::web_server_base
@@ -0,0 +1,16 @@
#pragma once
#include "esphome/core/defines.h"
// The long lived stream response the /events layer writes to, for the platform's web server
#ifdef USE_ESP32
#include "esphome/components/web_server_idf/web_server_idf.h"
#include <sys/socket.h>
namespace esphome::web_server_base {
using web_server_idf::AsyncStreamResponse;
// One entry of the gather list AsyncStreamResponse::write() takes
using StreamIovec = ::iovec;
} // namespace esphome::web_server_base
#endif
@@ -17,7 +17,6 @@
#include <freertos/task.h>
#include "utils.h"
#include "sse_chunk.h"
#include "web_server_idf.h"
#ifdef USE_WEBSERVER_AUTH_DIGEST
@@ -30,17 +29,15 @@
#include "multipart.h" // For parse_multipart_boundary and other utils
#endif
#ifdef USE_WEBSERVER
#include "esphome/components/web_server/web_server.h"
#include "esphome/components/web_server/list_entities.h"
#endif // USE_WEBSERVER
// Include socket headers after Arduino headers to avoid IPADDR_NONE/INADDR_NONE macro conflicts
#include <cerrno>
#include <sys/socket.h>
namespace esphome::web_server_idf {
using web_server_base::advance_iovecs;
using web_server_base::write_stream_head;
// Status strings not provided by esp_http_server.h
#ifndef HTTPD_401
#define HTTPD_401 "401 Unauthorized"
@@ -55,13 +52,10 @@ namespace esphome::web_server_idf {
#define HTTPD_422 "422 Unprocessable Entity"
#endif
#define CRLF_STR "\r\n"
#define CRLF_LEN (sizeof(CRLF_STR) - 1)
ESPHOME_LOG_TAG(TAG, "web_server_idf");
// Only prepare_json_() may hold the JSON arena: every other frame in this file is capped below one
// arena. Measured at -Os on GCC 14; newer toolchains stay checked on purpose, older ones skip it.
// Every frame in this file is capped below one JSON arena.
// Measured at -Os on GCC 14; newer toolchains stay checked on purpose, older ones skip it.
#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
#pragma GCC diagnostic error "-Wstack-usage=2048"
#endif
@@ -140,6 +134,8 @@ void AsyncWebServer::safe_close_with_shutdown(httpd_handle_t hd, int sockfd) {
//
// See: https://github.com/esphome/esphome-webserver/issues/163
AsyncStreamResponse::socket_closing(hd, sockfd);
// Attempt shutdown - ignore errors as socket may already be closed
shutdown(sockfd, SHUT_RD);
@@ -736,224 +732,84 @@ void AsyncResponseStream::printf(const char *fmt, ...) {
this->print(str);
}
#ifdef USE_WEBSERVER
AsyncEventSource::~AsyncEventSource() {
LockGuard guard{this->pending_mutex_};
for (auto *vec : {&this->sessions_, &this->pending_sessions_}) {
for (auto *ses : *vec) {
delete ses; // NOLINT(cppcoreguidelines-owning-memory)
}
}
}
void AsyncEventSource::handleRequest(AsyncWebServerRequest *request) {
// Httpd task: set up the live httpd_req_t and park the session; main loop does the rest.
// NOLINTNEXTLINE(cppcoreguidelines-owning-memory,clang-analyzer-cplusplus.NewDeleteLeaks)
auto *rsp = new AsyncEventSourceResponse(request, this, this->web_server_);
{
LockGuard guard{this->pending_mutex_};
this->pending_sessions_.push_back(rsp);
this->has_pending_sessions_.store(true, std::memory_order_release);
}
this->web_server_->enable_loop_soon_any_context();
}
// clang-analyzer traces a false-positive leak path from loop() through
// adopt_pending_sessions_main_loop_() into start_session_main_loop_() and
// finally ArduinoJson. Suppress along the entire in-our-code call chain.
// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks)
bool AsyncEventSource::loop() {
// Fast path: one atomic load per tick. Slow path is out-of-line on connect.
if (this->has_pending_sessions_.load(std::memory_order_acquire)) {
this->adopt_pending_sessions_main_loop_();
}
// Clean up dead sessions safely
// This follows the ESP-IDF pattern where free_ctx marks resources as dead
// and the main loop handles the actual cleanup to avoid race conditions
for (size_t i = 0; i < this->sessions_.size();) {
auto *ses = this->sessions_[i];
// If the session has a dead socket (marked by destroy callback)
if (ses->safe_to_delete_()) {
// destroy() already logged the close with the fd; don't double-log here.
delete ses; // NOLINT(cppcoreguidelines-owning-memory)
// Remove by swapping with last element (O(1) removal, order doesn't matter for sessions)
this->sessions_[i] = this->sessions_.back();
this->sessions_.pop_back();
} else {
ses->loop();
++i;
}
}
// A session still waiting for httpd to commit its context keeps the loop alive too
return !this->sessions_.empty() || this->has_pending_sessions_.load(std::memory_order_acquire);
}
void AsyncEventSource::adopt_pending_sessions_main_loop_() {
std::vector<AsyncEventSourceResponse *> incoming;
{
LockGuard guard{this->pending_mutex_};
incoming.swap(this->pending_sessions_);
this->has_pending_sessions_.store(false, std::memory_order_relaxed);
}
for (auto *rsp : incoming) {
// Already disconnected? Drop it; skip session start on a dead session.
if (rsp->safe_to_delete_()) {
delete rsp; // NOLINT(cppcoreguidelines-owning-memory)
continue;
}
// httpd commits the session context only after the creating handler returns, so stay
// pending until then; httpd_req_cleanup() always commits it or calls destroy()
if (httpd_sess_get_ctx(rsp->hd_, rsp->fd_.load()) != rsp) {
LockGuard guard{this->pending_mutex_};
this->pending_sessions_.push_back(rsp);
this->has_pending_sessions_.store(true, std::memory_order_release);
continue;
}
this->sessions_.push_back(rsp);
rsp->start_session_main_loop_();
}
}
// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
void AsyncEventSource::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
uint32_t reconnect) {
for (auto *ses : this->sessions_) {
if (ses->fd_.load() != 0) { // Skip dead sessions
ses->try_send_nodefer(message, message_len, event, id, reconnect);
}
}
}
void AsyncEventSource::deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator) {
// Skip if no connected clients to avoid unnecessary processing
if (this->empty())
return;
for (auto *ses : this->sessions_) {
if (ses->fd_.load() != 0) { // Skip dead sessions
ses->deferrable_send_state(source, detail_all, message_generator);
}
}
}
AsyncEventSourceResponse::AsyncEventSourceResponse(const AsyncWebServerRequest *request,
esphome::web_server_idf::AsyncEventSource *server,
esphome::web_server::WebServer *ws)
: server_(server), web_server_(ws), entities_iterator_(ws, server) {
// Httpd task only. start_session_main_loop_() sends the greeting and starts the iterator.
void AsyncStreamResponse::begin(AsyncWebServerRequest *request, const char *status, const char *content_type,
std::span<const HttpHeader> headers, const char *body) {
httpd_req_t *req = *request;
httpd_resp_set_status(req, HTTPD_200);
httpd_resp_set_type(req, "text/event-stream");
httpd_resp_set_hdr(req, "Cache-Control", "no-cache");
httpd_resp_set_hdr(req, "Connection", "keep-alive");
DefaultHeaders::Instance().apply_(req);
httpd_resp_send_chunk(req, CRLF_STR, CRLF_LEN);
// Same head httpd_resp_send_chunk() writes, without its first chunk. Blocking sends: the session's non-blocking
// send override is not installed yet.
const int fd = httpd_req_to_sockfd(req);
const auto &defaults = DefaultHeaders::Instance().headers_;
const bool head_sent = write_stream_head<struct iovec>(
[fd](struct iovec *iov, int count) {
while (count > 0) {
struct msghdr msg{};
msg.msg_iov = iov;
msg.msg_iovlen = count;
const ssize_t sent = sendmsg(fd, &msg, 0);
if (sent <= 0) {
return false;
}
advance_iovecs(iov, count, sent);
}
return true;
},
status, content_type, headers, std::span<const HttpHeader>(defaults.data(), defaults.size()), body);
req->sess_ctx = this;
req->free_ctx = AsyncEventSourceResponse::destroy;
req->free_ctx = AsyncStreamResponse::destroy;
this->hd_ = req->handle;
this->fd_.store(httpd_req_to_sockfd(req));
this->fd_.store(fd);
// Use non-blocking send to prevent watchdog timeouts when TCP buffers are full
httpd_sess_set_send_override(this->hd_, this->fd_.load(), nonblocking_send);
httpd_sess_set_send_override(this->hd_, fd, nonblocking_send);
if (!head_sent) {
// A partly sent head must not be continued by later writes. close() is main loop only, so shut the socket
// down here; the first write() then fails and closes the stream, and httpd frees the session as usual.
shutdown(fd, SHUT_RDWR);
}
}
// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson
void AsyncEventSourceResponse::start_session_main_loop_() {
auto *ws = this->web_server_;
// Held around a stream's ownership check and send, and while its socket is marked closed.
// Created at boot, so no allocation can fail once requests arrive.
static Mutex stream_send_mutex; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
// The tcp send buffer is empty on connect. A refusal is a closing session or a failed tail
// allocation; nothing retries the greeting, so close and let the client reconnect.
auto message = ws->get_config_json();
if (!this->try_send_nodefer(message.c_str(), message.size(), "ping", millis(), 30000)) {
ESP_LOGW(TAG, "Config not sent to fd %d", this->fd_.load());
this->request_close_();
void AsyncStreamResponse::socket_closing(httpd_handle_t hd, int fd) {
// Only streams set a session context
auto *rsp = static_cast<AsyncStreamResponse *>(httpd_sess_get_ctx(hd, fd));
if (rsp == nullptr)
return;
}
#ifdef USE_WEBSERVER_SORTING
for (auto &group : ws->sorting_groups_) {
json::JsonBuilder builder;
JsonObject root = builder.root();
root["name"] = json::linked(group.second.name);
root["sorting_weight"] = group.second.weight;
message = builder.serialize();
// a (very) large number of these should be able to be queued initially without defer
// since the only thing in the send buffer at this point is the initial ping/config.
// A refusal means the socket is full or closing; the remaining groups are not sent. The
// session stays up on purpose: partial grouping beats a reconnect loop on a slow link.
if (!this->try_send_nodefer(message.c_str(), message.size(), "sorting_group")) {
ESP_LOGW(TAG, "Sorting groups not sent to fd %d", this->fd_.load());
break;
}
}
#endif
this->entities_iterator_.begin(ws->include_internal_);
LockGuard guard{stream_send_mutex};
rsp->socket_closed_ = true;
}
// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
void AsyncEventSourceResponse::destroy(void *ptr) {
auto *rsp = static_cast<AsyncEventSourceResponse *>(ptr);
void AsyncStreamResponse::destroy(void *ptr) {
auto *rsp = static_cast<AsyncStreamResponse *>(ptr);
int fd = rsp->fd_.exchange(0); // Atomically get and clear fd
ESP_LOGD(TAG, "Event source connection closed (fd: %d)", fd);
// Mark as dead - will be cleaned up in the main loop
// Note: We don't delete or remove from set here to avoid race conditions
ESP_LOGD(TAG, "Stream connection closed (fd: %d)", fd);
// Mark as dead - the owner deletes it from the main loop once safe_to_delete().
// httpd will call our custom close_fn (safe_close_with_shutdown) which handles
// shutdown() before close() to prevent lwIP race conditions
}
// helper for allowing only unique entries in the queue
void AsyncEventSourceResponse::deq_push_back_with_dedup_(void *source, message_generator_t *message_generator) {
DeferredEvent item(source, message_generator);
// Use range-based for loop instead of std::find_if to reduce template instantiation overhead and binary size
for (auto &event : this->deferred_queue_) {
if (event == item) {
return; // Already in queue, no need to update since items are equal
}
}
this->deferred_queue_.push_back(item);
}
void AsyncEventSourceResponse::process_deferred_queue_() {
if (this->close_requested_) {
return;
}
while (!deferred_queue_.empty()) {
DeferredEvent &de = deferred_queue_.front();
if (this->send_json_(de.source_, de.message_generator_)) {
if (this->close_requested_ || deferred_queue_.empty()) {
return;
}
// O(n) but memory efficiency is more important than speed here which is why std::vector was chosen
deferred_queue_.erase(deferred_queue_.begin());
} else {
break;
}
}
}
void AsyncEventSourceResponse::request_close_() {
void AsyncStreamResponse::close() {
if (!this->close_requested_) {
this->close_requested_ = true;
this->deferred_queue_.clear();
this->tail_.reset();
this->tail_cap_ = 0;
this->tail_len_ = 0;
this->tail_sent_ = 0;
this->next_close_attempt_ms_ = App.get_loop_component_start_time();
}
this->process_close_();
}
void AsyncEventSourceResponse::process_close_() {
void AsyncStreamResponse::loop() {
if (this->close_requested_) {
this->process_close_();
} else {
this->refresh_lru_();
}
}
void AsyncStreamResponse::process_close_() {
if (!this->close_requested_ || this->close_work_queued_.load(std::memory_order_acquire)) {
return;
}
@@ -971,7 +827,7 @@ void AsyncEventSourceResponse::process_close_() {
// httpd_sess_trigger_close() queues only a reusable fd/session slot and can
// therefore close a new client if the original peer disconnects meanwhile.
this->close_work_queued_.store(true, std::memory_order_release);
const esp_err_t err = httpd_queue_work(this->hd_, &AsyncEventSourceResponse::close_session_work, this);
const esp_err_t err = httpd_queue_work(this->hd_, &AsyncStreamResponse::close_session_work, this);
this->next_close_attempt_ms_ = now + (err == ESP_OK ? CLOSE_CONFIRM_INTERVAL_MS : CLOSE_RETRY_INTERVAL_MS);
if (err == ESP_OK) {
return;
@@ -979,17 +835,17 @@ void AsyncEventSourceResponse::process_close_() {
this->close_work_queued_.store(false, std::memory_order_release);
if (!this->close_retry_warning_logged_) {
ESP_LOGW(TAG, "Failed to queue EventSource close (%s); retrying", esp_err_to_name(err));
ESP_LOGW(TAG, "Failed to queue stream close (%s); retrying", esp_err_to_name(err));
this->close_retry_warning_logged_ = true;
}
}
void AsyncEventSourceResponse::refresh_lru_() {
// httpd only advances a session's LRU counter when a request arrives, and an event stream receives one
void AsyncStreamResponse::refresh_lru_() {
// httpd only advances a session's LRU counter when a request arrives, and a 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)
if (this->stall_.is_running() || 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 ||
@@ -998,13 +854,13 @@ void AsyncEventSourceResponse::refresh_lru_() {
}
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) {
if (httpd_queue_work(this->hd_, &AsyncStreamResponse::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);
void AsyncStreamResponse::refresh_lru_work(void *arg) {
auto *response = static_cast<AsyncStreamResponse *>(arg);
const int fd = response->fd_.load();
if (response->owns_session_(fd)) {
httpd_sess_update_lru_counter(response->hd_, fd);
@@ -1013,8 +869,8 @@ void AsyncEventSourceResponse::refresh_lru_work(void *arg) {
response->lru_work_queued_.store(false, std::memory_order_release);
}
void AsyncEventSourceResponse::close_session_work(void *arg) {
auto *response = static_cast<AsyncEventSourceResponse *>(arg);
void AsyncStreamResponse::close_session_work(void *arg) {
auto *response = static_cast<AsyncStreamResponse *>(arg);
const int fd = response->fd_.load();
if (response->owns_session_(fd)) {
// The HTTPD task remains the session owner. Shutting the socket down makes
@@ -1026,320 +882,61 @@ void AsyncEventSourceResponse::close_session_work(void *arg) {
response->close_work_queued_.store(false, std::memory_order_release);
}
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.
ssize_t AsyncStreamResponse::write(struct iovec *iov, int iovcnt) {
if (this->close_requested_) {
return -1;
}
// Nothing to send is not a stall
size_t total = 0;
for (int i = 0; i < iovcnt; i++) {
total += iov[i].iov_len;
}
if (total == 0) {
return 0;
}
// httpd closes the socket before it clears the session, so the fd number 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 (!this->owns_session_(fd)) {
return 0;
ssize_t sent = 0;
int err = 0;
{
// Held across the check and the send, so socket_closing() cannot close the socket in between
LockGuard guard{stream_send_mutex};
if (!this->socket_closed_ && this->owns_session_(fd)) {
struct msghdr msg{};
msg.msg_iov = iov;
msg.msg_iovlen = iovcnt;
sent = sendmsg(fd, &msg, MSG_DONTWAIT);
err = errno;
}
}
struct msghdr msg{};
msg.msg_iov = iov;
msg.msg_iovlen = iovcnt;
const ssize_t sent = sendmsg(fd, &msg, MSG_DONTWAIT);
if (sent >= 0) {
if (sent > 0) {
this->stall_.clear();
return sent;
}
const int err = errno;
if (err == EAGAIN || err == EWOULDBLOCK) {
return 0;
if (sent < 0 && err != EAGAIN && err != EWOULDBLOCK) {
ESP_LOGD(TAG, "send error: errno %d", err);
// A hard error is not a timeout, so is_stalled() must not report one
this->stall_.clear();
this->close();
return -1;
}
ESP_LOGD(TAG, "send error: errno %d", err);
this->request_close_();
return -1;
}
void AsyncEventSourceResponse::drain_tail_() {
if (this->sending_ || this->close_requested_ || this->tail_len_ == 0) {
return;
}
SendGuard guard{*this};
const size_t remaining = this->tail_len_ - this->tail_sent_;
struct iovec iov = {this->tail_.get() + this->tail_sent_, remaining};
const ssize_t sent = this->send_(&iov, 1);
if (sent < 0) {
return;
}
if (sent == 0) {
// Socket buffer full, try again later
// NOTE: Similar logic exists in web_server/web_server.cpp in DeferredUpdateEventSource::process_deferred_queue_().
// The IDF path is intentionally time-based and closes through HTTPD to preserve session ownership.
const uint32_t now = App.get_loop_component_start_time();
if (this->send_failure_started_ms_ == 0) {
this->send_failure_started_ms_ = now != 0 ? now : 1; // Reserve zero for no stall.
}
if (static_cast<int32_t>(now - (this->send_failure_started_ms_ + SEND_STALL_TIMEOUT_MS)) >= 0) {
ESP_LOGW(TAG, "Closing stuck EventSource connection after %" PRIu32 " ms without send progress",
now - this->send_failure_started_ms_);
this->request_close_();
}
return;
}
this->send_failure_started_ms_ = 0;
this->tail_sent_ += sent;
if (this->tail_sent_ < this->tail_len_) {
ESP_LOGV(TAG, "Partial send: %zd/%zu bytes (total: %u/%u)", sent, remaining, this->tail_sent_, this->tail_len_);
return;
}
// Fully sent; the storage stays for the next stall
this->tail_len_ = 0;
}
bool AsyncEventSourceResponse::reserve_tail_(size_t len) {
if (this->tail_cap_ >= len) {
return true;
}
if (len > TAIL_MAX_SIZE) {
return false;
}
// Nothing is pending while the tail grows, so free the old block first. PREFER_INTERNAL keeps
// the tail where plain new put it.
this->tail_.reset();
this->tail_ = RAMAllocator<uint8_t>(RAMAllocator<uint8_t>::PREFER_INTERNAL).make_unique_array_for_overwrite(len);
this->tail_cap_ = this->tail_ ? len : 0;
return this->tail_cap_ != 0;
}
bool AsyncEventSourceResponse::stash_chunk_(const char *prefix, size_t prefix_len, const char *message,
size_t message_len, size_t total, size_t sent) {
// A log event of nothing but line breaks is the largest chunk; reserve_tail_() must never
// refuse it for size, only for memory
static_assert(SSE_SEP_LEN * web_server::LOG_EVENT_MAX_LEN + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN <= TAIL_MAX_SIZE,
"the log cut in web_server.h must keep a worst case log event inside the tail ceiling");
if (!this->reserve_tail_(total)) {
if (sent != 0) {
// Part of the chunk is on the wire, so the stream is broken and the client has to go
ESP_LOGW(TAG, "Cannot buffer a %zu byte chunk, closing", total);
this->request_close_();
} else {
this->tail_alloc_failed_(total); // nothing on the wire, the caller retries on the stall clock
}
return false;
}
uint8_t *dst = this->tail_.get();
std::memcpy(dst, prefix, prefix_len);
dst += prefix_len;
for_each_chunk_piece(
message, message_len,
[](void *ctx, const char *piece, size_t len) {
auto &out = *static_cast<uint8_t **>(ctx);
std::memcpy(out, piece, len);
out += len;
},
&dst);
this->tail_len_ = total;
this->tail_sent_ = sent;
return true;
}
void AsyncEventSourceResponse::loop() {
if (this->close_requested_) {
this->process_close_();
return;
}
this->refresh_lru_();
drain_tail_();
process_deferred_queue_();
if (this->close_requested_)
return;
// One step per loop; refusals retry next pass
this->entities_iterator_.try_advance(1);
}
#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wstack-usage=" // the one frame that holds the JSON arena
#endif
AsyncEventSourceResponse::JsonPrepareResult AsyncEventSourceResponse::prepare_json_(void *source,
message_generator_t *generator,
char *buffer, size_t *size_out) {
// Log lines emitted while the generator runs are dropped for this session: on the large path they would fill
// tail_ before it is overwritten, and on the small path they would go out ahead of the event being built.
SendGuard guard{*this};
json::JsonArena<json::JSON_ARENA_SIZE> arena;
json::JsonBuilder builder(&arena);
generator(this->web_server_, source, builder);
*size_out = builder.serialize_to(buffer, JSON_BUF_SIZE);
if (*size_out < JSON_BUF_SIZE) {
return JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER;
}
char prefix[PREFIX_BUF_SIZE];
const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), "state", 0, 0, true);
// Nothing reaches the wire until the complete document and its framing are owned by the caller.
size_t json_len = 0;
size_t cap = std::max<size_t>(JSON_BUF_SIZE * 2, this->tail_cap_);
for (;;) {
if (!this->reserve_tail_(cap)) {
*size_out = cap;
return JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY;
}
const size_t room = cap - prefix_len - SSE_SUFFIX_LEN;
json_len = builder.serialize_to(reinterpret_cast<char *>(this->tail_.get()) + prefix_len, room);
if (json_len < room) {
break;
}
if (cap >= TAIL_MAX_SIZE) {
*size_out = room;
return JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE;
}
cap = std::min<size_t>(cap * 2, TAIL_MAX_SIZE);
}
const size_t total = prefix_len + json_len + SSE_SUFFIX_LEN;
write_chunk_header(prefix, total - CHUNK_HDR_LEN - CHUNK_END_LEN);
uint8_t *dst = this->tail_.get();
std::memcpy(dst, prefix, prefix_len);
std::memcpy(dst + prefix_len + json_len, SSE_SUFFIX, SSE_SUFFIX_LEN);
this->tail_len_ = total;
this->tail_sent_ = 0;
return JsonPrepareResult::JSON_PREPARE_RESULT_TAIL;
}
#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
#pragma GCC diagnostic pop
#endif
bool AsyncEventSourceResponse::send_json_(void *source, message_generator_t *generator) {
if (!this->ready_to_send_()) {
return false;
}
char buffer[JSON_BUF_SIZE];
size_t size;
const auto result = this->prepare_json_(source, generator, buffer, &size);
switch (result) {
case JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER:
return this->try_send_nodefer(buffer, size, "state");
case JsonPrepareResult::JSON_PREPARE_RESULT_TAIL:
this->drain_tail_();
return true;
case JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY: {
SendGuard guard{*this};
this->tail_alloc_failed_(size);
return false;
}
case JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE: {
SendGuard guard{*this};
ESP_LOGW(TAG, "State event does not fit %zu bytes, dropped", size);
this->tail_.reset();
this->tail_cap_ = 0;
this->send_failure_started_ms_ = 0;
return true;
}
}
return false;
}
void AsyncEventSourceResponse::tail_alloc_failed_(size_t cap) {
// Same stall clock as a socket that stops draining, so a session cannot retry forever
// Socket buffer full, try again later; a socket that stays full past the stall timeout is closed
const uint32_t now = App.get_loop_component_start_time();
if (this->send_failure_started_ms_ == 0) {
this->send_failure_started_ms_ = now != 0 ? now : 1; // Reserve zero for no stall.
ESP_LOGW(TAG, "No memory for a %zu byte chunk", cap);
return;
}
if (static_cast<int32_t>(now - (this->send_failure_started_ms_ + SEND_STALL_TIMEOUT_MS)) >= 0) {
ESP_LOGW(TAG, "Closing EventSource after %" PRIu32 " ms without memory", now - this->send_failure_started_ms_);
this->request_close_();
this->stall_.start(now);
if (this->stall_.is_expired(now, SEND_STALL_TIMEOUT_MS)) {
ESP_LOGW(TAG, "Closing stuck stream after %" PRIu32 " ms without send progress", this->stall_.elapsed(now));
this->close();
return -1;
}
return 0;
}
bool AsyncEventSourceResponse::ready_to_send_() {
if (this->sending_ || this->fd_.load() == 0 || this->close_requested_) {
return false;
}
drain_tail_();
return !this->close_requested_ && this->tail_len_ == 0;
ssize_t AsyncStreamResponse::write(const void *data, size_t len) {
struct iovec iov = {const_cast<void *>(data), len};
return this->write(&iov, 1);
}
bool AsyncEventSourceResponse::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
uint32_t reconnect) {
if (!this->ready_to_send_()) {
return false;
}
SendGuard guard{*this};
// Everything after the prefix goes out straight from the caller's buffer
char prefix[PREFIX_BUF_SIZE];
const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), event, id, reconnect, message != nullptr);
if (message == nullptr && prefix_len == CHUNK_HDR_LEN) {
return true; // Match ESPAsyncWebServer: nothing to send
}
if (prefix_len >= PREFIX_BUF_SIZE - 1) {
// The appenders truncate silently, which would put a malformed event on the wire
ESP_LOGW(TAG, "Event name too long, dropped");
return true;
}
// Gather list: the prefix, then the data lines and their separators from the message
struct Gather {
struct iovec iov[MAX_SEND_IOV]; // left uninitialized on purpose
int iovcnt{1};
size_t total{0};
bool fits{true};
} g;
g.total = prefix_len;
for_each_chunk_piece(
message, message_len,
[](void *ctx, const char *piece, size_t len) {
auto &g = *static_cast<Gather *>(ctx);
g.total += len;
if (len == 0) {
return;
}
if (g.iovcnt == MAX_SEND_IOV) {
g.fits = false;
return;
}
g.iov[g.iovcnt++] = {const_cast<char *>(piece), len};
},
&g);
// The header and the terminator are not part of the chunk length
write_chunk_header(prefix, g.total - CHUNK_HDR_LEN - CHUNK_END_LEN);
g.iov[0] = {prefix, prefix_len};
// A message with more lines than the list holds skips straight to the tail
const ssize_t sent = g.fits ? this->send_(g.iov, g.iovcnt) : 0;
if (sent < 0) {
return false;
}
if (static_cast<size_t>(sent) == g.total) {
this->send_failure_started_ms_ = 0; // progress, whichever stall clock was running
return true;
}
// The caller's buffers do not outlive this call, so keep the chunk and continue from loop()
return this->stash_chunk_(prefix, prefix_len, message, message_len, g.total, sent);
}
void AsyncEventSourceResponse::deferrable_send_state(void *source, bool detail_all,
message_generator_t *message_generator) {
// Let the detail_all listing finish before bare state events, so the web UI never shows unnamed entries and
// initial connect sends less
if (!detail_all && !this->entities_iterator_.completed())
return;
drain_tail_();
process_deferred_queue_();
if (this->close_requested_) {
return;
}
if (this->tail_len_ != 0 || !deferred_queue_.empty()) {
// outgoing event buffer or deferred queue still not empty which means downstream tcp send buffer full, no point
// trying to send first
deq_push_back_with_dedup_(source, message_generator);
} else {
// A send error closes the session and clears the queue; nothing is queued after that
if (!this->send_json_(source, message_generator) && !this->close_requested_) {
deq_push_back_with_dedup_(source, message_generator);
}
}
}
#endif
#ifdef USE_WEBSERVER_OTA
esp_err_t AsyncWebServer::handle_multipart_upload_(httpd_req_t *r, const char *content_type) {
// Parse boundary and create reader
@@ -15,21 +15,14 @@
#include <utility>
#include <vector>
#ifdef USE_WEBSERVER
#include "esphome/components/json/json_util.h"
#include "esphome/components/web_server/list_entities.h"
#include "sse_chunk.h"
#endif
#include "esphome/components/web_server_base/stream_head.h"
struct iovec; // NOLINT(readability-identifier-naming) - forward decl of lwip's gather list entry
namespace esphome {
#ifdef USE_WEBSERVER
namespace web_server {
class WebServer;
}; // namespace web_server
#endif
namespace web_server_idf {
namespace esphome::web_server_idf {
using web_server_base::HttpHeader;
using web_server_base::StallClock;
class AsyncWebParameter {
public:
@@ -191,6 +184,7 @@ class AsyncWebServerRequest {
void onDisconnect(std::function<void()> fn) { this->on_disconnect_ = std::move(fn); }
#endif
/// sess_ctx is reserved for AsyncStreamResponse: the server's close_fn treats any session context as one.
operator httpd_req_t *() const { return this->req_; }
optional<std::string> get_header(const char *name) const;
/// Copy a header value into buf without heap allocation.
@@ -269,204 +263,85 @@ class AsyncWebHandler {
virtual bool isRequestHandlerTrivial() const { return true; }
};
#ifdef USE_WEBSERVER
class AsyncEventSource;
class AsyncEventSourceResponse;
using message_generator_t = void(esphome::web_server::WebServer *, void *, json::JsonBuilder &);
/*
This class holds a pointer to the source component that wants to publish a state event, and a pointer to a function
that will lazily generate that event. The two pointers allow dedup in the deferred queue if multiple publishes for
the same component are backed up, and take up only two pointers of memory. The entry in the deferred queue (a
std::vector) is the DeferredEvent instance itself (not a pointer to one elsewhere in heap) so still only two pointers
per entry (and no heap fragmentation). Even 100 backed up events (you'd have to have at least 100 sensors publishing
because of dedup) would take up only 0.8 kB.
*/
struct DeferredEvent {
friend class AsyncEventSourceResponse;
protected:
void *source_;
message_generator_t *message_generator_;
/// A response that keeps its connection open and is written from the main loop, such as an event stream or
/// an audio stream. Use it directly or derive from it. Call loop() and delete the object only through the type
/// it was created as: subclasses hide loop() instead of overriding it, and the destructor is not virtual.
///
/// Lifecycle:
/// 1. Httpd task, inside handleRequest(): create it and call begin(), then hand the pointer to the main loop.
/// 2. Main loop: once is_attached() is true, write() may go through. Call loop() on every pass.
/// 3. Main loop: delete it once safe_to_delete() is true. httpd owns the socket throughout.
class AsyncStreamResponse {
public:
DeferredEvent(void *source, message_generator_t *message_generator)
: source_(source), message_generator_(message_generator) {}
bool operator==(const DeferredEvent &test) const {
return (source_ == test.source_ && message_generator_ == test.message_generator_);
}
};
static_assert(sizeof(DeferredEvent) == sizeof(void *) + sizeof(message_generator_t *),
"DeferredEvent should have no padding");
/// Httpd task, once. Sends the status line (for example HTTPD_200), Content-Type, headers, the default headers
/// and body (a C string or null, sent as is), then makes this object the session context. A chunked response
/// passes {"Transfer-Encoding", "chunked"} as its first header and frames each chunk itself.
void begin(AsyncWebServerRequest *request, const char *status, const char *content_type,
std::span<const HttpHeader> headers, const char *body = nullptr);
class AsyncEventSourceResponse {
friend class AsyncEventSource;
public:
bool try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
uint32_t reconnect = 0);
// detail_all events go out while entities are still being listed; state events wait until listing is done
void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
// The old event name argument would otherwise convert to true
void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
/// Main loop. Non-blocking gather write. Returns the bytes accepted, which may be fewer than offered: the caller
/// keeps the rest and continues from there. 0 means the socket is full or not ours yet; -1 means the stream failed
/// and is closing. Refused writes for SEND_STALL_TIMEOUT_MS in a row also close it.
ssize_t write(struct iovec *iov, int iovcnt);
ssize_t write(const void *data, size_t len);
/// Main loop. Shuts the connection down through httpd; safe to call more than once.
void close();
/// Main loop, every pass. Keeps the session fresh in httpd's LRU and retries a pending close.
void loop();
/// Httpd task, from the server's close_fn: no write() reaches the socket after this returns.
static void socket_closing(httpd_handle_t hd, int fd);
protected:
AsyncEventSourceResponse(const AsyncWebServerRequest *request, esphome::web_server_idf::AsyncEventSource *server,
esphome::web_server::WebServer *ws);
// Main-loop only: sends initial ping/config/sorting_groups, starts entity iterator.
void start_session_main_loop_();
void deq_push_back_with_dedup_(void *source, message_generator_t *message_generator);
void process_deferred_queue_();
// A new chunk may go out: not re-entered from a log line, session alive, tail empty
bool ready_to_send_();
// Non-blocking gather write. Returns bytes written, 0 on would-block or a socket that is not
// ours, -1 after requesting the close on any other error.
ssize_t send_(struct iovec *iov, int iovcnt);
// Push what is left of the chunk in tail_ to the socket; owns the stall timer.
void drain_tail_();
// Grow tail_ to hold len bytes, kept at its high-water mark. False on OOM.
bool reserve_tail_(size_t len);
// Keep the whole chunk in tail_ and continue from sent; false when the tail cannot be allocated.
bool stash_chunk_(const char *prefix, size_t prefix_len, const char *message, size_t message_len, size_t total,
size_t sent);
// Send a state event; JSON too large for the stack buffer is serialized into tail_ instead
bool send_json_(void *source, message_generator_t *generator);
enum class JsonPrepareResult : uint8_t {
JSON_PREPARE_RESULT_BUFFER,
JSON_PREPARE_RESULT_TAIL,
JSON_PREPARE_RESULT_NO_MEMORY,
JSON_PREPARE_RESULT_TOO_LARGE,
};
// Keep the arena out of the caller's frame so it is gone before the state send.
// size_out: BUFFER = JSON length, NO_MEMORY = requested tail capacity, TOO_LARGE = available JSON room;
// unused for TAIL.
JsonPrepareResult __attribute__((noinline)) prepare_json_(void *source, message_generator_t *generator, char *buffer,
size_t *size_out);
// Warn once, and close the session once the stall timeout passes with no memory for the tail
void tail_alloc_failed_(size_t cap);
void request_close_();
// A log line emitted inside a send re-enters try_send_nodefer on this session; refuse it
struct SendGuard {
AsyncEventSourceResponse &owner;
explicit SendGuard(AsyncEventSourceResponse &owner) : owner(owner) { owner.sending_ = true; }
~SendGuard() { this->owner.sending_ = false; }
};
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 work still references this object.
bool safe_to_delete_() const {
/// httpd holds this response as the session context; begin() has sent the whole head by then.
bool is_attached() const { return this->owns_session_(this->fd_.load()); }
/// The client is still connected; httpd has not freed the session.
bool is_connected() const { return this->fd_.load() != 0; }
bool is_closing() const { return this->close_requested_; }
/// The socket fd, for log lines only
int log_id() const { return this->fd_.load(); }
/// Sends are making no progress; after write() returns -1 this tells a send timeout from a send error.
bool is_stalled() const { return this->stall_.is_running(); }
/// The session is gone 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) &&
!this->lru_work_queued_.load(std::memory_order_acquire);
}
static constexpr uint32_t SEND_STALL_TIMEOUT_MS = 20000;
protected:
void process_close_();
static void close_session_work(void *arg);
void refresh_lru_();
static void refresh_lru_work(void *arg);
// httpd closes the socket before it clears the session, so this can still match a closed fd; write() also checks
// socket_closed_
bool owns_session_(int fd) const { return fd != 0 && httpd_sess_get_ctx(this->hd_, fd) == this; }
static void destroy(void *p);
AsyncEventSource *server_;
httpd_handle_t hd_{};
std::atomic<int> fd_{};
std::vector<DeferredEvent> deferred_queue_;
esphome::web_server::WebServer *web_server_;
esphome::web_server::ListEntitiesIterator entities_iterator_;
// One chunk the socket did not take whole, allocated on the first stall; the only heap use
// on the send path
RAMUniquePtr<uint8_t[]> tail_;
uint32_t send_failure_started_ms_{0}; // Zero means no send stall in progress.
// Runs while sends make no progress; a running clock keeps its start time
StallClock stall_;
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};
// Guarded by the stream send mutex
bool socket_closed_{false};
// Main-loop only; the HTTPD task never reads or writes these flags.
bool close_requested_{false};
bool close_retry_warning_logged_{false};
bool sending_{false};
// The longest multi line log message in the tree (a climate dump_config) has 22 lines; a
// longer one goes through the tail
static constexpr size_t MAX_SEND_LINES = 22;
static constexpr size_t MAX_SEND_IOV = 1 + 2 * MAX_SEND_LINES;
// Chunk header, retry/id/event lines and the first "data: "
static constexpr size_t PREFIX_BUF_SIZE = 128;
// Stack buffer for a state event's JSON; a larger document is serialized into the tail
static constexpr size_t JSON_BUF_SIZE = 1024;
// Same ceiling JsonBuilder::serialize() applies (max_heap_size in json_util.cpp); a larger
// document is dropped before anything is on the wire
static constexpr size_t JSON_MAX_SIZE = 5120;
// Most RAM a stalled session keeps: the largest state document plus any accepted framing
static constexpr size_t TAIL_MAX_SIZE = JSON_MAX_SIZE + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN;
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;
class AsyncEventSource : public AsyncWebHandler {
friend class AsyncEventSourceResponse;
public:
AsyncEventSource(StringRef url, esphome::web_server::WebServer *ws) : url_(url), web_server_(ws) {}
~AsyncEventSource() override;
// NOLINTNEXTLINE(readability-identifier-naming)
bool canHandle(AsyncWebServerRequest *request) const override {
if (request->method() != HTTP_GET)
return false;
char url_buf[AsyncWebServerRequest::URL_BUF_SIZE];
return request->url_to(url_buf) == this->url_;
}
// NOLINTNEXTLINE(readability-identifier-naming)
void handleRequest(AsyncWebServerRequest *request) override;
void try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
uint32_t reconnect = 0);
void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
// The old event name argument would otherwise convert to true
void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
/// Returns true if there are sessions remaining (including pending cleanup).
bool loop();
bool empty() { return this->count() == 0; }
size_t count() const { return this->sessions_.size(); }
protected:
// Cold path: move sessions from pending_sessions_ into sessions_ and greet each one.
void __attribute__((noinline, cold)) adopt_pending_sessions_main_loop_();
StringRef url_; // Must outlive this object (string literal)
// Main-loop only. Vector: SSE sessions are 1-5 connections, linear search beats set.
std::vector<AsyncEventSourceResponse *> sessions_;
// Httpd-task intake; guarded by pending_mutex_, gated by has_pending_sessions_.
std::vector<AsyncEventSourceResponse *> pending_sessions_;
Mutex pending_mutex_;
esphome::web_server::WebServer *web_server_;
std::atomic<bool> has_pending_sessions_{false};
};
#endif // USE_WEBSERVER
struct HttpHeader {
const char *name;
const char *value;
};
class DefaultHeaders {
friend class AsyncWebServerRequest;
friend class AsyncStreamResponse;
friend class AsyncWebServerResponse;
#ifdef USE_WEBSERVER
friend class AsyncEventSourceResponse;
#endif
static_assert(WEB_SERVER_DEFAULT_HEADERS_COUNT <= 8, "AsyncWebServerResponse::replaced_defaults_ holds 8 bits");
public:
@@ -484,8 +359,7 @@ class DefaultHeaders {
void apply_(httpd_req_t *req, uint8_t skip_mask = 0) const;
};
} // namespace web_server_idf
} // namespace esphome
} // namespace esphome::web_server_idf
using namespace esphome::web_server_idf; // NOLINT(google-global-names-in-headers)
+51
View File
@@ -337,6 +337,57 @@ size_t truncate_name_to(char *buffer, size_t buffer_size, const char *name, size
// Parsing & formatting
// A plain decimal converts exactly with one division: the digits accumulate
// into an integer that stays below 2^24 and the fraction's scale is a power of
// ten up to 1e9 (2^9 * 5^9), so both convert to float exactly and IEEE division
// rounds the quotient correctly, the same result strtof produces.
static constexpr uint32_t PARSE_FLOAT_MANTISSA_LIMIT = 1000000; // one more digit stays below 2^24
static constexpr uint32_t PARSE_FLOAT_MAX_FRACTION_DIGITS = 9;
optional<float> parse_float(const char *str) {
const char *p = str;
const bool negative = *p == '-';
if (negative || *p == '+')
p++;
const char *digits = p;
const char *dot = nullptr;
uint32_t mantissa = 0;
for (;; p++) {
const uint32_t d = static_cast<unsigned char>(*p) - '0';
if (d > 9) {
if (*p == '.' && dot == nullptr) {
dot = p;
continue;
}
break;
}
if (mantissa >= PARSE_FLOAT_MANTISSA_LIMIT)
break;
mantissa = mantissa * 10 + d;
}
const uint32_t fraction_digits = dot != nullptr ? p - dot - 1 : 0;
const bool any_digit = p - digits > (dot != nullptr ? 1 : 0);
if (*p == '\0' && any_digit && fraction_digits <= PARSE_FLOAT_MAX_FRACTION_DIGITS) {
// The mantissa is exact in a float, so the sign goes on before the one rounding step
float value = static_cast<float>(mantissa);
if (negative)
value = -value;
if (fraction_digits != 0) {
uint32_t scale = 10;
for (uint32_t i = 1; i < fraction_digits; i++)
scale *= 10;
value /= static_cast<float>(scale);
}
return value;
}
// Exponents, hex, inf, nan, leading whitespace and longer inputs take the libc path.
char *end = nullptr;
const float value = ::strtof(str, &end);
if (end == str || *end != '\0' || value == HUGE_VALF)
return {};
return value;
}
size_t parse_hex(const char *str, size_t length, uint8_t *data, size_t count) {
size_t chars = std::min(length, 2 * count);
for (size_t i = 2 * count - chars; i < 2 * count; i++, str++) {
+2 -21
View File
@@ -25,7 +25,6 @@
#include <strings.h>
#include "esphome/core/optional.h"
#include "esphome/core/parse_float.h"
#include "esphome/core/time_conversion.h"
#ifdef USE_ESP8266
@@ -1034,26 +1033,6 @@ inline float pow10_int(int8_t exp) {
return result;
}
/// 10^n, exact while it fits in T.
template<typename T> constexpr T pow10_exact(uint32_t n) {
T result = 1;
T base = 10;
for (;;) {
if ((n & 1) != 0)
result *= base;
n >>= 1;
if (n == 0)
return result;
base *= base;
}
}
/// Whether c is '0' to '9'.
constexpr bool is_digit(char c) { return static_cast<uint8_t>(c - '0') <= 9; }
/// Whether c is C-locale white space, without isspace's ctype table.
constexpr bool is_space_c(char c) { return c == ' ' || (c >= '\t' && c <= '\r'); }
/// Remap \p value from the range (\p min, \p max) to (\p min_out, \p max_out).
template<typename T, typename U> T remap(U value, U min, U max, T min_out, T max_out) {
return (value - min) * (max_out - min_out) / (max - min) + min_out;
@@ -1594,6 +1573,8 @@ template<typename T, enable_if_t<(std::is_integral<T>::value && std::is_signed<T
optional<T> parse_number(const std::string &str) {
return parse_number<T>(str.c_str());
}
/// Parse a decimal floating-point number; plain decimals skip the libc parser.
optional<float> parse_float(const char *str);
/// Parse a decimal floating-point number from a null-terminated string.
template<typename T, enable_if_t<(std::is_same<T, float>::value), int> = 0> optional<T> parse_number(const char *str) {
return parse_float(str);
-372
View File
@@ -1,372 +0,0 @@
// picolibc 1.8.10's strtof, bit for bit, without libc's strtod: decimals round to 9 digits, then to the nearest float.
// Float bits are built from integers, so chips that flush denormals give the same result.
#include "esphome/core/parse_float.h"
#include "esphome/core/hal.h"
#include "esphome/core/helpers.h"
#include <algorithm>
#include <bit>
namespace esphome {
namespace {
constexpr uint32_t FLOAT_SIGN_BIT = 0x80000000, FLOAT_INF_BITS = 0x7F800000;
constexpr uint32_t FLOAT_NAN_BITS = 0x7FC00000; // nanf(""), which strtof returns for every NaN
constexpr uint32_t DECIMAL_DIGITS = 9, HEX_DIGITS = 8; // digits kept before the rest only round
constexpr int32_t EXPONENT_CAP = 224; // picolibc stops adding exponent digits once the exponent reaches this
// 0.<digits> * 10^exp10 is zero at or below MIN_EXP10 and infinity at or above MAX_EXP10, as in picolibc
constexpr int32_t MIN_EXP10 = -46, MAX_EXP10 = 40;
// Past this, one more digit could pass 2^24
constexpr uint32_t PLAIN_MANTISSA_LIMIT = 1677720;
// Bits of the float nearest to (m + sticky fraction) * 2^e, ties to even; m > 0
uint32_t round_to_float_bits(uint64_t m, int32_t e, bool sticky) {
const int32_t msb = e + static_cast<int32_t>(std::bit_width(m)) - 1; // the value is in [2^msb, 2^(msb + 1))
if (msb >= 128)
return FLOAT_INF_BITS;
if (msb < -150) // below half the smallest float
return 0;
// The unit of the float's last bit: 2^(msb - 23), or 2^-149 for a denormal
const int32_t shift = std::max<int32_t>(msb - 23, -149) - e; // at most 64
uint64_t q;
if (shift <= 0) {
q = m << -shift;
} else {
q = shift < 64 ? m >> shift : 0;
const uint64_t rem = shift < 64 ? m & ((uint64_t{1} << shift) - 1) : m;
const uint64_t half = uint64_t{1} << (shift - 1);
if (rem > half || (rem == half && (sticky || (q & 1) != 0)))
q++;
}
if (msb < -126)
return static_cast<uint32_t>(q); // 2^23 here is the smallest normal float, as it should be
// Rounding up past the largest float gives exactly FLOAT_INF_BITS
return (static_cast<uint32_t>(msb + 127) << 23) + static_cast<uint32_t>(q - (uint64_t{1} << 23));
}
// The decimal exponents that reach decimal_to_float_bits: 1 to 9 digits and MIN_EXP10 < digits + e10 < MAX_EXP10
constexpr int32_t MIN_E10 = MIN_EXP10 + 1 - static_cast<int32_t>(DECIMAL_DIGITS), MAX_E10 = MAX_EXP10 - 2;
// Top 64 bits of 10^(8k), rounded down: 10^(8k) = entry * 2^(floor_log2_pow10(8k) - 63), exact for k = 0..3
constexpr int32_t POW10_BY_8_FIRST = MIN_E10 >> 3, POW10_BY_8_LAST = MAX_E10 >> 3;
constexpr uint64_t POW10_BY_8[] PROGMEM = {
0xFB158592BE068D2E, 0xBB127C53B17EC159, 0x8B61313BBABCE2C6, 0xCFB11EAD453994BA,
0x9ABE14CD44753B52, 0xE69594BEC44DE15B, 0xABCC77118461CEFC, 0x8000000000000000,
0xBEBC200000000000, 0x8E1BC9BF04000000, 0xD3C21BCECCEDA100, 0x9DC5ADA82B70B59D,
};
static_assert(POW10_BY_8_FIRST == -7 && POW10_BY_8_LAST == 4 &&
sizeof(POW10_BY_8) / sizeof(POW10_BY_8[0]) == POW10_BY_8_LAST - POW10_BY_8_FIRST + 1,
"POW10_BY_8 must cover MIN_E10..MAX_E10");
// floor(q * log2(10)), for |q| < 100
constexpr int32_t floor_log2_pow10(int32_t q) { return (q * 217706) >> 16; }
// The top 64 bits of a * b, rounded down
uint64_t mul_high(uint64_t a, uint64_t b) {
const uint64_t a_lo = static_cast<uint32_t>(a), a_hi = a >> 32;
const uint64_t b_lo = static_cast<uint32_t>(b), b_hi = b >> 32;
const uint64_t hi_lo = a_hi * b_lo, lo_hi = a_lo * b_hi;
const uint64_t middle = ((a_lo * b_lo) >> 32) + static_cast<uint32_t>(hi_lo) + static_cast<uint32_t>(lo_hi);
return a_hi * b_hi + (hi_lo >> 32) + (lo_hi >> 32) + (middle >> 32);
}
// Cold: a decimal within 16 units of a midpoint is an exact tie when m10 * 5^e10 is an integer
__attribute__((noinline, cold)) uint32_t tie_float_bits(uint32_t m10, int32_t e10, uint32_t bits) {
if (e10 >= 0) {
if (e10 <= 14) // m10 * 5^14 stays below 2^64
return round_to_float_bits(m10 * (pow10_exact<uint64_t>(e10) >> e10), e10, false);
} else if (e10 >= -13) {
const auto pow5 = static_cast<uint32_t>(pow10_exact<uint64_t>(-e10) >> -e10);
if (m10 % pow5 == 0)
return round_to_float_bits(m10 / pow5, e10, false);
}
return bits;
}
// The float nearest to m10 * 10^e10 (0 < m10 <= 10^9). The product is under 4 units low; checked for all 85 billion
// (m10, e10), only exact ties and one non-tie (6 units off, keeps its float) come within 34 units of a midpoint.
uint32_t decimal_to_float_bits(uint32_t m10, int32_t e10) {
// Both operands are exact floats, so the one float operation gives the nearest float
const uint32_t power = e10 < 0 ? -e10 : e10;
if (m10 < (1u << 24) && power <= 10) {
const auto value = static_cast<float>(m10);
// 10^10 overflows uint32_t
const float factor = power == 10 ? 1e10f : static_cast<float>(pow10_exact<uint32_t>(power));
return bit_cast<uint32_t>(e10 < 0 ? value / factor : value * factor);
}
const int32_t k8 = e10 >> 3; // rounds down
const uint64_t x = uint64_t{m10} * pow10_exact<uint32_t>(e10 & 7);
const auto zeros = static_cast<int32_t>(std::countl_zero(x));
uint64_t m = x << zeros;
int32_t e = -zeros;
if (k8 == 0)
return round_to_float_bits(m, e, false); // exact
m = mul_high(m, POW10_BY_8[k8 - POW10_BY_8_FIRST]);
e += floor_log2_pow10(8 * k8) + 1;
if ((m >> 63) == 0) {
m <<= 1;
e--;
}
const uint32_t bits = round_to_float_bits(m, e, false);
// The bits below the float's last one, and the midpoint among them
const int32_t shift = std::max<int32_t>(40, -149 - e);
if (shift <= 64) {
const uint64_t low = shift < 64 ? m & ((uint64_t{1} << shift) - 1) : m;
const uint64_t half = uint64_t{1} << (shift - 1);
if (low - half + 16 > 32) // |low - half| > 16
return bits;
}
return tie_float_bits(m10, e10, bits);
}
uint32_t no_number(const char *str, const char **end, bool *negative) {
*end = str;
*negative = false;
return 0;
}
// The rest of an "nan" or "inf" after its first letter at p; returns the end, or nullptr for no number
const char *read_inf_nan(const char *p) {
const auto lower = [](char c) { return static_cast<char>(c | 0x20); };
if (lower(*p) == 'n') {
if (lower(p[1]) != 'a' || lower(p[2]) != 'n')
return nullptr;
p += 3;
if (*p != '(')
return p;
// "nan(" must be closed after letters, digits and '_'
do {
p++;
} while (static_cast<uint8_t>(lower(*p) - 'a') < 26 || is_digit(*p) || *p == '_');
return *p == ')' ? p + 1 : nullptr;
}
if (lower(p[1]) != 'n' || lower(p[2]) != 'f')
return nullptr;
p += 3;
if (lower(*p) != 'i')
return p;
// Past "infi", only "infinity" is a number
return lower(p[1]) == 'n' && lower(p[2]) == 'i' && lower(p[3]) == 't' && lower(p[4]) == 'y' ? p + 5 : nullptr;
}
// Reads a binary exponent "p" or decimal exponent "e" at p into *exp; returns the end of the number
const char *read_exponent(const char *p, char letter, int32_t *exp) {
if ((*p | 0x20) != letter)
return p;
const char *q = p + 1;
const bool negative = *q == '-';
if (negative || *q == '+')
q++;
if (!is_digit(*q))
return p;
int32_t value = 0;
for (; is_digit(*q); q++) {
if (value < EXPONENT_CAP)
value = value * 10 + (*q - '0');
}
*exp += negative ? -value : value;
return q;
}
// Hex digits after the "x" at p, with the point and decimal exponent read so far (no non-zero digit yet)
__attribute__((noinline, cold)) uint32_t read_hex(const char *p, bool dot, int32_t exp, const char **end) {
uint32_t m = 0, kept = 0; // kept: digits in m from its first non-zero one
bool sticky = false;
for (;; p++) {
const char lower = static_cast<char>(*p | 0x20);
uint32_t digit = static_cast<uint8_t>(*p - '0');
if (static_cast<uint8_t>(lower - 'a') <= 5) {
digit = static_cast<uint8_t>(lower - 'a') + 10;
} else if (digit > 9) {
digit = 16; // picolibc takes ':' to '?' as 10 to 15; we stop there like newlib
}
if (digit < 16) {
if (kept < HEX_DIGITS) {
if (dot)
exp--;
m = m * 16 + digit;
kept += m != 0 ? 1 : 0;
} else {
if (!dot)
exp++;
sticky |= digit != 0;
}
} else if (*p == '.' && !dot) {
dot = true;
} else if (lower != 'x' || m != 0) { // picolibc takes another "x" while the value is still zero
break;
}
}
exp *= 4;
*end = read_exponent(p, 'p', &exp);
return m == 0 ? 0 : round_to_float_bits(m, exp, sticky);
}
// picolibc's conv_flt for a float from the first digit or point at p: returns the bits; *end is str for no number
__attribute__((noinline)) uint32_t read_decimal(const char *str, const char *p, const char **end) {
uint32_t m = 0, kept = 0; // the first (up to 9) significant digits, and how many from the first non-zero one
int32_t exp = 0; // the value is m * 10^exp
bool dot = false, any = false;
uint32_t overflow = 0; // digits past the kept ones: 0 none, 1 below half, 2 half, 3 above half
bool full = false; // a digit past the kept ones was read
for (;; p++) {
const uint32_t digit = static_cast<uint8_t>(*p - '0');
if (digit <= 9) {
any = true;
if (kept < DECIMAL_DIGITS) {
if (dot)
exp--;
m = m * 10 + digit;
kept += m != 0 ? 1 : 0;
} else {
if (!dot)
exp++;
if (!full) {
full = true;
overflow = digit > 5 ? 3 : digit == 5 ? 2 : digit != 0 ? 1 : 0; // the first one decides the half
} else {
overflow |= digit != 0 ? 1 : 0;
}
}
} else if (*p == '.' && !dot) {
dot = true;
} else {
break;
}
}
if ((*p | 0x20) == 'x' && any && m == 0)
return read_hex(p + 1, dot, exp, end);
if (!any) {
*end = str;
return 0;
}
*end = read_exponent(p, 'e', &exp);
if (m == 0)
return 0;
// Ties to even on the decimal digits
overflow |= m & 1;
m += (overflow + 1) >> 2;
const int32_t exp10 = static_cast<int32_t>(kept) + exp;
if (exp10 <= MIN_EXP10)
return 0;
if (exp10 >= MAX_EXP10)
return FLOAT_INF_BITS;
return decimal_to_float_bits(m, exp);
}
// picolibc's strtof from the start: returns the bits without the sign, which goes in *negative
__attribute__((noinline)) uint32_t parse_float_full(const char *str, const char **end, bool *negative) {
// No number starts with a letter but i (inf) or n (nan): "unavailable", "on" and "unknown" stop here
const char first = static_cast<char>(*str | 0x20);
if (first >= 'a' && first <= 'z' && first != 'i' && first != 'n')
return no_number(str, end, negative);
const char *p = str;
while (is_space_c(*p))
p++;
*negative = *p == '-';
if (*negative || *p == '+')
p++;
const char lower = static_cast<char>(*p | 0x20);
if (lower == 'i' || lower == 'n') {
const char *stop = read_inf_nan(p);
if (stop == nullptr)
return no_number(str, end, negative);
*end = stop;
return lower == 'i' ? FLOAT_INF_BITS : FLOAT_NAN_BITS;
}
const uint32_t bits = read_decimal(str, p, end);
if (*end == str)
*negative = false;
return bits;
}
// A plain decimal below 2^24 with up to 9 digits after the point, in one float operation; nullptr otherwise
ESPHOME_ALWAYS_INLINE inline const char *parse_plain_decimal(const char *str, uint32_t *bits, bool *negative) {
const char *p = str;
*negative = *p == '-';
if (*negative || *p == '+')
p++;
const char *digits = p;
const char *point = nullptr;
uint32_t mantissa = 0;
for (;; p++) {
const uint32_t d = static_cast<uint8_t>(*p) - '0';
if (d > 9) {
if (*p == '.' && point == nullptr) {
point = p;
continue;
}
break;
}
if (mantissa > PLAIN_MANTISSA_LIMIT)
return nullptr;
mantissa = mantissa * 10 + d;
}
const uint32_t fraction = point != nullptr ? p - point - 1 : 0;
const char c = static_cast<char>(*p | 0x20);
if (p - digits == (point != nullptr ? 1 : 0) || fraction > 9 || c == 'e' || c == 'x')
return nullptr;
float value = static_cast<float>(mantissa);
if (fraction != 0)
value /= static_cast<float>(pow10_exact<uint32_t>(fraction)); // 10^9 is an exact float
*bits = bit_cast<uint32_t>(value);
return p;
}
} // namespace
optional<float> parse_float(const char *str) {
// Own copy of the plain-decimal loop: sharing parse_plain_decimal costs 12 to 18% more instructions
const char *p = str;
const bool negative = *p == '-';
if (negative || *p == '+')
p++;
const char *digits = p;
const char *point = nullptr;
uint32_t mantissa = 0;
for (;; p++) {
const uint32_t d = static_cast<uint8_t>(*p) - '0';
if (d > 9) {
if (*p == '.' && point == nullptr) {
point = p;
continue;
}
break;
}
if (mantissa > PLAIN_MANTISSA_LIMIT)
break;
mantissa = mantissa * 10 + d;
}
const uint32_t fraction = point != nullptr ? p - point - 1 : 0;
if (*p == '\0' && p - digits > (point != nullptr ? 1 : 0) && fraction <= 9) {
float value = negative ? -static_cast<float>(mantissa) : static_cast<float>(mantissa);
if (fraction != 0)
value /= static_cast<float>(pow10_exact<uint32_t>(fraction));
return value;
}
const char *end;
bool full_negative;
const uint32_t bits = parse_float_full(str, &end, &full_negative);
// HUGE_VALF, positive overflow or "inf", is rejected like before
if (end == str || *end != '\0' || (!full_negative && bits == FLOAT_INF_BITS))
return {};
return bit_cast<float>(full_negative ? bits | FLOAT_SIGN_BIT : bits);
}
float str_to_float(const char *str, char **end) {
const char *stop;
bool negative;
uint32_t bits;
if (const char *plain_end = parse_plain_decimal(str, &bits, &negative)) {
stop = plain_end;
} else {
bits = parse_float_full(str, &stop, &negative);
}
if (end != nullptr)
*end = const_cast<char *>(stop); // strtof's signature
return bit_cast<float>(negative ? bits | FLOAT_SIGN_BIT : bits);
}
} // namespace esphome
-17
View File
@@ -1,17 +0,0 @@
#pragma once
#include <cstdint>
#include "esphome/core/optional.h"
namespace esphome {
/// The whole string as str_to_float's syntax (leading space, hex, inf, nan); trailing characters, "inf" and positive
/// overflow are rejected, while -inf, negative overflow and NaN are accepted.
optional<float> parse_float(const char *str);
/// strtof with picolibc's results on every platform, but it never sets errno (overflow gives +-HUGE_VALF, underflow 0);
/// use it instead of strtof, or strtod/atof for a float result.
float str_to_float(const char *str, char **end);
} // namespace esphome
+2 -3
View File
@@ -7,7 +7,6 @@
#include <memory>
#include <string>
#include "esphome/core/defines.h"
#include "esphome/core/parse_float.h"
#ifdef USE_JSON
#include "esphome/components/json/json_util.h"
@@ -294,7 +293,7 @@ inline bool operator!=(const std::string &lhs, const ProgmemStringRef &rhs) { re
#endif
// String conversion functions for ADL compatibility (allows stoi(x) where x is StringRef)
// Must be in esphome namespace for ADL to find them. Uses strtol, strtod and str_to_float to avoid heap allocation.
// Must be in esphome namespace for ADL to find them. Uses strtol/strtod directly to avoid heap allocation.
namespace internal {
// NOLINTBEGIN(google-runtime-int)
template<typename R, typename F> inline R parse_number(const StringRef &str, size_t *pos, F conv) {
@@ -335,7 +334,7 @@ inline long stol(const StringRef &str, size_t *pos = nullptr, int base = 10) {
return internal::parse_number<long>(str, pos, base, std::strtol);
}
inline float stof(const StringRef &str, size_t *pos = nullptr) {
return internal::parse_number<float>(str, pos, str_to_float);
return internal::parse_number<float>(str, pos, std::strtof);
}
inline double stod(const StringRef &str, size_t *pos = nullptr) {
return internal::parse_number<double>(str, pos, std::strtod);
@@ -1,40 +0,0 @@
#include <benchmark/benchmark.h>
#include "esphome/core/helpers.h"
namespace esphome::benchmarks {
// Inner iteration count to amortize CodSpeed instrumentation overhead.
static constexpr int kInnerIterations = 2000;
static const char *const PLAIN[] = {"0.5", "255", "-12.75", "21.3"};
static const char *const EXPONENT[] = {"1e3", "2.5e-2", "-6.02E5", "1.5e+1"};
static const char *const PRECISE[] = {"3.14159265358979", "0.000123456789", "123456789.125", "-0.1000000000000000055"};
// Home Assistant states that are not numbers
static const char *const REJECT[] = {"unavailable", "unknown", "on", "off"};
template<size_t N, typename F> static void run(benchmark::State &state, const char *const (&inputs)[N], F parse) {
for (auto _ : state) {
float sum = 0.0f;
for (int i = 0; i < kInnerIterations; i++)
sum += parse(inputs[i % N]);
benchmark::DoNotOptimize(sum);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
static float parse_number_float(const char *str) { return parse_number<float>(str).value_or(0.0f); }
static void ParseNumberFloat_Plain(benchmark::State &state) { run(state, PLAIN, parse_number_float); }
BENCHMARK(ParseNumberFloat_Plain);
static void ParseNumberFloat_Exponent(benchmark::State &state) { run(state, EXPONENT, parse_number_float); }
BENCHMARK(ParseNumberFloat_Exponent);
static void ParseNumberFloat_Precise(benchmark::State &state) { run(state, PRECISE, parse_number_float); }
BENCHMARK(ParseNumberFloat_Precise);
static void ParseNumberFloat_Reject(benchmark::State &state) { run(state, REJECT, parse_number_float); }
BENCHMARK(ParseNumberFloat_Reject);
} // namespace esphome::benchmarks
+11 -342
View File
@@ -1,74 +1,34 @@
#include <gtest/gtest.h>
#include <cmath>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <random>
#include <string>
#include "esphome/core/helpers.h"
#include "esphome/core/parse_float.h"
#ifdef USE_HOST
namespace esphome::core::testing {
static uint32_t bits(float value) { return bit_cast<uint32_t>(value); }
// picolibc rounds a decimal to 9 significant digits and converts those to the nearest float, so up to 9
// significant digits its result is the correctly rounded float the host's strtof (glibc, macOS) gives. Longer
// input and the syntax where libcs differ are checked against values taken from picolibc 1.8.10's own strtof.
static float reference_strtof(const char *str, char **end) {
const float value = ::strtof(str, end);
if (std::isnan(value))
return std::signbit(value) ? -NAN : NAN;
return value;
static uint32_t bits(float value) {
uint32_t out;
memcpy(&out, &value, sizeof(out));
return out;
}
// Every accepted input must give the reference's bits.
// strtof is the reference: every accepted input must give the same bits.
static void expect_matches_strtof(const char *str) {
const float expected = reference_strtof(str, nullptr);
char *end = nullptr;
const float expected = ::strtof(str, &end);
const auto parsed = parse_number<float>(str);
if (!parsed.has_value()) {
ADD_FAILURE() << str << " was rejected";
return;
}
EXPECT_EQ(bits(parsed.value()), bits(expected)) << str;
const float actual = parsed.value();
EXPECT_EQ(bits(actual), bits(expected)) << str << " parsed " << actual << " expected " << expected;
}
// The same bits and end as the host's strtof; for decimals of up to 9 significant digits
static void expect_str_to_float_matches(const char *str) {
char *expected_end = nullptr;
const float expected = reference_strtof(str, &expected_end);
char *end = nullptr;
const float actual = str_to_float(str, &end);
EXPECT_EQ(bits(actual), bits(expected)) << '"' << str << '"';
EXPECT_EQ(end - str, expected_end - str) << '"' << str << '"';
}
struct PicolibcCase {
const char *str;
uint32_t bits;
int32_t end;
};
// Bits and end from picolibc 1.8.10's strtof
static void expect_picolibc(const PicolibcCase &c) {
char *end = nullptr;
const float value = str_to_float(c.str, &end);
EXPECT_EQ(bits(value), c.bits) << '"' << c.str << '"';
EXPECT_EQ(end - c.str, c.end) << '"' << c.str << '"';
}
static void expect_picolibc(const std::string &str, uint32_t expected) {
char *end = nullptr;
EXPECT_EQ(bits(str_to_float(str.c_str(), &end)), expected) << str.substr(0, 40);
EXPECT_EQ(static_cast<size_t>(end - str.c_str()), str.size()) << str.substr(0, 40);
}
static float float_from(uint32_t value) { return bit_cast<float>(value); }
TEST(ParseFloat, PlainDecimalsMatchStrtof) {
for (const char *str :
{"0", "-0", "+0", "1", "-1", "100.50", "99.75", "0.1", "0.3",
@@ -86,14 +46,6 @@ TEST(ParseFloat, SweepOfTwoDecimalValuesMatchesStrtof) {
}
}
// Eight-digit mantissas below 2^24 stay on the one-division path; the next digit past the limit falls back
TEST(ParseFloat, EightDigitPlainDecimalsMatchStrtof) {
for (const char *str : {"12345678", "-12345678", "1234.5678", "0.12345678", "16777209", "1677720.9", "0.16777209",
"16777210", "16777215", "1677721.5", "99999999", "9999999.9", "16777209."}) {
expect_matches_strtof(str);
}
}
// A fraction longer than the fast path accepts must fall back, however long it is.
TEST(ParseFloat, LongFractionsMatchStrtof) {
std::string str = "0.";
@@ -116,293 +68,10 @@ TEST(ParseFloat, LibcPathInputsMatchStrtof) {
// Positive overflow and a positive literal infinity both come back as HUGE_VALF from strtof, and the parser
// rejects both; the negative forms pass through, as they did before the fast path.
TEST(ParseFloat, RejectsInvalidInput) {
for (const char *str : {"", "-", "+", ".", "abc", "12abc", "1.2.3", "1,5", "12 ", "--1", "1e", "1e400", "inf",
"infinity", "infi", "nan(12 ab)", "nan(12"}) {
for (const char *str :
{"", "-", "+", ".", "abc", "12abc", "1.2.3", "1,5", "12 ", "--1", "1e", "1e400", "inf", "infinity"}) {
EXPECT_FALSE(parse_number<float>(str).has_value()) << str;
}
}
// Above the midpoint 1 + 2^-24 by less than half a double: newlib's strtof rounds to the midpoint and then to
// 1.0f; picolibc's 9 digits, 1.00000006, give the next float.
TEST(ParseFloat, FullPathRoundsLikePicolibc) {
EXPECT_EQ(bits(parse_number<float>("1.0000000596046448").value()), 0x3F800001u);
EXPECT_EQ(bits(parse_number<float>("-1.0000000596046448").value()), 0xBF800001u);
}
// picolibc takes "nan(" with letters, digits and '_' up to a ")", and nothing at all otherwise
TEST(ParseFloat, NanForms) {
EXPECT_TRUE(std::isnan(parse_number<float>("nan(12)").value()));
EXPECT_TRUE(std::isnan(parse_number<float>("nan(xyz_1)").value()));
EXPECT_FALSE(parse_number<float>("nan(x y)").has_value());
}
// The callers' overflow check: a positive overflow is HUGE_VALF and rejected, a negative one is -inf and accepted
TEST(ParseFloat, OverflowIsHugeValf) {
EXPECT_FALSE(parse_number<float>("1e39").has_value());
EXPECT_FALSE(parse_number<float>("3.40282357e38").has_value());
EXPECT_FALSE(parse_number<float>("1e99999").has_value());
EXPECT_EQ(bits(parse_number<float>("-1e39").value()), 0xFF800000u);
EXPECT_EQ(bits(parse_number<float>("3.4028235e38").value()), 0x7F7FFFFFu);
}
// Up to 9 significant digits, with leading and trailing zeros, a point anywhere and an exponent
TEST(StrToFloat, RandomDecimalsMatchStrtof) {
std::mt19937_64 rng(1); // NOLINT(cert-msc32-c,cert-msc51-cpp,bugprone-random-generator-seed) reproducible
for (int i = 0; i < 50000; i++) {
std::string str;
if (rng() % 3 == 0)
str += (rng() & 1) ? "-" : "+";
const int zeros = static_cast<int>(rng() % 4);
const int significant = 1 + static_cast<int>(rng() % 9);
const int trailing = static_cast<int>(rng() % 4);
std::string digits(zeros, '0');
digits += static_cast<char>('1' + rng() % 9);
for (int j = 1; j < significant; j++)
digits += static_cast<char>('0' + rng() % 10);
digits.append(trailing, '0');
const int point = static_cast<int>(rng() % (digits.size() + 2)) - 1;
for (int j = 0; j < static_cast<int>(digits.size()); j++) {
if (j == point)
str += '.';
str += digits[j];
}
if (rng() & 1) {
str += (rng() & 1) ? 'e' : 'E';
char exp[8];
snprintf(exp, sizeof(exp), "%d", static_cast<int>(rng() % 110) - 60);
str += exp;
}
expect_str_to_float_matches(str.c_str());
}
}
// Every 16411th float, printed with 6 and 9 digits
TEST(StrToFloat, FloatsMatchStrtof) {
char buf[64];
for (uint32_t u = 0; u < 0x7F800000; u += 16411) {
for (int precision : {6, 9}) {
snprintf(buf, sizeof(buf), "%.*e", precision - 1, static_cast<double>(float_from(u)));
expect_str_to_float_matches(buf);
}
}
}
TEST(StrToFloat, SyntaxMatchesStrtof) {
// Tokens whose combinations the libcs read alike: no "x", "nan(" or "infi"
static const char *const TOKENS[] = {"0", "1", "9", ".", "e", "E", "+", "-", "a", "N", " ", "\t", "y", "00", "#", ""};
for (const char *a : TOKENS) {
for (const char *b : TOKENS) {
for (const char *c : TOKENS) {
for (const char *d : TOKENS) {
const std::string str = std::string(a) + b + c + d;
expect_str_to_float_matches(str.c_str());
}
}
}
}
}
// Where newlib and glibc read the text differently from picolibc
TEST(StrToFloat, SyntaxMatchesPicolibc) {
static const PicolibcCase CASES[] = {
// "x" after any zero digits starts hex, even after a point or a second "0x"
{"000x", 0x00000000, 4},
{"0x", 0x00000000, 2},
{"-0x", 0x80000000, 3},
{"00x1", 0x3F800000, 4},
{"0x0x1", 0x3F800000, 5},
{"0.0x1", 0x3B800000, 5},
{".x1", 0x00000000, 0},
{"0x.p1", 0x00000000, 5},
{"0x.8", 0x3F000000, 4},
{"0X1.8", 0x3FC00000, 5},
{"0x1P-1", 0x3F000000, 6},
{"0x1p+", 0x3F800000, 3},
{"1x5", 0x3F800000, 1},
// "inf" stops before anything but "i"; past "infi" only "infinity" is a number
{"inf", 0x7F800000, 3},
{"-inf", 0xFF800000, 4},
{"INFINITY", 0x7F800000, 8},
{"+infi0", 0x00000000, 0},
{"infi", 0x00000000, 0},
{"infinit", 0x00000000, 0},
// "nan(" takes letters, digits and '_' up to ")", and is no number without the ")"
{"NaN", 0x7FC00000, 3},
{"nan()", 0x7FC00000, 5},
{"nan(xyz)", 0x7FC00000, 8},
{"nan(_a1)", 0x7FC00000, 8},
{"-nan(1)", 0xFFC00000, 7},
{"nan( )", 0x00000000, 0},
{"nan(12 ab)", 0x00000000, 0},
{"nan(12", 0x00000000, 0},
// An exponent needs a digit
{"1e", 0x3F800000, 1},
{"1e+", 0x3F800000, 1},
{"1e-x", 0x3F800000, 1},
{"\t 12", 0x41400000, 4},
{"- 1", 0x00000000, 0},
{"1.2.3", 0x3F99999A, 3},
};
for (const PicolibcCase &c : CASES)
expect_picolibc(c);
}
// The inputs where newlib's strtof (decimal to double to float) and picolibc's were found to differ; every platform
// now gives picolibc's result, which is also the correctly rounded float for these
TEST(StrToFloat, RoundsLikePicolibcWhereNewlibDoubleRounds) {
expect_picolibc({"1.0000000596046448", 0x3F800001, 18});
expect_picolibc({"0x1.100b0d0000000c", 0x3F880587, 18});
expect_picolibc("1.000000059604644775390625" + std::string(800, '0') + "1", 0x3F800001);
}
// picolibc rounds a decimal to 9 significant digits, ties to even, before converting it: past 9 digits its float can
// be one away from the nearest, and an exact tie between two floats can round up. Values from picolibc.
TEST(StrToFloat, NineDigitRoundingLikePicolibc) {
static const PicolibcCase CASES[] = {
{"1.0856183295439e10", 0x5021C50E, 18}, // nearest 0x5021C50D
{"3.97496754550e-4", 0x39D0671D, 16}, // nearest 0x39D0671C
{"9.76178673657108930e23", 0x674EB6C1, 22}, // nearest 0x674EB6C0
{"156.6562236e-26", 0x17F269ED, 15}, // nearest 0x17F269EC
{"1.000000059604644775390625", 0x3F800001, 26}, // the tie 1 + 2^-24: nearest-even 0x3F800000
{"1.000000059", 0x3F800001, 11}, // 1.00000006 after rounding
{"7.0064923216240854e-46", 0x00000000, 22}, // 7.00649232e-46, below half the smallest float
{"21.299999999999997", 0x41AA6666, 18},
{"0.30000000000000004", 0x3E99999A, 19},
{"0.1000000000000000055511151231257827", 0x3DCCCCCD, 36},
};
for (const PicolibcCase &c : CASES)
expect_picolibc(c);
}
// Nine digits converted exactly: ties to even between floats, where the 64-bit product cannot decide alone
TEST(StrToFloat, NineDigitTiesGoToEven) {
static const PicolibcCase CASES[] = {
{"16777217", 0x4B800000, 8}, {"16777219", 0x4B800002, 8}, {"16777217.0", 0x4B800000, 10},
{"8388608.5", 0x4B000000, 9}, {"8388609.5", 0x4B000002, 9}, {"3e10", 0x50DF8476, 4},
{"5e10", 0x513A43B7, 4}, {"1.00000006", 0x3F800001, 10},
};
for (const PicolibcCase &c : CASES)
expect_picolibc(c);
}
// RP2350 flushes float denormals in its double to float conversion; #22677 bench. The parser builds the bits
// with integer operations, so every denormal parses to its own value.
TEST(StrToFloat, DenormalsParseWithoutFlushing) {
char buf[64];
const auto check = [&buf](uint32_t k) {
snprintf(buf, sizeof(buf), "%.8e", std::ldexp(static_cast<double>(k), -149));
EXPECT_EQ(bits(str_to_float(buf, nullptr)), k) << buf;
};
for (uint32_t k = 0; k <= 0x800000; k += 4099)
check(k);
// The largest denormal and FLT_MIN, where rounding carries into the exponent
for (uint32_t k : {1u, 2u, 0x7FFFFEu, 0x7FFFFFu, 0x800000u})
check(k);
}
// Pico W and Pico 2 W bench, #22677: the RP2040 and RP2350 libcs flush float denormals to 0 in their double to
// float conversion, and the RP2040 one turns a NaN into +infinity
TEST(StrToFloat, Rp2040LibcFlushesAndNanToInfinity) {
static const PicolibcCase CASES[] = {
{"1e-40", 0x000116C2, 5}, {"1.4e-45", 0x00000001, 7}, {"1e-38", 0x006CE3EE, 5},
{"1.1754942e-38", 0x007FFFFF, 13}, {"0x1p-149", 0x00000001, 8}, {"1.17549435e-38", 0x00800000, 14},
{"nan(12)", 0x7FC00000, 7}, {"nan", 0x7FC00000, 3},
};
for (const PicolibcCase &c : CASES)
expect_picolibc(c);
}
// RP2350 bench: libc returned 0 and the first parser stalled stepping up from a flushed guess; #22677 bench
TEST(StrToFloat, DenormalsAndFltMinMatchStrtof) {
for (const char *str : {"1.4e-45", "-1.4e-45", "1e-40", "1e-38", "1.17549435e-38", "1.1754942e-38", "1.17549421e-38",
"1.1754945e-38", "7.0064923e-46", "7.0064924e-46"}) {
expect_str_to_float_matches(str);
}
}
// Home Assistant's non-numeric states; CodSpeed HomeassistantSensor_StateResponse_Unavailable on #22677
TEST(StrToFloat, NonNumericStatesAreRejected) {
for (const char *str : {"unavailable", "unknown", "on", "off", "", "-", ".", "none", "inactive"}) {
expect_str_to_float_matches(str);
EXPECT_FALSE(parse_number<float>(str).has_value()) << str;
}
}
// The one float operation path takes up to 2^24 - 1 and 10^10; #22677 review
TEST(StrToFloat, PlainPathBoundaries) {
for (const char *top : {"16777215", "16777216", "16777217"}) {
for (const char *exp : {"e10", "e-10", "e11", "e-11", "e9", "e-9", ".5", ".25"}) {
const std::string str = std::string(top) + exp;
expect_str_to_float_matches(str.c_str());
}
}
}
// The midpoint between FLT_MAX and 2^128 rounds to infinity; review
TEST(StrToFloat, OverflowMidpoint) {
expect_picolibc({"340282356779733661637539395458142568448", 0x7F800000, 39});
expect_picolibc({"3.40282356e38", 0x7F7FFFFF, 13});
expect_picolibc({"3.40282357e38", 0x7F800000, 13});
EXPECT_EQ(bits(str_to_float("0x1.ffffffp127", nullptr)), 0x7F800000u);
}
// picolibc stops adding exponent digits at 224, so the digits of the mantissa can move it no further
TEST(StrToFloat, HugeExponents) {
for (const char *str : {"1e-99999", "1e99999", "-1e99999", "0e99999", "1e9999999999999999999",
"1e-9999999999999999999", "1e-999999999999", "0.000e99999", "1e224", "1e2240", "1e-2240"}) {
expect_str_to_float_matches(str);
}
// 10^-3000 * 10^3000 is 1, but the exponent counts only as 300
expect_picolibc("0." + std::string(2999, '0') + "1e3000", 0x00000000);
expect_picolibc("1" + std::string(300, '0') + "e-300", 0x3F800000);
}
TEST(StrToFloat, LongMantissas) {
expect_picolibc("1" + std::string(799, '0') + "e-799", 0x3F800000);
expect_picolibc("1." + std::string(3000, '0') + "1e-3001", 0x00000000);
expect_picolibc("0." + std::string(400, '0') + "34028235" + std::string(500, '7') + "e440", 0x7F800000);
expect_picolibc(std::string(300, '9') + "e-261", 0x7F800000);
expect_picolibc("3" + std::string(799, '7') + "e-761", 0x7F800000);
}
// picolibc takes ':' to '?' as hex digits 10 to 15 ("0x1:" is 26); these stop at them, as newlib does. #22677 review
TEST(StrToFloat, HexStopsAtPunctuationAfterNine) {
struct Case {
const char *str;
uint32_t bits;
int32_t end;
};
const Case cases[] = {
{"0x1:", 0x3F800000, 3}, {"0x1;", 0x3F800000, 3}, {"0x1<", 0x3F800000, 3}, {"0x1=", 0x3F800000, 3},
{"0x1>", 0x3F800000, 3}, {"0x1?", 0x3F800000, 3}, {"0x1.:", 0x3F800000, 4}, {"0x:", 0x00000000, 2},
{"0x9?", 0x41100000, 3}, {"0xA:", 0x41200000, 3},
};
for (const auto &c : cases) {
char *end = nullptr;
const float value = str_to_float(c.str, &end);
EXPECT_EQ(bits(value), c.bits) << '"' << c.str << '"';
EXPECT_EQ(end - c.str, c.end) << '"' << c.str << '"';
EXPECT_FALSE(parse_number<float>(c.str).has_value()) << c.str;
}
}
TEST(StrToFloat, HexDenormalsAndOverflow) {
struct Case {
const char *str;
uint32_t bits;
};
static const Case CASES[] = {
{"0x1p-149", 0x00000001}, {"0x1p-150", 0x00000000},
{"0x1.8p-149", 0x00000002}, {"0x1.fffffcp-127", 0x007FFFFF},
{"0x1.fffffep-127", 0x00800000}, {"-0x1p-149", 0x80000001},
{"0x1p128", 0x7F800000}, {"0x1.fffffep127", 0x7F7FFFFF},
{"0x1.fffffefp127", 0x7F7FFFFF}, {"0x1.0000000000001p-150", 0x00000001},
{"0x1.000001p0", 0x3F800000}, {"0x1.000003p0", 0x3F800002},
{"0x1p-99999999999", 0x00000000}, {"0x1p99999999999", 0x7F800000},
};
for (const Case &c : CASES)
EXPECT_EQ(bits(str_to_float(c.str, nullptr)), c.bits) << c.str;
}
} // namespace esphome::core::testing
#endif // USE_HOST
@@ -6,9 +6,9 @@
#include <random>
#include <string>
#include "esphome/components/web_server_idf/sse_chunk.h"
#include "esphome/components/web_server/sse_chunk.h"
namespace esphome::web_server_idf::testing {
namespace esphome::web_server::testing {
// The chunk as try_send_nodefer lays it out: header, fields, first "data: ", then the pieces
static std::string build_chunk(const char *message, size_t message_len, const char *event, uint32_t id,
@@ -128,4 +128,4 @@ TEST(SseChunk, MatchesReferenceOnRandomMessages) {
}
}
} // namespace esphome::web_server_idf::testing
} // namespace esphome::web_server::testing
@@ -0,0 +1,300 @@
#ifdef USE_HOST
#include <gtest/gtest.h>
#include <sys/uio.h>
#include <cstdint>
#include <cstdio>
#include <string>
#include <vector>
#include "esphome/components/web_server/sse_chunk.h"
#include "esphome/components/web_server_base/stream_head.h"
namespace esphome::web_server::testing {
using namespace web_server_base; // NOLINT(google-build-using-namespace)
// Collects every batch write_stream_head() sends
struct Capture {
std::string bytes;
std::vector<int> batch_sizes;
int attempts{0};
int fail_after_batches{-1};
bool operator()(struct iovec *iov, int count) {
this->attempts++;
if (static_cast<int>(this->batch_sizes.size()) == this->fail_after_batches) {
return false;
}
this->batch_sizes.push_back(count);
for (int i = 0; i < count; i++) {
this->bytes.append(static_cast<const char *>(iov[i].iov_base), iov[i].iov_len);
}
return true;
}
};
static std::string build_head(Capture &cap, const char *status, const char *content_type,
std::span<const HttpHeader> headers, std::span<const HttpHeader> defaults,
const char *body) {
EXPECT_TRUE(write_stream_head<struct iovec>(cap, status, content_type, headers, defaults, body));
return cap.bytes;
}
// What httpd_resp_send_chunk() put on the wire for its first chunk (esp_http_server/src/httpd_txrx.c in
// ESP-IDF 5.5): the chunked head with Transfer-Encoding, each httpd_resp_set_hdr() header, the blank line,
// then "%lx\r\n", the chunk data and "\r\n".
static std::string reference_send_chunk(const char *status, const char *content_type,
const std::vector<HttpHeader> &resp_hdrs, const char *buf, size_t buf_len) {
char head[256];
std::string out(
head, snprintf(head, sizeof(head), "HTTP/1.1 %s\r\nContent-Type: %s\r\nTransfer-Encoding: chunked\r\n", status,
content_type));
for (const auto &hdr : resp_hdrs) {
out.append(hdr.name).append(": ").append(hdr.value).append("\r\n");
}
out.append("\r\n");
char len_str[10];
out.append(len_str, snprintf(len_str, sizeof(len_str), "%lx\r\n", static_cast<long>(buf_len)));
out.append(buf, buf_len);
out.append("\r\n");
return out;
}
static constexpr HttpHeader DEFAULTS[] = {
{"Access-Control-Allow-Origin", "*"},
{"Access-Control-Allow-Private-Network", "true"},
};
TEST(StreamHead, SseHeadMatchesHttpdRespSendChunk) {
// The old event source set Cache-Control and Connection, then the default headers, and sent "\r\n" as a chunk
const std::string expected = reference_send_chunk(
"200 OK", "text/event-stream",
{{"Cache-Control", "no-cache"}, {"Connection", "keep-alive"}, DEFAULTS[0], DEFAULTS[1]}, "\r\n", 2);
Capture cap;
EXPECT_EQ(build_head(cap, SSE_STATUS, SSE_CONTENT_TYPE, SSE_HEADERS, DEFAULTS, SSE_FIRST_CHUNK), expected);
}
TEST(StreamHead, HeaderReplacesDefaultOfSameName) {
// A default the caller also sets, in any case, goes out once with the caller's value
static constexpr HttpHeader HEADERS[] = {
{"access-control-allow-origin", "http://example.com"},
};
Capture cap;
EXPECT_EQ(build_head(cap, "200 OK", "text/plain", HEADERS, DEFAULTS, nullptr),
"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\naccess-control-allow-origin: http://example.com"
"\r\nAccess-Control-Allow-Private-Network: true\r\n\r\n");
}
TEST(StreamHead, RawHeadWithCustomHeadersAndNoBody) {
static constexpr HttpHeader HEADERS[] = {
{"Cache-Control", "no-store"},
{"X-Stream", "audio"},
};
Capture cap;
EXPECT_EQ(build_head(cap, "200 OK", "audio/wav", HEADERS, {}, nullptr), "HTTP/1.1 200 OK\r\n"
"Content-Type: audio/wav\r\n"
"Cache-Control: no-store\r\n"
"X-Stream: audio\r\n"
"\r\n");
}
TEST(StreamHead, BareHead) {
Capture cap;
EXPECT_EQ(build_head(cap, "204 No Content", "text/plain", {}, {}, nullptr),
"HTTP/1.1 204 No Content\r\nContent-Type: text/plain\r\n\r\n");
EXPECT_EQ(cap.batch_sizes, std::vector<int>{5});
}
TEST(StreamHead, ManyHeadersSplitIntoBatchesWithSameBytes) {
// 4 + 12 * 4 + 2 = 54 pieces, more than two batches hold
static constexpr HttpHeader HEADERS[] = {
{"X-Header-0", "value-0"}, {"X-Header-1", "value-1"}, {"X-Header-2", "value-2"}, {"X-Header-3", "value-3"},
{"X-Header-4", "value-4"}, {"X-Header-5", "value-5"}, {"X-Header-6", "value-6"}, {"X-Header-7", "value-7"},
{"X-Header-8", "value-8"}, {"X-Header-9", "value-9"},
};
const std::span<const HttpHeader> headers{HEADERS};
Capture batched;
const std::string bytes = build_head(batched, "200 OK", "text/plain", headers, DEFAULTS, "body");
EXPECT_EQ(batched.batch_sizes, (std::vector<int>{STREAM_HEAD_MAX_IOV, STREAM_HEAD_MAX_IOV, 6}));
std::string expected = "HTTP/1.1 200 OK\r\nContent-Type: text/plain";
const auto append_headers = [&expected](std::span<const HttpHeader> list) {
for (const auto &h : list) {
expected.append("\r\n").append(h.name).append(": ").append(h.value);
}
};
append_headers(headers);
append_headers(DEFAULTS);
expected.append("\r\n\r\nbody");
EXPECT_EQ(bytes, expected);
}
TEST(StreamHead, BatchBoundary) {
// 4 + 4 * 4 + 1 + 1 = 22 pieces fit one batch; a fifth header makes 4 + 5 * 4 = 24 before the blank line
static constexpr HttpHeader HEADERS[] = {{"A", "1"}, {"B", "2"}, {"C", "3"}, {"D", "4"}, {"E", "5"}};
Capture four;
build_head(four, "200 OK", "text/plain", std::span(HEADERS, 4), {}, "x");
EXPECT_EQ(four.batch_sizes, std::vector<int>{22});
EXPECT_EQ(stream_head_iov_count(4), 22u);
Capture five;
const std::string bytes = build_head(five, "200 OK", "text/plain", HEADERS, {}, "x");
EXPECT_EQ(five.batch_sizes, (std::vector<int>{STREAM_HEAD_MAX_IOV, 2}));
EXPECT_EQ(bytes, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nA: 1\r\nB: 2\r\nC: 3\r\nD: 4\r\nE: 5\r\n\r\nx");
}
TEST(StreamHead, FailedSendSkipsLaterBatches) {
// Three batches; the second fails, so the third is never sent
std::vector<HttpHeader> headers(12, HttpHeader{"X", "y"});
Capture full;
build_head(full, "200 OK", "text/plain", headers, {}, nullptr);
ASSERT_EQ(full.batch_sizes.size(), 3u);
Capture cap;
cap.fail_after_batches = 1;
EXPECT_FALSE(write_stream_head<struct iovec>(cap, "200 OK", "text/plain", headers, {}, nullptr));
EXPECT_EQ(cap.attempts, 2);
EXPECT_EQ(cap.batch_sizes.size(), 1u);
}
// Three elements of 3, 4 and 5 bytes
struct IovFixture {
char a[4] = "abc";
char b[5] = "defg";
char c[6] = "hijkl";
struct iovec iov[3] = {{a, 3}, {b, 4}, {c, 5}};
struct iovec *cur = iov;
int count = 3;
std::string rest() const {
std::string out;
for (int i = 0; i < this->count; i++) {
out.append(static_cast<const char *>(this->cur[i].iov_base), this->cur[i].iov_len);
}
return out;
}
};
TEST(AdvanceIovecs, ZeroLengthWriteChangesNothing) {
IovFixture f;
advance_iovecs(f.cur, f.count, 0);
EXPECT_EQ(f.cur, f.iov);
EXPECT_EQ(f.count, 3);
EXPECT_EQ(f.rest(), "abcdefghijkl");
}
TEST(AdvanceIovecs, PartialInsideFirstElement) {
IovFixture f;
advance_iovecs(f.cur, f.count, 2);
EXPECT_EQ(f.cur, f.iov);
EXPECT_EQ(f.count, 3);
EXPECT_EQ(f.cur->iov_len, 1u);
EXPECT_EQ(f.rest(), "cdefghijkl");
}
TEST(AdvanceIovecs, ExactlyOnElementBoundary) {
IovFixture f;
advance_iovecs(f.cur, f.count, 3);
EXPECT_EQ(f.cur, &f.iov[1]);
EXPECT_EQ(f.count, 2);
EXPECT_EQ(f.rest(), "defghijkl");
}
TEST(AdvanceIovecs, AcrossElementBoundaries) {
IovFixture f;
advance_iovecs(f.cur, f.count, 9);
EXPECT_EQ(f.cur, &f.iov[2]);
EXPECT_EQ(f.count, 1);
EXPECT_EQ(f.rest(), "jkl");
}
TEST(AdvanceIovecs, RepeatedShortWrites) {
IovFixture f;
std::string sent;
while (f.count > 0) {
const std::string before = f.rest();
advance_iovecs(f.cur, f.count, 2);
sent += before.substr(0, before.size() - f.rest().size());
}
EXPECT_EQ(sent, "abcdefghijkl");
}
TEST(AdvanceIovecs, EverythingWritten) {
IovFixture f;
advance_iovecs(f.cur, f.count, 12);
EXPECT_EQ(f.count, 0);
EXPECT_EQ(f.rest(), "");
}
TEST(AdvanceIovecs, SkipsZeroLengthElements) {
char a[] = "ab";
char c[] = "cd";
struct iovec iov[3] = {{a, 2}, {nullptr, 0}, {c, 2}};
struct iovec *cur = iov;
int count = 3;
advance_iovecs(cur, count, 2);
EXPECT_EQ(cur, &iov[2]);
EXPECT_EQ(count, 1);
}
static constexpr uint32_t TIMEOUT = 20000;
TEST(StallClock, NotStarted) {
StallClock clock;
EXPECT_FALSE(clock.is_running());
EXPECT_FALSE(clock.is_expired(0, TIMEOUT));
EXPECT_FALSE(clock.is_expired(UINT32_MAX, TIMEOUT));
}
TEST(StallClock, ExpiresAtExactlyTheTimeout) {
StallClock clock;
clock.start(1000);
EXPECT_TRUE(clock.is_running());
EXPECT_FALSE(clock.is_expired(1000, TIMEOUT));
EXPECT_FALSE(clock.is_expired(1000 + TIMEOUT - 1, TIMEOUT));
EXPECT_TRUE(clock.is_expired(1000 + TIMEOUT, TIMEOUT));
EXPECT_TRUE(clock.is_expired(1000 + TIMEOUT + 1, TIMEOUT));
EXPECT_EQ(clock.elapsed(1000 + TIMEOUT), TIMEOUT);
}
TEST(StallClock, RunningClockKeepsItsStart) {
StallClock clock;
clock.start(1000);
clock.start(5000);
EXPECT_TRUE(clock.is_expired(1000 + TIMEOUT, TIMEOUT));
}
TEST(StallClock, Cleared) {
StallClock clock;
clock.start(1000);
clock.clear();
EXPECT_FALSE(clock.is_running());
EXPECT_FALSE(clock.is_expired(1000 + TIMEOUT, TIMEOUT));
clock.start(50000);
EXPECT_FALSE(clock.is_expired(50000 + TIMEOUT - 1, TIMEOUT));
EXPECT_TRUE(clock.is_expired(50000 + TIMEOUT, TIMEOUT));
}
TEST(StallClock, StartAtZeroStillRuns) {
StallClock clock;
clock.start(0);
EXPECT_TRUE(clock.is_running());
// Recorded as 1, so it expires one millisecond late
EXPECT_FALSE(clock.is_expired(TIMEOUT, TIMEOUT));
EXPECT_TRUE(clock.is_expired(TIMEOUT + 1, TIMEOUT));
}
TEST(StallClock, AcrossMillisWraparound) {
StallClock clock;
const uint32_t start = UINT32_MAX - 5000;
clock.start(start);
EXPECT_FALSE(clock.is_expired(UINT32_MAX, TIMEOUT));
EXPECT_FALSE(clock.is_expired(start + TIMEOUT - 1, TIMEOUT)); // wrapped to 14998
EXPECT_TRUE(clock.is_expired(start + TIMEOUT, TIMEOUT));
EXPECT_EQ(clock.elapsed(start + TIMEOUT), TIMEOUT);
}
} // namespace esphome::web_server::testing
#endif // USE_HOST