put replay filters on transit hops to reduce retransmissions.

llarp/path/ihophandler.cpp

@@ -8,6 +8,8 @@ namespace llarp
     bool
     IHopHandler::HandleUpstream(const llarp_buffer_t& X, const TunnelNonce& Y, AbstractRouter*)
     {
+      if (not m_UpstreamReplayFilter.Insert(Y))
+        return false;
       if (m_UpstreamQueue == nullptr)
         m_UpstreamQueue = std::make_shared<TrafficQueue_t>();
       m_UpstreamQueue->emplace_back();
@@ -22,6 +24,8 @@ namespace llarp
     bool
     IHopHandler::HandleDownstream(const llarp_buffer_t& X, const TunnelNonce& Y, AbstractRouter*)
     {
+      if (not m_DownstreamReplayFilter.Insert(Y))
+        return false;
       if (m_DownstreamQueue == nullptr)
         m_DownstreamQueue = std::make_shared<TrafficQueue_t>();
       m_DownstreamQueue->emplace_back();
@@ -31,5 +35,12 @@ namespace llarp
       pkt.second = Y;
       return true;
     }
+
+    void
+    IHopHandler::DecayFilters(llarp_time_t now)
+    {
+      m_UpstreamReplayFilter.Decay(now);
+      m_DownstreamReplayFilter.Decay(now);
+    }
   }  // namespace path
 }  // namespace llarp

llarp/path/ihophandler.hpp

@@ -4,6 +4,7 @@
 #include <crypto/types.hpp>
 #include <util/types.hpp>
 #include <crypto/encrypted_frame.hpp>
+#include <util/decaying_hashset.hpp>
 #include <messages/relay.hpp>
 #include <vector>
 
@@ -30,6 +31,9 @@ namespace llarp
 
       virtual ~IHopHandler() = default;
 
+      void
+      DecayFilters(llarp_time_t now);
+
       virtual bool
       Expired(llarp_time_t now) const = 0;
 
@@ -70,6 +74,8 @@ namespace llarp
       uint64_t m_SequenceNum = 0;
       TrafficQueue_ptr m_UpstreamQueue;
       TrafficQueue_ptr m_DownstreamQueue;
+      util::DecayingHashSet<TunnelNonce> m_UpstreamReplayFilter;
+      util::DecayingHashSet<TunnelNonce> m_DownstreamReplayFilter;
 
       virtual void
       UpstreamWork(TrafficQueue_ptr queue, AbstractRouter* r) = 0;

llarp/path/path.cpp

@@ -58,19 +58,6 @@ namespace llarp
         EnterState(ePathBuilding, parent->Now());
     }
 
