sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-10-03 18:42:23 +01:00
Code Issues Releases Wiki Activity

Merge branch 'bound_tx_buf' into integration

Browse Source
This commit is contained in:
J. Nick Koston
2025-10-01 18:20:40 +02:00
parent 2b374529f4 adade2952a
commit cc5b9ba4d6
4 changed files with 52 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
esphome/components/api/__init__.py
View File

@@ -61,6 +61,7 @@ CONF_HOMEASSISTANT_SERVICES = "homeassistant_services"
CONF_HOMEASSISTANT_STATES = "homeassistant_states"
CONF_LISTEN_BACKLOG = "listen_backlog"
CONF_MAX_CONNECTIONS = "max_connections"
CONF_MAX_SEND_QUEUE = "max_send_queue"
def validate_encryption_key(value):
@@ -185,6 +186,19 @@ CONFIG_SCHEMA = cv.All(
host=8, # Abundant resources
ln882x=8, # Moderate RAM
): cv.int_range(min=1, max=20),
# 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:
cv.SplitDefault(
CONF_MAX_SEND_QUEUE,
esp8266=5, # Limited RAM, need to fail fast
esp32=8, # More RAM, can buffer more
rp2040=5, # Limited RAM
bk72xx=8, # Moderate RAM
rtl87xx=8, # Moderate RAM
host=16, # Abundant resources
ln882x=8, # Moderate RAM
): cv.int_range(min=1, max=64),
}
).extend(cv.COMPONENT_SCHEMA),
cv.rename_key(CONF_SERVICES, CONF_ACTIONS),
@@ -207,6 +221,7 @@ async def to_code(config):
cg.add(var.set_listen_backlog(config[CONF_LISTEN_BACKLOG]))
if CONF_MAX_CONNECTIONS in config:
cg.add(var.set_max_connections(config[CONF_MAX_CONNECTIONS]))
cg.add_define("API_MAX_SEND_QUEUE", config.get(CONF_MAX_SEND_QUEUE, 5))
# Set USE_API_SERVICES if any services are enabled
if config.get(CONF_ACTIONS) or config[CONF_CUSTOM_SERVICES]:

46
esphome/components/api/api_frame_helper.cpp
View File

@@ -81,7 +81,7 @@ const LogString *api_error_to_logstr(APIError err) {
// Default implementation for loop - handles sending buffered data
APIError APIFrameHelper::loop() {
if (!this->tx_buf_.empty()) {
if (this->tx_buf_count_ > 0) {
APIError err = try_send_tx_buf_();
if (err != APIError::OK && err != APIError::WOULD_BLOCK) {
return err;
@@ -103,9 +103,16 @@ APIError APIFrameHelper::handle_socket_write_error_() {
// 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) {
SendBuffer buffer;
buffer.size = total_write_len - offset;
buffer.data = std::make_unique<uint8_t[]>(buffer.size);
// 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;
}
auto buffer = std::make_unique<SendBuffer>();
buffer->size = total_write_len - offset;
buffer->data = std::make_unique<uint8_t[]>(buffer->size);
uint16_t to_skip = offset;
uint16_t write_pos = 0;
@@ -118,12 +125,16 @@ void APIFrameHelper::buffer_data_from_iov_(const struct iovec *iov, int iovcnt,
// Include this segment (partially or fully)
const uint8_t *src = reinterpret_cast<uint8_t *>(iov[i].iov_base) + to_skip;
uint16_t len = static_cast<uint16_t>(iov[i].iov_len) - to_skip;
std::memcpy(buffer.data.get() + write_pos, src, len);
std::memcpy(buffer->data.get() + write_pos, src, len);
write_pos += len;
to_skip = 0;
}
}
this->tx_buf_.push_back(std::move(buffer));
// Add to circular buffer
this->tx_buf_[this->tx_buf_tail_] = std::move(buffer);
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
@@ -141,7 +152,7 @@ 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_.empty()) {
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) {
@@ -150,7 +161,7 @@ APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt, uint16_
// If there is still data in the buffer, we can't send, buffer
// the new data and return
if (!this->tx_buf_.empty()) {
if (this->tx_buf_count_ > 0) {
this->buffer_data_from_iov_(iov, iovcnt, total_write_len, 0);
return APIError::OK; // Success, data buffered
}
@@ -178,32 +189,31 @@ APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt, uint16_
}
// Common implementation for trying to send buffered data
// IMPORTANT: Caller MUST ensure tx_buf_ is not empty before calling this method
// 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
bool tx_buf_empty = false;
while (!tx_buf_empty) {
while (this->tx_buf_count_ > 0) {
// Get the first buffer in the queue
SendBuffer &front_buffer = this->tx_buf_.front();
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());
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<uint16_t>(sent) < front_buffer.remaining()) {
} else if (static_cast<uint16_t>(sent) < front_buffer->remaining()) {
// Partially sent, update offset
// Cast to ensure no overflow issues with uint16_t
front_buffer.offset += static_cast<uint16_t>(sent);
front_buffer->offset += static_cast<uint16_t>(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_.pop_front();
// Update empty status for the loop condition
tx_buf_empty = this->tx_buf_.empty();
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
}
}

12
esphome/components/api/api_frame_helper.h
View File

@@ -1,7 +1,8 @@
#pragma once
#include <array>
#include <cstdint>
#include <deque>
#include <limits>
#include <memory>
#include <span>
#include <utility>
#include <vector>
@@ -89,7 +90,7 @@ class APIFrameHelper {
virtual APIError init() = 0;
virtual APIError loop();
virtual APIError read_packet(ReadPacketBuffer *buffer) = 0;
bool can_write_without_blocking() { return state_ == State::DATA && tx_buf_.empty(); }
bool can_write_without_blocking() { return state_ == State::DATA && tx_buf_count_ == 0; }
std::string getpeername() { return socket_->getpeername(); }
int getpeername(struct sockaddr *addr, socklen_t *addrlen) { return socket_->getpeername(addr, addrlen); }
APIError close() {
@@ -171,7 +172,7 @@ class APIFrameHelper {
};
// Containers (size varies, but typically 12+ bytes on 32-bit)
std::deque<SendBuffer> tx_buf_;
std::array<std::unique_ptr<SendBuffer>, API_MAX_SEND_QUEUE> tx_buf_;
std::vector<struct iovec> reusable_iovs_;
std::vector<uint8_t> rx_buf_;
@@ -184,7 +185,10 @@ class APIFrameHelper {
State state_{State::INITIALIZE};
uint8_t frame_header_padding_{0};
uint8_t frame_footer_size_{0};
// 5 bytes total, 3 bytes padding
uint8_t tx_buf_head_{0};
uint8_t tx_buf_tail_{0};
uint8_t tx_buf_count_{0};
// 8 bytes total, 0 bytes padding
// Common initialization for both plaintext and noise protocols
APIError init_common_();

1
esphome/core/defines.h
View File

@@ -115,6 +115,7 @@
#define USE_API_NOISE
#define USE_API_PLAINTEXT
#define USE_API_SERVICES
#define API_MAX_SEND_QUEUE 8
#define USE_MD5
#define USE_SHA256
#define USE_MQTT