refactor: move PeerManager out of CSigningManager ctor

kwvg · kwvg · commit 7498a380764c · 2024-12-05T22:37:49.000Z
diff --git a/src/llmq/context.cpp b/src/llmq/context.cpp
@@ -33,8 +33,7 @@ LLMQContext::LLMQContext(ChainstateManager& chainman, CConnman& connman, CDeterm
     qman{std::make_unique<llmq::CQuorumManager>(*bls_worker, chainman.ActiveChainstate(), dmnman, *qdkgsman, evo_db,
                                                 *quorum_block_processor, mn_activeman, mn_sync, sporkman, unit_tests,
                                                 wipe)},
-    sigman{std::make_unique<llmq::CSigningManager>(mn_activeman, chainman.ActiveChainstate(), *qman, peerman,
-                                                   unit_tests, wipe)},
+    sigman{std::make_unique<llmq::CSigningManager>(mn_activeman, chainman.ActiveChainstate(), *qman, unit_tests, wipe)},
     shareman{std::make_unique<llmq::CSigSharesManager>(connman, *sigman, mn_activeman, *qman, sporkman, peerman)},
     clhandler{[&]() -> llmq::CChainLocksHandler* const {
         assert(llmq::chainLocksHandler == nullptr);
@@ -85,7 +84,7 @@ void LLMQContext::Start(CConnman& connman, PeerManager& peerman)
     qman->Start();
     shareman->RegisterAsRecoveredSigsListener();
     shareman->StartWorkerThread();
-    sigman->StartWorkerThread();
+    sigman->StartWorkerThread(peerman);
 
     llmq::chainLocksHandler->Start();
     llmq::quorumInstantSendManager->Start();
diff --git a/src/llmq/signing.cpp b/src/llmq/signing.cpp
@@ -349,8 +349,11 @@ void CRecoveredSigsDb::CleanupOldVotes(int64_t maxAge)
 //////////////////
 
 CSigningManager::CSigningManager(const CActiveMasternodeManager* const mn_activeman, const CChainState& chainstate,
-                                 const CQuorumManager& _qman, const std::unique_ptr<PeerManager>& peerman, bool fMemory, bool fWipe) :
-    db(fMemory, fWipe), m_mn_activeman(mn_activeman), m_chainstate(chainstate), qman(_qman), m_peerman(peerman)
+                                 const CQuorumManager& _qman, bool fMemory, bool fWipe) :
+    db(fMemory, fWipe),
+    m_mn_activeman(mn_activeman),
+    m_chainstate(chainstate),
+    qman(_qman)
 {
 }
 
@@ -381,13 +384,14 @@ bool CSigningManager::GetRecoveredSigForGetData(const uint256& hash, CRecoveredS
     return true;
 }
 
-PeerMsgRet CSigningManager::ProcessMessage(const CNode& pfrom, const std::string& msg_type, CDataStream& vRecv)
+PeerMsgRet CSigningManager::ProcessMessage(const CNode& pfrom, PeerManager& peerman, const std::string& msg_type,
+                                           CDataStream& vRecv)
 {
     if (msg_type == NetMsgType::QSIGREC) {
         auto recoveredSig = std::make_shared<CRecoveredSig>();
         vRecv >> *recoveredSig;
 
-        return ProcessMessageRecoveredSig(pfrom, recoveredSig);
+        return ProcessMessageRecoveredSig(pfrom, peerman, recoveredSig);
     }
     return {};
 }
@@ -416,10 +420,11 @@ static bool PreVerifyRecoveredSig(const CQuorumManager& quorum_manager, const CR
     return true;
 }
 
-PeerMsgRet CSigningManager::ProcessMessageRecoveredSig(const CNode& pfrom, const std::shared_ptr<const CRecoveredSig>& recoveredSig)
+PeerMsgRet CSigningManager::ProcessMessageRecoveredSig(const CNode& pfrom, PeerManager& peerman,
+                                                       const std::shared_ptr<const CRecoveredSig>& recoveredSig)
 {
-    WITH_LOCK(::cs_main, Assert(m_peerman)->EraseObjectRequest(pfrom.GetId(),
-                                                               CInv(MSG_QUORUM_RECOVERED_SIG, recoveredSig->GetHash())));
+    WITH_LOCK(::cs_main,
+              peerman.EraseObjectRequest(pfrom.GetId(), CInv(MSG_QUORUM_RECOVERED_SIG, recoveredSig->GetHash())));
 
     bool ban = false;
     if (!PreVerifyRecoveredSig(qman, *recoveredSig, ban)) {
@@ -517,22 +522,22 @@ void CSigningManager::CollectPendingRecoveredSigsToVerify(
     }
 }
 
-void CSigningManager::ProcessPendingReconstructedRecoveredSigs()
+void CSigningManager::ProcessPendingReconstructedRecoveredSigs(PeerManager& peerman)
 {
     decltype(pendingReconstructedRecoveredSigs) m;
     WITH_LOCK(cs_pending, swap(m, pendingReconstructedRecoveredSigs));
 
     for (const auto& p : m) {
-        ProcessRecoveredSig(p.second);
+        ProcessRecoveredSig(p.second, peerman);
     }
 }
 
-bool CSigningManager::ProcessPendingRecoveredSigs()
+bool CSigningManager::ProcessPendingRecoveredSigs(PeerManager& peerman)
 {
     std::unordered_map<NodeId, std::list<std::shared_ptr<const CRecoveredSig>>> recSigsByNode;
     std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher> quorums;
 
-    ProcessPendingReconstructedRecoveredSigs();
+    ProcessPendingReconstructedRecoveredSigs(peerman);
 
     const size_t nMaxBatchSize{32};
     CollectPendingRecoveredSigsToVerify(nMaxBatchSize, recSigsByNode, quorums);
@@ -575,7 +580,7 @@ bool CSigningManager::ProcessPendingRecoveredSigs()
 
         if (batchVerifier.badSources.count(nodeId)) {
             LogPrint(BCLog::LLMQ, "CSigningManager::%s -- invalid recSig from other node, banning peer=%d\n", __func__, nodeId);
-            Assert(m_peerman)->Misbehaving(nodeId, 100);
+            peerman.Misbehaving(nodeId, 100);
             continue;
         }
 
@@ -584,15 +589,15 @@ bool CSigningManager::ProcessPendingRecoveredSigs()
                 continue;
             }
 
-            ProcessRecoveredSig(recSig);
+            ProcessRecoveredSig(recSig, peerman);
         }
     }
 
     return recSigsByNode.size() >= nMaxBatchSize;
 }
 
 // signature must be verified already
