feat: add -parbls and validate async quorumSig and membersSig in QC

It introduces new commandline argument -parbls to set up amount of parallel threads for BLS validation

New parallel BlsChecker validates asynchonously quorumSig and membersSig in Quorum Commitment

Author: knst

src/init.cpp

@@ -529,6 +529,8 @@ void SetupServerArgs(ArgsManager& argsman)
     argsman.AddArg("-minimumchainwork=<hex>", strprintf("Minimum work assumed to exist on a valid chain in hex (default: %s, testnet: %s)", defaultChainParams->GetConsensus().nMinimumChainWork.GetHex(), testnetChainParams->GetConsensus().nMinimumChainWork.GetHex()), ArgsManager::ALLOW_ANY | ArgsManager::DEBUG_ONLY, OptionsCategory::OPTIONS);
     argsman.AddArg("-par=<n>", strprintf("Set the number of script verification threads (%u to %d, 0 = auto, <0 = leave that many cores free, default: %d)",
         -GetNumCores(), MAX_SCRIPTCHECK_THREADS, DEFAULT_SCRIPTCHECK_THREADS), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS);
+    argsman.AddArg("-parbls=<n>", strprintf("Set the number of bls verification threads (%u to %d, 0 = auto, <0 = leave that many cores free, default: %d)",
+        -GetNumCores(), llmq::MAX_BLSCHECK_THREADS, llmq::DEFAULT_BLSCHECK_THREADS), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS);
     argsman.AddArg("-persistmempool", strprintf("Whether to save the mempool on shutdown and load on restart (default: %u)", DEFAULT_PERSIST_MEMPOOL), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS);
     argsman.AddArg("-pid=<file>", strprintf("Specify pid file. Relative paths will be prefixed by a net-specific datadir location. (default: %s)", BITCOIN_PID_FILENAME), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS);
     argsman.AddArg("-prune=<n>", strprintf("Reduce storage requirements by enabling pruning (deleting) of old blocks. This allows the pruneblockchain RPC to be called to delete specific blocks, and enables automatic pruning of old blocks if a target size in MiB is provided. This mode is incompatible with -txindex, -rescan and -disablegovernance=false. "

src/llmq/blockprocessor.cpp

@@ -12,6 +12,7 @@
 
 #include <chain.h>
 #include <chainparams.h>
+#include <checkqueue.h>
 #include <consensus/params.h>
 #include <consensus/validation.h>
 #include <deploymentstatus.h>
@@ -52,8 +53,25 @@ CQuorumBlockProcessor::CQuorumBlockProcessor(CChainState& chainstate, CDetermini
     m_qsnapman(qsnapman)
 {
     utils::InitQuorumsCache(mapHasMinedCommitmentCache);
+
+    int bls_threads = gArgs.GetIntArg("-parbls", DEFAULT_BLSCHECK_THREADS);
+    if (bls_threads <= 0) {
+        // -parbls=0 means autodetect (number of cores - 1 validator threads)
+        // -parbls=-n means "leave n cores free" (number of cores - n - 1 validator threads)
+        bls_threads += GetNumCores();
+    }
+    // Subtract 1 because the main thread counts towards the par threads
+    bls_threads = std::max(bls_threads - 1, 0);
+
+    // Number of script-checking threads <= MAX_BLSCHECK_THREADS
+    bls_threads = std::min(bls_threads, MAX_BLSCHECK_THREADS);
+
+    LogPrintf("Bls verification uses %d additional threads\n", bls_threads);
+    m_bls_queue.StartWorkerThreads(bls_threads);
 }
 
+CQuorumBlockProcessor::~CQuorumBlockProcessor() { m_bls_queue.StopWorkerThreads(); }
+
 MessageProcessingResult CQuorumBlockProcessor::ProcessMessage(const CNode& peer, std::string_view msg_type,
                                                               CDataStream& vRecv)
 {
@@ -197,18 +215,18 @@ bool CQuorumBlockProcessor::ProcessBlock(const CBlock& block, gsl::not_null<cons
     }
 
     if (fBLSChecks) {
+        CCheckQueueControl<utils::BlsCheck> queue_control(&m_bls_queue);
         for (const auto& [_, qc] : qcs) {
             if (qc.IsNull()) continue;
             const auto* pQuorumBaseBlockIndex = m_chainstate.m_blockman.LookupBlockIndex(qc.quorumHash);
-            if (!qc.VerifySignature(m_dmnman, m_qsnapman, pQuorumBaseBlockIndex)) {
-                LogPrintf("[ProcessBlock] failed h[%d] llmqType[%d] version[%d] quorumIndex[%d] quorumHash[%s]\n",
-                          pindex->nHeight, ToUnderlying(qc.llmqType), qc.nVersion, qc.quorumIndex,
-                          qc.quorumHash.ToString());
-                return false;
-            }
+            qc.VerifySignatureAsync(m_dmnman, m_qsnapman, pQuorumBaseBlockIndex, &queue_control);
         }
-    }
 
+        if (!queue_control.Wait()) {
+            // at least one check failed
+            return false;
+        }
+    }
     for (const auto& [_, qc] : qcs) {
         if (!ProcessCommitment(pindex->nHeight, blockHash, qc, state, fJustCheck, false)) {
             LogPrintf("[ProcessBlock] failed h[%d] llmqType[%d] version[%d] quorumIndex[%d] quorumHash[%s]\n", pindex->nHeight, ToUnderlying(qc.llmqType), qc.nVersion, qc.quorumIndex, qc.quorumHash.ToString());

src/llmq/blockprocessor.h

@@ -7,7 +7,10 @@
 
 #include <unordered_lru_cache.h>
 
+#include <bls/bls.h>
+#include <checkqueue.h>
 #include <llmq/params.h>
+#include <llmq/utils.h>
 #include <protocol.h>
 #include <saltedhasher.h>
 #include <sync.h>
@@ -19,6 +22,7 @@
 class BlockValidationState;
 class CBlock;
 class CBlockIndex;
+class CBLSSignature;
 class CChain;
 class CChainState;
 class CDataStream;
@@ -41,6 +45,8 @@ class CQuorumBlockProcessor
     CEvoDB& m_evoDb;
     CQuorumSnapshotManager& m_qsnapman;
 
+    CCheckQueue<utils::BlsCheck> m_bls_queue{4};
+
     mutable Mutex minableCommitmentsCs;
     std::map<std::pair<Consensus::LLMQType, uint256>, uint256> minableCommitmentsByQuorum GUARDED_BY(minableCommitmentsCs);
     std::map<uint256, CFinalCommitment> minableCommitments GUARDED_BY(minableCommitmentsCs);
@@ -50,6 +56,7 @@ class CQuorumBlockProcessor
 public:
     explicit CQuorumBlockProcessor(CChainState& chainstate, CDeterministicMNManager& dmnman, CEvoDB& evoDb,
                                    CQuorumSnapshotManager& qsnapman);
+    ~CQuorumBlockProcessor();
 
     [[nodiscard]] MessageProcessingResult ProcessMessage(const CNode& peer, std::string_view msg_type, CDataStream& vRecv);
 

src/llmq/commitment.cpp

@@ -8,13 +8,14 @@
 #include <evo/specialtx.h>
 
 #include <chainparams.h>
+#include <checkqueue.h>
 #include <consensus/validation.h>
 #include <deploymentstatus.h>
 #include <llmq/options.h>
 #include <llmq/utils.h>
 #include <logging.h>
-#include <validation.h>
 #include <util/underlying.h>
+#include <validation.h>
 
 namespace llmq
 {
@@ -27,8 +28,9 @@ CFinalCommitment::CFinalCommitment(const Consensus::LLMQParams& params, const ui
 {
 }
 
-bool CFinalCommitment::VerifySignature(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
-                                       gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex) const
+bool CFinalCommitment::VerifySignatureAsync(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
+                                            gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex,
+                                            CCheckQueueControl<utils::BlsCheck>* queue_control) const
 {
     auto members = utils::GetAllQuorumMembers(llmqType, dmnman, qsnapman, pQuorumBaseBlockIndex);
     const auto& llmq_params_opt = Params().GetLLMQ(llmqType);
@@ -63,16 +65,31 @@ bool CFinalCommitment::VerifySignature(CDeterministicMNManager& dmnman, CQuorumS
             }
             memberPubKeys.emplace_back(members[i]->pdmnState->pubKeyOperator.Get());
         }
-
-        if (!membersSig.VerifySecureAggregated(memberPubKeys, commitmentHash)) {
-            LogPrint(BCLog::LLMQ, "CFinalCommitment -- q[%s] invalid aggregated members signature\n",
-                     quorumHash.ToString());
-            return false;
+        std::string members_id_string{
+            strprintf("CFinalCommitment -- q[%s] invalid aggregated members signature", quorumHash.ToString())};
+        if (queue_control) {
+            std::vector<utils::BlsCheck> vChecks;
+            vChecks.emplace_back(membersSig, memberPubKeys, commitmentHash, members_id_string);
+            queue_control->Add(vChecks);
+        } else {
+            if (!membersSig.VerifySecureAggregated(memberPubKeys, commitmentHash)) {
+                LogPrint(BCLog::LLMQ, "%s\n", members_id_string);
+                return false;
+            }
         }
     }
-    if (!quorumSig.VerifyInsecure(quorumPublicKey, commitmentHash)) {
-        LogPrint(BCLog::LLMQ, "CFinalCommitment -- q[%s] invalid quorum signature\n", quorumHash.ToString());
-        return false;
+    std::string qsig_id_string{strprintf("CFinalCommitment -- q[%s] invalid quorum signature", quorumHash.ToString())};
+    if (queue_control) {
+        std::vector<utils::BlsCheck> vChecks;
+        std::vector<CBLSPublicKey> public_keys;
+        public_keys.push_back(quorumPublicKey);
+        vChecks.emplace_back(quorumSig, public_keys, commitmentHash, qsig_id_string);
+        queue_control->Add(vChecks);
+    } else {
+        if (!quorumSig.VerifyInsecure(quorumPublicKey, commitmentHash)) {
+            LogPrint(BCLog::LLMQ, "%s\n", qsig_id_string);
+            return false;
+        }
     }
     return true;
 }
@@ -157,7 +174,7 @@ bool CFinalCommitment::Verify(CDeterministicMNManager& dmnman, CQuorumSnapshotMa
 
     // sigs are only checked when the block is processed
     if (checkSigs) {
-        if (!VerifySignature(dmnman, qsnapman, pQuorumBaseBlockIndex)) {
+        if (!VerifySignatureAsync(dmnman, qsnapman, pQuorumBaseBlockIndex, nullptr)) {
             return false;
         }
     }

src/llmq/commitment.h

@@ -25,11 +25,16 @@ class CBlockIndex;
 class CDeterministicMNManager;
 class ChainstateManager;
 class TxValidationState;
+template <typename T>
+class CCheckQueueControl;
 
 namespace llmq
 {
 class CQuorumSnapshotManager;
 
+namespace utils {
+struct BlsCheck;
+} // namespace utils
 // This message is an aggregation of all received premature commitments and only valid if
 // enough (>=threshold) premature commitments were aggregated
 // This is mined on-chain as part of TRANSACTION_QUORUM_COMMITMENT
@@ -67,8 +72,9 @@ class CFinalCommitment
         return int(std::count(validMembers.begin(), validMembers.end(), true));
     }
 
-    bool VerifySignature(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
-                         gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex) const;
+    bool VerifySignatureAsync(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
+                              gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex,
+                              CCheckQueueControl<utils::BlsCheck>* queue_control) const;
     bool Verify(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
                 gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex, bool checkSigs) const;
     bool VerifyNull() const;

src/llmq/options.h

@@ -24,6 +24,11 @@ enum class QvvecSyncMode {
     OnlyIfTypeMember = 1,
 };
 
+/** Maximum number of dedicated script-checking threads allowed */
+static const int MAX_BLSCHECK_THREADS = 31;
+/** -parbls default (number of bls-checking threads, 0 = auto) */
+static const int DEFAULT_BLSCHECK_THREADS = 0;
+
 static constexpr bool DEFAULT_ENABLE_QUORUM_DATA_RECOVERY{true};
 
 // If true, we will connect to all new quorums and watch their communication

src/llmq/utils.cpp

@@ -940,6 +940,25 @@ void AddQuorumProbeConnections(const Consensus::LLMQParams& llmqParams, CConnman
     }
 }
 
+bool BlsCheck::operator()()
+{
+    if (m_pubkeys.size() > 1) {
+        if (!m_sig.VerifySecureAggregated(m_pubkeys, m_msg_hash)) {
+            LogPrint(BCLog::LLMQ, "%s\n", m_id_string);
+            return false;
+        }
+    } else if (m_pubkeys.size() == 1) {
+        if (!m_sig.VerifyInsecure(m_pubkeys.back(), m_msg_hash)) {
+            LogPrint(BCLog::LLMQ, "%s\n", m_id_string);
+            return false;
+        }
+    } else {
+        // It is supposed to be at least one public key!
+        return false;
+    }
+    return true;
+}
+
 template <typename CacheType>
 void InitQuorumsCache(CacheType& cache, bool limit_by_connections)
 {

src/llmq/utils.h

@@ -5,6 +5,7 @@
 #ifndef BITCOIN_LLMQ_UTILS_H
 #define BITCOIN_LLMQ_UTILS_H
 
+#include <bls/bls.h>
 #include <gsl/pointers.h>
 #include <llmq/params.h>
 #include <saltedhasher.h>
@@ -56,8 +57,36 @@ void AddQuorumProbeConnections(const Consensus::LLMQParams& llmqParams, CConnman
                                const CSporkManager& sporkman, const CDeterministicMNList& tip_mn_list,
                                gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex, const uint256& myProTxHash);
 
+struct BlsCheck {
+    CBLSSignature m_sig;
+    std::vector<CBLSPublicKey> m_pubkeys;
+    uint256 m_msg_hash;
+    std::string m_id_string;
+
+    BlsCheck() = default;
+
+    BlsCheck(CBLSSignature sig, std::vector<CBLSPublicKey> pubkeys, uint256 msg_hash, std::string id_string) :
+        m_sig(sig),
+        m_pubkeys(pubkeys),
+        m_msg_hash(msg_hash),
+        m_id_string(id_string)
+    {
+    }
+
+    void swap(BlsCheck& obj)
+    {
+        std::swap(m_sig, obj.m_sig);
+        std::swap(m_pubkeys, obj.m_pubkeys);
+        std::swap(m_msg_hash, obj.m_msg_hash);
+        std::swap(m_id_string, obj.m_id_string);
+    }
+
+    bool operator()();
+};
+
 template <typename CacheType>
 void InitQuorumsCache(CacheType& cache, bool limit_by_connections = true);
+
 } // namespace utils
 } // namespace llmq