From 2271ac64701b5524997d2c1c7335b585a97cc939 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 18 Mar 2026 18:41:45 -1000 Subject: [PATCH] [api] Extract overflow buffer from frame helper into APIOverflowBuffer (#14871) --- esphome/components/api/__init__.py | 5 +- esphome/components/api/api_frame_helper.cpp | 161 ++++-------------- esphome/components/api/api_frame_helper.h | 43 ++--- .../components/api/api_frame_helper_noise.cpp | 6 +- .../api/api_frame_helper_plaintext.cpp | 6 +- .../components/api/api_overflow_buffer.cpp | 73 ++++++++ esphome/components/api/api_overflow_buffer.h | 76 +++++++++ 7 files changed, 214 insertions(+), 156 deletions(-) create mode 100644 esphome/components/api/api_overflow_buffer.cpp create mode 100644 esphome/components/api/api_overflow_buffer.h diff --git a/esphome/components/api/__init__.py b/esphome/components/api/__init__.py index 9772e6afca7..4c3cf81927d 100644 --- a/esphome/components/api/__init__.py +++ b/esphome/components/api/__init__.py @@ -301,11 +301,12 @@ CONFIG_SCHEMA = cv.All( # Maximum queued send buffers per connection before dropping connection # Each buffer uses ~8-12 bytes overhead plus actual message size # Platform defaults based on available RAM and typical message rates: + # CONF_MAX_SEND_QUEUE defaults are power of 2 for efficient modulo cv.SplitDefault( CONF_MAX_SEND_QUEUE, - esp8266=5, # Limited RAM, need to fail fast + esp8266=4, # Limited RAM, need to fail fast esp32=8, # More RAM, can buffer more - rp2040=5, # Limited RAM + rp2040=8, # Moderate RAM bk72xx=8, # Moderate RAM nrf52=8, # Moderate RAM rtl87xx=8, # Moderate RAM diff --git a/esphome/components/api/api_frame_helper.cpp b/esphome/components/api/api_frame_helper.cpp index fbee2940226..6d3bd51b586 100644 --- a/esphome/components/api/api_frame_helper.cpp +++ b/esphome/components/api/api_frame_helper.cpp @@ -100,150 +100,61 @@ const LogString *api_error_to_logstr(APIError err) { return LOG_STR("UNKNOWN"); } -// Default implementation for loop - handles sending buffered data -APIError APIFrameHelper::loop() { - if (this->tx_buf_count_ > 0) { - APIError err = try_send_tx_buf_(); - if (err != APIError::OK && err != APIError::WOULD_BLOCK) { - return err; +APIError APIFrameHelper::drain_overflow_and_handle_errors_() { + if (this->overflow_buf_.try_drain(this->socket_.get()) == -1) { + int err = errno; + if (this->check_socket_write_err_(err) != APIError::WOULD_BLOCK) { + HELPER_LOG("Socket write failed with errno %d", err); + return APIError::SOCKET_WRITE_FAILED; } } - return APIError::OK; // Convert WOULD_BLOCK to OK to avoid connection termination + return APIError::OK; } -// Common socket write error handling -APIError APIFrameHelper::handle_socket_write_error_() { - const int err = errno; - if (err == EWOULDBLOCK || err == EAGAIN) { - return APIError::WOULD_BLOCK; - } - HELPER_LOG("Socket write failed with errno %d", err); - this->state_ = State::FAILED; - return APIError::SOCKET_WRITE_FAILED; -} - -// Helper method to buffer data from IOVs -void APIFrameHelper::buffer_data_from_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len, - uint16_t offset) { - // Check if queue is full - if (this->tx_buf_count_ >= API_MAX_SEND_QUEUE) { - HELPER_LOG("Send queue full (%u buffers), dropping connection", this->tx_buf_count_); - this->state_ = State::FAILED; - return; - } - - uint16_t buffer_size = total_write_len - offset; - auto &buffer = this->tx_buf_[this->tx_buf_tail_]; - buffer = std::make_unique(SendBuffer{ - .data = std::make_unique(buffer_size), - .size = buffer_size, - .offset = 0, - }); - - uint16_t to_skip = offset; - uint16_t write_pos = 0; - - for (int i = 0; i < iovcnt; i++) { - if (to_skip >= iov[i].iov_len) { - // Skip this entire segment - to_skip -= static_cast(iov[i].iov_len); - } else { - // Include this segment (partially or fully) - const uint8_t *src = reinterpret_cast(iov[i].iov_base) + to_skip; - uint16_t len = static_cast(iov[i].iov_len) - to_skip; - std::memcpy(buffer->data.get() + write_pos, src, len); - write_pos += len; - to_skip = 0; - } - } - - // Update circular buffer tracking - this->tx_buf_tail_ = (this->tx_buf_tail_ + 1) % API_MAX_SEND_QUEUE; - this->tx_buf_count_++; -} - -// This method writes data to socket or buffers it +// Write data to socket, overflow to backlog buffer if LWIP TCP send buffer is full. +// Returns OK if all data was sent or successfully queued. +// Returns SOCKET_WRITE_FAILED on hard error (sets state to FAILED). APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt, uint16_t total_write_len) { - // Returns APIError::OK if successful (or would block, but data has been buffered) - // Returns APIError::SOCKET_WRITE_FAILED if socket write failed, and sets state to FAILED - - if (iovcnt == 0) - return APIError::OK; // Nothing to do, success - #ifdef HELPER_LOG_PACKETS for (int i = 0; i < iovcnt; i++) { LOG_PACKET_SENDING(reinterpret_cast(iov[i].iov_base), iov[i].iov_len); } #endif - // Try to send any existing buffered data first if there is any - if (this->tx_buf_count_ > 0) { - APIError send_result = try_send_tx_buf_(); - // If real error occurred (not just WOULD_BLOCK), return it - if (send_result != APIError::OK && send_result != APIError::WOULD_BLOCK) { - return send_result; - } + uint16_t skip = 0; - // If there is still data in the buffer, we can't send, buffer - // the new data and return - if (this->tx_buf_count_ > 0) { - this->buffer_data_from_iov_(iov, iovcnt, total_write_len, 0); - return APIError::OK; // Success, data buffered - } + // Drain any existing backlog first + if (!this->overflow_buf_.empty()) [[unlikely]] { + APIError err = this->drain_overflow_and_handle_errors_(); + if (err != APIError::OK) + return err; } - // Try to send directly if no buffered data - // Optimize for single iovec case (common for plaintext API) - ssize_t sent = - (iovcnt == 1) ? this->socket_->write(iov[0].iov_base, iov[0].iov_len) : this->socket_->writev(iov, iovcnt); + // If backlog is clear, try direct send + if (this->overflow_buf_.empty()) [[likely]] { + ssize_t sent = + (iovcnt == 1) ? this->socket_->write(iov[0].iov_base, iov[0].iov_len) : this->socket_->writev(iov, iovcnt); - if (sent == -1) { - APIError err = this->handle_socket_write_error_(); - if (err == APIError::WOULD_BLOCK) { - // Socket would block, buffer the data - this->buffer_data_from_iov_(iov, iovcnt, total_write_len, 0); - return APIError::OK; // Success, data buffered - } - return err; // Socket write failed - } else if (static_cast(sent) < total_write_len) { - // Partially sent, buffer the remaining data - this->buffer_data_from_iov_(iov, iovcnt, total_write_len, static_cast(sent)); - } - - return APIError::OK; // Success, all data sent or buffered -} - -// Common implementation for trying to send buffered data -// IMPORTANT: Caller MUST ensure tx_buf_count_ > 0 before calling this method -APIError APIFrameHelper::try_send_tx_buf_() { - // Try to send from tx_buf - we assume it's not empty as it's the caller's responsibility to check - while (this->tx_buf_count_ > 0) { - // Get the first buffer in the queue - SendBuffer *front_buffer = this->tx_buf_[this->tx_buf_head_].get(); - - // Try to send the remaining data in this buffer - ssize_t sent = this->socket_->write(front_buffer->current_data(), front_buffer->remaining()); - - if (sent == -1) { - return this->handle_socket_write_error_(); - } else if (sent == 0) { - // Nothing sent but not an error - return APIError::WOULD_BLOCK; - } else if (static_cast(sent) < front_buffer->remaining()) { - // Partially sent, update offset - // Cast to ensure no overflow issues with uint16_t - front_buffer->offset += static_cast(sent); - return APIError::WOULD_BLOCK; // Stop processing more buffers if we couldn't send a complete buffer + if (sent == -1) [[unlikely]] { + int err = errno; + if (this->check_socket_write_err_(err) != APIError::WOULD_BLOCK) { + HELPER_LOG("Socket write failed with errno %d", err); + return APIError::SOCKET_WRITE_FAILED; + } + } else if (static_cast(sent) >= total_write_len) [[likely]] { + return APIError::OK; } else { - // Buffer completely sent, remove it from the queue - this->tx_buf_[this->tx_buf_head_].reset(); - this->tx_buf_head_ = (this->tx_buf_head_ + 1) % API_MAX_SEND_QUEUE; - this->tx_buf_count_--; - // Continue loop to try sending the next buffer + skip = static_cast(sent); } } - return APIError::OK; // All buffers sent successfully + // Queue unsent data into overflow buffer + if (!this->overflow_buf_.enqueue_iov(iov, iovcnt, total_write_len, skip)) { + HELPER_LOG("Overflow buffer full, dropping connection"); + this->state_ = State::FAILED; + return APIError::SOCKET_WRITE_FAILED; + } + return APIError::OK; } const char *APIFrameHelper::get_peername_to(std::span buf) const { diff --git a/esphome/components/api/api_frame_helper.h b/esphome/components/api/api_frame_helper.h index e78c71507c8..72ccf8aa562 100644 --- a/esphome/components/api/api_frame_helper.h +++ b/esphome/components/api/api_frame_helper.h @@ -9,6 +9,7 @@ #include "esphome/core/defines.h" #ifdef USE_API #include "esphome/components/api/api_buffer.h" +#include "esphome/components/api/api_overflow_buffer.h" #include "esphome/components/socket/socket.h" #include "esphome/core/application.h" #include "esphome/core/log.h" @@ -104,9 +105,9 @@ class APIFrameHelper { } virtual ~APIFrameHelper() = default; virtual APIError init() = 0; - virtual APIError loop(); + virtual APIError loop() = 0; virtual APIError read_packet(ReadPacketBuffer *buffer) = 0; - bool can_write_without_blocking() { return this->state_ == State::DATA && this->tx_buf_count_ == 0; } + bool can_write_without_blocking() { return this->state_ == State::DATA && this->overflow_buf_.empty(); } int getpeername(struct sockaddr *addr, socklen_t *addrlen) { return socket_->getpeername(addr, addrlen); } APIError close() { if (state_ == State::CLOSED) @@ -189,28 +190,23 @@ class APIFrameHelper { } protected: - // Buffer containing data to be sent - struct SendBuffer { - std::unique_ptr data; - uint16_t size{0}; // Total size of the buffer - uint16_t offset{0}; // Current offset within the buffer - - // Using uint16_t reduces memory usage since ESPHome API messages are limited to UINT16_MAX (65535) bytes - uint16_t remaining() const { return size - offset; } - const uint8_t *current_data() const { return data.get() + offset; } - }; + // Drain backlogged overflow data to the socket and handle errors. + // Called when overflow_buf_.empty() is false. Out-of-line to keep the + // fast path (empty check) inline at call sites. + // Returns OK for transient errors (WOULD_BLOCK), SOCKET_WRITE_FAILED for hard errors. + APIError drain_overflow_and_handle_errors_(); // Common implementation for writing raw data to socket APIError write_raw_(const struct iovec *iov, int iovcnt, uint16_t total_write_len); - // Try to send data from the tx buffer - APIError try_send_tx_buf_(); - - // Helper method to buffer data from IOVs - void buffer_data_from_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len, uint16_t offset); - - // Common socket write error handling - APIError handle_socket_write_error_(); + // Check if a socket write errno is a hard error (not WOULD_BLOCK/EAGAIN). + // Returns WOULD_BLOCK for transient errors, SOCKET_WRITE_FAILED for hard errors. + APIError check_socket_write_err_(int err) { + if (err == EWOULDBLOCK || err == EAGAIN) + return APIError::WOULD_BLOCK; + this->state_ = State::FAILED; + return APIError::SOCKET_WRITE_FAILED; + } // Socket ownership (4 bytes on 32-bit, 8 bytes on 64-bit) std::unique_ptr socket_; @@ -245,8 +241,8 @@ class APIFrameHelper { return APIError::WOULD_BLOCK; } - // Containers (size varies, but typically 12+ bytes on 32-bit) - std::array, API_MAX_SEND_QUEUE> tx_buf_; + // Backlog for unsent data when TCP send buffer is full (rarely used in production) + APIOverflowBuffer overflow_buf_; APIBuffer rx_buf_; // Client name buffer - stores name from Hello message or initial peername @@ -257,9 +253,6 @@ class APIFrameHelper { State state_{State::INITIALIZE}; uint8_t frame_header_padding_{0}; uint8_t frame_footer_size_{0}; - uint8_t tx_buf_head_{0}; - uint8_t tx_buf_tail_{0}; - uint8_t tx_buf_count_{0}; // Nagle batching counter for log messages. 0 means NODELAY is enabled (immediate send). // Values 1..LOG_NAGLE_COUNT count log messages in the current Nagle batch. // After LOG_NAGLE_COUNT logs, we flush by re-enabling NODELAY and resetting to 0. diff --git a/esphome/components/api/api_frame_helper_noise.cpp b/esphome/components/api/api_frame_helper_noise.cpp index b635d84f168..78e87793fc8 100644 --- a/esphome/components/api/api_frame_helper_noise.cpp +++ b/esphome/components/api/api_frame_helper_noise.cpp @@ -153,8 +153,10 @@ APIError APINoiseFrameHelper::loop() { } } - // Use base class implementation for buffer sending - return APIFrameHelper::loop(); + if (!this->overflow_buf_.empty()) [[unlikely]] { + return this->drain_overflow_and_handle_errors_(); + } + return APIError::OK; } /** Read a packet into the rx_buf_. diff --git a/esphome/components/api/api_frame_helper_plaintext.cpp b/esphome/components/api/api_frame_helper_plaintext.cpp index e97b558fa39..9e669b31ee2 100644 --- a/esphome/components/api/api_frame_helper_plaintext.cpp +++ b/esphome/components/api/api_frame_helper_plaintext.cpp @@ -64,8 +64,10 @@ APIError APIPlaintextFrameHelper::loop() { if (state_ != State::DATA) { return APIError::BAD_STATE; } - // Use base class implementation for buffer sending - return APIFrameHelper::loop(); + if (!this->overflow_buf_.empty()) [[unlikely]] { + return this->drain_overflow_and_handle_errors_(); + } + return APIError::OK; } /** Read a packet into the rx_buf_. diff --git a/esphome/components/api/api_overflow_buffer.cpp b/esphome/components/api/api_overflow_buffer.cpp new file mode 100644 index 00000000000..e242d4553e4 --- /dev/null +++ b/esphome/components/api/api_overflow_buffer.cpp @@ -0,0 +1,73 @@ +#include "api_overflow_buffer.h" +#ifdef USE_API +#include + +namespace esphome::api { + +APIOverflowBuffer::~APIOverflowBuffer() { + for (auto *entry : this->queue_) { + if (entry != nullptr) + Entry::destroy(entry); + } +} + +ssize_t APIOverflowBuffer::try_drain(socket::Socket *socket) { + while (this->count_ > 0) { + Entry *front = this->queue_[this->head_]; + + ssize_t sent = socket->write(front->current_data(), front->remaining()); + + if (sent <= 0) { + // -1 = error (caller checks errno for EWOULDBLOCK vs hard error) + // 0 = nothing sent (treat as no progress) + return sent; + } + + if (static_cast(sent) < front->remaining()) { + // Partially sent, update offset and stop + front->offset += static_cast(sent); + return sent; + } + + // Entry fully sent — free it and advance + Entry::destroy(front); + this->queue_[this->head_] = nullptr; + this->head_ = (this->head_ + 1) % API_MAX_SEND_QUEUE; + this->count_--; + } + + return 0; // All drained +} + +bool APIOverflowBuffer::enqueue_iov(const struct iovec *iov, int iovcnt, uint16_t total_len, uint16_t skip) { + if (this->count_ >= API_MAX_SEND_QUEUE) + return false; + + uint16_t buffer_size = total_len - skip; + // NOLINTNEXTLINE(cppcoreguidelines-owning-memory) + auto *entry = new Entry{new uint8_t[buffer_size], buffer_size, 0}; + this->queue_[this->tail_] = entry; + + uint16_t to_skip = skip; + uint16_t write_pos = 0; + + for (int i = 0; i < iovcnt; i++) { + if (to_skip >= iov[i].iov_len) { + to_skip -= static_cast(iov[i].iov_len); + } else { + const uint8_t *src = reinterpret_cast(iov[i].iov_base) + to_skip; + uint16_t len = static_cast(iov[i].iov_len) - to_skip; + std::memcpy(entry->data + write_pos, src, len); + write_pos += len; + to_skip = 0; + } + } + + this->tail_ = (this->tail_ + 1) % API_MAX_SEND_QUEUE; + this->count_++; + return true; +} + +} // namespace esphome::api + +#endif // USE_API diff --git a/esphome/components/api/api_overflow_buffer.h b/esphome/components/api/api_overflow_buffer.h new file mode 100644 index 00000000000..19aae680f01 --- /dev/null +++ b/esphome/components/api/api_overflow_buffer.h @@ -0,0 +1,76 @@ +#pragma once +#include +#include +#include + +#include "esphome/core/defines.h" +#ifdef USE_API + +#include "esphome/components/socket/headers.h" +#include "esphome/components/socket/socket.h" +#include "esphome/core/helpers.h" + +namespace esphome::api { + +/// Circular queue of heap-allocated byte buffers used as a TCP send backlog. +/// +/// Under normal operation this buffer is **never used** — data goes straight +/// from the frame helper to the socket. It only fills when the LWIP TCP +/// send buffer is full (slow client, congested network, heavy logging). +/// The queue drains automatically on subsequent write/loop calls once the +/// socket becomes writable again. +/// +/// Capacity is compile-time-fixed via API_MAX_SEND_QUEUE (set from Python +/// config). If the queue fills completely the connection is marked failed. +class APIOverflowBuffer { + public: + /// A single heap-allocated send-backlog entry. + /// Lifetime is manually managed — see destroy(). + struct Entry { + uint8_t *data; + uint16_t size; // Total size of the buffer + uint16_t offset; // Current send offset within the buffer + + uint16_t remaining() const { return this->size - this->offset; } + const uint8_t *current_data() const { return this->data + this->offset; } + + /// Free this entry and its data buffer. + static ESPHOME_ALWAYS_INLINE void destroy(Entry *entry) { + delete[] entry->data; + delete entry; // NOLINT(cppcoreguidelines-owning-memory) + } + }; + + ~APIOverflowBuffer(); + + /// True when no backlogged data is waiting. + bool empty() const { return this->count_ == 0; } + + /// True when the queue has no room for another entry. + bool full() const { return this->count_ >= API_MAX_SEND_QUEUE; } + + /// Number of entries currently queued. + uint8_t count() const { return this->count_; } + + /// Try to drain queued data to the socket. + /// Returns bytes-written > 0 on success/partial, 0 if all drained or no progress, + /// -1 on error (caller must check errno to distinguish EWOULDBLOCK from hard errors). + /// Callers only need to act on -1; 0 and positive values both mean "no error". + /// Frees entries as they are fully sent. + ssize_t try_drain(socket::Socket *socket); + + /// Enqueue unsent IOV data into the backlog. + /// Copies iov data starting at byte offset `skip` into a new entry. + /// Returns false if the queue is full (caller should fail the connection). + bool enqueue_iov(const struct iovec *iov, int iovcnt, uint16_t total_len, uint16_t skip); + + protected: + std::array queue_{}; + uint8_t head_{0}; + uint8_t tail_{0}; + uint8_t count_{0}; +}; + +} // namespace esphome::api + +#endif // USE_API