refactor: remove boost::thread_group usage

This commit is contained in:
fanquake
2021-01-27 15:04:34 +08:00
parent c8b83510f4
commit dc8be12510
12 changed files with 44 additions and 39 deletions

View File

@@ -10,7 +10,6 @@
#include <util/time.h>
#include <boost/test/unit_test.hpp>
#include <boost/thread/thread.hpp>
#include <atomic>
#include <condition_variable>
@@ -363,11 +362,11 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
{
auto queue = MakeUnique<Standard_Queue>(QUEUE_BATCH_SIZE);
{
boost::thread_group tg;
std::vector<std::thread> tg;
std::atomic<int> nThreads {0};
std::atomic<int> fails {0};
for (size_t i = 0; i < 3; ++i) {
tg.create_thread(
tg.emplace_back(
[&]{
CCheckQueueControl<FakeCheck> control(queue.get());
// While sleeping, no other thread should execute to this point
@@ -376,11 +375,13 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
fails += observed != nThreads;
});
}
tg.join_all();
for (auto& thread: tg) {
if (thread.joinable()) thread.join();
}
BOOST_REQUIRE_EQUAL(fails, 0);
}
{
boost::thread_group tg;
std::vector<std::thread> tg;
std::mutex m;
std::condition_variable cv;
bool has_lock{false};
@@ -389,7 +390,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
bool done_ack{false};
{
std::unique_lock<std::mutex> l(m);
tg.create_thread([&]{
tg.emplace_back([&]{
CCheckQueueControl<FakeCheck> control(queue.get());
std::unique_lock<std::mutex> ll(m);
has_lock = true;
@@ -415,7 +416,9 @@ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks)
cv.notify_one();
BOOST_REQUIRE(!fails);
}
tg.join_all();
for (auto& thread: tg) {
if (thread.joinable()) thread.join();
}
}
}
BOOST_AUTO_TEST_SUITE_END()

View File

@@ -2,6 +2,8 @@
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
#include <boost/test/unit_test.hpp>
#include <boost/thread/lock_types.hpp>
#include <boost/thread/shared_mutex.hpp>
#include <cuckoocache.h>
#include <deque>
#include <random.h>

View File

@@ -7,10 +7,11 @@
#include <util/time.h>
#include <boost/test/unit_test.hpp>
#include <boost/thread/thread.hpp>
#include <functional>
#include <mutex>
#include <thread>
#include <vector>
BOOST_AUTO_TEST_SUITE(scheduler_tests)
@@ -69,16 +70,16 @@ BOOST_AUTO_TEST_CASE(manythreads)
BOOST_CHECK(last > now);
// As soon as these are created they will start running and servicing the queue
boost::thread_group microThreads;
std::vector<std::thread> microThreads;
for (int i = 0; i < 5; i++)
microThreads.create_thread(std::bind(&CScheduler::serviceQueue, &microTasks));
microThreads.emplace_back(std::bind(&CScheduler::serviceQueue, &microTasks));
UninterruptibleSleep(std::chrono::microseconds{600});
now = std::chrono::system_clock::now();
// More threads and more tasks:
for (int i = 0; i < 5; i++)
microThreads.create_thread(std::bind(&CScheduler::serviceQueue, &microTasks));
microThreads.emplace_back(std::bind(&CScheduler::serviceQueue, &microTasks));
for (int i = 0; i < 100; i++) {
std::chrono::system_clock::time_point t = now + std::chrono::microseconds(randomMsec(rng));
std::chrono::system_clock::time_point tReschedule = now + std::chrono::microseconds(500 + randomMsec(rng));
@@ -91,7 +92,10 @@ BOOST_AUTO_TEST_CASE(manythreads)
// Drain the task queue then exit threads
microTasks.StopWhenDrained();
microThreads.join_all(); // ... wait until all the threads are done
// wait until all the threads are done
for (auto& thread: microThreads) {
if (thread.joinable()) thread.join();
}
int counterSum = 0;
for (int i = 0; i < 10; i++) {
@@ -131,9 +135,9 @@ BOOST_AUTO_TEST_CASE(singlethreadedscheduler_ordered)
// if the queues only permit execution of one task at once then
// the extra threads should effectively be doing nothing
// if they don't we'll get out of order behaviour
boost::thread_group threads;
std::vector<std::thread> threads;
for (int i = 0; i < 5; ++i) {
threads.create_thread(std::bind(&CScheduler::serviceQueue, &scheduler));
threads.emplace_back(std::bind(&CScheduler::serviceQueue, &scheduler));
}
// these are not atomic, if SinglethreadedSchedulerClient prevents
@@ -157,7 +161,9 @@ BOOST_AUTO_TEST_CASE(singlethreadedscheduler_ordered)
// finish up
scheduler.StopWhenDrained();
threads.join_all();
for (auto& thread: threads) {
if (thread.joinable()) thread.join();
}
BOOST_CHECK_EQUAL(counter1, 100);
BOOST_CHECK_EQUAL(counter2, 100);

View File

@@ -131,7 +131,7 @@ ChainTestingSetup::ChainTestingSetup(const std::string& chainName, const std::ve
// We have to run a scheduler thread to prevent ActivateBestChain
// from blocking due to queue overrun.
m_node.scheduler = MakeUnique<CScheduler>();
threadGroup.create_thread([&] { TraceThread("scheduler", [&] { m_node.scheduler->serviceQueue(); }); });
m_node.scheduler->m_service_thread = std::thread([&] { TraceThread("scheduler", [&] { m_node.scheduler->serviceQueue(); }); });
GetMainSignals().RegisterBackgroundSignalScheduler(*m_node.scheduler);
pblocktree.reset(new CBlockTreeDB(1 << 20, true));
@@ -150,8 +150,6 @@ ChainTestingSetup::ChainTestingSetup(const std::string& chainName, const std::ve
ChainTestingSetup::~ChainTestingSetup()
{
if (m_node.scheduler) m_node.scheduler->stop();
threadGroup.interrupt_all();
threadGroup.join_all();
StopScriptCheckWorkerThreads();
GetMainSignals().FlushBackgroundCallbacks();
GetMainSignals().UnregisterBackgroundSignalScheduler();

View File

@@ -17,8 +17,7 @@
#include <util/string.h>
#include <type_traits>
#include <boost/thread/thread.hpp>
#include <vector>
/** This is connected to the logger. Can be used to redirect logs to any other log */
extern const std::function<void(const std::string&)> G_TEST_LOG_FUN;
@@ -88,7 +87,6 @@ private:
* initialization behaviour.
*/
struct ChainTestingSetup : public BasicTestingSetup {
boost::thread_group threadGroup;
explicit ChainTestingSetup(const std::string& chainName = CBaseChainParams::MAIN, const std::vector<const char*>& extra_args = {});
~ChainTestingSetup();