Merge pull request #873 from tewinget/path-queues

Add per-path queues, prioritize control messages over traffic

llarp/ev/ev.hpp

@@ -671,8 +671,7 @@ namespace llarp
 #ifdef _WIN32
 struct llarp_fd_promise
 {
-  void
-  Set(std::pair< int, int >)
+  void Set(std::pair< int, int >)
   {
   }
 

llarp/messages/link_message.hpp

@@ -4,6 +4,7 @@
 #include <link/session.hpp>
 #include <router_id.hpp>
 #include <util/bencode.hpp>
+#include <path/path_types.hpp>
 
 #include <vector>
 
@@ -19,6 +20,8 @@ namespace llarp
     ILinkSession* session = nullptr;
     uint64_t version      = LLARP_PROTO_VERSION;
 
+    PathID_t pathid;
+
     ILinkMessage() = default;
 
     virtual ~ILinkMessage() = default;

llarp/messages/relay.hpp

@@ -12,7 +12,6 @@ namespace llarp
 {
   struct RelayUpstreamMessage : public ILinkMessage
   {
-    PathID_t pathid;
     Encrypted< MAX_LINK_MSG_SIZE - 128 > X;
     TunnelNonce Y;
 
@@ -37,7 +36,6 @@ namespace llarp
 
   struct RelayDownstreamMessage : public ILinkMessage
   {
-    PathID_t pathid;
     Encrypted< MAX_LINK_MSG_SIZE - 128 > X;
     TunnelNonce Y;
 

llarp/messages/relay_commit.cpp

@@ -322,6 +322,9 @@ namespace llarp
       self->hop = nullptr;
     }
 
+    // TODO: If decryption has succeeded here but we otherwise don't
+    //       want to or can't accept the path build request, send
+    //       a status message saying as much.
     static void
     HandleDecrypted(llarp_buffer_t* buf,
                     std::shared_ptr< LRCMFrameDecrypt > self)
@@ -344,8 +347,16 @@ namespace llarp
         return;
       }
 
-      info.txID     = self->record.txid;
-      info.rxID     = self->record.rxid;
+      info.txID = self->record.txid;
+      info.rxID = self->record.rxid;
+
+      if(info.txID.IsZero() || info.rxID.IsZero())
+      {
+        llarp::LogError("LRCM refusing zero pathid");
+        self->decrypter = nullptr;
+        return;
+      }
+
       info.upstream = self->record.nextHop;
 
       // generate path key as we are in a worker thread

llarp/path/path.cpp

@@ -31,8 +31,15 @@ namespace llarp
       for(size_t idx = 0; idx < hsz; ++idx)
       {
         hops[idx].rc = h[idx];
-        hops[idx].txID.Randomize();
-        hops[idx].rxID.Randomize();
+        do
+        {
+          hops[idx].txID.Randomize();
+        } while(hops[idx].txID.IsZero());
+
+        do
+        {
+          hops[idx].rxID.Randomize();
+        } while(hops[idx].rxID.IsZero());
       }
 
       for(size_t idx = 0; idx < hsz - 1; ++idx)

llarp/path/path_context.cpp

@@ -289,6 +289,7 @@ namespace llarp
         {
           if(itr->second->Expired(now))
           {
+            m_Router->outboundMessageHandler().QueueRemoveEmptyPath(itr->first);
             itr = map.erase(itr);
           }
           else

llarp/path/path_types.hpp

@@ -8,6 +8,7 @@ namespace llarp
 {
   struct PathID_t final : public AlignedBuffer< PATHIDSIZE >
   {
+    using Hash = AlignedBuffer< PATHIDSIZE >::Hash;
   };
 
 }  // namespace llarp

llarp/path/pathbuilder.cpp

@@ -165,7 +165,7 @@ namespace llarp
     void Builder::Tick(llarp_time_t)
     {
       const auto now = llarp::time_now_ms();
-      ExpirePaths(now);
+      ExpirePaths(now, m_router);
       if(ShouldBuildMore(now))
         BuildOne();
       TickPaths(m_router);

llarp/path/pathset.cpp

@@ -73,7 +73,7 @@ namespace llarp
     }
 
     void
-    PathSet::ExpirePaths(llarp_time_t now)
+    PathSet::ExpirePaths(llarp_time_t now, AbstractRouter* router)
     {
       Lock_t l(&m_PathsMutex);
       if(m_Paths.size() == 0)
@@ -83,6 +83,8 @@ namespace llarp
       {
         if(itr->second->Expired(now))
         {
+          router->outboundMessageHandler().QueueRemoveEmptyPath(
+              itr->second->TXID());
           itr = m_Paths.erase(itr);
         }
         else

llarp/path/pathset.hpp

@@ -148,7 +148,7 @@ namespace llarp
       GetByUpstream(RouterID remote, PathID_t rxid) const;
 
       void
-      ExpirePaths(llarp_time_t now);
+      ExpirePaths(llarp_time_t now, AbstractRouter* router);
 
       /// get the number of paths in this status
       size_t
@@ -199,15 +199,15 @@ namespace llarp
       }
 
       /// override me in subtype
-      virtual bool
-      HandleGotIntroMessage(std::shared_ptr< const dht::GotIntroMessage >)
+      virtual bool HandleGotIntroMessage(
+          std::shared_ptr< const dht::GotIntroMessage >)
       {
         return false;
       }
 
       /// override me in subtype
-      virtual bool
-      HandleGotRouterMessage(std::shared_ptr< const dht::GotRouterMessage >)
+      virtual bool HandleGotRouterMessage(
+          std::shared_ptr< const dht::GotRouterMessage >)
       {
         return false;
       }

llarp/router/i_outbound_message_handler.hpp

@@ -20,9 +20,14 @@ namespace llarp
 
   struct ILinkMessage;
   struct RouterID;
+  struct PathID_t;
 
   using SendStatusHandler = std::function< void(SendStatus) >;
 
+  static const size_t MAX_PATH_QUEUE_SIZE            = 40;
+  static const size_t MAX_OUTBOUND_QUEUE_SIZE        = 200;
+  static const size_t MAX_OUTBOUND_MESSAGES_PER_TICK = 20;
+
   struct IOutboundMessageHandler
   {
     virtual ~IOutboundMessageHandler() = default;
@@ -31,6 +36,12 @@ namespace llarp
     QueueMessage(const RouterID &remote, const ILinkMessage *msg,
                  SendStatusHandler callback) = 0;
 
+    virtual void
+    Tick() = 0;
+
+    virtual void
+    QueueRemoveEmptyPath(const PathID_t &pathid) = 0;
+
     virtual util::StatusObject
     ExtractStatus() const = 0;
   };