src/scheduler.cpp

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