diff --git a/contrib/devtools/benchmark_inventory.py b/contrib/devtools/benchmark_inventory.py new file mode 100755 index 000000000000..f3a207b06fb3 --- /dev/null +++ b/contrib/devtools/benchmark_inventory.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +# 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. +"""Measure chain-height RPC latency during inventory download and NOTFOUND fallback. + +Run from the repository root with PYTHONPATH=test/functional and DASHD pointing +to the binary under test. See doc/benchmarking.md for methodology and examples. +""" + +import json +import multiprocessing +import subprocess +import time +from pathlib import Path + +from test_framework.authproxy import AuthServiceProxy +from test_framework.messages import CInv, msg_inv, msg_notfound +from test_framework.p2p import P2PInterface, p2p_lock +from test_framework.test_framework import BitcoinTestFramework + + +def query_rpc(url, stop, output): + rpc = AuthServiceProxy(url, timeout=60) + latencies = [] + error = None + try: + while not stop.is_set(): + start = time.perf_counter() + rpc.getblockcount() + latencies.append((time.perf_counter() - start) * 1000) + stop.wait(0.001) + except Exception as exc: + error = str(exc) + output.send((latencies, error)) + output.close() + + +class Peer(P2PInterface): + def __init__(self): + super().__init__() + self.requested = 0 + + def on_getdata(self, message): + self.requested += len(message.inv) + self.send_message(msg_notfound(message.inv)) + + +class InventoryBenchmark(BitcoinTestFramework): + def set_test_params(self): + self.num_nodes = 1 + self.setup_clean_chain = True + self.extra_args = [["-debug=0", "-logtimemicros=1"]] + + def add_options(self, parser): + parser.add_argument("--batch", type=int, default=5000) + parser.add_argument("--rounds", type=int, default=20) + parser.add_argument("--peers", type=int, default=4) + parser.add_argument("--sample", action="store_true", help="Capture a macOS sample trace (separate from timing comparisons)") + parser.add_argument("--inv-type", type=int, choices=(6, 17, 18, 29, 31), default=6) + + def run_test(self): + assert 1 <= self.options.batch <= 50000 + assert self.options.rounds > 0 + assert 1 <= self.options.peers <= 16 + node = self.nodes[0] + self.generate(node, 1) + if self.options.inv_type in (17, 18): + while not node.mnsync("status")["IsBlockchainSynced"]: + node.mnsync("next") + node.logging([], ["all"]) + node.logging(["lock"], []) + peers = [node.add_p2p_connection(Peer()) for _ in range(self.options.peers)] + # Construct inventories before timing; transport serialization remains included. + messages = [msg_inv([CInv(self.options.inv_type, 1 + r * self.options.batch + i) + for i in range(self.options.batch)]) + for r in range(self.options.rounds)] + # A separate process avoids measuring the P2P driver's Python GIL contention as RPC latency. + context = multiprocessing.get_context("spawn") + stop = context.Event() + receive, output = context.Pipe(duplex=False) + observer = context.Process(target=query_rpc, args=(node.url, stop, output)) + sample = None + if self.options.sample: + sample = subprocess.Popen(["sample", str(node.process.pid), "10", "1", "-file", + str(Path(self.options.tmpdir) / "sample.txt")], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + observer.start() + start = time.perf_counter() + try: + for round_index, message in enumerate(messages, start=1): + for peer in peers: + peer.send_message(message) + # Drain this round before announcing more: otherwise outstanding requests can + # activate the overload delay, which cannot expire while mocktime is fixed. + self.wait_until(lambda: all(peer.requested >= round_index * self.options.batch for peer in peers)) + for peer in peers: + peer.sync_with_ping() + for peer in peers: + peer.sync_with_ping() + expected = len(peers) * len(messages) * self.options.batch + self.wait_until(lambda: sum(peer.requested for peer in peers) >= expected) + finally: + elapsed = time.perf_counter() - start + stop.set() + if not receive.poll(65): + observer.terminate() + observer.join(timeout=5) + raise RuntimeError("RPC observer did not finish") + latencies, error = receive.recv() + observer.join(timeout=5) + assert not observer.is_alive() + assert not error, error + receive.close() + output.close() + if sample: + sample.wait(timeout=30) + latencies.sort() + assert latencies, "No RPC samples were collected" + with p2p_lock: + requested = sum(peer.requested for peer in peers) + assert all(peer.requested == len(messages) * self.options.batch for peer in peers) + assert requested == expected, (requested, expected) + result = {"seconds": elapsed, "announcements": len(peers) * len(messages) * self.options.batch, + "requested": requested, "rpc_count": len(latencies), + "rpc_ms_p50": latencies[len(latencies)//2], + "rpc_ms_p95": latencies[int(len(latencies)*.95)], + "rpc_ms_p99": latencies[int(len(latencies)*.99)], "rpc_ms_max": max(latencies)} + Path(self.options.tmpdir, "rpc-latencies-ms.json").write_text(json.dumps(latencies) + "\n") + Path(self.options.tmpdir, "metrics.json").write_text(json.dumps(result, indent=2) + "\n") + self.log.info("PROFILE %s", json.dumps(result)) + + +if __name__ == "__main__": + InventoryBenchmark().main() diff --git a/doc/benchmarking.md b/doc/benchmarking.md index e02563b61c66..943544d80385 100644 --- a/doc/benchmarking.md +++ b/doc/benchmarking.md @@ -53,6 +53,52 @@ More benchmarks are needed for, in no particular order: - Cuckoo Cache - P2P throughput +Inventory processing and cs_main contention +------------------------------------------ + +`InventoryBatch100` and `InventoryBatch50000` exercise the actual peer-manager +INV, GETDATA scheduling, and NOTFOUND paths without sockets or a Python driver. +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. + +```sh +src/bench/bench_dash -filter='InventoryBatch.*' -min-time=5000 -output-json=inventory.json +``` + +For an end-to-end contention workload, use the functional-test dependencies +(including `dash_hash`) and `contrib/devtools/benchmark_inventory.py`. It starts +an isolated regtest node, announces batches of unknown objects from its peers, +and answers GETDATA with NOTFOUND so every fallback peer is tried. The run +checks that the number of requested objects matches the announcements. A +separate process measures `getblockcount` latency, avoiding the Python P2P +driver's GIL contention in the RPC measurements. + +```sh +PYTHONPATH=test/functional DASHD=/absolute/path/to/dashd \ + python3 contrib/devtools/benchmark_inventory.py --configfile=test/config.ini \ + --tmpdir=/tmp/inventory-baseline-1 --nocleanup --randomseed=6990 \ + --batch=50000 --rounds=8 --peers=1 +``` + +The output directory contains `metrics.json`, individual RPC samples in +`rpc-latencies-ms.json`, and the node's log. Compile both compared binaries +with `CPPFLAGS=-DDEBUG_LOCKCONTENTION` to also record lock acquisition waits. +For CPU attribution on macOS, `--sample` captures a stack sample; run it +separately from timing comparisons because sampling can perturb the process. + +Compare repeated runs of the same workload and alternate baseline/candidate +order on the same host, without simultaneous builds or tests. Also test small +batches (`--batch=100 --rounds=20 --peers=4`) and governance-vote inventories +(`--inv-type=18 --batch=5000 --rounds=20 --peers=4`). Governance profiling advances +the regtest masternode sync state before measuring. + +These are synthetic request/fallback workloads, not live-network throughput +measurements. Report native CPU cost, RPC latency, and lock waits separately: +the socket workload's completion time also includes Python serialization and +polling. Record host load, build options, and exact revisions; do not infer a +mainnet speedup from reduced tail latency in the large-batch workload. + 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..3fb0777402d5 --- /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; + 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..ebbb3ec99db0 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,12 @@ 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, ::cs_main); /** Implements external handlers logic */ void AddExtraHandler(std::unique_ptr&& handler) override; @@ -635,10 +640,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 +654,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, ::cs_main); + 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 +679,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 +757,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 +1106,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 +1629,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,6 +1656,7 @@ 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)); } @@ -1649,6 +1664,7 @@ void PeerManagerImpl::ForgetTx(const uint256& txid) size_t PeerManagerImpl::GetRequestedObjectCount(NodeId nodeid) const { AssertLockHeld(cs_main); + LOCK(m_object_request_mutex); return m_object_request.Count(nodeid); } @@ -1711,7 +1727,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 +1793,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 +1810,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 +2466,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 +3767,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 +3775,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 +4830,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 +5619,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 +5673,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 +6250,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto) MaybeSendAddr(*pto, *peer, current_time); + std::vector vGetData; { LOCK(cs_main); @@ -6717,7 +6736,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 +6753,50 @@ 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); - } - 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 { - // 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); + // 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); + } + for (const CInv& inv : requestable) { + LOCK(cs_main); + 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(); + } + WITH_LOCK(m_object_request_mutex, + 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(pto->GetId())->m_recent_object_requests.insert(inv.hash, RequestedObject{inv.type, current_time}); } + } else { + // 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. + WITH_LOCK(m_object_request_mutex, 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()); - } - } // release cs_main + 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 +6816,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 +6857,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/test/net_tests.cpp b/src/test/net_tests.cpp index 84e727b5df5b..16e1501efa92 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(WITH_LOCK(cs_main, return peerman.GetRequestedObjectCount(peer->GetId())), objects.size()); + BOOST_CHECK_EQUAL(WITH_LOCK(cs_main, return 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(WITH_LOCK(cs_main, return 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(WITH_LOCK(cs_main, return 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(WITH_LOCK(cs_main, return peerman.GetRequestedObjectCount(fallback->GetId())), 0U); + BOOST_CHECK_EQUAL(WITH_LOCK(cs_main, return 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(WITH_LOCK(cs_main, return 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(WITH_LOCK(cs_main, return 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);