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