Separate init/destroy and start/stop steps in LLMQ flow (#2709)

src/init.cpp

@@ -236,6 +236,7 @@ void PrepareShutdown()
     StopREST();
     StopRPC();
     StopHTTPServer();
+    llmq::StopLLMQSystem();
 
     // fRPCInWarmup should be `false` if we completed the loading sequence
     // before a shutdown request was received
@@ -2066,6 +2067,8 @@ bool AppInitMain(boost::thread_group& threadGroup, CScheduler& scheduler)
 #endif // ENABLE_WALLET
     }
 
+    llmq::StartLLMQSystem();
+
     // ********************************************************* Step 12: start node
 
     //// debug print

src/llmq/quorums_chainlocks.cpp

@@ -28,10 +28,18 @@ std::string CChainLockSig::ToString() const
 CChainLocksHandler::CChainLocksHandler(CScheduler* _scheduler) :
     scheduler(_scheduler)
 {
-    quorumSigningManager->RegisterRecoveredSigsListener(this);
 }
 
 CChainLocksHandler::~CChainLocksHandler()
+{
+}
+
+void CChainLocksHandler::RegisterAsRecoveredSigsListener()
+{
+    quorumSigningManager->RegisterRecoveredSigsListener(this);
+}
+
+void CChainLocksHandler::UnregisterAsRecoveredSigsListener()
 {
     quorumSigningManager->UnregisterRecoveredSigsListener(this);
 }

src/llmq/quorums_chainlocks.h

@@ -68,7 +68,9 @@ public:
     CChainLocksHandler(CScheduler* _scheduler);
     ~CChainLocksHandler();
 
-public:
+    void RegisterAsRecoveredSigsListener();
+    void UnregisterAsRecoveredSigsListener();
+
     bool AlreadyHave(const CInv& inv);
     bool GetChainLockByHash(const uint256& hash, CChainLockSig& ret);
 

src/llmq/quorums_debug.cpp

@@ -110,6 +110,10 @@ UniValue CDKGDebugSessionStatus::ToJson(int detailLevel) const
 
 CDKGDebugManager::CDKGDebugManager(CScheduler* _scheduler) :
     scheduler(_scheduler)
+{
+}
+
+void CDKGDebugManager::StartScheduler()
 {
     if (scheduler) {
         scheduler->scheduleEvery([&]() {

src/llmq/quorums_debug.h

@@ -158,6 +158,8 @@ private:
 public:
     CDKGDebugManager(CScheduler* _scheduler);
 
+    void StartScheduler();
+
     void ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStream& vRecv, CConnman& connman);
     bool PreVerifyDebugStatusMessage(const uint256& hash, CDKGDebugStatus& status, bool& retBan);
     void ScheduleProcessPending();

src/llmq/quorums_dkgsessionmgr.cpp

@@ -24,18 +24,26 @@ static const std::string DB_SKCONTRIB = "qdkg_S";
 CDKGSessionManager::CDKGSessionManager(CEvoDB& _evoDb, CBLSWorker& _blsWorker) :
     evoDb(_evoDb),
     blsWorker(_blsWorker)
+{
+}
+
+CDKGSessionManager::~CDKGSessionManager()
+{
+}
+
+void CDKGSessionManager::StartMessageHandlerPool()
 {
     for (const auto& qt : Params().GetConsensus().llmqs) {
         dkgSessionHandlers.emplace(std::piecewise_construct,
                 std::forward_as_tuple(qt.first),
-                std::forward_as_tuple(qt.second, _evoDb, messageHandlerPool, blsWorker, *this));
+                std::forward_as_tuple(qt.second, evoDb, messageHandlerPool, blsWorker, *this));
     }
 
     messageHandlerPool.resize(2);
     RenameThreadPool(messageHandlerPool, "quorum-msg");
 }
 
-CDKGSessionManager::~CDKGSessionManager()
+void CDKGSessionManager::StopMessageHandlerPool()
 {
     messageHandlerPool.stop(true);
 }

src/llmq/quorums_dkgsessionmgr.h

@@ -50,6 +50,9 @@ public:
     CDKGSessionManager(CEvoDB& _evoDb, CBLSWorker& _blsWorker);
     ~CDKGSessionManager();
 
+    void StartMessageHandlerPool();
+    void StopMessageHandlerPool();
+
     void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload);
 
     void ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStream& vRecv, CConnman& connman);

src/llmq/quorums_init.cpp

@@ -31,13 +31,6 @@ void InitLLMQSystem(CEvoDB& evoDb, CScheduler* scheduler, bool unitTests)
     chainLocksHandler = new CChainLocksHandler(scheduler);
 }
 
