Merge pull request #1060 from loki-project/dev

0.6.4

llarp/constants/version.h

@@ -8,7 +8,7 @@
 
 #define LLARP_VERSION_MAJ 0
 #define LLARP_VERSION_MIN 6
-#define LLARP_VERSION_PATCH 3
+#define LLARP_VERSION_PATCH 4
 
 #define LLARP_DEFAULT_NETID "lokinet"
 

llarp/messages/link_intro.hpp

@@ -45,6 +45,13 @@ namespace llarp
     {
       return "LinkIntro";
     }
+
+    // always first
+    uint16_t
+    Priority() const override
+    {
+      return std::numeric_limits< uint16_t >::max();
+    }
   };
 }  // namespace llarp
 

llarp/messages/link_message.hpp

@@ -55,6 +55,13 @@ namespace llarp
     // the name of this kind of message
     virtual const char*
     Name() const = 0;
+
+    /// get message prority, higher value means more important
+    virtual uint16_t
+    Priority() const
+    {
+      return 1;
+    }
   };
 
 }  // namespace llarp

llarp/messages/relay.hpp

@@ -32,6 +32,11 @@ namespace llarp
     {
       return "RelayUpstream";
     }
+    uint16_t
+    Priority() const override
+    {
+      return 0;
+    }
   };
 
   struct RelayDownstreamMessage : public ILinkMessage
@@ -56,6 +61,12 @@ namespace llarp
     {
       return "RelayDownstream";
     }
+
+    uint16_t
+    Priority() const override
+    {
+      return 0;
+    }
   };
 }  // namespace llarp
 

llarp/messages/relay_commit.hpp

@@ -79,6 +79,12 @@ namespace llarp
     {
       return "RelayCommit";
     }
+
+    virtual uint16_t
+    Priority() const override
+    {
+      return 5;
+    }
   };
 }  // namespace llarp
 

llarp/messages/relay_status.hpp

@@ -103,6 +103,11 @@ namespace llarp
     {
       return "RelayStatus";
     }
+    virtual uint16_t
+    Priority() const override
+    {
+      return 6;
+    }
   };
 }  // namespace llarp
 

llarp/router/outbound_message_handler.cpp

@@ -24,6 +24,7 @@ namespace llarp
                                        const ILinkMessage *msg,
                                        SendStatusHandler callback)
   {
+    const uint16_t priority = msg->Priority();
     std::array< byte_t, MAX_LINK_MSG_SIZE > linkmsg_buffer;
     llarp_buffer_t buf(linkmsg_buffer);
 
@@ -40,7 +41,7 @@ namespace llarp
 
     if(_linkManager->HasSessionTo(remote))
     {
-      QueueOutboundMessage(remote, std::move(message), msg->pathid);
+      QueueOutboundMessage(remote, std::move(message), msg->pathid, priority);
       return true;
     }
 
@@ -53,8 +54,9 @@ namespace llarp
           pendingSessionMessageQueues.emplace(remote, MessageQueue());
 
       MessageQueueEntry entry;
-      entry.message = message;
-      entry.router  = remote;
+      entry.priority = priority;
+      entry.message  = message;
+      entry.router   = remote;
       itr_pair.first->second.push(std::move(entry));
 
       shouldCreateSession = itr_pair.second;
@@ -89,7 +91,14 @@ namespace llarp
   util::StatusObject
   OutboundMessageHandler::ExtractStatus() const
   {
-    util::StatusObject status{};
+    util::StatusObject status{"queueStats",
+                              {{"queued", m_queueStats.queued},
+                               {"dropped", m_queueStats.dropped},
+                               {"sent", m_queueStats.sent},
+                               {"queueWatermark", m_queueStats.queueWatermark},
+                               {"perTickMax", m_queueStats.perTickMax},
+                               {"numTicks", m_queueStats.numTicks}}};
+
     return status;
   }
 
@@ -201,6 +210,7 @@ namespace llarp
   {
     const llarp_buffer_t buf(msg.first);
     auto callback = msg.second;
+    m_queueStats.sent++;
     return _linkManager->SendTo(
         remote, buf, [=](ILinkSession::DeliveryStatus status) {
           if(status == ILinkSession::DeliveryStatus::eDeliverySuccess)
@@ -224,18 +234,29 @@ namespace llarp
   bool
   OutboundMessageHandler::QueueOutboundMessage(const RouterID &remote,
                                                Message &&msg,
-                                               const PathID_t &pathid)
+                                               const PathID_t &pathid,
+                                               uint16_t priority)
   {
     MessageQueueEntry entry;
     entry.message      = std::move(msg);
     auto callback_copy = entry.message.second;
     entry.router       = remote;
     entry.pathid       = pathid;
+    entry.priority     = priority;
     if(outboundQueue.tryPushBack(std::move(entry))
        != llarp::thread::QueueReturn::Success)
     {
+      m_queueStats.dropped++;
       DoCallback(callback_copy, SendStatus::Congestion);
     }
+    else
+    {
+      m_queueStats.queued++;
+
+      uint32_t queueSize = outboundQueue.size();
+      m_queueStats.queueWatermark =
+          std::max(queueSize, m_queueStats.queueWatermark);
+    }
 
     return true;
   }
@@ -257,11 +278,16 @@ namespace llarp
       }
 
       MessageQueue &path_queue = itr_pair.first->second;
-      if(path_queue.size() >= MAX_PATH_QUEUE_SIZE)
+
+      if(path_queue.size() < MAX_PATH_QUEUE_SIZE)
       {
-        path_queue.pop();  // head drop
+        path_queue.push(std::move(entry));
+      }
+      else
+      {
+        DoCallback(entry.message.second, SendStatus::Congestion);
+        m_queueStats.dropped++;
       }
-      path_queue.push(std::move(entry));
     }
   }
 
