Optimize sleeping behavior in CSigSharesManager::WorkThreadMain (#2707)

* Don't sleep in WorkThreadMain when CPU intensive work was done

When the current iteration resulted in CPU intensive work, it's likely that
the next iteration will result in work as well. Do not sleep in that case,
as we're otherwise wasting (unused) CPU resources.

* No matter how fast we process sig shares, always force 100ms between sending

* Apply review suggestions

Author: codablock

src/llmq/quorums_signing.cpp

@@ -421,14 +421,14 @@ void CSigningManager::CollectPendingRecoveredSigsToVerify(
     }
 }
 
-void CSigningManager::ProcessPendingRecoveredSigs(CConnman& connman)
+bool CSigningManager::ProcessPendingRecoveredSigs(CConnman& connman)
 {
     std::map<NodeId, std::list<CRecoveredSig>> recSigsByNode;
     std::map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr> quorums;
 
     CollectPendingRecoveredSigsToVerify(32, recSigsByNode, quorums);
     if (recSigsByNode.empty()) {
-        return;
+        return false;
     }
 
     // It's ok to perform insecure batched verification here as we verify against the quorum public keys, which are not
@@ -474,6 +474,8 @@ void CSigningManager::ProcessPendingRecoveredSigs(CConnman& connman)
             ProcessRecoveredSig(nodeId, recSig, quorum, connman);
         }
     }
+
+    return true;
 }
 
 // signature must be verified already

src/llmq/quorums_signing.h

@@ -157,7 +157,7 @@ class CSigningManager
     bool PreVerifyRecoveredSig(NodeId nodeId, const CRecoveredSig& recoveredSig, bool& retBan);
 
     void CollectPendingRecoveredSigsToVerify(size_t maxUniqueSessions, std::map<NodeId, std::list<CRecoveredSig>>& retSigShares, std::map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr>& retQuorums);
-    void ProcessPendingRecoveredSigs(CConnman& connman); // called from the worker thread of CSigSharesManager
+    bool ProcessPendingRecoveredSigs(CConnman& connman); // called from the worker thread of CSigSharesManager
     void ProcessRecoveredSig(NodeId nodeId, const CRecoveredSig& recoveredSig, const CQuorumCPtr& quorum, CConnman& connman);
     void Cleanup(); // called from the worker thread of CSigSharesManager
 

src/llmq/quorums_signing_shares.cpp

@@ -454,14 +454,14 @@ void CSigSharesManager::CollectPendingSigSharesToVerify(
     }
 }
 
-void CSigSharesManager::ProcessPendingSigShares(CConnman& connman)
+bool CSigSharesManager::ProcessPendingSigShares(CConnman& connman)
 {
     std::map<NodeId, std::vector<CSigShare>> sigSharesByNodes;
     std::map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr> quorums;
 
     CollectPendingSigSharesToVerify(32, sigSharesByNodes, quorums);
     if (sigSharesByNodes.empty()) {
-        return;
+        return false;
     }
 
     // It's ok to perform insecure batched verification here as we verify against the quorum public key shares,
@@ -521,6 +521,8 @@ void CSigSharesManager::ProcessPendingSigShares(CConnman& connman)
 
         ProcessPendingSigSharesFromNode(nodeId, v, quorums, connman);
     }
+
+    return true;
 }
 
 // It's ensured that no duplicates are passed to this method
@@ -881,7 +883,7 @@ void CSigSharesManager::CollectSigSharesToAnnounce(std::map<NodeId, std::map<uin
     this->sigSharesToAnnounce.clear();
 }
 
