perf: using asynchronous worker to validate BLS signatures in quorum commitments

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)
 {
@@ -196,8 +214,21 @@ 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);
+            qc.VerifySignatureAsync(m_dmnman, m_qsnapman, pQuorumBaseBlockIndex, &queue_control);
+        }
+
+        if (!queue_control.Wait()) {
+            // at least one check failed
+            return state.Invalid(BlockValidationResult::BLOCK_CONSENSUS, "bad-qc-invalid");
+        }
+    }
     for (const auto& [_, qc] : qcs) {
-        if (!ProcessCommitment(pindex->nHeight, blockHash, qc, state, fJustCheck, fBLSChecks)) {
+        if (!ProcessCommitment(pindex->nHeight, blockHash, qc, state, fJustCheck)) {
             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;
         }
@@ -237,7 +268,8 @@ static bool IsMiningPhase(const Consensus::LLMQParams& llmqParams, const CChain&
     return nHeight >= quorumCycleMiningStartHeight && nHeight <= quorumCycleMiningEndHeight;
 }
 
-bool CQuorumBlockProcessor::ProcessCommitment(int nHeight, const uint256& blockHash, const CFinalCommitment& qc, BlockValidationState& state, bool fJustCheck, bool fBLSChecks)
+bool CQuorumBlockProcessor::ProcessCommitment(int nHeight, const uint256& blockHash, const CFinalCommitment& qc,
+                                              BlockValidationState& state, bool fJustCheck)
 {
     AssertLockHeld(::cs_main);
 
@@ -303,7 +335,8 @@ bool CQuorumBlockProcessor::ProcessCommitment(int nHeight, const uint256& blockH
 
     const auto* pQuorumBaseBlockIndex = m_chainstate.m_blockman.LookupBlockIndex(qc.quorumHash);
 
-    if (!qc.Verify(m_dmnman, m_qsnapman, pQuorumBaseBlockIndex, /*checkSigs=*/fBLSChecks)) {
+    // we don't validate signatures here; they already validated on previous step
+    if (!qc.Verify(m_dmnman, m_qsnapman, pQuorumBaseBlockIndex, /*checksigs=*/false)) {
         LogPrint(BCLog::LLMQ, /* Continued */
                  "%s -- height=%d, type=%d, quorumIndex=%d, quorumHash=%s, signers=%s, validMembers=%d, "
                  "quorumPublicKey=%s qc verify failed.\n",

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);
 
@@ -75,7 +82,8 @@ class CQuorumBlockProcessor
     std::optional<const CBlockIndex*> GetLastMinedCommitmentsByQuorumIndexUntilBlock(Consensus::LLMQType llmqType, const CBlockIndex* pindex, int quorumIndex, size_t cycle) const;
 private:
     static bool GetCommitmentsFromBlock(const CBlock& block, gsl::not_null<const CBlockIndex*> pindex, std::multimap<Consensus::LLMQType, CFinalCommitment>& ret, BlockValidationState& state) EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
-    bool ProcessCommitment(int nHeight, const uint256& blockHash, const CFinalCommitment& qc, BlockValidationState& state, bool fJustCheck, bool fBLSChecks) EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
+    bool ProcessCommitment(int nHeight, const uint256& blockHash, const CFinalCommitment& qc,
+                           BlockValidationState& state, bool fJustCheck) EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
     size_t GetNumCommitmentsRequired(const Consensus::LLMQParams& llmqParams, int nHeight) const EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
     static uint256 GetQuorumBlockHash(const Consensus::LLMQParams& llmqParams, const CChain& active_chain, int nHeight, int quorumIndex) EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
 };

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,6 +28,73 @@ CFinalCommitment::CFinalCommitment(const Consensus::LLMQParams& params, const ui
 {
 }
 
+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);
+    if (!llmq_params_opt.has_value()) {
+        LogPrint(BCLog::LLMQ, "CFinalCommitment -- q[%s] invalid llmqType=%d\n", quorumHash.ToString(),
+                 ToUnderlying(llmqType));
+        return false;
+    }
+    const auto& llmq_params = llmq_params_opt.value();
+
+    uint256 commitmentHash = BuildCommitmentHash(llmq_params.type, quorumHash, validMembers, quorumPublicKey,
+                                                 quorumVvecHash);
+    if (LogAcceptDebug(BCLog::LLMQ)) {
+        std::stringstream ss3;
+        for (const auto& mn : members) {
+            ss3 << mn->proTxHash.ToString().substr(0, 4) << " | ";
+        }
+        LogPrint(BCLog::LLMQ, "CFinalCommitment::%s members[%s] quorumPublicKey[%s] commitmentHash[%s]\n", __func__,
+                 ss3.str(), quorumPublicKey.ToString(), commitmentHash.ToString());
+    }
+    if (llmq_params.size == 1) {
+        LogPrintf("pubkey operator: %s\n", members[0]->pdmnState->pubKeyOperator.Get().ToString());
+        if (!membersSig.VerifyInsecure(members[0]->pdmnState->pubKeyOperator.Get(), commitmentHash)) {
+            LogPrint(BCLog::LLMQ, "CFinalCommitment -- q[%s] invalid member signature\n", quorumHash.ToString());
+            return false;
+        }
+    } else {
+        std::vector<CBLSPublicKey> memberPubKeys;
+        for (const auto i : irange::range(members.size())) {
+            if (!signers[i]) {
+                continue;
+            }
+            memberPubKeys.emplace_back(members[i]->pdmnState->pubKeyOperator.Get());
+        }
+        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;
+            }
+        }
+    }
+    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;
+}
+
+
 bool CFinalCommitment::Verify(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
                               gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex, bool checkSigs) const
 {
@@ -106,38 +174,7 @@ bool CFinalCommitment::Verify(CDeterministicMNManager& dmnman, CQuorumSnapshotMa
 
     // sigs are only checked when the block is processed
     if (checkSigs) {
-        uint256 commitmentHash = BuildCommitmentHash(llmq_params.type, quorumHash, validMembers, quorumPublicKey, quorumVvecHash);
-        if (LogAcceptDebug(BCLog::LLMQ)) {
-            std::stringstream ss3;
-            for (const auto &mn: members) {
-                ss3 << mn->proTxHash.ToString().substr(0, 4) << " | ";
-            }
-            LogPrint(BCLog::LLMQ, "CFinalCommitment::%s members[%s] quorumPublicKey[%s] commitmentHash[%s]\n",
-                                     __func__, ss3.str(), quorumPublicKey.ToString(), commitmentHash.ToString());
-        }
-        if (llmq_params.size == 1) {
-            LogPrintf("pubkey operator: %s\n", members[0]->pdmnState->pubKeyOperator.Get().ToString());
-            if (!membersSig.VerifyInsecure(members[0]->pdmnState->pubKeyOperator.Get(), commitmentHash)) {
-                LogPrint(BCLog::LLMQ, "CFinalCommitment -- q[%s] invalid member signature\n", quorumHash.ToString());
-                return false;
-            }
-        } else {
-            std::vector<CBLSPublicKey> memberPubKeys;
-            for (const auto i : irange::range(members.size())) {
-                if (!signers[i]) {
-                    continue;
-                }
-                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;
-            }
-        }
-        if (!quorumSig.VerifyInsecure(quorumPublicKey, commitmentHash)) {
-            LogPrint(BCLog::LLMQ, "CFinalCommitment -- q[%s] invalid quorum signature\n", quorumHash.ToString());
+        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,6 +72,9 @@ class CFinalCommitment
         return int(std::count(validMembers.begin(), validMembers.end(), true));
     }
 
+    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 = 33;
+/** -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,26 @@ 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 {
+        // we should not get there ever
+        LogPrint(BCLog::LLMQ, "%s - no public keys are provided\n", m_id_string);
+        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