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 "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                    newTaskScheduled.wait_until<>(lock, taskQueue.begin()->first) != boost::cv_status::timeout) {
  59                 // Keep waiting until timeout
  60             }
  61 #endif
  62             // If there are multiple threads, the queue can empty while we're waiting (another
  63             // thread may service the task we were waiting on).
  64             if (shouldStop() || taskQueue.empty())
  65                 continue;
  66
  67             Function f = taskQueue.begin()->second;
  68             taskQueue.erase(taskQueue.begin());
  69
  70             {
  71                 // Unlock before calling f, so it can reschedule itself or another task
  72                 // without deadlocking:
  73                 reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock);
  74                 f();
  75             }
  76         } catch (...) {
  77             --nThreadsServicingQueue;
  78             throw;
  79         }
  80     }
  81     --nThreadsServicingQueue;
  82 }
  83
  84 void CScheduler::stop(bool drain)
  85 {
  86     {
  87         boost::unique_lock<boost::mutex> lock(newTaskMutex);
  88         if (drain)
  89             stopWhenEmpty = true;
  90         else
  91             stopRequested = true;
  92     }
  93     newTaskScheduled.notify_all();
  94 }
  95
  96 void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::time_point t)
  97 {
  98     {
  99         boost::unique_lock<boost::mutex> lock(newTaskMutex);
 100         taskQueue.insert(std::make_pair(t, f));
 101     }
 102     newTaskScheduled.notify_one();
 103 }
 104
 105 void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaSeconds)
 106 {
 107     schedule(f, boost::chrono::system_clock::now() + boost::chrono::seconds(deltaSeconds));
 108 }
 109
 110 static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaSeconds)
 111 {
 112     f();
 113     s->scheduleFromNow(boost::bind(&Repeat, s, f, deltaSeconds), deltaSeconds);
 114 }
 115
 116 void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaSeconds)
 117 {
 118     scheduleFromNow(boost::bind(&Repeat, this, f, deltaSeconds), deltaSeconds);
 119 }
 120
 121 size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,
 122                              boost::chrono::system_clock::time_point &last) const
 123 {
 124     boost::unique_lock<boost::mutex> lock(newTaskMutex);
 125     size_t result = taskQueue.size();
 126     if (!taskQueue.empty()) {
 127         first = taskQueue.begin()->first;
 128         last = taskQueue.rbegin()->first;
 129     }
 130     return result;
 131 }