Merge #18037: Util: Allow scheduler to be mocked

8bca30ea17 [rpc] expose ability to mock scheduler via the rpc (Amiti Uttarwar)
7c8b6e5b52 [lib] add scheduler to node context (Amiti Uttarwar)
930d837542 [test] add chainparams property to indicate chain allows time mocking (Amiti Uttarwar)
1cd43e83c6 [test] unit test for new MockForward scheduler method (Amiti Uttarwar)
a6f63598ad [util] allow scheduler to be mocked (Amiti Uttarwar)

Pull request description:

  This PR is to support functional tests by allowing the scheduler to be mocked via the RPC.

  It adds a `MockForward` method to the scheduler class that iterates through the task queue and reschedules them to be `delta_seconds` sooner.

  This is currently used to support functional testing of the "unbroadcast" set tracking in #18038. If this patch is accepted, it would also be useful to simplify the code in #16698.

ACKs for top commit:
  MarcoFalke:
    ACK 8bca30ea17, only change is some style fixups 🕓

Tree-SHA512: 2a97fe8ade2b7fd1fb5cdfa1dcafb3227a377d7a847e3845a228bc119eb77824b4aefa43d922a06d583939b22725e223f308cf092961048079d36f6b1d9a639b

src/chainparams.cpp

@@ -135,6 +135,7 @@ public:
         fDefaultConsistencyChecks = false;
         fRequireStandard = true;
         m_is_test_chain = false;
