Merge #6954: feat: convert CSigSharesManager to use Mutex instead of RecursiveMutex

e98ba12 refactor: streamline signature recovery process in CSigSharesManager (pasta)
0a08b2a refactor: enhance CSigSharesManager for single-member quorum handling (pasta)
582e6e4 chore: run clang-format (pasta)
8832775 feat: convert CSigSharesManager to use Mutex instead of RecursiveMutex (pasta)

Pull request description:

  update lock requirements for various methods. This change enhances thread safety and simplifies locking semantics throughout the class.

  ## Issue being fixed or feature implemented
  After other recent lock contention improvements, this RecursiveMutex is now the source of the most contention during high load. Convert the mutex from Recursive to non-recursive. Surprisingly compiler is happy with this and we don't need more invasive changes

  ## What was done?

  ## How Has This Been Tested?
  Builds

  ## Breaking Changes
  None

  ## Checklist:
    _Go over all the following points, and put an `x` in all the boxes that apply._
  - [ ] I have performed a self-review of my own code
  - [ ] I have commented my code, particularly in hard-to-understand areas
  - [ ] I have added or updated relevant unit/integration/functional/e2e tests
  - [ ] I have made corresponding changes to the documentation
  - [x] I have assigned this pull request to a milestone _(for repository code-owners and collaborators only)_

ACKs for top commit:
  knst:
    utACK e98ba12
  UdjinM6:
    utACK e98ba12

Tree-SHA512: ac9a9df73ed099e8518954b80cc83cf5e92122162ecb26d2a80ace65c7ef7c65dcd4e07fddbd60fece549dcfbcf42b2698cb60b4149aea9720f9be3dc1bbcff5

Author: PastaPastaPasta

src/llmq/signing_shares.cpp

@@ -783,6 +783,7 @@ void CSigSharesManager::TryRecoverSig(const CQuorum& quorum, const uint256& id,
 
     std::vector<CBLSSignature> sigSharesForRecovery;
     std::vector<CBLSId> idsForRecovery;
+    std::shared_ptr<CRecoveredSig> singleMemberRecoveredSig;
     {
         LOCK(cs);
 
@@ -805,10 +806,8 @@ void CSigSharesManager::TryRecoverSig(const CQuorum& quorum, const uint256& id,
             LogPrint(BCLog::LLMQ_SIGS, "CSigSharesManager::%s -- recover single-node signature. id=%s, msgHash=%s\n",
                      __func__, id.ToString(), msgHash.ToString());
 
-            auto rs = std::make_shared<CRecoveredSig>(quorum.params.type, quorum.qc->quorumHash, id, msgHash,
+            singleMemberRecoveredSig = std::make_shared<CRecoveredSig>(quorum.params.type, quorum.qc->quorumHash, id, msgHash,
                                                       recoveredSig);
-            sigman.ProcessRecoveredSig(rs, m_peerman);
-            return; // end of single-quorum processing
         }
 
         sigSharesForRecovery.reserve((size_t) quorum.params.threshold);
@@ -825,6 +824,12 @@ void CSigSharesManager::TryRecoverSig(const CQuorum& quorum, const uint256& id,
         }
     }
 
+    // Handle single-member quorum case after releasing the lock
+    if (singleMemberRecoveredSig) {
+        sigman.ProcessRecoveredSig(singleMemberRecoveredSig, m_peerman);
+        return; // end of single-quorum processing
+    }
+
     // now recover it
     cxxtimer::Timer t(true);
     CBLSSignature recoveredSig;

src/llmq/signing_shares.h

@@ -376,7 +376,7 @@ class CSigSharesManager : public CRecoveredSigsListener
     static constexpr int64_t MAX_SEND_FOR_RECOVERY_TIMEOUT{10000};
     static constexpr size_t MAX_MSGS_SIG_SHARES{32};
 
-    RecursiveMutex cs;
+    Mutex cs;
 
     std::thread workThread;
     CThreadInterrupt workInterrupt;
@@ -424,73 +424,79 @@ class CSigSharesManager : public CRecoveredSigsListener
                                const CQuorumManager& _qman, const CSporkManager& sporkman);
     ~CSigSharesManager() override;
 
-    void StartWorkerThread();
-    void StopWorkerThread();
-    void RegisterAsRecoveredSigsListener();
-    void UnregisterAsRecoveredSigsListener();
-    void InterruptWorkerThread();
+    void StartWorkerThread() EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    void StopWorkerThread() EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    void RegisterAsRecoveredSigsListener() EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    void UnregisterAsRecoveredSigsListener() EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    void InterruptWorkerThread() EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
-    void ProcessMessage(const CNode& pnode, const std::string& msg_type, CDataStream& vRecv);
+    void ProcessMessage(const CNode& pnode, const std::string& msg_type, CDataStream& vRecv) EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
     void AsyncSign(CQuorumCPtr quorum, const uint256& id, const uint256& msgHash)
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns);
-    std::optional<CSigShare> CreateSigShare(const CQuorum& quorum, const uint256& id, const uint256& msgHash) const;
-    void ForceReAnnouncement(const CQuorum& quorum, Consensus::LLMQType llmqType, const uint256& id, const uint256& msgHash);
+        EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns, !cs);
+    std::optional<CSigShare> CreateSigShare(const CQuorum& quorum, const uint256& id, const uint256& msgHash) const
+        EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    void ForceReAnnouncement(const CQuorum& quorum, Consensus::LLMQType llmqType, const uint256& id,
+                             const uint256& msgHash) EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
-    [[nodiscard]] MessageProcessingResult HandleNewRecoveredSig(const CRecoveredSig& recoveredSig) override;
+    [[nodiscard]] MessageProcessingResult HandleNewRecoveredSig(const CRecoveredSig& recoveredSig) override
+        EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
     static CDeterministicMNCPtr SelectMemberForRecovery(const CQuorum& quorum, const uint256& id, int attempt);
 
     bool AsyncSignIfMember(Consensus::LLMQType llmqType, CSigningManager& sigman, const uint256& id,
                            const uint256& msgHash, const uint256& quorumHash = uint256(), bool allowReSign = false,
-                           bool allowDiffMsgHashSigning = false)
-        EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns);
+                           bool allowDiffMsgHashSigning = false) EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns, !cs);
 
