[LLMQ] Improve thread handling

src/init.cpp

@@ -191,6 +191,7 @@ void Interrupt()
     InterruptREST();
     InterruptTorControl();
     InterruptMapPort();
+    InterruptTierTwo();
     if (g_connman)
         g_connman->Interrupt();
 }

src/llmq/quorums_chainlocks.cpp

@@ -20,7 +20,7 @@ namespace llmq
 
 static const std::string CLSIG_REQUESTID_PREFIX = "clsig";
 
-CChainLocksHandler* chainLocksHandler;
+std::unique_ptr<CChainLocksHandler> chainLocksHandler{nullptr};
 
 std::string CChainLockSig::ToString() const
 {
@@ -30,10 +30,18 @@ std::string CChainLockSig::ToString() const
 CChainLocksHandler::CChainLocksHandler(CScheduler* _scheduler) :
     scheduler(_scheduler)
 {
-    quorumSigningManager->RegisterRecoveredSigsListener(this);
 }
 
 CChainLocksHandler::~CChainLocksHandler()
+{
+}
+
+void CChainLocksHandler::Start()
+{
+    quorumSigningManager->RegisterRecoveredSigsListener(this);
+}
+
+void CChainLocksHandler::Stop()
 {
     quorumSigningManager->UnregisterRecoveredSigsListener(this);
 }

src/llmq/quorums_chainlocks.h

@@ -64,6 +64,8 @@ class CChainLocksHandler : public CRecoveredSigsListener
 public:
     CChainLocksHandler(CScheduler* _scheduler);
     ~CChainLocksHandler();
+    void Start();
+    void Stop();
 
 public:
     bool AlreadyHave(const CInv& inv);
@@ -90,9 +92,7 @@ class CChainLocksHandler : public CRecoveredSigsListener
     void Cleanup();
 };
 
-extern CChainLocksHandler* chainLocksHandler;
-
-
+extern std::unique_ptr<CChainLocksHandler> chainLocksHandler;
 }
 
 #endif //PIVX_QUORUMS_CHAINLOCKS_H

src/llmq/quorums_init.cpp

@@ -27,19 +27,16 @@ void InitLLMQSystem(CEvoDB& evoDb, CScheduler* scheduler, bool unitTests)
     quorumBlockProcessor.reset(new CQuorumBlockProcessor(evoDb));
     quorumDKGSessionManager.reset(new CDKGSessionManager(evoDb, *blsWorker));
     quorumManager.reset(new CQuorumManager(evoDb, *blsWorker, *quorumDKGSessionManager));
-    quorumSigSharesManager = new CSigSharesManager();
-    quorumSigningManager = new CSigningManager(unitTests);
-    chainLocksHandler = new CChainLocksHandler(scheduler);
+    quorumSigSharesManager.reset(new CSigSharesManager());
+    quorumSigningManager.reset(new CSigningManager(unitTests));
+    chainLocksHandler.reset(new CChainLocksHandler(scheduler));
 }
 
 void DestroyLLMQSystem()
 {
-    delete chainLocksHandler;
-    chainLocksHandler = nullptr;
-    delete quorumSigningManager;
-    quorumSigningManager = nullptr;
-    delete quorumSigSharesManager;
-    quorumSigSharesManager = nullptr;
+    chainLocksHandler.reset();
+    quorumSigningManager.reset();
+    quorumSigSharesManager.reset();
     quorumDKGSessionManager.reset();
     quorumBlockProcessor.reset();
     quorumDKGDebugManager.reset();
@@ -59,19 +56,32 @@ void StartLLMQSystem()
     if (quorumSigSharesManager) {
         quorumSigSharesManager->StartWorkerThread();
     }
+    if (chainLocksHandler) {
+        chainLocksHandler->Start();
+    }
 }
 
 void StopLLMQSystem()
 {
-    if (quorumDKGSessionManager) {
-        quorumDKGSessionManager->StopThreads();
+    if (chainLocksHandler) {
+        chainLocksHandler->Stop();
     }
     if (quorumSigSharesManager) {
         quorumSigSharesManager->StopWorkerThread();
     }
+    if (quorumDKGSessionManager) {
+        quorumDKGSessionManager->StopThreads();
+    }
     if (blsWorker) {
         blsWorker->Stop();
     }
 }
 
+void InterruptLLMQSystem()
+{
+    if (quorumSigSharesManager) {
+        quorumSigSharesManager->Interrupt();
+    }
+}
+
 } // namespace llmq

src/llmq/quorums_init.h

@@ -21,6 +21,7 @@ void DestroyLLMQSystem();
 // Manage scheduled tasks, threads, listeners etc.
 void StartLLMQSystem();
 void StopLLMQSystem();
+void InterruptLLMQSystem();
 
 } // namespace llmq
 

src/llmq/quorums_signing.cpp