-void CSigningManager::ProcessRecoveredSig(const std::shared_ptr<const CRecoveredSig>& recoveredSig)
+void CSigningManager::ProcessRecoveredSig(const std::shared_ptr<const CRecoveredSig>& recoveredSig, PeerManager& peerman)
 {
     auto llmqType = recoveredSig->getLlmqType();
 
@@ -631,12 +636,12 @@ void CSigningManager::ProcessRecoveredSig(const std::shared_ptr<const CRecovered
     WITH_LOCK(cs_pending, pendingReconstructedRecoveredSigs.erase(recoveredSig->GetHash()));
 
     if (m_mn_activeman != nullptr) {
-        Assert(m_peerman)->RelayRecoveredSig(recoveredSig->GetHash());
+        peerman.RelayRecoveredSig(recoveredSig->GetHash());
     }
 
     auto listeners = WITH_LOCK(cs_listeners, return recoveredSigsListeners);
     for (auto& l : listeners) {
-        Assert(m_peerman)->PostProcessMessage(l->HandleNewRecoveredSig(*recoveredSig));
+        peerman.PostProcessMessage(l->HandleNewRecoveredSig(*recoveredSig));
     }
 
     GetMainSignals().NotifyRecoveredSig(recoveredSig);
@@ -799,14 +804,14 @@ bool CSigningManager::GetVoteForId(Consensus::LLMQType llmqType, const uint256&
     return db.GetVoteForId(llmqType, id, msgHashRet);
 }
 
-void CSigningManager::StartWorkerThread()
+void CSigningManager::StartWorkerThread(PeerManager& peerman)
 {
     // can't start new thread if we have one running already
     if (workThread.joinable()) {
         assert(false);
     }
 
-    workThread = std::thread(&util::TraceThread, "sigshares", [this] { WorkThreadMain(); });
+    workThread = std::thread(&util::TraceThread, "sigshares", [this, &peerman] { WorkThreadMain(peerman); });
 }
 
 void CSigningManager::StopWorkerThread()
@@ -826,10 +831,10 @@ void CSigningManager::InterruptWorkerThread()
     workInterrupt();
 }
 
-void CSigningManager::WorkThreadMain()
+void CSigningManager::WorkThreadMain(PeerManager& peerman)
 {
     while (!workInterrupt) {
-        bool fMoreWork = ProcessPendingRecoveredSigs();
+        bool fMoreWork = ProcessPendingRecoveredSigs(peerman);
 
         Cleanup();
 
diff --git a/src/llmq/signing.h b/src/llmq/signing.h
@@ -162,7 +162,6 @@ class CSigningManager
     const CActiveMasternodeManager* const m_mn_activeman;
     const CChainState& m_chainstate;
     const CQuorumManager& qman;
-    const std::unique_ptr<PeerManager>& m_peerman;
 
     mutable Mutex cs_pending;
     // Incoming and not verified yet
@@ -178,12 +177,12 @@ class CSigningManager
 
 public:
     CSigningManager(const CActiveMasternodeManager* const mn_activeman, const CChainState& chainstate,
-                    const CQuorumManager& _qman, const std::unique_ptr<PeerManager>& peerman, bool fMemory, bool fWipe);
+                    const CQuorumManager& _qman, bool fMemory, bool fWipe);
 
     bool AlreadyHave(const CInv& inv) const;
     bool GetRecoveredSigForGetData(const uint256& hash, CRecoveredSig& ret) const;
 
-    PeerMsgRet ProcessMessage(const CNode& pnode, const std::string& msg_type, CDataStream& vRecv);
+    PeerMsgRet ProcessMessage(const CNode& pnode, PeerManager& peerman, const std::string& msg_type, CDataStream& vRecv);
 
     // This is called when a recovered signature was was reconstructed from another P2P message and is known to be valid
     // This is the case for example when a signature appears as part of InstantSend or ChainLocks
@@ -196,16 +195,18 @@ class CSigningManager
     void TruncateRecoveredSig(Consensus::LLMQType llmqType, const uint256& id);
 
 private:
-    PeerMsgRet ProcessMessageRecoveredSig(const CNode& pfrom, const std::shared_ptr<const CRecoveredSig>& recoveredSig);
+    PeerMsgRet ProcessMessageRecoveredSig(const CNode& pfrom, PeerManager& peerman,
+                                          const std::shared_ptr<const CRecoveredSig>& recoveredSig);
 
     void CollectPendingRecoveredSigsToVerify(size_t maxUniqueSessions,
             std::unordered_map<NodeId, std::list<std::shared_ptr<const CRecoveredSig>>>& retSigShares,
             std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>& retQuorums);
-    void ProcessPendingReconstructedRecoveredSigs();
-    bool ProcessPendingRecoveredSigs(); // called from the worker thread of CSigSharesManager
+    void ProcessPendingReconstructedRecoveredSigs(PeerManager& peerman);
+    bool ProcessPendingRecoveredSigs(PeerManager& peerman); // called from the worker thread of CSigSharesManager
 public:
     // TODO - should not be public!
-    void ProcessRecoveredSig(const std::shared_ptr<const CRecoveredSig>& recoveredSig);
+    void ProcessRecoveredSig(const std::shared_ptr<const CRecoveredSig>& recoveredSig, PeerManager& peerman);
+
 private:
     void Cleanup(); // called from the worker thread of CSigSharesManager
 
@@ -228,10 +229,10 @@ class CSigningManager
 private:
     std::thread workThread;
     CThreadInterrupt workInterrupt;
-    void WorkThreadMain();
+    void WorkThreadMain(PeerManager& peerman);
 
 public:
-    void StartWorkerThread();
+    void StartWorkerThread(PeerManager& peerman);
     void StopWorkerThread();
     void InterruptWorkerThread();
 };
diff --git a/src/llmq/signing_shares.cpp b/src/llmq/signing_shares.cpp
@@ -817,7 +817,7 @@ void CSigSharesManager::TryRecoverSig(const CQuorumCPtr& quorum, const uint256&
         }
     }
 