-void CSigSharesManager::SendMessages()
+bool CSigSharesManager::SendMessages()
 {
     std::multimap<CService, NodeId> nodesByAddress;
     g_connman->ForEachNode([&nodesByAddress](CNode* pnode) {
@@ -943,6 +945,8 @@ void CSigSharesManager::SendMessages()
     if (didSend) {
         g_connman->WakeSelect();
     }
+
+    return didSend;
 }
 
 void CSigSharesManager::Cleanup()
@@ -1110,6 +1114,8 @@ void CSigSharesManager::WorkThreadMain()
 {
     workInterrupt.reset();
 
+    int64_t lastSendTime = 0;
+
     while (!workInterrupt) {
         if (!quorumSigningManager || !g_connman || !quorumSigningManager) {
             if (!workInterrupt.sleep_for(std::chrono::milliseconds(100))) {
@@ -1118,17 +1124,26 @@ void CSigSharesManager::WorkThreadMain()
             continue;
         }
 
+        bool didWork = false;
+
         RemoveBannedNodeStates();
-        quorumSigningManager->ProcessPendingRecoveredSigs(*g_connman);
-        ProcessPendingSigShares(*g_connman);
-        SignPendingSigShares();
-        SendMessages();
+        didWork |= quorumSigningManager->ProcessPendingRecoveredSigs(*g_connman);
+        didWork |= ProcessPendingSigShares(*g_connman);
+        didWork |= SignPendingSigShares();
+
+        if (GetTimeMillis() - lastSendTime > 100) {
+            SendMessages();
+            lastSendTime = GetTimeMillis();
+        }
+
         Cleanup();
         quorumSigningManager->Cleanup();
 
         // TODO Wakeup when pending signing is needed?
-        if (!workInterrupt.sleep_for(std::chrono::milliseconds(100))) {
-            return;
+        if (!didWork) {
+            if (!workInterrupt.sleep_for(std::chrono::milliseconds(100))) {
+                return;
+            }
         }
     }
 }
@@ -1139,7 +1154,7 @@ void CSigSharesManager::AsyncSign(const CQuorumCPtr& quorum, const uint256& id,
     pendingSigns.emplace_back(quorum, id, msgHash);
 }
 
-void CSigSharesManager::SignPendingSigShares()
+bool CSigSharesManager::SignPendingSigShares()
 {
     std::vector<std::tuple<const CQuorumCPtr, uint256, uint256>> v;
     {
@@ -1150,6 +1165,8 @@ void CSigSharesManager::SignPendingSigShares()
     for (auto& t : v) {
         Sign(std::get<0>(t), std::get<1>(t), std::get<2>(t));
     }
+
+    return !v.empty();
 }
 
 void CSigSharesManager::Sign(const CQuorumCPtr& quorum, const uint256& id, const uint256& msgHash)

src/llmq/quorums_signing_shares.h

@@ -212,7 +212,7 @@ class CSigSharesManager
     bool PreVerifyBatchedSigShares(NodeId nodeId, const CBatchedSigShares& batchedSigShares, bool& retBan);
 
     void CollectPendingSigSharesToVerify(size_t maxUniqueSessions, std::map<NodeId, std::vector<CSigShare>>& retSigShares, std::map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr>& retQuorums);
-    void ProcessPendingSigShares(CConnman& connman);
+    bool ProcessPendingSigShares(CConnman& connman);
 
     void ProcessPendingSigSharesFromNode(NodeId nodeId, const std::vector<CSigShare>& sigShares, const std::map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr>& quorums, CConnman& connman);
 
@@ -226,11 +226,11 @@ class CSigSharesManager
 
     void BanNode(NodeId nodeId);
 
-    void SendMessages();
+    bool SendMessages();
     void CollectSigSharesToRequest(std::map<NodeId, std::map<uint256, CSigSharesInv>>& sigSharesToRequest);
     void CollectSigSharesToSend(std::map<NodeId, std::map<uint256, CBatchedSigShares>>& sigSharesToSend);
     void CollectSigSharesToAnnounce(std::map<NodeId, std::map<uint256, CSigSharesInv>>& sigSharesToAnnounce);
-    void SignPendingSigShares();
+    bool SignPendingSigShares();
     void WorkThreadMain();
 };