+        m_is_mockable_chain = false;
 
         checkpointData = {
             {
@@ -231,7 +232,7 @@ public:
         fDefaultConsistencyChecks = false;
         fRequireStandard = false;
         m_is_test_chain = true;
-
+        m_is_mockable_chain = false;
 
         checkpointData = {
             {
@@ -303,6 +304,7 @@ public:
         fDefaultConsistencyChecks = true;
         fRequireStandard = true;
         m_is_test_chain = true;
+        m_is_mockable_chain = true;
 
         checkpointData = {
             {

src/chainparams.h

@@ -68,6 +68,8 @@ public:
     bool RequireStandard() const { return fRequireStandard; }
     /** If this chain is exclusively used for testing */
     bool IsTestChain() const { return m_is_test_chain; }
+    /** If this chain allows time to be mocked */
+    bool IsMockableChain() const { return m_is_mockable_chain; }
     uint64_t PruneAfterHeight() const { return nPruneAfterHeight; }
     /** Minimum free space (in GB) needed for data directory */
     uint64_t AssumedBlockchainSize() const { return m_assumed_blockchain_size; }
@@ -102,6 +104,7 @@ protected:
     bool fDefaultConsistencyChecks;
     bool fRequireStandard;
     bool m_is_test_chain;
+    bool m_is_mockable_chain;
     CCheckpointData checkpointData;
     ChainTxData chainTxData;
 };

src/init.cpp

@@ -157,7 +157,6 @@ NODISCARD static bool CreatePidFile()
 static std::unique_ptr<ECCVerifyHandle> globalVerifyHandle;
 
 static boost::thread_group threadGroup;
-static CScheduler scheduler;
 
 void Interrupt(NodeContext& node)
 {
@@ -295,6 +294,7 @@ void Shutdown(NodeContext& node)
     globalVerifyHandle.reset();
     ECC_Stop();
     if (node.mempool) node.mempool = nullptr;
+    node.scheduler.reset();
     LogPrintf("%s: done\n", __func__);
 }
 
@@ -1268,16 +1268,19 @@ bool AppInitMain(NodeContext& node)
         }
     }
 
+    assert(!node.scheduler);
+    node.scheduler = MakeUnique<CScheduler>();
+
     // Start the lightweight task scheduler thread
-    CScheduler::Function serviceLoop = std::bind(&CScheduler::serviceQueue, &scheduler);
+    CScheduler::Function serviceLoop = [&node]{ node.scheduler->serviceQueue(); };
     threadGroup.create_thread(std::bind(&TraceThread<CScheduler::Function>, "scheduler", serviceLoop));
 
     // Gather some entropy once per minute.
-    scheduler.scheduleEvery([]{
+    node.scheduler->scheduleEvery([]{
         RandAddPeriodic();
     }, 60000);
 
-    GetMainSignals().RegisterBackgroundSignalScheduler(scheduler);
+    GetMainSignals().RegisterBackgroundSignalScheduler(*node.scheduler);
 
     // Create client interfaces for wallets that are supposed to be loaded
     // according to -wallet and -disablewallet options. This only constructs
@@ -1327,7 +1330,7 @@ bool AppInitMain(NodeContext& node)
     assert(!node.connman);
     node.connman = std::unique_ptr<CConnman>(new CConnman(GetRand(std::numeric_limits<uint64_t>::max()), GetRand(std::numeric_limits<uint64_t>::max())));
 
-    node.peer_logic.reset(new PeerLogicValidation(node.connman.get(), node.banman.get(), scheduler));
+    node.peer_logic.reset(new PeerLogicValidation(node.connman.get(), node.banman.get(), *node.scheduler));
     RegisterValidationInterface(node.peer_logic.get());
 
     // sanitize comments per BIP-0014, format user agent and check total size
@@ -1819,7 +1822,7 @@ bool AppInitMain(NodeContext& node)
             connOptions.m_specified_outgoing = connect;
         }
     }
-    if (!node.connman->Start(scheduler, connOptions)) {
+    if (!node.connman->Start(*node.scheduler, connOptions)) {
         return false;
     }
 
@@ -1848,11 +1851,11 @@ bool AppInitMain(NodeContext& node)
     uiInterface.InitMessage(_("Done loading").translated);
 
     for (const auto& client : node.chain_clients) {
-        client->start(scheduler);
+        client->start(*node.scheduler);
     }
 
     BanMan* banman = node.banman.get();
-    scheduler.scheduleEvery([banman]{
+    node.scheduler->scheduleEvery([banman]{
         banman->DumpBanlist();
     }, DUMP_BANS_INTERVAL * 1000);
 

src/node/context.cpp

@@ -8,6 +8,7 @@
 #include <interfaces/chain.h>
 #include <net.h>
 #include <net_processing.h>
+#include <scheduler.h>
 
 NodeContext::NodeContext() {}
 NodeContext::~NodeContext() {}

src/node/context.h

@@ -10,6 +10,7 @@
 
 class BanMan;
 class CConnman;
+class CScheduler;
 class CTxMemPool;
 class PeerLogicValidation;
 namespace interfaces {
@@ -34,6 +35,7 @@ struct NodeContext {
     std::unique_ptr<BanMan> banman;
     std::unique_ptr<interfaces::Chain> chain;
     std::vector<std::unique_ptr<interfaces::ChainClient>> chain_clients;
+    std::unique_ptr<CScheduler> scheduler;
 
     //! Declare default constructor and destructor that are not inline, so code
     //! instantiating the NodeContext struct doesn't need to #include class

src/rpc/client.cpp

@@ -27,6 +27,7 @@ public:
 static const CRPCConvertParam vRPCConvertParams[] =
 {
     { "setmocktime", 0, "timestamp" },
+    { "mockscheduler", 0, "delta_time" },
     { "utxoupdatepsbt", 1, "descriptors" },
     { "generatetoaddress", 0, "nblocks" },
     { "generatetoaddress", 2, "maxtries" },

src/rpc/misc.cpp

@@ -5,10 +5,12 @@
 
 #include <httpserver.h>
 #include <key_io.h>
+#include <node/context.h>
 #include <outputtype.h>
 #include <rpc/blockchain.h>
 #include <rpc/server.h>
 #include <rpc/util.h>
+#include <scheduler.h>
 #include <script/descriptor.h>
 #include <util/check.h>
 #include <util/strencodings.h>
@@ -371,6 +373,36 @@ static UniValue setmocktime(const JSONRPCRequest& request)
     return NullUniValue;
 }
 
+static UniValue mockscheduler(const JSONRPCRequest& request)
+{
+    RPCHelpMan{"mockscheduler",
+        "\nBump the scheduler into the future (-regtest only)\n",
+        {
+            {"delta_time", RPCArg::Type::NUM, RPCArg::Optional::NO, "Number of seconds to forward the scheduler into the future." },
+        },
+        RPCResults{},
+        RPCExamples{""},
+    }.Check(request);
+
+    if (!Params().IsMockableChain()) {
+        throw std::runtime_error("mockscheduler is for regression testing (-regtest mode) only");
+    }
+
+    // check params are valid values
+    RPCTypeCheck(request.params, {UniValue::VNUM});
+    int64_t delta_seconds = request.params[0].get_int64();
+    if ((delta_seconds <= 0) || (delta_seconds > 3600)) {
+        throw std::runtime_error("delta_time must be between 1 and 3600 seconds (1 hr)");
+    }
+
+    // protect against null pointer dereference
+    CHECK_NONFATAL(g_rpc_node);
+    CHECK_NONFATAL(g_rpc_node->scheduler);
+    g_rpc_node->scheduler->MockForward(boost::chrono::seconds(delta_seconds));
+
+    return NullUniValue;
+}
+
 static UniValue RPCLockedMemoryInfo()
 {
     LockedPool::Stats stats = LockedPoolManager::Instance().stats();
@@ -575,6 +607,7 @@ static const CRPCCommand commands[] =
 
     /* Not shown in help */
     { "hidden",             "setmocktime",            &setmocktime,            {"timestamp"}},
+    { "hidden",             "mockscheduler",          &mockscheduler,          {"delta_time"}},
     { "hidden",             "echo",                   &echo,                   {"arg0","arg1","arg2","arg3","arg4","arg5","arg6","arg7","arg8","arg9"}},
     { "hidden",             "echojson",               &echo,                   {"arg0","arg1","arg2","arg3","arg4","arg5","arg6","arg7","arg8","arg9"}},
 };

src/scheduler.cpp

@@ -114,6 +114,28 @@ void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSecon
     schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds));
 }
 
+void CScheduler::MockForward(boost::chrono::seconds delta_seconds)
+{
+    assert(delta_seconds.count() > 0 && delta_seconds < boost::chrono::hours{1});
+
+    {
+        boost::unique_lock<boost::mutex> lock(newTaskMutex);
+
+        // use temp_queue to maintain updated schedule
+        std::multimap<boost::chrono::system_clock::time_point, Function> temp_queue;
+
+        for (const auto& element : taskQueue) {
+            temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second);
+        }
+
+        // point taskQueue to temp_queue
+        taskQueue = std::move(temp_queue);
+    }
+
+    // notify that the taskQueue needs to be processed
+    newTaskScheduled.notify_one();
+}
+
 static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)
 {
     f();

src/scheduler.h

@@ -55,6 +55,13 @@ public:
     // need more accurate scheduling, don't use this method.
     void scheduleEvery(Function f, int64_t deltaMilliSeconds);
 
+    /**
+     * Mock the scheduler to fast forward in time.
+     * Iterates through items on taskQueue and reschedules them
+     * to be delta_seconds sooner.
+     */
+    void MockForward(boost::chrono::seconds delta_seconds);
+
     // To keep things as simple as possible, there is no unschedule.
 
     // Services the queue 'forever'. Should be run in a thread,

src/test/denialofservice_tests.cpp

@@ -78,7 +78,7 @@ BOOST_FIXTURE_TEST_SUITE(denialofservice_tests, TestingSetup)
 BOOST_AUTO_TEST_CASE(outbound_slow_chain_eviction)
 {
     auto connman = MakeUnique<CConnman>(0x1337, 0x1337);
-    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), nullptr, scheduler);
+    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), nullptr, *m_node.scheduler);
 
     // Mock an outbound peer
     CAddress addr1(ip(0xa0b0c001), NODE_NONE);
@@ -148,7 +148,7 @@ static void AddRandomOutboundPeer(std::vector<CNode *> &vNodes, PeerLogicValidat
 BOOST_AUTO_TEST_CASE(stale_tip_peer_management)
 {
     auto connman = MakeUnique<CConnmanTest>(0x1337, 0x1337);
-    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), nullptr, scheduler);
+    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), nullptr, *m_node.scheduler);
 
     const Consensus::Params& consensusParams = Params().GetConsensus();
     constexpr int max_outbound_full_relay = 8;
