From 9cf89f7a5b81197e38f58b24be0793b28fe41477 Mon Sep 17 00:00:00 2001 From: Hennadii Stepanov <32963518+hebasto@users.noreply.github.com> Date: Fri, 7 Jul 2023 10:40:13 +0100 Subject: [PATCH] refactor: Make `CCheckQueue` constructor start worker threads --- src/bench/checkqueue.cpp | 5 +++-- src/checkqueue.h | 19 ++++--------------- src/test/checkqueue_tests.cpp | 23 +++++++---------------- src/test/fuzz/checkqueue.cpp | 4 ++-- src/test/transaction_tests.cpp | 4 +--- src/validation.cpp | 3 +-- 6 files changed, 18 insertions(+), 40 deletions(-) diff --git a/src/bench/checkqueue.cpp b/src/bench/checkqueue.cpp index c7e1ad4f838..114dd9d39c8 100644 --- a/src/bench/checkqueue.cpp +++ b/src/bench/checkqueue.cpp @@ -37,10 +37,11 @@ static void CCheckQueueSpeedPrevectorJob(benchmark::Bench& bench) return true; } }; - CCheckQueue queue {QUEUE_BATCH_SIZE}; + // The main thread should be counted to prevent thread oversubscription, and // to decrease the variance of benchmark results. - queue.StartWorkerThreads(GetNumCores() - 1); + int worker_threads_num{GetNumCores() - 1}; + CCheckQueue queue{QUEUE_BATCH_SIZE, worker_threads_num}; // create all the data once, then submit copies in the benchmark. FastRandomContext insecure_rand(true); diff --git a/src/checkqueue.h b/src/checkqueue.h index a89be2cfd53..f8b288c10a6 100644 --- a/src/checkqueue.h +++ b/src/checkqueue.h @@ -130,22 +130,11 @@ public: Mutex m_control_mutex; //! Create a new check queue - explicit CCheckQueue(unsigned int nBatchSizeIn) - : nBatchSize(nBatchSizeIn) + explicit CCheckQueue(unsigned int batch_size, int worker_threads_num) + : nBatchSize(batch_size) { - } - - //! Create a pool of new worker threads. - void StartWorkerThreads(const int threads_num) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) - { - { - LOCK(m_mutex); - nIdle = 0; - nTotal = 0; - fAllOk = true; - } - assert(m_worker_threads.empty()); - for (int n = 0; n < threads_num; ++n) { + m_worker_threads.reserve(worker_threads_num); + for (int n = 0; n < worker_threads_num; ++n) { m_worker_threads.emplace_back([this, n]() { util::ThreadRename(strprintf("scriptch.%i", n)); Loop(false /* worker thread */); diff --git a/src/test/checkqueue_tests.cpp b/src/test/checkqueue_tests.cpp index 9c2a5d1ae66..023a5e8e70d 100644 --- a/src/test/checkqueue_tests.cpp +++ b/src/test/checkqueue_tests.cpp @@ -158,8 +158,7 @@ typedef CCheckQueue FrozenCleanup_Queue; */ static void Correct_Queue_range(std::vector range) { - auto small_queue = std::make_unique(QUEUE_BATCH_SIZE); - small_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS); + auto small_queue = std::make_unique(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS); // Make vChecks here to save on malloc (this test can be slow...) std::vector vChecks; vChecks.reserve(9); @@ -217,9 +216,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Correct_Random) /** Test that failing checks are caught */ BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure) { - auto fail_queue = std::make_unique(QUEUE_BATCH_SIZE); - fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS); - + auto fail_queue = std::make_unique(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS); for (size_t i = 0; i < 1001; ++i) { CCheckQueueControl control(fail_queue.get()); size_t remaining = i; @@ -244,9 +241,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure) // future blocks, ie, the bad state is cleared. BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure) { - auto fail_queue = std::make_unique(QUEUE_BATCH_SIZE); - fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS); - + auto fail_queue = std::make_unique(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS); for (auto times = 0; times < 10; ++times) { for (const bool end_fails : {true, false}) { CCheckQueueControl control(fail_queue.get()); @@ -267,9 +262,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure) // more than once as well BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck) { - auto queue = std::make_unique(QUEUE_BATCH_SIZE); - queue->StartWorkerThreads(SCRIPT_CHECK_THREADS); - + auto queue = std::make_unique(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS); size_t COUNT = 100000; size_t total = COUNT; { @@ -301,8 +294,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck) // time could leave the data hanging across a sequence of blocks. BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory) { - auto queue = std::make_unique(QUEUE_BATCH_SIZE); - queue->StartWorkerThreads(SCRIPT_CHECK_THREADS); + auto queue = std::make_unique(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS); for (size_t i = 0; i < 1000; ++i) { size_t total = i; { @@ -327,9 +319,8 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory) // have been destructed BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup) { - auto queue = std::make_unique(QUEUE_BATCH_SIZE); + auto queue = std::make_unique(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS); bool fails = false; - queue->StartWorkerThreads(SCRIPT_CHECK_THREADS); std::thread t0([&]() { CCheckQueueControl control(queue.get()); std::vector vChecks(1); @@ -362,7 +353,7 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup) /** Test that CCheckQueueControl is threadsafe */ BOOST_AUTO_TEST_CASE(test_CheckQueueControl_Locks) { - auto queue = std::make_unique(QUEUE_BATCH_SIZE); + auto queue = std::make_unique(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS); { std::vector tg; std::atomic nThreads {0}; diff --git a/src/test/fuzz/checkqueue.cpp b/src/test/fuzz/checkqueue.cpp index 429570526f1..6320b500b6f 100644 --- a/src/test/fuzz/checkqueue.cpp +++ b/src/test/fuzz/checkqueue.cpp @@ -31,8 +31,8 @@ FUZZ_TARGET(checkqueue) FuzzedDataProvider fuzzed_data_provider(buffer.data(), buffer.size()); const unsigned int batch_size = fuzzed_data_provider.ConsumeIntegralInRange(0, 1024); - CCheckQueue check_queue_1{batch_size}; - CCheckQueue check_queue_2{batch_size}; + CCheckQueue check_queue_1{batch_size, /*worker_threads_num=*/0}; + CCheckQueue check_queue_2{batch_size, /*worker_threads_num=*/0}; std::vector checks_1; std::vector checks_2; const int size = fuzzed_data_provider.ConsumeIntegralInRange(0, 1024); diff --git a/src/test/transaction_tests.cpp b/src/test/transaction_tests.cpp index 830cfeb25a9..932c40cdec1 100644 --- a/src/test/transaction_tests.cpp +++ b/src/test/transaction_tests.cpp @@ -530,11 +530,9 @@ BOOST_AUTO_TEST_CASE(test_big_witness_transaction) // check all inputs concurrently, with the cache PrecomputedTransactionData txdata(tx); - CCheckQueue scriptcheckqueue(128); + CCheckQueue scriptcheckqueue(/*batch_size=*/128, /*worker_threads_num=*/20); CCheckQueueControl control(&scriptcheckqueue); - scriptcheckqueue.StartWorkerThreads(20); - std::vector coins; for(uint32_t i = 0; i < mtx.vin.size(); i++) { Coin coin; diff --git a/src/validation.cpp b/src/validation.cpp index 91027872b02..ba88ef2ec5d 100644 --- a/src/validation.cpp +++ b/src/validation.cpp @@ -5739,12 +5739,11 @@ static ChainstateManager::Options&& Flatten(ChainstateManager::Options&& opts) } ChainstateManager::ChainstateManager(const util::SignalInterrupt& interrupt, Options options, node::BlockManager::Options blockman_options) - : m_script_check_queue{/*nBatchSizeIn=*/128}, + : m_script_check_queue{/*batch_size=*/128, options.worker_threads_num}, m_interrupt{interrupt}, m_options{Flatten(std::move(options))}, m_blockman{interrupt, std::move(blockman_options)} { - m_script_check_queue.StartWorkerThreads(m_options.worker_threads_num); } ChainstateManager::~ChainstateManager()