Move global scriptcheckqueue into ChainstateManager class

This commit is contained in:
Hennadii Stepanov 2023-07-07 10:33:11 +01:00
parent 4e78834ec1
commit be4ff3060b
No known key found for this signature in database
GPG key ID: 410108112E7EA81F
7 changed files with 38 additions and 42 deletions

View file

@ -290,7 +290,7 @@ epilogue:
// dereferencing and UB.
scheduler.stop();
if (chainman.m_thread_load.joinable()) chainman.m_thread_load.join();
StopScriptCheckWorkerThreads();
chainman.StopScriptCheckWorkerThreads();
GetMainSignals().FlushBackgroundCallbacks();
{

View file

@ -271,7 +271,7 @@ void Shutdown(NodeContext& node)
// CScheduler/checkqueue, 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();
if (node.chainman) node.chainman->StopScriptCheckWorkerThreads();
// After the threads that potentially access these pointers have been stopped,
// destruct and reset all to nullptr.
@ -1109,24 +1109,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>();

View file

@ -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

View file

@ -6,7 +6,9 @@
#include <arith_uint256.h>
#include <common/args.h>
#include <common/system.h>
#include <kernel/chainstatemanager_opts.h>
#include <logging.h>
#include <node/coins_view_args.h>
#include <node/database_args.h>
#include <tinyformat.h>
@ -16,6 +18,7 @@
#include <util/translation.h>
#include <validation.h>
#include <algorithm>
#include <chrono>
#include <string>
@ -41,6 +44,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

View file

@ -176,6 +176,7 @@ ChainTestingSetup::ChainTestingSetup(const ChainType chainType, const std::vecto
.adjusted_time_callback = GetAdjustedTime,
.check_block_index = true,
.notifications = *m_node.notifications,
.worker_threads_num = 2,
};
const BlockManager::Options blockman_opts{
.chainparams = chainman_opts.chainparams,
@ -187,15 +188,12 @@ ChainTestingSetup::ChainTestingSetup(const ChainType chainType, const std::vecto
.path = m_args.GetDataDirNet() / "blocks" / "index",
.cache_bytes = static_cast<size_t>(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();
m_node.chainman->StopScriptCheckWorkerThreads();
GetMainSignals().FlushBackgroundCallbacks();
GetMainSignals().UnregisterBackgroundSignalScheduler();
m_node.connman.reset();

View file

@ -2047,16 +2047,9 @@ DisconnectResult Chainstate::DisconnectBlock(const CBlock& block, const CBlockIn
return fClean ? DISCONNECT_OK : DISCONNECT_UNCLEAN;
}
static CCheckQueue<CScriptCheck> scriptcheckqueue(128);
void StartScriptCheckWorkerThreads(int threads_num)
void ChainstateManager::StopScriptCheckWorkerThreads()
{
scriptcheckqueue.StartWorkerThreads(threads_num);
}
void StopScriptCheckWorkerThreads()
{
scriptcheckqueue.StopWorkerThreads();
m_script_check_queue.StopWorkerThreads();
}
/**
@ -2147,7 +2140,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()};
@ -2336,7 +2329,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<CScriptCheck> control(fScriptChecks && parallel_script_checks ? &scriptcheckqueue : nullptr);
CCheckQueueControl<CScriptCheck> control(fScriptChecks && parallel_script_checks ? &m_chainman.GetCheckQueue() : nullptr);
std::vector<PrecomputedTransactionData> txsdata(block.vtx.size());
std::vector<int> prevheights;
@ -5751,12 +5744,18 @@ 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{/*nBatchSizeIn=*/128},
m_interrupt{interrupt},
m_options{Flatten(std::move(options))},
m_blockman{interrupt, std::move(blockman_options)} {}
m_blockman{interrupt, std::move(blockman_options)}
{
m_script_check_queue.StartWorkerThreads(m_options.worker_threads_num);
}
ChainstateManager::~ChainstateManager()
{
StopScriptCheckWorkerThreads();
LOCK(::cs_main);
m_versionbitscache.Clear();

View file

@ -13,6 +13,7 @@
#include <arith_uint256.h>
#include <attributes.h>
#include <chain.h>
#include <checkqueue.h>
#include <kernel/chain.h>
#include <consensus/amount.h>
#include <deploymentstatus.h>
@ -98,11 +99,6 @@ extern uint256 g_best_block;
/** Documentation for argument 'checklevel'. */
extern const std::vector<std::string> 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 = {});
@ -896,6 +892,9 @@ private:
return cs && !cs->m_disabled;
}
//! A queue for script verifications that have to be performed by worker threads.
CCheckQueue<CScriptCheck> m_script_check_queue;
public:
using Options = kernel::ChainstateManagerOpts;
@ -1246,6 +1245,9 @@ public:
//! nullopt.
std::optional<int> GetSnapshotBaseHeight() const EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
CCheckQueue<CScriptCheck>& GetCheckQueue() { return m_script_check_queue; }
void StopScriptCheckWorkerThreads();
~ChainstateManager();
};