Merge bitcoin#11824: Block ActivateBestChain to empty validationinter…

…face queue

97d2b09 Add helper to wait for validation interface queue to catch up (Matt Corallo)
3613749 Block ActivateBestChain to empty validationinterface queue (Matt Corallo)
5a933ce Add an interface to get the queue depth out of CValidationInterface (Matt Corallo)
a99b76f Require no cs_main lock for ProcessNewBlock/ActivateBestChain (Matt Corallo)
a734896 Avoid cs_main in net_processing ActivateBestChain calls (Matt Corallo)
66aa1d5 Refactor ProcessGetData in anticipation of avoiding cs_main for ABC (Matt Corallo)
818075a Create new mutex for orphans, no cs_main in PLV::BlockConnected (Matt Corallo)

Pull request description:

  This should fix bitcoin#11822.

  It ended up bigger than I hoped for, but its not too gnarly. Note that "
  Require no cs_main lock for ProcessNewBlock/ActivateBestChain" is mostly pure code-movement.

Tree-SHA512: 1127688545926f6099449dca6a4e6609eefc3abbd72f1c66e03d32bd8c7b31e82097d8307822cfd1dec0321703579cfdd82069cab6e17b1024e75eac694122cb

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

@@ -66,38 +66,26 @@ BOOST_FIXTURE_TEST_CASE(tx_mempool_block_doublespend, TestChain100Setup)
 
     // Test 1: block with both of those transactions should be rejected.
     block = CreateAndProcessBlock(spends, scriptPubKey);
-    {
-        LOCK(cs_main);
-        BOOST_CHECK(chainActive.Tip()->GetBlockHash() != block.GetHash());
-    }
+    BOOST_CHECK(chainActive.Tip()->GetBlockHash() != block.GetHash());
 
     // Test 2: ... and should be rejected if spend1 is in the memory pool
     BOOST_CHECK(ToMemPool(spends[0]));
     block = CreateAndProcessBlock(spends, scriptPubKey);
-    {
-        LOCK(cs_main);
-        BOOST_CHECK(chainActive.Tip()->GetBlockHash() != block.GetHash());
-    }
+    BOOST_CHECK(chainActive.Tip()->GetBlockHash() != block.GetHash());
     mempool.clear();
 
     // Test 3: ... and should be rejected if spend2 is in the memory pool
     BOOST_CHECK(ToMemPool(spends[1]));
     block = CreateAndProcessBlock(spends, scriptPubKey);
-    {
-        LOCK(cs_main);
-        BOOST_CHECK(chainActive.Tip()->GetBlockHash() != block.GetHash());
-    }
+    BOOST_CHECK(chainActive.Tip()->GetBlockHash() != block.GetHash());
     mempool.clear();
 
     // Final sanity test: first spend in mempool, second in block, that's OK:
     std::vector<CMutableTransaction> oneSpend;
     oneSpend.push_back(spends[0]);
     BOOST_CHECK(ToMemPool(spends[1]));
     block = CreateAndProcessBlock(oneSpend, scriptPubKey);
-    {
-        LOCK(cs_main);
-        BOOST_CHECK(chainActive.Tip()->GetBlockHash() == block.GetHash());
-    }
+    BOOST_CHECK(chainActive.Tip()->GetBlockHash() == block.GetHash());
     // spends[1] should have been removed from the mempool when the
     // block with spends[0] is accepted:
     BOOST_CHECK_EQUAL(mempool.size(), 0);
@@ -206,8 +194,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));
@@ -231,10 +220,11 @@ BOOST_FIXTURE_TEST_CASE(checkinputs_test, TestChain100Setup)
     CBlock block;
 
     block = CreateAndProcessBlock({spend_tx}, p2pk_scriptPubKey);
-    LOCK(cs_main);
     BOOST_CHECK(chainActive.Tip()->GetBlockHash() == block.GetHash());
     BOOST_CHECK(pcoinsTip->GetBestBlock() == block.GetHash());
 
+    LOCK(cs_main);
+
     // Test P2SH: construct a transaction that is valid without P2SH, and
     // then test validity with P2SH.
     {

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,13 @@ 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();
 }