diff --git a/esphome/components/api/api_frame_helper.cpp b/esphome/components/api/api_frame_helper.cpp index fbee294022..5d51204cb6 100644 --- a/esphome/components/api/api_frame_helper.cpp +++ b/esphome/components/api/api_frame_helper.cpp @@ -100,72 +100,22 @@ const LogString *api_error_to_logstr(APIError err) { return LOG_STR("UNKNOWN"); } -// Default implementation for loop - handles sending buffered data +// Default implementation for loop - handles draining overflow buffer 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; - } + if (!this->overflow_buf_.empty() && this->overflow_buf_.try_drain(this->socket_.get()) == -1) { + const int sav_errno = errno; + HELPER_LOG("Socket write failed with errno %d", sav_errno); + if (this->check_socket_write_err_(sav_errno) != APIError::WOULD_BLOCK) + return APIError::SOCKET_WRITE_FAILED; } - return APIError::OK; // Convert WOULD_BLOCK to OK to avoid connection termination -} - -// 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_++; + // Convert WOULD_BLOCK to OK to avoid connection termination + return APIError::OK; } // This method writes data to socket or buffers it 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 + // Returns APIError::OK if all data was sent or successfully queued. + // Returns APIError::SOCKET_WRITE_FAILED if socket write failed, and sets state to FAILED. if (iovcnt == 0) return APIError::OK; // Nothing to do, success @@ -176,74 +126,34 @@ APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt, uint16_ } #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; - } - - // 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 - } + // If there is already backlogged data, try to drain then queue behind it + if (!this->overflow_buf_.empty()) { + this->overflow_buf_.try_drain(this->socket_.get()); + // If still backlogged, queue new data behind it; otherwise fall through to direct send + if (!this->overflow_buf_.empty()) + return this->enqueue_or_fail_(iov, iovcnt, total_write_len, 0); } - // Try to send directly if no buffered data + // No backlog — try to send directly // 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 (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)); + const int sav_errno = errno; + HELPER_LOG("Socket write failed with errno %d", sav_errno); + if (this->check_socket_write_err_(sav_errno) != APIError::WOULD_BLOCK) + return APIError::SOCKET_WRITE_FAILED; + // Socket would block — queue everything + return this->enqueue_or_fail_(iov, iovcnt, total_write_len, 0); } - 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 - } 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 - } + if (static_cast(sent) < total_write_len) { + // Partially sent — queue the remainder + return this->enqueue_or_fail_(iov, iovcnt, total_write_len, static_cast(sent)); } - return APIError::OK; // All buffers sent successfully + 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 e78c71507c..133f18c104 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" @@ -106,7 +107,7 @@ class APIFrameHelper { virtual APIError init() = 0; virtual APIError loop(); 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,26 @@ 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; } - }; - // 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_(); + // 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; + } - // 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_(); + // Enqueue IOV data into the overflow buffer, or fail the connection if full + APIError enqueue_or_fail_(const struct iovec *iov, int iovcnt, uint16_t total_len, uint16_t skip) { + if (!this->overflow_buf_.enqueue_iov(iov, iovcnt, total_len, skip)) { + this->state_ = State::FAILED; + return APIError::SOCKET_WRITE_FAILED; + } + return APIError::OK; + } // Socket ownership (4 bytes on 32-bit, 8 bytes on 64-bit) std::unique_ptr socket_; @@ -245,8 +244,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 +256,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_overflow_buffer.cpp b/esphome/components/api/api_overflow_buffer.cpp new file mode 100644 index 0000000000..af3b670e0e --- /dev/null +++ b/esphome/components/api/api_overflow_buffer.cpp @@ -0,0 +1,72 @@ +#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), 0 = would block + 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 0000000000..b0f79e1e9a --- /dev/null +++ b/esphome/components/api/api_overflow_buffer.h @@ -0,0 +1,72 @@ +#pragma once +#include +#include +#include + +#include "esphome/core/defines.h" +#ifdef USE_API + +#include "esphome/components/socket/socket.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 kernel 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 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, -1 on hard error (errno set). + /// 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