[WIP] rebase: Call ProcessNewBlock() asynchronously

src/init.cpp

@@ -1764,6 +1764,8 @@ bool AppInitMain(NodeContext& node)
         return false;
     }
 
+    threadGroup.create_thread(std::bind(&TraceThread<std::function<void()>>, "blockconn", std::function<void()>(std::bind(&CChainState::ProcessBlockValidationQueue, &ChainstateActive()))));
+
     fs::path est_path = GetDataDir() / FEE_ESTIMATES_FILENAME;
     CAutoFile est_filein(fsbridge::fopen(est_path, "rb"), SER_DISK, CLIENT_VERSION);
     // Allowed to fail as this file IS missing on first startup.

src/net_processing.h

@@ -51,6 +51,10 @@ class PeerLogicValidation final : public CValidationInterface, public NetEventsI
      * Overridden from CValidationInterface.
      */
     void NewPoWValidBlock(const CBlockIndex *pindex, const std::shared_ptr<const CBlock>& pblock) override;
+    /**
+     * Overridden from CValidationInterface.
+     */
+    void BlockProcessed() override;
 
     /** Initialize a peer by adding it to mapNodeState and pushing a message requesting its version */
     void InitializeNode(CNode* pnode) override;

src/rpc/mining.cpp

@@ -124,7 +124,9 @@ static bool GenerateBlock(CBlock& block, uint64_t& max_tries, unsigned int& extr
     }
 
     std::shared_ptr<const CBlock> shared_pblock = std::make_shared<const CBlock>(block);
-    if (!ProcessNewBlock(chainparams, shared_pblock, true, nullptr))
+    BlockValidationState state;
+    ProcessNewBlock(Params(), shared_pblock, state, true).wait();
+    if (ChainActive().Tip()->GetBlockHash() != block.GetHash())
         throw JSONRPCError(RPC_INTERNAL_ERROR, "ProcessNewBlock, block not accepted");
 
     block_hash = block.GetHash();
@@ -941,14 +943,18 @@ static UniValue submitblock(const JSONRPCRequest& request)
         }
     }
 
-    bool new_block;
     auto sc = std::make_shared<submitblock_StateCatcher>(block.GetHash());
     RegisterSharedValidationInterface(sc);
-    bool accepted = ProcessNewBlock(Params(), blockptr, /* fForceProcessing */ true, /* fNewBlock */ &new_block);
+    BlockValidationState dos_state;
+    bool new_block = ProcessNewBlock(Params(), blockptr, dos_state, /* fForceProcessing */ true).get();
+    SyncWithValidationInterfaceQueue();
     UnregisterSharedValidationInterface(sc);
