Merge pull request #767 from majestrate/utp-write-split-2019-08-09

utp fixes

llarp/utp/linklayer.cpp

@@ -302,7 +302,7 @@ namespace llarp
       {
         if(arg->state == UTP_STATE_WRITABLE)
         {
-          session->PumpWrite();
+          session->Pump();
         }
         else if(arg->state == UTP_STATE_EOF)
         {

llarp/utp/session.cpp

@@ -20,7 +20,7 @@ namespace llarp
 
     /// pump tx queue
     void
-    Session::PumpWrite()
+    Session::PumpWrite(size_t numSend)
     {
       if(!sock)
         return;
@@ -31,10 +31,11 @@ namespace llarp
       {
         for(const auto& vec : msg.vecs)
         {
-          if(vec.iov_len)
+          if(vec.iov_len > 0)
           {
             expect += vec.iov_len;
             send.emplace_back(vec);
+            numSend--;
           }
         }
       }
@@ -147,10 +148,15 @@ namespace llarp
     Session::Tick(llarp_time_t now)
     {
       PruneInboundMessages(now);
-      m_TXRate = 0;
+      // ensure that this section is called every 1s or so
-      m_RXRate = 0;
+      if(now - m_LastTick >= 1000)
-      metrics::integerTick("utp.session.sendq", "size", sendq.size(), "id",
+      {
-                           RouterID(remoteRC.pubkey).ToString());
+        m_TXRate = 0;
+        m_RXRate = 0;
+        metrics::integerTick("utp.session.sendq", "size", sendq.size(), "id",
+                             RouterID(remoteRC.pubkey).ToString());
+        m_LastTick = now;
+      }
     }
 
     /// low level read
@@ -209,15 +215,29 @@ namespace llarp
         return now - lastActive > 5000;
       if(state == eSessionReady)
       {
-        // don't time out the connection if backlogged in downstream direction
+        const bool remoteIsSNode = remoteRC.IsPublicRouter();
-        // for clients dangling off the side of the network
+        const bool weAreSnode    = parent->GetOurRC().IsPublicRouter();
-        const auto timestamp =
+        const bool recvTimeout =
-            remoteRC.IsPublicRouter() && parent->GetOurRC().IsPublicRouter()
+            (now > lastActive) && now - lastActive > sessionTimeout;
-            ? lastSend
+        const bool sendTimeout =
-            : lastActive;
+            (now > lastSend) && now - lastSend > sessionTimeout;
-        if(now <= timestamp)
+        if(remoteIsSNode && weAreSnode)
-          return false;
+        {
-        return now - timestamp > 30000;
+          /// for s2s connections only check write direction
+          return sendTimeout;
+        }
+        else if(weAreSnode)
+        {
+          // for edge connection as service node check either direction for
+          // timeout
+          return recvTimeout || sendTimeout;
+        }
+        else
+        {
+          /// for edge connections as client we check if both directions have
+          /// been silent for 60s
+          return recvTimeout && sendTimeout;
+        }
       }
       if(state == eLinkEstablished)
         return now - lastActive
@@ -270,8 +290,10 @@ namespace llarp
     void
     Session::Pump()
     {
-      // pump write queue
+      if(sendq.size() > (MaxSendQueueSize / 4))
-      PumpWrite();
+        PumpWrite(sendq.size() / 2);
+      else
+        PumpWrite(sendq.size());
       // prune inbound messages
       PruneInboundMessages(parent->Now());
     }
@@ -284,7 +306,7 @@ namespace llarp
       if(SendQueueBacklog() >= MaxSendQueueSize)
       {
         // pump write queue if we seem to be full
-        PumpWrite();
+        PumpWrite(MaxSendQueueSize / 2);
       }
       if(SendQueueBacklog() >= MaxSendQueueSize)
       {
@@ -310,7 +332,7 @@ namespace llarp
         sz -= s;
       }
       if(state != eSessionReady)
-        PumpWrite();
+        PumpWrite(sendq.size());
       return true;
     }
 

llarp/utp/session.hpp

@@ -101,6 +101,8 @@ namespace llarp
       uint64_t m_RXRate = 0;
       uint64_t m_TXRate = 0;
 
+      llarp_time_t m_LastTick = 0;
+
       /// mark session as alive
       void
       Alive();
@@ -147,7 +149,7 @@ namespace llarp
 
       /// pump tx queue
       void
-      PumpWrite();
+      PumpWrite(size_t numMessages);
 
       void
       Pump() override;