more

llarp/dns/server.hpp

@@ -15,7 +15,9 @@ namespace llarp
     /// handler of dns query hooking
     struct IQueryHandler
     {
-      virtual ~IQueryHandler(){}
+      virtual ~IQueryHandler()
+      {
+      }
 
       /// return true if we should hook this message
       virtual bool

llarp/service/endpoint.cpp

@@ -1122,6 +1122,17 @@ namespace llarp
       return false;
     }
 
+    void
+    Endpoint::Pump(llarp_time_t now)
+    {
+      auto itr = m_RemoteSessions.begin();
+      while(itr != m_RemoteSessions.end())
+      {
+        itr->second->Pump(now);
+        ++itr;
+      }
+    }
+
     bool
     Endpoint::SendToServiceOrQueue(const RouterID& addr,
                                    const llarp_buffer_t& data, ProtocolType t)

llarp/service/endpoint.hpp

@@ -182,6 +182,10 @@ namespace llarp
       bool
       LookupRouterAnon(RouterID router);
 
+      /// called on event loop pump
+      virtual void
+      Pump(llarp_time_t now);
+
       /// stop this endpoint
       bool
       Stop() override;

llarp/service/sendcontext.cpp

@@ -27,16 +27,36 @@ namespace llarp
     {
       auto transfer = std::make_shared< const routing::PathTransferMessage >(
           msg, remoteIntro.pathID);
-      m_Endpoint->RouterLogic()->queue_func([=]() {
-        if(path->SendRoutingMessage(*transfer, m_Endpoint->Router()))
+      {
+        util::Lock lock(&m_SendQueueMutex);
+        const auto sz = m_SendQueue.size();
+        m_SendQueue.emplace_back(transfer, path);
+        if(sz == 0)
+        {
+          // TODO: use shared_from_this()
+          m_Endpoint->RouterLogic()->queue_func(
+              std::bind(&SendContext::FlushSend, this));
+        }
+      }
+      return true;
+    }
+
+    void
+    SendContext::FlushSend()
+    {
+      auto r = m_Endpoint->Router();
+      util::Lock lock(&m_SendQueueMutex);
+      for(const auto& item : m_SendQueue)
+      {
+        if(item.second->SendRoutingMessage(*item.first, r))
         {
-          lastGoodSend = m_Endpoint->Now();
+          lastGoodSend = r->Now();
           ++sequenceNo;
         }
         else
           LogError("Failed to send frame on path");
-      });
-      return true;
+      }
+      m_SendQueue.clear();
     }
 
     /// send on an established convo tag

llarp/service/sendcontext.hpp

@@ -1,11 +1,12 @@
 #ifndef LLARP_SERVICE_SENDCONTEXT_HPP
 #define LLARP_SERVICE_SENDCONTEXT_HPP
-
+#include <messages/path_transfer.hpp>
 #include <path/pathset.hpp>
 #include <service/intro.hpp>
 #include <service/protocol.hpp>
 #include <util/buffer.hpp>
 #include <util/types.hpp>
+#include <deque>
 
 namespace llarp
 {
@@ -23,10 +24,15 @@ namespace llarp
       void
       AsyncEncryptAndSendTo(const llarp_buffer_t& payload, ProtocolType t);
 
-      /// send a fully encrypted hidden service frame
+      /// queue send a fully encrypted hidden service frame
       /// via a path
       bool
-      Send(const ProtocolFrame& f, path::Path_ptr path);
+      Send(const ProtocolFrame& f, path::Path_ptr path)
+          LOCKS_EXCLUDED(m_SendQueueMutex);
+
+      /// flush send when in router thread
+      void
+      FlushSend() LOCKS_EXCLUDED(m_SendQueueMutex);
 
       SharedSecret sharedKey;
       ServiceInfo remoteIdent;
@@ -41,6 +47,10 @@ namespace llarp
       llarp_time_t sendTimeout    = 40 * 1000;
       llarp_time_t connectTimeout = 60 * 1000;
       bool markedBad              = false;
+      using Msg_ptr     = std::shared_ptr< const routing::PathTransferMessage >;
+      using SendEvent_t = std::pair< Msg_ptr, path::Path_ptr >;
+      util::Mutex m_SendQueueMutex;
+      std::deque< SendEvent_t > m_SendQueue;
 
       virtual bool
       ShiftIntroduction(bool rebuild = true)

llarp/util/json.hpp

@@ -14,7 +14,9 @@ namespace llarp
   {
     struct IParser
     {
-      virtual ~IParser(){}
+      virtual ~IParser()
+      {
+      }
 
       /// result from feeding data to parser
       enum Result