-    if (!new_block && accepted) {
+    if (!new_block && dos_state.IsValid()) {
         return "duplicate";
     }
+    if (!dos_state.IsValid()) {
+        return BIP22ValidationResult(dos_state);
+    }
     if (!sc->found) {
         return "inconclusive";
     }

src/test/blockfilter_index_tests.cpp

@@ -171,7 +171,9 @@ BOOST_FIXTURE_TEST_CASE(blockfilter_index_initial_sync, BuildChainTestingSetup)
     uint256 chainA_last_header = last_header;
     for (size_t i = 0; i < 2; i++) {
         const auto& block = chainA[i];
-        BOOST_REQUIRE(ProcessNewBlock(Params(), block, true, nullptr));
+        BlockValidationState dos_state;
+        ProcessNewBlock(Params(), block, dos_state, true).wait();
+        BOOST_REQUIRE(dos_state.IsValid());
     }
     for (size_t i = 0; i < 2; i++) {
         const auto& block = chainA[i];
@@ -189,7 +191,9 @@ BOOST_FIXTURE_TEST_CASE(blockfilter_index_initial_sync, BuildChainTestingSetup)
     uint256 chainB_last_header = last_header;
     for (size_t i = 0; i < 3; i++) {
         const auto& block = chainB[i];
-        BOOST_REQUIRE(ProcessNewBlock(Params(), block, true, nullptr));
+        BlockValidationState dos_state;
+        ProcessNewBlock(Params(), block, dos_state, true).wait();
+        BOOST_REQUIRE(dos_state.IsValid());
     }
     for (size_t i = 0; i < 3; i++) {
         const auto& block = chainB[i];
@@ -218,10 +222,12 @@ BOOST_FIXTURE_TEST_CASE(blockfilter_index_initial_sync, BuildChainTestingSetup)
     }
 
     // Reorg back to chain A.
-     for (size_t i = 2; i < 4; i++) {
-         const auto& block = chainA[i];
-         BOOST_REQUIRE(ProcessNewBlock(Params(), block, true, nullptr));
-     }
+    for (size_t i = 2; i < 4; i++) {
+        const auto& block = chainA[i];
+        BlockValidationState dos_state;
+        ProcessNewBlock(Params(), block, dos_state, true).wait();
+        BOOST_REQUIRE(dos_state.IsValid());
+    }
 
      // Check that chain A and B blocks can be retrieved.
      chainA_last_header = last_header;

src/test/denialofservice_tests.cpp

@@ -98,11 +98,11 @@ BOOST_AUTO_TEST_CASE(outbound_slow_chain_eviction)
 
     // Test starts here
     {
-        LOCK2(cs_main, dummyNode1.cs_sendProcessing);
+        LOCK(dummyNode1.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode1)); // should result in getheaders
     }
     {
-        LOCK2(cs_main, dummyNode1.cs_vSend);
+        LOCK(dummyNode1.cs_vSend);
         BOOST_CHECK(dummyNode1.vSendMsg.size() > 0);
         dummyNode1.vSendMsg.clear();
     }
@@ -111,17 +111,17 @@ BOOST_AUTO_TEST_CASE(outbound_slow_chain_eviction)
     // Wait 21 minutes
     SetMockTime(nStartTime+21*60);
     {
-        LOCK2(cs_main, dummyNode1.cs_sendProcessing);
+        LOCK(dummyNode1.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode1)); // should result in getheaders
     }
     {
-        LOCK2(cs_main, dummyNode1.cs_vSend);
+        LOCK(dummyNode1.cs_vSend);
         BOOST_CHECK(dummyNode1.vSendMsg.size() > 0);
     }
     // Wait 3 more minutes
     SetMockTime(nStartTime+24*60);
     {
-        LOCK2(cs_main, dummyNode1.cs_sendProcessing);
+        LOCK(dummyNode1.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode1)); // should result in disconnect
     }
     BOOST_CHECK(dummyNode1.fDisconnect == true);
@@ -235,7 +235,7 @@ BOOST_AUTO_TEST_CASE(DoS_banning)
         Misbehaving(dummyNode1.GetId(), 100); // Should get banned
     }
     {
-        LOCK2(cs_main, dummyNode1.cs_sendProcessing);
+        LOCK(dummyNode1.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode1));
     }
     BOOST_CHECK(banman->IsBanned(addr1));
@@ -252,7 +252,7 @@ BOOST_AUTO_TEST_CASE(DoS_banning)
         Misbehaving(dummyNode2.GetId(), 50);
     }
     {
-        LOCK2(cs_main, dummyNode2.cs_sendProcessing);
+        LOCK(dummyNode2.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode2));
     }
     BOOST_CHECK(!banman->IsBanned(addr2)); // 2 not banned yet...
@@ -262,7 +262,7 @@ BOOST_AUTO_TEST_CASE(DoS_banning)
         Misbehaving(dummyNode2.GetId(), 50);
     }
     {
-        LOCK2(cs_main, dummyNode2.cs_sendProcessing);
+        LOCK(dummyNode2.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode2));
     }
     BOOST_CHECK(banman->IsBanned(addr2));
@@ -291,7 +291,7 @@ BOOST_AUTO_TEST_CASE(DoS_banscore)
         Misbehaving(dummyNode1.GetId(), 100);
     }
     {
-        LOCK2(cs_main, dummyNode1.cs_sendProcessing);
+        LOCK(dummyNode1.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode1));
     }
     BOOST_CHECK(!banman->IsBanned(addr1));
@@ -300,7 +300,7 @@ BOOST_AUTO_TEST_CASE(DoS_banscore)
         Misbehaving(dummyNode1.GetId(), 10);
     }
     {
-        LOCK2(cs_main, dummyNode1.cs_sendProcessing);
+        LOCK(dummyNode1.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode1));
     }
     BOOST_CHECK(!banman->IsBanned(addr1));
@@ -309,7 +309,7 @@ BOOST_AUTO_TEST_CASE(DoS_banscore)
         Misbehaving(dummyNode1.GetId(), 1);
     }
     {
-        LOCK2(cs_main, dummyNode1.cs_sendProcessing);
+        LOCK(dummyNode1.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode1));
     }
     BOOST_CHECK(banman->IsBanned(addr1));
