diff --git a/src/llmq/net_signing.cpp b/src/llmq/net_signing.cpp index b8f937b634bb..3b622e5b05aa 100644 --- a/src/llmq/net_signing.cpp +++ b/src/llmq/net_signing.cpp @@ -320,19 +320,31 @@ 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. + // + // 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; } + ++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/llmq/signing_shares.cpp b/src/llmq/signing_shares.cpp index 4849e2940ee9..940b8a3d9691 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,8 +480,35 @@ 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) { + // 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; + } + return nodeState.pendingIncomingSigShares.Add(sigShare.GetKey(), sigShare); +} + 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}; @@ -487,16 +519,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 }, @@ -508,8 +543,8 @@ 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()); return !ns.pendingIncomingSigShares.Empty(); @@ -1425,6 +1460,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 ef8aa42edc0f..332f553fc895 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() { - internalMap.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(); } [[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) { @@ -464,9 +524,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); @@ -484,6 +548,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); diff --git a/src/test/llmq_utils_tests.cpp b/src/test/llmq_utils_tests.cpp index ea51601a8b5a..839652b73f8e 100644 --- a/src/test/llmq_utils_tests.cpp +++ b/src/test/llmq_utils_tests.cpp @@ -103,6 +103,74 @@ 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()); + 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(sig_share_map_bucket_erase_updates_size) +{ + 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); + + sig_share_map.EraseAllForSignHash(sign_hash); + BOOST_CHECK(sig_share_map.Empty()); +} + +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())); + + // 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) { // Test deterministic behavior