diff --git a/src/bench/checkqueue.cpp b/src/bench/checkqueue.cpp index 948c3c2608..b69b59d5d0 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); @@ -62,9 +63,8 @@ static void CCheckQueueSpeedPrevectorJob(benchmark::Bench& bench) // it is done explicitly here for clarity control.Wait(); }); - queue.StopWorkerThreads(); - // ELEMENTS: ...and deallocate them + // ELEMENTS: deallocate vChecks for (auto& vChecks : vBatches) for (size_t x = 0; x < BATCH_SIZE; ++x) delete vChecks[x]; diff --git a/src/bitcoin-chainstate.cpp b/src/bitcoin-chainstate.cpp index 995b4781fc..31edb86ce2 100644 --- a/src/bitcoin-chainstate.cpp +++ b/src/bitcoin-chainstate.cpp @@ -290,7 +290,6 @@ epilogue: // dereferencing and UB. scheduler.stop(); if (chainman.m_thread_load.joinable()) chainman.m_thread_load.join(); - StopScriptCheckWorkerThreads(); GetMainSignals().FlushBackgroundCallbacks(); { diff --git a/src/checkqueue.h b/src/checkqueue.h index 12d5bd7af7..91bf1d2add 100644 --- a/src/checkqueue.h +++ b/src/checkqueue.h @@ -13,9 +13,6 @@ #include #include -template -class CCheckQueueControl; - /** * Queue for verifications that have to be performed. * The verifications are represented by a type T, which must provide an @@ -134,22 +131,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 */); @@ -157,6 +143,13 @@ public: } } + // 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) { @@ -183,25 +176,17 @@ public: } } - //! Stop all of the worker threads. - void StopWorkerThreads() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + ~CCheckQueue() { WITH_LOCK(m_mutex, m_request_stop = true); + for (auto remaining : queue) delete remaining; // ELEMENTS 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() - { - for (auto remaining : queue) delete remaining; - assert(m_worker_threads.empty()); - } }; /** diff --git a/src/init.cpp b/src/init.cpp index e3108625d5..acc0956c70 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -276,11 +276,10 @@ 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. - if (node.chainman && node.chainman->m_thread_load.joinable()) node.chainman->m_thread_load.join(); + // scheduler and load block thread. if (node.scheduler) node.scheduler->stop(); - if (node.reverification_scheduler) node.reverification_scheduler->stop(); - StopScriptCheckWorkerThreads(); + if (node.reverification_scheduler) node.reverification_scheduler->stop(); // ELEMENTS + if (node.chainman && node.chainman->m_thread_load.joinable()) node.chainman->m_thread_load.join(); // After the threads that potentially access these pointers have been stopped, // destruct and reset all to nullptr. @@ -1249,24 +1248,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(); assert(!node.reverification_scheduler); diff --git a/src/kernel/chainstatemanager_opts.h b/src/kernel/chainstatemanager_opts.h index 917f7d226c..ee20eabd79 100644 --- a/src/kernel/chainstatemanager_opts.h +++ b/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 diff --git a/src/node/chainstatemanager_args.cpp b/src/node/chainstatemanager_args.cpp index 87d9238c18..1cc126cb05 100644 --- a/src/node/chainstatemanager_args.cpp +++ b/src/node/chainstatemanager_args.cpp @@ -6,7 +6,8 @@ #include #include -#include +#include +#include #include #include #include @@ -16,6 +17,7 @@ #include #include +#include #include #include @@ -41,6 +43,16 @@ util::Result 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 diff --git a/src/node/chainstatemanager_args.h b/src/node/chainstatemanager_args.h index 701515953e..b2cdba68b8 100644 --- a/src/node/chainstatemanager_args.h +++ b/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 ApplyArgsManOptions(const ArgsManager& args, ChainstateManager::Options& opts); } // namespace node diff --git a/src/qt/optionsdialog.cpp b/src/qt/optionsdialog.cpp index cc5c772038..64569064ae 100644 --- a/src/qt/optionsdialog.cpp +++ b/src/qt/optionsdialog.cpp @@ -17,9 +17,9 @@ #include #include +#include #include #include -#include #include diff --git a/src/qt/optionsmodel.cpp b/src/qt/optionsmodel.cpp index 88ab25c58b..6cb6f64151 100644 --- a/src/qt/optionsmodel.cpp +++ b/src/qt/optionsmodel.cpp @@ -17,6 +17,7 @@ #include #include #include +#include #include // for -dbcache defaults #include #include // For DEFAULT_SCRIPTCHECK_THREADS diff --git a/src/test/checkqueue_tests.cpp b/src/test/checkqueue_tests.cpp index 2d2e4d1b31..81ce0fb43a 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); @@ -179,7 +178,6 @@ static void Correct_Queue_range(std::vector range) BOOST_REQUIRE(control.Wait()); BOOST_REQUIRE_EQUAL(FakeCheckCheckCompletion::n_calls, i); } - small_queue->StopWorkerThreads(); } /** Test that 0 checks is correct @@ -221,9 +219,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,15 +240,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(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()); @@ -269,7 +262,6 @@ 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 @@ -277,9 +269,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; { @@ -302,7 +292,6 @@ BOOST_AUTO_TEST_CASE(test_CheckQueue_UniqueCheck) } BOOST_REQUIRE(r); } - queue->StopWorkerThreads(); } @@ -313,8 +302,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; { @@ -333,16 +321,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(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; @@ -374,14 +360,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(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 627c2d7918..c71f30be5c 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 f31d633d51..59a9d76e6a 100644 --- a/src/test/transaction_tests.cpp +++ b/src/test/transaction_tests.cpp @@ -545,11 +545,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; @@ -568,7 +566,6 @@ BOOST_AUTO_TEST_CASE(test_big_witness_transaction) bool controlCheck = control.Wait(); assert(controlCheck); - scriptcheckqueue.StopWorkerThreads(); } SignatureData CombineSignatures(const CMutableTransaction& input1, const CMutableTransaction& input2, const CTransactionRef tx) diff --git a/src/test/util/setup_common.cpp b/src/test/util/setup_common.cpp index b99ee9f485..63483cb3ba 100644 --- a/src/test/util/setup_common.cpp +++ b/src/test/util/setup_common.cpp @@ -208,6 +208,7 @@ ChainTestingSetup::ChainTestingSetup(const ChainType chainType, const std::vecto .minimum_chain_work = std::nullopt, .assumed_valid_block = std::nullopt, .notifications = *m_node.notifications, + .worker_threads_num = 2, }; const BlockManager::Options blockman_opts{ .chainparams = chainman_opts.chainparams, @@ -219,15 +220,11 @@ ChainTestingSetup::ChainTestingSetup(const ChainType chainType, const std::vecto .path = m_args.GetDataDirNet() / "blocks" / "index", .cache_bytes = static_cast(m_cache_sizes.block_tree_db), .memory_only = true}); - - constexpr int script_check_threads = 2; - StartScriptCheckWorkerThreads(script_check_threads); } ChainTestingSetup::~ChainTestingSetup() { if (m_node.scheduler) m_node.scheduler->stop(); - StopScriptCheckWorkerThreads(); GetMainSignals().FlushBackgroundCallbacks(); GetMainSignals().UnregisterBackgroundSignalScheduler(); m_node.connman.reset(); diff --git a/src/validation.cpp b/src/validation.cpp index 0e98a265d2..9698f31bbe 100644 --- a/src/validation.cpp +++ b/src/validation.cpp @@ -2228,18 +2228,6 @@ DisconnectResult Chainstate::DisconnectBlock(const CBlock& block, const CBlockIn return fClean ? DISCONNECT_OK : DISCONNECT_UNCLEAN; } -static CCheckQueue scriptcheckqueue(128); - -void StartScriptCheckWorkerThreads(int threads_num) -{ - scriptcheckqueue.StartWorkerThreads(threads_num); -} - -void StopScriptCheckWorkerThreads() -{ - scriptcheckqueue.StopWorkerThreads(); -} - /** * Threshold condition checker that triggers when unknown versionbits are seen on the network. */ @@ -2359,7 +2347,7 @@ bool Chainstate::ConnectBlock(const CBlock& block, BlockValidationState& state, uint256 block_hash{block.GetHash()}; assert(*pindex->phashBlock == block_hash); - const bool parallel_script_checks{scriptcheckqueue.HasThreads()}; + const bool parallel_script_checks{m_chainman.GetCheckQueue().HasThreads()}; const auto time_start{SteadyClock::now()}; const CChainParams& params{m_chainman.GetParams()}; @@ -2569,7 +2557,7 @@ bool Chainstate::ConnectBlock(const CBlock& block, BlockValidationState& state, // in multiple threads). Preallocate the vector size so a new allocation // doesn't invalidate pointers into the vector, and keep txsdata in scope // for as long as `control`. - CCheckQueueControl control(fScriptChecks && parallel_script_checks ? &scriptcheckqueue : nullptr); + CCheckQueueControl control(fScriptChecks && parallel_script_checks ? &m_chainman.GetCheckQueue() : nullptr); std::vector txsdata; for (unsigned int i = 0; i < block.vtx.size(); i++){ txsdata.emplace_back(m_chainman.GetParams().HashGenesisBlock()); @@ -6268,9 +6256,12 @@ static ChainstateManager::Options&& Flatten(ChainstateManager::Options&& opts) } ChainstateManager::ChainstateManager(const util::SignalInterrupt& interrupt, Options options, node::BlockManager::Options blockman_options) - : m_interrupt{interrupt}, + : 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_blockman{interrupt, std::move(blockman_options)} +{ +} ChainstateManager::~ChainstateManager() { diff --git a/src/validation.h b/src/validation.h index 80b58a177a..f932aea41e 100644 --- a/src/validation.h +++ b/src/validation.h @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -68,10 +69,6 @@ namespace util { class SignalInterrupt; } // namespace util -/** Maximum number of dedicated script-checking threads allowed */ -static const int MAX_SCRIPTCHECK_THREADS = 15; -/** -par default (number of script-checking threads, 0 = auto) */ -static const int DEFAULT_SCRIPTCHECK_THREADS = 0; /** Block files containing a block-height within MIN_BLOCKS_TO_KEEP of ActiveChain().Tip() will not be pruned. */ static const unsigned int MIN_BLOCKS_TO_KEEP = 288; static const signed int DEFAULT_CHECKBLOCKS = 6; @@ -108,11 +105,6 @@ extern uint256 g_best_block; /** Documentation for argument 'checklevel'. */ extern const std::vector CHECKLEVEL_DOC; -/** Run instances of script checking worker threads */ -void StartScriptCheckWorkerThreads(int threads_num); -/** Stop all of the script checking worker threads */ -void StopScriptCheckWorkerThreads(); - CAmount GetBlockSubsidy(int nHeight, const Consensus::Params& consensusParams); bool FatalError(kernel::Notifications& notifications, BlockValidationState& state, const std::string& strMessage, const bilingual_str& userMessage = {}); @@ -936,6 +928,9 @@ private: return cs && !cs->m_disabled; } + //! A queue for script verifications that have to be performed by worker threads. + CCheckQueue m_script_check_queue; + public: using Options = kernel::ChainstateManagerOpts; @@ -1286,6 +1281,8 @@ public: //! nullopt. std::optional GetSnapshotBaseHeight() const EXCLUSIVE_LOCKS_REQUIRED(::cs_main); + CCheckQueue& GetCheckQueue() { return m_script_check_queue; } + ~ChainstateManager(); };