Bitcoin Core  22.99.0
P2P Digital Currency
scheduler_tests.cpp
Go to the documentation of this file.
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 
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(&microTask, std::ref(s), std::ref(mutex), std::ref(counter), -delta + 1, noTime);
27  s.schedule(f, rescheduleTime);
28  }
29 }
30 
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(&microTask, 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, &microTasks));
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, &microTasks));
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(&microTask, 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 
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 
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 
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 
CScheduler
Simple class for background tasks that should be run periodically or once "after a while".
Definition: scheduler.h:33
CScheduler::MockForward
void MockForward(std::chrono::seconds delta_seconds)
Mock the scheduler to fast forward in time.
Definition: scheduler.cpp:82
assert
assert(!tx.IsCoinBase())
BOOST_AUTO_TEST_SUITE
BOOST_AUTO_TEST_SUITE(cuckoocache_tests)
Test Suite for CuckooCache.
AnnotatedMixin< std::mutex >
microTask
static void microTask(CScheduler &s, std::mutex &mutex, int &counter, int delta, std::chrono::system_clock::time_point rescheduleTime)
Definition: scheduler_tests.cpp:18
BOOST_AUTO_TEST_SUITE_END
BOOST_AUTO_TEST_SUITE_END()
scheduler.h
CScheduler::serviceQueue
void serviceQueue()
Services the queue 'forever'.
Definition: scheduler.cpp:25
CScheduler::getQueueInfo
size_t getQueueInfo(std::chrono::system_clock::time_point &first, std::chrono::system_clock::time_point &last) const
Returns number of tasks waiting to be serviced, and first and last task times.
Definition: scheduler.cpp:115
random.h
CScheduler::Function
std::function< void()> Function
Definition: scheduler.h:41
time.h
CScheduler::stop
void stop()
Tell any threads running serviceQueue to stop as soon as the current task is done.
Definition: scheduler.h:73
UninterruptibleSleep
void UninterruptibleSleep(const std::chrono::microseconds &n)
Definition: time.cpp:22
CScheduler::StopWhenDrained
void StopWhenDrained()
Tell any threads running serviceQueue to stop when there is no work left to be done.
Definition: scheduler.h:80
SingleThreadedSchedulerClient::AddToProcessQueue
void AddToProcessQueue(std::function< void()> func)
Add a callback to be executed.
Definition: scheduler.cpp:178
FastRandomContext::randrange
uint64_t randrange(uint64_t range) noexcept
Generate a random integer in the range [0..range).
Definition: random.h:190
CScheduler::scheduleFromNow
void scheduleFromNow(Function f, std::chrono::milliseconds delta)
Call f once after the delta has passed.
Definition: scheduler.h:47
SingleThreadedSchedulerClient
Class used by CScheduler clients which may schedule multiple jobs which are required to be run serial...
Definition: scheduler.h:117
BOOST_AUTO_TEST_CASE
BOOST_AUTO_TEST_CASE(manythreads)
Definition: scheduler_tests.cpp:31
CScheduler::schedule
void schedule(Function f, std::chrono::system_clock::time_point t)
Call func at/after time t.
Definition: scheduler.cpp:73
FastRandomContext
Fast randomness source.
Definition: random.h:119
BOOST_CHECK
#define BOOST_CHECK(expr)
Definition: object.cpp:17
BOOST_CHECK_EQUAL
#define BOOST_CHECK_EQUAL(v1, v2)
Definition: object.cpp:18
WAIT_LOCK
#define WAIT_LOCK(cs, name)
Definition: sync.h:231