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     newTaskScheduled.notify_one();
  83 }
  84
  85 void CScheduler::stop(bool drain)
  86 {
  87     {
  88         boost::unique_lock<boost::mutex> lock(newTaskMutex);
  89         if (drain)
  90             stopWhenEmpty = true;
  91         else
  92             stopRequested = true;
  93     }
  94     newTaskScheduled.notify_all();
  95 }
  96
  97 void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::time_point t)
  98 {
  99     {
 100         boost::unique_lock<boost::mutex> lock(newTaskMutex);
 101         taskQueue.insert(std::make_pair(t, f));
 102     }
 103     newTaskScheduled.notify_one();
 104 }
 105
 106 void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaSeconds)
 107 {
 108     schedule(f, boost::chrono::system_clock::now() + boost::chrono::seconds(deltaSeconds));
 109 }
 110
 111 static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaSeconds)
 112 {
 113     f();
 114     s->scheduleFromNow(boost::bind(&Repeat, s, f, deltaSeconds), deltaSeconds);
 115 }
 116
 117 void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaSeconds)
 118 {
 119     scheduleFromNow(boost::bind(&Repeat, this, f, deltaSeconds), deltaSeconds);
 120 }
 121
 122 size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first,
 123                              boost::chrono::system_clock::time_point &last) const
 124 {
 125     boost::unique_lock<boost::mutex> lock(newTaskMutex);
 126     size_t result = taskQueue.size();
 127     if (!taskQueue.empty()) {
 128         first = taskQueue.begin()->first;
 129         last = taskQueue.rbegin()->first;
 130     }
 131     return result;
 132 }