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