@@ -221,7 +221,7 @@ BOOST_AUTO_TEST_CASE(DoS_banning)
 {
     auto banman = MakeUnique<BanMan>(GetDataDir() / "banlist.dat", nullptr, DEFAULT_MISBEHAVING_BANTIME);
     auto connman = MakeUnique<CConnman>(0x1337, 0x1337);
-    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), banman.get(), scheduler);
+    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), banman.get(), *m_node.scheduler);
 
     banman->ClearBanned();
     CAddress addr1(ip(0xa0b0c001), NODE_NONE);
@@ -276,7 +276,7 @@ BOOST_AUTO_TEST_CASE(DoS_banscore)
 {
     auto banman = MakeUnique<BanMan>(GetDataDir() / "banlist.dat", nullptr, DEFAULT_MISBEHAVING_BANTIME);
     auto connman = MakeUnique<CConnman>(0x1337, 0x1337);
-    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), banman.get(), scheduler);
+    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), banman.get(), *m_node.scheduler);
 
     banman->ClearBanned();
     gArgs.ForceSetArg("-banscore", "111"); // because 11 is my favorite number
@@ -323,7 +323,7 @@ BOOST_AUTO_TEST_CASE(DoS_bantime)
 {
     auto banman = MakeUnique<BanMan>(GetDataDir() / "banlist.dat", nullptr, DEFAULT_MISBEHAVING_BANTIME);
     auto connman = MakeUnique<CConnman>(0x1337, 0x1337);
-    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), banman.get(), scheduler);
+    auto peerLogic = MakeUnique<PeerLogicValidation>(connman.get(), banman.get(), *m_node.scheduler);
 
     banman->ClearBanned();
     int64_t nStartTime = GetTime();

