refactor: Make CCheckQueue constructor start worker threads

Author: hebasto

src/bench/checkqueue.cpp

@@ -37,10 +37,11 @@ static void CCheckQueueSpeedPrevectorJob(benchmark::Bench& bench)
             return true;
         }
     };
-    CCheckQueue<PrevectorJob> 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<PrevectorJob> queue{QUEUE_BATCH_SIZE, worker_threads_num};
 
     // create all the data once, then submit copies in the benchmark.
     FastRandomContext insecure_rand(true);

src/checkqueue.h

@@ -130,22 +130,11 @@ class CCheckQueue
     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 */);

src/test/checkqueue_tests.cpp

@@ -158,8 +158,7 @@ typedef CCheckQueue<FrozenCleanupCheck> FrozenCleanup_Queue;
  */
 static void Correct_Queue_range(std::vector<size_t> range)
 {
-    auto small_queue = std::make_unique<Correct_Queue>(QUEUE_BATCH_SIZE);
-    small_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
+    auto small_queue = std::make_unique<Correct_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     // Make vChecks here to save on malloc (this test can be slow...)
     std::vector<FakeCheckCheckCompletion> 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<Failing_Queue>(QUEUE_BATCH_SIZE);
-    fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
-
+    auto fail_queue = std::make_unique<Failing_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     for (size_t i = 0; i < 1001; ++i) {
         CCheckQueueControl<FailingCheck> 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<Failing_Queue>(QUEUE_BATCH_SIZE);
-    fail_queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
-
+    auto fail_queue = std::make_unique<Failing_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     for (auto times = 0; times < 10; ++times) {
         for (const bool end_fails : {true, false}) {
             CCheckQueueControl<FailingCheck> 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<Unique_Queue>(QUEUE_BATCH_SIZE);
-    queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
-
+    auto queue = std::make_unique<Unique_Queue>(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<Memory_Queue>(QUEUE_BATCH_SIZE);
-    queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
+    auto queue = std::make_unique<Memory_Queue>(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<FrozenCleanup_Queue>(QUEUE_BATCH_SIZE);
+    auto queue = std::make_unique<FrozenCleanup_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     bool fails = false;
-    queue->StartWorkerThreads(SCRIPT_CHECK_THREADS);
     std::thread t0([&]() {
         CCheckQueueControl<FrozenCleanupCheck> control(queue.get());
         std::vector<FrozenCleanupCheck> 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<Standard_Queue>(QUEUE_BATCH_SIZE);
+    auto queue = std::make_unique<Standard_Queue>(QUEUE_BATCH_SIZE, SCRIPT_CHECK_THREADS);
     {
         std::vector<std::thread> tg;
         std::atomic<int> nThreads {0};

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<unsigned int>(0, 1024);
-    CCheckQueue<DumbCheck> check_queue_1{batch_size};
-    CCheckQueue<DumbCheck> check_queue_2{batch_size};
+    CCheckQueue<DumbCheck> check_queue_1{batch_size, /*worker_threads_num=*/0};
+    CCheckQueue<DumbCheck> check_queue_2{batch_size, /*worker_threads_num=*/0};
     std::vector<DumbCheck> checks_1;
     std::vector<DumbCheck> checks_2;
     const int size = fuzzed_data_provider.ConsumeIntegralInRange<int>(0, 1024);

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<CScriptCheck> scriptcheckqueue(128);
+    CCheckQueue<CScriptCheck> scriptcheckqueue(/*batch_size=*/128, /*worker_threads_num=*/20);
     CCheckQueueControl<CScriptCheck> control(&scriptcheckqueue);
 
-    scriptcheckqueue.StartWorkerThreads(20);
-
     std::vector<Coin> coins;
     for(uint32_t i = 0; i < mtx.vin.size(); i++) {
         Coin coin;

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()