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