Bitcoin Core  0.19.99
P2P Digital Currency
scheduler.cpp
Go to the documentation of this file.
1 // Copyright (c) 2015-2019 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 <scheduler.h>
6 
7 #include <random.h>
8 #include <reverselock.h>
9 
10 #include <assert.h>
11 #include <utility>
12 
13 CScheduler::CScheduler() : nThreadsServicingQueue(0), stopRequested(false), stopWhenEmpty(false)
14 {
15 }
16 
18 {
19  assert(nThreadsServicingQueue == 0);
20 }
21 
22 
23 #if BOOST_VERSION < 105000
24 static boost::system_time toPosixTime(const boost::chrono::system_clock::time_point& t)
25 {
26  // Creating the posix_time using from_time_t loses sub-second precision. So rather than exporting the time_point to time_t,
27  // start with a posix_time at the epoch (0) and add the milliseconds that have passed since then.
28  return boost::posix_time::from_time_t(0) + boost::posix_time::milliseconds(boost::chrono::duration_cast<boost::chrono::milliseconds>(t.time_since_epoch()).count());
29 }
30 #endif
31 
33 {
34  boost::unique_lock<boost::mutex> lock(newTaskMutex);
36 
37  // newTaskMutex is locked throughout this loop EXCEPT
38  // when the thread is waiting or when the user's function
39  // is called.
40  while (!shouldStop()) {
41  try {
42  if (!shouldStop() && taskQueue.empty()) {
44  }
45  while (!shouldStop() && taskQueue.empty()) {
46  // Wait until there is something to do.
47  newTaskScheduled.wait(lock);
48  }
49 
50  // Wait until either there is a new task, or until
51  // the time of the first item on the queue:
52 
53 // wait_until needs boost 1.50 or later; older versions have timed_wait:
54 #if BOOST_VERSION < 105000
55  while (!shouldStop() && !taskQueue.empty() &&
56  newTaskScheduled.timed_wait(lock, toPosixTime(taskQueue.begin()->first))) {
57  // Keep waiting until timeout
58  }
59 #else
60  // Some boost versions have a conflicting overload of wait_until that returns void.
61  // Explicitly use a template here to avoid hitting that overload.
62  while (!shouldStop() && !taskQueue.empty()) {
63  boost::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first;
64  if (newTaskScheduled.wait_until<>(lock, timeToWaitFor) == boost::cv_status::timeout)
65  break; // Exit loop after timeout, it means we reached the time of the event
66  }
67 #endif
68  // If there are multiple threads, the queue can empty while we're waiting (another
69  // thread may service the task we were waiting on).
70  if (shouldStop() || taskQueue.empty())
71  continue;
72 
73  Function f = taskQueue.begin()->second;
74  taskQueue.erase(taskQueue.begin());
75 
76  {
77  // Unlock before calling f, so it can reschedule itself or another task
78  // without deadlocking:
80  f();
81  }
82  } catch (...) {
84  throw;
85  }
86  }
88  newTaskScheduled.notify_one();
89 }
90 
91 void CScheduler::stop(bool drain)
92 {
93  {
94  boost::unique_lock<boost::mutex> lock(newTaskMutex);
95  if (drain)
96  stopWhenEmpty = true;
97  else
98  stopRequested = true;
99  }
100  newTaskScheduled.notify_all();
101 }
102 
104 {
105  {
106  boost::unique_lock<boost::mutex> lock(newTaskMutex);
107  taskQueue.insert(std::make_pair(t, f));
108  }
109  newTaskScheduled.notify_one();
110 }
111 
112 void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
113 {
114  schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds));
115 }
116 
117 void CScheduler::MockForward(boost::chrono::seconds delta_seconds)
118 {
119  assert(delta_seconds.count() > 0 && delta_seconds < boost::chrono::hours{1});
120 
121  {
122  boost::unique_lock<boost::mutex> lock(newTaskMutex);
123 
124  // use temp_queue to maintain updated schedule
125  std::multimap<boost::chrono::system_clock::time_point, Function> temp_queue;
126 
127  for (const auto& element : taskQueue) {
128  temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second);
129  }
130 
131  // point taskQueue to temp_queue
132  taskQueue = std::move(temp_queue);
133  }
134 
135  // notify that the taskQueue needs to be processed
136  newTaskScheduled.notify_one();
137 }
138 
139 static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)
140 {
141  f();
142  s->scheduleFromNow(std::bind(&Repeat, s, f, deltaMilliSeconds), deltaMilliSeconds);
143 }
144 
145 void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds)
146 {
147  scheduleFromNow(std::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);
148 }
149 
152 {
153  boost::unique_lock<boost::mutex> lock(newTaskMutex);
154  size_t result = taskQueue.size();
155  if (!taskQueue.empty()) {
156  first = taskQueue.begin()->first;
157  last = taskQueue.rbegin()->first;
158  }
159  return result;
160 }
161 
163  boost::unique_lock<boost::mutex> lock(newTaskMutex);
164  return nThreadsServicingQueue;
165 }
166 
167 
169  {
170  LOCK(m_cs_callbacks_pending);
171  // Try to avoid scheduling too many copies here, but if we
172  // accidentally have two ProcessQueue's scheduled at once its
173  // not a big deal.
174  if (m_are_callbacks_running) return;
175  if (m_callbacks_pending.empty()) return;
176  }
177  m_pscheduler->schedule(std::bind(&SingleThreadedSchedulerClient::ProcessQueue, this));
178 }
179 
181  std::function<void ()> callback;
182  {
183  LOCK(m_cs_callbacks_pending);
184  if (m_are_callbacks_running) return;
185  if (m_callbacks_pending.empty()) return;
186  m_are_callbacks_running = true;
187 
188  callback = std::move(m_callbacks_pending.front());
189  m_callbacks_pending.pop_front();
190  }
191 
192  // RAII the setting of fCallbacksRunning and calling MaybeScheduleProcessQueue
193  // to ensure both happen safely even if callback() throws.
194  struct RAIICallbacksRunning {
196  explicit RAIICallbacksRunning(SingleThreadedSchedulerClient* _instance) : instance(_instance) {}
197  ~RAIICallbacksRunning() {
198  {
199  LOCK(instance->m_cs_callbacks_pending);
200  instance->m_are_callbacks_running = false;
201  }
202  instance->MaybeScheduleProcessQueue();
203  }
204  } raiicallbacksrunning(this);
205 
206  callback();
207 }
208 
209 void SingleThreadedSchedulerClient::AddToProcessQueue(std::function<void ()> func) {
210  assert(m_pscheduler);
211 
212  {
213  LOCK(m_cs_callbacks_pending);
214  m_callbacks_pending.emplace_back(std::move(func));
215  }
216  MaybeScheduleProcessQueue();
217 }
218 
220  assert(!m_pscheduler->AreThreadsServicingQueue());
221  bool should_continue = true;
222  while (should_continue) {
223  ProcessQueue();
224  LOCK(m_cs_callbacks_pending);
225  should_continue = !m_callbacks_pending.empty();
226  }
227 }
228 
230  LOCK(m_cs_callbacks_pending);
231  return m_callbacks_pending.size();
232 }
Class used by CScheduler clients which may schedule multiple jobs which are required to be run serial...
Definition: scheduler.h:104
bool stopWhenEmpty
Definition: scheduler.h:90
std::multimap< boost::chrono::system_clock::time_point, Function > taskQueue
Definition: scheduler.h:85
bool shouldStop() const
Definition: scheduler.h:91
void scheduleEvery(Function f, int64_t deltaMilliSeconds)
Definition: scheduler.cpp:145
std::function< void()> Function
Definition: scheduler.h:43
void scheduleFromNow(Function f, int64_t deltaMilliSeconds)
Definition: scheduler.cpp:112
void AddToProcessQueue(std::function< void()> func)
Add a callback to be executed.
Definition: scheduler.cpp:209
void schedule(Function f, boost::chrono::system_clock::time_point t=boost::chrono::system_clock::now())
Definition: scheduler.cpp:103
static void Repeat(CScheduler *s, CScheduler::Function f, int64_t deltaMilliSeconds)
Definition: scheduler.cpp:139
static boost::system_time toPosixTime(const boost::chrono::system_clock::time_point &t)
Definition: scheduler.cpp:24
void stop(bool drain=false)
Definition: scheduler.cpp:91
#define LOCK(cs)
Definition: sync.h:179
RecursiveMutex m_cs_callbacks_pending
Definition: scheduler.h:108
boost::condition_variable newTaskScheduled
Definition: scheduler.h:86
void serviceQueue()
Definition: scheduler.cpp:32
An RAII-style reverse lock.
Definition: reverselock.h:12
void MockForward(boost::chrono::seconds delta_seconds)
Mock the scheduler to fast forward in time.
Definition: scheduler.cpp:117
boost::mutex newTaskMutex
Definition: scheduler.h:87
bool stopRequested
Definition: scheduler.h:89
bool AreThreadsServicingQueue() const
Definition: scheduler.cpp:162
clock::time_point time_point
Definition: bench.h:51
static int count
Definition: tests.c:45
int nThreadsServicingQueue
Definition: scheduler.h:88
size_t getQueueInfo(boost::chrono::system_clock::time_point &first, boost::chrono::system_clock::time_point &last) const
Definition: scheduler.cpp:150