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 }