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 "reverselock.h"
   8
   9 #include <assert.h>
  10 #include <boost/bind.hpp>
  11 #include <utility>
  12
  13 CScheduler::CScheduler() : nThreadsServicingQueue(0), stopRequested(false), stopWhenEmpty(false)
  14 {
  15 }
  16
  17 CScheduler::~CScheduler()
  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
  32 void CScheduler::serviceQueue()
  33 {
  34     boost::unique_lock<boost::mutex> lock(newTaskMutex);
  35     ++nThreadsServicingQueue;
  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             while (!shouldStop() && taskQueue.empty()) {
  43                 // Wait until there is something to do.
  44                 newTaskScheduled.wait(lock);
  45             }
  46
  47             // Wait until either there is a new task, or until
  48             // the time of the first item on the queue:
  49
  50 // wait_until needs boost 1.50 or later; older versions have timed_wait:
  51 #if BOOST_VERSION < 105000
  52             while (!shouldStop() && !taskQueue.empty() &&
  53                    newTaskScheduled.timed_wait(lock, toPosixTime(taskQueue.begin()->first))) {
  54                 // Keep waiting until timeout
  55             }
  56 #else
  57             // Some boost versions have a conflicting overload of wait_until that returns void.
  58             // Explicitly use a template here to avoid hitting that overload.
  59             while (!shouldStop() && !taskQueue.empty()) {
  60                 boost::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first;
  61                 if (newTaskScheduled.wait_until<>(lock, timeToWaitFor) == boost::cv_status::timeout)
  62                     break; // Exit loop after timeout, it means we reached the time of the event
  63             }
  64 #endif
  65             // If there are multiple threads, the queue can empty while we're waiting (another
  66             // thread may service the task we were waiting on).
  67             if (shouldStop() || taskQueue.empty())
  68                 continue;
  69
  70             Function f = taskQueue.begin()->second;
  71             taskQueue.erase(taskQueue.begin());
  72
  73             {
  74                 // Unlock before calling f, so it can reschedule itself or another task
  75                 // without deadlocking:
  76                 reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
  77                 f();
  78             }
  79         } catch (...) {
  80             --nThreadsServicingQueue;
  81             throw;
  82         }
  83     }
  84     --nThreadsServicingQueue;
  85     newTaskScheduled.notify_one();
  86 }
  87
  88 void CScheduler::stop(bool drain)
  89 {
  90     {
  91         boost::unique_lock<boost::mutex> lock(newTaskMutex);
  92         if (drain)
  93             stopWhenEmpty = true;
  94         else
  95             stopRequested = true;
  96     }
  97     newTaskScheduled.notify_all();
  98 }
  99
 100 void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::time_point t)
 101 {
 102     {
 103         boost::unique_lock<boost::mutex> lock(newTaskMutex);
 104         taskQueue.insert(std::make_pair(t, f));
 105     }
 106     newTaskScheduled.notify_one();
 107 }
 108
 109 void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaMilliSeconds)
 110 {
 111     schedule(f, boost::chrono::system_clock::now() + boost::chrono::milliseconds(deltaMilliSeconds));
 112 }
 113
 114 static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaMilliSeconds)
 115 {
 116     f();
 117     s->scheduleFromNow(boost::bind(&Repeat, s, f, deltaMilliSeconds), deltaMilliSeconds);
 118 }
 119
 120 void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaMilliSeconds)
 121 {
 122     scheduleFromNow(boost::bind(&Repeat, this, f, deltaMilliSeconds), deltaMilliSeconds);
 123 }
 124
 125 size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,
 126                              boost::chrono::system_clock::time_point &last) const
 127 {
 128     boost::unique_lock<boost::mutex> lock(newTaskMutex);
 129     size_t result = taskQueue.size();
 130     if (!taskQueue.empty()) {
 131         first = taskQueue.begin()->first;
 132         last = taskQueue.rbegin()->first;
 133     }
 134     return result;
 135 }