From d0e9fe36dc6ae01c6c4c085adf6c30118fe5b25c Mon Sep 17 00:00:00 2001 From: Alexander Block Date: Fri, 11 Oct 2019 12:09:07 +0200 Subject: [PATCH] Replace vecAskFor with a priority queue 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. --- src/net.cpp | 21 +++++++++++---------- src/net.h | 4 +++- src/net_processing.cpp | 16 ++++++++++------ 3 files changed, 24 insertions(+), 17 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 5b37104b3477..e4bdd609188d 100644 --- a/src/net.cpp +++ b/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& 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) diff --git a/src/net.h b/src/net.h index 5b2cbd658d92..e6d1d9a9b2e7 100644 --- a/src/net.h +++ b/src/net.h @@ -32,6 +32,7 @@ #include #include #include +#include #ifndef WIN32 #include @@ -843,7 +844,8 @@ class CNode std::vector vInventoryOtherToSend; CCriticalSection cs_inventory; std::unordered_set setAskFor; - std::vector> vecAskFor; + std::unordered_set setAskForInQueue; + std::priority_queue, std::vector>, std::greater<>> queueAskFor; int64_t nNextInvSend; // Used for headers announcements - unfiltered blocks to relay // Also protected by cs_inventory diff --git a/src/net_processing.cpp b/src/net_processing.cpp index cc7fc68e3870..a93b04a6d846 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -3964,11 +3964,15 @@ bool PeerLogicValidation::SendMessages(CNode* pto, std::atomic& 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& 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());