Add a packet deduplicator and tests using a bitset window

Author: TrentHouliston

src/extension/network/NUClearNetwork.cpp

@@ -675,39 +675,35 @@ namespace extension {
                             // Check if this packet is a retransmission of data
                             if (header.type == DATA_RETRANSMISSION) {
 
-                                // See if we recently processed this packet
-                                // NOLINTNEXTLINE(readability-qualified-auto) MSVC disagrees
-                                auto it = std::find(remote->recent_packets.begin(),
-                                                    remote->recent_packets.end(),
-                                                    packet.packet_id);
-
                                 // We recently processed this packet, this is just a failed ack
-                                // Send the ack again if it was reliable
-                                if (it != remote->recent_packets.end() && packet.reliable) {
+                                if (remote->deduplicator.is_duplicate(packet.packet_id)) {
 
-                                    // Allocate room for the whole ack packet
-                                    std::vector<uint8_t> r(sizeof(ACKPacket) + (packet.packet_count / 8), 0);
-                                    ACKPacket& response   = *reinterpret_cast<ACKPacket*>(r.data());
-                                    response              = ACKPacket();
-                                    response.packet_id    = packet.packet_id;
-                                    response.packet_no    = packet.packet_no;
-                                    response.packet_count = packet.packet_count;
+                                    // Send the ack again if it was reliable
+                                    if (packet.reliable) {
+                                        // Allocate room for the whole ack packet
+                                        std::vector<uint8_t> r(sizeof(ACKPacket) + (packet.packet_count / 8), 0);
+                                        ACKPacket& response   = *reinterpret_cast<ACKPacket*>(r.data());
+                                        response              = ACKPacket();
+                                        response.packet_id    = packet.packet_id;
+                                        response.packet_no    = packet.packet_no;
+                                        response.packet_count = packet.packet_count;
 
-                                    // Set the bits for all packets (we got the whole thing)
-                                    for (int i = 0; i < packet.packet_count; ++i) {
-                                        (&response.packets)[i / 8] |= uint8_t(1 << (i % 8));
-                                    }
+                                        // Set the bits for all packets (we got the whole thing)
+                                        for (int i = 0; i < packet.packet_count; ++i) {
+                                            (&response.packets)[i / 8] |= uint8_t(1 << (i % 8));
+                                        }
 
-                                    // Make who we are sending it to into a useable address
-                                    const sock_t& to = remote->target;
+                                        // Make who we are sending it to into a useable address
+                                        const sock_t& to = remote->target;
 
-                                    // Send the packet
-                                    ::sendto(data_fd,
-                                             reinterpret_cast<const char*>(r.data()),
-                                             static_cast<socklen_t>(r.size()),
-                                             0,
-                                             &to.sock,
-                                             to.size());
+                                        // Send the packet
+                                        ::sendto(data_fd,
+                                                 reinterpret_cast<const char*>(r.data()),
+                                                 static_cast<socklen_t>(r.size()),
+                                                 0,
+                                                 &to.sock,
+                                                 to.size());
+                                    }
 
                                     // We don't need to process this packet we already did
                                     return;
@@ -739,13 +735,11 @@ namespace extension {
                                              0,
                                              &to.sock,
                                              to.size());
-
-                                    // Set this packet to have been recently received
-                                    remote->recent_packets[remote->recent_packets_index
-                                                               .fetch_add(1, std::memory_order_relaxed)] =
-                                        packet.packet_id;
                                 }
 
+                                // Add the packet to our deduplicator
+                                remote->deduplicator.add_packet(packet.packet_id);
+
                                 packet_callback(*remote, packet.hash, packet.reliable, std::move(out));
                             }
                             else {
@@ -851,17 +845,12 @@ namespace extension {
                                                    &part.data + p.second.size() - sizeof(DataPacket) + 1);
                                     }
 
+                                    // Add the packet to our deduplicator
+                                    remote->deduplicator.add_packet(packet.packet_id);
+
                                     // Send our assembled data packet
                                     packet_callback(*remote, packet.hash, packet.reliable, std::move(out));
 
-                                    // If the packet was reliable add that it was recently received
-                                    if (packet.reliable) {
-                                        // Set this packet to have been recently received
-                                        remote->recent_packets[remote->recent_packets_index
-                                                                   .fetch_add(1, std::memory_order_relaxed)] =
-                                            packet.packet_id;
-                                    }
-
                                     // We have completed this packet, discard the data
                                     assemblers.erase(assemblers.find(packet.packet_id));
                                 }

src/extension/network/NUClearNetwork.hpp

@@ -39,6 +39,7 @@
 
 #include "../../util/network/sock_t.hpp"
 #include "../../util/platform.hpp"
+#include "PacketDeduplicator.hpp"
 #include "RTTEstimator.hpp"
 #include "wire_protocol.hpp"
 
@@ -57,22 +58,14 @@ namespace extension {
                     std::string name,
                     const sock_t& target,
                     const std::chrono::steady_clock::time_point& last_update = std::chrono::steady_clock::now())
-                    : name(std::move(name)), target(target), last_update(last_update) {
-
-                    // Set our recent packets to an invalid value
-                    recent_packets.fill(-1);
-                }
+                    : name(std::move(name)), target(target), last_update(last_update) {}
 
                 /// The name of the remote target
                 std::string name;
                 /// The socket address for the remote target
                 sock_t target{};
                 /// When we last received data from the remote target
                 std::chrono::steady_clock::time_point last_update;
-                /// A list of the last n packet groups to be received
-                std::array<int, std::numeric_limits<uint8_t>::max()> recent_packets{};
-                /// An index for the recent_packets (circular buffer)
-                std::atomic<uint8_t> recent_packets_index{0};
                 /// Mutex to protect the fragmented packet storage
                 std::mutex assemblers_mutex;
                 /// Storage for fragmented packets while we build them
@@ -82,6 +75,9 @@ namespace extension {
 
                 /// RTT estimator for this network target
                 RTTEstimator rtt;
+
+                /// Packet deduplicator for this network target
+                PacketDeduplicator deduplicator;
             };
 
             NUClearNetwork() = default;

src/extension/network/PacketDeduplicator.cpp

@@ -0,0 +1,75 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 NUClear Contributors
+ *
+ * This file is part of the NUClear codebase.
+ * See https://github.com/Fastcode/NUClear for further info.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+#include "PacketDeduplicator.hpp"
+
+namespace NUClear {
+namespace extension {
+    namespace network {
+
+        PacketDeduplicator::PacketDeduplicator() : newest_seen(0) {}
+
+        bool PacketDeduplicator::is_duplicate(uint16_t packet_id) const {
+            // If we haven't seen any packets yet, nothing is a duplicate
+            if (!initialized) {
+                return false;
+            }
+
+            // Calculate relative position in window using unsigned subtraction
+            uint16_t relative_id = newest_seen - packet_id;
+
+            // If the packet is too old or too new, it's not a duplicate
+            if (relative_id >= 256) {
+                return false;
+            }
+
+            return window[relative_id];
+        }
+
+        void PacketDeduplicator::add_packet(uint16_t packet_id) {
+            // If this is our first packet, just set it as newest_seen
+            if (!initialized) {
+                newest_seen = packet_id;
+                window[0]   = true;
+                initialized = true;
+                return;
+            }
+
+            // Calculate relative position in window using unsigned subtraction
+            uint16_t relative_id = newest_seen - packet_id;
+
+            // If the distance is more than half the range, the packet is newer than our newest_seen
+            if (relative_id > 32768) {
+                // Calculate how far to shift to make this packet our newest
+                uint16_t shift_amount = packet_id - newest_seen;
+                window <<= shift_amount;
+                newest_seen = packet_id;
+                window[0]   = true;
+            }
+            // Packet is recent enough to be counted
+            else if (relative_id < 256) {
+                window[relative_id] = true;
+            }
+        }
+
+    }  // namespace network
+}  // namespace extension
+}  // namespace NUClear

src/extension/network/PacketDeduplicator.hpp

@@ -0,0 +1,69 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 NUClear Contributors
+ *
+ * This file is part of the NUClear codebase.
+ * See https://github.com/Fastcode/NUClear for further info.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+#ifndef NUCLEAR_EXTENSION_NETWORK_PACKET_DEDUPLICATOR_HPP
+#define NUCLEAR_EXTENSION_NETWORK_PACKET_DEDUPLICATOR_HPP
+
+#include <bitset>
+#include <cstdint>
+
+namespace NUClear {
+namespace extension {
+    namespace network {
+
+        /**
+         * A class that implements a sliding window bitset for packet deduplication.
+         * Maintains a 256-bit window of recently seen packet IDs, sliding forward as new packets are added.
+         */
+        class PacketDeduplicator {
+        public:
+            PacketDeduplicator();
+
+            /**
+             * Check if a packet ID has been seen recently
+             *
+             * @param packet_id The packet ID to check
+             *
+             * @return true if the packet has been seen recently, false otherwise
+             */
+            bool is_duplicate(uint16_t packet_id) const;
+
+            /**
+             * Add a packet ID to the window
+             *
+             * @param packet_id The packet ID to add
+             */
+            void add_packet(uint16_t packet_id);
+
+        private:
+            /// Whether we've seen any packets yet
+            bool initialized{false};
+            /// The newest packet ID we've seen
+            uint16_t newest_seen;
+            /// The 256-bit window of seen packets (newest at 0, older at higher indices)
+            std::bitset<256> window;
+        };
+
+    }  // namespace network
+}  // namespace extension
+}  // namespace NUClear
+
+#endif  // NUCLEAR_EXTENSION_NETWORK_PACKET_DEDUPLICATOR_HPP