Skip to content

Commit

Permalink
refactor: Drop boost::thread stuff in CCheckQueue
Browse files Browse the repository at this point in the history
Summary:
This is a backport of [[bitcoin/bitcoin#18710 | core#18710]] [4/4]
bitcoin/bitcoin@bb6fcc7

Depends on D10987

Test Plan:
With TSAN: `ninja && ninja check check-functional`

```
cmake .. -GNinja -DCMAKE_BUILD_TYPE=Debug -DCMAKE_C_COMPILER=clang -DCMAKE_CXX_COMPILER=clang++
ninja all check-all
```

Reviewers: #bitcoin_abc, Fabien

Reviewed By: #bitcoin_abc, Fabien

Subscribers: Fabien

Differential Revision: https://reviews.bitcoinabc.org/D10988
  • Loading branch information
hebasto authored and PiRK committed Feb 4, 2022
1 parent 21d559e commit e376425
Show file tree
Hide file tree
Showing 3 changed files with 27 additions and 35 deletions.
57 changes: 25 additions & 32 deletions src/checkqueue.h
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,6 @@
#include <algorithm>
#include <vector>

#include <boost/thread/condition_variable.hpp>
#include <boost/thread/mutex.hpp>

template <typename T> class CCheckQueueControl;

/**
Expand All @@ -30,50 +27,50 @@ template <typename T> class CCheckQueueControl;
template <typename T> 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) {
Expand All @@ -82,7 +79,7 @@ template <typename T> class CCheckQueue {
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
Expand Down Expand Up @@ -119,9 +116,9 @@ template <typename T> class CCheckQueue {
(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.
// 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();
}
Expand All @@ -140,7 +137,7 @@ template <typename T> class CCheckQueue {

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)
Expand All @@ -149,7 +146,7 @@ template <typename T> class CCheckQueue {
//! Create a pool of new worker threads.
void StartWorkerThreads(const int threads_num) {
{
boost::unique_lock<boost::mutex> lock(mutex);
LOCK(m_mutex);
nIdle = 0;
nTotal = 0;
fAllOk = true;
Expand All @@ -169,32 +166,28 @@ template <typename T> class CCheckQueue {

//! 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() { assert(m_worker_threads.empty()); }
Expand All @@ -217,7 +210,7 @@ template <typename T> class CCheckQueueControl {
: pqueue(pqueueIn), fDone(false) {
// passed queue is supposed to be unused, or nullptr
if (pqueue != nullptr) {
ENTER_CRITICAL_SECTION(pqueue->ControlMutex);
ENTER_CRITICAL_SECTION(pqueue->m_control_mutex);
}
}

Expand All @@ -241,7 +234,7 @@ template <typename T> class CCheckQueueControl {
Wait();
}
if (pqueue != nullptr) {
LEAVE_CRITICAL_SECTION(pqueue->ControlMutex);
LEAVE_CRITICAL_SECTION(pqueue->m_control_mutex);
}
}
};
Expand Down
4 changes: 2 additions & 2 deletions src/test/checkqueue_tests.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -321,7 +321,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup) {
}
// Try to get control of the queue a bunch of times
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)
Expand Down Expand Up @@ -381,7 +381,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks) {
cv.wait(l, [&]() { return has_lock; });
bool fails = false;
for (auto x = 0; x < 100 && !fails; ++x) {
fails = queue->ControlMutex.try_lock();
fails = queue->m_control_mutex.try_lock();
}
has_tried = true;
cv.notify_one();
Expand Down
1 change: 0 additions & 1 deletion test/lint/lint-boost-dependencies.sh
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@ EXPECTED_BOOST_INCLUDES=(
boost/signals2/signal.hpp
boost/test/unit_test.hpp
boost/thread/condition_variable.hpp
boost/thread/mutex.hpp
boost/thread/shared_mutex.hpp
boost/thread/thread.hpp
boost/variant.hpp
Expand Down

0 comments on commit e376425

Please sign in to comment.