@@ -286,14 +312,15 @@ namespace llarp
   void
   OutboundMessageHandler::SendRoundRobin()
   {
+    m_queueStats.numTicks++;
+
     // send non-routing messages first priority
     auto &non_routing_mq = outboundMessageQueues[zeroID];
     while(not non_routing_mq.empty())
     {
-      MessageQueueEntry entry = std::move(non_routing_mq.front());
-      non_routing_mq.pop();
-
+      const MessageQueueEntry &entry = non_routing_mq.top();
       Send(entry.router, entry.message);
+      non_routing_mq.pop();
     }
 
     size_t empty_count = 0;
@@ -329,10 +356,11 @@ namespace llarp
       auto &message_queue = outboundMessageQueues[pathid];
       if(message_queue.size() > 0)
       {
-        MessageQueueEntry entry = std::move(message_queue.front());
-        message_queue.pop();
+        const MessageQueueEntry &entry = message_queue.top();
 
         Send(entry.router, entry.message);
+        message_queue.pop();
+
         empty_count = 0;
         sent_count++;
       }
@@ -349,6 +377,9 @@ namespace llarp
         break;
       }
     }
+
+    m_queueStats.perTickMax =
+        std::max((uint32_t)sent_count, m_queueStats.perTickMax);
   }
 
   void
@@ -372,8 +403,7 @@ namespace llarp
 
     while(!movedMessages.empty())
     {
-      MessageQueueEntry entry = std::move(movedMessages.front());
-      movedMessages.pop();
+      const MessageQueueEntry &entry = movedMessages.top();
 
       if(status == SendStatus::Success)
       {
@@ -383,6 +413,7 @@ namespace llarp
       {
         DoCallback(entry.message.second, status);
       }
+      movedMessages.pop();
     }
   }
 

llarp/router/outbound_message_handler.hpp

@@ -49,12 +49,30 @@ namespace llarp
 
     struct MessageQueueEntry
     {
+      uint16_t priority;
       Message message;
       PathID_t pathid;
       RouterID router;
+
+      bool
+      operator<(const MessageQueueEntry &other) const
+      {
+        return other.priority < priority;
+      }
+    };
+
+    struct MessageQueueStats
+    {
+      uint64_t queued         = 0;
+      uint64_t dropped        = 0;
+      uint64_t sent           = 0;
+      uint32_t queueWatermark = 0;
+
+      uint32_t perTickMax = 0;
+      uint32_t numTicks   = 0;
     };
 
-    using MessageQueue = std::queue< MessageQueueEntry >;
+    using MessageQueue = std::priority_queue< MessageQueueEntry >;
 
     void
     OnSessionEstablished(const RouterID &router);
@@ -91,7 +109,7 @@ namespace llarp
 
     bool
     QueueOutboundMessage(const RouterID &remote, Message &&msg,
-                         const PathID_t &pathid);
+                         const PathID_t &pathid, uint16_t priority = 0);
 
     void
     ProcessOutboundQueue();
@@ -128,6 +146,8 @@ namespace llarp
     // paths cannot have pathid "0", so it can be used as the "pathid"
     // for non-traffic (control) messages, so they can be prioritized.
     static const PathID_t zeroID;
+
+    MessageQueueStats m_queueStats;
   };
 
 }  // namespace llarp

llarp/router/rc_lookup_handler.cpp

@@ -1,3 +1,4 @@
+#include <chrono>
 #include <router/rc_lookup_handler.hpp>
 
 #include <link/i_link_manager.hpp>
@@ -16,6 +17,8 @@
 #include <functional>
 #include <random>
 
+using namespace std::chrono_literals;
+
 namespace llarp
 {
   void
@@ -53,7 +56,7 @@ namespace llarp
   RCLookupHandler::HaveReceivedWhitelist()
   {
     util::Lock l(&_mutex);
-    return whitelistRouters.empty();
+    return not whitelistRouters.empty();
   }
 
   void
@@ -113,6 +116,10 @@ namespace llarp
       {
         FinalizeRequest(router, nullptr, RCRequestResult::RouterNotFound);
       }
+      else
+      {
+        _routerLookupTimes[router] = std::chrono::steady_clock::now();
+      }
     }
   }
 
