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