src/checkqueue.h

   1 // Copyright (c) 2012 The Bitcoin developers
   2 // Distributed under the MIT/X11 software license, see the accompanying
   3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
   4
   5 #ifndef CHECKQUEUE_H
   6 #define CHECKQUEUE_H
   7
   8 #include <algorithm>
   9 #include <vector>
  10
  11 #include <boost/thread/condition_variable.hpp>
  12 #include <boost/thread/locks.hpp>
  13 #include <boost/thread/mutex.hpp>
  14
  15 extern bool fShutdown;
  16
  17 template<typename T> class CCheckQueueControl;
  18
  19 /** Queue for verifications that have to be performed.
  20   * The verifications are represented by a type T, which must provide an
  21   * operator(), returning a bool.
  22   *
  23   * One thread (the master) is assumed to push batches of verifications
  24   * onto the queue, where they are processed by N-1 worker threads. When
  25   * the master is done adding work, it temporarily joins the worker pool
  26   * as an N'th worker, until all jobs are done.
  27   */
  28 template<typename T> class CCheckQueue {
  29 private:
  30     // Mutex to protect the inner state
  31     boost::mutex mutex;
  32
  33     // Worker threads block on this when out of work
  34     boost::condition_variable condWorker;
  35
  36     // Master thread blocks on this when out of work
  37     boost::condition_variable condMaster;
  38
  39     // Quit method blocks on this until all workers are gone
  40     boost::condition_variable condQuit;
  41
  42     // The queue of elements to be processed.
  43     // As the order of booleans doesn't matter, it is used as a LIFO (stack)
  44     std::vector<T> queue;
  45
  46     // The number of workers (including the master) that are idle.
  47     int nIdle;
  48
  49     // The total number of workers (including the master).
  50     int nTotal;
  51
  52     // The temporary evaluation result.
  53     bool fAllOk;
  54
  55     // Number of verifications that haven't completed yet.
  56     // This includes elements that are not anymore in queue, but still in
  57     // worker's own batches.
  58     unsigned int nTodo;
  59
  60     // Whether we're shutting down.
  61     bool fQuit;
  62
  63     // The maximum number of elements to be processed in one batch
  64     unsigned int nBatchSize;
  65
  66     // Internal function that does bulk of the verification work.
  67     bool Loop(bool fMaster = false) {
  68         boost::condition_variable &cond = fMaster ? condMaster : condWorker;
  69         std::vector<T> vChecks;
  70         vChecks.reserve(nBatchSize);
  71         unsigned int nNow = 0;
  72         bool fOk = true;
  73         do {
  74             {
  75                 boost::unique_lock<boost::mutex> lock(mutex);
  76                 // first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
  77                 if (nNow) {
  78                     fAllOk &= fOk;
  79                     nTodo -= nNow;
  80                     if (nTodo == 0 && !fMaster)
  81                         // We processed the last element; inform the master he can exit and return the result
  82                         condMaster.notify_one();
  83                 } else {
  84                     // first iteration
  85                     nTotal++;
  86                 }
  87                 // logically, the do loop starts here
  88                 while (queue.empty()) {
  89                     if ((fMaster || fQuit) && nTodo == 0) {
  90                         nTotal--;
  91                         if (nTotal==0)
  92                             condQuit.notify_one();
  93                         bool fRet = fAllOk;
  94                         // reset the status for new work later
  95                         if (fMaster)
  96                             fAllOk = true;
  97                         // return the current status
  98                         return fRet;
  99                     }
 100                     nIdle++;
 101                     cond.wait(lock); // wait
 102                     nIdle--;
 103                 }
 104                 // Decide how many work units to process now.
 105                 // * Do not try to do everything at once, but aim for increasingly smaller batches so
 106                 //   all workers finish approximately simultaneously.
 107                 // * Try to account for idle jobs which will instantly start helping.
 108                 // * Don't do batches smaller than 1 (duh), or larger than nBatchSize.
 109                 nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1)));
 110                 vChecks.resize(nNow);
 111                 for (unsigned int i = 0; i < nNow; i++) {
 112                      // We want the lock on the mutex to be as short as possible, so swap jobs from the global
 113                      // queue to the local batch vector instead of copying.
 114                      vChecks[i].swap(queue.back());
 115                      queue.pop_back();
 116                 }
 117                 // Check whether we need to do work at all
 118                 fOk = fAllOk;
 119             }
 120             // execute work
 121             for (T &check : vChecks)
 122                 if (fOk)
 123                     fOk = check();
 124             vChecks.clear();
 125         } while(true && !fShutdown); // HACK: force queue to shut down
 126         return false;
 127     }
 128
 129 public:
 130     // Create a new check queue
 131     CCheckQueue(unsigned int nBatchSizeIn) :
 132         nIdle(0), nTotal(0), fAllOk(true), nTodo(0), fQuit(false), nBatchSize(nBatchSizeIn) {}
 133
 134     // Worker thread
 135     void Thread() {
 136         Loop();
 137     }
 138
 139     // Wait until execution finishes, and return whether all evaluations where succesful.
 140     bool Wait() {
 141         return Loop(true);
 142     }
 143
 144     // Add a batch of checks to the queue
 145     void Add(std::vector<T> &vChecks) {
 146         boost::unique_lock<boost::mutex> lock(mutex);
 147         for (T &check : vChecks) {
 148             queue.push_back(T());
 149             check.swap(queue.back());
 150         }
 151         nTodo += vChecks.size();
 152         if (vChecks.size() == 1)
 153             condWorker.notify_one();
 154         else if (vChecks.size() > 1)
 155             condWorker.notify_all();
 156     }
 157
 158     // Shut the queue down
 159     void Quit() {
 160         boost::unique_lock<boost::mutex> lock(mutex);
 161         fQuit = true;
 162         // No need to wake the master, as he will quit automatically when all jobs are
 163         // done.
 164         condWorker.notify_all();
 165
 166         while (nTotal > 0)
 167             condQuit.wait(lock);
 168     }
 169
 170     ~CCheckQueue() {
 171         Quit();
 172     }
 173
 174     bool IsIdle()
 175     {
 176         boost::unique_lock<boost::mutex> lock(mutex);
 177         return (nTotal == nIdle && nTodo == 0 && fAllOk == true);
 178     }
 179 };
 180
 181 /** RAII-style controller object for a CCheckQueue that guarantees the passed
 182  *  queue is finished before continuing.
 183  */
 184 template<typename T> class CCheckQueueControl {
 185 private:
 186     CCheckQueue<T> *pqueue;
 187     bool fDone;
 188
 189 public:
 190     CCheckQueueControl(CCheckQueue<T> *pqueueIn) : pqueue(pqueueIn), fDone(false) {
 191         // passed queue is supposed to be unused, or NULL
 192         if (pqueue != NULL) {
 193             bool isIdle = pqueue->IsIdle();
 194             assert(isIdle);
 195         }
 196     }
 197
 198     bool Wait() {
 199         if (pqueue == NULL)
 200             return true;
 201         bool fRet = pqueue->Wait();
 202         fDone = true;
 203         return fRet;
 204     }
 205
 206     void Add(std::vector<T> &vChecks) {
 207         if (pqueue != NULL)
 208             pqueue->Add(vChecks);
 209     }
 210
 211     ~CCheckQueueControl() {
 212         if (!fDone)
 213             Wait();
 214     }
 215 };
 216
 217 #endif