[Validation][Test] Fix concurrency issues in ActivateBestChain + optimization

src/Makefile.test.include

@@ -101,7 +101,8 @@ BITCOIN_TESTS =\
   test/univalue_tests.cpp \
   test/util_tests.cpp \
   test/sha256compress_tests.cpp \
-  test/upgrades_tests.cpp
+  test/upgrades_tests.cpp \
+  test/validation_block_tests.cpp
 
 SAPLING_TESTS =\
     test/librust/libsapling_utils_tests.cpp \

src/blockassembler.h

@@ -66,8 +66,8 @@ class BlockAssembler
     BlockAssembler(const CChainParams& chainparams, const bool defaultPrintPriority);
     /** Construct a new block template with coinbase to scriptPubKeyIn */
     std::unique_ptr<CBlockTemplate> CreateNewBlock(const CScript& scriptPubKeyIn,
-                                   CWallet* pwallet,
-                                   bool fProofOfStake,
+                                   CWallet* pwallet = nullptr,
+                                   bool fProofOfStake = false,
                                    std::vector<CStakeableOutput>* availableCoins = nullptr);
 
 private:

src/scheduler.h

@@ -87,9 +87,13 @@ class CScheduler
 
 /**
  * Class used by CScheduler clients which may schedule multiple jobs
- * which are required to be run serially. Does not require such jobs
- * to be executed on the same thread, but no two jobs will be executed
- * at the same time.
+ * which are required to be run serially. Jobs may not be run on the
+ * same thread, but no two jobs will be executed
+ * at the same time and memory will be release-acquire consistent
+ * (the scheduler will internally do an acquire before invoking a callback
+ * as well as a release at the end). In practice this means that a callback
+ * B() will be able to observe all of the effects of callback A() which executed
+ * before it.
  */
 class SingleThreadedSchedulerClient {
 private:
@@ -104,6 +108,13 @@ class SingleThreadedSchedulerClient {
 
 public:
     explicit SingleThreadedSchedulerClient(CScheduler *pschedulerIn) : m_pscheduler(pschedulerIn) {}
+
+    /**
+     * Add a callback to be executed. Callbacks are executed serially
+     * and memory is release-acquire consistent between callback executions.
+     * Practially, this means that callbacks can behave as if they are executed
+     * in order by a single thread.
+     */
     void AddToProcessQueue(std::function<void (void)> func);
 
     // Processes all remaining queue members on the calling thread, blocking until queue is empty

src/test/CMakeLists.txt

@@ -133,6 +133,7 @@ set(BITCOIN_TESTS
         ${CMAKE_CURRENT_SOURCE_DIR}/validation_tests.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/sha256compress_tests.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/upgrades_tests.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/validation_block_tests.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/librust/sapling_rpc_wallet_tests.cpp
         ${CMAKE_SOURCE_DIR}/src/wallet/test/wallet_tests.cpp
         ${CMAKE_SOURCE_DIR}/src/wallet/test/crypto_tests.cpp

src/test/scheduler_tests.cpp

@@ -61,7 +61,7 @@ BOOST_AUTO_TEST_CASE(manythreads)
     size_t nTasks = microTasks.getQueueInfo(first, last);
     BOOST_CHECK(nTasks == 0);
 
-    for (int i = 0; i < 100; i++) {
+    for (int i = 0; i < 100; ++i) {
         boost::chrono::system_clock::time_point t = now + boost::chrono::microseconds(randomMsec(rng));
         boost::chrono::system_clock::time_point tReschedule = now + boost::chrono::microseconds(500 + randomMsec(rng));
         int whichCounter = zeroToNine(rng);
@@ -108,4 +108,46 @@ BOOST_AUTO_TEST_CASE(manythreads)
     BOOST_CHECK_EQUAL(counterSum, 200);
 }
 
+BOOST_AUTO_TEST_CASE(singlethreadedscheduler_ordered)
+{
+    CScheduler scheduler;
+
+    // each queue should be well ordered with respect to itself but not other queues
+    SingleThreadedSchedulerClient queue1(&scheduler);
+    SingleThreadedSchedulerClient queue2(&scheduler);
+
+    // create more threads than queues
+    // if the queues only permit execution of one task at once then
+    // the extra threads should effectively be doing nothing
+    // if they don't we'll get out of order behaviour
+    boost::thread_group threads;
+    for (int i = 0; i < 5; ++i) {
+        threads.create_thread(boost::bind(&CScheduler::serviceQueue, &scheduler));
+    }
+
+    // these are not atomic, if SinglethreadedSchedulerClient prevents
+    // parallel execution at the queue level no synchronization should be required here
+    int counter1 = 0;
+    int counter2 = 0;
+
+    // just simply count up on each queue - if execution is properly ordered then
+    // the callbacks should run in exactly the order in which they were enqueued
+    for (int i = 0; i < 100; ++i) {
+        queue1.AddToProcessQueue([i, &counter1]() {
+            BOOST_CHECK_EQUAL(i, counter1++);
+        });
+
+        queue2.AddToProcessQueue([i, &counter2]() {
+            BOOST_CHECK_EQUAL(i, counter2++);
+        });
+    }
+
+    // finish up
+    scheduler.stop(true);
+    threads.join_all();
+
+    BOOST_CHECK_EQUAL(counter1, 100);
+    BOOST_CHECK_EQUAL(counter2, 100);
+}
+
 BOOST_AUTO_TEST_SUITE_END()

src/test/test_pivx.cpp

@@ -32,6 +32,12 @@ FastRandomContext insecure_rand_ctx(insecure_rand_seed);
 extern bool fPrintToConsole;
 extern void noui_connect();
 
+std::ostream& operator<<(std::ostream& os, const uint256& num)
+{
+    os << num.ToString();
+    return os;
+}
+
 BasicTestingSetup::BasicTestingSetup()
 {
         RandomInit();

src/test/test_pivx.h

@@ -117,4 +117,7 @@ struct TestMemPoolEntryHelper
     TestMemPoolEntryHelper &SigOps(unsigned int _sigops) { sigOpCount = _sigops; return *this; }
 };
 
+// define an implicit conversion here so that uint256 may be used directly in BOOST_CHECK_*
+std::ostream& operator<<(std::ostream& os, const uint256& num);
+
 #endif

src/test/validation_block_tests.cpp

@@ -0,0 +1,178 @@
+// Copyright (c) 2018 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or https://www.opensource.org/licenses/mit-license.php.
+
+#include <boost/test/unit_test.hpp>
+
+#include "blockassembler.h"
+#include "chainparams.h"
+#include "consensus/merkle.h"
+#include "consensus/validation.h"
+#include "pow.h"
+#include "random.h"
+#include "test/test_pivx.h"
+#include "validation.h"
+#include "validationinterface.h"
+
+struct RegtestingSetup : public TestingSetup {
+    RegtestingSetup() : TestingSetup() {
+        SelectParams(CBaseChainParams::REGTEST);
+    }
+};
+
+BOOST_FIXTURE_TEST_SUITE(validation_block_tests, RegtestingSetup)
+
+struct TestSubscriber : public CValidationInterface {
+    uint256 m_expected_tip;
+
+    TestSubscriber(uint256 tip) : m_expected_tip(std::move(tip)) {}
+
+    void UpdatedBlockTip(const CBlockIndex* pindexNew, const CBlockIndex* pindexFork, bool fInitialDownload)
+    {
+        BOOST_CHECK_EQUAL(m_expected_tip, pindexNew->GetBlockHash());
+    }
+
+    void BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex, const std::vector<CTransactionRef>& txnConflicted)
+    {
+        BOOST_CHECK_EQUAL(m_expected_tip, block->hashPrevBlock);
+        BOOST_CHECK_EQUAL(m_expected_tip, pindex->pprev->GetBlockHash());
+
+        m_expected_tip = block->GetHash();
+    }
+
+    void BlockDisconnected(const std::shared_ptr<const CBlock> &block, const uint256& blockHash, int nBlockHeight, int64_t blockTime)
+    {
+        BOOST_CHECK_EQUAL(m_expected_tip, block->GetHash());
+
+        m_expected_tip = block->hashPrevBlock;
+    }
+};
+
+std::shared_ptr<CBlock> Block(const uint256& prev_hash)
+{
+    static int i = 0;
+    static uint64_t time = Params().GenesisBlock().nTime;
+
+    CScript pubKey;
+    pubKey << i++ << OP_TRUE;
+
+    auto ptemplate = BlockAssembler(Params(), false).CreateNewBlock(pubKey);
+    auto pblock = std::make_shared<CBlock>(ptemplate->block);
+    pblock->hashPrevBlock = prev_hash;
+    pblock->nTime = ++time;
+
+    CMutableTransaction txCoinbase(*pblock->vtx[0]);
+    txCoinbase.vout.resize(1);
+    pblock->vtx[0] = MakeTransactionRef(std::move(txCoinbase));
+
+    return pblock;
+}
+
+std::shared_ptr<CBlock> FinalizeBlock(std::shared_ptr<CBlock> pblock)
+{
+    pblock->hashMerkleRoot = BlockMerkleRoot(*pblock);
+
+    while (!CheckProofOfWork(pblock->GetHash(), pblock->nBits)) {
+        ++(pblock->nNonce);
+    }
+
+    return pblock;
+}
+
+// construct a valid block
+const std::shared_ptr<const CBlock> GoodBlock(const uint256& prev_hash)
+{
+    return FinalizeBlock(Block(prev_hash));
+}
+
+// construct an invalid block (but with a valid header)
+const std::shared_ptr<const CBlock> BadBlock(const uint256& prev_hash)
+{
+    auto pblock = Block(prev_hash);
+
+    CMutableTransaction coinbase_spend;
+    coinbase_spend.vin.emplace_back(CTxIn(COutPoint(pblock->vtx[0]->GetHash(), 0), CScript(), 0));
+    coinbase_spend.vout.emplace_back(pblock->vtx[0]->vout[0]);
+
+    CTransactionRef tx = MakeTransactionRef(coinbase_spend);
+    pblock->vtx.emplace_back(tx);
+
+    auto ret = FinalizeBlock(pblock);
+    return ret;
+}
+
+void BuildChain(const uint256& root, int height, const unsigned int invalid_rate, const unsigned int branch_rate, const unsigned int max_size, std::vector<std::shared_ptr<const CBlock>>& blocks)
+{
+    if (height <= 0 || blocks.size() >= max_size) return;
+
+    bool gen_invalid = GetRand(100) < invalid_rate;
+    bool gen_fork = GetRand(100) < branch_rate;
+
+    const std::shared_ptr<const CBlock> pblock = gen_invalid ? BadBlock(root) : GoodBlock(root);
+    blocks.emplace_back(pblock);
+    if (!gen_invalid) {
+        BuildChain(pblock->GetHash(), height - 1, invalid_rate, branch_rate, max_size, blocks);
+    }
+
+    if (gen_fork) {
+        blocks.emplace_back(GoodBlock(root));
+        BuildChain(blocks.back()->GetHash(), height - 1, invalid_rate, branch_rate, max_size, blocks);
+    }
+}
+
+BOOST_AUTO_TEST_CASE(processnewblock_signals_ordering)
+{
+    // build a large-ish chain that's likely to have some forks
+    std::vector<std::shared_ptr<const CBlock>> blocks;
+    while (blocks.size() < 50) {
+        blocks.clear();
+        BuildChain(Params().GenesisBlock().GetHash(), 100, 15, 10, 500, blocks);
+    }
+
+    CValidationState state;
+    // Connect the genesis block and drain any outstanding events
+    BOOST_CHECK_MESSAGE(ProcessNewBlock(state, nullptr, std::make_shared<CBlock>(Params().GenesisBlock()), nullptr), "Error: genesis not connected");
+    SyncWithValidationInterfaceQueue();
+
+    // subscribe to events (this subscriber will validate event ordering)
+    const CBlockIndex* initial_tip = WITH_LOCK(cs_main, return chainActive.Tip());
+    TestSubscriber sub(initial_tip->GetBlockHash());
+    RegisterValidationInterface(&sub);
+
+    // create a bunch of threads that repeatedly process a block generated above at random
+    // this will create parallelism and randomness inside validation - the ValidationInterface
+    // will subscribe to events generated during block validation and assert on ordering invariance
+    boost::thread_group threads;
+    for (int i = 0; i < 10; i++) {
+        threads.create_thread([&blocks]() {
+            CValidationState state;
+            for (int i = 0; i < 1000; i++) {
+                auto block = blocks[GetRand(blocks.size() - 1)];
+                ProcessNewBlock(state, nullptr, block, nullptr);
+            }
+
+            // to make sure that eventually we process the full chain - do it here
+            for (const auto& block : blocks) {
+                if (block->vtx.size() == 1) {
+                    bool processed = ProcessNewBlock(state, nullptr, block, nullptr);
+                    // Future to do: "prevblk-not-found" here is the only valid reason to not check processed flag.
+                    if (state.GetRejectReason() == "duplicate" ||
+                        state.GetRejectReason() == "prevblk-not-found" ||
+                        state.GetRejectReason() == "bad-prevblk") continue;
+                    BOOST_ASSERT_MSG(processed,  ("Error: " + state.GetRejectReason()).c_str());
+                }
+            }
+        });
+    }
+
+    threads.join_all();
+    while (GetMainSignals().CallbacksPending() > 0) {
+        MilliSleep(100);
+    }
+
+    UnregisterValidationInterface(&sub);
+
+    BOOST_CHECK_EQUAL(sub.m_expected_tip, WITH_LOCK(cs_main, return chainActive.Tip()->GetBlockHash()));
+}
+
+BOOST_AUTO_TEST_SUITE_END()