Merge bitcoin#26762: bugfix: Make CCheckQueue RAII-styled (attempt 2)

5b3ea5f refactor: Move `{MAX,DEFAULT}_SCRIPTCHECK_THREADS` constants (Hennadii Stepanov)
6e17b31 refactor: Make `CCheckQueue` non-copyable and non-movable explicitly (Hennadii Stepanov)
8111e74 refactor: Drop unneeded declaration (Hennadii Stepanov)
9cf89f7 refactor: Make `CCheckQueue` constructor start worker threads (Hennadii Stepanov)
d03eaac Make `CCheckQueue` destructor stop worker threads (Hennadii Stepanov)
be4ff30 Move global `scriptcheckqueue` into `ChainstateManager` class (Hennadii Stepanov)

Pull request description:

  This PR:
  - makes `CCheckQueue` RAII-styled
  - gets rid of the global `scriptcheckqueue`
  - fixes bitcoin#25448

  The previous attempt was in bitcoin#18731.

ACKs for top commit:
  martinus:
    ACK 5b3ea5f
  achow101:
    ACK 5b3ea5f
  TheCharlatan:
    ACK 5b3ea5f

Tree-SHA512: 45cca846e7ed107e3930149f0b616ddbaf2648d6cde381f815331b861b5d67ab39e154883ae174b8abb1dae485bc904318c50c51e5d6b46923d89de51c5eadb0

Author: achow101

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);
@@ -61,7 +62,6 @@ static void CCheckQueueSpeedPrevectorJob(benchmark::Bench& bench)
         // it is done explicitly here for clarity
         control.Wait();
     });
-    queue.StopWorkerThreads();
     ECC_Stop();
 }
 BENCHMARK(CCheckQueueSpeedPrevectorJob, benchmark::PriorityLevel::HIGH);

src/bitcoin-chainstate.cpp

@@ -290,7 +290,6 @@ int main(int argc, char* argv[])
     // dereferencing and UB.
     scheduler.stop();
     if (chainman.m_thread_load.joinable()) chainman.m_thread_load.join();
-    StopScriptCheckWorkerThreads();
 
     GetMainSignals().FlushBackgroundCallbacks();
     {

src/checkqueue.h

@@ -13,9 +13,6 @@
 #include <iterator>
 #include <vector>
 
-template <typename T>
-class CCheckQueueControl;
-
 /**
  * Queue for verifications that have to be performed.
   * The verifications are represented by a type T, which must provide an
@@ -130,29 +127,25 @@ 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 */);
             });
         }
     }
 
+    // Since this class manages its own resources, which is a thread
+    // pool `m_worker_threads`, copy and move operations are not appropriate.
+    CCheckQueue(const CCheckQueue&) = delete;
+    CCheckQueue& operator=(const CCheckQueue&) = delete;
+    CCheckQueue(CCheckQueue&&) = delete;
+    CCheckQueue& operator=(CCheckQueue&&) = delete;
+
     //! Wait until execution finishes, and return whether all evaluations were successful.
     bool Wait() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
     {
@@ -179,24 +172,16 @@ class CCheckQueue
         }
     }
 
-    //! Stop all of the worker threads.
-    void StopWorkerThreads() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+    ~CCheckQueue()
     {
         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();
-        WITH_LOCK(m_mutex, m_request_stop = false);
     }
 
     bool HasThreads() const { return !m_worker_threads.empty(); }
