Merge dashpay#7225: fix: collect sigshares synchronously in dispatcher for parallel verification

018298b fix: collect sigshares synchronously in dispatcher for parallel verification (PastaClaw)

Pull request description:

  ## Summary

  The dispatcher thread in `WorkThreadDispatcher()` unconditionally enqueued a
  `ProcessPendingSigShares` worker into the thread pool every 10ms whenever any
  `pendingIncomingSigShares` existed. Since the workers would then find nothing
  to do and immediately exit, this wasted allocations and queue operations.

  This was flagged during review of dashpay#7004 by CodeRabbit as a nitpick but was not
  addressed at the time.

  ## Fix

  Move `CollectPendingSigSharesToVerify` out of `ProcessPendingSigShares` and
  into the dispatcher loop, where it runs synchronously (it holds `cs` briefly
  and does no BLS work). Each collected batch is moved into its own worker for
  parallel BLS verification.

  This approach (suggested knst) has two advantages over the initial
  atomic-guard implementation:
  1. **Parallel verification**: Multiple verification batches can run
     concurrently across the thread pool (up to 4 workers), instead of being
     serialised to a single worker.
  2. **Natural queue bounding**: Each `Collect` call dequeues its batch from
     the pending queue, so only as many tasks are enqueued as there are actual
     batches of work — no redundant empty-exit tasks.

  **Performance impact: slight improvement.** Multiple BLS verification batches
  can now run in parallel when there are more than 32 pending sig shares
  (`nMaxBatchSize`). The collection step is cheap (no BLS ops).

  ## Validation

  - Code review of the dispatch path confirms collect-then-dispatch preserves
    the existing drain behavior
  - Each `CollectPendingSigSharesToVerify` call removes its batch from the
    pending queue, preventing duplicate work across workers
  - No functional change to sigshare processing semantics

ACKs for top commit:
  UdjinM6:
    utACK 018298b
  knst:
    utACK 018298b

Tree-SHA512: 667f510f7b4a75025cf840bc2835895bd09c25563da1617116b5bb721bf950e9823a4b911f6e86b5a6b8096e70d08d8238e2a2a80f5c2f06db221bdc2a0b2b68

Author: PastaPastaPasta

src/llmq/net_signing.cpp

@@ -306,17 +306,25 @@ void NetSigning::WorkThreadDispatcher()
             }
         }
 
-        if (m_shares_manager->IsAnyPendingProcessing()) {
-            // If there's processing work, spawn a helper worker
-            worker_pool.push([this](int) {
-                while (!workInterrupt) {
-                    bool moreWork = ProcessPendingSigShares();
-
-                    if (!moreWork) {
-                        return; // No work found, exit immediately
-                    }
-                }
+        // Collect pending sig shares synchronously and dispatch each batch to a worker for parallel BLS verification
+        while (!workInterrupt) {
+            std::unordered_map<NodeId, std::vector<CSigShare>> sigSharesByNodes;
+            std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher> quorums;
+
+            const size_t nMaxBatchSize{32};
+            bool more_work = m_shares_manager->CollectPendingSigSharesToVerify(nMaxBatchSize, sigSharesByNodes, quorums);
+
+            if (sigSharesByNodes.empty()) {
+                break;
+            }
+
+            worker_pool.push([this, sigSharesByNodes = std::move(sigSharesByNodes), quorums = std::move(quorums)](int) mutable {
+                ProcessPendingSigShares(std::move(sigSharesByNodes), std::move(quorums));
             });
+
+            if (!more_work) {
+                break;
+            }
         }
 
         // Always sleep briefly between checks
@@ -329,17 +337,10 @@ void NetSigning::NotifyRecoveredSig(const std::shared_ptr<const CRecoveredSig>&
     m_peer_manager->PeerRelayRecoveredSig(*sig, proactive_relay);
 }
 
-bool NetSigning::ProcessPendingSigShares()
+void NetSigning::ProcessPendingSigShares(
+    std::unordered_map<NodeId, std::vector<CSigShare>>&& sigSharesByNodes,
+    std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>&& quorums)
 {
-    std::unordered_map<NodeId, std::vector<CSigShare>> sigSharesByNodes;
-    std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher> quorums;
-
-    const size_t nMaxBatchSize{32};
-    bool more_work = m_shares_manager->CollectPendingSigSharesToVerify(nMaxBatchSize, sigSharesByNodes, quorums);
-    if (sigSharesByNodes.empty()) {
-        return false;
-    }
-
     // It's ok to perform insecure batched verification here as we verify against the quorum public key shares,
     // which are not craftable by individual entities, making the rogue public key attack impossible
     CBLSBatchVerifier<NodeId, SigShareKey> batchVerifier(false, true);
@@ -400,8 +401,6 @@ bool NetSigning::ProcessPendingSigShares()
             ProcessRecoveredSig(rs, true);
         }
     }
-
-    return more_work;
 }
 
 } // namespace llmq

src/llmq/net_signing.h

@@ -6,6 +6,8 @@
 #define BITCOIN_LLMQ_NET_SIGNING_H
 
 #include <ctpl_stl.h>
+#include <llmq/signing_shares.h>
+#include <llmq/types.h>
 #include <net_processing.h>
 #include <util/threadinterrupt.h>
 #include <util/time.h>
@@ -52,7 +54,9 @@ class NetSigning final : public NetHandler, public CValidationInterface
     void WorkThreadCleaning();
     void WorkThreadDispatcher();
 
-    bool ProcessPendingSigShares();
+    void ProcessPendingSigShares(
+        std::unordered_map<NodeId, std::vector<CSigShare>>&& sigSharesByNodes,
+        std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>&& quorums);
 
     void RemoveBannedNodeStates();
     void BanNode(NodeId nodeid);