Merge remaining bits of #11824: Block ActivateBestChain to empty validationinterface queue

src/net_processing.cpp

@@ -1183,16 +1183,16 @@ void static ProcessGetBlockData(CNode* pfrom, const Consensus::Params& consensus
         send = false;
     }
     // Avoid leaking prune-height by never sending blocks below the NODE_NETWORK_LIMITED threshold
-                if (send && !pfrom->fWhitelisted && (
-                        (((pfrom->GetLocalServices() & NODE_NETWORK_LIMITED) == NODE_NETWORK_LIMITED) && ((pfrom->GetLocalServices() & NODE_NETWORK) != NODE_NETWORK) && (chainActive.Tip()->nHeight - mi->second->nHeight > (int)NODE_NETWORK_LIMITED_MIN_BLOCKS + 2 /* add two blocks buffer extension for possible races */) )
-                   )) {
-                    LogPrint(BCLog::NET, "Ignore block request below NODE_NETWORK_LIMITED threshold from peer=%d\n", pfrom->GetId());
+    if (send && !pfrom->fWhitelisted && (
+            (((pfrom->GetLocalServices() & NODE_NETWORK_LIMITED) == NODE_NETWORK_LIMITED) && ((pfrom->GetLocalServices() & NODE_NETWORK) != NODE_NETWORK) && (chainActive.Tip()->nHeight - mi->second->nHeight > (int)NODE_NETWORK_LIMITED_MIN_BLOCKS + 2 /* add two blocks buffer extension for possible races */) )
+       )) {
+        LogPrint(BCLog::NET, "Ignore block request below NODE_NETWORK_LIMITED threshold from peer=%d\n", pfrom->GetId());
 
-                    //disconnect node and prevent it from stalling (would otherwise wait for the missing block)
-                    pfrom->fDisconnect = true;
-                    send = false;
-                }
-                // Pruned nodes may have deleted the block, so check whether
+        //disconnect node and prevent it from stalling (would otherwise wait for the missing block)
+        pfrom->fDisconnect = true;
+        send = false;
+    }
+    // Pruned nodes may have deleted the block, so check whether
     // it's available before trying to send.
     if (send && (mi->second->nStatus & BLOCK_HAVE_DATA))
     {

src/scheduler.cpp

@@ -206,3 +206,8 @@ void SingleThreadedSchedulerClient::EmptyQueue() {
         should_continue = !m_callbacks_pending.empty();
     }
 }
+
+size_t SingleThreadedSchedulerClient::CallbacksPending() {
+    LOCK(m_cs_callbacks_pending);
+    return m_callbacks_pending.size();
+}

src/scheduler.h

@@ -108,6 +108,8 @@ class SingleThreadedSchedulerClient {
     // Processes all remaining queue members on the calling thread, blocking until queue is empty
     // Must be called after the CScheduler has no remaining processing threads!
     void EmptyQueue();
+
+    size_t CallbacksPending();
 };
 
 #endif

src/test/test_dash.cpp

