From 28c32cf1f667d7ffd3d6233fa5a168f95d3c0664 Mon Sep 17 00:00:00 2001 From: clydebarrow <2366188+clydebarrow@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:52:44 +1100 Subject: [PATCH] [packet_interface] Route broadcasts to the broadcast port, and keep a join A broadcast from a device with its own port went to that port, whose layers drop it, so a broadcast port could not reach reliable peers. It now goes to the broadcast port when there is one. Fragmentation allocates its buffers with RAMAllocator and fails setup when they cannot be allocated. A fragment from another source no longer drops the packet being joined. Co-Authored-By: Claude Opus 5.5 --- .../packet_interface/basic_addressing.cpp | 16 +++--- .../packet_interface/basic_addressing.h | 3 +- .../packet_interface/basic_fragmentation.cpp | 53 ++++++++++++------- .../packet_interface/basic_fragmentation.h | 13 ++--- .../layers/basic_fragmentation.py | 4 +- .../packet_interface/basic_addressing.cpp | 18 +++++++ .../packet_interface/basic_fragmentation.cpp | 17 +++++- 7 files changed, 87 insertions(+), 37 deletions(-) diff --git a/esphome/components/packet_interface/basic_addressing.cpp b/esphome/components/packet_interface/basic_addressing.cpp index 3519fca0d6f..a606ec48345 100644 --- a/esphome/components/packet_interface/basic_addressing.cpp +++ b/esphome/components/packet_interface/basic_addressing.cpp @@ -40,16 +40,20 @@ void BasicAddressing::receive_(const PacketBuffer &data, const PacketMetaData &m uint16_t source = encode_uint16(packet[3], packet[2]); if (source == this->address_ || (destination != this->address_ && destination != BROADCAST_ADDRESS)) return; - // A device with a port of its own is heard only there; the broadcast port hears the rest - AddressingPort *target = nullptr; + AddressingPort *source_port = nullptr; + AddressingPort *broadcast_port = nullptr; for (auto *port : this->ports_) { if (port->destination_ == source) { - target = port; - break; + source_port = port; + } else if (port->destination_ == BROADCAST_ADDRESS) { + broadcast_port = port; } - if (port->destination_ == BROADCAST_ADDRESS) - target = port; } + // A broadcast goes to the broadcast port, even from a device with a port of its own, as that port's layers expect + // only its point-to-point packets; anything else goes to the source's port, or to the broadcast port if it has none + AddressingPort *target = source_port; + if (broadcast_port != nullptr && (destination == BROADCAST_ADDRESS || source_port == nullptr)) + target = broadcast_port; if (target == nullptr) { ESP_LOGV(TAG, "%s: Packet from 0x%04X dropped", this->link_->get_name(), source); return; diff --git a/esphome/components/packet_interface/basic_addressing.h b/esphome/components/packet_interface/basic_addressing.h index 0dbd56d5a92..a9e6438e856 100644 --- a/esphome/components/packet_interface/basic_addressing.h +++ b/esphome/components/packet_interface/basic_addressing.h @@ -19,7 +19,8 @@ class AddressingPort; * Above this layer are one or more ports, each the bottom of its own stack of layers. A port with a destination * address is a point-to-point link to that device: it sends to it and receives only its packets, so the reliability * and session layers and the rolling code, which work between two devices only, can be used above it. A port with - * the broadcast destination sends to every device, and receives packets from devices that have no port of their own. + * the broadcast destination sends to every device, and receives every broadcast, as well as packets from devices + * that have no port of their own. Without a broadcast port, a device's broadcasts go to its own port. */ class BasicAddressing : public Component { public: diff --git a/esphome/components/packet_interface/basic_fragmentation.cpp b/esphome/components/packet_interface/basic_fragmentation.cpp index 119d4c59b48..1823bfd7c30 100644 --- a/esphome/components/packet_interface/basic_fragmentation.cpp +++ b/esphome/components/packet_interface/basic_fragmentation.cpp @@ -33,10 +33,16 @@ void BasicFragmentation::setup() { this->mark_failed(LOG_STR("max_packet_size needs more than 128 fragments")); return; } - // Checked now, while the layer below holds nothing - this->queue_too_short_ = this->fragments_for_(this->max_packet_size_) > this->lower_->get_send_capacity(); - this->tx_buffer_ = std::make_unique(this->fragment_size_()); - this->rx_buffer_ = std::make_unique(this->max_packet_size_); + // Checked now, while the layer below holds nothing; a failed layer below reports its own failure instead + this->queue_too_short_ = + !this->lower_->is_failed() && this->fragments_for_(this->max_packet_size_) > this->lower_->get_send_capacity(); + RAMAllocator allocator; + this->tx_buffer_ = allocator.allocate(this->fragment_size_() + this->max_packet_size_); + if (this->tx_buffer_ == nullptr) { + this->mark_failed(LOG_STR("Not enough memory for the packet buffers")); + return; + } + this->rx_buffer_ = this->tx_buffer_ + this->fragment_size_(); // Numbering from a random start makes it unlikely that a packet joins fragments from before a restart this->tx_number_ = random_uint32(); this->lower_->add_packet_interface_listener( @@ -58,7 +64,7 @@ void BasicFragmentation::dump_config() { } bool BasicFragmentation::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) { - if (this->is_failed()) + if (this->tx_buffer_ == nullptr) return false; size_t size = data.size(); if (size > this->max_packet_size_) { @@ -77,9 +83,9 @@ bool BasicFragmentation::send_to_interface(const PacketBuffer &data, const Packe } size_t fragment_size = this->fragment_size_(); for (size_t index = 0; index != fragments; index++) { - size_t length = data.copy_to(this->tx_buffer_.get(), fragment_size, index * fragment_size); + size_t length = data.copy_to(this->tx_buffer_, fragment_size, index * fragment_size); header[1] = index | (index + 1 == fragments ? LAST_FRAGMENT : 0); - PacketBuffer body(this->tx_buffer_.get(), length); + PacketBuffer body(this->tx_buffer_, length); if (!this->lower_->send_to_interface(PacketBuffer(header, HEADER_SIZE, &body, nullptr, 0), meta_data)) { // The peer drops the part already sent when the next packet starts ESP_LOGV(TAG, "%s: Fragment %zu of %zu not accepted", this->name_, index, fragments); @@ -103,26 +109,33 @@ void BasicFragmentation::receive_(const PacketBuffer &data, const PacketMetaData bool last = (fragment[1] & LAST_FRAGMENT) != 0; const uint8_t *payload = fragment.data() + HEADER_SIZE; size_t length = fragment.size() - HEADER_SIZE; + if (index == 0 && last) { + // A packet in one fragment leaves a packet being joined from another device intact + if (this->rx_joining_ && meta_data.source_address == this->rx_source_) { + ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_); + this->rx_joining_ = false; + } + if (length > this->max_packet_size_) { + ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_); + return; + } + this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data); + return; + } if (index == 0) { if (this->rx_joining_) { ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_); } - this->rx_joining_ = false; - if (last) { - if (length > this->max_packet_size_) { - ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_); - return; - } - this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data); - return; - } this->rx_joining_ = true; this->rx_number_ = number; this->rx_source_ = meta_data.source_address; this->rx_next_index_ = 0; this->rx_length_ = 0; - } else if (!this->rx_joining_ || number != this->rx_number_ || meta_data.source_address != this->rx_source_ || - index != this->rx_next_index_) { + } else if (this->rx_joining_ && meta_data.source_address != this->rx_source_) { + // Only this fragment is lost, so of two devices sending at once, the one that started last still gets through + ESP_LOGV(TAG, "%s: Fragment %u from 0x%04X dropped", this->name_, index, meta_data.source_address); + return; + } else if (!this->rx_joining_ || number != this->rx_number_ || index != this->rx_next_index_) { ESP_LOGV(TAG, "%s: Fragment %u of packet %u dropped", this->name_, index, number); this->rx_joining_ = false; return; @@ -132,12 +145,12 @@ void BasicFragmentation::receive_(const PacketBuffer &data, const PacketMetaData this->rx_joining_ = false; return; } - memcpy(this->rx_buffer_.get() + this->rx_length_, payload, length); + memcpy(this->rx_buffer_ + this->rx_length_, payload, length); this->rx_length_ += length; this->rx_next_index_++; if (last) { this->rx_joining_ = false; - this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_.get(), this->rx_length_), meta_data); + this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_, this->rx_length_), meta_data); } } diff --git a/esphome/components/packet_interface/basic_fragmentation.h b/esphome/components/packet_interface/basic_fragmentation.h index a3c0652eba7..9b6e337569e 100644 --- a/esphome/components/packet_interface/basic_fragmentation.h +++ b/esphome/components/packet_interface/basic_fragmentation.h @@ -2,7 +2,6 @@ #include "esphome/core/defines.h" #ifdef USE_PACKET_INTERFACE_BASIC_FRAGMENTATION -#include #include "packet_interface.h" @@ -18,8 +17,9 @@ namespace esphome::packet_interface { * * A packet is started only if the layer below can take all of its fragments now (get_send_capacity()), so a full * reliability queue does not leave the peer with part of a packet. The receiver joins one packet at a time, and drops - * it when a fragment is missing, or one arrives with another packet number or from another source address. Above a - * reliability layer, fragments arrive in order and none are missing. + * it when a fragment is missing or arrives with another packet number, or when the same source, or another one, + * starts a new packet of several fragments. Other fragments from another source address are dropped, leaving the + * packet being joined intact. Above a reliability layer, fragments arrive in order and none are missing. */ class BasicFragmentation : public PacketInterface { public: @@ -43,9 +43,10 @@ class BasicFragmentation : public PacketInterface { size_t fragments_for_(size_t size) const; PacketInterface *lower_; - // One fragment being sent, and the packet being joined; both allocated in setup() - std::unique_ptr tx_buffer_; - std::unique_ptr rx_buffer_; + // One fragment being sent, and the packet being joined; one block allocated in setup(), and tx_buffer_ stays null + // if that fails + uint8_t *tx_buffer_{nullptr}; + uint8_t *rx_buffer_{nullptr}; size_t rx_length_{0}; uint16_t rx_source_{0}; uint8_t tx_number_{0}; diff --git a/esphome/components/packet_interface/layers/basic_fragmentation.py b/esphome/components/packet_interface/layers/basic_fragmentation.py index cb7ef02604b..06826520b05 100644 --- a/esphome/components/packet_interface/layers/basic_fragmentation.py +++ b/esphome/components/packet_interface/layers/basic_fragmentation.py @@ -21,8 +21,8 @@ BASIC_FRAGMENTATION_SCHEMA = cv.Schema( ).extend(cv.COMPONENT_SCHEMA) -# Fragments from different devices are told apart by the addressing layer's source address, so several devices can -# share a channel; without addressing, two devices sending fragmented packets at once can make the receiver drop both +# Fragments from different devices are told apart by the addressing layer's source address, so of two devices sending +# fragmented packets at once, one still gets through; without addressing, the receiver can drop both @FRAGMENTATION.register("basic", schema=BASIC_FRAGMENTATION_SCHEMA) async def build_basic_fragmentation(config: ConfigType, lower: MockObj) -> MockObj: cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION") diff --git a/tests/components/packet_interface/basic_addressing.cpp b/tests/components/packet_interface/basic_addressing.cpp index 6efa5ae900f..37e718d1b1e 100644 --- a/tests/components/packet_interface/basic_addressing.cpp +++ b/tests/components/packet_interface/basic_addressing.cpp @@ -143,6 +143,24 @@ TEST(BasicAddressingTest, PeersHaveSeparateStacks) { EXPECT_TRUE(d.port().received.empty()); } +TEST(BasicAddressingTest, BroadcastFromPeerReachesBroadcastPort) { + Device hub(1, {2, BROADCAST}), b(2, {1, BROADCAST}); + hub.port(1).send({0x10}); // a state every device needs + hub.port(0).send({0x11}); // point to point + hub.deliver({&b}); + EXPECT_EQ(b.port(0).received, std::vector{{0x11}}); + EXPECT_EQ(b.port(1).received, std::vector{{0x10}}); + EXPECT_EQ(b.port(1).sources, std::vector{1}); +} + +TEST(BasicAddressingTest, WithoutBroadcastPortPeerBroadcastReachesItsPort) { + Device hub(1, {2, 3}), b(2, BROADCAST); + b.port().send({0x20}); + b.deliver({&hub}); + EXPECT_EQ(hub.port(0).received, std::vector{{0x20}}); + EXPECT_TRUE(hub.port(1).received.empty()); +} + TEST(BasicAddressingTest, WithoutBroadcastPortUnknownDevicesDropped) { Device hub(1, {2, 3}), d(4); d.port().send({0x04}); diff --git a/tests/components/packet_interface/basic_fragmentation.cpp b/tests/components/packet_interface/basic_fragmentation.cpp index 01586cb9ae6..ba0c7efa342 100644 --- a/tests/components/packet_interface/basic_fragmentation.cpp +++ b/tests/components/packet_interface/basic_fragmentation.cpp @@ -161,8 +161,21 @@ TEST_F(BasicFragmentationTest, InterleavedSourcesNotMixed) { this->link_b_.inject(from_c[0], 2); this->link_b_.inject(from_a[1], 1); this->link_b_.inject(from_c[1], 2); - for (auto &packet : b.received) - EXPECT_EQ(packet, make_packet(20)); // nothing mixed from the two + // Nothing is mixed from the two, and the packet started last gets through + EXPECT_EQ(b.received, std::vector{make_packet(20)}); +} + +TEST_F(BasicFragmentationTest, OneFragmentPacketFromAnotherSourceKeepsJoin) { + Peer a(&this->link_a_), c(&this->link_a_), b(&this->link_b_); + a.send(make_packet(20)); + auto from_a = std::move(this->link_a_.sent); + this->link_a_.sent.clear(); + c.send(make_packet(5)); + auto from_c = std::move(this->link_a_.sent); + this->link_b_.inject(from_a[0], 1); + this->link_b_.inject(from_c[0], 2); + this->link_b_.inject(from_a[1], 1); + EXPECT_EQ(b.received, (std::vector{make_packet(5), make_packet(20)})); } TEST_F(BasicFragmentationTest, OverReliabilityWithLosses) {