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