Browse Source
* During block verification (when parallelism is requested), script check actions are stored instead of being executed immediately. * After every processed transactions, its signature actions are pushed to a CScriptCheckQueue, which maintains a queue and some synchronization mechanism. * Two or more threads (if enabled) start processing elements from this queue, * When the block connection code is finished processing transactions, it joins the worker pool until the queue is empty. As cs_main is held the entire time, and all verification must be finished before the block continues processing, this does not reach the best possible performance. It is a less drastic change than some more advanced mechanisms (like doing verification out-of-band entirely, and rolling back blocks when a failure is detected). The -par=N flag controls the number of threads (1-16). 0 means auto, and is the default.miguelfreitas
Pieter Wuille
12 years ago
7 changed files with 281 additions and 6 deletions
@ -0,0 +1,206 @@ |
|||||||
|
// Copyright (c) 2012 The Bitcoin developers
|
||||||
|
// Distributed under the MIT/X11 software license, see the accompanying
|
||||||
|
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||||
|
#ifndef CHECKQUEUE_H |
||||||
|
#define CHECKQUEUE_H |
||||||
|
|
||||||
|
#include <boost/thread/mutex.hpp> |
||||||
|
#include <boost/thread/locks.hpp> |
||||||
|
#include <boost/thread/condition_variable.hpp> |
||||||
|
|
||||||
|
#include <vector> |
||||||
|
#include <algorithm> |
||||||
|
|
||||||
|
template<typename T> class CCheckQueueControl; |
||||||
|
|
||||||
|
/** Queue for verifications that have to be performed.
|
||||||
|
* The verifications are represented by a type T, which must provide an |
||||||
|
* operator(), returning a bool. |
||||||
|
* |
||||||
|
* One thread (the master) is assumed to push batches of verifications |
||||||
|
* onto the queue, where they are processed by N-1 worker threads. When |
||||||
|
* the master is done adding work, it temporarily joins the worker pool |
||||||
|
* as an N'th worker, until all jobs are done. |
||||||
|
*/ |
||||||
|
template<typename T> class CCheckQueue { |
||||||
|
private: |
||||||
|
// Mutex to protect the inner state
|
||||||
|
boost::mutex mutex; |
||||||
|
|
||||||
|
// Worker threads block on this when out of work
|
||||||
|
boost::condition_variable condWorker; |
||||||
|
|
||||||
|
// Master thread blocks on this when out of work
|
||||||
|
boost::condition_variable condMaster; |
||||||
|
|
||||||
|
// Quit method blocks on this until all workers are gone
|
||||||
|
boost::condition_variable condQuit; |
||||||
|
|
||||||
|
// The queue of elements to be processed.
|
||||||
|
// As the order of booleans doesn't matter, it is used as a LIFO (stack)
|
||||||
|
std::vector<T> queue; |
||||||
|
|
||||||
|
// The number of workers (including the master) that are idle.
|
||||||
|
int nIdle; |
||||||
|
|
||||||
|
// The total number of workers (including the master).
|
||||||
|
int nTotal; |
||||||
|
|
||||||
|
// The temporary evaluation result.
|
||||||
|
bool fAllOk; |
||||||
|
|
||||||
|
// Number of verifications that haven't completed yet.
|
||||||
|
// This includes elements that are not anymore in queue, but still in
|
||||||
|
// worker's own batches.
|
||||||
|
unsigned int nTodo; |
||||||
|
|
||||||
|
// Whether we're shutting down.
|
||||||
|
bool fQuit; |
||||||
|
|
||||||
|
// The maximum number of elements to be processed in one batch
|
||||||
|
unsigned int nBatchSize; |
||||||
|
|
||||||
|
// Internal function that does bulk of the verification work.
|
||||||
|
bool Loop(bool fMaster = false) { |
||||||
|
boost::condition_variable &cond = fMaster ? condMaster : condWorker; |
||||||
|
std::vector<T> vChecks; |
||||||
|
vChecks.reserve(nBatchSize); |
||||||
|
unsigned int nNow = 0; |
||||||
|
bool fOk = true; |
||||||
|
do { |
||||||
|
{ |
||||||
|
boost::unique_lock<boost::mutex> lock(mutex); |
||||||
|
// first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
|
||||||
|
if (nNow) { |
||||||
|
fAllOk &= fOk; |
||||||
|
nTodo -= nNow; |
||||||
|
if (nTodo == 0 && !fMaster) |
||||||
|
// We processed the last element; inform the master he can exit and return the result
|
||||||
|
condMaster.notify_one(); |
||||||
|
} else { |
||||||
|
// first iteration
|
||||||
|
nTotal++; |
||||||
|
} |
||||||
|
// logically, the do loop starts here
|
||||||
|
while (queue.empty()) { |
||||||
|
if ((fMaster || fQuit) && nTodo == 0) { |
||||||
|
nTotal--; |
||||||
|
if (nTotal==0) |
||||||
|
condQuit.notify_one(); |
||||||
|
bool fRet = fAllOk; |
||||||
|
// reset the status for new work later
|
||||||
|
if (fMaster) |
||||||
|
fAllOk = true; |
||||||
|
// return the current status
|
||||||
|
return fRet; |
||||||
|
} |
||||||
|
nIdle++; |
||||||
|
cond.wait(lock); // wait
|
||||||
|
nIdle--; |
||||||
|
} |
||||||
|
// Decide how many work units to process now.
|
||||||
|
// * Do not try to do everything at once, but aim for increasingly smaller batches so
|
||||||
|
// all workers finish approximately simultaneously.
|
||||||
|
// * Try to account for idle jobs which will instantly start helping.
|
||||||
|
// * Don't do batches smaller than 1 (duh), or larger than nBatchSize.
|
||||||
|
nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1))); |
||||||
|
vChecks.resize(nNow); |
||||||
|
for (unsigned int i = 0; i < nNow; i++) { |
||||||
|
// We want the lock on the mutex to be as short as possible, so swap jobs from the global
|
||||||
|
// queue to the local batch vector instead of copying.
|
||||||
|
vChecks[i].swap(queue.back()); |
||||||
|
queue.pop_back(); |
||||||
|
} |
||||||
|
// Check whether we need to do work at all
|
||||||
|
fOk = fAllOk; |
||||||
|
} |
||||||
|
// execute work
|
||||||
|
BOOST_FOREACH(T &check, vChecks) |
||||||
|
if (fOk) |
||||||
|
fOk = check(); |
||||||
|
vChecks.clear(); |
||||||
|
} while(true); |
||||||
|
} |
||||||
|
|
||||||
|
public: |
||||||
|
// Create a new check queue
|
||||||
|
CCheckQueue(unsigned int nBatchSizeIn) : |
||||||
|
nIdle(0), nTotal(0), fAllOk(true), nTodo(0), fQuit(false), nBatchSize(nBatchSizeIn) {} |
||||||
|
|
||||||
|
// Worker thread
|
||||||
|
void Thread() { |
||||||
|
Loop(); |
||||||
|
} |
||||||
|
|
||||||
|
// Wait until execution finishes, and return whether all evaluations where succesful.
|
||||||
|
bool Wait() { |
||||||
|
return Loop(true); |
||||||
|
} |
||||||
|
|
||||||
|
// Add a batch of checks to the queue
|
||||||
|
void Add(std::vector<T> &vChecks) { |
||||||
|
boost::unique_lock<boost::mutex> lock(mutex); |
||||||
|
BOOST_FOREACH(T &check, vChecks) { |
||||||
|
queue.push_back(T()); |
||||||
|
check.swap(queue.back()); |
||||||
|
} |
||||||
|
nTodo += vChecks.size(); |
||||||
|
if (vChecks.size() == 1) |
||||||
|
condWorker.notify_one(); |
||||||
|
else if (vChecks.size() > 1) |
||||||
|
condWorker.notify_all(); |
||||||
|
} |
||||||
|
|
||||||
|
// Shut the queue down
|
||||||
|
void Quit() { |
||||||
|
boost::unique_lock<boost::mutex> lock(mutex); |
||||||
|
fQuit = true; |
||||||
|
// No need to wake the master, as he will quit automatically when all jobs are
|
||||||
|
// done.
|
||||||
|
condWorker.notify_all(); |
||||||
|
|
||||||
|
while (nTotal > 0) |
||||||
|
condQuit.wait(lock); |
||||||
|
} |
||||||
|
|
||||||
|
friend class CCheckQueueControl<T>; |
||||||
|
}; |
||||||
|
|
||||||
|
/** RAII-style controller object for a CCheckQueue that guarantees the passed
|
||||||
|
* queue is finished before continuing. |
||||||
|
*/ |
||||||
|
template<typename T> class CCheckQueueControl { |
||||||
|
private: |
||||||
|
CCheckQueue<T> *pqueue; |
||||||
|
bool fDone; |
||||||
|
|
||||||
|
public: |
||||||
|
CCheckQueueControl(CCheckQueue<T> *pqueueIn) : pqueue(pqueueIn), fDone(false) { |
||||||
|
// passed queue is supposed to be unused, or NULL
|
||||||
|
if (pqueue != NULL) { |
||||||
|
assert(pqueue->nTotal == pqueue->nIdle); |
||||||
|
assert(pqueue->nTodo == 0); |
||||||
|
assert(pqueue->fAllOk == true); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
bool Wait() { |
||||||
|
if (pqueue == NULL) |
||||||
|
return true; |
||||||
|
bool fRet = pqueue->Wait(); |
||||||
|
fDone = true; |
||||||
|
return fRet; |
||||||
|
} |
||||||
|
|
||||||
|
void Add(std::vector<T> &vChecks) { |
||||||
|
if (pqueue != NULL) |
||||||
|
pqueue->Add(vChecks); |
||||||
|
} |
||||||
|
|
||||||
|
~CCheckQueueControl() { |
||||||
|
if (!fDone) |
||||||
|
Wait(); |
||||||
|
} |
||||||
|
}; |
||||||
|
|
||||||
|
#endif |
||||||
Loading…
Reference in new issue