mirror of
https://github.com/bitcoin/bitcoin.git
synced 2025-01-25 10:43:19 -03:00
refactor: Drop boost::thread stuff in CCheckQueue
This commit is contained in:
parent
6784ac471b
commit
bb6fcc75d1
3 changed files with 25 additions and 33 deletions
|
@ -12,9 +12,6 @@
|
||||||
#include <algorithm>
|
#include <algorithm>
|
||||||
#include <vector>
|
#include <vector>
|
||||||
|
|
||||||
#include <boost/thread/condition_variable.hpp>
|
|
||||||
#include <boost/thread/mutex.hpp>
|
|
||||||
|
|
||||||
template <typename T>
|
template <typename T>
|
||||||
class CCheckQueueControl;
|
class CCheckQueueControl;
|
||||||
|
|
||||||
|
@ -33,58 +30,58 @@ class CCheckQueue
|
||||||
{
|
{
|
||||||
private:
|
private:
|
||||||
//! Mutex to protect the inner state
|
//! Mutex to protect the inner state
|
||||||
boost::mutex mutex;
|
Mutex m_mutex;
|
||||||
|
|
||||||
//! Worker threads block on this when out of work
|
//! Worker threads block on this when out of work
|
||||||
boost::condition_variable condWorker;
|
std::condition_variable m_worker_cv;
|
||||||
|
|
||||||
//! Master thread blocks on this when out of work
|
//! Master thread blocks on this when out of work
|
||||||
boost::condition_variable condMaster;
|
std::condition_variable m_master_cv;
|
||||||
|
|
||||||
//! The queue of elements to be processed.
|
//! The queue of elements to be processed.
|
||||||
//! As the order of booleans doesn't matter, it is used as a LIFO (stack)
|
//! As the order of booleans doesn't matter, it is used as a LIFO (stack)
|
||||||
std::vector<T> queue;
|
std::vector<T> queue GUARDED_BY(m_mutex);
|
||||||
|
|
||||||
//! The number of workers (including the master) that are idle.
|
//! The number of workers (including the master) that are idle.
|
||||||
int nIdle{0};
|
int nIdle GUARDED_BY(m_mutex){0};
|
||||||
|
|
||||||
//! The total number of workers (including the master).
|
//! The total number of workers (including the master).
|
||||||
int nTotal{0};
|
int nTotal GUARDED_BY(m_mutex){0};
|
||||||
|
|
||||||
//! The temporary evaluation result.
|
//! The temporary evaluation result.
|
||||||
bool fAllOk{true};
|
bool fAllOk GUARDED_BY(m_mutex){true};
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Number of verifications that haven't completed yet.
|
* Number of verifications that haven't completed yet.
|
||||||
* This includes elements that are no longer queued, but still in the
|
* This includes elements that are no longer queued, but still in the
|
||||||
* worker's own batches.
|
* worker's own batches.
|
||||||
*/
|
*/
|
||||||
unsigned int nTodo{0};
|
unsigned int nTodo GUARDED_BY(m_mutex){0};
|
||||||
|
|
||||||
//! The maximum number of elements to be processed in one batch
|
//! The maximum number of elements to be processed in one batch
|
||||||
const unsigned int nBatchSize;
|
const unsigned int nBatchSize;
|
||||||
|
|
||||||
std::vector<std::thread> m_worker_threads;
|
std::vector<std::thread> m_worker_threads;
|
||||||
bool m_request_stop{false};
|
bool m_request_stop GUARDED_BY(m_mutex){false};
|
||||||
|
|
||||||
/** Internal function that does bulk of the verification work. */
|
/** Internal function that does bulk of the verification work. */
|
||||||
bool Loop(bool fMaster)
|
bool Loop(bool fMaster)
|
||||||
{
|
{
|
||||||
boost::condition_variable& cond = fMaster ? condMaster : condWorker;
|
std::condition_variable& cond = fMaster ? m_master_cv : m_worker_cv;
|
||||||
std::vector<T> vChecks;
|
std::vector<T> vChecks;
|
||||||
vChecks.reserve(nBatchSize);
|
vChecks.reserve(nBatchSize);
|
||||||
unsigned int nNow = 0;
|
unsigned int nNow = 0;
|
||||||
bool fOk = true;
|
bool fOk = true;
|
||||||
do {
|
do {
|
||||||
{
|
{
|
||||||
boost::unique_lock<boost::mutex> lock(mutex);
|
WAIT_LOCK(m_mutex, lock);
|
||||||
// first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
|
// first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
|
||||||
if (nNow) {
|
if (nNow) {
|
||||||
fAllOk &= fOk;
|
fAllOk &= fOk;
|
||||||
nTodo -= nNow;
|
nTodo -= nNow;
|
||||||
if (nTodo == 0 && !fMaster)
|
if (nTodo == 0 && !fMaster)
|
||||||
// We processed the last element; inform the master it can exit and return the result
|
// We processed the last element; inform the master it can exit and return the result
|
||||||
condMaster.notify_one();
|
m_master_cv.notify_one();
|
||||||
} else {
|
} else {
|
||||||
// first iteration
|
// first iteration
|
||||||
nTotal++;
|
nTotal++;
|
||||||
|
@ -115,7 +112,7 @@ private:
|
||||||
nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1)));
|
nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1)));
|
||||||
vChecks.resize(nNow);
|
vChecks.resize(nNow);
|
||||||
for (unsigned int i = 0; i < nNow; i++) {
|
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
|
// We want the lock on the m_mutex to be as short as possible, so swap jobs from the global
|
||||||
// queue to the local batch vector instead of copying.
|
// queue to the local batch vector instead of copying.
|
||||||
vChecks[i].swap(queue.back());
|
vChecks[i].swap(queue.back());
|
||||||
queue.pop_back();
|
queue.pop_back();
|
||||||
|
@ -133,7 +130,7 @@ private:
|
||||||
|
|
||||||
public:
|
public:
|
||||||
//! Mutex to ensure only one concurrent CCheckQueueControl
|
//! Mutex to ensure only one concurrent CCheckQueueControl
|
||||||
boost::mutex ControlMutex;
|
Mutex m_control_mutex;
|
||||||
|
|
||||||
//! Create a new check queue
|
//! Create a new check queue
|
||||||
explicit CCheckQueue(unsigned int nBatchSizeIn)
|
explicit CCheckQueue(unsigned int nBatchSizeIn)
|
||||||
|
@ -145,7 +142,7 @@ public:
|
||||||
void StartWorkerThreads(const int threads_num)
|
void StartWorkerThreads(const int threads_num)
|
||||||
{
|
{
|
||||||
{
|
{
|
||||||
boost::unique_lock<boost::mutex> lock(mutex);
|
LOCK(m_mutex);
|
||||||
nIdle = 0;
|
nIdle = 0;
|
||||||
nTotal = 0;
|
nTotal = 0;
|
||||||
fAllOk = true;
|
fAllOk = true;
|
||||||
|
@ -168,32 +165,28 @@ public:
|
||||||
//! Add a batch of checks to the queue
|
//! Add a batch of checks to the queue
|
||||||
void Add(std::vector<T>& vChecks)
|
void Add(std::vector<T>& vChecks)
|
||||||
{
|
{
|
||||||
boost::unique_lock<boost::mutex> lock(mutex);
|
LOCK(m_mutex);
|
||||||
for (T& check : vChecks) {
|
for (T& check : vChecks) {
|
||||||
queue.push_back(T());
|
queue.push_back(T());
|
||||||
check.swap(queue.back());
|
check.swap(queue.back());
|
||||||
}
|
}
|
||||||
nTodo += vChecks.size();
|
nTodo += vChecks.size();
|
||||||
if (vChecks.size() == 1)
|
if (vChecks.size() == 1)
|
||||||
condWorker.notify_one();
|
m_worker_cv.notify_one();
|
||||||
else if (vChecks.size() > 1)
|
else if (vChecks.size() > 1)
|
||||||
condWorker.notify_all();
|
m_worker_cv.notify_all();
|
||||||
}
|
}
|
||||||
|
|
||||||
//! Stop all of the worker threads.
|
//! Stop all of the worker threads.
|
||||||
void StopWorkerThreads()
|
void StopWorkerThreads()
|
||||||
{
|
{
|
||||||
{
|
WITH_LOCK(m_mutex, m_request_stop = true);
|
||||||
boost::unique_lock<boost::mutex> lock(mutex);
|
m_worker_cv.notify_all();
|
||||||
m_request_stop = true;
|
|
||||||
}
|
|
||||||
condWorker.notify_all();
|
|
||||||
for (std::thread& t : m_worker_threads) {
|
for (std::thread& t : m_worker_threads) {
|
||||||
t.join();
|
t.join();
|
||||||
}
|
}
|
||||||
m_worker_threads.clear();
|
m_worker_threads.clear();
|
||||||
boost::unique_lock<boost::mutex> lock(mutex);
|
WITH_LOCK(m_mutex, m_request_stop = false);
|
||||||
m_request_stop = false;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
~CCheckQueue()
|
~CCheckQueue()
|
||||||
|
@ -222,7 +215,7 @@ public:
|
||||||
{
|
{
|
||||||
// passed queue is supposed to be unused, or nullptr
|
// passed queue is supposed to be unused, or nullptr
|
||||||
if (pqueue != nullptr) {
|
if (pqueue != nullptr) {
|
||||||
ENTER_CRITICAL_SECTION(pqueue->ControlMutex);
|
ENTER_CRITICAL_SECTION(pqueue->m_control_mutex);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -246,7 +239,7 @@ public:
|
||||||
if (!fDone)
|
if (!fDone)
|
||||||
Wait();
|
Wait();
|
||||||
if (pqueue != nullptr) {
|
if (pqueue != nullptr) {
|
||||||
LEAVE_CRITICAL_SECTION(pqueue->ControlMutex);
|
LEAVE_CRITICAL_SECTION(pqueue->m_control_mutex);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
|
@ -342,7 +342,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup)
|
||||||
}
|
}
|
||||||
// Try to get control of the queue a bunch of times
|
// Try to get control of the queue a bunch of times
|
||||||
for (auto x = 0; x < 100 && !fails; ++x) {
|
for (auto x = 0; x < 100 && !fails; ++x) {
|
||||||
fails = queue->ControlMutex.try_lock();
|
fails = queue->m_control_mutex.try_lock();
|
||||||
}
|
}
|
||||||
{
|
{
|
||||||
// Unfreeze (we need lock n case of spurious wakeup)
|
// Unfreeze (we need lock n case of spurious wakeup)
|
||||||
|
@ -405,7 +405,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
|
||||||
cv.wait(l, [&](){return has_lock;});
|
cv.wait(l, [&](){return has_lock;});
|
||||||
bool fails = false;
|
bool fails = false;
|
||||||
for (auto x = 0; x < 100 && !fails; ++x) {
|
for (auto x = 0; x < 100 && !fails; ++x) {
|
||||||
fails = queue->ControlMutex.try_lock();
|
fails = queue->m_control_mutex.try_lock();
|
||||||
}
|
}
|
||||||
has_tried = true;
|
has_tried = true;
|
||||||
cv.notify_one();
|
cv.notify_one();
|
||||||
|
|
|
@ -69,7 +69,6 @@ EXPECTED_BOOST_INCLUDES=(
|
||||||
boost/signals2/signal.hpp
|
boost/signals2/signal.hpp
|
||||||
boost/test/unit_test.hpp
|
boost/test/unit_test.hpp
|
||||||
boost/thread/condition_variable.hpp
|
boost/thread/condition_variable.hpp
|
||||||
boost/thread/mutex.hpp
|
|
||||||
boost/thread/shared_mutex.hpp
|
boost/thread/shared_mutex.hpp
|
||||||
boost/thread/thread.hpp
|
boost/thread/thread.hpp
|
||||||
boost/variant.hpp
|
boost/variant.hpp
|
||||||
|
|
Loading…
Add table
Reference in a new issue