aboutsummaryrefslogtreecommitdiff
path: root/src/checkqueue.h
diff options
context:
space:
mode:
authorHennadii Stepanov <32963518+hebasto@users.noreply.github.com>2020-08-21 09:25:07 +0300
committerHennadii Stepanov <32963518+hebasto@users.noreply.github.com>2020-09-24 06:55:34 +0300
commitbb6fcc75d1ec94b733d1477c816351c50be5faf9 (patch)
treea9a00f07504cb9f41f521ce2f46662c6404b5e53 /src/checkqueue.h
parent6784ac471bb32b6bb8e2de60986f123eb4990706 (diff)
refactor: Drop boost::thread stuff in CCheckQueue
Diffstat (limited to 'src/checkqueue.h')
-rw-r--r--src/checkqueue.h53
1 files changed, 23 insertions, 30 deletions
diff --git a/src/checkqueue.h b/src/checkqueue.h
index 83b371fd42..4ceeb3600a 100644
--- a/src/checkqueue.h
+++ b/src/checkqueue.h
@@ -12,9 +12,6 @@
#include <algorithm>
#include <vector>
-#include <boost/thread/condition_variable.hpp>
-#include <boost/thread/mutex.hpp>
-
template <typename T>
class CCheckQueueControl;
@@ -33,58 +30,58 @@ class CCheckQueue
{
private:
//! Mutex to protect the inner state
- boost::mutex mutex;
+ Mutex m_mutex;
//! 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
- boost::condition_variable condMaster;
+ std::condition_variable m_master_cv;
//! 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;
+ std::vector<T> queue GUARDED_BY(m_mutex);
//! 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).
- int nTotal{0};
+ int nTotal GUARDED_BY(m_mutex){0};
//! The temporary evaluation result.
- bool fAllOk{true};
+ bool fAllOk GUARDED_BY(m_mutex){true};
/**
* Number of verifications that haven't completed yet.
* This includes elements that are no longer queued, but still in the
* 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
const unsigned int nBatchSize;
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. */
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;
vChecks.reserve(nBatchSize);
unsigned int nNow = 0;
bool fOk = true;
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)
if (nNow) {
fAllOk &= fOk;
nTodo -= nNow;
if (nTodo == 0 && !fMaster)
// We processed the last element; inform the master it can exit and return the result
- condMaster.notify_one();
+ m_master_cv.notify_one();
} else {
// first iteration
nTotal++;
@@ -115,7 +112,7 @@ private:
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
+ // 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.
vChecks[i].swap(queue.back());
queue.pop_back();
@@ -133,7 +130,7 @@ private:
public:
//! Mutex to ensure only one concurrent CCheckQueueControl
- boost::mutex ControlMutex;
+ Mutex m_control_mutex;
//! Create a new check queue
explicit CCheckQueue(unsigned int nBatchSizeIn)
@@ -145,7 +142,7 @@ public:
void StartWorkerThreads(const int threads_num)
{
{
- boost::unique_lock<boost::mutex> lock(mutex);
+ LOCK(m_mutex);
nIdle = 0;
nTotal = 0;
fAllOk = true;
@@ -168,32 +165,28 @@ public:
//! Add a batch of checks to the queue
void Add(std::vector<T>& vChecks)
{
- boost::unique_lock<boost::mutex> lock(mutex);
+ LOCK(m_mutex);
for (T& check : vChecks) {
queue.push_back(T());
check.swap(queue.back());
}
nTodo += vChecks.size();
if (vChecks.size() == 1)
- condWorker.notify_one();
+ m_worker_cv.notify_one();
else if (vChecks.size() > 1)
- condWorker.notify_all();
+ m_worker_cv.notify_all();
}
//! Stop all of the worker threads.
void StopWorkerThreads()
{
- {
- boost::unique_lock<boost::mutex> lock(mutex);
- m_request_stop = true;
- }
- condWorker.notify_all();
+ WITH_LOCK(m_mutex, m_request_stop = true);
+ m_worker_cv.notify_all();
for (std::thread& t : m_worker_threads) {
t.join();
}
m_worker_threads.clear();
- boost::unique_lock<boost::mutex> lock(mutex);
- m_request_stop = false;
+ WITH_LOCK(m_mutex, m_request_stop = false);
}
~CCheckQueue()
@@ -222,7 +215,7 @@ public:
{
// passed queue is supposed to be unused, or nullptr
if (pqueue != nullptr) {
- ENTER_CRITICAL_SECTION(pqueue->ControlMutex);
+ ENTER_CRITICAL_SECTION(pqueue->m_control_mutex);
}
}
@@ -246,7 +239,7 @@ public:
if (!fDone)
Wait();
if (pqueue != nullptr) {
- LEAVE_CRITICAL_SECTION(pqueue->ControlMutex);
+ LEAVE_CRITICAL_SECTION(pqueue->m_control_mutex);
}
}
};