From ec15f90cc91435079a7d48e18771afba476b4bb2 Mon Sep 17 00:00:00 2001 From: Konstantin Akimov Date: Thu, 9 Jul 2026 22:50:39 +0700 Subject: [PATCH 1/6] refactor: use dedicated util structure to count internal objects in signing-shares --- src/llmq/signing_shares.h | 146 +++++++++++++++++++++++++++----------- 1 file changed, 103 insertions(+), 43 deletions(-) diff --git a/src/llmq/signing_shares.h b/src/llmq/signing_shares.h index ef8aa42edc0f..8a7a483b66a1 100644 --- a/src/llmq/signing_shares.h +++ b/src/llmq/signing_shares.h @@ -9,7 +9,6 @@ #include #include #include -#include #include #include @@ -160,40 +159,114 @@ class CBatchedSigShares [[nodiscard]] std::string ToInvString() const; }; +/** + * Two-level (signHash -> quorumMember) map with a running entry count, so Size() is O(1) + * instead of a fold over all sign hash buckets. All structural mutations go through the + * counted methods; Buckets() is for lookups and in-place value updates only. + */ template -class SigShareMap +class CountedBucketMap { +public: + using BucketMap = Uint256HashMap>; + private: - Uint256HashMap> internalMap; + BucketMap m_data; + size_t m_num_entries{0}; public: - bool Add(const SigShareKey& k, const T& v) + BucketMap& Buckets() { return m_data; } + const BucketMap& Buckets() const { return m_data; } + [[nodiscard]] size_t Size() const { return m_num_entries; } + + bool Emplace(const SigShareKey& k, const T& v) { - auto& m = internalMap[k.first]; - return m.emplace(k.second, v).second; + if (!m_data[k.first].emplace(k.second, v).second) { + return false; + } + ++m_num_entries; + return true; } void Erase(const SigShareKey& k) { - auto it = internalMap.find(k.first); - if (it == internalMap.end()) { + auto it = m_data.find(k.first); + if (it == m_data.end()) { return; } - it->second.erase(k.second); + m_num_entries -= it->second.erase(k.second); if (it->second.empty()) { - internalMap.erase(it); + m_data.erase(it); } } + void EraseBucket(const uint256& signHash) + { + auto it = m_data.find(signHash); + if (it == m_data.end()) { + return; + } + m_num_entries -= it->second.size(); + m_data.erase(it); + } + + template + void EraseIf(F&& f) + { + for (auto it = m_data.begin(); it != m_data.end(); ) { + SigShareKey k; + k.first = it->first; + for (auto jt = it->second.begin(); jt != it->second.end(); ) { + k.second = jt->first; + if (f(k, jt->second)) { + jt = it->second.erase(jt); + --m_num_entries; + } else { + ++jt; + } + } + if (it->second.empty()) { + it = m_data.erase(it); + } else { + ++it; + } + } + } + + void Clear() + { + m_data.clear(); + m_num_entries = 0; + } +}; + +template +class SigShareMap +{ +private: + CountedBucketMap internalMap; + +public: + bool Add(const SigShareKey& k, const T& v) + { + return internalMap.Emplace(k, v); + } + + void Erase(const SigShareKey& k) + { + internalMap.Erase(k); + } + void Clear() { - internalMap.clear(); + internalMap.Clear(); } [[nodiscard]] bool Has(const SigShareKey& k) const { - auto it = internalMap.find(k.first); - if (it == internalMap.end()) { + auto& m = internalMap.Buckets(); + auto it = m.find(k.first); + if (it == m.end()) { return false; } return it->second.count(k.second) != 0; @@ -201,8 +274,9 @@ class SigShareMap T* Get(const SigShareKey& k) { - auto it = internalMap.find(k.first); - if (it == internalMap.end()) { + auto& m = internalMap.Buckets(); + auto it = m.find(k.first); + if (it == m.end()) { return nullptr; } @@ -226,22 +300,23 @@ class SigShareMap const T* GetFirst() const { - if (internalMap.empty()) { + auto& m = internalMap.Buckets(); + if (m.empty()) { return nullptr; } - return &internalMap.begin()->second.begin()->second; + return &m.begin()->second.begin()->second; } [[nodiscard]] size_t Size() const { - return std23::ranges::fold_left(internalMap, size_t{0}, - [](size_t s, const auto& p) { return s + p.second.size(); }); + return internalMap.Size(); } [[nodiscard]] size_t CountForSignHash(const uint256& signHash) const { - auto it = internalMap.find(signHash); - if (it == internalMap.end()) { + auto& m = internalMap.Buckets(); + auto it = m.find(signHash); + if (it == m.end()) { return 0; } return it->second.size(); @@ -249,13 +324,14 @@ class SigShareMap [[nodiscard]] bool Empty() const { - return internalMap.empty(); + return internalMap.Buckets().empty(); } const std::unordered_map* GetAllForSignHash(const uint256& signHash) const { - auto it = internalMap.find(signHash); - if (it == internalMap.end()) { + auto& m = internalMap.Buckets(); + auto it = m.find(signHash); + if (it == m.end()) { return nullptr; } return &it->second; @@ -263,35 +339,19 @@ class SigShareMap void EraseAllForSignHash(const uint256& signHash) { - internalMap.erase(signHash); + internalMap.EraseBucket(signHash); } template void EraseIf(F&& f) { - for (auto it = internalMap.begin(); it != internalMap.end(); ) { - SigShareKey k; - k.first = it->first; - for (auto jt = it->second.begin(); jt != it->second.end(); ) { - k.second = jt->first; - if (f(k, jt->second)) { - jt = it->second.erase(jt); - } else { - ++jt; - } - } - if (it->second.empty()) { - it = internalMap.erase(it); - } else { - ++it; - } - } + internalMap.EraseIf(f); } template void ForEach(F&& f) { - for (auto& p : internalMap) { + for (auto& p : internalMap.Buckets()) { SigShareKey k; k.first = p.first; for (auto& p2 : p.second) { From 4666ef3a1fb00e6f9934be839cb4e300ab27d822 Mon Sep 17 00:00:00 2001 From: Konstantin Akimov Date: Thu, 9 Jul 2026 22:51:12 +0700 Subject: [PATCH 2/6] test: add regression tests for CountedBucketMap --- src/test/llmq_utils_tests.cpp | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/src/test/llmq_utils_tests.cpp b/src/test/llmq_utils_tests.cpp index ea51601a8b5a..75bc0cfe99c3 100644 --- a/src/test/llmq_utils_tests.cpp +++ b/src/test/llmq_utils_tests.cpp @@ -103,6 +103,34 @@ BOOST_AUTO_TEST_CASE(sig_ses_ann_limit_is_per_llmq_type) BOOST_CHECK_EQUAL(node_state.GetSessionCount(Consensus::LLMQType::LLMQ_400_60), 1U); } +BOOST_AUTO_TEST_CASE(sig_share_map_size_tracks_mutations) +{ + SigShareMap sig_share_map; + const CSigShare sig_share1{MakeSigShare(1)}; + const CSigShare sig_share2{MakeSigShare(2)}; + + BOOST_CHECK(sig_share_map.Add(sig_share1.GetKey(), sig_share1)); + BOOST_CHECK(!sig_share_map.Add(sig_share1.GetKey(), sig_share1)); + BOOST_CHECK(sig_share_map.Add(sig_share2.GetKey(), sig_share2)); + BOOST_CHECK_EQUAL(sig_share_map.Size(), 2U); + + sig_share_map.Erase(sig_share1.GetKey()); + sig_share_map.Erase(sig_share1.GetKey()); + BOOST_CHECK_EQUAL(sig_share_map.Size(), 1U); + + sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()); + BOOST_CHECK_EQUAL(sig_share_map.Size(), 0U); + BOOST_CHECK(sig_share_map.Empty()); + + BOOST_CHECK(sig_share_map.Add(sig_share1.GetKey(), sig_share1)); + BOOST_CHECK(sig_share_map.Add(sig_share2.GetKey(), sig_share2)); + sig_share_map.EraseIf([&](const SigShareKey& k, const CSigShare&) { return k == sig_share1.GetKey(); }); + BOOST_CHECK_EQUAL(sig_share_map.Size(), 1U); + + sig_share_map.Clear(); + BOOST_CHECK_EQUAL(sig_share_map.Size(), 0U); +} + BOOST_AUTO_TEST_CASE(deterministic_outbound_connection_test) { // Test deterministic behavior From e2862fe90df4ccfdd2953e7f27df25e7e9d4d673 Mon Sep 17 00:00:00 2001 From: Konstantin Akimov Date: Fri, 10 Jul 2026 01:56:20 +0700 Subject: [PATCH 3/6] fix: the actual cap enforcement for pending signatures --- src/llmq/signing_shares.cpp | 35 +++++++++++++++++++++++++++++++++-- src/llmq/signing_shares.h | 2 ++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/src/llmq/signing_shares.cpp b/src/llmq/signing_shares.cpp index 4849e2940ee9..efd83fadbbb7 100644 --- a/src/llmq/signing_shares.cpp +++ b/src/llmq/signing_shares.cpp @@ -31,6 +31,11 @@ namespace { constexpr size_t MAX_SESSIONS_PER_PEER_FACTOR{4}; constexpr size_t MIN_SESSIONS_PER_PEER{100}; +// Incoming QSIGSHARE/QBSIGSHARES traffic is cheap to admit but drains only at BLS verification +// speed, so unverified shares are bounded and over-cap shares dropped without misbehaviour scoring. +constexpr size_t MAX_PENDING_SIG_SHARES_PER_NODE{1000}; +constexpr size_t MAX_PENDING_SIG_SHARES_TOTAL{10000}; + size_t GetMaxSessionsForPeer(const Consensus::LLMQParams& params) { return std::max(size_t(params.size) * MAX_SESSIONS_PER_PEER_FACTOR, MIN_SESSIONS_PER_PEER); @@ -422,7 +427,7 @@ bool CSigSharesManager::ProcessMessageBatchedSigShares(const CNode& pfrom, const LOCK(cs); auto& nodeState = nodeStates[pfrom.GetId()]; for (const auto& s : sigSharesToProcess) { - nodeState.pendingIncomingSigShares.Add(s.GetKey(), s); + TryAddPendingIncomingSigShare(pfrom.GetId(), nodeState, s); } return true; } @@ -467,7 +472,7 @@ bool CSigSharesManager::ProcessMessageSigShare(NodeId fromId, const CSigShare& s } auto& nodeState = nodeStates[fromId]; - nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare); + TryAddPendingIncomingSigShare(fromId, nodeState, sigShare); } LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- signHash=%s, id=%s, msgHash=%s, member=%d, node=%d\n", __func__, @@ -475,6 +480,31 @@ bool CSigSharesManager::ProcessMessageSigShare(NodeId fromId, const CSigShare& s return true; } +bool CSigSharesManager::TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesNodeState& nodeState, + const CSigShare& sigShare) +{ + AssertLockHeld(cs); + + if (nodeState.banned) { + return false; + } + if (nodeState.pendingIncomingSigShares.Size() >= MAX_PENDING_SIG_SHARES_PER_NODE) { + LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- per-node pending sig shares cap reached (%d), dropping sigShare. node=%d\n", + __func__, MAX_PENDING_SIG_SHARES_PER_NODE, nodeId); + return false; + } + size_t total{0}; + for (const auto& [_, ns] : nodeStates) { + total += ns.pendingIncomingSigShares.Size(); + } + if (total >= MAX_PENDING_SIG_SHARES_TOTAL) { + LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- global pending sig shares cap reached (%d), dropping sigShare. node=%d\n", + __func__, MAX_PENDING_SIG_SHARES_TOTAL, nodeId); + return false; + } + return nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare); +} + bool CSigSharesManager::CollectPendingSigSharesToVerify( size_t maxUniqueSessions, std::unordered_map>& retSigShares, std::unordered_map, CQuorumCPtr, StaticSaltedHasher>& retQuorums) @@ -1425,6 +1455,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId) sigSharesRequested.Erase(k); }); nodeState.requestedSigShares.Clear(); + nodeState.pendingIncomingSigShares.Clear(); nodeState.banned = true; } diff --git a/src/llmq/signing_shares.h b/src/llmq/signing_shares.h index 8a7a483b66a1..7e0e95ab714f 100644 --- a/src/llmq/signing_shares.h +++ b/src/llmq/signing_shares.h @@ -544,6 +544,8 @@ class CSigSharesManager : public llmq::CRecoveredSigsListener bool GetSessionInfoByRecvId(NodeId nodeId, uint32_t sessionId, CSigSharesNodeState::SessionInfo& retInfo) EXCLUSIVE_LOCKS_REQUIRED(!cs); static CSigShare RebuildSigShare(const CSigSharesNodeState::SessionInfo& session, const std::pair& in); + bool TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesNodeState& nodeState, const CSigShare& sigShare) + EXCLUSIVE_LOCKS_REQUIRED(cs); void RemoveSigSharesForSession(const uint256& signHash) EXCLUSIVE_LOCKS_REQUIRED(cs); From 08d95330926310ee189248920f639b3b2768d837 Mon Sep 17 00:00:00 2001 From: Konstantin Akimov Date: Fri, 10 Jul 2026 02:01:25 +0700 Subject: [PATCH 4/6] fix: cap amount of unverified batches by 4 --- src/llmq/net_signing.cpp | 13 +++++++++++-- src/llmq/net_signing.h | 2 ++ src/test/llmq_utils_tests.cpp | 16 ++++++++++++++++ 3 files changed, 29 insertions(+), 2 deletions(-) diff --git a/src/llmq/net_signing.cpp b/src/llmq/net_signing.cpp index b8f937b634bb..1c3561bf2b4d 100644 --- a/src/llmq/net_signing.cpp +++ b/src/llmq/net_signing.cpp @@ -320,8 +320,11 @@ void NetSigning::WorkThreadDispatcher() } } - // Collect pending sig shares synchronously and dispatch each batch to a worker for parallel BLS verification - while (!workInterrupt) { + // Collect pending sig shares synchronously and dispatch each batch to a worker for parallel BLS verification. + // Batches awaiting verification are bounded so that under flood shares back up in the capped + // pending maps instead of migrating into the unbounded worker pool task queue. + static constexpr int MAX_UNVERIFIED_BATCHES{4}; + while (!workInterrupt && unverified_batches < MAX_UNVERIFIED_BATCHES) { std::unordered_map> sigSharesByNodes; std::unordered_map, CQuorumCPtr, StaticSaltedHasher> quorums; @@ -332,7 +335,13 @@ void NetSigning::WorkThreadDispatcher() break; } + ++unverified_batches; worker_pool.push([this, sigSharesByNodes = std::move(sigSharesByNodes), quorums = std::move(quorums)](int) mutable { + // Ensures unverified_batches is decremented on every exit path, including exceptions. + struct UnverifiedBatchGuard { + std::atomic& count; + ~UnverifiedBatchGuard() { --count; } + } guard{unverified_batches}; ProcessPendingSigShares(std::move(sigSharesByNodes), std::move(quorums)); }); diff --git a/src/llmq/net_signing.h b/src/llmq/net_signing.h index a9effd2575e3..431e6efb1537 100644 --- a/src/llmq/net_signing.h +++ b/src/llmq/net_signing.h @@ -15,6 +15,7 @@ #include +#include #include class CSporkManager; @@ -72,6 +73,7 @@ class NetSigning final : public NetHandler, public CValidationInterface std::thread shares_cleaning_thread; std::thread shares_dispatcher_thread; mutable ctpl::thread_pool worker_pool; + std::atomic unverified_batches{0}; CThreadInterrupt workInterrupt; }; diff --git a/src/test/llmq_utils_tests.cpp b/src/test/llmq_utils_tests.cpp index 75bc0cfe99c3..fb565c3fe1d4 100644 --- a/src/test/llmq_utils_tests.cpp +++ b/src/test/llmq_utils_tests.cpp @@ -131,6 +131,22 @@ BOOST_AUTO_TEST_CASE(sig_share_map_size_tracks_mutations) BOOST_CHECK_EQUAL(sig_share_map.Size(), 0U); } +BOOST_AUTO_TEST_CASE(pending_sig_shares_session_removal_updates_count) +{ + CSigSharesNodeState node_state; + const CSigShare sig_share1{MakeSigShare(1)}; + const CSigShare sig_share2{MakeSigShare(2)}; + + BOOST_CHECK(node_state.pendingIncomingSigShares.Add(sig_share1.GetKey(), sig_share1)); + BOOST_CHECK(node_state.pendingIncomingSigShares.Add(sig_share2.GetKey(), sig_share2)); + BOOST_CHECK_EQUAL(node_state.pendingIncomingSigShares.Size(), 2U); + + node_state.RemoveSession(sig_share1.GetSignHash()); + BOOST_CHECK_EQUAL(node_state.pendingIncomingSigShares.Size(), 1U); + BOOST_CHECK(!node_state.pendingIncomingSigShares.Has(sig_share1.GetKey())); + BOOST_CHECK(node_state.pendingIncomingSigShares.Has(sig_share2.GetKey())); +} + BOOST_AUTO_TEST_CASE(deterministic_outbound_connection_test) { // Test deterministic behavior From 355da45d4a45e7ceac552fd005a5519de3c13c62 Mon Sep 17 00:00:00 2001 From: PastaClaw Date: Thu, 9 Jul 2026 18:04:41 -0500 Subject: [PATCH 5/6] fix: cap sig-share verification batch by share count CollectPendingSigSharesToVerify used to bound each dispatched batch by unique (nodeId, signHash) sessions. A single session can carry up to the full MAX_PENDING_SIG_SHARES_TOTAL (10000) shares, so with the four in-flight batches allowed on the worker pool, tens of thousands of BLS shares could sit outside the pending accounting. Bound each batch by the actual number of shares added (32) instead. The rotation-based fairness across peers is preserved: one share at a time in randomized round-robin order. Combined with MAX_UNVERIFIED_BATCHES=4, at most 128 shares can be in-flight in / on the worker pool at once. Also collapse the O(peers) global-cap sum in TryAddPendingIncomingSigShare into a running m_pending_sig_shares_total maintained on every mutation site. The primitives now return the count they erased so the running total stays in lockstep with the aggregate node-state size. Add regression coverage for the count-returning primitives and for CSigSharesNodeState::RemoveSession's returned drop count. --- src/llmq/net_signing.cpp | 7 +++-- src/llmq/signing_shares.cpp | 45 +++++++++++++++------------ src/llmq/signing_shares.h | 58 +++++++++++++++++++++++------------ src/test/llmq_utils_tests.cpp | 42 +++++++++++++++++++++---- 4 files changed, 105 insertions(+), 47 deletions(-) diff --git a/src/llmq/net_signing.cpp b/src/llmq/net_signing.cpp index 1c3561bf2b4d..3b622e5b05aa 100644 --- a/src/llmq/net_signing.cpp +++ b/src/llmq/net_signing.cpp @@ -323,13 +323,16 @@ void NetSigning::WorkThreadDispatcher() // Collect pending sig shares synchronously and dispatch each batch to a worker for parallel BLS verification. // Batches awaiting verification are bounded so that under flood shares back up in the capped // pending maps instead of migrating into the unbounded worker pool task queue. + // + // Each batch is bounded by actual share count (not unique-session count), so at most + // MAX_UNVERIFIED_BATCHES * MAX_SHARES_PER_BATCH shares can be sitting in / on the worker pool at once. static constexpr int MAX_UNVERIFIED_BATCHES{4}; + static constexpr size_t MAX_SHARES_PER_BATCH{32}; while (!workInterrupt && unverified_batches < MAX_UNVERIFIED_BATCHES) { std::unordered_map> sigSharesByNodes; std::unordered_map, CQuorumCPtr, StaticSaltedHasher> quorums; - const size_t nMaxBatchSize{32}; - bool more_work = m_shares_manager->CollectPendingSigSharesToVerify(nMaxBatchSize, sigSharesByNodes, quorums); + bool more_work = m_shares_manager->CollectPendingSigSharesToVerify(MAX_SHARES_PER_BATCH, sigSharesByNodes, quorums); if (sigSharesByNodes.empty()) { break; diff --git a/src/llmq/signing_shares.cpp b/src/llmq/signing_shares.cpp index efd83fadbbb7..8101dc70871e 100644 --- a/src/llmq/signing_shares.cpp +++ b/src/llmq/signing_shares.cpp @@ -206,14 +206,14 @@ bool CSigSharesNodeState::GetSessionInfoByRecvId(uint32_t sessionId, SessionInfo return true; } -void CSigSharesNodeState::RemoveSession(const uint256& signHash) +size_t CSigSharesNodeState::RemoveSession(const uint256& signHash) { if (const auto it = sessions.find(signHash); it != sessions.end()) { sessionByRecvId.erase(it->second.recvSessionId); sessions.erase(it); } requestedSigShares.EraseAllForSignHash(signHash); - pendingIncomingSigShares.EraseAllForSignHash(signHash); + return pendingIncomingSigShares.EraseAllForSignHash(signHash); } ////////////////////// @@ -493,20 +493,20 @@ bool CSigSharesManager::TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesN __func__, MAX_PENDING_SIG_SHARES_PER_NODE, nodeId); return false; } - size_t total{0}; - for (const auto& [_, ns] : nodeStates) { - total += ns.pendingIncomingSigShares.Size(); - } - if (total >= MAX_PENDING_SIG_SHARES_TOTAL) { + if (m_pending_sig_shares_total >= MAX_PENDING_SIG_SHARES_TOTAL) { LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- global pending sig shares cap reached (%d), dropping sigShare. node=%d\n", __func__, MAX_PENDING_SIG_SHARES_TOTAL, nodeId); return false; } - return nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare); + if (!nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare)) { + return false; + } + ++m_pending_sig_shares_total; + return true; } bool CSigSharesManager::CollectPendingSigSharesToVerify( - size_t maxUniqueSessions, std::unordered_map>& retSigShares, + size_t maxShares, std::unordered_map>& retSigShares, std::unordered_map, CQuorumCPtr, StaticSaltedHasher>& retQuorums) { bool more_work{false}; @@ -517,16 +517,19 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify( return false; } - // This will iterate node states in random order and pick one sig share at a time. This avoids processing - // of large batches at once from the same node while other nodes also provided shares. If we wouldn't do this, - // other nodes would be able to poison us with a large batch with N-1 valid shares and the last one being - // invalid, making batch verification fail and revert to per-share verification, which in turn would slow down - // the whole verification process - std::unordered_set, StaticSaltedHasher> uniqueSignHashes; + // Iterate node states in random order and pick one sig share at a time. This ensures no single peer can + // dominate a batch and that a large flood from one peer cannot poison batch verification (an N-1 valid / + // 1 invalid batch would fall back to per-share verification and slow the whole pipeline). + // + // The batch is bounded by the number of shares actually added (maxShares), not by the count of unique + // (nodeId, signHash) sessions. Bounding by sessions could otherwise let a single session inflate the + // batch to the full pending-share cap, and, together with the in-flight batch cap, keep tens of thousands + // of shares outside the pending accounting. + size_t sharesAdded{0}; IterateNodesRandom( nodeStates, [&]() { - return uniqueSignHashes.size() < maxUniqueSessions; + return sharesAdded < maxShares; // TODO: remove NO_THREAD_SAFETY_ANALYSIS // using here template IterateNodesRandom makes impossible to use lock annotation }, @@ -538,10 +541,10 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify( AssertLockHeld(cs); if (const bool alreadyHave = this->sigShares.Has(sigShare.GetKey()); !alreadyHave) { - uniqueSignHashes.emplace(nodeId, sigShare.GetSignHash()); retSigShares[nodeId].emplace_back(sigShare); + ++sharesAdded; } - ns.pendingIncomingSigShares.Erase(sigShare.GetKey()); + m_pending_sig_shares_total -= ns.pendingIncomingSigShares.Erase(sigShare.GetKey()); return !ns.pendingIncomingSigShares.Empty(); }, rnd); @@ -1396,6 +1399,7 @@ void CSigSharesManager::Cleanup() AssertLockHeld(cs); sigSharesRequested.Erase(k); }); + m_pending_sig_shares_total -= it->second.pendingIncomingSigShares.Size(); nodeStates.erase(nodeId); } } @@ -1405,7 +1409,7 @@ void CSigSharesManager::RemoveSigSharesForSession(const uint256& signHash) AssertLockHeld(cs); for (auto& [_, nodeState] : nodeStates) { - nodeState.RemoveSession(signHash); + m_pending_sig_shares_total -= nodeState.RemoveSession(signHash); } sigSharesRequested.EraseAllForSignHash(signHash); @@ -1427,6 +1431,7 @@ void CSigSharesManager::RemoveNodesIf(std::function predicate) AssertLockHeld(cs); sigSharesRequested.Erase(k); }); + m_pending_sig_shares_total -= it->second.pendingIncomingSigShares.Size(); it = nodeStates.erase(it); } else { ++it; @@ -1455,7 +1460,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId) sigSharesRequested.Erase(k); }); nodeState.requestedSigShares.Clear(); - nodeState.pendingIncomingSigShares.Clear(); + m_pending_sig_shares_total -= nodeState.pendingIncomingSigShares.Clear(); nodeState.banned = true; } diff --git a/src/llmq/signing_shares.h b/src/llmq/signing_shares.h index 7e0e95ab714f..75da5061f28a 100644 --- a/src/llmq/signing_shares.h +++ b/src/llmq/signing_shares.h @@ -188,31 +188,36 @@ class CountedBucketMap return true; } - void Erase(const SigShareKey& k) + size_t Erase(const SigShareKey& k) { auto it = m_data.find(k.first); if (it == m_data.end()) { - return; + return 0; } - m_num_entries -= it->second.erase(k.second); + const size_t n = it->second.erase(k.second); + m_num_entries -= n; if (it->second.empty()) { m_data.erase(it); } + return n; } - void EraseBucket(const uint256& signHash) + size_t EraseBucket(const uint256& signHash) { auto it = m_data.find(signHash); if (it == m_data.end()) { - return; + return 0; } - m_num_entries -= it->second.size(); + const size_t n = it->second.size(); + m_num_entries -= n; m_data.erase(it); + return n; } template - void EraseIf(F&& f) + size_t EraseIf(F&& f) { + size_t erased{0}; for (auto it = m_data.begin(); it != m_data.end(); ) { SigShareKey k; k.first = it->first; @@ -221,6 +226,7 @@ class CountedBucketMap if (f(k, jt->second)) { jt = it->second.erase(jt); --m_num_entries; + ++erased; } else { ++jt; } @@ -231,12 +237,15 @@ class CountedBucketMap ++it; } } + return erased; } - void Clear() + size_t Clear() { + const size_t prev = m_num_entries; m_data.clear(); m_num_entries = 0; + return prev; } }; @@ -252,14 +261,14 @@ class SigShareMap return internalMap.Emplace(k, v); } - void Erase(const SigShareKey& k) + size_t Erase(const SigShareKey& k) { - internalMap.Erase(k); + return internalMap.Erase(k); } - void Clear() + size_t Clear() { - internalMap.Clear(); + return internalMap.Clear(); } [[nodiscard]] bool Has(const SigShareKey& k) const @@ -337,15 +346,15 @@ class SigShareMap return &it->second; } - void EraseAllForSignHash(const uint256& signHash) + size_t EraseAllForSignHash(const uint256& signHash) { - internalMap.EraseBucket(signHash); + return internalMap.EraseBucket(signHash); } template - void EraseIf(F&& f) + size_t EraseIf(F&& f) { - internalMap.EraseIf(f); + return internalMap.EraseIf(f); } template @@ -414,7 +423,10 @@ class CSigSharesNodeState Session* GetSessionByRecvId(uint32_t sessionId); bool GetSessionInfoByRecvId(uint32_t sessionId, SessionInfo& retInfo); - void RemoveSession(const uint256& signHash); + // Returns the number of pending incoming sig shares that were dropped + // as part of tearing this session down; other per-session structures + // are erased unconditionally. + size_t RemoveSession(const uint256& signHash); }; class CSignedSession @@ -459,6 +471,10 @@ class CSigSharesManager : public llmq::CRecoveredSigsListener mutable Mutex cs; SigShareMap sigShares GUARDED_BY(cs); + // Running total of shares held in nodeStates[*].pendingIncomingSigShares. + // Maintained explicitly so the global cap check in TryAddPendingIncomingSigShare + // doesn't need to walk every peer on every admission. + size_t m_pending_sig_shares_total GUARDED_BY(cs){0}; Uint256HashMap signedSessions GUARDED_BY(cs); // stores time of last receivedSigShare. Used to detect timeouts @@ -524,9 +540,13 @@ class CSigSharesManager : public llmq::CRecoveredSigsListener // if ProcessMessageSigShare returns false the node should be banned bool ProcessMessageSigShare(NodeId fromId, const CSigShare& sigShare) EXCLUSIVE_LOCKS_REQUIRED(!cs); - // CollectPendingSigSharesToVerify returns true if there's more work to do + // CollectPendingSigSharesToVerify returns true if there's more work to do. + // The returned batch contains at most maxShares actual sig shares, drawn one + // at a time in randomized round-robin order across peers so that no single + // peer can dominate a batch. Bounding by shares (not by unique sessions) + // caps the amount of BLS work that can be in-flight in the worker pool. bool CollectPendingSigSharesToVerify( - size_t maxUniqueSessions, std::unordered_map>& retSigShares, + size_t maxShares, std::unordered_map>& retSigShares, std::unordered_map, CQuorumCPtr, StaticSaltedHasher>& retQuorums) EXCLUSIVE_LOCKS_REQUIRED(!cs); diff --git a/src/test/llmq_utils_tests.cpp b/src/test/llmq_utils_tests.cpp index fb565c3fe1d4..f1292c722b33 100644 --- a/src/test/llmq_utils_tests.cpp +++ b/src/test/llmq_utils_tests.cpp @@ -114,23 +114,45 @@ BOOST_AUTO_TEST_CASE(sig_share_map_size_tracks_mutations) BOOST_CHECK(sig_share_map.Add(sig_share2.GetKey(), sig_share2)); BOOST_CHECK_EQUAL(sig_share_map.Size(), 2U); - sig_share_map.Erase(sig_share1.GetKey()); - sig_share_map.Erase(sig_share1.GetKey()); + BOOST_CHECK_EQUAL(sig_share_map.Erase(sig_share1.GetKey()), 1U); + BOOST_CHECK_EQUAL(sig_share_map.Erase(sig_share1.GetKey()), 0U); BOOST_CHECK_EQUAL(sig_share_map.Size(), 1U); - sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()); + BOOST_CHECK_EQUAL(sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()), 1U); + BOOST_CHECK_EQUAL(sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()), 0U); BOOST_CHECK_EQUAL(sig_share_map.Size(), 0U); BOOST_CHECK(sig_share_map.Empty()); BOOST_CHECK(sig_share_map.Add(sig_share1.GetKey(), sig_share1)); BOOST_CHECK(sig_share_map.Add(sig_share2.GetKey(), sig_share2)); - sig_share_map.EraseIf([&](const SigShareKey& k, const CSigShare&) { return k == sig_share1.GetKey(); }); + BOOST_CHECK_EQUAL(sig_share_map.EraseIf([&](const SigShareKey& k, const CSigShare&) { return k == sig_share1.GetKey(); }), 1U); BOOST_CHECK_EQUAL(sig_share_map.Size(), 1U); - sig_share_map.Clear(); + BOOST_CHECK_EQUAL(sig_share_map.Clear(), 1U); + BOOST_CHECK_EQUAL(sig_share_map.Clear(), 0U); BOOST_CHECK_EQUAL(sig_share_map.Size(), 0U); } +BOOST_AUTO_TEST_CASE(sig_share_map_bucket_erase_returns_bucket_size) +{ + // EraseAllForSignHash must report the total number of entries evicted for the bucket, + // so callers can maintain accurate running totals (e.g. the global pending sig-share cap). + SigShareMap sig_share_map; + const auto sign_hash = MakeSigShare(1).GetSignHash(); + + for (uint16_t member = 0; member < 5; ++member) { + CSigShare s{Consensus::LLMQType::LLMQ_50_60, GetTestQuorumHash(1), GetTestQuorumHash(2), GetTestQuorumHash(1), + member, CBLSLazySignature{}}; + s.UpdateKey(); + BOOST_CHECK_EQUAL(s.GetSignHash(), sign_hash); + BOOST_CHECK(sig_share_map.Add(s.GetKey(), s)); + } + BOOST_CHECK_EQUAL(sig_share_map.Size(), 5U); + + BOOST_CHECK_EQUAL(sig_share_map.EraseAllForSignHash(sign_hash), 5U); + BOOST_CHECK(sig_share_map.Empty()); +} + BOOST_AUTO_TEST_CASE(pending_sig_shares_session_removal_updates_count) { CSigSharesNodeState node_state; @@ -141,10 +163,18 @@ BOOST_AUTO_TEST_CASE(pending_sig_shares_session_removal_updates_count) BOOST_CHECK(node_state.pendingIncomingSigShares.Add(sig_share2.GetKey(), sig_share2)); BOOST_CHECK_EQUAL(node_state.pendingIncomingSigShares.Size(), 2U); - node_state.RemoveSession(sig_share1.GetSignHash()); + // RemoveSession must report the number of pending shares actually dropped so the manager's + // global pending-share counter can be decremented in lockstep. + BOOST_CHECK_EQUAL(node_state.RemoveSession(sig_share1.GetSignHash()), 1U); BOOST_CHECK_EQUAL(node_state.pendingIncomingSigShares.Size(), 1U); BOOST_CHECK(!node_state.pendingIncomingSigShares.Has(sig_share1.GetKey())); BOOST_CHECK(node_state.pendingIncomingSigShares.Has(sig_share2.GetKey())); + + // Removing the same session twice is idempotent and reports zero on the second call. + BOOST_CHECK_EQUAL(node_state.RemoveSession(sig_share1.GetSignHash()), 0U); + + // Removing a session that never had pending shares must also report zero. + BOOST_CHECK_EQUAL(node_state.RemoveSession(MakeSigShare(3).GetSignHash()), 0U); } BOOST_AUTO_TEST_CASE(deterministic_outbound_connection_test) From bdca48cde09153c54f67ef5b9b3faaf22be7f81d Mon Sep 17 00:00:00 2001 From: Konstantin Akimov Date: Fri, 10 Jul 2026 15:03:16 +0700 Subject: [PATCH 6/6] refactor: use CountedBucketMap for counting total amount of pending sigs --- src/llmq/signing_shares.cpp | 26 +++++++++--------- src/llmq/signing_shares.h | 50 ++++++++++++----------------------- src/test/llmq_utils_tests.cpp | 32 +++++++++------------- 3 files changed, 43 insertions(+), 65 deletions(-) diff --git a/src/llmq/signing_shares.cpp b/src/llmq/signing_shares.cpp index 8101dc70871e..940b8a3d9691 100644 --- a/src/llmq/signing_shares.cpp +++ b/src/llmq/signing_shares.cpp @@ -206,14 +206,14 @@ bool CSigSharesNodeState::GetSessionInfoByRecvId(uint32_t sessionId, SessionInfo return true; } -size_t CSigSharesNodeState::RemoveSession(const uint256& signHash) +void CSigSharesNodeState::RemoveSession(const uint256& signHash) { if (const auto it = sessions.find(signHash); it != sessions.end()) { sessionByRecvId.erase(it->second.recvSessionId); sessions.erase(it); } requestedSigShares.EraseAllForSignHash(signHash); - return pendingIncomingSigShares.EraseAllForSignHash(signHash); + pendingIncomingSigShares.EraseAllForSignHash(signHash); } ////////////////////// @@ -493,16 +493,18 @@ bool CSigSharesManager::TryAddPendingIncomingSigShare(NodeId nodeId, CSigSharesN __func__, MAX_PENDING_SIG_SHARES_PER_NODE, nodeId); return false; } - if (m_pending_sig_shares_total >= MAX_PENDING_SIG_SHARES_TOTAL) { + size_t total{0}; + for (const auto& [_, ns] : nodeStates) { + // the size of nodeStates is limited by DEFAULT_MAX_PEER_CONNECTIONS(125) so it should not be performance issue + // The name of variable is intentionally mentioned in comment to make this code snippet relevant for possible changes in future + total += ns.pendingIncomingSigShares.Size(); + } + if (total >= MAX_PENDING_SIG_SHARES_TOTAL) { LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- global pending sig shares cap reached (%d), dropping sigShare. node=%d\n", __func__, MAX_PENDING_SIG_SHARES_TOTAL, nodeId); return false; } - if (!nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare)) { - return false; - } - ++m_pending_sig_shares_total; - return true; + return nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare); } bool CSigSharesManager::CollectPendingSigSharesToVerify( @@ -544,7 +546,7 @@ bool CSigSharesManager::CollectPendingSigSharesToVerify( retSigShares[nodeId].emplace_back(sigShare); ++sharesAdded; } - m_pending_sig_shares_total -= ns.pendingIncomingSigShares.Erase(sigShare.GetKey()); + ns.pendingIncomingSigShares.Erase(sigShare.GetKey()); return !ns.pendingIncomingSigShares.Empty(); }, rnd); @@ -1399,7 +1401,6 @@ void CSigSharesManager::Cleanup() AssertLockHeld(cs); sigSharesRequested.Erase(k); }); - m_pending_sig_shares_total -= it->second.pendingIncomingSigShares.Size(); nodeStates.erase(nodeId); } } @@ -1409,7 +1410,7 @@ void CSigSharesManager::RemoveSigSharesForSession(const uint256& signHash) AssertLockHeld(cs); for (auto& [_, nodeState] : nodeStates) { - m_pending_sig_shares_total -= nodeState.RemoveSession(signHash); + nodeState.RemoveSession(signHash); } sigSharesRequested.EraseAllForSignHash(signHash); @@ -1431,7 +1432,6 @@ void CSigSharesManager::RemoveNodesIf(std::function predicate) AssertLockHeld(cs); sigSharesRequested.Erase(k); }); - m_pending_sig_shares_total -= it->second.pendingIncomingSigShares.Size(); it = nodeStates.erase(it); } else { ++it; @@ -1460,7 +1460,7 @@ void CSigSharesManager::MarkAsBanned(NodeId nodeId) sigSharesRequested.Erase(k); }); nodeState.requestedSigShares.Clear(); - m_pending_sig_shares_total -= nodeState.pendingIncomingSigShares.Clear(); + nodeState.pendingIncomingSigShares.Clear(); nodeState.banned = true; } diff --git a/src/llmq/signing_shares.h b/src/llmq/signing_shares.h index 75da5061f28a..332f553fc895 100644 --- a/src/llmq/signing_shares.h +++ b/src/llmq/signing_shares.h @@ -188,36 +188,31 @@ class CountedBucketMap return true; } - size_t Erase(const SigShareKey& k) + void Erase(const SigShareKey& k) { auto it = m_data.find(k.first); if (it == m_data.end()) { - return 0; + return; } - const size_t n = it->second.erase(k.second); - m_num_entries -= n; + m_num_entries -= it->second.erase(k.second); if (it->second.empty()) { m_data.erase(it); } - return n; } - size_t EraseBucket(const uint256& signHash) + void EraseBucket(const uint256& signHash) { auto it = m_data.find(signHash); if (it == m_data.end()) { - return 0; + return; } - const size_t n = it->second.size(); - m_num_entries -= n; + m_num_entries -= it->second.size(); m_data.erase(it); - return n; } template - size_t EraseIf(F&& f) + void EraseIf(F&& f) { - size_t erased{0}; for (auto it = m_data.begin(); it != m_data.end(); ) { SigShareKey k; k.first = it->first; @@ -226,7 +221,6 @@ class CountedBucketMap if (f(k, jt->second)) { jt = it->second.erase(jt); --m_num_entries; - ++erased; } else { ++jt; } @@ -237,15 +231,12 @@ class CountedBucketMap ++it; } } - return erased; } - size_t Clear() + void Clear() { - const size_t prev = m_num_entries; m_data.clear(); m_num_entries = 0; - return prev; } }; @@ -261,14 +252,14 @@ class SigShareMap return internalMap.Emplace(k, v); } - size_t Erase(const SigShareKey& k) + void Erase(const SigShareKey& k) { - return internalMap.Erase(k); + internalMap.Erase(k); } - size_t Clear() + void Clear() { - return internalMap.Clear(); + internalMap.Clear(); } [[nodiscard]] bool Has(const SigShareKey& k) const @@ -346,15 +337,15 @@ class SigShareMap return &it->second; } - size_t EraseAllForSignHash(const uint256& signHash) + void EraseAllForSignHash(const uint256& signHash) { - return internalMap.EraseBucket(signHash); + internalMap.EraseBucket(signHash); } template - size_t EraseIf(F&& f) + void EraseIf(F&& f) { - return internalMap.EraseIf(f); + internalMap.EraseIf(f); } template @@ -423,10 +414,7 @@ class CSigSharesNodeState Session* GetSessionByRecvId(uint32_t sessionId); bool GetSessionInfoByRecvId(uint32_t sessionId, SessionInfo& retInfo); - // Returns the number of pending incoming sig shares that were dropped - // as part of tearing this session down; other per-session structures - // are erased unconditionally. - size_t RemoveSession(const uint256& signHash); + void RemoveSession(const uint256& signHash); }; class CSignedSession @@ -471,10 +459,6 @@ class CSigSharesManager : public llmq::CRecoveredSigsListener mutable Mutex cs; SigShareMap sigShares GUARDED_BY(cs); - // Running total of shares held in nodeStates[*].pendingIncomingSigShares. - // Maintained explicitly so the global cap check in TryAddPendingIncomingSigShare - // doesn't need to walk every peer on every admission. - size_t m_pending_sig_shares_total GUARDED_BY(cs){0}; Uint256HashMap signedSessions GUARDED_BY(cs); // stores time of last receivedSigShare. Used to detect timeouts diff --git a/src/test/llmq_utils_tests.cpp b/src/test/llmq_utils_tests.cpp index f1292c722b33..839652b73f8e 100644 --- a/src/test/llmq_utils_tests.cpp +++ b/src/test/llmq_utils_tests.cpp @@ -114,29 +114,26 @@ BOOST_AUTO_TEST_CASE(sig_share_map_size_tracks_mutations) BOOST_CHECK(sig_share_map.Add(sig_share2.GetKey(), sig_share2)); BOOST_CHECK_EQUAL(sig_share_map.Size(), 2U); - BOOST_CHECK_EQUAL(sig_share_map.Erase(sig_share1.GetKey()), 1U); - BOOST_CHECK_EQUAL(sig_share_map.Erase(sig_share1.GetKey()), 0U); + sig_share_map.Erase(sig_share1.GetKey()); + sig_share_map.Erase(sig_share1.GetKey()); BOOST_CHECK_EQUAL(sig_share_map.Size(), 1U); - BOOST_CHECK_EQUAL(sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()), 1U); - BOOST_CHECK_EQUAL(sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()), 0U); + sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()); + sig_share_map.EraseAllForSignHash(sig_share2.GetSignHash()); BOOST_CHECK_EQUAL(sig_share_map.Size(), 0U); BOOST_CHECK(sig_share_map.Empty()); BOOST_CHECK(sig_share_map.Add(sig_share1.GetKey(), sig_share1)); BOOST_CHECK(sig_share_map.Add(sig_share2.GetKey(), sig_share2)); - BOOST_CHECK_EQUAL(sig_share_map.EraseIf([&](const SigShareKey& k, const CSigShare&) { return k == sig_share1.GetKey(); }), 1U); + sig_share_map.EraseIf([&](const SigShareKey& k, const CSigShare&) { return k == sig_share1.GetKey(); }); BOOST_CHECK_EQUAL(sig_share_map.Size(), 1U); - BOOST_CHECK_EQUAL(sig_share_map.Clear(), 1U); - BOOST_CHECK_EQUAL(sig_share_map.Clear(), 0U); + sig_share_map.Clear(); BOOST_CHECK_EQUAL(sig_share_map.Size(), 0U); } -BOOST_AUTO_TEST_CASE(sig_share_map_bucket_erase_returns_bucket_size) +BOOST_AUTO_TEST_CASE(sig_share_map_bucket_erase_updates_size) { - // EraseAllForSignHash must report the total number of entries evicted for the bucket, - // so callers can maintain accurate running totals (e.g. the global pending sig-share cap). SigShareMap sig_share_map; const auto sign_hash = MakeSigShare(1).GetSignHash(); @@ -149,7 +146,7 @@ BOOST_AUTO_TEST_CASE(sig_share_map_bucket_erase_returns_bucket_size) } BOOST_CHECK_EQUAL(sig_share_map.Size(), 5U); - BOOST_CHECK_EQUAL(sig_share_map.EraseAllForSignHash(sign_hash), 5U); + sig_share_map.EraseAllForSignHash(sign_hash); BOOST_CHECK(sig_share_map.Empty()); } @@ -163,18 +160,15 @@ BOOST_AUTO_TEST_CASE(pending_sig_shares_session_removal_updates_count) BOOST_CHECK(node_state.pendingIncomingSigShares.Add(sig_share2.GetKey(), sig_share2)); BOOST_CHECK_EQUAL(node_state.pendingIncomingSigShares.Size(), 2U); - // RemoveSession must report the number of pending shares actually dropped so the manager's - // global pending-share counter can be decremented in lockstep. - BOOST_CHECK_EQUAL(node_state.RemoveSession(sig_share1.GetSignHash()), 1U); + node_state.RemoveSession(sig_share1.GetSignHash()); BOOST_CHECK_EQUAL(node_state.pendingIncomingSigShares.Size(), 1U); BOOST_CHECK(!node_state.pendingIncomingSigShares.Has(sig_share1.GetKey())); BOOST_CHECK(node_state.pendingIncomingSigShares.Has(sig_share2.GetKey())); - // Removing the same session twice is idempotent and reports zero on the second call. - BOOST_CHECK_EQUAL(node_state.RemoveSession(sig_share1.GetSignHash()), 0U); - - // Removing a session that never had pending shares must also report zero. - BOOST_CHECK_EQUAL(node_state.RemoveSession(MakeSigShare(3).GetSignHash()), 0U); + // Removing the same session twice, or a session with no pending shares, is a no-op. + node_state.RemoveSession(sig_share1.GetSignHash()); + node_state.RemoveSession(MakeSigShare(3).GetSignHash()); + BOOST_CHECK_EQUAL(node_state.pendingIncomingSigShares.Size(), 1U); } BOOST_AUTO_TEST_CASE(deterministic_outbound_connection_test)