@@ -21,7 +21,7 @@
 namespace llmq
 {
 
-CSigningManager* quorumSigningManager;
+std::unique_ptr<CSigningManager> quorumSigningManager{nullptr};
 
 CRecoveredSigsDb::CRecoveredSigsDb(bool fMemory) : db(fMemory ? "" : (GetDataDir() / "llmq"), 1 << 20, fMemory, false, CLIENT_VERSION | ADDRV2_FORMAT)
 {

src/llmq/quorums_signing.h

@@ -156,7 +156,7 @@ class CSigningManager
     bool VerifyRecoveredSig(Consensus::LLMQType llmqType, int signedAtHeight, const uint256& id, const uint256& msgHash, const CBLSSignature& sig);
 };
 
-extern CSigningManager* quorumSigningManager;
+extern std::unique_ptr<CSigningManager> quorumSigningManager;
 
 } // namespace llmq
 

src/llmq/quorums_signing_shares.cpp

@@ -22,7 +22,7 @@
 namespace llmq
 {
 
-CSigSharesManager* quorumSigSharesManager = nullptr;
+std::unique_ptr<CSigSharesManager> quorumSigSharesManager{nullptr};
 
 template <typename M>
 static std::pair<typename M::const_iterator, typename M::const_iterator> FindBySignHash(const M& m, const uint256& signHash)
@@ -179,32 +179,30 @@ CSigSharesInv CBatchedSigShares::ToInv() const
 
 CSigSharesManager::CSigSharesManager()
 {
+    interruptSigningShare.reset();
 }
 
 CSigSharesManager::~CSigSharesManager()
 {
-    StopWorkerThread();
 }
 
 void CSigSharesManager::StartWorkerThread()
 {
-    workThread = std::thread(&TraceThread<std::function<void()>>, "quorum-sigshares", [this]() {
-        WorkThreadMain();
-    });
+    workThread = std::thread(&TraceThread<std::function<void()>>, "quorum-sigshares", std::function<void()>(std::bind(&CSigSharesManager::WorkThreadMain, this)));
 }
 
 void CSigSharesManager::StopWorkerThread()
 {
-    if (stopWorkThread) {
-        return;
-    }
-
-    stopWorkThread = true;
     if (workThread.joinable()) {
         workThread.join();
     }
 }
 
+void CSigSharesManager::Interrupt()
+{
+    interruptSigningShare();
+}
+
 void CSigSharesManager::ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStream& vRecv, CConnman& connman)
 {
     // non-masternodes are not interested in sigshares
@@ -1105,7 +1103,7 @@ void CSigSharesManager::BanNode(NodeId nodeId)
 void CSigSharesManager::WorkThreadMain()
 {
     int64_t lastProcessTime = GetTimeMillis();
-    while (!stopWorkThread && !ShutdownRequested()) {
+    while (!interruptSigningShare) {
         RemoveBannedNodeStates();
         quorumSigningManager->ProcessPendingRecoveredSigs(*g_connman);
         ProcessPendingSigShares(*g_connman);
@@ -1115,7 +1113,9 @@ void CSigSharesManager::WorkThreadMain()
         quorumSigningManager->Cleanup();
 
         // TODO Wakeup when pending signing is needed?
-        MilliSleep(100);
+        if (!interruptSigningShare.sleep_for(std::chrono::milliseconds(100))) {
+            return;
+        }
     }
 }
 

src/llmq/quorums_signing_shares.h

@@ -199,7 +199,7 @@ class CSigSharesManager
     RecursiveMutex cs;
 
     std::thread workThread;
-    std::atomic<bool> stopWorkThread{false};
+    CThreadInterrupt interruptSigningShare;
 
     std::map<SigShareKey, CSigShare> sigShares;
     std::map<uint256, int64_t> firstSeenForSessions;
@@ -221,6 +221,7 @@ class CSigSharesManager
 
     void StartWorkerThread();
     void StopWorkerThread();
+    void Interrupt();
 
 public:
     void ProcessMessage(CNode* pnode, const std::string& strCommand, CDataStream& vRecv, CConnman& connman);
@@ -259,7 +260,7 @@ class CSigSharesManager
     void WorkThreadMain();
 };
 
-extern CSigSharesManager* quorumSigSharesManager;
+extern std::unique_ptr<CSigSharesManager> quorumSigSharesManager;
 
 } // namespace llmq
 

src/test/test_pivx.cpp

@@ -141,6 +141,7 @@ TestingSetup::TestingSetup(const std::string& chainName) : BasicTestingSetup(cha
 TestingSetup::~TestingSetup()
 {
         scheduler.stop();
+        llmq::InterruptLLMQSystem();
         threadGroup.interrupt_all();
         threadGroup.join_all();
         GetMainSignals().FlushBackgroundCallbacks();

src/tiertwo/init.cpp

@@ -328,3 +328,8 @@ void DeleteTierTwo()
     deterministicMNManager.reset();
     evoDb.reset();
 }
+
+void InterruptTierTwo()
+{
+    llmq::InterruptLLMQSystem();
+}

src/tiertwo/init.h

@@ -56,5 +56,8 @@ void StopTierTwoThreads();
 /** Cleans manager and worker objects pointers */
 void DeleteTierTwo();
 
+/** Interrupt tier two threads */
+void InterruptTierTwo();
+
 
 #endif //PIVX_TIERTWO_INIT_H