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