src/test/scheduler_tests.cpp

@@ -5,8 +5,6 @@
 #include <random.h>
 #include <scheduler.h>
 
-#include <test/util/setup_common.h>
-
 #include <boost/thread.hpp>
 #include <boost/test/unit_test.hpp>
 
@@ -155,4 +153,45 @@ BOOST_AUTO_TEST_CASE(singlethreadedscheduler_ordered)
     BOOST_CHECK_EQUAL(counter2, 100);
 }
 
+BOOST_AUTO_TEST_CASE(mockforward)
+{
+    CScheduler scheduler;
+
+    int counter{0};
+    CScheduler::Function dummy = [&counter]{counter++;};
+
+    // schedule jobs for 2, 5 & 8 minutes into the future
+    int64_t min_in_milli = 60*1000;
+    scheduler.scheduleFromNow(dummy, 2*min_in_milli);
+    scheduler.scheduleFromNow(dummy, 5*min_in_milli);
+    scheduler.scheduleFromNow(dummy, 8*min_in_milli);
+
+    // check taskQueue
+    boost::chrono::system_clock::time_point first, last;
+    size_t num_tasks = scheduler.getQueueInfo(first, last);
+    BOOST_CHECK_EQUAL(num_tasks, 3ul);
+
+    std::thread scheduler_thread([&]() { scheduler.serviceQueue(); });
+
+    // bump the scheduler forward 5 minutes
+    scheduler.MockForward(boost::chrono::seconds(5*60));
+
+    // ensure scheduler has chance to process all tasks queued for before 1 ms from now.
+    scheduler.scheduleFromNow([&scheduler]{ scheduler.stop(false); }, 1);
+    scheduler_thread.join();
+
+    // check that the queue only has one job remaining
+    num_tasks = scheduler.getQueueInfo(first, last);
+    BOOST_CHECK_EQUAL(num_tasks, 1ul);
+
+    // check that the dummy function actually ran
+    BOOST_CHECK_EQUAL(counter, 2);
+
+    // check that the time of the remaining job has been updated
+    boost::chrono::system_clock::time_point now = boost::chrono::system_clock::now();
+    int delta = boost::chrono::duration_cast<boost::chrono::seconds>(first - now).count();
+    // should be between 2 & 3 minutes from now
+    BOOST_CHECK(delta > 2*60 && delta < 3*60);
+}
+
 BOOST_AUTO_TEST_SUITE_END()

src/test/util/setup_common.cpp

@@ -103,10 +103,12 @@ TestingSetup::TestingSetup(const std::string& chainName) : BasicTestingSetup(cha
     g_rpc_node = &m_node;
     RegisterAllCoreRPCCommands(tableRPC);
 
+    m_node.scheduler = MakeUnique<CScheduler>();
+
     // We have to run a scheduler thread to prevent ActivateBestChain
     // from blocking due to queue overrun.
-    threadGroup.create_thread(std::bind(&CScheduler::serviceQueue, &scheduler));
-    GetMainSignals().RegisterBackgroundSignalScheduler(scheduler);
+    threadGroup.create_thread([&]{ m_node.scheduler->serviceQueue(); });
+    GetMainSignals().RegisterBackgroundSignalScheduler(*g_rpc_node->scheduler);
 
     pblocktree.reset(new CBlockTreeDB(1 << 20, true));
     g_chainstate = MakeUnique<CChainState>();
@@ -147,6 +149,7 @@ TestingSetup::~TestingSetup()
     m_node.connman.reset();
     m_node.banman.reset();
     m_node.mempool = nullptr;
+    m_node.scheduler.reset();
     UnloadBlockIndex();
     g_chainstate.reset();
     pblocktree.reset();

src/test/util/setup_common.h

@@ -85,7 +85,6 @@ private:
 struct TestingSetup : public BasicTestingSetup {
     NodeContext m_node;
     boost::thread_group threadGroup;
-    CScheduler scheduler;
 
     explicit TestingSetup(const std::string& chainName = CBaseChainParams::MAIN);
     ~TestingSetup();