-    void NotifyRecoveredSig(const std::shared_ptr<const CRecoveredSig>& sig) const;
+    void NotifyRecoveredSig(const std::shared_ptr<const CRecoveredSig>& sig) const EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
 private:
     // all of these return false when the currently processed message should be aborted (as each message actually contains multiple messages)
-    bool ProcessMessageSigSesAnn(const CNode& pfrom, const CSigSesAnn& ann);
-    bool ProcessMessageSigSharesInv(const CNode& pfrom, const CSigSharesInv& inv);
-    bool ProcessMessageGetSigShares(const CNode& pfrom, const CSigSharesInv& inv);
-    bool ProcessMessageBatchedSigShares(const CNode& pfrom, const CBatchedSigShares& batchedSigShares);
-    void ProcessMessageSigShare(NodeId fromId, const CSigShare& sigShare);
+    bool ProcessMessageSigSesAnn(const CNode& pfrom, const CSigSesAnn& ann) EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    bool ProcessMessageSigSharesInv(const CNode& pfrom, const CSigSharesInv& inv) EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    bool ProcessMessageGetSigShares(const CNode& pfrom, const CSigSharesInv& inv) EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    bool ProcessMessageBatchedSigShares(const CNode& pfrom, const CBatchedSigShares& batchedSigShares)
+        EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    void ProcessMessageSigShare(NodeId fromId, const CSigShare& sigShare) EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
     static bool VerifySigSharesInv(Consensus::LLMQType llmqType, const CSigSharesInv& inv);
     static bool PreVerifyBatchedSigShares(const CActiveMasternodeManager& mn_activeman, const CQuorumManager& quorum_manager,
                                           const CSigSharesNodeState::SessionInfo& session, const CBatchedSigShares& batchedSigShares, bool& retBan);
 
     bool CollectPendingSigSharesToVerify(
         size_t maxUniqueSessions, std::unordered_map<NodeId, std::vector<CSigShare>>& retSigShares,
-        std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>& retQuorums);
-    bool ProcessPendingSigShares();
+        std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>& retQuorums)
+        EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    bool ProcessPendingSigShares() EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
     void ProcessPendingSigShares(
         const std::vector<CSigShare>& sigSharesToProcess,
-        const std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>& quorums);
+        const std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher>& quorums)
+        EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
-    void ProcessSigShare(const CSigShare& sigShare, const CQuorumCPtr& quorum);
-    void TryRecoverSig(const CQuorum& quorum, const uint256& id, const uint256& msgHash);
+    void ProcessSigShare(const CSigShare& sigShare, const CQuorumCPtr& quorum) EXCLUSIVE_LOCKS_REQUIRED(!cs);
+    void TryRecoverSig(const CQuorum& quorum, const uint256& id, const uint256& msgHash) EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
-    bool GetSessionInfoByRecvId(NodeId nodeId, uint32_t sessionId, CSigSharesNodeState::SessionInfo& retInfo);
+    bool GetSessionInfoByRecvId(NodeId nodeId, uint32_t sessionId, CSigSharesNodeState::SessionInfo& retInfo)
+        EXCLUSIVE_LOCKS_REQUIRED(!cs);
     static CSigShare RebuildSigShare(const CSigSharesNodeState::SessionInfo& session, const std::pair<uint16_t, CBLSLazySignature>& in);
 
-    void Cleanup();
+    void Cleanup() EXCLUSIVE_LOCKS_REQUIRED(!cs);
     void RemoveSigSharesForSession(const uint256& signHash) EXCLUSIVE_LOCKS_REQUIRED(cs);
-    void RemoveBannedNodeStates();
+    void RemoveBannedNodeStates() EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
-    void BanNode(NodeId nodeId);
+    void BanNode(NodeId nodeId) EXCLUSIVE_LOCKS_REQUIRED(!cs);
 
-    bool SendMessages();
+    bool SendMessages() EXCLUSIVE_LOCKS_REQUIRED(!cs);
     void CollectSigSharesToRequest(std::unordered_map<NodeId, Uint256HashMap<CSigSharesInv>>& sigSharesToRequest)
         EXCLUSIVE_LOCKS_REQUIRED(cs);
     void CollectSigSharesToSend(std::unordered_map<NodeId, Uint256HashMap<CBatchedSigShares>>& sigSharesToSend)
         EXCLUSIVE_LOCKS_REQUIRED(cs);
     void CollectSigSharesToSendConcentrated(std::unordered_map<NodeId, std::vector<CSigShare>>& sigSharesToSend, const std::vector<CNode*>& vNodes) EXCLUSIVE_LOCKS_REQUIRED(cs);
     void CollectSigSharesToAnnounce(std::unordered_map<NodeId, Uint256HashMap<CSigSharesInv>>& sigSharesToAnnounce)
         EXCLUSIVE_LOCKS_REQUIRED(cs);
-    void SignPendingSigShares() EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns);
-    void WorkThreadMain() EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns);
+    void SignPendingSigShares() EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns, !cs);
+    void WorkThreadMain() EXCLUSIVE_LOCKS_REQUIRED(!cs_pendingSigns, !cs);
 };
 } // namespace llmq