@@ -248,19 +255,24 @@ namespace llarp
 
     if(useWhitelist)
     {
-      static constexpr size_t LookupPerTick = 25;
+      static constexpr auto RerequestInterval = 10min;
+      static constexpr size_t LookupPerTick   = 5;
 
       std::vector< RouterID > lookupRouters;
       lookupRouters.reserve(LookupPerTick);
 
+      const auto now = std::chrono::steady_clock::now();
+
       {
         // if we are using a whitelist look up a few routers we don't have
         util::Lock l(&_mutex);
         for(const auto &r : whitelistRouters)
         {
-          if(_nodedb->Has(r))
-            continue;
-          lookupRouters.emplace_back(r);
+          if(now > _routerLookupTimes[r] + RerequestInterval
+             and not _nodedb->Has(r))
+          {
+            lookupRouters.emplace_back(r);
+          }
         }
       }
 

llarp/router/rc_lookup_handler.hpp

@@ -1,6 +1,7 @@
 #ifndef LLARP_RC_LOOKUP_HANDLER_HPP
 #define LLARP_RC_LOOKUP_HANDLER_HPP
 
+#include <chrono>
 #include <router/i_rc_lookup_handler.hpp>
 
 #include <util/thread/threading.hpp>
@@ -115,6 +116,10 @@ namespace llarp
     bool isServiceNode = false;
 
     std::set< RouterID > whitelistRouters GUARDED_BY(_mutex);
+
+    using TimePoint = std::chrono::steady_clock::time_point;
+    std::unordered_map< RouterID, TimePoint, RouterID::Hash >
+        _routerLookupTimes;
   };
 
 }  // namespace llarp

llarp/router/router.cpp

@@ -59,7 +59,8 @@ namespace llarp
 
     _stopping.store(false);
     _running.store(false);
-    _lastTick = llarp::time_now_ms();
+    _lastTick       = llarp::time_now_ms();
+    m_NextExploreAt = Clock_t::now();
   }
 
   Router::~Router()
@@ -78,7 +79,8 @@ namespace llarp
           {"dht", _dht->impl->ExtractStatus()},
           {"services", _hiddenServiceContext.ExtractStatus()},
           {"exit", _exitContext.ExtractStatus()},
-          {"links", _linkManager.ExtractStatus()}};
+          {"links", _linkManager.ExtractStatus()},
+          {"outboundMessages", _outboundMessageHandler.ExtractStatus()}};
     }
     else
     {
@@ -697,8 +699,13 @@ namespace llarp
       connected += _linkManager.NumberOfPendingConnections();
     }
 
-    _rcLookupHandler.ExploreNetwork();
-
+    const int interval       = isSvcNode ? 5 : 2;
+    const auto timepoint_now = Clock_t::now();
+    if(timepoint_now >= m_NextExploreAt)
+    {
+      _rcLookupHandler.ExploreNetwork();
+      m_NextExploreAt = timepoint_now + std::chrono::seconds(interval);
+    }
     size_t connectToNum      = _outboundSessionMaker.minConnectedRouters;
     const auto strictConnect = _rcLookupHandler.NumberOfStrictConnectRouters();
     if(strictConnect > 0 && connectToNum > strictConnect)

llarp/router/router.hpp

@@ -274,6 +274,11 @@ namespace llarp
     LinkManager _linkManager;
     RCLookupHandler _rcLookupHandler;
 
+    using Clock_t     = std::chrono::steady_clock;
+    using TimePoint_t = Clock_t::time_point;
+
+    TimePoint_t m_NextExploreAt;
+
     IOutboundMessageHandler &
     outboundMessageHandler() override
     {

llarp/router_contact.cpp

@@ -32,10 +32,11 @@ namespace llarp
   /// 1 day for real network
   llarp_time_t RouterContact::Lifetime = 24 * 60 * 60 * 1000;
 #endif
-  /// update RCs every 5 minutes
-  llarp_time_t RouterContact::UpdateInterval = 5 * 60 * 1000;
-  /// an RC inserted long enough ago (30 min) is considered stale and is removed
-  llarp_time_t RouterContact::StaleInsertionAge = 30 * 60 * 1000;
+  /// an RC inserted long enough ago (4 hrs) is considered stale and is removed
+  llarp_time_t RouterContact::StaleInsertionAge = 4 * 60 * 60 * 1000;
+  /// update RCs shortly before they are about to expire
+  llarp_time_t RouterContact::UpdateInterval =
+      RouterContact::StaleInsertionAge - (5 * 60 * 1000);
 
   NetID::NetID(const byte_t *val)
   {