instantsend|sigs: Sleep when there is no more work (#3988)

* instantsend|sigs: Sleep when there is no more work

Instead of sleeping only when no work has been done.
Avoids useless cycles, improves batching.

* llmq: Add and use nMaxBatchSize

* llmq: Compare to what we got in return, not what we verified at the end

It might happen that we get 32 pending but do only verify less than 32 and in this case we would assume there is no more work but it could still be more in the pipeline from my understanding.

* llmq: Rename more_work -> fMoreWork

* llmq: Be consistent with the other fMoreWork initialization

Co-authored-by: xdustinface <xdustinfacex@gmail.com>

src/llmq/quorums_instantsend.cpp

@@ -731,6 +731,7 @@ bool CInstantSendManager::PreVerifyInstantSendLock(const llmq::CInstantSendLock&
 bool CInstantSendManager::ProcessPendingInstantSendLocks()
 {
     decltype(pendingInstantSendLocks) pend;
+    bool fMoreWork{false};
 
     {
         LOCK(cs);
@@ -745,6 +746,7 @@ bool CInstantSendManager::ProcessPendingInstantSendLocks()
                 pend.emplace(it->first, std::move(it->second));
                 pendingInstantSendLocks.erase(it);
             }
+            fMoreWork = true;
         }
     }
 
@@ -776,7 +778,7 @@ bool CInstantSendManager::ProcessPendingInstantSendLocks()
         ProcessPendingInstantSendLocks(dkgInterval, pend, true);
     }
 
-    return true;
+    return fMoreWork;
 }
 
 std::unordered_set<uint256> CInstantSendManager::ProcessPendingInstantSendLocks(int signOffset, const std::unordered_map<uint256, std::pair<NodeId, CInstantSendLockPtr>, StaticSaltedHasher>& pend, bool ban)
@@ -1365,7 +1367,7 @@ void CInstantSendManager::AskNodesForLockedTx(const uint256& txid)
     }
 }
 
-bool CInstantSendManager::ProcessPendingRetryLockTxs()
+void CInstantSendManager::ProcessPendingRetryLockTxs()
 {
     decltype(pendingRetryTxs) retryTxs;
     {
@@ -1374,11 +1376,11 @@ bool CInstantSendManager::ProcessPendingRetryLockTxs()
     }
 
     if (retryTxs.empty()) {
-        return false;
+        return;
     }
 
     if (!IsInstantSendEnabled()) {
-        return false;
+        return;
     }
 
     int retryCount = 0;
@@ -1428,8 +1430,6 @@ bool CInstantSendManager::ProcessPendingRetryLockTxs()
         LogPrint(BCLog::INSTANTSEND, "CInstantSendManager::%s -- retried %d TXs. nonLockedTxs.size=%d\n", __func__,
                  retryCount, nonLockedTxs.size());
     }
-
-    return retryCount != 0;
 }
 
 bool CInstantSendManager::AlreadyHave(const CInv& inv)
@@ -1521,18 +1521,14 @@ size_t CInstantSendManager::GetInstantSendLockCount()
 void CInstantSendManager::WorkThreadMain()
 {
     while (!workInterrupt) {
-        bool didWork = false;
+        bool fMoreWork = ProcessPendingInstantSendLocks();
+        ProcessPendingRetryLockTxs();
 
-        didWork |= ProcessPendingInstantSendLocks();
+        if (!fMoreWork && !workInterrupt.sleep_for(std::chrono::milliseconds(100))) {
-        didWork |= ProcessPendingRetryLockTxs();
-
-        if (!didWork) {
-            if (!workInterrupt.sleep_for(std::chrono::milliseconds(100))) {
             return;
         }
     }
 }
-}
 
 bool IsInstantSendEnabled()
 {

src/llmq/quorums_instantsend.h

@@ -159,7 +159,7 @@ public:
     void ResolveBlockConflicts(const uint256& islockHash, const CInstantSendLock& islock);
     void RemoveChainLockConflictingLock(const uint256& islockHash, const CInstantSendLock& islock);
     static void AskNodesForLockedTx(const uint256& txid);
-    bool ProcessPendingRetryLockTxs();
+    void ProcessPendingRetryLockTxs();
 
     bool AlreadyHave(const CInv& inv);
     bool GetInstantSendLockByHash(const uint256& hash, CInstantSendLock& ret);

src/llmq/quorums_signing.cpp

@@ -621,7 +621,8 @@ bool CSigningManager::ProcessPendingRecoveredSigs()
 
     ProcessPendingReconstructedRecoveredSigs();
 
-    CollectPendingRecoveredSigsToVerify(32, recSigsByNode, quorums);
+    const size_t nMaxBatchSize{32};
+    CollectPendingRecoveredSigsToVerify(nMaxBatchSize, recSigsByNode, quorums);
     if (recSigsByNode.empty()) {
         return false;
     }
@@ -675,7 +676,7 @@ bool CSigningManager::ProcessPendingRecoveredSigs()
         }
     }
 
