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) {