src/scheduler.cpp

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