-void InterruptLLMQSystem()
-{
-    if (quorumSigSharesManager) {
-        quorumSigSharesManager->InterruptWorkerThread();
-    }
-}
-
 void DestroyLLMQSystem()
 {
     delete chainLocksHandler;
@@ -56,4 +49,40 @@ void DestroyLLMQSystem()
     quorumDKGDebugManager = nullptr;
 }
 
+void StartLLMQSystem()
+{
+    if (quorumDKGDebugManager) {
+        quorumDKGDebugManager->StartScheduler();
+    }
+    if (quorumDKGSessionManager) {
+        quorumDKGSessionManager->StartMessageHandlerPool();
+    }
+    if (quorumSigSharesManager) {
+        quorumSigSharesManager->StartWorkerThread();
+    }
+    if (chainLocksHandler) {
+        chainLocksHandler->RegisterAsRecoveredSigsListener();
+    }
+}
+
+void StopLLMQSystem()
+{
+    if (chainLocksHandler) {
+        chainLocksHandler->UnregisterAsRecoveredSigsListener();
+    }
+    if (quorumSigSharesManager) {
+        quorumSigSharesManager->StopWorkerThread();
+    }
+    if (quorumDKGSessionManager) {
+        quorumDKGSessionManager->StopMessageHandlerPool();
+    }
+}
+
+void InterruptLLMQSystem()
+{
+    if (quorumSigSharesManager) {
+        quorumSigSharesManager->InterruptWorkerThread();
+    }
+}
+
 }

src/llmq/quorums_init.h

@@ -14,10 +14,14 @@ namespace llmq
 // If true, we will connect to all new quorums and watch their communication
 static const bool DEFAULT_WATCH_QUORUMS = false;
 
+// Init/destroy LLMQ globals
 void InitLLMQSystem(CEvoDB& evoDb, CScheduler* scheduler, bool unitTests);
-void InterruptLLMQSystem();
 void DestroyLLMQSystem();
 
+// Manage scheduled tasks, threads, listeners etc.
+void StartLLMQSystem();
+void StopLLMQSystem();
+void InterruptLLMQSystem();
 }
 
 #endif //DASH_QUORUMS_INIT_H

src/llmq/quorums_signing_shares.cpp

@@ -148,16 +148,20 @@ CSigSharesInv CBatchedSigShares::ToInv() const
 
 CSigSharesManager::CSigSharesManager()
 {
-    StartWorkerThread();
+    workInterrupt.reset();
 }
 
 CSigSharesManager::~CSigSharesManager()
 {
-    StopWorkerThread();
 }
 
 void CSigSharesManager::StartWorkerThread()
 {
+    // can't start new thread if we have one running already
+    if (workThread.joinable()) {
+        assert(false);
+    }
+
     workThread = std::thread(&TraceThread<std::function<void()> >,
         "sigshares",
         std::function<void()>(std::bind(&CSigSharesManager::WorkThreadMain, this)));
@@ -165,6 +169,11 @@ void CSigSharesManager::StartWorkerThread()
 
 void CSigSharesManager::StopWorkerThread()
 {
+    // make sure to call InterruptWorkerThread() first
+    if (!workInterrupt) {
+        assert(false);
+    }
+
     if (workThread.joinable()) {
         workThread.join();
     }
@@ -1086,8 +1095,6 @@ void CSigSharesManager::BanNode(NodeId nodeId)
 
 void CSigSharesManager::WorkThreadMain()
 {
-    workInterrupt.reset();
-
     int64_t lastSendTime = 0;
 
     while (!workInterrupt) {

src/llmq/quorums_signing_shares.h

@@ -361,6 +361,8 @@ public:
     CSigSharesManager();
     ~CSigSharesManager();
 
+    void StartWorkerThread();
+    void StopWorkerThread();
     void InterruptWorkerThread();
 
 public:
@@ -370,9 +372,6 @@ public:
     void Sign(const CQuorumCPtr& quorum, const uint256& id, const uint256& msgHash);
 
 private:
-    void StartWorkerThread();
-    void StopWorkerThread();
-
     void ProcessMessageSigSharesInv(CNode* pfrom, const CSigSharesInv& inv, CConnman& connman);
     void ProcessMessageGetSigShares(CNode* pfrom, const CSigSharesInv& inv, CConnman& connman);
     void ProcessMessageBatchedSigShares(CNode* pfrom, const CBatchedSigShares& batchedSigShares, CConnman& connman);