src/checkqueue.h

   1 // Copyright (c) 2012-2015 The Bitcoin Core developers
   2 // Distributed under the MIT software license, see the accompanying
   3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
   4
   5 #ifndef BITCOIN_CHECKQUEUE_H
   6 #define BITCOIN_CHECKQUEUE_H
   7
   8 #include "sync.h"
   9
  10 #include <algorithm>
  11 #include <vector>
  12
  13 #include <boost/thread/condition_variable.hpp>
  14 #include <boost/thread/mutex.hpp>
  15
  16 template <typename T>
  17 class CCheckQueueControl;
  18
  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>
  30 class CCheckQueue
  31 {
  32 private:
  33     //! Mutex to protect the inner state
  34     boost::mutex mutex;
  35
  36     //! Worker threads block on this when out of work
  37     boost::condition_variable condWorker;
  38
  39     //! Master thread blocks on this when out of work
  40     boost::condition_variable condMaster;
  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     /**
  56      * Number of verifications that haven't completed yet.
  57      * This includes elements that are no longer queued, but still in the
  58      * worker's own batches.
  59      */
  60     unsigned int nTodo;
  61
  62     //! Whether we're shutting down.
  63     bool fQuit;
  64
  65     //! The maximum number of elements to be processed in one batch
  66     unsigned int nBatchSize;
  67
  68     /** Internal function that does bulk of the verification work. */
  69     bool Loop(bool fMaster = false)
  70     {
  71         boost::condition_variable& cond = fMaster ? condMaster : condWorker;
  72         std::vector<T> vChecks;
  73         vChecks.reserve(nBatchSize);
  74         unsigned int nNow = 0;
  75         bool fOk = true;
  76         do {
  77             {
  78                 boost::unique_lock<boost::mutex> lock(mutex);
  79                 // first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
  80                 if (nNow) {
  81                     fAllOk &= fOk;
  82                     nTodo -= nNow;
  83                     if (nTodo == 0 && !fMaster)
  84                         // We processed the last element; inform the master it can exit and return the result
  85                         condMaster.notify_one();
  86                 } else {
  87                     // first iteration
  88                     nTotal++;
  89                 }
  90                 // logically, the do loop starts here
  91                 while (queue.empty()) {
  92                     if ((fMaster || fQuit) && nTodo == 0) {
  93                         nTotal--;
  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             for (T& check : vChecks)
 123                 if (fOk)
 124                     fOk = check();
 125             vChecks.clear();
 126         } while (true);
 127     }
 128
 129 public:
 130     //! Mutex to ensure only one concurrent CCheckQueueControl
 131     boost::mutex ControlMutex;
 132
 133     //! Create a new check queue
 134     CCheckQueue(unsigned int nBatchSizeIn) : nIdle(0), nTotal(0), fAllOk(true), nTodo(0), fQuit(false), nBatchSize(nBatchSizeIn) {}
 135
 136     //! Worker thread
 137     void Thread()
 138     {
 139         Loop();
 140     }
 141
 142     //! Wait until execution finishes, and return whether all evaluations were successful.
 143     bool Wait()
 144     {
 145         return Loop(true);
 146     }
 147
 148     //! Add a batch of checks to the queue
 149     void Add(std::vector<T>& vChecks)
 150     {
 151         boost::unique_lock<boost::mutex> lock(mutex);
 152         for (T& check : vChecks) {
 153             queue.push_back(T());
 154             check.swap(queue.back());
 155         }
 156         nTodo += vChecks.size();
 157         if (vChecks.size() == 1)
 158             condWorker.notify_one();
 159         else if (vChecks.size() > 1)
 160             condWorker.notify_all();
 161     }
 162
 163     ~CCheckQueue()
 164     {
 165     }
 166
 167 };
 168
 169 /**
 170  * RAII-style controller object for a CCheckQueue that guarantees the passed
 171  * queue is finished before continuing.
 172  */
 173 template <typename T>
 174 class CCheckQueueControl
 175 {
 176 private:
 177     CCheckQueue<T> * const pqueue;
 178     bool fDone;
 179
 180 public:
 181     CCheckQueueControl() = delete;
 182     CCheckQueueControl(const CCheckQueueControl&) = delete;
 183     CCheckQueueControl& operator=(const CCheckQueueControl&) = delete;
 184     explicit CCheckQueueControl(CCheckQueue<T> * const pqueueIn) : pqueue(pqueueIn), fDone(false)
 185     {
 186         // passed queue is supposed to be unused, or nullptr
 187         if (pqueue != nullptr) {
 188             ENTER_CRITICAL_SECTION(pqueue->ControlMutex);
 189         }
 190     }
 191
 192     bool Wait()
 193     {
 194         if (pqueue == nullptr)
 195             return true;
 196         bool fRet = pqueue->Wait();
 197         fDone = true;
 198         return fRet;
 199     }
 200
 201     void Add(std::vector<T>& vChecks)
 202     {
 203         if (pqueue != nullptr)
 204             pqueue->Add(vChecks);
 205     }
 206
 207     ~CCheckQueueControl()
 208     {
 209         if (!fDone)
 210             Wait();
 211         if (pqueue != nullptr) {
 212             LEAVE_CRITICAL_SECTION(pqueue->ControlMutex);
 213         }
 214     }
 215 };
 216
 217 #endif // BITCOIN_CHECKQUEUE_H