Branch data Line data Source code
# 1 : : // Copyright (c) 2012-2021 The Bitcoin Core developers
# 2 : : // Distributed under the MIT software license, see the accompanying
# 3 : : // file COPYING or http://www.opensource.org/licenses/mit-license.php.
# 4 : :
# 5 : : #include <random.h>
# 6 : : #include <scheduler.h>
# 7 : : #include <util/time.h>
# 8 : :
# 9 : : #include <boost/test/unit_test.hpp>
# 10 : :
# 11 : : #include <functional>
# 12 : : #include <mutex>
# 13 : : #include <thread>
# 14 : : #include <vector>
# 15 : :
# 16 : : BOOST_AUTO_TEST_SUITE(scheduler_tests)
# 17 : :
# 18 : : static void microTask(CScheduler& s, std::mutex& mutex, int& counter, int delta, std::chrono::system_clock::time_point rescheduleTime)
# 19 : 800 : {
# 20 : 800 : {
# 21 : 800 : std::lock_guard<std::mutex> lock(mutex);
# 22 : 800 : counter += delta;
# 23 : 800 : }
# 24 : 800 : std::chrono::system_clock::time_point noTime = std::chrono::system_clock::time_point::min();
# 25 [ + + ]: 800 : if (rescheduleTime != noTime) {
# 26 : 400 : CScheduler::Function f = std::bind(µTask, std::ref(s), std::ref(mutex), std::ref(counter), -delta + 1, noTime);
# 27 : 400 : s.schedule(f, rescheduleTime);
# 28 : 400 : }
# 29 : 800 : }
# 30 : :
# 31 : : BOOST_AUTO_TEST_CASE(manythreads)
# 32 : 2 : {
# 33 : : // Stress test: hundreds of microsecond-scheduled tasks,
# 34 : : // serviced by 10 threads.
# 35 : : //
# 36 : : // So... ten shared counters, which if all the tasks execute
# 37 : : // properly will sum to the number of tasks done.
# 38 : : // Each task adds or subtracts a random amount from one of the
# 39 : : // counters, and then schedules another task 0-1000
# 40 : : // microseconds in the future to subtract or add from
# 41 : : // the counter -random_amount+1, so in the end the shared
# 42 : : // counters should sum to the number of initial tasks performed.
# 43 : 2 : CScheduler microTasks;
# 44 : :
# 45 : 2 : std::mutex counterMutex[10];
# 46 : 2 : int counter[10] = { 0 };
# 47 : 2 : FastRandomContext rng{/*fDeterministic=*/true};
# 48 : 400 : auto zeroToNine = [](FastRandomContext& rc) -> int { return rc.randrange(10); }; // [0, 9]
# 49 : 800 : auto randomMsec = [](FastRandomContext& rc) -> int { return -11 + (int)rc.randrange(1012); }; // [-11, 1000]
# 50 : 400 : auto randomDelta = [](FastRandomContext& rc) -> int { return -1000 + (int)rc.randrange(2001); }; // [-1000, 1000]
# 51 : :
# 52 : 2 : std::chrono::system_clock::time_point start = std::chrono::system_clock::now();
# 53 : 2 : std::chrono::system_clock::time_point now = start;
# 54 : 2 : std::chrono::system_clock::time_point first, last;
# 55 : 2 : size_t nTasks = microTasks.getQueueInfo(first, last);
# 56 : 2 : BOOST_CHECK(nTasks == 0);
# 57 : :
# 58 [ + + ]: 202 : for (int i = 0; i < 100; ++i) {
# 59 : 200 : std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
# 60 : 200 : std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
# 61 : 200 : int whichCounter = zeroToNine(rng);
# 62 : 200 : CScheduler::Function f = std::bind(µTask, std::ref(microTasks),
# 63 : 200 : std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),
# 64 : 200 : randomDelta(rng), tReschedule);
# 65 : 200 : microTasks.schedule(f, t);
# 66 : 200 : }
# 67 : 2 : nTasks = microTasks.getQueueInfo(first, last);
# 68 : 2 : BOOST_CHECK(nTasks == 100);
# 69 : 2 : BOOST_CHECK(first < last);
# 70 : 2 : BOOST_CHECK(last > now);
# 71 : :
# 72 : : // As soon as these are created they will start running and servicing the queue
# 73 : 2 : std::vector<std::thread> microThreads;
# 74 [ + + ]: 12 : for (int i = 0; i < 5; i++)
# 75 : 10 : microThreads.emplace_back(std::bind(&CScheduler::serviceQueue, µTasks));
# 76 : :
# 77 : 2 : UninterruptibleSleep(std::chrono::microseconds{600});
# 78 : 2 : now = std::chrono::system_clock::now();
# 79 : :
# 80 : : // More threads and more tasks:
# 81 [ + + ]: 12 : for (int i = 0; i < 5; i++)
# 82 : 10 : microThreads.emplace_back(std::bind(&CScheduler::serviceQueue, µTasks));
# 83 [ + + ]: 202 : for (int i = 0; i < 100; i++) {
# 84 : 200 : std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
# 85 : 200 : std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
# 86 : 200 : int whichCounter = zeroToNine(rng);
# 87 : 200 : CScheduler::Function f = std::bind(µTask, std::ref(microTasks),
# 88 : 200 : std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),
# 89 : 200 : randomDelta(rng), tReschedule);
# 90 : 200 : microTasks.schedule(f, t);
# 91 : 200 : }
# 92 : :
# 93 : : // Drain the task queue then exit threads
# 94 : 2 : microTasks.StopWhenDrained();
# 95 : : // wait until all the threads are done
# 96 [ + + ]: 20 : for (auto& thread: microThreads) {
# 97 [ + - ]: 20 : if (thread.joinable()) thread.join();
# 98 : 20 : }
# 99 : :
# 100 : 2 : int counterSum = 0;
# 101 [ + + ]: 22 : for (int i = 0; i < 10; i++) {
# 102 : 20 : BOOST_CHECK(counter[i] != 0);
# 103 : 20 : counterSum += counter[i];
# 104 : 20 : }
# 105 : 2 : BOOST_CHECK_EQUAL(counterSum, 200);
# 106 : 2 : }
# 107 : :
# 108 : : BOOST_AUTO_TEST_CASE(wait_until_past)
# 109 : 2 : {
# 110 : 2 : std::condition_variable condvar;
# 111 : 2 : Mutex mtx;
# 112 : 2 : WAIT_LOCK(mtx, lock);
# 113 : :
# 114 : 12 : const auto no_wait= [&](const std::chrono::seconds& d) {
# 115 : 12 : return condvar.wait_until(lock, std::chrono::system_clock::now() - d);
# 116 : 12 : };
# 117 : :
# 118 : 2 : BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::seconds{1}));
# 119 : 2 : BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::minutes{1}));
# 120 : 2 : BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{1}));
# 121 : 2 : BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{10}));
# 122 : 2 : BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{100}));
# 123 : 2 : BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{1000}));
# 124 : 2 : }
# 125 : :
# 126 : : BOOST_AUTO_TEST_CASE(singlethreadedscheduler_ordered)
# 127 : 2 : {
# 128 : 2 : CScheduler scheduler;
# 129 : :
# 130 : : // each queue should be well ordered with respect to itself but not other queues
# 131 : 2 : SingleThreadedSchedulerClient queue1(&scheduler);
# 132 : 2 : SingleThreadedSchedulerClient queue2(&scheduler);
# 133 : :
# 134 : : // create more threads than queues
# 135 : : // if the queues only permit execution of one task at once then
# 136 : : // the extra threads should effectively be doing nothing
# 137 : : // if they don't we'll get out of order behaviour
# 138 : 2 : std::vector<std::thread> threads;
# 139 [ + + ]: 12 : for (int i = 0; i < 5; ++i) {
# 140 : 10 : threads.emplace_back(std::bind(&CScheduler::serviceQueue, &scheduler));
# 141 : 10 : }
# 142 : :
# 143 : : // these are not atomic, if SinglethreadedSchedulerClient prevents
# 144 : : // parallel execution at the queue level no synchronization should be required here
# 145 : 2 : int counter1 = 0;
# 146 : 2 : int counter2 = 0;
# 147 : :
# 148 : : // just simply count up on each queue - if execution is properly ordered then
# 149 : : // the callbacks should run in exactly the order in which they were enqueued
# 150 [ + + ]: 202 : for (int i = 0; i < 100; ++i) {
# 151 : 200 : queue1.AddToProcessQueue([i, &counter1]() {
# 152 : 200 : bool expectation = i == counter1++;
# 153 : 200 : assert(expectation);
# 154 : 200 : });
# 155 : :
# 156 : 200 : queue2.AddToProcessQueue([i, &counter2]() {
# 157 : 200 : bool expectation = i == counter2++;
# 158 : 200 : assert(expectation);
# 159 : 200 : });
# 160 : 200 : }
# 161 : :
# 162 : : // finish up
# 163 : 2 : scheduler.StopWhenDrained();
# 164 [ + + ]: 10 : for (auto& thread: threads) {
# 165 [ + - ]: 10 : if (thread.joinable()) thread.join();
# 166 : 10 : }
# 167 : :
# 168 : 2 : BOOST_CHECK_EQUAL(counter1, 100);
# 169 : 2 : BOOST_CHECK_EQUAL(counter2, 100);
# 170 : 2 : }
# 171 : :
# 172 : : BOOST_AUTO_TEST_CASE(mockforward)
# 173 : 2 : {
# 174 : 2 : CScheduler scheduler;
# 175 : :
# 176 : 2 : int counter{0};
# 177 : 4 : CScheduler::Function dummy = [&counter]{counter++;};
# 178 : :
# 179 : : // schedule jobs for 2, 5 & 8 minutes into the future
# 180 : :
# 181 : 2 : scheduler.scheduleFromNow(dummy, std::chrono::minutes{2});
# 182 : 2 : scheduler.scheduleFromNow(dummy, std::chrono::minutes{5});
# 183 : 2 : scheduler.scheduleFromNow(dummy, std::chrono::minutes{8});
# 184 : :
# 185 : : // check taskQueue
# 186 : 2 : std::chrono::system_clock::time_point first, last;
# 187 : 2 : size_t num_tasks = scheduler.getQueueInfo(first, last);
# 188 : 2 : BOOST_CHECK_EQUAL(num_tasks, 3ul);
# 189 : :
# 190 : 2 : std::thread scheduler_thread([&]() { scheduler.serviceQueue(); });
# 191 : :
# 192 : : // bump the scheduler forward 5 minutes
# 193 : 2 : scheduler.MockForward(std::chrono::minutes{5});
# 194 : :
# 195 : : // ensure scheduler has chance to process all tasks queued for before 1 ms from now.
# 196 : 2 : scheduler.scheduleFromNow([&scheduler] { scheduler.stop(); }, std::chrono::milliseconds{1});
# 197 : 2 : scheduler_thread.join();
# 198 : :
# 199 : : // check that the queue only has one job remaining
# 200 : 2 : num_tasks = scheduler.getQueueInfo(first, last);
# 201 : 2 : BOOST_CHECK_EQUAL(num_tasks, 1ul);
# 202 : :
# 203 : : // check that the dummy function actually ran
# 204 : 2 : BOOST_CHECK_EQUAL(counter, 2);
# 205 : :
# 206 : : // check that the time of the remaining job has been updated
# 207 : 2 : std::chrono::system_clock::time_point now = std::chrono::system_clock::now();
# 208 : 2 : int delta = std::chrono::duration_cast<std::chrono::seconds>(first - now).count();
# 209 : : // should be between 2 & 3 minutes from now
# 210 : 2 : BOOST_CHECK(delta > 2*60 && delta < 3*60);
# 211 : 2 : }
# 212 : :
# 213 : : BOOST_AUTO_TEST_SUITE_END()
|
