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