perf: warm quorum public key shares on the BLS worker pool

CQuorumManager ran a dedicated thread that polled a queue every 100 ms only to precompute the public key shares of freshly built quorums.
That is BLS work with no ordering or latency requirement of its own, so push each quorum as a job to the CBLSWorker pool instead.
One thread less per node (0.9 MB of thread-local storage each while the BLS library keeps its context per thread),
no idle polling, and the queue, its mutex and the interrupt go away with it.

A job only holds a shared_ptr to its quorum, so it needs nothing from the manager and Stop() of the pool drains it on shutdown.

Author: knst

src/bls/bls_worker.cpp

@@ -69,6 +69,11 @@ void CBLSWorker::Stop()
     workerPool.stop(true);
 }
 
+void CBLSWorker::PushJob(std::function<void()> job)
+{
+    workerPool.push([job = std::move(job)](int) { job(); });
+}
+
 #ifndef BUILD_BITCOIN_INTERNAL
 bool CBLSWorker::GenerateContributions(int quorumThreshold, Span<CBLSId> ids, BLSVerificationVectorPtr& vvecRet, std::vector<CBLSSecretKey>& skSharesRet)
 {

src/bls/bls_worker.h

@@ -56,6 +56,7 @@ class CBLSWorker
 
     void Start(int16_t worker_count);
     void Stop();
+    void PushJob(std::function<void()> job);
 
 #ifndef BUILD_BITCOIN_INTERNAL
     bool GenerateContributions(int threshold, Span<CBLSId> ids, BLSVerificationVectorPtr& vvecRet, std::vector<CBLSSecretKey>& skSharesRet);

src/llmq/context.cpp

@@ -28,5 +28,6 @@ LLMQContext::LLMQContext(CDeterministicMNManager& dmnman, CEvoDB& evo_db, Chains
 
 LLMQContext::~LLMQContext()
 {
+    qman->InterruptWarming();
     bls_worker->Stop();
 }

src/llmq/quorumsman.cpp

@@ -22,7 +22,6 @@
 #include <chainparams.h>
 #include <dbwrapper.h>
 #include <logging.h>
-#include <util/thread.h>
 #include <util/time.h>
 #include <validation.h>
 
@@ -40,18 +39,10 @@ CQuorumManager::CQuorumManager(CBLSWorker& _blsWorker, CDeterministicMNManager&
     db{util::MakeDbWrapper({db_params.path / "llmq" / "quorumdb", db_params.memory, db_params.wipe, /*cache_size=*/1 << 20})}
 {
     mapQuorumsCache.Init(m_chainman.GetConsensus(), /*limit_by_connections=*/false);
-    m_cache_interrupt.reset();
-    m_cache_thread = std::thread(&util::TraceThread, "q-cache", [this] { CacheWarmingThreadMain(); });
     MigrateOldQuorumDB(_evoDb);
 }
 
-CQuorumManager::~CQuorumManager()
-{
-    if (m_cache_thread.joinable()) {
-        m_cache_interrupt();
-        m_cache_thread.join();
-    }
-}
+CQuorumManager::~CQuorumManager() = default;
 
 bool CQuorumManager::GetEncryptedContributions(Consensus::LLMQType llmq_type, const CBlockIndex* block_index,
                                                const std::vector<bool>& valid_members, const uint256& protx_hash,
@@ -505,50 +496,29 @@ void CQuorumManager::WriteContributions(const CQuorumPtr& quorum) const
     quorum->WriteContributions(*db);
 }
 
-void CQuorumManager::CacheWarmingThreadMain() const
+void CQuorumManager::QueueQuorumForWarming(CQuorumCPtr pQuorum) const
 {
-    while (!m_cache_interrupt) {
-        CQuorumCPtr pQuorum;
-        {
-            LOCK(m_cache_cs);
-            if (!m_cache_queue.empty()) {
-                pQuorum = std::move(m_cache_queue.front());
-                m_cache_queue.pop_front();
-            };
-        }
-
-        if (!pQuorum) {
-            m_cache_interrupt.sleep_for(std::chrono::milliseconds(100));
-            continue;
-        }
-
+    if (!pQuorum->HasVerificationVector()) {
+        return;
+    }
+    // The job may reference this manager: ~LLMQContext stops the worker pool before destroying it.
+    blsWorker.PushJob([this, pQuorum = std::move(pQuorum)]() {
         cxxtimer::Timer t(true);
-        LogPrint(BCLog::LLMQ, "CQuorumManager::%s -- type=%d height=%d hash=%s start\n", __func__,
+        LogPrint(BCLog::LLMQ, "CQuorumManager::QueueQuorumForWarming -- type=%d height=%d hash=%s start\n",
                  std23::to_underlying(pQuorum->params.type), pQuorum->m_quorum_base_block_index->nHeight,
                  pQuorum->m_quorum_base_block_index->GetBlockHash().ToString());
-
-        // when then later some other thread tries to get keys, it will be much faster
         for (const auto i : util::irange(pQuorum->members.size())) {
-            if (m_cache_interrupt) {
-                break;
+            if (m_warming_interrupted) {
+                return;
             }
             if (pQuorum->qc->validMembers[i]) {
                 pQuorum->GetPubKeyShare(i);
             }
         }
-
-        LogPrint(BCLog::LLMQ, "CQuorumManager::%s -- type=%d height=%d hash=%s done. time=%d\n", __func__,
+        LogPrint(BCLog::LLMQ, "CQuorumManager::QueueQuorumForWarming -- type=%d height=%d hash=%s done. time=%d\n",
                  std23::to_underlying(pQuorum->params.type), pQuorum->m_quorum_base_block_index->nHeight,
                  pQuorum->m_quorum_base_block_index->GetBlockHash().ToString(), t.count());
-    }
-}
-
-void CQuorumManager::QueueQuorumForWarming(CQuorumCPtr pQuorum) const
-{
-    if (pQuorum->HasVerificationVector()) {
-        LOCK(m_cache_cs);
-        m_cache_queue.push_back(std::move(pQuorum));
-    }
+    });
 }
 
 // TODO: remove in v23

src/llmq/quorumsman.h

@@ -15,14 +15,12 @@
 #include <unordered_lru_cache.h>
 
 #include <sync.h>
-#include <util/threadinterrupt.h>
 
 #include <gsl/pointers.h>
 
-#include <deque>
+#include <atomic>
 #include <map>
 #include <memory>
-#include <thread>
 
 class CBLSSignature;
 class CBLSWorker;
@@ -96,10 +94,7 @@ class CQuorumManager final
     mutable Uint256LruHashMap<const CBlockIndex*, /*max_size=*/128> quorumBaseBlockIndexCache
         GUARDED_BY(cs_quorumBaseBlockIndexCache);
 
-    mutable Mutex m_cache_cs;
-    mutable std::deque<CQuorumCPtr> m_cache_queue GUARDED_BY(m_cache_cs);
-    mutable CThreadInterrupt m_cache_interrupt;
-    mutable std::thread m_cache_thread;
+    std::atomic<bool> m_warming_interrupted{false};
 
 public:
     CQuorumManager() = delete;
@@ -135,20 +130,20 @@ class CQuorumManager final
 
     // all these methods will lock cs_main for a short period of time
     CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, const uint256& quorumHash) const
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps);
     CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, const uint256& quorumHash, const CChain& chain) const
-        EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps);
     std::vector<CQuorumCPtr> ScanQuorums(Consensus::LLMQType llmqType, size_t nCountRequested) const
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps);
 
     // this one is cs_main-free
     std::vector<CQuorumCPtr> ScanQuorums(Consensus::LLMQType llmqType, gsl::not_null<const CBlockIndex*> pindexStart,
                                          size_t nCountRequested) const
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps);
     std::vector<CQuorumCPtr> ScanQuorums(Consensus::LLMQType llmqType,
                                          gsl::not_null<const CBlockIndex*> pindexStart,
                                          size_t nCountRequested, const CChain& chain) const
