perf(net): isolate object request scheduling from cs_main

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()

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
 --------------------
 

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 \

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 <arith_uint256.h>
+#include <bench/bench.h>
+#include <net_processing.h>
+#include <test/util/net.h>
+#include <test/util/setup_common.h>
+#include <test/util/validation.h>
+#include <validation.h>
+
+using namespace std::literals;
+
+static void InventoryBatch(benchmark::Bench& bench, uint32_t count)
+{
+    const auto setup = MakeNoLogFileContext<TestingSetup>();
+    auto& chainstate = *static_cast<TestChainState*>(&setup->m_node.chainman->ActiveChainstate());
+    chainstate.JumpOutOfIbd();
+    auto& peerman = *setup->m_node.peerman;
+    auto& connman = *static_cast<ConnmanTestMsg*>(setup->m_node.connman.get());
+    auto peer{MakeTestPeer(/*id=*/0)};
+    peerman.InitializeNode(*peer, NODE_NETWORK);
+
+    std::vector<CInv> 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<bool> interrupt{false};
+    SetMockTime(1'700'000'000s);
+    const auto now{GetTime<std::chrono::microseconds>()};
+
+    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);