Replace vecAskFor with a priority queue (#3147)

This avoids sorting before looping through it to figure out what to
request. The assumption that sorting would be cheap when vecAskFor is
already mostly sorted (only unsorted at the end) turned out to be false.
In reality, ~50% of CPU time was consumed by the sort when a lot of traffic
(thousands of TXs) happen.

Author: codablock

src/net.cpp

@@ -3221,11 +3221,11 @@ CNode::~CNode()
 
 void CNode::AskFor(const CInv& inv, int64_t doubleRequestDelay)
 {
-    if (vecAskFor.size() > MAPASKFOR_MAX_SZ || setAskFor.size() > SETASKFOR_MAX_SZ) {
+    if (queueAskFor.size() > MAPASKFOR_MAX_SZ || setAskFor.size() > SETASKFOR_MAX_SZ) {
         int64_t nNow = GetTime();
         if(nNow - nLastWarningTime > WARNING_INTERVAL) {
             LogPrintf("CNode::AskFor -- WARNING: inventory message dropped: vecAskFor.size = %d, setAskFor.size = %d, MAPASKFOR_MAX_SZ = %d, SETASKFOR_MAX_SZ = %d, nSkipped = %d, peer=%d\n",
-                      vecAskFor.size(), setAskFor.size(), MAPASKFOR_MAX_SZ, SETASKFOR_MAX_SZ, nNumWarningsSkipped, id);
+                      queueAskFor.size(), setAskFor.size(), MAPASKFOR_MAX_SZ, SETASKFOR_MAX_SZ, nNumWarningsSkipped, id);
             nLastWarningTime = nNow;
             nNumWarningsSkipped = 0;
         }
@@ -3235,10 +3235,10 @@ void CNode::AskFor(const CInv& inv, int64_t doubleRequestDelay)
         return;
     }
     // a peer may not have multiple non-responded queue positions for a single inv item
-    if (!setAskFor.insert(inv.hash).second)
+    if (!setAskFor.emplace(inv.hash).second)
         return;
 
-    // We're using vecAskFor as a priority queue,
+    // We're using queueAskFor as a priority queue,
     // the key is the earliest time the request can be sent
     int64_t nRequestTime;
     auto it = mapAlreadyAskedFor.find(inv.hash);
@@ -3262,16 +3262,17 @@ void CNode::AskFor(const CInv& inv, int64_t doubleRequestDelay)
         mapAlreadyAskedFor.update(it, nRequestTime);
     else
         mapAlreadyAskedFor.insert(std::make_pair(inv.hash, nRequestTime));
-    vecAskFor.emplace_back(nRequestTime, inv);
+
+    queueAskFor.emplace(nRequestTime, inv);
+    setAskForInQueue.emplace(inv.hash);
 }
 
 void CNode::RemoveAskFor(const uint256& hash)
 {
-    if (setAskFor.erase(hash)) {
-        vecAskFor.erase(std::remove_if(vecAskFor.begin(), vecAskFor.end(), [&](const std::pair<int64_t, CInv>& item) {
-            return item.second.hash == hash;
-        }), vecAskFor.end());
-    }
+    setAskFor.erase(hash);
+    // we don't really remove it from queueAskFor as it would be too expensive to rebuild the heap
+    // instead, we're ignoring the entry later as it won't be found in setAskForInQueue anymore
+    setAskForInQueue.erase(hash);
 }
 
 bool CConnman::NodeFullyConnected(const CNode* pnode)

src/net.h

@@ -32,6 +32,7 @@
 #include <memory>
 #include <condition_variable>
 #include <unordered_set>
+#include <queue>
 
 #ifndef WIN32
 #include <arpa/inet.h>
@@ -843,7 +844,8 @@ class CNode
     std::vector<CInv> vInventoryOtherToSend;
     CCriticalSection cs_inventory;
     std::unordered_set<uint256, StaticSaltedHasher> setAskFor;
-    std::vector<std::pair<int64_t, CInv>> vecAskFor;
+    std::unordered_set<uint256, StaticSaltedHasher> setAskForInQueue;
+    std::priority_queue<std::pair<int64_t, CInv>, std::vector<std::pair<int64_t, CInv>>, std::greater<>> queueAskFor;
     int64_t nNextInvSend;
     // Used for headers announcements - unfiltered blocks to relay
     // Also protected by cs_inventory

src/net_processing.cpp

@@ -3964,11 +3964,15 @@ bool PeerLogicValidation::SendMessages(CNode* pto, std::atomic<bool>& interruptM
         //
         // Message: getdata (non-blocks)
         //
-        std::sort(pto->vecAskFor.begin(), pto->vecAskFor.end());
-        auto it = pto->vecAskFor.begin();
-        while (it != pto->vecAskFor.end() && it->first <= nNow)
+        while (!pto->queueAskFor.empty() && pto->queueAskFor.top().first <= nNow)
         {
-            const CInv& inv = it->second;
+            const CInv& inv = pto->queueAskFor.top().second;
+            auto jt = pto->setAskForInQueue.find(inv.hash);
+            if (jt == pto->setAskForInQueue.end()) {
+                pto->queueAskFor.pop();
+                continue;
+            }
+
             if (!AlreadyHave(inv))
             {
                 LogPrint(BCLog::NET, "SendMessages -- GETDATA -- requesting inv = %s peer=%d\n", inv.ToString(), pto->GetId());
@@ -3984,9 +3988,9 @@ bool PeerLogicValidation::SendMessages(CNode* pto, std::atomic<bool>& interruptM
                 LogPrint(BCLog::NET, "SendMessages -- GETDATA -- already have inv = %s peer=%d\n", inv.ToString(), pto->GetId());
                 pto->setAskFor.erase(inv.hash);
             }
-            ++it;
+            pto->queueAskFor.pop();
+            pto->setAskForInQueue.erase(jt);
         }
-        pto->vecAskFor.erase(pto->vecAskFor.begin(), it);
         if (!vGetData.empty()) {
             connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData));
             LogPrint(BCLog::NET, "SendMessages -- GETDATA -- pushed size = %lu peer=%d\n", vGetData.size(), pto->GetId());