@@ -341,7 +341,7 @@ BOOST_AUTO_TEST_CASE(DoS_bantime)
         Misbehaving(dummyNode.GetId(), 100);
     }
     {
-        LOCK2(cs_main, dummyNode.cs_sendProcessing);
+        LOCK(dummyNode.cs_sendProcessing);
         BOOST_CHECK(peerLogic->SendMessages(&dummyNode));
     }
     BOOST_CHECK(banman->IsBanned(addr));

src/test/miner_tests.cpp

@@ -7,6 +7,7 @@
 #include <consensus/consensus.h>
 #include <consensus/merkle.h>
 #include <consensus/tx_verify.h>
+#include <consensus/validation.h>
 #include <miner.h>
 #include <policy/policy.h>
 #include <script/standard.h>
@@ -253,7 +254,9 @@ BOOST_AUTO_TEST_CASE(CreateNewBlock_validity)
             pblock->nNonce = blockinfo[i].nonce;
         }
         std::shared_ptr<const CBlock> shared_pblock = std::make_shared<const CBlock>(*pblock);
-        BOOST_CHECK(ProcessNewBlock(chainparams, shared_pblock, true, nullptr));
+        BlockValidationState dos_state;
+        ProcessNewBlock(chainparams, shared_pblock, dos_state, true).wait();
+        BOOST_CHECK(dos_state.IsValid());
         pblock->hashPrevBlock = pblock->GetHash();
     }
 

src/test/util/mining.cpp

@@ -6,6 +6,7 @@
 
 #include <chainparams.h>
 #include <consensus/merkle.h>
+#include <consensus/validation.h>
 #include <key_io.h>
 #include <miner.h>
 #include <node/context.h>
@@ -31,8 +32,9 @@ CTxIn MineBlock(const NodeContext& node, const CScript& coinbase_scriptPubKey)
         assert(block->nNonce);
     }
 
-    bool processed{ProcessNewBlock(Params(), block, true, nullptr)};
-    assert(processed);
+    BlockValidationState dos_state;
+    ProcessNewBlock(Params(), block, dos_state, true).wait();
+    assert(dos_state.IsValid());
 
     return CTxIn{block->vtx[0]->GetHash(), 0};
 }

src/test/util/setup_common.cpp

