remove mutex

llarp/service/sendcontext.cpp

@@ -11,6 +11,8 @@ namespace llarp
 {
   namespace service
   {
+    static constexpr size_t SendContextQueueSize = 512;
+
     SendContext::SendContext(ServiceInfo ident, const Introduction& intro,
                              path::PathSet* send, Endpoint* ep)
         : remoteIdent(std::move(ident))
@@ -18,6 +20,7 @@ namespace llarp
         , m_PathSet(send)
         , m_DataHandler(ep)
         , m_Endpoint(ep)
+        , m_SendQueue(SendContextQueueSize)
     {
       createdAt = ep->Now();
     }
@@ -25,11 +28,15 @@ namespace llarp
     bool
     SendContext::Send(std::shared_ptr< ProtocolFrame > msg, path::Path_ptr path)
     {
-      util::Lock lock(&m_SendQueueMutex);
-      m_SendQueue.emplace_back(
-          std::make_shared< const routing::PathTransferMessage >(
-              *msg, remoteIntro.pathID),
-          path);
+      if(m_SendQueue.empty() or m_SendQueue.full())
+      {
+        LogicCall(m_Endpoint->RouterLogic(),
+                  [self = this]() { self->FlushUpstream(); });
+      }
+      m_SendQueue.pushBack(
+          std::make_pair(std::make_shared< const routing::PathTransferMessage >(
+                             *msg, remoteIntro.pathID),
+                         path));
       return true;
     }
 
@@ -39,17 +46,19 @@ namespace llarp
       auto r = m_Endpoint->Router();
       std::unordered_set< path::Path_ptr, path::Path::Ptr_Hash > flushpaths;
       {
-        util::Lock lock(&m_SendQueueMutex);
-        for(const auto& item : m_SendQueue)
+        do
         {
+          auto maybe = m_SendQueue.tryPopFront();
+          if(not maybe.has_value())
+            break;
+          auto& item = maybe.value();
           if(item.second->SendRoutingMessage(*item.first, r))
           {
             lastGoodSend = r->Now();
             flushpaths.emplace(item.second);
             m_Endpoint->MarkConvoTagActive(item.first->T.T);
           }
-        }
-        m_SendQueue.clear();
+        } while(not m_SendQueue.empty());
       }
       // flush the select path's upstream
       for(const auto& path : flushpaths)

llarp/service/sendcontext.hpp

@@ -7,6 +7,7 @@
 #include <service/protocol.hpp>
 #include <util/buffer.hpp>
 #include <util/types.hpp>
+#include <util/thread/queue.hpp>
 
 #include <deque>
 
@@ -29,12 +30,11 @@ namespace llarp
       /// queue send a fully encrypted hidden service frame
       /// via a path
       bool
-      Send(std::shared_ptr< ProtocolFrame > f, path::Path_ptr path)
-          LOCKS_EXCLUDED(m_SendQueueMutex);
+      Send(std::shared_ptr< ProtocolFrame > f, path::Path_ptr path);
 
       /// flush upstream traffic when in router thread
       void
-      FlushUpstream() LOCKS_EXCLUDED(m_SendQueueMutex);
+      FlushUpstream();
 
       SharedSecret sharedKey;
       ServiceInfo remoteIdent;
@@ -51,8 +51,7 @@ namespace llarp
       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;
+      thread::Queue< SendEvent_t > m_SendQueue;
 
       virtual bool
       ShiftIntroduction(bool rebuild = true)