1 // Copyright (c) 2012-2020 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
BOOST_AUTO_TEST_SUITE(scheduler_tests)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 {
20 {
21 std::lock_guard<std::mutex> lock(mutex);
22 counter += delta;
23 }
24 std::chrono::system_clock::time_point noTime = std::chrono::system_clock::time_point::min();
25 if (rescheduleTime != noTime) {
26 CScheduler::Function f = std::bind(µTask, std::ref(s), std::ref(mutex), std::ref(counter), -delta + 1, noTime);
27 s.schedule(f, rescheduleTime);
28 }
29 }
30
BOOST_AUTO_TEST_CASE(manythreads)31 BOOST_AUTO_TEST_CASE(manythreads)
32 {
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 CScheduler microTasks;
44
45 std::mutex counterMutex[10];
46 int counter[10] = { 0 };
47 FastRandomContext rng{/* fDeterministic */ true};
48 auto zeroToNine = [](FastRandomContext& rc) -> int { return rc.randrange(10); }; // [0, 9]
49 auto randomMsec = [](FastRandomContext& rc) -> int { return -11 + (int)rc.randrange(1012); }; // [-11, 1000]
50 auto randomDelta = [](FastRandomContext& rc) -> int { return -1000 + (int)rc.randrange(2001); }; // [-1000, 1000]
51
52 std::chrono::system_clock::time_point start = std::chrono::system_clock::now();
53 std::chrono::system_clock::time_point now = start;
54 std::chrono::system_clock::time_point first, last;
55 size_t nTasks = microTasks.getQueueInfo(first, last);
56 BOOST_CHECK(nTasks == 0);
57
58 for (int i = 0; i < 100; ++i) {
59 std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
60 std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
61 int whichCounter = zeroToNine(rng);
62 CScheduler::Function f = std::bind(µTask, std::ref(microTasks),
63 std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),
64 randomDelta(rng), tReschedule);
65 microTasks.schedule(f, t);
66 }
67 nTasks = microTasks.getQueueInfo(first, last);
68 BOOST_CHECK(nTasks == 100);
69 BOOST_CHECK(first < last);
70 BOOST_CHECK(last > now);
71
72 // As soon as these are created they will start running and servicing the queue
73 std::vector<std::thread> microThreads;
74 for (int i = 0; i < 5; i++)
75 microThreads.emplace_back(std::bind(&CScheduler::serviceQueue, µTasks));
76
77 UninterruptibleSleep(std::chrono::microseconds{600});
78 now = std::chrono::system_clock::now();
79
80 // More threads and more tasks:
81 for (int i = 0; i < 5; i++)
82 microThreads.emplace_back(std::bind(&CScheduler::serviceQueue, µTasks));
83 for (int i = 0; i < 100; i++) {
84 std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
85 std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
86 int whichCounter = zeroToNine(rng);
87 CScheduler::Function f = std::bind(µTask, std::ref(microTasks),
88 std::ref(counterMutex[whichCounter]), std::ref(counter[whichCounter]),
89 randomDelta(rng), tReschedule);
90 microTasks.schedule(f, t);
91 }
92
93 // Drain the task queue then exit threads
94 microTasks.StopWhenDrained();
95 // wait until all the threads are done
96 for (auto& thread: microThreads) {
97 if (thread.joinable()) thread.join();
98 }
99
100 int counterSum = 0;
101 for (int i = 0; i < 10; i++) {
102 BOOST_CHECK(counter[i] != 0);
103 counterSum += counter[i];
104 }
105 BOOST_CHECK_EQUAL(counterSum, 200);
106 }
107
BOOST_AUTO_TEST_CASE(wait_until_past)108 BOOST_AUTO_TEST_CASE(wait_until_past)
109 {
110 std::condition_variable condvar;
111 Mutex mtx;
112 WAIT_LOCK(mtx, lock);
113
114 const auto no_wait= [&](const std::chrono::seconds& d) {
115 return condvar.wait_until(lock, std::chrono::system_clock::now() - d);
116 };
117
118 BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::seconds{1}));
119 BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::minutes{1}));
120 BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{1}));
121 BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{10}));
122 BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{100}));
123 BOOST_CHECK(std::cv_status::timeout == no_wait(std::chrono::hours{1000}));
124 }
125
BOOST_AUTO_TEST_CASE(singlethreadedscheduler_ordered)126 BOOST_AUTO_TEST_CASE(singlethreadedscheduler_ordered)
127 {
128 CScheduler scheduler;
129
130 // each queue should be well ordered with respect to itself but not other queues
131 SingleThreadedSchedulerClient queue1(&scheduler);
132 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 std::vector<std::thread> threads;
139 for (int i = 0; i < 5; ++i) {
140 threads.emplace_back(std::bind(&CScheduler::serviceQueue, &scheduler));
141 }
142
143 // these are not atomic, if SinglethreadedSchedulerClient prevents
144 // parallel execution at the queue level no synchronization should be required here
145 int counter1 = 0;
146 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 for (int i = 0; i < 100; ++i) {
151 queue1.AddToProcessQueue([i, &counter1]() {
152 bool expectation = i == counter1++;
153 assert(expectation);
154 });
155
156 queue2.AddToProcessQueue([i, &counter2]() {
157 bool expectation = i == counter2++;
158 assert(expectation);
159 });
160 }
161
162 // finish up
163 scheduler.StopWhenDrained();
164 for (auto& thread: threads) {
165 if (thread.joinable()) thread.join();
166 }
167
168 BOOST_CHECK_EQUAL(counter1, 100);
169 BOOST_CHECK_EQUAL(counter2, 100);
170 }
171
BOOST_AUTO_TEST_CASE(mockforward)172 BOOST_AUTO_TEST_CASE(mockforward)
173 {
174 CScheduler scheduler;
175
176 int counter{0};
177 CScheduler::Function dummy = [&counter]{counter++;};
178
179 // schedule jobs for 2, 5 & 8 minutes into the future
180
181 scheduler.scheduleFromNow(dummy, std::chrono::minutes{2});
182 scheduler.scheduleFromNow(dummy, std::chrono::minutes{5});
183 scheduler.scheduleFromNow(dummy, std::chrono::minutes{8});
184
185 // check taskQueue
186 std::chrono::system_clock::time_point first, last;
187 size_t num_tasks = scheduler.getQueueInfo(first, last);
188 BOOST_CHECK_EQUAL(num_tasks, 3ul);
189
190 std::thread scheduler_thread([&]() { scheduler.serviceQueue(); });
191
192 // bump the scheduler forward 5 minutes
193 scheduler.MockForward(std::chrono::minutes{5});
194
195 // ensure scheduler has chance to process all tasks queued for before 1 ms from now.
196 scheduler.scheduleFromNow([&scheduler] { scheduler.stop(); }, std::chrono::milliseconds{1});
197 scheduler_thread.join();
198
199 // check that the queue only has one job remaining
200 num_tasks = scheduler.getQueueInfo(first, last);
201 BOOST_CHECK_EQUAL(num_tasks, 1ul);
202
203 // check that the dummy function actually ran
204 BOOST_CHECK_EQUAL(counter, 2);
205
206 // check that the time of the remaining job has been updated
207 std::chrono::system_clock::time_point now = std::chrono::system_clock::now();
208 int delta = std::chrono::duration_cast<std::chrono::seconds>(first - now).count();
209 // should be between 2 & 3 minutes from now
210 BOOST_CHECK(delta > 2*60 && delta < 3*60);
211 }
212
213 BOOST_AUTO_TEST_SUITE_END()
214