refactor: extract CActiveMasternodeManager from LLMQContext (4/n, followups)

src/active/context.cpp

@@ -35,7 +35,7 @@ ActiveContext::ActiveContext(CBLSWorker& bls_worker, ChainstateManager& chainman
     m_isman{isman},
     m_qman{qman},
     nodeman{std::make_unique<CActiveMasternodeManager>(connman, dmnman, operator_sk)},
-    dkgdbgman{std::make_unique<llmq::CDKGDebugManager>()},
+    dkgdbgman{std::make_unique<llmq::CDKGDebugManager>(dmnman, qsnapman, chainman)},
     qdkgsman{std::make_unique<llmq::CDKGSessionManager>(dmnman, qsnapman, chainman, sporkman, db_params, quorums_watch)},
     shareman{std::make_unique<llmq::CSigSharesManager>(connman, chainman, sigman, *nodeman, qman, sporkman)},
     gov_signer{std::make_unique<GovernanceSigner>(connman, dmnman, govman, *nodeman, chainman, mn_sync)},
@@ -63,9 +63,9 @@ ActiveContext::~ActiveContext()
     m_isman.DisconnectSigner();
 }
 
-void ActiveContext::Start(CConnman& connman, PeerManager& peerman)
+void ActiveContext::Start(CConnman& connman, PeerManager& peerman, int16_t worker_count)
 {
-    qman_handler->Start();
+    qman_handler->Start(worker_count);
     qdkgsman->StartThreads(connman, peerman);
     cl_signer->Start();
     cl_signer->RegisterRecoveryInterface();

src/active/context.h

@@ -72,7 +72,7 @@ struct ActiveContext final : public CValidationInterface {
                            bool quorums_recovery, bool quorums_watch);
     ~ActiveContext();
 
-    void Start(CConnman& connman, PeerManager& peerman);
+    void Start(CConnman& connman, PeerManager& peerman, int16_t worker_count);
     void Stop();
 
     CCoinJoinServer& GetCJServer() const;

src/active/dkgsession.cpp

@@ -37,14 +37,14 @@ ActiveDKGSession::~ActiveDKGSession() = default;
 
 void ActiveDKGSession::Contribute(CDKGPendingMessages& pendingMessages, PeerManager& peerman)
 {
-    CDKGLogger logger(*this, __func__, __LINE__);
-
     if (!AreWeMember()) {
         return;
     }
 
     assert(params.threshold > 1); // we should not get there with single-node-quorums
 
+    CDKGLogger logger(*this, __func__, __LINE__);
+
     cxxtimer::Timer t1(true);
     logger.Batch("generating contributions");
     if (!blsWorker.GenerateContributions(params.threshold, memberIds, vvecContribution, m_sk_contributions)) {
@@ -304,7 +304,7 @@ void ActiveDKGSession::VerifyAndJustify(CDKGPendingMessages& pendingMessages, Pe
 
     CDKGLogger logger(*this, __func__, __LINE__);
 
-    std::set<uint256> justifyFor;
+    Uint256HashSet justifyFor;
 
     for (const auto& m : members) {
         if (m->bad) {
@@ -338,7 +338,7 @@ void ActiveDKGSession::VerifyAndJustify(CDKGPendingMessages& pendingMessages, Pe
 }
 
 void ActiveDKGSession::SendJustification(CDKGPendingMessages& pendingMessages, PeerManager& peerman,
-                                         const std::set<uint256>& forMembers)
+                                         const Uint256HashSet& forMembers)
 {
     CDKGLogger logger(*this, __func__, __LINE__);
 

src/active/dkgsession.h

@@ -43,7 +43,7 @@ class ActiveDKGSession final : public llmq::CDKGSession
     void VerifyAndJustify(CDKGPendingMessages& pendingMessages, PeerManager& peerman) override
         EXCLUSIVE_LOCKS_REQUIRED(!invCs);
     void SendJustification(CDKGPendingMessages& pendingMessages, PeerManager& peerman,
-                           const std::set<uint256>& forMembers) override;
+                           const Uint256HashSet& forMembers) override;
 
     // Phase 4: commit
     void VerifyAndCommit(CDKGPendingMessages& pendingMessages, PeerManager& peerman) override;

src/active/dkgsessionhandler.cpp

@@ -251,19 +251,19 @@ void ActiveDKGSessionHandler::HandlePhase(QuorumPhase curPhase, QuorumPhase next
 
 // returns a set of NodeIds which sent invalid messages
 template<typename Message>
-std::set<NodeId> BatchVerifyMessageSigs(CDKGSession& session, const std::vector<std::pair<NodeId, std::shared_ptr<Message>>>& messages)
+std::unordered_set<NodeId> BatchVerifyMessageSigs(CDKGSession& session, const std::vector<std::pair<NodeId, std::shared_ptr<Message>>>& messages)
 {
     if (messages.empty()) {
         return {};
     }
 
-    std::set<NodeId> ret;
+    std::unordered_set<NodeId> ret;
     bool revertToSingleVerification = false;
 
     CBLSSignature aggSig;
     std::vector<CBLSPublicKey> pubKeys;
     std::vector<uint256> messageHashes;
-    std::set<uint256> messageHashesSet;
+    Uint256HashSet messageHashesSet;
     pubKeys.reserve(messages.size());
     messageHashes.reserve(messages.size());
     bool first = true;

src/bench/bls.cpp

@@ -3,7 +3,10 @@
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
 #include <bench/bench.h>
+
 #include <bls/bls_worker.h>
+#include <llmq/options.h>
+
 #include <random.h>
 #include <util/time.h>
 
@@ -321,7 +324,7 @@ static void BLS_Verify_BatchedParallel(benchmark::Bench& bench)
     };
 
     CBLSWorker blsWorker;
-    blsWorker.Start();
+    blsWorker.Start(llmq::DEFAULT_WORKER_COUNT);
 
     // Benchmark.
     bench.minEpochIterations(bench.output() ? 1000 : 1).run([&] {

src/bench/bls_dkg.cpp

@@ -3,10 +3,13 @@
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
 #include <bench/bench.h>
+
 #include <bls/bls_worker.h>
-#include <random.h>
+#include <llmq/options.h>
 #include <util/irange.h>
 
+#include <random.h>
+
 struct Member {
     CBLSId id;
 
@@ -67,7 +70,7 @@ class DKG
             ids.emplace_back(id);
         }
 
-        blsWorker.Start();
+        blsWorker.Start(llmq::DEFAULT_WORKER_COUNT);
         for (auto& member : members) {
             blsWorker.GenerateContributions(quorumSize / 2 + 1, ids, member.vvec, member.skShares);
         }
@@ -110,7 +113,7 @@ class DKG
 static void BLSDKG_GenerateContributions(benchmark::Bench& bench, uint32_t epoch_iters, int quorumSize)
 {
     CBLSWorker blsWorker;
-    blsWorker.Start();
+    blsWorker.Start(llmq::DEFAULT_WORKER_COUNT);
     std::vector<CBLSId> ids;
     std::vector<Member> members;
     if (!bench.output()) {

src/bls/bls_worker.cpp

@@ -56,11 +56,10 @@ CBLSWorker::~CBLSWorker()
     Stop();
 }
 
-void CBLSWorker::Start()
+void CBLSWorker::Start(int16_t worker_count)
 {
-    int workerCount = std::thread::hardware_concurrency() / 2;
-    workerCount = std::clamp(workerCount, 1, 4);
-    workerPool.resize(workerCount);
+    assert(worker_count > 0);
+    workerPool.resize(worker_count);
     RenameThreadPool(workerPool, "bls-work");
 }
 

src/bls/bls_worker.h

@@ -54,7 +54,7 @@ class CBLSWorker
     CBLSWorker();
     ~CBLSWorker();
 
-    void Start();
+    void Start(int16_t worker_count);
     void Stop();
 
     bool GenerateContributions(int threshold, Span<CBLSId> ids, BLSVerificationVectorPtr& vvecRet, std::vector<CBLSSecretKey>& skSharesRet);

src/init.cpp

@@ -2018,6 +2018,7 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
                                                   // Subtract 1 because the main thread counts towards the par threads
                                                   return std::clamp<int8_t>(threads - 1, 0, llmq::MAX_BLSCHECK_THREADS);
                                               }(),
+                                              llmq::DEFAULT_WORKER_COUNT,
                                               args.GetIntArg("-maxrecsigsage", llmq::DEFAULT_MAX_RECOVERED_SIGS_AGE),
                                               /*shutdown_requested=*/ShutdownRequested,
                                               /*coins_error_cb=*/[]() {
@@ -2321,15 +2322,15 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
 
     node.peerman->StartHandlers();
     node.clhandler->Start();
-    if (node.observer_ctx) node.observer_ctx->Start();
+    if (node.observer_ctx) node.observer_ctx->Start(llmq::DEFAULT_WORKER_COUNT);
 
     node.scheduler->scheduleEvery(std::bind(&CNetFulfilledRequestManager::DoMaintenance, std::ref(*node.netfulfilledman)), std::chrono::minutes{1});
     node.scheduler->scheduleEvery(std::bind(&CMasternodeUtils::DoMaintenance, std::ref(*node.connman), std::ref(*node.dmnman), std::ref(*node.mn_sync), node.cj_walletman.get()), std::chrono::minutes{1});
     node.scheduler->scheduleEvery(std::bind(&CDeterministicMNManager::DoMaintenance, std::ref(*node.dmnman)), std::chrono::seconds{10});
     node.peerman->ScheduleHandlers(*node.scheduler);
 
     if (node.active_ctx) {
-        node.active_ctx->Start(*node.connman, *node.peerman);
+        node.active_ctx->Start(*node.connman, *node.peerman, llmq::DEFAULT_WORKER_COUNT);
         node.scheduler->scheduleEvery(std::bind(&llmq::CDKGSessionManager::CleanupOldContributions, std::ref(*node.active_ctx->qdkgsman)), std::chrono::hours{1});
     }
 

src/llmq/context.cpp

@@ -15,7 +15,7 @@
 LLMQContext::LLMQContext(CDeterministicMNManager& dmnman, CEvoDB& evo_db, CSporkManager& sporkman,
                          chainlock::Chainlocks& chainlocks, CTxMemPool& mempool, ChainstateManager& chainman,
                          const CMasternodeSync& mn_sync, const util::DbWrapperParams& db_params, int8_t bls_threads,
-                         int64_t max_recsigs_age) :
+                         int16_t worker_count, int64_t max_recsigs_age) :
     bls_worker{std::make_shared<CBLSWorker>()},
     qsnapman{std::make_unique<llmq::CQuorumSnapshotManager>(evo_db)},
     quorum_block_processor{std::make_unique<llmq::CQuorumBlockProcessor>(chainman.ActiveChainstate(), dmnman, evo_db,
@@ -27,7 +27,7 @@ LLMQContext::LLMQContext(CDeterministicMNManager& dmnman, CEvoDB& evo_db, CSpork
                                                       mempool, mn_sync, db_params)}
 {
     // Have to start it early to let VerifyDB check ChainLock signatures in coinbase
-    bls_worker->Start();
+    bls_worker->Start(worker_count);
 }
 
 LLMQContext::~LLMQContext()

src/llmq/context.h

@@ -41,7 +41,7 @@ struct LLMQContext {
     explicit LLMQContext(CDeterministicMNManager& dmnman, CEvoDB& evo_db, CSporkManager& sporkman,
                          chainlock::Chainlocks& chainlocks, CTxMemPool& mempool, ChainstateManager& chainman,
                          const CMasternodeSync& mn_sync, const util::DbWrapperParams& db_params, int8_t bls_threads,
-                         int64_t max_recsigs_age);
+                         int16_t worker_count, int64_t max_recsigs_age);
     ~LLMQContext();
 
     /** Guaranteed if LLMQContext is initialized then all members are valid too

src/llmq/core_write.cpp

@@ -77,8 +77,8 @@ RPCResult CDKGDebugSessionStatus::GetJsonHelp(const std::string& key, bool optio
     }};
 }
 
-// CDKGDebugStatus::ToJson() defined in llmq/debug.cpp
-RPCResult CDKGDebugStatus::GetJsonHelp(const std::string& key, bool optional, bool inner_optional)
+// CDKGDebugManager::ToJson() defined in llmq/debug.cpp
+RPCResult CDKGDebugManager::GetJsonHelp(const std::string& key, bool optional, bool inner_optional)
 {
     return {RPCResult::Type::OBJ, key, optional, key.empty() ? "" : "The state of the node's DKG sessions",
     {

src/llmq/debug.cpp

@@ -107,29 +107,40 @@ UniValue CDKGDebugSessionStatus::ToJson(CDeterministicMNManager& dmnman, CQuorum
     return ret;
 }
 
-CDKGDebugManager::CDKGDebugManager() = default;
+CDKGDebugManager::CDKGDebugManager(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
+                                   const ChainstateManager& chainman) :
+    m_dmnman{dmnman},
+    m_qsnapman{qsnapman},
+    m_chainman{chainman}
+{
+}
 
 CDKGDebugManager::~CDKGDebugManager() = default;
 
-UniValue CDKGDebugStatus::ToJson(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
-                                 const ChainstateManager& chainman, int detailLevel) const
+size_t CDKGDebugManager::GetSessionCount() const
 {
-    UniValue ret(UniValue::VOBJ);
+    return WITH_LOCK(cs_lockStatus, return localStatus.sessions.size());
+}
+
+UniValue CDKGDebugManager::ToJson(int detailLevel) const
+{
+    LOCK(cs_lockStatus);
 
-    ret.pushKV("time", nTime);
-    ret.pushKV("timeStr", FormatISO8601DateTime(nTime));
+    UniValue ret(UniValue::VOBJ);
+    ret.pushKV("time", localStatus.nTime);
+    ret.pushKV("timeStr", FormatISO8601DateTime(localStatus.nTime));
 
     // TODO Support array of sessions
     UniValue sessionsArrJson(UniValue::VARR);
-    for (const auto& p : sessions) {
+    for (const auto& p : localStatus.sessions) {
         const auto& llmq_params_opt = Params().GetLLMQ(p.first.first);
         if (!llmq_params_opt.has_value()) {
             continue;
         }
         UniValue s(UniValue::VOBJ);
         s.pushKV("llmqType", std::string(llmq_params_opt->name));
         s.pushKV("quorumIndex", p.first.second);
-        s.pushKV("status", p.second.ToJson(dmnman, qsnapman, chainman, p.first.second, detailLevel));
+        s.pushKV("status", p.second.ToJson(m_dmnman, m_qsnapman, m_chainman, p.first.second, detailLevel));
 
         sessionsArrJson.push_back(s);
     }
@@ -138,12 +149,6 @@ UniValue CDKGDebugStatus::ToJson(CDeterministicMNManager& dmnman, CQuorumSnapsho
     return ret;
 }
 
-void CDKGDebugManager::GetLocalDebugStatus(llmq::CDKGDebugStatus& ret) const
-{
-    LOCK(cs_lockStatus);
-    ret = localStatus;
-}
-
 void CDKGDebugManager::ResetLocalSessionStatus(Consensus::LLMQType llmqType, int quorumIndex)
 {
     LOCK(cs_lockStatus);

src/llmq/debug.h

@@ -10,7 +10,7 @@
 #include <univalue.h>
 
 #include <functional>
-#include <set>
+#include <unordered_set>
 
 class CDataStream;
 class CDeterministicMNManager;
@@ -45,7 +45,7 @@ class CDKGDebugMemberStatus
         uint8_t statusBitset;
     };
 
-    std::set<uint16_t> complaintsFromMembers;
+    std::unordered_set<uint16_t> complaintsFromMembers;
 
 public:
     CDKGDebugMemberStatus() : statusBitset(0) {}
@@ -83,34 +83,28 @@ class CDKGDebugSessionStatus
                                   const ChainstateManager& chainman, int quorumIndex, int detailLevel) const;
 };
 
-class CDKGDebugStatus
-{
-public:
+struct CDKGDebugStatus {
     int64_t nTime{0};
-
     std::map<std::pair<Consensus::LLMQType, int>, CDKGDebugSessionStatus> sessions;
-    //std::map<Consensus::LLMQType, CDKGDebugSessionStatus> sessions;
-
-public:
-    [[nodiscard]] static RPCResult GetJsonHelp(const std::string& key, bool optional, bool inner_optional = false);
-    [[nodiscard]] UniValue ToJson(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman,
-                                  const ChainstateManager& chainman, int detailLevel) const;
 };
 
 class CDKGDebugManager
 {
+private:
+    CDeterministicMNManager& m_dmnman;
+    CQuorumSnapshotManager& m_qsnapman;
+    const ChainstateManager& m_chainman;
+
 private:
     mutable Mutex cs_lockStatus;
     CDKGDebugStatus localStatus GUARDED_BY(cs_lockStatus);
 
 public:
     CDKGDebugManager(const CDKGDebugManager&) = delete;
     CDKGDebugManager& operator=(const CDKGDebugManager&) = delete;
-    CDKGDebugManager();
+    CDKGDebugManager(CDeterministicMNManager& dmnman, CQuorumSnapshotManager& qsnapman, const ChainstateManager& chainman);
     ~CDKGDebugManager();
 
-    void GetLocalDebugStatus(CDKGDebugStatus& ret) const EXCLUSIVE_LOCKS_REQUIRED(!cs_lockStatus);
-
     void ResetLocalSessionStatus(Consensus::LLMQType llmqType, int quorumIndex) EXCLUSIVE_LOCKS_REQUIRED(!cs_lockStatus);
     void InitLocalSessionStatus(const Consensus::LLMQParams& llmqParams, int quorumIndex, const uint256& quorumHash,
                                 int quorumHeight) EXCLUSIVE_LOCKS_REQUIRED(!cs_lockStatus);
@@ -121,8 +115,13 @@ class CDKGDebugManager
     void UpdateLocalMemberStatus(Consensus::LLMQType llmqType, int quorumIndex, size_t memberIdx,
                                  std::function<bool(CDKGDebugMemberStatus& status)>&& func)
         EXCLUSIVE_LOCKS_REQUIRED(!cs_lockStatus);
-};
 
+    size_t GetSessionCount() const
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_lockStatus);
+    [[nodiscard]] static RPCResult GetJsonHelp(const std::string& key, bool optional, bool inner_optional = false);
+    [[nodiscard]] UniValue ToJson(int detailLevel) const
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_lockStatus);
+};
 } // namespace llmq
 
 #endif // BITCOIN_LLMQ_DEBUG_H