diff --git a/esphome/components/packet_interface/__init__.py b/esphome/components/packet_interface/__init__.py index 4685ab45713..c663d5b84ff 100644 --- a/esphome/components/packet_interface/__init__.py +++ b/esphome/components/packet_interface/__init__.py @@ -1,3 +1,5 @@ +import importlib +import pkgutil from typing import Any import esphome.codegen as cg @@ -7,14 +9,20 @@ from esphome.const import CONF_ID from esphome.cpp_generator import MockObj, MockObjClass from esphome.types import ConfigType +from . import layers +from .layers import PacketInterface, build_stack, packet_interface_ns, stack_schema + CODEOWNERS = ["@clydebarrow"] DOMAIN = "packet_interface" IS_PLATFORM_COMPONENT = True CONF_PACKET_INTERFACE = "packet_interface" -packet_interface_ns = cg.esphome_ns.namespace("packet_interface") -PacketInterface = packet_interface_ns.class_("PacketInterface", cg.Component) +# Each module in layers registers a layer type; import them all before building the schemas from the registry +for module_info in pkgutil.iter_modules(layers.__path__): + importlib.import_module(f".layers.{module_info.name}", package=__package__) + +LINK_LAYER_SCHEMAS = stack_schema() __all__ = [ "CONF_MAX_PACKET_SIZE", @@ -31,19 +39,20 @@ def packet_interface_schema( default_max_packet_size: int, max_packet_size_limit: int = 4096, ) -> cv.Schema: - """Schema for a link platform.""" + """Schema for a link platform, with the layers that can be stacked on it.""" return cv.Schema( { cv.GenerateID(): cv.declare_id(class_), cv.Optional( CONF_MAX_PACKET_SIZE, default=default_max_packet_size ): cv.int_range(16, max_packet_size_limit), + **LINK_LAYER_SCHEMAS, } ).extend(cv.COMPONENT_SCHEMA) async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj: - """Create the link layer. + """Create the link layer and the layers configured on it, from the bottom up. The link's constructor takes its maximum packet size, usually as a std::span over a buffer that code generation sized from the configuration, so the two cannot differ. @@ -53,4 +62,5 @@ async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj: var = cg.new_Pvariable(config[CONF_ID], *args) cg.add(var.set_name(config[CONF_ID].id)) await cg.register_component(var, config) + await build_stack(config, var) return var diff --git a/esphome/components/packet_interface/hdlc_reliability.cpp b/esphome/components/packet_interface/hdlc_reliability.cpp new file mode 100644 index 00000000000..cad0ba25b09 --- /dev/null +++ b/esphome/components/packet_interface/hdlc_reliability.cpp @@ -0,0 +1,320 @@ +#include "hdlc_reliability.h" +#ifdef USE_PACKET_INTERFACE_HDLC +#include "esphome/core/helpers.h" +#include "esphome/core/log.h" + +namespace esphome::packet_interface { + +ESPHOME_LOG_TAG(TAG, "packet_interface.hdlc"); + +static constexpr uint32_t RETRANSMIT_TIMER_ID = 0; +static constexpr uint32_t ACK_TIMER_ID = 1; +static constexpr uint32_t SEND_RETRY_TIMER_ID = 2; +// A frame or acknowledgement the lower layer refused, for example while a radio was receiving, is sent again after this +static constexpr uint32_t SEND_RETRY_MS = 50; +// SABM is repeated at this interval, or the timeout if longer, until the peer answers +static constexpr uint32_t RESET_INTERVAL_MS = 1000; + +// Control byte formats, with the poll/final bit (0x10) ignored +static constexpr uint8_t POLL_FINAL_BIT = 0x10; +static constexpr uint8_t SUPERVISORY_FRAME = 0x01; +static constexpr uint8_t SUPERVISORY_RR = 0x00; +static constexpr uint8_t SUPERVISORY_REJ = 0x02; +static constexpr uint8_t UNNUMBERED_SABM = 0x2F; +static constexpr uint8_t UNNUMBERED_UA = 0x63; + +static uint16_t frame_crc(uint8_t control, const uint8_t *payload, size_t length) { + uint16_t crc = crc16(&control, 1); + return crc16(payload, length, crc); +} + +HdlcReliability::HdlcReliability(PacketInterface *lower) : lower_(lower) { + this->stack_on_(lower, CAPABILITY_RELIABILITY); + this->max_packet_size_ = lower->get_max_packet_size() - FRAME_OVERHEAD; +} + +void HdlcReliability::setup() { + this->tx_slots_ = std::make_unique(this->max_queue_length_ * this->max_packet_size_); + this->lower_->add_packet_interface_listener( + [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); }); +} + +void HdlcReliability::dump_config() { + ESP_LOGCONFIG(TAG, + "HDLC Reliability on '%s':\n" + " Max Packet Size: %zu\n" + " Timeout: %" PRIu32 "ms\n" + " Ack Delay: %" PRIu32 "ms\n" + " Max Queue Length: %u\n" + " Max Retries: %u", + this->name_, this->max_packet_size_, this->timeout_ms_, this->ack_delay_ms_, this->max_queue_length_, + this->max_retries_); +} + +void HdlcReliability::loop() { + // Disabled first, as sending may deliver a reply to this layer and request the loop again + this->disable_loop(); + if (this->state_ == LinkState::LINK_STATE_RESET_PENDING) { + this->state_ = LinkState::LINK_STATE_RESETTING; + this->start_timer_(); + this->send_frame_(UNNUMBERED_SABM, nullptr, 0); + } + if (this->state_ != LinkState::LINK_STATE_CONNECTED) + return; + // Queued frames go first, as each carries the acknowledgement + if (this->send_pending_() && this->ack_due_ && !this->send_supervisory_(SUPERVISORY_RR)) + this->retry_send_(); +} + +void HdlcReliability::retry_send_() { + this->set_timeout(SEND_RETRY_TIMER_ID, SEND_RETRY_MS, [this]() { this->request_loop_(); }); +} + +void HdlcReliability::owe_ack_(bool now) { + if (now || this->ack_delay_ms_ == 0) { + this->ack_due_ = true; + this->request_loop_(); + } else if (!this->ack_pending_) { + // Waiting lets one acknowledgement cover several frames, or a reply carry it + this->set_timeout(ACK_TIMER_ID, this->ack_delay_ms_, [this]() { + this->ack_due_ = true; + this->request_loop_(); + }); + } + this->ack_pending_ = true; +} + +void HdlcReliability::ack_sent_() { + this->ack_pending_ = false; + this->ack_due_ = false; + this->cancel_timeout(ACK_TIMER_ID); +} + +void HdlcReliability::request_loop_() { this->enable_loop(); } + +bool HdlcReliability::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) { + size_t size = data.size(); + if (size > this->max_packet_size_) { + ESP_LOGW(TAG, "%s: Packet of %zu bytes is larger than %zu", this->name_, size, this->max_packet_size_); + return false; + } + uint8_t outstanding = this->outstanding_(); + if (outstanding >= this->max_queue_length_) { + ESP_LOGV(TAG, "%s: Send queue full", this->name_); + return false; + } + uint8_t seq = this->vs_; + data.copy_to(this->slot_(seq), size); + this->tx_lengths_[this->slot_index_(seq)] = size; + this->vs_ = (seq + 1) & 7; + // While the link is being reset, the frame waits in the queue and is sent when UA arrives + if (this->state_ == LinkState::LINK_STATE_CONNECTED) { + this->send_pending_(); + if (outstanding == 0) + this->start_timer_(); + } + return true; +} + +bool HdlcReliability::send_frame_(uint8_t control, const uint8_t *payload, size_t length) { + uint16_t crc = frame_crc(control, payload, length); + uint8_t trailer[2] = {static_cast(crc), static_cast(crc >> 8)}; + PacketBuffer body(payload, length); + PacketBuffer frame(&control, 1, length != 0 ? &body : nullptr, trailer, sizeof(trailer)); + if (this->lower_->send_to_interface(frame, {})) + return true; + ESP_LOGV(TAG, "%s: Lower layer did not accept frame 0x%02X", this->name_, control); + return false; +} + +bool HdlcReliability::send_supervisory_(uint8_t type) { + if (!this->send_frame_(SUPERVISORY_FRAME | (type << 2) | (this->vr_ << 5), nullptr, 0)) + return false; + this->ack_sent_(); + return true; +} + +// Sends the next queued frame not yet sent, and returns true if none are left. One frame per loop pass lets other +// components run between frames: on a radio, sending blocks, and the peer's frames are received in between. If the +// lower layer refuses the frame, it is sent again shortly. +bool HdlcReliability::send_pending_() { + if (this->vn_ == this->vs_) + return true; + uint8_t control = (this->vn_ << 1) | (this->vr_ << 5); + if (!this->send_frame_(control, this->slot_(this->vn_), this->tx_lengths_[this->slot_index_(this->vn_)])) { + this->retry_send_(); + return false; + } + // Every I-frame carries an acknowledgement, so a pending RR is no longer needed + this->ack_sent_(); + this->vn_ = (this->vn_ + 1) & 7; + if (this->vn_ == this->vs_) + return true; + this->request_loop_(); + return false; +} + +void HdlcReliability::resend_outstanding_() { + this->vn_ = this->va_; + this->send_pending_(); +} + +void HdlcReliability::start_timer_() { + uint32_t interval = this->timeout_ms_; + if (this->state_ != LinkState::LINK_STATE_CONNECTED && interval < RESET_INTERVAL_MS) + interval = RESET_INTERVAL_MS; + // Up to a quarter longer at random, so devices whose frames collided on a shared channel do not resend together + interval += random_uint32() % (interval / 4 + 1); + this->set_timeout(RETRANSMIT_TIMER_ID, interval, [this]() { this->on_timeout_(); }); +} + +void HdlcReliability::on_timeout_() { + if (this->state_ != LinkState::LINK_STATE_CONNECTED) { + this->send_frame_(UNNUMBERED_SABM, nullptr, 0); + this->start_timer_(); + return; + } + if (this->outstanding_() == 0) + return; + if (++this->retries_ > this->max_retries_) { + ESP_LOGW(TAG, "%s: No acknowledgement after %u retries, resetting link", this->name_, this->max_retries_); + this->start_reset_(); + return; + } + ESP_LOGV(TAG, "%s: Timeout, resending %u frames", this->name_, this->outstanding_()); + this->resend_outstanding_(); + this->start_timer_(); +} + +// Unacknowledged frames are discarded; frames queued from now on are numbered from 0 and sent once the link is up. +// Resetting a connected link can lose data both ways, which is reported to the layers above. +void HdlcReliability::start_reset_() { + bool was_connected = this->state_ == LinkState::LINK_STATE_CONNECTED; + this->state_ = LinkState::LINK_STATE_RESET_PENDING; + this->va_ = this->vs_ = this->vn_ = 0; + this->head_ = 0; + this->vr_ = 0; + this->retries_ = 0; + this->rej_sent_ = false; + this->ack_sent_(); + this->cancel_timeout(RETRANSMIT_TIMER_ID); + this->request_loop_(); + if (was_connected) + this->session_callback_.call(false); +} + +void HdlcReliability::link_up_() { + ESP_LOGD(TAG, "%s: Link connected", this->name_); + this->state_ = LinkState::LINK_STATE_CONNECTED; + this->retries_ = 0; + this->cancel_timeout(RETRANSMIT_TIMER_ID); + if (this->outstanding_() != 0) { + this->resend_outstanding_(); + this->start_timer_(); + } +} + +void HdlcReliability::process_ack_(uint8_t nr) { + uint8_t acked = (nr - this->va_) & 7; + if (acked > this->outstanding_()) { + ESP_LOGV(TAG, "%s: Ignoring invalid N(R) %u", this->name_, nr); + return; + } + if (acked == 0) + return; + // The peer may have had frames that were refused when last resent + if (acked > ((this->vn_ - this->va_) & 7)) + this->vn_ = nr; + this->head_ = (this->head_ + acked) % this->max_queue_length_; + this->va_ = nr; + this->retries_ = 0; + if (this->outstanding_() == 0) { + this->cancel_timeout(RETRANSMIT_TIMER_ID); + } else { + this->start_timer_(); + } +} + +void HdlcReliability::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) { + // Lower layers deliver packets in one piece; the payload is passed up without copying + if (!data.is_contiguous()) { + ESP_LOGW(TAG, "%s: Packet split across buffers is not supported", this->name_); + return; + } + auto frame = data.as_span(); + if (frame.size() < FRAME_OVERHEAD) { + this->note_received_(false); + return; + } + size_t length = frame.size() - FRAME_OVERHEAD; + uint8_t control = frame[0]; + const uint8_t *payload = frame.data() + 1; + uint16_t crc = frame[length + 1] | (frame[length + 2] << 8); + if (frame_crc(control, payload, length) != crc) { + ESP_LOGV(TAG, "%s: CRC error, frame dropped", this->name_); + this->note_received_(false); + return; + } + this->note_received_(true); + + uint8_t command = control & ~POLL_FINAL_BIT; + if (command == UNNUMBERED_SABM) { + if (this->state_ == LinkState::LINK_STATE_CONNECTED) { + // The peer has restarted, so frames it has not acknowledged are lost + ESP_LOGD(TAG, "%s: Link reset by peer", this->name_); + this->start_reset_(); + } + this->vr_ = 0; + this->rej_sent_ = false; + this->send_frame_(UNNUMBERED_UA, nullptr, 0); + // Both sides now start from 0, so the link is up even if our own SABM is unanswered + this->link_up_(); + return; + } + if (command == UNNUMBERED_UA) { + if (this->state_ == LinkState::LINK_STATE_RESETTING) + this->link_up_(); + return; + } + // Numbered frames are ignored until the link is reset; the peer answers our SABM first + if (this->state_ != LinkState::LINK_STATE_CONNECTED) + return; + + uint8_t nr = control >> 5; + if ((control & 1) == 0) { + this->process_ack_(nr); + this->receive_info_(control, payload, length, meta_data); + return; + } + if ((control & 3) != SUPERVISORY_FRAME) + return; + this->process_ack_(nr); + if (((control >> 2) & 3) == SUPERVISORY_REJ && this->outstanding_() != 0) { + ESP_LOGV(TAG, "%s: Peer rejected from %u", this->name_, nr); + this->resend_outstanding_(); + this->start_timer_(); + } +} + +void HdlcReliability::receive_info_(uint8_t control, const uint8_t *payload, size_t length, + const PacketMetaData &meta_data) { + uint8_t ns = (control >> 1) & 7; + if (ns != this->vr_) { + // A gap or a duplicate; ask once for a resend from the expected frame, then just acknowledge at once + if (!this->rej_sent_) { + this->rej_sent_ = this->send_supervisory_(SUPERVISORY_REJ); + if (this->rej_sent_) + return; + } + this->owe_ack_(true); + return; + } + this->vr_ = (this->vr_ + 1) & 7; + this->rej_sent_ = false; + // The acknowledgement is sent from loop(), so a reply sent from the callback can carry it instead + this->owe_ack_(false); + this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data); +} + +} // namespace esphome::packet_interface +#endif diff --git a/esphome/components/packet_interface/hdlc_reliability.h b/esphome/components/packet_interface/hdlc_reliability.h new file mode 100644 index 00000000000..07f741f403e --- /dev/null +++ b/esphome/components/packet_interface/hdlc_reliability.h @@ -0,0 +1,93 @@ +#pragma once + +#include "esphome/core/defines.h" +#ifdef USE_PACKET_INTERFACE_HDLC +#include +#include + +#include "packet_interface.h" + +namespace esphome::packet_interface { + +/** + * A reliability layer that delivers packets in order, without loss or duplicates, between two peers. + * + * It follows HDLC asynchronous balanced mode, with sequence numbers modulo 8 and go-back-N retransmission: + * - I-frames carry a payload, the sequence number N(S) and an acknowledgement N(R) of the peer's frames. + * - RR frames acknowledge without a payload; REJ frames ask the peer to resend from N(R). + * - SABM resets both peers' sequence numbers, and UA answers it. + * + * Each frame is the HDLC control byte, the payload, then a CRC-16 of both in little endian order. + * Unacknowledged frames are resent after the timeout; after max_retries the link is reset. + */ +class HdlcReliability : public PacketInterface { + public: + /// Most frames that can be unacknowledged at once, set by the 3 bit sequence numbers. + static constexpr uint8_t MAX_WINDOW = 7; + /// Control byte plus CRC-16. + static constexpr size_t FRAME_OVERHEAD = 3; + + explicit HdlcReliability(PacketInterface *lower); + + void setup() override; + void loop() override; + void dump_config() override; + bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override; + size_t get_send_capacity() const override { return this->max_queue_length_ - this->outstanding_(); } + + void set_ack_timeout(uint32_t timeout_ms) { this->timeout_ms_ = timeout_ms; } + void set_ack_delay(uint32_t ack_delay_ms) { this->ack_delay_ms_ = ack_delay_ms; } + void set_max_queue_length(uint8_t length) { this->max_queue_length_ = length; } + void set_max_retries(uint8_t retries) { this->max_retries_ = retries; } + + bool is_connected() const { return this->state_ == LinkState::LINK_STATE_CONNECTED; } + + protected: + enum class LinkState : uint8_t { + LINK_STATE_RESET_PENDING, // SABM to be sent from loop() + LINK_STATE_RESETTING, // SABM sent, waiting for UA + LINK_STATE_CONNECTED, + }; + + void receive_(const PacketBuffer &data, const PacketMetaData &meta_data); + void receive_info_(uint8_t control, const uint8_t *payload, size_t length, const PacketMetaData &meta_data); + bool send_frame_(uint8_t control, const uint8_t *payload, size_t length); + bool send_supervisory_(uint8_t type); + bool send_pending_(); + void resend_outstanding_(); + void retry_send_(); + /// An RR is owed: at once, or after the ack delay unless an I-frame carries it first. + void owe_ack_(bool now); + void ack_sent_(); + void process_ack_(uint8_t nr); + void on_timeout_(); + void start_reset_(); + void link_up_(); + void start_timer_(); + void request_loop_(); + uint8_t outstanding_() const { return (this->vs_ - this->va_) & 7; } + uint8_t slot_index_(uint8_t seq) const { return (this->head_ + ((seq - this->va_) & 7)) % this->max_queue_length_; } + uint8_t *slot_(uint8_t seq) { return this->tx_slots_.get() + this->slot_index_(seq) * this->max_packet_size_; } + + PacketInterface *lower_; + // max_queue_length_ slots of max_packet_size_ bytes, a ring starting at head_ which holds frame va_ + std::unique_ptr tx_slots_; + std::array tx_lengths_{}; // indexed like the slots + uint32_t timeout_ms_{200}; + uint32_t ack_delay_ms_{0}; + uint8_t max_queue_length_{4}; + uint8_t max_retries_{5}; + uint8_t retries_{0}; + uint8_t head_{0}; + uint8_t vs_{0}; // sequence number of the next new frame + uint8_t vn_{0}; // next frame to send; frames from va_ up to it have been sent + uint8_t va_{0}; // oldest unacknowledged frame + uint8_t vr_{0}; // next frame expected from the peer + LinkState state_{LinkState::LINK_STATE_RESET_PENDING}; + bool ack_pending_{false}; // an RR is owed + bool ack_due_{false}; // and should be sent now + bool rej_sent_{false}; +}; + +} // namespace esphome::packet_interface +#endif diff --git a/esphome/components/packet_interface/layers/__init__.py b/esphome/components/packet_interface/layers/__init__.py new file mode 100644 index 00000000000..4f48f3c2b20 --- /dev/null +++ b/esphome/components/packet_interface/layers/__init__.py @@ -0,0 +1,92 @@ +"""The kinds of layer that can be stacked on a packet interface, and the registry of their implementations. + +Each module in this package implements one layer type and registers it with its kind, for example: + + @RELIABILITY.register("hdlc", schema=HDLC_SCHEMA) + async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj: + ... + +The packet_interface component imports every module here before it builds its schemas, so a new layer type needs +only a new module. The kinds, and the order in which they are stacked, are fixed here. +""" + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Any + +import esphome.codegen as cg +import esphome.config_validation as cv +from esphome.const import CONF_TYPE +from esphome.cpp_generator import MockObj +from esphome.types import ConfigType + +packet_interface_ns = cg.esphome_ns.namespace("packet_interface") +PacketInterface = packet_interface_ns.class_("PacketInterface", cg.Component) + +CONF_RELIABILITY = "reliability" + +# Builds a layer above lower and returns it +LayerBuilder = Callable[[ConfigType, MockObj], Awaitable[Any]] + + +@dataclass(frozen=True) +class LayerType: + """One implementation of a kind of layer.""" + + schema: Any + build: LayerBuilder + + +class LayerKind: + """A kind of layer, such as reliability, with the types that implement it.""" + + def __init__(self, key: str, default_type: str) -> None: + self.key = key + self.default_type = default_type + self.types: dict[str, LayerType] = {} + + def register( + self, name: str, *, schema: Any + ) -> Callable[[LayerBuilder], LayerBuilder]: + """Register a type with its schema.""" + + def decorator(build: LayerBuilder) -> LayerBuilder: + self.types[name] = LayerType(schema, build) + return build + + return decorator + + @staticmethod + def _expand(value: Any) -> Any: + """Allow a layer to be enabled with no options, as in `reliability:`.""" + return {} if value is None else value + + def schema(self) -> Callable[[Any], ConfigType]: + """The schema for this kind, from the types registered so far.""" + return cv.All( + self._expand, + cv.typed_schema( + {name: layer.schema for name, layer in self.types.items()}, + default_type=self.default_type, + ), + ) + + def get_type(self, config: ConfigType) -> LayerType: + return self.types[config[CONF_TYPE]] + + +RELIABILITY = LayerKind(CONF_RELIABILITY, "hdlc") +# The kinds stacked above the link, from the bottom up +STACK_ORDER = (RELIABILITY,) + + +def stack_schema() -> dict: + """Schema entries for the layers of one stack; call only once all layer types are registered.""" + return {cv.Optional(kind.key): kind.schema() for kind in STACK_ORDER} + + +async def build_stack(config: ConfigType, lower: MockObj) -> None: + """Create the layers configured in config, from the bottom up, above lower.""" + for kind in STACK_ORDER: + if (layer := config.get(kind.key)) is not None: + lower = await kind.get_type(layer).build(layer, lower) diff --git a/esphome/components/packet_interface/layers/hdlc_reliability.py b/esphome/components/packet_interface/layers/hdlc_reliability.py new file mode 100644 index 00000000000..22dd942e3bf --- /dev/null +++ b/esphome/components/packet_interface/layers/hdlc_reliability.py @@ -0,0 +1,57 @@ +"""Reliability layer using HDLC style acknowledgement and go-back-N retransmission.""" + +import esphome.codegen as cg +import esphome.config_validation as cv +from esphome.const import CONF_ID, CONF_TIMEOUT +from esphome.cpp_generator import MockObj +from esphome.types import ConfigType + +from . import RELIABILITY, PacketInterface, packet_interface_ns + +CONF_ACK_DELAY = "ack_delay" +CONF_MAX_QUEUE_LENGTH = "max_queue_length" +CONF_MAX_RETRIES = "max_retries" + +HdlcReliability = packet_interface_ns.class_("HdlcReliability", PacketInterface) + + +def _ack_delay_within_timeout(config: ConfigType) -> ConfigType: + """The peer, configured the same, must hear the acknowledgement before its timeout.""" + if ( + config[CONF_ACK_DELAY].total_milliseconds + >= config[CONF_TIMEOUT].total_milliseconds + ): + raise cv.Invalid( + f"Must be shorter than '{CONF_TIMEOUT}'", path=[CONF_ACK_DELAY] + ) + return config + + +HDLC_SCHEMA = cv.All( + cv.Schema( + { + cv.GenerateID(): cv.declare_id(HdlcReliability), + cv.Optional( + CONF_TIMEOUT, default="200ms" + ): cv.positive_time_period_milliseconds, + cv.Optional( + CONF_ACK_DELAY, default="0ms" + ): cv.positive_time_period_milliseconds, + cv.Optional(CONF_MAX_QUEUE_LENGTH, default=4): cv.int_range(1, 7), + cv.Optional(CONF_MAX_RETRIES, default=5): cv.int_range(1, 255), + } + ).extend(cv.COMPONENT_SCHEMA), + _ack_delay_within_timeout, +) + + +@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA) +async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj: + cg.add_define("USE_PACKET_INTERFACE_HDLC") + var = cg.new_Pvariable(config[CONF_ID], lower) + cg.add(var.set_ack_timeout(config[CONF_TIMEOUT])) + cg.add(var.set_ack_delay(config[CONF_ACK_DELAY])) + cg.add(var.set_max_queue_length(config[CONF_MAX_QUEUE_LENGTH])) + cg.add(var.set_max_retries(config[CONF_MAX_RETRIES])) + await cg.register_component(var, config) + return var diff --git a/esphome/core/defines.h b/esphome/core/defines.h index 931ccf3c789..9e376277716 100644 --- a/esphome/core/defines.h +++ b/esphome/core/defines.h @@ -257,6 +257,7 @@ #define USE_OTA_STATE_LISTENER #define USE_OUTPUT #define USE_OUTPUT_FLOAT_POWER_SCALING +#define USE_PACKET_INTERFACE_HDLC #define USE_POWER_SUPPLY // Only defined by key-lookup preference backends; the slot-based platforms // (esp8266, rp2040) never set it in generated builds, and their preferences diff --git a/tests/components/packet_interface/__init__.py b/tests/components/packet_interface/__init__.py new file mode 100644 index 00000000000..a49be401949 --- /dev/null +++ b/tests/components/packet_interface/__init__.py @@ -0,0 +1,9 @@ +import esphome.codegen as cg +from tests.testing_helpers import ComponentManifestOverride + + +def override_manifest(manifest: ComponentManifestOverride) -> None: + async def to_code_testing(config: object) -> None: + cg.add_define("USE_PACKET_INTERFACE_HDLC") + + manifest.to_code = to_code_testing diff --git a/tests/components/packet_interface/common.yaml b/tests/components/packet_interface/common.yaml index d0049f8a9ef..836d6497028 100644 --- a/tests/components/packet_interface/common.yaml +++ b/tests/components/packet_interface/common.yaml @@ -9,9 +9,15 @@ packet_interface: - platform: uart id: packet_interface_uart max_packet_size: 256 + reliability: + timeout: 300ms + ack_delay: 50ms + max_queue_length: 3 + max_retries: 4 - platform: udp id: packet_interface_network udp_id: packet_interface_udp + reliability: sensor: - platform: template diff --git a/tests/components/packet_interface/hdlc_reliability.cpp b/tests/components/packet_interface/hdlc_reliability.cpp new file mode 100644 index 00000000000..0d7347c46c9 --- /dev/null +++ b/tests/components/packet_interface/hdlc_reliability.cpp @@ -0,0 +1,308 @@ +#include +#include +#include +#include +#include +#include "esphome/components/packet_interface/hdlc_reliability.h" +#include "esphome/core/application.h" +#include "esphome/core/helpers.h" + +namespace esphome::packet_interface::testing { + +using Frame = std::vector; + +// A link layer that records what is sent, so a test can deliver it to the peer +class MockLink : public PacketInterface { + public: + MockLink() { this->set_max_packet_size(64); } + bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override { + if (this->budget == 0) { + this->refused++; + return false; + } + this->budget--; + this->sent.push_back(data.to_vector()); + return true; + } + void inject(const Frame &frame) { this->on_receive_from_interface_(PacketBuffer(frame), {}); } + + std::vector sent; + size_t budget{SIZE_MAX}; // frames accepted before the link refuses, like a full network stack + size_t refused{0}; +}; + +class TestHdlc : public HdlcReliability { + public: + using HdlcReliability::HdlcReliability; + using HdlcReliability::on_timeout_; + using HdlcReliability::outstanding_; +}; + +struct Peer { + explicit Peer(uint8_t queue_length = 4) { + this->hdlc = std::make_unique(&this->link); + this->hdlc->set_max_queue_length(queue_length); + this->hdlc->set_max_retries(3); + this->hdlc->setup(); + this->hdlc->add_packet_interface_listener( + [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(data.to_vector()); }); + this->hdlc->add_on_session_state_callback([this](bool connected) { this->states.push_back(connected); }); + } + bool send(const Frame &data) { return this->hdlc->send_to_interface(PacketBuffer(data), {}); } + + MockLink link; + std::unique_ptr hdlc; + std::vector received; + std::vector states; +}; + +// Returns true to drop a frame on its way from one peer to the other +using DropFilter = std::function; + +static bool is_info(const Frame &frame) { return (frame[0] & 1) == 0; } +static bool is_ack(const Frame &frame) { return (frame[0] & 0x0F) == 0x01; } + +// Move frames between the peers and run their loops until nothing more is sent +static void pump(Peer &a, Peer &b, const DropFilter &drop_a_to_b = nullptr, const DropFilter &drop_b_to_a = nullptr) { + for (int i = 0; i != 50; i++) { + a.hdlc->loop(); + b.hdlc->loop(); + if (a.link.sent.empty() && b.link.sent.empty()) + return; + auto from_a = std::move(a.link.sent); + auto from_b = std::move(b.link.sent); + a.link.sent.clear(); + b.link.sent.clear(); + for (auto &frame : from_a) { + if (!drop_a_to_b || !drop_a_to_b(frame)) + b.link.inject(frame); + } + for (auto &frame : from_b) { + if (!drop_b_to_a || !drop_b_to_a(frame)) + a.link.inject(frame); + } + } + FAIL() << "Peers did not go quiet"; +} + +static Frame make_frame(uint8_t control, const Frame &payload) { + Frame frame{control}; + frame.insert(frame.end(), payload.begin(), payload.end()); + uint16_t crc = crc16(frame.data(), frame.size()); + frame.push_back(crc & 0xFF); + frame.push_back(crc >> 8); + return frame; +} + +// The test main does not construct App; the retransmit timer needs its scheduler +class HdlcReliabilityTest : public ::testing::Test { + protected: + void SetUp() override { + new (&App) Application; + App.pre_setup("hdlc-tests", 10, "", 0); + } + void TearDown() override { App.~Application(); } +}; + +TEST_F(HdlcReliabilityTest, MaxPacketSizeAllowsForOverhead) { + Peer a; + EXPECT_EQ(a.hdlc->get_max_packet_size(), 64 - HdlcReliability::FRAME_OVERHEAD); + EXPECT_EQ(a.link.top(), a.hdlc.get()); + EXPECT_EQ(a.hdlc->get_capabilities(), CAPABILITY_RELIABILITY); +} + +TEST_F(HdlcReliabilityTest, ResetHandshake) { + Peer a, b; + a.hdlc->loop(); + ASSERT_EQ(a.link.sent.size(), 1); + EXPECT_EQ(a.link.sent[0], make_frame(0x2F, {})); // SABM + pump(a, b); + EXPECT_TRUE(a.hdlc->is_connected()); + EXPECT_TRUE(b.hdlc->is_connected()); +} + +TEST_F(HdlcReliabilityTest, QueuedBeforeConnectIsSentAfter) { + Peer a, b; + ASSERT_TRUE(a.send({0x01})); + ASSERT_TRUE(a.send({0x02})); + pump(a, b); + EXPECT_EQ(b.received, (std::vector{{0x01}, {0x02}})); +} + +TEST_F(HdlcReliabilityTest, DeliversInOrderBothWays) { + Peer a, b; + pump(a, b); + a.send({0x10}); + b.send({0x20}); + a.send({0x11}); + pump(a, b); + EXPECT_EQ(b.received, (std::vector{{0x10}, {0x11}})); + EXPECT_EQ(a.received, (std::vector{{0x20}})); +} + +TEST_F(HdlcReliabilityTest, SequenceNumbersWrap) { + Peer a(3), b(3); + pump(a, b); + std::vector expected; + for (uint8_t i = 0; i != 20; i++) { + expected.push_back({i, static_cast(i * 3)}); + ASSERT_TRUE(a.send(expected.back())); + if (i % 2 == 1) + pump(a, b); + } + pump(a, b); + EXPECT_EQ(b.received, expected); +} + +TEST_F(HdlcReliabilityTest, GapRecoveredByReject) { + Peer a, b; + pump(a, b); + bool dropped = false; + a.send({0x01}); + a.send({0x02}); + a.send({0x03}); + pump(a, b, [&dropped](const Frame &frame) { + if (dropped || !is_info(frame)) + return false; + dropped = true; + return true; + }); + EXPECT_EQ(b.received, (std::vector{{0x01}, {0x02}, {0x03}})); +} + +TEST_F(HdlcReliabilityTest, LossRecoveredByTimeout) { + Peer a, b; + pump(a, b); + a.send({0x42}); + pump(a, b, [](const Frame &frame) { return is_info(frame); }); + EXPECT_TRUE(b.received.empty()); + a.hdlc->on_timeout_(); + pump(a, b); + EXPECT_EQ(b.received, (std::vector{{0x42}})); +} + +TEST_F(HdlcReliabilityTest, LostAckDoesNotDuplicate) { + Peer a, b; + pump(a, b); + a.send({0x42}); + pump(a, b, nullptr, [](const Frame &frame) { return is_ack(frame); }); + ASSERT_EQ(b.received.size(), 1); + a.hdlc->on_timeout_(); + pump(a, b); + EXPECT_EQ(b.received.size(), 1); + // The resend was acknowledged, so the full queue is available again + for (int i = 0; i != 4; i++) + EXPECT_TRUE(a.send({0x01})); +} + +TEST_F(HdlcReliabilityTest, QueueFull) { + Peer a(2), b(2); + pump(a, b); + EXPECT_TRUE(a.send({0x01})); + EXPECT_TRUE(a.send({0x02})); + EXPECT_FALSE(a.send({0x03})); +} + +TEST_F(HdlcReliabilityTest, OversizedPacketRejected) { + Peer a; + EXPECT_FALSE(a.send(Frame(a.hdlc->get_max_packet_size() + 1))); +} + +TEST_F(HdlcReliabilityTest, CrcErrorDropped) { + Peer a, b; + pump(a, b); + Frame frame = make_frame(0x00, {0x55}); // I-frame N(S)=0 N(R)=0 + frame[1] ^= 0xFF; + b.link.inject(frame); + EXPECT_TRUE(b.received.empty()); + b.link.inject(make_frame(0x00, {0x55})); + EXPECT_EQ(b.received, (std::vector{{0x55}})); +} + +TEST_F(HdlcReliabilityTest, RetriesExhaustedResetsLink) { + Peer a, b; + pump(a, b); + a.send({0x01}); + pump(a, b, [](const Frame &frame) { return is_info(frame); }); + for (int i = 0; i != 4; i++) { + a.hdlc->on_timeout_(); + pump(a, b, [](const Frame &frame) { return is_info(frame); }); + } + // The unacknowledged frame was discarded and the link was reset, which both report as data lost + EXPECT_TRUE(b.received.empty()); + EXPECT_TRUE(a.hdlc->is_connected()); + EXPECT_EQ(a.states, std::vector{false}); + EXPECT_EQ(b.states, std::vector{false}); + a.send({0x02}); + pump(a, b); + EXPECT_EQ(b.received, (std::vector{{0x02}})); +} + +TEST_F(HdlcReliabilityTest, PeerRestart) { + Peer a; + auto b = std::make_unique(); + pump(a, *b); + a.send({0x01}); + pump(a, *b); + // The peer restarts with fresh sequence numbers + b = std::make_unique(); + pump(a, *b); + a.send({0x02}); + b->send({0x03}); + pump(a, *b); + EXPECT_EQ(b->received, (std::vector{{0x02}})); + EXPECT_EQ(a.received, (std::vector{{0x03}})); + // The restarted peer's reset discarded what a had not yet had acknowledged + EXPECT_EQ(a.states, std::vector{false}); + EXPECT_TRUE(b->states.empty()); +} + +TEST_F(HdlcReliabilityTest, RefusedFramesSentAgainWithoutTimeout) { + Peer a, b; + for (uint8_t i = 0; i != 3; i++) + ASSERT_TRUE(a.send({i})); + a.link.budget = 3; // SABM, UA to the peer's SABM, and the first queued frame + pump(a, b); + EXPECT_GE(a.link.refused, 1); + EXPECT_EQ(b.received, (std::vector{{0}})); + a.link.budget = SIZE_MAX; // the lower layer accepts again, as when a radio stops receiving + pump(a, b); + EXPECT_EQ(b.received, (std::vector{{0}, {1}, {2}})); + EXPECT_EQ(a.hdlc->outstanding_(), 0); +} + +TEST_F(HdlcReliabilityTest, RefusedAckSentAgain) { + Peer a, b; + pump(a, b); + a.send({1}); + for (auto &frame : a.link.sent) + b.link.inject(frame); + a.link.sent.clear(); + b.link.budget = 0; + b.hdlc->loop(); + EXPECT_TRUE(b.link.sent.empty()); + b.link.budget = SIZE_MAX; + pump(a, b); + EXPECT_EQ(a.hdlc->outstanding_(), 0); +} + +TEST_F(HdlcReliabilityTest, AckDelayLetsReplyCarryAck) { + Peer a, b; + b.hdlc->set_ack_delay(100); + pump(a, b); + a.send({1}); + for (auto &frame : a.link.sent) + b.link.inject(frame); + a.link.sent.clear(); + b.hdlc->loop(); + EXPECT_TRUE(b.link.sent.empty()); // no RR yet + b.send({2}); + ASSERT_EQ(b.link.sent.size(), 1); + EXPECT_TRUE(is_info(b.link.sent[0])); + EXPECT_EQ(b.link.sent[0][0] >> 5, 1); // N(R) acknowledges the frame from a + b.link.sent.clear(); + b.hdlc->loop(); + EXPECT_TRUE(b.link.sent.empty()); // the RR is no longer owed +} + +} // namespace esphome::packet_interface::testing diff --git a/tests/unit_tests/test_packet_interface.py b/tests/unit_tests/test_packet_interface.py index 3e594592dec..e3343380d96 100644 --- a/tests/unit_tests/test_packet_interface.py +++ b/tests/unit_tests/test_packet_interface.py @@ -1,10 +1,20 @@ """Unit tests for the packet_interface component.""" +from typing import Any + import pytest from esphome import config_validation as cv from esphome.components import packet_interface from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE +from esphome.components.packet_interface.layers import ( + CONF_RELIABILITY, + RELIABILITY, + STACK_ORDER, +) +from esphome.components.packet_interface.layers.hdlc_reliability import ( + CONF_MAX_QUEUE_LENGTH, +) from esphome.components.uart import packet_interface as uart_packet_interface from esphome.components.udp import packet_interface as udp_packet_interface @@ -34,3 +44,55 @@ def test_schema_default(setup_core) -> None: packet_interface.PacketInterface, 100 ) assert schema({})[CONF_MAX_PACKET_SIZE] == 100 + + +def test_reliability_defaults(setup_core) -> None: + """An empty reliability key selects the HDLC layer with its defaults.""" + config = uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: None}) + reliability = config[CONF_RELIABILITY] + assert reliability["type"] == "hdlc" + assert reliability["timeout"].total_milliseconds == 200 + assert reliability[CONF_MAX_QUEUE_LENGTH] == 4 + assert reliability["max_retries"] == 5 + + +def test_reliability_queue_length_limit(setup_core) -> None: + """The queue cannot hold more frames than 3 bit sequence numbers allow.""" + with pytest.raises(cv.Invalid): + uart_packet_interface.CONFIG_SCHEMA( + {CONF_RELIABILITY: {CONF_MAX_QUEUE_LENGTH: 8}} + ) + + +def test_ack_delay_shorter_than_timeout(setup_core) -> None: + """The acknowledgement must not wait longer than the peer's timeout.""" + reliability = uart_packet_interface.CONFIG_SCHEMA( + {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}} + )[CONF_RELIABILITY] + assert reliability["ack_delay"].total_milliseconds == 100 + with pytest.raises(cv.Invalid, match="shorter"): + uart_packet_interface.CONFIG_SCHEMA( + {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}} + ) + + +def test_layer_types_registered() -> None: + """Importing packet_interface registers the built-in layer types.""" + assert {kind.key: set(kind.types) for kind in STACK_ORDER} == { + CONF_RELIABILITY: {"hdlc"}, + } + + +def test_registered_type_selectable(setup_core) -> None: + """A newly registered layer type can be selected with type.""" + + async def build(config: dict, lower: Any) -> Any: + return lower + + RELIABILITY.register("test", schema=cv.Schema({}))(build) + try: + config = RELIABILITY.schema()({"type": "test"}) + assert config["type"] == "test" + assert RELIABILITY.get_type(config).build is build + finally: + del RELIABILITY.types["test"]