@@ -159,6 +159,9 @@ TestingSetup::TestingSetup(const std::string& chainName, const std::vector<const
 
     m_node.mempool = &::mempool;
     m_node.mempool->setSanityCheck(1.0);
+
+    threadGroup.create_thread(std::function<void()>(std::bind(&CChainState::ProcessBlockValidationQueue, &ChainstateActive())));
+
     m_node.banman = MakeUnique<BanMan>(GetDataDir() / "banlist.dat", nullptr, DEFAULT_MISBEHAVING_BANTIME);
     m_node.connman = MakeUnique<CConnman>(0x1337, 0x1337); // Deterministic randomness for tests.
     m_node.peer_logic = MakeUnique<PeerLogicValidation>(m_node.connman.get(), m_node.banman.get(), *m_node.scheduler, *m_node.mempool);
@@ -171,6 +174,10 @@ TestingSetup::TestingSetup(const std::string& chainName, const std::vector<const
 
 TestingSetup::~TestingSetup()
 {
+    // If eg the block connection thread is waiting on the queue to drain,
+    // killing the scheduler thread now will hang us, so wait on the queue
+    // to drain first.
+    ChainstateActive().AwaitBlockValidationParked();
     if (m_node.scheduler) m_node.scheduler->stop();
     threadGroup.interrupt_all();
     threadGroup.join_all();
@@ -228,7 +235,8 @@ CBlock TestChain100Setup::CreateAndProcessBlock(const std::vector<CMutableTransa
     while (!CheckProofOfWork(block.GetHash(), block.nBits, chainparams.GetConsensus())) ++block.nNonce;
 
     std::shared_ptr<const CBlock> shared_pblock = std::make_shared<const CBlock>(block);
-    ProcessNewBlock(chainparams, shared_pblock, true, nullptr);
+    BlockValidationState dos_state;
+    ProcessNewBlock(chainparams, shared_pblock, dos_state, true).wait();
 
     CBlock result = block;
     return result;

src/test/validation_block_tests.cpp

@@ -157,7 +157,6 @@ BOOST_AUTO_TEST_CASE(processnewblock_signals_ordering)
         BuildChain(Params().GenesisBlock().GetHash(), 100, 15, 10, 500, blocks);
     }
 
-    bool ignored;
     BlockValidationState state;
     std::vector<CBlockHeader> headers;
     std::transform(blocks.begin(), blocks.end(), std::back_inserter(headers), [](std::shared_ptr<const CBlock> b) { return b->GetBlockHeader(); });
@@ -166,7 +165,9 @@ BOOST_AUTO_TEST_CASE(processnewblock_signals_ordering)
     BOOST_CHECK(ProcessNewBlockHeaders(headers, state, Params()));
 
     // Connect the genesis block and drain any outstanding events
-    BOOST_CHECK(ProcessNewBlock(Params(), std::make_shared<CBlock>(Params().GenesisBlock()), true, &ignored));
+    BlockValidationState dos_state;
+    ProcessNewBlock(Params(), std::make_shared<CBlock>(Params().GenesisBlock()), dos_state, true).wait();
+    BOOST_CHECK(dos_state.IsValid());
     SyncWithValidationInterfaceQueue();
 
     // subscribe to events (this subscriber will validate event ordering)
@@ -184,20 +185,25 @@ BOOST_AUTO_TEST_CASE(processnewblock_signals_ordering)
     std::vector<std::thread> threads;
     for (int i = 0; i < 10; i++) {
         threads.emplace_back([&blocks]() {
-            bool ignored;
+            std::vector<std::future<bool>> thread_futures;
             FastRandomContext insecure;
             for (int i = 0; i < 1000; i++) {
                 auto block = blocks[insecure.randrange(blocks.size() - 1)];
-                ProcessNewBlock(Params(), block, true, &ignored);
+                BlockValidationState dos_state;
+                thread_futures.push_back(ProcessNewBlock(Params(), block, dos_state, true));
             }
 
             // to make sure that eventually we process the full chain - do it here
             for (auto block : blocks) {
                 if (block->vtx.size() == 1) {
-                    bool processed = ProcessNewBlock(Params(), block, true, &ignored);
-                    assert(processed);
+                    BlockValidationState dos_state;
+                    thread_futures.push_back(ProcessNewBlock(Params(), block, dos_state, true));
+                    assert(dos_state.IsValid());
                 }
             }
+            for (std::future<bool>& future: thread_futures) {
+                future.wait();
+            }
         });
     }
 
@@ -233,7 +239,9 @@ BOOST_AUTO_TEST_CASE(mempool_locks_reorg)
 {
     bool ignored;
     auto ProcessBlock = [&ignored](std::shared_ptr<const CBlock> block) -> bool {
-        return ProcessNewBlock(Params(), block, /* fForceProcessing */ true, /* fNewBlock */ &ignored);
+        BlockValidationState dos_state;
+        ProcessNewBlock(Params(), block, dos_state, /* fForceProcessing */ true).wait();
+        return dos_state.IsValid();
     };
 
     // Process all mined blocks

src/test/validationinterface_tests.cpp

@@ -22,7 +22,7 @@ BOOST_AUTO_TEST_CASE(unregister_validation_interface_race)
 
     // Start thread to generate notifications
     std::thread gen{[&] {
-        const CBlock block_dummy;
+        const std::shared_ptr<const CBlock> block_dummy = std::make_shared<const CBlock>();
         const BlockValidationState state_dummy;
         while (generate) {
             GetMainSignals().BlockChecked(block_dummy, state_dummy);
@@ -57,15 +57,15 @@ class TestInterface : public CValidationInterface
     {
         if (m_on_destroy) m_on_destroy();
     }
-    void BlockChecked(const CBlock& block, const BlockValidationState& state) override
+    void NewPoWValidBlock(const CBlockIndex *pindex, const std::shared_ptr<const CBlock> &block) override
     {
         if (m_on_call) m_on_call();
     }
     static void Call()
     {
-        CBlock block;
+        std::shared_ptr<const CBlock> block = std::make_shared<CBlock>();
         BlockValidationState state;
-        GetMainSignals().BlockChecked(block, state);
+        GetMainSignals().NewPoWValidBlock(nullptr, std::make_shared<CBlock>());
     }
     std::function<void()> m_on_call;
     std::function<void()> m_on_destroy;