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