-        EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps);
 
     bool IsMasternode() const;
     bool IsWatching() const;
@@ -175,29 +170,30 @@ class CQuorumManager final
     CQuorumPtr GetCachedMutableQuorum(Consensus::LLMQType llmqType, const uint256& quorumHash) const
         EXCLUSIVE_LOCKS_REQUIRED(!m_cs_maps);
     void WriteContributions(const CQuorumPtr& quorum) const EXCLUSIVE_LOCKS_REQUIRED(!cs_db);
-    void QueueQuorumForWarming(CQuorumCPtr pQuorum) const EXCLUSIVE_LOCKS_REQUIRED(!m_cache_cs);
+    void QueueQuorumForWarming(CQuorumCPtr pQuorum) const;
+    /** Make queued and running warming jobs return; call before stopping the BLS worker pool */
+    void InterruptWarming() { m_warming_interrupted = true; }
 
 private:
     // all private methods here are cs_main-free
     std::vector<CQuorumCPtr> ScanQuorums(Consensus::LLMQType llmqType,
                                          gsl::not_null<const CBlockIndex*> pindexStart,
                                          size_t nCountRequested, const CChain* chain) const
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps);
     bool BuildQuorumContributions(const CFinalCommitmentPtr& fqc, const std::shared_ptr<CQuorum>& quorum) const;
 
     CQuorumPtr BuildQuorumFromCommitment(Consensus::LLMQType llmqType,
                                          gsl::not_null<const CBlockIndex*> pQuorumBaseBlockIndex,
                                          bool populate_cache) const
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps);
 
     CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, gsl::not_null<const CBlockIndex*> pindex,
                           bool populate_cache = true) const
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_db, !m_cs_maps);
     CQuorumCPtr GetQuorum(Consensus::LLMQType llmqType, gsl::not_null<const CBlockIndex*> pindex,
                           const CChain& chain, bool populate_cache = true) const
-        EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps, !m_cache_cs);
+        EXCLUSIVE_LOCKS_REQUIRED(::cs_main, !cs_db, !m_cs_maps);
 
-    void CacheWarmingThreadMain() const EXCLUSIVE_LOCKS_REQUIRED(!m_cache_cs);
     void MigrateOldQuorumDB(CEvoDB& evoDb) const EXCLUSIVE_LOCKS_REQUIRED(!cs_db);
 };