-    bool
-    Path::HandleUpstream(const llarp_buffer_t& X, const TunnelNonce& Y, AbstractRouter* r)
-
-    {
-      return m_UpstreamReplayFilter.Insert(Y) and IHopHandler::HandleUpstream(X, Y, r);
-    }
-
-    bool
-    Path::HandleDownstream(const llarp_buffer_t& X, const TunnelNonce& Y, AbstractRouter* r)
-    {
-      return m_DownstreamReplayFilter.Insert(Y) and IHopHandler::HandleDownstream(X, Y, r);
-    }
-
     void
     Path::SetBuildResultHook(BuildResultHookFunc func)
     {
@@ -372,9 +359,6 @@ namespace llarp
       m_RXRate = 0;
       m_TXRate = 0;
 
-      m_UpstreamReplayFilter.Decay(now);
-      m_DownstreamReplayFilter.Decay(now);
-
       if (_status == ePathBuilding)
       {
         if (buildStarted == 0s)

llarp/path/path.hpp

@@ -26,8 +26,6 @@
 #include <unordered_set>
 #include <vector>
 
-#include <util/decaying_hashset.hpp>
-
 namespace llarp
 {
   class Logic;
@@ -282,11 +280,6 @@ namespace llarp
       void
       Rebuild();
 
-      bool
-      HandleUpstream(const llarp_buffer_t& X, const TunnelNonce& Y, AbstractRouter*) override;
-      bool
-      HandleDownstream(const llarp_buffer_t& X, const TunnelNonce& Y, AbstractRouter*) override;
-
       void
       Tick(llarp_time_t now, AbstractRouter* r);
 
@@ -420,8 +413,6 @@ namespace llarp
       uint64_t m_ExitObtainTX = 0;
       PathStatus _status;
       PathRole _role;
-      util::DecayingHashSet<TunnelNonce> m_UpstreamReplayFilter;
-      util::DecayingHashSet<TunnelNonce> m_DownstreamReplayFilter;
       uint64_t m_LastRXRate = 0;
       uint64_t m_RXRate = 0;
       uint64_t m_LastTXRate = 0;

llarp/path/path_context.cpp

@@ -308,6 +308,7 @@ namespace llarp
         auto itr = map.begin();
         while (itr != map.end())
         {
+          itr->second->DecayFilters(now);
           if (itr->second->Expired(now))
           {
             m_Router->outboundMessageHandler().QueueRemoveEmptyPath(itr->first);
@@ -323,6 +324,7 @@ namespace llarp
         auto itr = map.begin();
         while (itr != map.end())
         {
+          itr->second->DecayFilters(now);
           if (itr->second->Expired(now))
           {
             itr = map.erase(itr);

llarp/util/decaying_hashset.hpp

@@ -20,7 +20,7 @@ namespace llarp
       bool
       Contains(const Val_t& v) const
       {
-        return m_Values.find(v) != m_Values.end();
+        return m_Values.count(v) != 0;
       }
 
       /// return true if inserted
@@ -30,7 +30,7 @@ namespace llarp
       {
         if (now == 0s)
           now = llarp::time_now_ms();
-        return m_Values.emplace(v, now).second;
+        return m_Values.try_emplace(v, now).second;
       }
 
       /// decay hashset entries
@@ -56,6 +56,12 @@ namespace llarp
         return m_CacheInterval;
       }
 
+      bool
+      Empty() const
+      {
+        return m_Values.empty();
+      }
+
       void
       DecayInterval(Time_t interval)
       {

test/util/test_llarp_util_decaying_hashset.cpp

@@ -2,11 +2,42 @@
 #include <router_id.hpp>
 #include <catch2/catch.hpp>
 
+TEST_CASE("Thrash DecayingHashSet", "[decaying-hashset]")
+{
+  static constexpr auto duration = 5s;
+
+  static constexpr auto decayInterval = 50ms;
+  llarp::util::DecayingHashSet<llarp::AlignedBuffer<32>> hashset(decayInterval);
+  const llarp_time_t started = llarp::time_now_ms();
+  const auto end = duration + started;
+  llarp_time_t nextDecay = started + decayInterval;
+  do
+  {
+    const auto now = llarp::time_now_ms();
+    for (size_t i = 0; i < 500; i++)
+    {
+      llarp::AlignedBuffer<32> rando;
+      rando.Randomize();
+      hashset.Insert(rando, now);
+      /// maybe reinsert to simulate filter hits
+      if (i % 20 == 0)
+        hashset.Insert(rando, now);
+    }
+    if (now >= nextDecay)
+    {
+      REQUIRE(not hashset.Empty());
+      hashset.Decay(now);
+      nextDecay += decayInterval;
+    }
+
+  } while (llarp::time_now_ms() <= end);
+}
+
 TEST_CASE("DecayingHashSet test decay static time", "[decaying-hashset]")
 {
   static constexpr auto timeout = 5s;
-  static constexpr auto now     = 1s;
-  llarp::util::DecayingHashSet< llarp::RouterID > hashset(timeout);
+  static constexpr auto now = 1s;
+  llarp::util::DecayingHashSet<llarp::RouterID> hashset(timeout);
   const llarp::RouterID zero;
   REQUIRE(zero.IsZero());
   REQUIRE(not hashset.Contains(zero));
@@ -23,8 +54,8 @@ TEST_CASE("DecayingHashSet test decay static time", "[decaying-hashset]")
 TEST_CASE("DecayingHashSet tset decay dynamic time", "[decaying-hashset]")
 {
   static constexpr llarp_time_t timeout = 5s;
-  const llarp_time_t now                = llarp::time_now_ms();
-  llarp::util::DecayingHashSet< llarp::RouterID > hashset(timeout);
+  const llarp_time_t now = llarp::time_now_ms();
+  llarp::util::DecayingHashSet<llarp::RouterID> hashset(timeout);
   const llarp::RouterID zero;
   REQUIRE(zero.IsZero());
   REQUIRE(not hashset.Contains(zero));