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