-    return true;
+    return recSigsByNode.size() >= nMaxBatchSize;
 }
 
 // signature must be verified already

src/llmq/quorums_signing_shares.cpp

@@ -639,7 +639,8 @@ bool CSigSharesManager::ProcessPendingSigShares(CConnman& connman)
     std::unordered_map<NodeId, std::vector<CSigShare>> sigSharesByNodes;
     std::unordered_map<std::pair<Consensus::LLMQType, uint256>, CQuorumCPtr, StaticSaltedHasher> quorums;
 
-    CollectPendingSigSharesToVerify(32, sigSharesByNodes, quorums);
+    const size_t nMaxBatchSize{32};
+    CollectPendingSigSharesToVerify(nMaxBatchSize, sigSharesByNodes, quorums);
     if (sigSharesByNodes.empty()) {
         return false;
     }
@@ -704,7 +705,7 @@ bool CSigSharesManager::ProcessPendingSigShares(CConnman& connman)
         ProcessPendingSigShares(v, quorums, connman);
     }
 
-    return true;
+    return sigSharesByNodes.size() >= nMaxBatchSize;
 }
 
 // It's ensured that no duplicates are passed to this method
@@ -1501,12 +1502,12 @@ void CSigSharesManager::WorkThreadMain()
             continue;
         }
 
-        bool didWork = false;
+        bool fMoreWork{false};
 
         RemoveBannedNodeStates();
-        didWork |= quorumSigningManager->ProcessPendingRecoveredSigs();
+        fMoreWork |= quorumSigningManager->ProcessPendingRecoveredSigs();
-        didWork |= ProcessPendingSigShares(*g_connman);
+        fMoreWork |= ProcessPendingSigShares(*g_connman);
-        didWork |= SignPendingSigShares();
+        SignPendingSigShares();
 
         if (GetTimeMillis() - lastSendTime > 100) {
             SendMessages();
@@ -1517,13 +1518,11 @@ void CSigSharesManager::WorkThreadMain()
         quorumSigningManager->Cleanup();
 
         // TODO Wakeup when pending signing is needed?
-        if (!didWork) {
+        if (!fMoreWork && !workInterrupt.sleep_for(std::chrono::milliseconds(100))) {
-            if (!workInterrupt.sleep_for(std::chrono::milliseconds(100))) {
             return;
         }
     }
 }
-}
 
 void CSigSharesManager::AsyncSign(const CQuorumCPtr& quorum, const uint256& id, const uint256& msgHash)
 {
@@ -1531,7 +1530,7 @@ void CSigSharesManager::AsyncSign(const CQuorumCPtr& quorum, const uint256& id,
     pendingSigns.emplace_back(quorum, id, msgHash);
 }
 
-bool CSigSharesManager::SignPendingSigShares()
+void CSigSharesManager::SignPendingSigShares()
 {
     std::vector<std::tuple<const CQuorumCPtr, uint256, uint256>> v;
     {
@@ -1557,8 +1556,6 @@ bool CSigSharesManager::SignPendingSigShares()
             }
         }
     }
-
-    return !v.empty();
 }
 
 CSigShare CSigSharesManager::CreateSigShare(const CQuorumCPtr& quorum, const uint256& id, const uint256& msgHash)

src/llmq/quorums_signing_shares.h

@@ -451,7 +451,7 @@ private:
     void CollectSigSharesToSend(std::unordered_map<NodeId, std::unordered_map<uint256, CBatchedSigShares, StaticSaltedHasher>>& sigSharesToSend);
     void CollectSigSharesToSendConcentrated(std::unordered_map<NodeId, std::vector<CSigShare>>& sigSharesToSend, const std::vector<CNode*>& vNodes);
     void CollectSigSharesToAnnounce(std::unordered_map<NodeId, std::unordered_map<uint256, CSigSharesInv, StaticSaltedHasher>>& sigSharesToAnnounce);
-    bool SignPendingSigShares();
+    void SignPendingSigShares();
     void WorkThreadMain();
 };