-
-    ~CCheckQueue()
-    {
-        assert(m_worker_threads.empty());
-    }
 };
 
 /**

src/init.cpp

@@ -268,10 +268,9 @@ void Shutdown(NodeContext& node)
     StopTorControl();
 
     // After everything has been shut down, but before things get flushed, stop the
-    // CScheduler/checkqueue, scheduler and load block thread.
+    // scheduler and load block thread.
     if (node.scheduler) node.scheduler->stop();
     if (node.chainman && node.chainman->m_thread_load.joinable()) node.chainman->m_thread_load.join();
-    StopScriptCheckWorkerThreads();
 
     // After the threads that potentially access these pointers have been stopped,
     // destruct and reset all to nullptr.
@@ -1114,24 +1113,6 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
         return InitError(strprintf(_("Unable to allocate memory for -maxsigcachesize: '%s' MiB"), args.GetIntArg("-maxsigcachesize", DEFAULT_MAX_SIG_CACHE_BYTES >> 20)));
     }
 
-    int script_threads = args.GetIntArg("-par", DEFAULT_SCRIPTCHECK_THREADS);
-    if (script_threads <= 0) {
-        // -par=0 means autodetect (number of cores - 1 script threads)
-        // -par=-n means "leave n cores free" (number of cores - n - 1 script threads)
-        script_threads += GetNumCores();
-    }
-
-    // Subtract 1 because the main thread counts towards the par threads
-    script_threads = std::max(script_threads - 1, 0);
-
-    // Number of script-checking threads <= MAX_SCRIPTCHECK_THREADS
-    script_threads = std::min(script_threads, MAX_SCRIPTCHECK_THREADS);
-
-    LogPrintf("Script verification uses %d additional threads\n", script_threads);
-    if (script_threads >= 1) {
-        StartScriptCheckWorkerThreads(script_threads);
-    }
-
     assert(!node.scheduler);
     node.scheduler = std::make_unique<CScheduler>();
 

src/kernel/chainstatemanager_opts.h

@@ -45,6 +45,8 @@ struct ChainstateManagerOpts {
     DBOptions coins_db{};
     CoinsViewOptions coins_view{};
     Notifications& notifications;
+    //! Number of script check worker threads. Zero means no parallel verification.
+    int worker_threads_num{0};
 };
 
 } // namespace kernel

src/node/chainstatemanager_args.cpp

@@ -6,7 +6,8 @@
 
 #include <arith_uint256.h>
 #include <common/args.h>
-#include <kernel/chainstatemanager_opts.h>
+#include <common/system.h>
+#include <logging.h>
 #include <node/coins_view_args.h>
 #include <node/database_args.h>
 #include <tinyformat.h>
@@ -16,6 +17,7 @@
 #include <util/translation.h>
 #include <validation.h>
 
+#include <algorithm>
 #include <chrono>
 #include <string>
 
@@ -41,6 +43,16 @@ util::Result<void> ApplyArgsManOptions(const ArgsManager& args, ChainstateManage
     ReadDatabaseArgs(args, opts.coins_db);
     ReadCoinsViewArgs(args, opts.coins_view);
 
+    int script_threads = args.GetIntArg("-par", DEFAULT_SCRIPTCHECK_THREADS);
+    if (script_threads <= 0) {
+        // -par=0 means autodetect (number of cores - 1 script threads)
+        // -par=-n means "leave n cores free" (number of cores - n - 1 script threads)
+        script_threads += GetNumCores();
+    }
+    // Subtract 1 because the main thread counts towards the par threads.
+    opts.worker_threads_num = std::clamp(script_threads - 1, 0, MAX_SCRIPTCHECK_THREADS);
+    LogPrintf("Script verification uses %d additional threads\n", opts.worker_threads_num);
+
     return {};
 }
 } // namespace node

src/node/chainstatemanager_args.h

@@ -10,6 +10,11 @@
 
 class ArgsManager;
 
+/** Maximum number of dedicated script-checking threads allowed */
+static constexpr int MAX_SCRIPTCHECK_THREADS{15};
+/** -par default (number of script-checking threads, 0 = auto) */
+static constexpr int DEFAULT_SCRIPTCHECK_THREADS{0};
+
 namespace node {
 [[nodiscard]] util::Result<void> ApplyArgsManOptions(const ArgsManager& args, ChainstateManager::Options& opts);
 } // namespace node

src/qt/optionsdialog.cpp

@@ -17,9 +17,9 @@
 
 #include <common/system.h>
 #include <interfaces/node.h>
+#include <node/chainstatemanager_args.h>
 #include <netbase.h>
 #include <txdb.h>
-#include <validation.h>
 
 #include <chrono>
 

src/qt/optionsmodel.cpp

@@ -17,6 +17,7 @@
 #include <mapport.h>
 #include <net.h>
 #include <netbase.h>
+#include <node/chainstatemanager_args.h>
 #include <txdb.h> // for -dbcache defaults
 #include <util/string.h>
 #include <validation.h>    // For DEFAULT_SCRIPTCHECK_THREADS

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);
@@ -176,7 +175,6 @@ static void Correct_Queue_range(std::vector<size_t> range)
         BOOST_REQUIRE(control.Wait());
         BOOST_REQUIRE_EQUAL(FakeCheckCheckCompletion::n_calls, i);
     }
-    small_queue->StopWorkerThreads();
 }
 
 /** Test that 0 checks is correct
@@ -218,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;
@@ -240,15 +236,12 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Catches_Failure)
             BOOST_REQUIRE(success);
         }
     }
-    fail_queue->StopWorkerThreads();
 }
 // Test that a block validation which fails does not interfere with
 // 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());
@@ -262,17 +255,14 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Recovers_From_Failure)
             BOOST_REQUIRE(r != end_fails);
         }
     }
-    fail_queue->StopWorkerThreads();
 }
 
 // Test that unique checks are actually all called individually, rather than
 // just one check being called repeatedly. Test that checks are not called
 // 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;
     {
@@ -294,7 +284,6 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck)
         }
         BOOST_REQUIRE(r);
     }
-    queue->StopWorkerThreads();
 }
 
 
@@ -305,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;
         {
@@ -325,16 +313,14 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_Memory)
         }
         BOOST_REQUIRE_EQUAL(MemoryCheck::fake_allocated_memory, 0U);
     }
-    queue->StopWorkerThreads();
 }
 
 // Test that a new verification cannot occur until all checks
 // 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);
@@ -361,14 +347,13 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_FrozenCleanup)
     // Wait for control to finish
     t0.join();
     BOOST_REQUIRE(!fails);
-    queue->StopWorkerThreads();
 }
 
 
 /** 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};