@@ -86,9 +86,9 @@ TestingSetup::TestingSetup(const std::string& chainName) : BasicTestingSetup(cha
         fs::create_directories(pathTemp);
         gArgs.ForceSetArg("-datadir", pathTemp.string());
 
-        // Note that because we don't bother running a scheduler thread here,
-        // callbacks via CValidationInterface are unreliable, but that's OK,
-        // our unit tests aren't testing multiple parts of the code at once.
+        // We have to run a scheduler thread to prevent ActivateBestChain
+        // from blocking due to queue overrun.
+        threadGroup.create_thread(boost::bind(&CScheduler::serviceQueue, &scheduler));
         GetMainSignals().RegisterBackgroundSignalScheduler(scheduler);
         mempool.setSanityCheck(1.0);
         g_connman = std::unique_ptr<CConnman>(new CConnman(0x1337, 0x1337)); // Deterministic randomness for tests.

src/test/txvalidationcache_tests.cpp

@@ -206,8 +206,9 @@ BOOST_FIXTURE_TEST_CASE(checkinputs_test, TestChain100Setup)
     // under other (eg consensus) flags.
     // spend_tx is invalid according to DERSIG
     {
-        CValidationState state;
         LOCK(cs_main);
+
+        CValidationState state;
         PrecomputedTransactionData ptd_spend_tx(spend_tx);
 
         BOOST_CHECK(!CheckInputs(spend_tx, state, pcoinsTip.get(), true, SCRIPT_VERIFY_P2SH | SCRIPT_VERIFY_DERSIG, true, true, ptd_spend_tx, nullptr));

src/validation.cpp

@@ -54,6 +54,7 @@
 #include <llmq/quorums_chainlocks.h>
 
 #include <atomic>
+#include <future>
 #include <sstream>
 
 #include <boost/algorithm/string/replace.hpp>
@@ -2849,6 +2850,14 @@ bool ActivateBestChain(CValidationState &state, const CChainParams& chainparams,
     do {
         boost::this_thread::interruption_point();
 
+        if (GetMainSignals().CallbacksPending() > 10) {
+            // Block until the validation queue drains. This should largely
+            // never happen in normal operation, however may happen during
+            // reindex, causing memory blowup  if we run too far ahead.
+            SyncWithValidationInterfaceQueue();
+        }
+
+
         const CBlockIndex *pindexFork;
         bool fInitialDownload;
         {

src/validationinterface.cpp

@@ -11,9 +11,11 @@
 #include <sync.h>
 #include <txmempool.h>
 #include <util.h>
+#include <validation.h>
 
 #include <list>
 #include <atomic>
+#include <future>
 
 #include <boost/signals2/signal.hpp>
 
@@ -62,6 +64,11 @@ void CMainSignals::FlushBackgroundCallbacks() {
     }
 }
 
+size_t CMainSignals::CallbacksPending() {
+    if (!m_internals) return 0;
+    return m_internals->m_schedulerClient.CallbacksPending();
+}
+
 void CMainSignals::RegisterWithMempoolSignals(CTxMemPool& pool) {
     pool.NotifyEntryRemoved.connect(boost::bind(&CMainSignals::MempoolEntryRemoved, this, _1, _2));
 }
@@ -148,6 +155,16 @@ void CallFunctionInValidationInterfaceQueue(std::function<void ()> func) {
     g_signals.m_internals->m_schedulerClient.AddToProcessQueue(std::move(func));
 }
 
+void SyncWithValidationInterfaceQueue() {
+    AssertLockNotHeld(cs_main);
+    // Block until the validation queue drains
+    std::promise<void> promise;
+    CallFunctionInValidationInterfaceQueue([&promise] {
+        promise.set_value();
+    });
+    promise.get_future().wait();
+}
+
 void CMainSignals::MempoolEntryRemoved(CTransactionRef ptx, MemPoolRemovalReason reason) {
     if (reason != MemPoolRemovalReason::BLOCK && reason != MemPoolRemovalReason::CONFLICT) {
         m_internals->m_schedulerClient.AddToProcessQueue([ptx, this] {

src/validationinterface.h

@@ -50,6 +50,16 @@ void UnregisterAllValidationInterfaces();
  * will result in a deadlock (that DEBUG_LOCKORDER will miss).
  */
 void CallFunctionInValidationInterfaceQueue(std::function<void ()> func);
+/**
+ * This is a synonym for the following, which asserts certain locks are not
+ * held:
+ *     std::promise<void> promise;
+ *     CallFunctionInValidationInterfaceQueue([&promise] {
+ *         promise.set_value();
+ *     });
+ *     promise.get_future().wait();
+ */
+void SyncWithValidationInterfaceQueue();
 
 class CValidationInterface {
 protected:
@@ -151,6 +161,8 @@ class CMainSignals {
     /** Call any remaining callbacks on the calling thread */
     void FlushBackgroundCallbacks();
 
+    size_t CallbacksPending();
+
     /** Register with mempool to call TransactionRemovedFromMempool callbacks */
     void RegisterWithMempoolSignals(CTxMemPool& pool);
     /** Unregister with mempool */

src/wallet/wallet.cpp

@@ -1485,12 +1485,7 @@ void CWallet::BlockUntilSyncedToCurrentChain() {
     // ...otherwise put a callback in the validation interface queue and wait
     // for the queue to drain enough to execute it (indicating we are caught up
     // at least with the time we entered this function).
-
-    std::promise<void> promise;
-    CallFunctionInValidationInterfaceQueue([&promise] {
-        promise.set_value();
-    });
-    promise.get_future().wait();
+    SyncWithValidationInterfaceQueue();
 }