Compare commits

...
20 changed files with 750 additions and 44 deletions
+8 -8
View File
@@ -412,15 +412,15 @@ void APIConnection::finalize_iterator_sync_() {
}
void APIConnection::process_iterator_batch_(ComponentIterator &iterator) {
size_t initial_size = this->deferred_batch_.size();
size_t max_batch = MAX_INITIAL_PER_BATCH;
while (!iterator.completed() && (this->deferred_batch_.size() - initial_size) < max_batch) {
iterator.advance();
}
// Budget by remaining batch capacity so a pass cannot overfill the batch;
// stops early on a refused send and resumes next loop pass
size_t batch_size = this->deferred_batch_.size();
if (batch_size < MAX_INITIAL_BATCH_SIZE)
iterator.try_advance(MAX_INITIAL_BATCH_SIZE - batch_size);
// If the batch is full, process it immediately
// Note: iterator.advance() already calls schedule_batch_() via schedule_message_()
if (this->deferred_batch_.size() >= max_batch) {
// Flush immediately once enough is queued (not guaranteed every pass);
// partial batches go out via the batch timer or finalize_iterator_sync_()
if (this->deferred_batch_.size() >= MAX_INITIAL_BATCH_SIZE) {
this->process_batch_();
}
}
+4 -4
View File
@@ -53,11 +53,11 @@ void log_dropped_message(const char *tag, int line, const LogString *what);
// Keepalive timeout in milliseconds
static constexpr uint32_t KEEPALIVE_TIMEOUT_MS = 60000;
// Maximum number of entities to process in a single batch during initial state/info sending
static constexpr size_t MAX_INITIAL_PER_BATCH = 34;
// Deferred batch size cap during initial state/info sync
static constexpr size_t MAX_INITIAL_BATCH_SIZE = 34;
// Verify MAX_MESSAGES_PER_BATCH (defined in api_frame_helper.h) can hold the initial batch
static_assert(MAX_MESSAGES_PER_BATCH >= MAX_INITIAL_PER_BATCH,
"MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_PER_BATCH");
static_assert(MAX_MESSAGES_PER_BATCH >= MAX_INITIAL_BATCH_SIZE,
"MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_BATCH_SIZE");
#ifdef USE_BENCHMARK
class APIConnection;
+1 -1
View File
@@ -36,7 +36,7 @@ static constexpr uint16_t MAX_MESSAGE_SIZE = 32768; // 32 KiB for ESP32 and oth
static constexpr uint16_t RX_BUF_NULL_TERMINATOR = 1;
// Maximum number of messages to batch in a single write operation
// Must be >= MAX_INITIAL_PER_BATCH in api_connection.h (enforced by static_assert there)
// Must be >= MAX_INITIAL_BATCH_SIZE in api_connection.h (enforced by static_assert there)
static constexpr size_t MAX_MESSAGES_PER_BATCH = 34;
// Max client name length (e.g., "Home Assistant 2026.1.0.dev0" = 28 chars)
+9 -1
View File
@@ -95,9 +95,17 @@ bool ListEntitiesIterator::on_end() { return this->client_->send_list_info_done(
ListEntitiesIterator::ListEntitiesIterator(APIConnection *client) : client_(client) {}
#ifdef USE_API_USER_DEFINED_ACTIONS
// Yield after every Nth service; bounds direct (non-batched) writes per loop pass
static constexpr uint8_t SERVICE_YIELD_INTERVAL = 3;
bool ListEntitiesIterator::on_service(UserServiceDescriptor *service) {
auto resp = service->encode_list_service_response();
return this->client_->send_message(resp);
if (!this->client_->send_message(resp))
return false;
// at_ is this service's index
if ((this->at_ + 1) % SERVICE_YIELD_INTERVAL == 0)
this->yield_after_step_();
return true;
}
#endif
@@ -35,7 +35,6 @@ class ListEntitiesIterator final : public ComponentIterator {
#undef ENTITY_TYPE_
#undef ENTITY_CONTROLLER_TYPE_
// NOLINTEND(bugprone-macro-parentheses)
bool completed() { return this->state_ == IteratorState::NONE; }
protected:
const WebServer *web_server_;
+2 -8
View File
@@ -214,8 +214,8 @@ void DeferredUpdateEventSource::process_deferred_queue_() {
void DeferredUpdateEventSource::loop() {
process_deferred_queue_();
if (!this->entities_iterator_.completed())
this->entities_iterator_.advance();
// One step per loop; refusals retry next pass
this->entities_iterator_.try_advance(1);
}
void DeferredUpdateEventSource::deferrable_send_state(void *source, const char *event_type,
@@ -321,12 +321,6 @@ void DeferredUpdateEventSourceList::on_client_connect_(DeferredUpdateEventSource
#endif
source->entities_iterator_.begin(ws->include_internal_);
// just dump them all up-front and take advantage of the deferred queue
// on second thought that takes too long, but leaving the commented code here for debug purposes
// while(!source->entities_iterator_.completed()) {
// source->entities_iterator_.advance();
//}
});
}
@@ -935,8 +935,8 @@ void AsyncEventSourceResponse::process_buffer_() {
void AsyncEventSourceResponse::loop() {
process_buffer_();
process_deferred_queue_();
if (!this->entities_iterator_.completed())
this->entities_iterator_.advance();
// One step per loop; refusals retry next pass
this->entities_iterator_.try_advance(1);
}
bool AsyncEventSourceResponse::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
+14 -11
View File
@@ -22,23 +22,23 @@ void ComponentIterator::advance_platform_() {
this->at_ = 0;
}
void ComponentIterator::advance() {
bool ComponentIterator::advance_step_() {
switch (this->state_) {
case IteratorState::NONE:
// not started
return;
return false;
case IteratorState::BEGIN:
if (this->on_begin()) {
advance_platform_();
return true;
}
break;
return false;
// Entity iterator cases (generated from entity_types.h)
// NOLINTBEGIN(bugprone-macro-parentheses)
#define ENTITY_TYPE_(type, singular, plural, count, upper) \
case IteratorState::upper: \
this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular); \
break;
return this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular);
#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
ENTITY_TYPE_(type, singular, plural, count, upper)
#include "esphome/core/entity_types.h"
@@ -48,26 +48,29 @@ void ComponentIterator::advance() {
#ifdef USE_API_USER_DEFINED_ACTIONS
case IteratorState::SERVICE:
this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service);
break;
return this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service);
#endif
#ifdef USE_CAMERA
case IteratorState::CAMERA: {
camera::Camera *camera_instance = camera::Camera::instance();
if (camera_instance != nullptr && (!camera_instance->is_internal() || this->include_internal_)) {
this->on_camera(camera_instance);
if (camera_instance != nullptr && (!camera_instance->is_internal() || this->include_internal_) &&
!this->on_camera(camera_instance)) {
return false;
}
advance_platform_();
} break;
return true;
}
#endif
case IteratorState::MAX:
if (this->on_end()) {
this->state_ = IteratorState::NONE;
return true;
}
return;
return false;
}
return false;
}
bool ComponentIterator::on_end() { return true; }
+35 -8
View File
@@ -30,7 +30,23 @@ class RadioFrequency;
class ComponentIterator {
public:
void begin(bool include_internal = false);
void advance();
/// Run up to max_steps iteration steps; stops early when iteration
/// completes or a callback refuses (that step is retried on the next
/// call). Inline so an idle (completed) iterator costs one compare, no call.
ESPHOME_ALWAYS_INLINE void try_advance(size_t max_steps) {
size_t steps = 0;
while (steps < max_steps && !this->completed()) {
this->yield_requested_ = false;
if (!this->advance_step_())
break;
steps++;
if (this->yield_requested_)
break;
}
}
// Remove before 2027.3.0
ESPDEPRECATED("Use try_advance() instead. Removed in 2027.3.0", "2026.8.1")
void advance() { this->try_advance(1); }
bool completed() const { return this->state_ == IteratorState::NONE; }
virtual bool on_begin();
// Pure virtual entity callbacks (generated from entity_types.h)
@@ -73,23 +89,34 @@ class ComponentIterator {
#endif
MAX,
};
/// End the current try_advance() pass after this step; lets callbacks
/// that write directly to the socket cap direct writes per pass.
void yield_after_step_() { this->yield_requested_ = true; }
uint16_t at_{0}; // Supports up to 65,535 entities per type
IteratorState state_{IteratorState::NONE};
bool include_internal_{false};
bool yield_requested_ : 1 {false};
bool include_internal_ : 1 {false};
template<typename Container>
void process_platform_item_(const Container &items,
bool process_platform_item_(const Container &items,
bool (ComponentIterator::*on_item)(typename Container::value_type)) {
if (this->at_ >= items.size()) {
this->advance_platform_();
} else {
typename Container::value_type item = items[this->at_];
if ((item->is_internal() && !this->include_internal_) || (this->*on_item)(item)) {
this->at_++;
}
return true;
}
typename Container::value_type item = items[this->at_];
if ((item->is_internal() && !this->include_internal_) || (this->*on_item)(item)) {
this->at_++;
return true;
}
return false;
}
/// One iteration step; false if no progress was made (callback refused
/// or iterator not running).
bool advance_step_();
void advance_platform_();
};
+11
View File
@@ -0,0 +1,11 @@
import esphome.codegen as cg
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# No host camera platform exists to emit USE_CAMERA; define it here so
# the iterator CAMERA state compiles into the test binary.
async def to_code_testing(config):
cg.add_define("USE_CAMERA")
manifest.to_code = to_code_testing
@@ -0,0 +1,79 @@
#include <gtest/gtest.h>
#include "esphome/core/component_iterator.h"
#ifdef USE_CAMERA
#include "esphome/components/camera/camera.h"
namespace esphome::testing {
class StubCamera : public camera::Camera {
public:
void add_listener(camera::CameraListener *listener) override {}
camera::CameraImageReader *create_image_reader() override { return nullptr; }
void request_image(camera::CameraRequester requester) override {}
void start_stream(camera::CameraRequester requester) override {}
void stop_stream(camera::CameraRequester requester) override {}
};
// Iterator that accepts everything except the camera, which can refuse a
// configurable number of times. The CAMERA state is a singleton path
// distinct from process_platform_item_; this pins the same contract:
// a refused camera is re-offered, never skipped.
class CameraRefusingIterator : public ComponentIterator {
public:
// NOLINTBEGIN(bugprone-macro-parentheses)
#define ENTITY_TYPE_(type, singular, plural, count, upper) \
bool on_##singular(type *obj) override { return true; }
#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
ENTITY_TYPE_(type, singular, plural, count, upper)
#include "esphome/core/entity_types.h"
#undef ENTITY_TYPE_
#undef ENTITY_CONTROLLER_TYPE_
// NOLINTEND(bugprone-macro-parentheses)
bool on_camera(camera::Camera *obj) override {
this->camera_calls++;
if (this->camera_refusals > 0) {
this->camera_refusals--;
return false;
}
return true;
}
int camera_calls{0};
int camera_refusals{0};
};
// Far above the fixed number of iterator states
static constexpr size_t BIG_BUDGET = 1000;
class ComponentIteratorCameraTest : public ::testing::Test {
protected:
void SetUp() override {
// Constructing a Camera installs the process-wide singleton
static StubCamera stub_camera;
ASSERT_EQ(camera::Camera::instance(), &stub_camera);
}
};
TEST_F(ComponentIteratorCameraTest, RefusedCameraIsReofferedNotSkipped) {
CameraRefusingIterator it;
it.camera_refusals = 2;
it.begin();
// Runs until the camera refuses, which stops the pass
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.camera_calls, 1);
EXPECT_FALSE(it.completed());
// The camera is re-offered once per call, not skipped
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.camera_calls, 2);
EXPECT_FALSE(it.completed());
// Once accepted, the iteration completes
it.try_advance(BIG_BUDGET);
EXPECT_TRUE(it.completed());
EXPECT_EQ(it.camera_calls, 3);
}
} // namespace esphome::testing
#endif // USE_CAMERA
+11
View File
@@ -0,0 +1,11 @@
# Pulls in sensor so entity iteration paths compile (USE_SENSOR);
# tests register their own instances. Plain yaml.safe_load, no ESPHome tags.
# An alphabetically-earlier component's sensor: block shadows this one in
# combined builds; the tests' sensor-count ASSERT catches a capacity drop.
sensor:
- platform: template
id: bench_sensor_a
name: "Bench A"
- platform: template
id: bench_sensor_b
name: "Bench B"
@@ -0,0 +1,195 @@
#include <gtest/gtest.h>
#include "esphome/core/component_iterator.h"
#ifdef USE_SENSOR
#include "esphome/components/sensor/sensor.h"
#include "esphome/core/application.h"
#endif
namespace esphome::testing {
// Iterator whose begin/end callbacks can refuse a configurable number of
// times; all entity callbacks accept (any registered entities are accepted).
class RefusingIterator : public ComponentIterator {
public:
// NOLINTBEGIN(bugprone-macro-parentheses)
#define ENTITY_TYPE_(type, singular, plural, count, upper) \
bool on_##singular(type *obj) override { return true; }
#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
ENTITY_TYPE_(type, singular, plural, count, upper)
#include "esphome/core/entity_types.h"
#undef ENTITY_TYPE_
#undef ENTITY_CONTROLLER_TYPE_
// NOLINTEND(bugprone-macro-parentheses)
bool on_begin() override { return step(this->begin_calls, this->begin_refusals); }
bool on_end() override { return step(this->end_calls, this->end_refusals); }
int begin_calls{0};
int end_calls{0};
int begin_refusals{0};
int end_refusals{0};
protected:
static bool step(int &calls, int &refusals) {
calls++;
if (refusals > 0) {
refusals--;
return false;
}
return true;
}
};
// Far above the fixed number of iterator states
static constexpr size_t BIG_BUDGET = 1000;
TEST(ComponentIterator, NotRunningMakesNoProgress) {
RefusingIterator it;
it.try_advance(BIG_BUDGET);
EXPECT_TRUE(it.completed());
EXPECT_EQ(it.begin_calls, 0);
EXPECT_EQ(it.end_calls, 0);
}
TEST(ComponentIterator, CompletesInOneCallWithoutRefusals) {
RefusingIterator it;
it.begin();
it.try_advance(BIG_BUDGET);
EXPECT_TRUE(it.completed());
EXPECT_EQ(it.begin_calls, 1);
EXPECT_EQ(it.end_calls, 1);
}
TEST(ComponentIterator, StepBudgetIsHonored) {
RefusingIterator it;
it.begin();
it.try_advance(1);
EXPECT_EQ(it.begin_calls, 1);
EXPECT_EQ(it.end_calls, 0);
EXPECT_FALSE(it.completed());
}
TEST(ComponentIterator, RefusedStepStopsBatchAndRetriesSameStep) {
RefusingIterator it;
it.end_refusals = 3;
it.begin();
// First call runs until the refused end step, which stops the pass
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.end_calls, 1);
EXPECT_FALSE(it.completed());
// The refused step is retried once per call, not skipped
it.try_advance(BIG_BUDGET);
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.end_calls, 3);
EXPECT_FALSE(it.completed());
// Once accepted, the iteration completes
it.try_advance(BIG_BUDGET);
EXPECT_TRUE(it.completed());
EXPECT_EQ(it.end_calls, 4);
}
TEST(ComponentIterator, RefusedBeginStopsBatchAndRetries) {
RefusingIterator it;
it.begin_refusals = 2;
it.begin();
it.try_advance(BIG_BUDGET);
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.begin_calls, 2);
EXPECT_FALSE(it.completed());
it.try_advance(BIG_BUDGET);
EXPECT_TRUE(it.completed());
EXPECT_EQ(it.begin_calls, 3);
}
// The deprecated advance() wrapper must keep the legacy once-per-loop
// pattern working during the deprecation window.
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
TEST(ComponentIterator, DeprecatedAdvanceKeepsLegacyPatternWorking) {
RefusingIterator it;
it.end_refusals = 2;
it.begin();
size_t guard = 0;
while (!it.completed() && guard++ < BIG_BUDGET) {
it.advance();
}
EXPECT_TRUE(it.completed());
// Two refused end steps were retried, then accepted
EXPECT_EQ(it.end_calls, 3);
}
#pragma GCC diagnostic pop
#ifdef USE_SENSOR
// Iterator whose sensor callback can refuse or yield; pins the per-item
// contract: a refused item is re-offered with at_ unchanged, never skipped.
class ItemRefusingIterator : public RefusingIterator {
public:
bool on_sensor(sensor::Sensor *obj) override {
this->last_sensor = obj;
if (!step(this->sensor_calls, this->sensor_refusals))
return false;
if (this->yield_on_sensor)
this->yield_after_step_();
return true;
}
sensor::Sensor *last_sensor{nullptr};
int sensor_calls{0};
int sensor_refusals{0};
bool yield_on_sensor{false};
};
class ComponentIteratorSensorTest : public ::testing::Test {
protected:
void SetUp() override {
static sensor::Sensor sensor_a;
static sensor::Sensor sensor_b;
static bool registered = false;
if (!registered) {
App.register_sensor(&sensor_a);
App.register_sensor(&sensor_b);
registered = true;
}
// StaticVector drops silently when full; fail the fixture, not the contract
ASSERT_EQ(App.get_sensors().size(), 2u) << "benchmark.yaml sensor count too small";
}
};
TEST_F(ComponentIteratorSensorTest, RefusedItemIsReofferedNotSkipped) {
ItemRefusingIterator it;
it.sensor_refusals = 2;
it.begin();
// Runs until the first sensor refuses
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.sensor_calls, 1);
EXPECT_FALSE(it.completed());
// The refused item is re-offered, not skipped
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.sensor_calls, 2);
sensor::Sensor *refused = it.last_sensor;
// Once accepted, iteration continues through the second sensor to the end
it.try_advance(BIG_BUDGET);
EXPECT_TRUE(it.completed());
EXPECT_NE(it.last_sensor, refused);
EXPECT_EQ(it.sensor_calls, 4);
}
TEST_F(ComponentIteratorSensorTest, YieldAfterStepEndsPassAndResumes) {
ItemRefusingIterator it;
it.yield_on_sensor = true;
it.begin();
// The pass ends right after the first sensor despite a big budget
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.sensor_calls, 1);
EXPECT_FALSE(it.completed());
// The next pass ends after the second sensor
it.try_advance(BIG_BUDGET);
EXPECT_EQ(it.sensor_calls, 2);
// Remaining states then run to completion in one pass
it.try_advance(BIG_BUDGET);
EXPECT_TRUE(it.completed());
}
#endif // USE_SENSOR
} // namespace esphome::testing
+2
View File
@@ -7,6 +7,7 @@ This directory contains end-to-end integration tests for ESPHome, focusing on te
- `conftest.py` - Common fixtures and utilities
- `const.py` - Constants used throughout the integration tests
- `types.py` - Type definitions for fixtures and functions
- `raw_api_client.py` - Minimal plaintext api client whose reads happen only on request (for backpressure tests)
- `state_utils.py` - State handling utilities (e.g., `InitialStateHelper`, `find_entity`, `require_entity`)
- `fixtures/` - YAML configuration files for tests
- `test_*.py` - Individual test files
@@ -347,6 +348,7 @@ Create C++ components in `fixtures/external_components/` for:
- Custom entity behaviors
- Scheduler testing
- Memory management tests
- Deterministic network backpressure (`sndbuf_pin_component` pins socket send buffers; assert on its log line to prove the pin took effect)
##### Log Line Monitoring
```python
@@ -0,0 +1,23 @@
esphome:
name: api-backpressure-test
host:
api:
# Smallest queue so a non-draining client blocks the send path quickly
max_send_queue: 1
actions:
# GENERATED_ACTIONS
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
components: [sndbuf_pin_component]
# Pins the device's socket send buffers for deterministic TCP backpressure
sndbuf_pin_component:
buffer_size: SERVER_SNDBUF
logger:
level: DEBUG
@@ -0,0 +1,20 @@
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.const import CONF_BUFFER_SIZE, CONF_ID
DEPENDENCIES = ["api"]
sndbuf_pin_ns = cg.esphome_ns.namespace("sndbuf_pin")
SndbufPinComponent = sndbuf_pin_ns.class_("SndbufPinComponent", cg.Component)
CONFIG_SCHEMA = cv.Schema(
{
cv.GenerateID(): cv.declare_id(SndbufPinComponent),
cv.Required(CONF_BUFFER_SIZE): cv.int_range(min=1),
}
).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
var = cg.new_Pvariable(config[CONF_ID], config[CONF_BUFFER_SIZE])
await cg.register_component(var, config)
@@ -0,0 +1,55 @@
#include "sndbuf_pin_component.h"
#include <netinet/in.h>
#include <sys/socket.h>
#include <cerrno>
#include "esphome/components/api/api_server.h"
#include "esphome/core/log.h"
namespace esphome::sndbuf_pin {
static const char *const TAG = "sndbuf_pin";
// Skip stdio; scan the low fd range where the listeners land
static constexpr int FIRST_USER_FD = 3;
static constexpr int MAX_FD_SCAN = 128;
void SndbufPinComponent::setup() {
int pinned = 0;
for (int fd = FIRST_USER_FD; fd < MAX_FD_SCAN; fd++) {
int type = 0;
socklen_t len = sizeof(type);
if (::getsockopt(fd, SOL_SOCKET, SO_TYPE, &type, &len) != 0 || type != SOCK_STREAM)
continue;
struct sockaddr_in addr {};
socklen_t addr_len = sizeof(addr);
if (::getsockname(fd, reinterpret_cast<struct sockaddr *>(&addr), &addr_len) != 0) {
ESP_LOGW(TAG, "fd %d: getsockname failed, errno %d", fd, errno);
continue;
}
if (ntohs(addr.sin_port) != api::global_api_server->get_port())
continue;
if (::setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &this->buffer_size_, sizeof(this->buffer_size_)) != 0) {
ESP_LOGW(TAG, "fd %d: SO_SNDBUF pin failed, errno %d", fd, errno);
continue;
}
int applied = 0;
len = sizeof(applied);
if (::getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &applied, &len) != 0 || applied < this->buffer_size_) {
// Linux doubles the requested value; anything below it means clamped
ESP_LOGW(TAG, "fd %d: SO_SNDBUF readback %d below requested %d", fd, applied, this->buffer_size_);
continue;
}
// Tests assert on this line; accepted sockets inherit the pinned size
ESP_LOGD(TAG, "fd %d port %d: SO_SNDBUF pinned to %d (effective %d)", fd, ntohs(addr.sin_port), this->buffer_size_,
applied);
pinned++;
}
if (pinned == 0) {
ESP_LOGE(TAG, "api listener socket was not pinned");
this->mark_failed();
}
}
} // namespace esphome::sndbuf_pin
@@ -0,0 +1,21 @@
#pragma once
#include "esphome/core/component.h"
namespace esphome::sndbuf_pin {
// Test-only (host): pins SO_SNDBUF on every open TCP socket so integration
// tests get deterministic backpressure; an explicit SO_SNDBUF also disables
// kernel autotuning, and accepted sockets inherit it from the listener.
class SndbufPinComponent : public Component {
public:
explicit SndbufPinComponent(int buffer_size) : buffer_size_(buffer_size) {}
void setup() override;
// After the api server so its listening socket exists
float get_setup_priority() const override { return setup_priority::LATE; }
protected:
int buffer_size_;
};
} // namespace esphome::sndbuf_pin
+148
View File
@@ -0,0 +1,148 @@
"""Minimal plaintext native-api client over a raw socket.
Reads only when told to, so tests control when the TCP pipe backs up toward
the device; payloads are skipped and only message types are counted.
"""
from __future__ import annotations
import asyncio
from collections import Counter
import socket
from typing import Self
from aioesphomeapi import api_pb2
import aioesphomeapi.core as api_core
from google.protobuf import message
from .const import LOCALHOST
# Message type ids are protocol constants; derive them from aioesphomeapi so
# they cannot drift from the client library in use.
MESSAGE_TYPE_OF = {cls: num for num, cls in api_core.MESSAGE_TYPE_TO_PROTO.items()}
_READ_CHUNK = 4096
def encode_varint(value: int) -> bytes:
out = bytearray()
while True:
byte = value & 0x7F
value >>= 7
if value:
out.append(byte | 0x80)
else:
out.append(byte)
return bytes(out)
def decode_varint(buf: bytearray, pos: int) -> tuple[int, int] | None:
"""Decode one varint at pos; return (value, new_pos) or None if short."""
value = shift = 0
while pos < len(buf):
byte = buf[pos]
pos += 1
value |= (byte & 0x7F) << shift
if not byte & 0x80:
return value, pos
shift += 7
return None
def encode_frame(msg_type: int, payload: bytes) -> bytes:
"""Encode one plaintext api frame: 0x00, payload length, message type."""
return b"\x00" + encode_varint(len(payload)) + encode_varint(msg_type) + payload
class FrameParser:
"""Incremental parser for the plaintext api frame stream."""
def __init__(self) -> None:
self._buf = bytearray()
def feed(self, data: bytes) -> list[int]:
self._buf.extend(data)
types: list[int] = []
while (msg_type := self._try_parse()) is not None:
types.append(msg_type)
return types
def _try_parse(self) -> int | None:
buf = self._buf
if not buf:
return None
assert buf[0] == 0, f"expected plaintext frame, got indicator {buf[0]}"
if (size_decoded := decode_varint(buf, 1)) is None:
return None
size, pos = size_decoded
if (type_decoded := decode_varint(buf, pos)) is None:
return None
msg_type, pos = type_decoded
if len(buf) - pos < size:
return None
del buf[: pos + size]
return msg_type
class RawApiClient:
"""Plaintext api client whose reads happen only on request."""
def __init__(self, port: int, recv_buffer_size: int | None = None) -> None:
self._port = port
self._parser = FrameParser()
self.bytes_received = 0
self.frame_counts: Counter[int] = Counter()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
if recv_buffer_size is not None:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, recv_buffer_size)
# Kernels may round up (Linux doubles) but must not clamp below
applied = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
assert applied >= recv_buffer_size, (
f"SO_RCVBUF clamped to {applied}, requested {recv_buffer_size}"
)
sock.setblocking(False)
except Exception:
sock.close()
raise
self._sock = sock
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, *exc_info: object) -> None:
self.close()
async def connect(self, client_info: str = "raw-api-client") -> None:
"""Connect and complete the Hello handshake (no auth step since 2026.1.0)."""
loop = asyncio.get_running_loop()
await loop.sock_connect(self._sock, (LOCALHOST, self._port))
hello = api_pb2.HelloRequest()
hello.client_info = client_info
hello.api_version_major = 1
hello.api_version_minor = 10
await self.send_message(hello)
await self.read_until_frame(MESSAGE_TYPE_OF[api_pb2.HelloResponse])
async def send_message(self, msg: message.Message) -> None:
loop = asyncio.get_running_loop()
await loop.sock_sendall(
self._sock,
encode_frame(MESSAGE_TYPE_OF[type(msg)], msg.SerializeToString()),
)
async def read_until_frame(self, msg_type: int, timeout: float = 10.0) -> None:
"""Read until at least one frame of msg_type has been received."""
loop = asyncio.get_running_loop()
async def _read_loop() -> None:
while not self.frame_counts[msg_type]:
data = await loop.sock_recv(self._sock, _READ_CHUNK)
assert data, "server closed the connection unexpectedly"
self.bytes_received += len(data)
self.frame_counts.update(self._parser.feed(data))
await asyncio.wait_for(_read_loop(), timeout)
def close(self) -> None:
self._sock.close()
@@ -0,0 +1,110 @@
"""A client that stops reading the entity listing must not starve other clients.
Service responses are sent directly (not via the deferred batch), so a full
TCP pipe makes the send path refuse; the drive loop now lives in
try_advance(), which stops on refusal instead of retrying forever. Not a
before/after regression test: pre-fix builds survive here because the
refusal path yields and pumps the socket each retry.
The sndbuf_pin_component fixture pins the device's send buffers so the pipe
fills deterministically regardless of kernel autotuning; the test waits for
its log line before proceeding.
"""
from __future__ import annotations
import asyncio
from aioesphomeapi import api_pb2
import pytest
from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient
from .types import APIClientConnectedFactory, RunCompiledFunction
SERVICES_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesServicesResponse]
LIST_DONE_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesDoneResponse]
# Both ends of the pipe are pinned small; only tens of KB fit in the kernel
RECV_BUFFER_SIZE = 4096
SERVER_SNDBUF = 8192 # substituted into the fixture yaml
# Logged by the sndbuf_pin_component fixture when it pins a socket
SNDBUF_PIN_LOG = "SO_SNDBUF pinned to"
# One response (~6.4 KB) must stay smaller than the pinned send buffer; an
# oversized message parks in the overflow buffer and reports as sent.
ARGS_PER_SERVICE = 8
ARG_NAME_LEN = 800
# ~160 KB listing versus a tens-of-KB pipe guarantees a mid-services block
NUM_SERVICES = 25
assert ARGS_PER_SERVICE * ARG_NAME_LEN < SERVER_SNDBUF
# The pipe fills in well under a second
STALL_SECONDS = 0.5
# Well above pipe capacity, well below the listing size
MIN_DRAINED_BYTES = 60_000
def _generated_actions() -> str:
"""Build the api actions block: services with long argument names."""
lines: list[str] = []
for i in range(NUM_SERVICES):
lines.append(f" - action: backpressure_service_{i:04d}")
lines.append(" variables:")
for j in range(ARGS_PER_SERVICE):
prefix = f"arg_{i:04d}_{j:02d}_"
lines.append(
f" {prefix}{'x' * (ARG_NAME_LEN - len(prefix))}: string"
)
lines.append(" then:")
lines.append(" - logger.log: service called")
return "\n".join(lines)
@pytest.mark.asyncio
async def test_api_list_entities_backpressure(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
unused_tcp_port: int,
) -> None:
"""A stalled reader mid-services must not block other api clients."""
assert "# GENERATED_ACTIONS" in yaml_config
config = yaml_config.replace("# GENERATED_ACTIONS", _generated_actions())
config = config.replace("SERVER_SNDBUF", str(SERVER_SNDBUF))
pin_applied = asyncio.Event()
def _on_log_line(line: str) -> None:
if SNDBUF_PIN_LOG in line:
pin_applied.set()
async with run_compiled(config, line_callback=_on_log_line):
# Fails loudly if the pin never applied
await asyncio.wait_for(pin_applied.wait(), 10)
async with RawApiClient(
unused_tcp_port, recv_buffer_size=RECV_BUFFER_SIZE
) as stalled:
await stalled.connect(client_info="backpressure-stall-client")
await stalled.send_message(api_pb2.ListEntitiesRequest())
# The client now stops reading entirely.
# Let the server run against the full pipe
await asyncio.sleep(STALL_SECONDS)
# Other clients must still be served while the first is blocked
async with api_client_connected(timeout=20) as client:
device_info = await asyncio.wait_for(client.device_info(), 20)
assert device_info.name == "api-backpressure-test"
_, services = await asyncio.wait_for(
client.list_entities_services(), 30
)
assert len(services) == NUM_SERVICES
# Fixture-size guard: the listing must dwarf the pinned pipe
before = stalled.bytes_received
await stalled.read_until_frame(LIST_DONE_RESPONSE, timeout=60)
drained = stalled.bytes_received - before
assert drained > MIN_DRAINED_BYTES, (
f"only {drained} bytes drained; the listing never backed up"
)
assert stalled.frame_counts[SERVICES_RESPONSE] == NUM_SERVICES
assert stalled.frame_counts[LIST_DONE_RESPONSE] == 1