-    sigman.ProcessRecoveredSig(rs);
+    sigman.ProcessRecoveredSig(rs, *m_peerman);
 }
 
 CDeterministicMNCPtr CSigSharesManager::SelectMemberForRecovery(const CQuorumCPtr& quorum, const uint256 &id, int attempt)
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
@@ -5265,7 +5265,7 @@ void PeerManagerImpl::ProcessMessage(
         ProcessPeerMsgRet(m_llmq_ctx->qdkgsman->ProcessMessage(pfrom, *this, is_masternode, msg_type, vRecv), pfrom);
         ProcessPeerMsgRet(m_llmq_ctx->qman->ProcessMessage(pfrom, m_connman, msg_type, vRecv), pfrom);
         m_llmq_ctx->shareman->ProcessMessage(pfrom, m_sporkman, msg_type, vRecv);
-        ProcessPeerMsgRet(m_llmq_ctx->sigman->ProcessMessage(pfrom, msg_type, vRecv), pfrom);
+        ProcessPeerMsgRet(m_llmq_ctx->sigman->ProcessMessage(pfrom, *this, msg_type, vRecv), pfrom);
 
         if (msg_type == NetMsgType::CLSIG) {
             if (llmq::AreChainLocksEnabled(m_sporkman)) {

Original file line number	Diff line number	Diff line change
`@@ -349,8 +349,11 @@ void CRecoveredSigsDb::CleanupOldVotes(int64_t maxAge)`
`349`	`349`	`//////////////////`
`350`	`350`
`351`	`351`	`CSigningManager::CSigningManager(const CActiveMasternodeManager* const mn_activeman, const CChainState& chainstate,`
`352`		`- const CQuorumManager& _qman, const std::unique_ptr<PeerManager>& peerman, bool fMemory, bool fWipe) :`
`353`		`- db(fMemory, fWipe), m_mn_activeman(mn_activeman), m_chainstate(chainstate), qman(_qman), m_peerman(peerman)`
	`352`	`+ const CQuorumManager& _qman, bool fMemory, bool fWipe) :`
	`353`	`+ db(fMemory, fWipe),`
	`354`	`+ m_mn_activeman(mn_activeman),`
	`355`	`+ m_chainstate(chainstate),`
	`356`	`+ qman(_qman)`
`354`	`357`	`{`
`355`	`358`	`}`
`356`	`359`
`@@ -381,13 +384,14 @@ bool CSigningManager::GetRecoveredSigForGetData(const uint256& hash, CRecoveredS`
`381`	`384`	`return true;`
`382`	`385`	`}`
`383`	`386`
`384`		`-PeerMsgRet CSigningManager::ProcessMessage(const CNode& pfrom, const std::string& msg_type, CDataStream& vRecv)`
	`387`	`+PeerMsgRet CSigningManager::ProcessMessage(const CNode& pfrom, PeerManager& peerman, const std::string& msg_type,`
	`388`	`+ CDataStream& vRecv)`
`385`	`389`	`{`
`386`	`390`	`if (msg_type == NetMsgType::QSIGREC) {`
`387`	`391`	`auto recoveredSig = std::make_shared<CRecoveredSig>();`
`388`	`392`	`vRecv >> *recoveredSig;`
`389`	`393`
`390`		`- return ProcessMessageRecoveredSig(pfrom, recoveredSig);`
	`394`	`+ return ProcessMessageRecoveredSig(pfrom, peerman, recoveredSig);`
`391`	`395`	`}`
`392`	`396`	`return {};`
`393`	`397`	`}`
`@@ -416,10 +420,11 @@ static bool PreVerifyRecoveredSig(const CQuorumManager& quorum_manager, const CR`
`416`	`420`	`return true;`
`417`	`421`	`}`
`418`	`422`
`419`		`-PeerMsgRet CSigningManager::ProcessMessageRecoveredSig(const CNode& pfrom, const std::shared_ptr<const CRecoveredSig>& recoveredSig)`
	`423`	`+PeerMsgRet CSigningManager::ProcessMessageRecoveredSig(const CNode& pfrom, PeerManager& peerman,`
	`424`	`+ const std::shared_ptr<const CRecoveredSig>& recoveredSig)`
`420`	`425`	`{`
`421`		`- WITH_LOCK(::cs_main, Assert(m_peerman)->EraseObjectRequest(pfrom.GetId(),`
`422`		`- CInv(MSG_QUORUM_RECOVERED_SIG, recoveredSig->GetHash())));`
	`426`	`+ WITH_LOCK(::cs_main,`
	`427`	`+ peerman.EraseObjectRequest(pfrom.GetId(), CInv(MSG_QUORUM_RECOVERED_SIG, recoveredSig->GetHash())));`
`423`	`428`
`424`	`429`	`bool ban = false;`
`425`	`430`	`if (!PreVerifyRecoveredSig(qman, *recoveredSig, ban)) {`
`@@ -517,22 +522,22 @@ void CSigningManager::CollectPendingRecoveredSigsToVerify(`
`517`	`522`	`}`
`518`	`523`	`}`
`519`	`524`
`520`		`-void CSigningManager::ProcessPendingReconstructedRecoveredSigs()`
	`525`	`+void CSigningManager::ProcessPendingReconstructedRecoveredSigs(PeerManager& peerman)`
`521`	`526`	`{`
`522`	`527`	`decltype(pendingReconstructedRecoveredSigs) m;`
`523`	`528`	`WITH_LOCK(cs_pending, swap(m, pendingReconstructedRecoveredSigs));`
`524`	`529`
`525`	`530`	`for (const auto& p : m) {`
`526`		`- ProcessRecoveredSig(p.second);`
	`531`	`+ ProcessRecoveredSig(p.second, peerman);`
`527`	`532`	`}`
`528`	`533`	`}`
`529`	`534`
`530`		`-bool CSigningManager::ProcessPendingRecoveredSigs()`
	`535`	`+bool CSigningManager::ProcessPendingRecoveredSigs(PeerManager& peerman)`
`531`	`536`	`{`
`532`	`537`	`std::unordered_map<NodeId, std::list<std::shared_ptr<const CRecoveredSig>>> recSigsByNode;`
`533`	`538`	`std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher> quorums;`
`534`	`539`
`535`		`- ProcessPendingReconstructedRecoveredSigs();`
	`540`	`+ ProcessPendingReconstructedRecoveredSigs(peerman);`
`536`	`541`
`537`	`542`	`const size_t nMaxBatchSize{32};`
`538`	`543`	`CollectPendingRecoveredSigsToVerify(nMaxBatchSize, recSigsByNode, quorums);`
`@@ -575,7 +580,7 @@ bool CSigningManager::ProcessPendingRecoveredSigs()`
`575`	`580`
`576`	`581`	`if (batchVerifier.badSources.count(nodeId)) {`
`577`	`582`	`LogPrint(BCLog::LLMQ, "CSigningManager::%s -- invalid recSig from other node, banning peer=%d\n", __func__, nodeId);`
`578`		`- Assert(m_peerman)->Misbehaving(nodeId, 100);`
	`583`	`+ peerman.Misbehaving(nodeId, 100);`
`579`	`584`	`continue;`
`580`	`585`	`}`
`581`	`586`
`@@ -584,15 +589,15 @@ bool CSigningManager::ProcessPendingRecoveredSigs()`
`584`	`589`	`continue;`
`585`	`590`	`}`
`586`	`591`
`587`		`- ProcessRecoveredSig(recSig);`
	`592`	`+ ProcessRecoveredSig(recSig, peerman);`
`588`	`593`	`}`
`589`	`594`	`}`
`590`	`595`
`591`	`596`	`return recSigsByNode.size() >= nMaxBatchSize;`
`592`	`597`	`}`
`593`	`598`
`594`	`599`	`// signature must be verified already`
`595`		`-void CSigningManager::ProcessRecoveredSig(const std::shared_ptr<const CRecoveredSig>& recoveredSig)`
	`600`	`+void CSigningManager::ProcessRecoveredSig(const std::shared_ptr<const CRecoveredSig>& recoveredSig, PeerManager& peerman)`
`596`	`601`	`{`
`597`	`602`	`auto llmqType = recoveredSig->getLlmqType();`
`598`	`603`
`@@ -631,12 +636,12 @@ void CSigningManager::ProcessRecoveredSig(const std::shared_ptr<const CRecovered`
`631`	`636`	`WITH_LOCK(cs_pending, pendingReconstructedRecoveredSigs.erase(recoveredSig->GetHash()));`
`632`	`637`
`633`	`638`	`if (m_mn_activeman != nullptr) {`
`634`		`- Assert(m_peerman)->RelayRecoveredSig(recoveredSig->GetHash());`
	`639`	`+ peerman.RelayRecoveredSig(recoveredSig->GetHash());`
`635`	`640`	`}`
`636`	`641`
`637`	`642`	`auto listeners = WITH_LOCK(cs_listeners, return recoveredSigsListeners);`
`638`	`643`	`for (auto& l : listeners) {`
`639`		`- Assert(m_peerman)->PostProcessMessage(l->HandleNewRecoveredSig(*recoveredSig));`
	`644`	`+ peerman.PostProcessMessage(l->HandleNewRecoveredSig(*recoveredSig));`
`640`	`645`	`}`
`641`	`646`
`642`	`647`	`GetMainSignals().NotifyRecoveredSig(recoveredSig);`
`@@ -799,14 +804,14 @@ bool CSigningManager::GetVoteForId(Consensus::LLMQType llmqType, const uint256&`
`799`	`804`	`return db.GetVoteForId(llmqType, id, msgHashRet);`
`800`	`805`	`}`
`801`	`806`
`802`		`-void CSigningManager::StartWorkerThread()`
	`807`	`+void CSigningManager::StartWorkerThread(PeerManager& peerman)`
`803`	`808`	`{`
`804`	`809`	`// can't start new thread if we have one running already`
`805`	`810`	`if (workThread.joinable()) {`
`806`	`811`	`assert(false);`
`807`	`812`	`}`
`808`	`813`
`809`		`- workThread = std::thread(&util::TraceThread, "sigshares", [this] { WorkThreadMain(); });`
	`814`	`+ workThread = std::thread(&util::TraceThread, "sigshares", [this, &peerman] { WorkThreadMain(peerman); });`
`810`	`815`	`}`
`811`	`816`
`812`	`817`	`void CSigningManager::StopWorkerThread()`
`@@ -826,10 +831,10 @@ void CSigningManager::InterruptWorkerThread()`
`826`	`831`	`workInterrupt();`
`827`	`832`	`}`
`828`	`833`
`829`		`-void CSigningManager::WorkThreadMain()`
	`834`	`+void CSigningManager::WorkThreadMain(PeerManager& peerman)`
`830`	`835`	`{`
`831`	`836`	`while (!workInterrupt) {`
`832`		`- bool fMoreWork = ProcessPendingRecoveredSigs();`
	`837`	`+ bool fMoreWork = ProcessPendingRecoveredSigs(peerman);`
`833`	`838`
`834`	`839`	`Cleanup();`
`835`	`840`
Original file line number	Diff line number	Diff line change
`@@ -817,7 +817,7 @@ void CSigSharesManager::TryRecoverSig(const CQuorumCPtr& quorum, const uint256&`
`817`	`817`	`}`
`818`	`818`	`}`
`819`	`819`
`820`		`- sigman.ProcessRecoveredSig(rs);`
	`820`	`+ sigman.ProcessRecoveredSig(rs, *m_peerman);`
`821`	`821`	`}`
`822`	`822`
`823`	`823`	`CDeterministicMNCPtr CSigSharesManager::SelectMemberForRecovery(const CQuorumCPtr& quorum, const uint256 &id, int attempt)`