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