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