From 78829822c888853261105d7dac53bc96cfec83fb Mon Sep 17 00:00:00 2001 From: pasta Date: Tue, 8 Sep 2026 15:49:03 -0500 Subject: [PATCH] perf(net): isolate object request scheduling from cs_main Give the object request tracker its own mutex so that inventory selection, request-state transitions, NOTFOUND handling, and disconnect cleanup no longer serialize behind cs_main. Block scheduling, availability checks, and per-peer response authorization stay under cs_main; when both locks are needed, cs_main is taken first. In SendMessages the getdata loop takes cs_main once to partition the requestable set by AlreadyHave, then takes the tracker mutex once to apply RequestedTx/ForgetTxHash for the batch, instead of acquiring both locks per inventory. GetRequestedObjectCount only touches the tracker, so it no longer requires cs_main. The governance vote sync path reads it without pinning cs_main while waiting on the tracker. Add native InventoryBatch benchmarks, an accounting test, and a regression test proving NOTFOUND completes while another thread holds cs_main. --- doc/benchmarking.md | 11 ++ src/Makefile.bench.include | 1 + src/bench/net_processing.cpp | 56 ++++++++++ src/net_processing.cpp | 174 ++++++++++++++++++------------ src/node/sync_manager.cpp | 6 +- src/test/net_tests.cpp | 111 +++++++++++++++++-- test/util/data/non-backported.txt | 1 + 7 files changed, 275 insertions(+), 85 deletions(-) create mode 100644 src/bench/net_processing.cpp diff --git a/doc/benchmarking.md b/doc/benchmarking.md index e02563b61c66..0ea678033388 100644 --- a/doc/benchmarking.md +++ b/doc/benchmarking.md @@ -53,6 +53,17 @@ More benchmarks are needed for, in no particular order: - Cuckoo Cache - P2P throughput +Inventory processing +-------------------- + +`InventoryBatch100` and `InventoryBatch50000` exercise the peer-manager INV, +GETDATA scheduling, and NOTFOUND paths without sockets. They use spork +inventories whose hashes are absent locally and clear the request state after +every iteration. Use an optimized build to measure CPU cost; debug builds +additionally enable expensive container and lock-order checks. + + src/bench/bench_dash -filter='InventoryBatch.*' -min-time=5000 + Going Further -------------------- diff --git a/src/Makefile.bench.include b/src/Makefile.bench.include index 7dec51f2b0c1..5b929dc9c51e 100644 --- a/src/Makefile.bench.include +++ b/src/Makefile.bench.include @@ -47,6 +47,7 @@ bench_bench_dash_SOURCES = \ bench/merkle_root.cpp \ bench/nanobench.cpp \ bench/nanobench.h \ + bench/net_processing.cpp \ bench/peer_eviction.cpp \ bench/poly1305.cpp \ bench/pool.cpp \ diff --git a/src/bench/net_processing.cpp b/src/bench/net_processing.cpp new file mode 100644 index 000000000000..1f6c7c497020 --- /dev/null +++ b/src/bench/net_processing.cpp @@ -0,0 +1,56 @@ +// Copyright (c) 2026 The Dash Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#include +#include +#include +#include +#include +#include +#include + +using namespace std::literals; + +static void InventoryBatch(benchmark::Bench& bench, uint32_t count) +{ + const auto setup = MakeNoLogFileContext(); + auto& chainstate = *static_cast(&setup->m_node.chainman->ActiveChainstate()); + chainstate.JumpOutOfIbd(); + auto& peerman = *setup->m_node.peerman; + const auto& connman = *static_cast(setup->m_node.connman.get()); + auto peer{MakeTestPeer(/*id=*/0)}; + peerman.InitializeNode(*peer, NODE_NETWORK); + + std::vector invs; + invs.reserve(count); + for (uint32_t i = 1; i <= count; ++i) { + invs.emplace_back(MSG_SPORK, ArithToUint256(arith_uint256{i})); + } + CDataStream inventory{SER_NETWORK, PROTOCOL_VERSION}; + inventory << invs; + const std::atomic interrupt{false}; + SetMockTime(1'700'000'000s); + const auto now{GetTime()}; + + bench.batch(count).unit("inventory").run([&] { + LOCK(NetEventsInterface::g_msgproc_mutex); + auto announcements = inventory; + peerman.ProcessMessage(*peer, NetMsgType::INV, announcements, now, interrupt); + peerman.SendMessages(peer.get()); + connman.FlushSendBuffer(*peer); + auto notfound = inventory; + peerman.ProcessMessage(*peer, NetMsgType::NOTFOUND, notfound, now, interrupt); + }); + + peerman.FinalizeNode(*peer); + chainstate.ResetIbd(); + SetMockTime(0s); +} + +static void InventoryBatch100(benchmark::Bench& bench) { InventoryBatch(bench, 100); } + +static void InventoryBatch50000(benchmark::Bench& bench) { InventoryBatch(bench, 50'000); } + +BENCHMARK(InventoryBatch100, benchmark::PriorityLevel::HIGH); +BENCHMARK(InventoryBatch50000, benchmark::PriorityLevel::HIGH); diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 62071ca1dcf1..dbf210a77465 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -583,7 +583,7 @@ class PeerManagerImpl final : public PeerManager /** Overridden from CValidationInterface. */ void BlockConnected(const std::shared_ptr& pblock, const CBlockIndex* pindexConnected) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_recent_confirmed_transactions_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex, !m_recent_confirmed_transactions_mutex); void BlockDisconnected(const std::shared_ptr &block, const CBlockIndex* pindex) override EXCLUSIVE_LOCKS_REQUIRED(!m_recent_confirmed_transactions_mutex); void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) override @@ -594,12 +594,15 @@ class PeerManagerImpl final : public PeerManager EXCLUSIVE_LOCKS_REQUIRED(!m_most_recent_block_mutex); /** Implement NetEventsInterface */ - void InitializeNode(CNode& node, ServiceFlags our_services) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - void FinalizeNode(const CNode& node) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void InitializeNode(CNode& node, ServiceFlags our_services) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); + void FinalizeNode(const CNode& node) override EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); bool ProcessMessages(CNode* pfrom, std::atomic& interrupt) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_recent_confirmed_transactions_mutex, !m_most_recent_block_mutex, g_msgproc_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex, !m_recent_confirmed_transactions_mutex, + !m_most_recent_block_mutex, g_msgproc_mutex); bool SendMessages(CNode* pto) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_recent_confirmed_transactions_mutex, !m_most_recent_block_mutex, g_msgproc_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex, !m_recent_confirmed_transactions_mutex, + !m_most_recent_block_mutex, g_msgproc_mutex); /** Implement PeerManager */ void StartScheduledTasks(CScheduler& scheduler) override; @@ -619,10 +622,11 @@ class PeerManagerImpl final : public PeerManager void UnitTestMisbehaving(NodeId peer_id, int howmuch) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex) { Misbehaving(*Assert(GetPeerRef(peer_id)), howmuch, ""); }; void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv, const std::chrono::microseconds time_received, const std::atomic& interruptMsgProc) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_recent_confirmed_transactions_mutex, !m_most_recent_block_mutex, g_msgproc_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex, !m_recent_confirmed_transactions_mutex, + !m_most_recent_block_mutex, g_msgproc_mutex); void UpdateLastBlockAnnounceTime(NodeId node, int64_t time_in_seconds) override; bool IsBanned(NodeId pnode) override EXCLUSIVE_LOCKS_REQUIRED(cs_main, !m_peer_mutex); - size_t GetRequestedObjectCount(NodeId nodeid) const override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + size_t GetRequestedObjectCount(NodeId nodeid) const override EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex); /** Implements external handlers logic */ void AddExtraHandler(std::unique_ptr&& handler) override; @@ -635,10 +639,13 @@ class PeerManagerImpl final : public PeerManager /** Implement PeerManagerInternal */ void PeerMisbehaving(const NodeId pnode, const int howmuch, const std::string& message = "") override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); bool PeerIsBanned(const NodeId node_id) override EXCLUSIVE_LOCKS_REQUIRED(cs_main, !m_peer_mutex); - void PeerEraseObjectRequest(const NodeId nodeid, const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); - bool PeerConsumeObjectRequest(NodeId nodeid, const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); - GetDataResponse PeerConsumeGetDataResponse(NodeId nodeid, const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); - void PeerForgetObjectRequest(const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + void PeerEraseObjectRequest(const NodeId nodeid, const CInv& inv) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, ::cs_main); + bool PeerConsumeObjectRequest(NodeId nodeid, const CInv& inv) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, ::cs_main); + GetDataResponse PeerConsumeGetDataResponse(NodeId nodeid, const CInv& inv) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, ::cs_main); + void PeerForgetObjectRequest(const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, ::cs_main); void PeerPushInventory(NodeId nodeid, const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayInv(const CInv& inv) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayInvFiltered(const CInv& inv, const CTransaction& relatedTx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); @@ -646,15 +653,18 @@ class PeerManagerImpl final : public PeerManager void PeerRelayDSQ(const CCoinJoinQueue& queue) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayTransaction(const uint256& txid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void PeerRelayRecoveredSig(const llmq::CRecoveredSig& sig, bool proactive_relay) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - void PeerAskPeersForTransaction(const uint256& txid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - size_t PeerGetRequestedObjectCount(NodeId nodeid) const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, ::cs_main); - void PeerPostProcessMessage(MessageProcessingResult&& ret) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void PeerAskPeersForTransaction(const uint256& txid) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); + size_t PeerGetRequestedObjectCount(NodeId nodeid) const override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); + void PeerPostProcessMessage(MessageProcessingResult&& ret) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); private: void _RelayTransaction(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(cs_main, !m_peer_mutex); /** Ask peers that have a transaction in their inventory to relay it to us. */ - void AskPeersForTransaction(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void AskPeersForTransaction(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); /** Relay inventories to peers that find it relevant */ void RelayInvFiltered(const CInv& inv, const CTransaction& relatedTx) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); @@ -668,14 +678,15 @@ class PeerManagerImpl final : public PeerManager /** Register with m_object_request that an inv has been received from a peer, computing the * request delay from the peer's preferredness and in-flight load. */ void AddObjectAnnouncement(const CNode& node, const CInv& inv, std::chrono::microseconds current_time) - EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, ::cs_main); /** Delete all announcements of a transaction across all peers, under both inv types it may * have been announced with (MSG_TX and MSG_DSTX). */ - void ForgetTx(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + void ForgetTx(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, ::cs_main); /** Helper to process result of external handlers of message */ - void PostProcessMessage(MessageProcessingResult&& ret, NodeId node) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void PostProcessMessage(MessageProcessingResult&& ret, NodeId node) override + EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex); /** Consider evicting an outbound peer based on the amount of time they've been behind our tip */ void ConsiderEviction(CNode& pto, Peer& peer, std::chrono::seconds time_in_seconds) EXCLUSIVE_LOCKS_REQUIRED(cs_main, g_msgproc_mutex); @@ -745,8 +756,7 @@ class PeerManagerImpl final : public PeerManager * reconsidered. * @return True if there are still orphans in this peer's work set. */ - bool ProcessOrphanTx(NodeId node_id) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, cs_main); + bool ProcessOrphanTx(NodeId node_id) EXCLUSIVE_LOCKS_REQUIRED(!m_object_request_mutex, !m_peer_mutex, cs_main); /** Process a single headers message from a peer. */ void ProcessHeadersMessage(CNode& pfrom, Peer& peer, const std::vector& headers, @@ -1095,8 +1105,10 @@ class PeerManagerImpl final : public PeerManager /** Tracks announced inventories (transactions and all Dash-specific object types), and which * peer to request them from next. All policy (preferredness, delays, per-type expiry) is - * decided by the callers; see AddObjectAnnouncement and the getdata section of SendMessages. */ - TxRequestTracker m_object_request GUARDED_BY(::cs_main); + * decided by the callers; see AddObjectAnnouncement and the getdata section of SendMessages. + * Acquire cs_main before m_object_request_mutex when both are needed. */ + mutable Mutex m_object_request_mutex; + TxRequestTracker m_object_request GUARDED_BY(m_object_request_mutex); void AddToCompactExtraTransactions(const CTransactionRef& tx) EXCLUSIVE_LOCKS_REQUIRED(g_msgproc_mutex); @@ -1616,6 +1628,7 @@ void PeerManagerImpl::AddObjectAnnouncement(const CNode& node, const CInv& inv, const CNodeState* state = State(node.GetId()); if (state == nullptr) return; + LOCK(m_object_request_mutex); if (m_object_request.Count(node.GetId()) >= MAX_PEER_OBJECT_ANNOUNCEMENTS) { // Too many queued announcements from this peer return; @@ -1642,13 +1655,14 @@ void PeerManagerImpl::AddObjectAnnouncement(const CNode& node, const CInv& inv, void PeerManagerImpl::ForgetTx(const uint256& txid) { AssertLockHeld(cs_main); + LOCK(m_object_request_mutex); m_object_request.ForgetTxHash(CInv(MSG_TX, txid)); m_object_request.ForgetTxHash(CInv(MSG_DSTX, txid)); } size_t PeerManagerImpl::GetRequestedObjectCount(NodeId nodeid) const { - AssertLockHeld(cs_main); + LOCK(m_object_request_mutex); return m_object_request.Count(nodeid); } @@ -1711,7 +1725,7 @@ void PeerManagerImpl::InitializeNode(CNode& node, ServiceFlags our_services) { { LOCK(cs_main); m_node_states.emplace_hint(m_node_states.end(), std::piecewise_construct, std::forward_as_tuple(nodeid), std::forward_as_tuple(node.IsInboundConn())); - assert(m_object_request.Count(nodeid) == 0); + assert(WITH_LOCK(m_object_request_mutex, return m_object_request.Count(nodeid)) == 0); } PeerRef peer = std::make_shared(nodeid, our_services); { @@ -1777,7 +1791,7 @@ void PeerManagerImpl::FinalizeNode(const CNode& node) { } } m_orphanage.EraseForPeer(nodeid); - m_object_request.DisconnectedPeer(nodeid); + WITH_LOCK(m_object_request_mutex, m_object_request.DisconnectedPeer(nodeid)); if (m_txreconciliation) m_txreconciliation->ForgetPeer(nodeid); m_num_preferred_download_peers -= state->fPreferredDownload; m_peers_downloading_from -= (!state->vBlocksInFlight.empty()); @@ -1794,7 +1808,7 @@ void PeerManagerImpl::FinalizeNode(const CNode& node) { assert(m_peers_downloading_from == 0); assert(m_outbound_peers_with_protect_from_disconnect == 0); assert(m_orphanage.Size() == 0); - assert(m_object_request.Size() == 0); + assert(WITH_LOCK(m_object_request_mutex, return m_object_request.Size()) == 0); } } // cs_main @@ -2450,7 +2464,8 @@ void PeerManagerImpl::AskPeersForTransaction(const uint256& txid) LogPrintf("PeerManagerImpl::%s -- txid=%s: asking other peer %d for correct TX\n", __func__, txid.ToString(), peer->m_id); - m_object_request.ReceivedInv(peer->m_id, CInv(MSG_TX, txid), /*preferred=*/true, current_time); + WITH_LOCK(m_object_request_mutex, + m_object_request.ReceivedInv(peer->m_id, CInv(MSG_TX, txid), /*preferred=*/true, current_time)); } } } @@ -3750,7 +3765,7 @@ void PeerManagerImpl::PostProcessMessage(MessageProcessingResult&& result, NodeI if (peer) Misbehaving(*peer, result.m_error->score, result.m_error->message); } if (result.m_to_erase) { - WITH_LOCK(cs_main, m_object_request.ReceivedResponse(node, result.m_to_erase.value())); + WITH_LOCK(m_object_request_mutex, m_object_request.ReceivedResponse(node, result.m_to_erase.value())); } for (const auto& tx : result.m_transactions) { WITH_LOCK(cs_main, _RelayTransaction(tx)); @@ -3758,7 +3773,7 @@ void PeerManagerImpl::PostProcessMessage(MessageProcessingResult&& result, NodeI for (const auto& inv : result.m_inventory) { // An inv being relayed is available locally, so there is no need to request it from // anyone anymore. - WITH_LOCK(cs_main, m_object_request.ForgetTxHash(inv)); + WITH_LOCK(m_object_request_mutex, m_object_request.ForgetTxHash(inv)); RelayInv(inv); } } @@ -4813,6 +4828,7 @@ void PeerManagerImpl::ProcessMessage( // A MSG_TX request may be answered with a DSTX message and vice versa (a getdata for // either type serves the underlying transaction), so complete whichever announcement // type the request was tracked under. + LOCK(m_object_request_mutex); m_object_request.ReceivedResponse(pfrom.GetId(), CInv(MSG_TX, txid)); m_object_request.ReceivedResponse(pfrom.GetId(), CInv(MSG_DSTX, txid)); } @@ -5601,13 +5617,13 @@ void PeerManagerImpl::ProcessMessage( uint256 hash = spork.GetHash(); CInv spork_inv{MSG_SPORK, hash}; - WITH_LOCK(::cs_main, m_object_request.ReceivedResponse(pfrom.GetId(), spork_inv)); + WITH_LOCK(m_object_request_mutex, m_object_request.ReceivedResponse(pfrom.GetId(), spork_inv)); if (!m_sporkman.IsValidSpork(spork)) { Misbehaving(*peer, 100, strprintf("invalid spork received. peer=%d", pfrom.GetId())); return; } if (m_sporkman.ProcessSpork(spork, strprintf(" peer=%d", pfrom.GetId()))) { - WITH_LOCK(::cs_main, m_object_request.ForgetTxHash(spork_inv)); + WITH_LOCK(m_object_request_mutex, m_object_request.ForgetTxHash(spork_inv)); RelayInv(spork_inv); } return; @@ -5655,7 +5671,7 @@ void PeerManagerImpl::ProcessMessage( return; } - LOCK(cs_main); + LOCK(m_object_request_mutex); for (CInv &inv : vInv) { if (inv.IsKnownType()) { // If we receive a NOTFOUND message for an inv we requested, mark the announcement @@ -6232,6 +6248,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto) MaybeSendAddr(*pto, *peer, current_time); + std::vector vGetData; { LOCK(cs_main); @@ -6717,7 +6734,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto) // // Message: getdata (blocks) // - std::vector vGetData; if (CanServeBlocks(*peer) && pto->CanRelay() && ((sync_blocks_and_headers_from_peer && !IsLimitedPeer(*peer)) || !m_chainman.ActiveChainstate().IsInitialBlockDownload()) && state.vBlocksInFlight.size() < MAX_BLOCKS_IN_TRANSIT_PER_PEER) { std::vector vToDownload; NodeId staller = -1; @@ -6735,49 +6751,65 @@ bool PeerManagerImpl::SendMessages(CNode* pto) } } } + } // release cs_main - // - // Message: getdata (non-blocks) - // + // + // Message: getdata (non-blocks) + // - // DASH unlike Bitcoin, this loop requests all Dash-specific object types too. The request - // expiry doubles as the fallback-to-another-peer trigger, so time-sensitive object types - // use a shorter per-type interval (see GetObjectInterval). - std::vector> expired; - auto requestable = m_object_request.GetRequestable(pto->GetId(), current_time, &expired); - for (const auto& entry : expired) { - LogPrint(BCLog::NET, "timeout of inflight object %s from peer=%d\n", entry.second.ToString(), entry.first); - } + // DASH unlike Bitcoin, this loop requests all Dash-specific object types too. The request + // expiry doubles as the fallback-to-another-peer trigger, so time-sensitive object types + // use a shorter per-type interval (see GetObjectInterval). + std::vector> expired; + auto requestable = WITH_LOCK(m_object_request_mutex, + return m_object_request.GetRequestable(pto->GetId(), current_time, &expired)); + for (const auto& entry : expired) { + LogPrint(BCLog::NET, "timeout of inflight object %s from peer=%d\n", entry.second.ToString(), entry.first); + } + // Availability checks need cs_main, tracker updates need m_object_request_mutex. Take each + // once for the whole batch rather than once per inventory. + std::vector requested; + std::vector already_have; + { + LOCK(cs_main); + CNodeState& state{*Assert(State(pto->GetId()))}; for (const CInv& inv : requestable) { - if (!AlreadyHave(inv)) { - LogPrint(BCLog::NET, "Requesting %s peer=%d\n", inv.ToString(), pto->GetId()); - vGetData.push_back(inv); - if (vGetData.size() >= MAX_GETDATA_SZ) { - m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); - vGetData.clear(); - } - m_object_request.RequestedTx(pto->GetId(), inv, current_time + GetObjectInterval(inv.type)); - if (IsGetDataOnlyObject(inv.type)) { - // Remember that we asked, so that an answer arriving after the tracker entry is - // gone -- expired, or erased because the object turned up elsewhere -- is not - // mistaken for an unsolicited push. See GetDataResponse. - state.m_recent_object_requests.insert(inv.hash, - RequestedObject{inv.type, current_time}); - } - } else { + if (AlreadyHave(inv)) { // We have already seen this object, no need to download. This is for belated // announcements of objects which arrived via another peer; the tracker has no // direct means to remove them once the object is received elsewhere. - m_object_request.ForgetTxHash(inv); + already_have.push_back(inv); + continue; + } + LogPrint(BCLog::NET, "Requesting %s peer=%d\n", inv.ToString(), pto->GetId()); + vGetData.push_back(inv); + requested.push_back(inv); + if (vGetData.size() >= MAX_GETDATA_SZ) { + m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); + vGetData.clear(); + } + if (IsGetDataOnlyObject(inv.type)) { + // Remember that we asked, so that an answer arriving after the tracker entry is + // gone -- expired, or erased because the object turned up elsewhere -- is not + // mistaken for an unsolicited push. See GetDataResponse. + state.m_recent_object_requests.insert(inv.hash, RequestedObject{inv.type, current_time}); } } - - - if (!vGetData.empty()) { - m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); - LogPrint(BCLog::NET, "SendMessages -- GETDATA -- pushed size = %lu peer=%d\n", vGetData.size(), pto->GetId()); + } + { + LOCK(m_object_request_mutex); + for (const CInv& inv : requested) { + m_object_request.RequestedTx(pto->GetId(), inv, current_time + GetObjectInterval(inv.type)); } - } // release cs_main + for (const CInv& inv : already_have) { + m_object_request.ForgetTxHash(inv); + } + } + + if (!vGetData.empty()) { + m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); + LogPrint(BCLog::NET, "SendMessages -- GETDATA -- pushed size = %lu peer=%d\n", vGetData.size(), pto->GetId()); + } return true; } @@ -6797,18 +6829,18 @@ void PeerManagerImpl::PeerEraseObjectRequest(const NodeId nodeid, const CInv& in // Completing only this peer's announcement is deliberate: an invalid or unusable object must // not stop us from fetching it from honest peers. Cleanup across peers happens once the object // is accepted and AlreadyHave(inv) turns true. - m_object_request.ReceivedResponse(nodeid, inv); + WITH_LOCK(m_object_request_mutex, m_object_request.ReceivedResponse(nodeid, inv)); } bool PeerManagerImpl::PeerConsumeObjectRequest(NodeId nodeid, const CInv& inv) { - return m_object_request.ReceivedResponse(nodeid, inv); + return WITH_LOCK(m_object_request_mutex, return m_object_request.ReceivedResponse(nodeid, inv)); } GetDataResponse PeerManagerImpl::PeerConsumeGetDataResponse(NodeId nodeid, const CInv& inv) { CNodeState* state = State(nodeid); - if (m_object_request.ReceivedRequestedResponse(nodeid, inv)) { + if (WITH_LOCK(m_object_request_mutex, return m_object_request.ReceivedRequestedResponse(nodeid, inv))) { // Answered on time. Spend the late-answer grace too, so the GETDATA cannot also pay for a // replay of the same payload. if (state != nullptr) state->m_recent_object_requests.erase(inv.hash); @@ -6838,7 +6870,7 @@ GetDataResponse PeerManagerImpl::PeerConsumeGetDataResponse(NodeId nodeid, const void PeerManagerImpl::PeerForgetObjectRequest(const CInv& inv) { - m_object_request.ForgetTxHash(inv); + WITH_LOCK(m_object_request_mutex, m_object_request.ForgetTxHash(inv)); } void PeerManagerImpl::PeerPushInventory(NodeId nodeid, const CInv& inv) diff --git a/src/node/sync_manager.cpp b/src/node/sync_manager.cpp index 6d6d53259e36..4eb1598e284f 100644 --- a/src/node/sync_manager.cpp +++ b/src/node/sync_manager.cpp @@ -111,11 +111,7 @@ int SyncManager::RequestGovernanceObjectVotes(const std::vector& vNodesC // initiated from another node, so skip it too. if (!pnode->CanRelay() || (m_connman.IsActiveMasternode() && pnode->IsInboundConn())) continue; // stop early to prevent setAskFor overflow - { - LOCK(::cs_main); - size_t nProjectedSize = m_peer_manager->PeerGetRequestedObjectCount(pnode->GetId()) + nProjectedVotes; - if (nProjectedSize > MAX_INV_SZ) continue; - } + if (m_peer_manager->PeerGetRequestedObjectCount(pnode->GetId()) + nProjectedVotes > MAX_INV_SZ) continue; // to early to ask the same node if (mapAskedRecently[nHashGovobj].count(pnode->addr)) continue; diff --git a/src/test/net_tests.cpp b/src/test/net_tests.cpp index 84e727b5df5b..7125a674dfda 100644 --- a/src/test/net_tests.cpp +++ b/src/test/net_tests.cpp @@ -4,6 +4,7 @@ #include +#include #include #include #include @@ -29,6 +30,7 @@ #include #include +#include #include #include #include @@ -73,6 +75,97 @@ BOOST_AUTO_TEST_CASE(cnode_listen_port) BOOST_CHECK(port == altPort); } +BOOST_AUTO_TEST_CASE(inventory_request_accounting) +{ + LOCK(NetEventsInterface::g_msgproc_mutex); + auto& chainstate = *static_cast(&m_node.chainman->ActiveChainstate()); + chainstate.JumpOutOfIbd(); + auto peer{MakeTestPeer(/*id=*/0)}; + auto fallback{MakeTestPeer(/*id=*/1)}; + auto& peerman = *m_node.peerman; + peerman.InitializeNode(*peer, NODE_NETWORK); + peerman.InitializeNode(*fallback, NODE_NETWORK); + + std::vector objects; + for (uint32_t i = 1; i <= 128; ++i) { + objects.emplace_back(MSG_SPORK, ArithToUint256(arith_uint256{i})); + objects.emplace_back(MSG_CLSIG, ArithToUint256(arith_uint256{i})); + } + auto batch = objects; + batch.insert(batch.end(), objects.begin(), objects.end()); // Duplicate announcements. + batch.emplace_back(0, objects.front().hash); // Unknown types must not affect accounting. + const std::atomic interrupt{false}; + const auto process_batch = + [&](CNode& node, const std::string& command, const std::vector& invs) + EXCLUSIVE_LOCKS_REQUIRED(NetEventsInterface::g_msgproc_mutex) { + CDataStream stream{SER_NETWORK, PROTOCOL_VERSION}; + stream << invs; + peerman.ProcessMessage(node, command, stream, GetTime(), interrupt); + }; + process_batch(*peer, NetMsgType::INV, batch); + process_batch(*fallback, NetMsgType::INV, batch); + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(peer->GetId()), objects.size()); + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(fallback->GetId()), objects.size()); + + SetMockTime(GetTime() + 61s); + peerman.SendMessages(peer.get()); + // Mix requested entries with duplicate, unknown, and unsolicited NOTFOUND entries. + batch.emplace_back(MSG_SPORK, uint256S("ffff")); + process_batch(*peer, NetMsgType::NOTFOUND, batch); + // Completed entries remain tracked until the fallback also completes. + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(peer->GetId()), objects.size()); + for (const auto& inv : objects) { + BOOST_CHECK(!WITH_LOCK(cs_main, return peerman.PeerConsumeObjectRequest(peer->GetId(), inv))); + } + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(fallback->GetId()), objects.size()); + peerman.SendMessages(fallback.get()); + for (const auto& inv : objects) { + BOOST_CHECK(WITH_LOCK(cs_main, return peerman.PeerConsumeObjectRequest(fallback->GetId(), inv))); + BOOST_CHECK(!WITH_LOCK(cs_main, return peerman.PeerConsumeObjectRequest(fallback->GetId(), inv))); + } + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(fallback->GetId()), 0U); + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(peer->GetId()), 0U); + + // Fresh announcements after completion remain requestable and are removed on disconnect. + process_batch(*peer, NetMsgType::INV, objects); + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(peer->GetId()), objects.size()); + peerman.FinalizeNode(*peer); + peerman.FinalizeNode(*fallback); + chainstate.ResetIbd(); + SetMockTime(0s); +} + +BOOST_AUTO_TEST_CASE(notfound_does_not_wait_for_chainstate) +{ + auto peer{MakeTestPeer(/*id=*/0)}; + auto& peerman = *m_node.peerman; + peerman.InitializeNode(*peer, NODE_NETWORK); + const CInv inv{MSG_SPORK, uint256S("01")}; + { + LOCK(NetEventsInterface::g_msgproc_mutex); + ProcessInv(peerman, *peer, inv); + } + std::future response; + std::future_status status; + { + LOCK(cs_main); + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(peer->GetId()), 1U); + response = std::async(std::launch::async, [&] { + LOCK(NetEventsInterface::g_msgproc_mutex); + CDataStream stream{SER_NETWORK, PROTOCOL_VERSION}; + stream << std::vector{inv}; + const std::atomic interrupt{false}; + peerman.ProcessMessage(*peer, NetMsgType::NOTFOUND, stream, GetTime(), interrupt); + }); + // The timeout bounds failure cleanup; completion while cs_main is held is the invariant. + status = response.wait_for(5s); + } + response.get(); + BOOST_CHECK(status == std::future_status::ready); + BOOST_CHECK_EQUAL(peerman.GetRequestedObjectCount(peer->GetId()), 0U); + peerman.FinalizeNode(*peer); +} + BOOST_AUTO_TEST_CASE(peer_requested_object_authorizes_and_erases_per_peer_state) { LOCK(NetEventsInterface::g_msgproc_mutex); @@ -87,14 +180,14 @@ BOOST_AUTO_TEST_CASE(peer_requested_object_authorizes_and_erases_per_peer_state) const CInv announced_inv{MSG_SPORK, uint256S("01")}; ProcessInv(*m_node.peerman, *peer, announced_inv); // The announcement is queued for a GETDATA that hasn't been sent yet. - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 1U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 1U); // Consuming completes the peer's announcement and returns true exactly once. BOOST_CHECK(WITH_LOCK(::cs_main, return m_node.peerman->PeerConsumeObjectRequest(peer->GetId(), announced_inv))); BOOST_CHECK(!WITH_LOCK(::cs_main, return m_node.peerman->PeerConsumeObjectRequest(peer->GetId(), announced_inv))); // A consumed announcement must not be requested by SendMessages. SetMockTime(GetTime() + 61s); m_node.peerman->SendMessages(peer.get()); - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 0U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 0U); // Not re-requested: consuming did not resurrect the announcement. BOOST_CHECK(!WITH_LOCK(::cs_main, return m_node.peerman->PeerConsumeObjectRequest(peer->GetId(), announced_inv))); @@ -145,7 +238,7 @@ BOOST_AUTO_TEST_CASE(peer_getdata_response_requires_an_inflight_request) // Announced but not yet requested: the looser check accepts this, the stricter one must not. BOOST_CHECK(ConsumeGetDataResponse(*m_node.peerman, *peer, inv) == GetDataResponse::UNREQUESTED); // The rejection left the candidate intact, so the GETDATA is still pending. - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 1U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 1U); // After SendMessages issues the GETDATA the announcement is REQUESTED and authorises once. SetMockTime(GetTime() + 61s); @@ -156,7 +249,7 @@ BOOST_AUTO_TEST_CASE(peer_getdata_response_requires_an_inflight_request) // Never announced at all: rejected, and no trace left behind. const CInv never_announced{MSG_SPORK, uint256S("05")}; BOOST_CHECK(ConsumeGetDataResponse(*m_node.peerman, *peer, never_announced) == GetDataResponse::UNREQUESTED); - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 0U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 0U); m_node.peerman->FinalizeNode(*peer); chainstate.ResetIbd(); @@ -185,14 +278,14 @@ BOOST_AUTO_TEST_CASE(expired_getdata_response_is_late_not_unrequested) // Nudge past the announcement's reqtime so SendMessages issues the GETDATA. SetMockTime(GetTime() + 2s); m_node.peerman->SendMessages(peer.get()); - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 1U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 1U); // Answer on the last moment of the grace: the request expired at CLSIG_REQUEST_INTERVAL, and // this peer was the only announcer, so the tracker drops the record entirely rather than keeping // a COMPLETED one -- there is nothing left for it to consult. SetMockTime(GetTime() + CLSIG_LATE_GRACE); m_node.peerman->SendMessages(peer.get()); - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 0U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 0U); // The answer is late, not unsolicited: it must not be scored. BOOST_CHECK(ConsumeGetDataResponse(*m_node.peerman, *peer, inv) == GetDataResponse::LATE); @@ -227,11 +320,11 @@ BOOST_AUTO_TEST_CASE(forgotten_getdata_response_is_late_not_unrequested) SetMockTime(GetTime() + 2s); m_node.peerman->SendMessages(peer.get()); - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 1U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 1U); // The object arrives from somewhere else while our GETDATA is still in flight. WITH_LOCK(::cs_main, m_node.peerman->PeerForgetObjectRequest(inv)); - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 0U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 0U); BOOST_CHECK(ConsumeGetDataResponse(*m_node.peerman, *peer, inv) == GetDataResponse::LATE); BOOST_CHECK(ConsumeGetDataResponse(*m_node.peerman, *peer, inv) == GetDataResponse::UNREQUESTED); @@ -296,7 +389,7 @@ BOOST_AUTO_TEST_CASE(getdata_response_grace_expires) // sits on, so the two cases together pin it from both sides. SetMockTime(GetTime() + CLSIG_LATE_GRACE + 1s); m_node.peerman->SendMessages(peer.get()); - BOOST_CHECK_EQUAL(WITH_LOCK(::cs_main, return m_node.peerman->GetRequestedObjectCount(peer->GetId())), 0U); + BOOST_CHECK_EQUAL(m_node.peerman->GetRequestedObjectCount(peer->GetId()), 0U); BOOST_CHECK(ConsumeGetDataResponse(*m_node.peerman, *peer, inv) == GetDataResponse::UNREQUESTED); diff --git a/test/util/data/non-backported.txt b/test/util/data/non-backported.txt index b27dc1cde71b..e06e682ac56a 100644 --- a/test/util/data/non-backported.txt +++ b/test/util/data/non-backported.txt @@ -3,6 +3,7 @@ src/active/*.h src/batchedlogger.* src/bench/blockfilter_index.cpp src/bench/bls*.cpp +src/bench/net_processing.cpp src/bls/*.cpp src/bls/*.h src/cachemap.h