net: measure send buffer fullness based on memory usage

This more accurately captures the intent of limiting send buffer size, as
many small messages can have a larger overhead that is not counted with the
current approach.

It also means removing the dependency on the header size (which will become
a function of the transport choice) from the send buffer calculations.

Author: sipa

src/init.cpp

@@ -490,7 +490,7 @@ void SetupServerArgs(ArgsManager& argsman)
     argsman.AddArg("-listenonion", strprintf("Automatically create Tor onion service (default: %d)", DEFAULT_LISTEN_ONION), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-maxconnections=<n>", strprintf("Maintain at most <n> connections to peers (default: %u). This limit does not apply to connections manually added via -addnode or the addnode RPC, which have a separate limit of %u.", DEFAULT_MAX_PEER_CONNECTIONS, MAX_ADDNODE_CONNECTIONS), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-maxreceivebuffer=<n>", strprintf("Maximum per-connection receive buffer, <n>*1000 bytes (default: %u)", DEFAULT_MAXRECEIVEBUFFER), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
-    argsman.AddArg("-maxsendbuffer=<n>", strprintf("Maximum per-connection send buffer, <n>*1000 bytes (default: %u)", DEFAULT_MAXSENDBUFFER), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
+    argsman.AddArg("-maxsendbuffer=<n>", strprintf("Maximum per-connection memory usage for the send buffer, <n>*1000 bytes (default: %u)", DEFAULT_MAXSENDBUFFER), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-maxtimeadjustment", strprintf("Maximum allowed median peer time offset adjustment. Local perspective of time may be influenced by outbound peers forward or backward by this amount (default: %u seconds).", DEFAULT_MAX_TIME_ADJUSTMENT), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-maxuploadtarget=<n>", strprintf("Tries to keep outbound traffic under the given target per 24h. Limit does not apply to peers with 'download' permission or blocks created within past week. 0 = no limit (default: %s). Optional suffix units [k|K|m|M|g|G|t|T] (default: M). Lowercase is 1000 base while uppercase is 1024 base", DEFAULT_MAX_UPLOAD_TARGET), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);
     argsman.AddArg("-onion=<ip:port>", "Use separate SOCKS5 proxy to reach peers via Tor onion services, set -noonion to disable (default: -proxy)", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION);

src/net.cpp

@@ -19,6 +19,7 @@
 #include <crypto/sha256.h>
 #include <i2p.h>
 #include <logging.h>
+#include <memusage.h>
 #include <net_permissions.h>
 #include <netaddress.h>
 #include <netbase.h>
@@ -116,6 +117,14 @@ std::map<CNetAddr, LocalServiceInfo> mapLocalHost GUARDED_BY(g_maplocalhost_mute
 static bool vfLimited[NET_MAX] GUARDED_BY(g_maplocalhost_mutex) = {};
 std::string strSubVersion;
 
+size_t CSerializedNetMsg::GetMemoryUsage() const noexcept
+{
+    // Don't count the dynamic memory used for the m_type string, by assuming it fits in the
+    // "small string" optimization area (which stores data inside the object itself, up to some
+    // size; 15 bytes in modern libstdc++).
+    return sizeof(*this) + memusage::DynamicUsage(data);
+}
+
 void CConnman::AddAddrFetch(const std::string& strDest)
 {
     LOCK(m_addr_fetches_mutex);
@@ -894,6 +903,14 @@ void V1Transport::MarkBytesSent(size_t bytes_sent) noexcept
     }
 }
 
+size_t V1Transport::GetSendMemoryUsage() const noexcept
+{
+    AssertLockNotHeld(m_send_mutex);
+    LOCK(m_send_mutex);
+    // Don't count sending-side fields besides m_message_to_send, as they're all small and bounded.
+    return m_message_to_send.GetMemoryUsage();
+}
+
 std::pair<size_t, bool> CConnman::SocketSendData(CNode& node) const
 {
     auto it = node.vSendMsg.begin();
@@ -923,8 +940,8 @@ std::pair<size_t, bool> CConnman::SocketSendData(CNode& node) const
             nSentSize += nBytes;
             if (node.nSendOffset == data.size()) {
                 node.nSendOffset = 0;
-                node.nSendSize -= data.size();
-                node.fPauseSend = node.nSendSize > nSendBufferMaxSize;
+                // Update memory usage of send buffer (as *it will be deleted).
+                node.m_send_memusage -= sizeof(data) + memusage::DynamicUsage(data);
                 it++;
             } else {
                 // could not send full message; stop sending more
@@ -944,9 +961,11 @@ std::pair<size_t, bool> CConnman::SocketSendData(CNode& node) const
         }
     }
 
+    node.fPauseSend = node.m_send_memusage + node.m_transport->GetSendMemoryUsage() > nSendBufferMaxSize;
+
     if (it == node.vSendMsg.end()) {
         assert(node.nSendOffset == 0);
-        assert(node.nSendSize == 0);
+        assert(node.m_send_memusage == 0);
     }
     node.vSendMsg.erase(node.vSendMsg.begin(), it);
     return {nSentSize, !node.vSendMsg.empty()};
@@ -2985,14 +3004,21 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
             if (bytes.empty()) break;
             // Update statistics per message type.
             pnode->AccountForSentBytes(msg_type, bytes.size());
-            // Update number of bytes in the send buffer.
-            pnode->nSendSize += bytes.size();
-            if (pnode->nSendSize > nSendBufferMaxSize) pnode->fPauseSend = true;
             pnode->vSendMsg.push_back({bytes.begin(), bytes.end()});
+            // Update memory usage of send buffer. For now, use static + dynamic memory usage of
+            // byte vectors in vSendMsg as send memory. In a future commit, vSendMsg will be
+            // replaced with a queue of CSerializedNetMsg objects, and we'll use their memory usage
+            // instead.
+            pnode->m_send_memusage += sizeof(pnode->vSendMsg.back()) + memusage::DynamicUsage(pnode->vSendMsg.back());
             // Notify transport that bytes have been processed (they're not actually sent yet,
             // but pushed onto the vSendMsg queue of bytes to send).
             pnode->m_transport->MarkBytesSent(bytes.size());
         }
+        // At this point, m_transport->GetSendMemoryUsage() isn't very interesting as the
+        // transport's message is fully flushed (and converted to byte arrays). It's still included
+        // here for correctness, and will become relevant in a future commit when a queued message
+        // inside the transport may survive PushMessage calls.
+        if (pnode->m_send_memusage + pnode->m_transport->GetSendMemoryUsage() > nSendBufferMaxSize) pnode->fPauseSend = true;
 
         // If the write queue was empty before and isn't now, attempt "optimistic write":
         // because the poll/select loop may pause for SELECT_TIMEOUT_MILLISECONDS before actually

src/net.h

@@ -122,6 +122,9 @@ struct CSerializedNetMsg {
 
     std::vector<unsigned char> data;
     std::string m_type;
+
+    /** Compute total memory usage of this object (own memory + any dynamic memory). */
+    size_t GetMemoryUsage() const noexcept;
 };
 
 /**
@@ -313,6 +316,9 @@ class Transport {
      * If bytes_sent=0, this call has no effect.
      */
     virtual void MarkBytesSent(size_t bytes_sent) noexcept = 0;
+
+    /** Return the memory usage of this transport attributable to buffered data to send. */
+    virtual size_t GetSendMemoryUsage() const noexcept = 0;
 };
 
 class V1Transport final : public Transport
@@ -399,6 +405,7 @@ class V1Transport final : public Transport
     bool SetMessageToSend(CSerializedNetMsg& msg) noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex);
     BytesToSend GetBytesToSend() const noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex);
     void MarkBytesSent(size_t bytes_sent) noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex);
+    size_t GetSendMemoryUsage() const noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex);
 };
 
 struct CNodeOptions
@@ -429,8 +436,9 @@ class CNode
      */
     std::shared_ptr<Sock> m_sock GUARDED_BY(m_sock_mutex);
 
-    /** Total size of all vSendMsg entries */
-    size_t nSendSize GUARDED_BY(cs_vSend){0};
+    /** Total memory usage of vSendMsg (counting the vectors and their dynamic usage, but not the
+     *  deque overhead). */
+    size_t m_send_memusage GUARDED_BY(cs_vSend){0};
     /** Offset inside the first vSendMsg already sent */
     size_t nSendOffset GUARDED_BY(cs_vSend){0};
     uint64_t nSendBytes GUARDED_BY(cs_vSend){0};