diff --git a/src/init.cpp b/src/init.cpp index f22302ff21..53cd1d8967 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -540,6 +540,8 @@ void SetupServerArgs(ArgsManager& argsman, bool can_listen_ipc) argsman.AddArg("-par=", strprintf("Set the number of script verification threads (0 = auto, up to %d, <0 = leave that many cores free, default: %d)", MAX_SCRIPTCHECK_THREADS, DEFAULT_SCRIPTCHECK_THREADS), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS); argsman.AddArg("-prevoutfetchthreads=", strprintf("Set the number of threads used to prefetch block input prevouts from the chainstate database (0 disables, up to %d, default: %d). Negative values are rejected.", MAX_PREVOUTFETCH_THREADS, DEFAULT_PREVOUTFETCH_THREADS), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS); + argsman.AddArg("-blockreadaheadthreads=", strprintf("Set the number of threads reading blocks ahead of the connect cursor (0 disables, up to %d, default: %d). Negative values are rejected.", MAX_BLOCK_READAHEAD_THREADS, DEFAULT_BLOCK_READAHEAD_THREADS), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS); + argsman.AddArg("-blockreadahead=", strprintf("Set how many blocks a background thread reads and deserializes ahead of the connect cursor during sequential connection such as -reindex-chainstate and IBD (0 disables, up to %d, default: %d). Negative values are rejected.", MAX_BLOCK_READAHEAD, DEFAULT_BLOCK_READAHEAD), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS); argsman.AddArg("-persistmempool", strprintf("Whether to save the mempool on shutdown and load on restart (default: %u)", DEFAULT_PERSIST_MEMPOOL), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS); argsman.AddArg("-persistmempoolv1", strprintf("Whether a mempool.dat file created by -persistmempool or the savemempool RPC will be written in the legacy format " diff --git a/src/kernel/chainstatemanager_opts.h b/src/kernel/chainstatemanager_opts.h index 554d032edd..dac19fa69b 100644 --- a/src/kernel/chainstatemanager_opts.h +++ b/src/kernel/chainstatemanager_opts.h @@ -23,6 +23,9 @@ class ValidationSignals; static constexpr auto DEFAULT_MAX_TIP_AGE{24h}; static constexpr int32_t DEFAULT_PREVOUTFETCH_THREADS{8}; +static constexpr int32_t DEFAULT_BLOCK_READAHEAD{4}; + +static constexpr int32_t DEFAULT_BLOCK_READAHEAD_THREADS{2}; namespace kernel { @@ -49,6 +52,8 @@ struct ChainstateManagerOpts { int worker_threads_num{0}; //! Number of worker threads used for prefetching block input prevouts. Zero means no parallel fetching. int32_t prevoutfetch_threads_num{DEFAULT_PREVOUTFETCH_THREADS}; + int32_t block_readahead_depth{DEFAULT_BLOCK_READAHEAD}; + int32_t block_readahead_threads{DEFAULT_BLOCK_READAHEAD_THREADS}; size_t script_execution_cache_bytes{DEFAULT_SCRIPT_EXECUTION_CACHE_BYTES}; size_t signature_cache_bytes{DEFAULT_SIGNATURE_CACHE_BYTES}; }; diff --git a/src/node/block_read_ahead.h b/src/node/block_read_ahead.h new file mode 100644 index 0000000000..6485d68b69 --- /dev/null +++ b/src/node/block_read_ahead.h @@ -0,0 +1,145 @@ +// Copyright (c) 2025-present The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#ifndef BITCOIN_NODE_BLOCK_READ_AHEAD_H +#define BITCOIN_NODE_BLOCK_READ_AHEAD_H + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace node { + +class BlockReadAhead +{ +public: + explicit BlockReadAhead(const BlockManager& blockman) : m_blockman{blockman} {} + ~BlockReadAhead() { Stop(); } + + BlockReadAhead(const BlockReadAhead&) = delete; + BlockReadAhead& operator=(const BlockReadAhead&) = delete; + + void Start(int32_t depth, int32_t threads) + { + if (depth <= 0 || threads <= 0 || Enabled()) return; + m_depth = static_cast(depth); + WITH_LOCK(m_mutex, m_stop = false); + m_threads.reserve(threads); + for (int32_t i{0}; i < threads; ++i) { + m_threads.emplace_back(&util::TraceThread, strprintf("blockread.%d", i), [this] { ReaderThread(); }); + } + } + + void Stop() + { + if (Enabled()) { + WITH_LOCK(m_mutex, m_stop = true); + m_cv.notify_all(); + for (auto& thread : m_threads) thread.join(); + m_threads.clear(); + } + DropBuffered(); + } + + bool Enabled() const { return !m_threads.empty(); } + + void Prime(const std::vector& to_connect) EXCLUSIVE_LOCKS_REQUIRED(::cs_main) + { + if (!Enabled()) return; + InFlightMap next; + bool queued_any{false}; + { + LOCK(m_mutex); + for (CBlockIndex* pindex : to_connect | std::views::reverse) { + if (next.size() >= m_depth) break; + if (next.contains(pindex)) continue; + if (auto in_flight{m_inflight.extract(pindex)}) { + next.insert(std::move(in_flight)); + continue; + } + if (!(pindex->nStatus & BLOCK_HAVE_DATA)) continue; + const FlatFilePos pos{pindex->GetBlockPos()}; + if (pos.IsNull()) continue; + const uint256 hash{pindex->GetBlockHash()}; + ReadTask task{[blockman = &m_blockman, pos, hash]() -> std::shared_ptr { + auto block{std::make_shared()}; + if (!blockman->ReadBlock(*block, pos, hash)) return nullptr; + return block; + }}; + next.emplace(pindex, task.get_future()); + m_queue.push_back(std::move(task)); + queued_any = true; + } + } + m_inflight = std::move(next); + if (queued_any) m_cv.notify_all(); + } + + std::shared_ptr Take(const CBlockIndex* pindex) EXCLUSIVE_LOCKS_REQUIRED(::cs_main) + { + auto it{m_inflight.find(pindex)}; + if (it == m_inflight.end()) return nullptr; + std::shared_ptr block{it->second.get()}; + m_inflight.erase(it); + return block; + } + + void Clear() EXCLUSIVE_LOCKS_REQUIRED(::cs_main) { DropBuffered(); } + +private: + using ReadTask = std::packaged_task()>; + using InFlightMap = std::map>>; + + void DropBuffered() + { + m_inflight.clear(); + WITH_LOCK(m_mutex, m_queue.clear()); + } + + void ReaderThread() + { + for (;;) { + ReadTask task; + { + WAIT_LOCK(m_mutex, lock); + m_cv.wait(lock, [&]() EXCLUSIVE_LOCKS_REQUIRED(m_mutex) { return m_stop || !m_queue.empty(); }); + if (m_stop) return; + task = std::move(m_queue.front()); + m_queue.pop_front(); + } + task(); + } + } + + const BlockManager& m_blockman; + std::vector m_threads; + size_t m_depth{0}; + InFlightMap m_inflight; + + Mutex m_mutex; + std::condition_variable m_cv; + std::deque m_queue GUARDED_BY(m_mutex); + bool m_stop GUARDED_BY(m_mutex){false}; +}; + +} // namespace node + +#endif // BITCOIN_NODE_BLOCK_READ_AHEAD_H diff --git a/src/node/chainstatemanager_args.cpp b/src/node/chainstatemanager_args.cpp index b44314908d..4d615bd1a5 100644 --- a/src/node/chainstatemanager_args.cpp +++ b/src/node/chainstatemanager_args.cpp @@ -67,6 +67,20 @@ util::Result ApplyArgsManOptions(const ArgsManager& args, ChainstateManage opts.prevoutfetch_threads_num = std::min(*value, MAX_PREVOUTFETCH_THREADS); } + if (auto value{args.GetArg("-blockreadahead")}) { + if (*value < 0) { + return util::Error{Untranslated(strprintf("-blockreadahead must be non-negative (got %d). Use 0 to disable block read-ahead.", *value))}; + } + opts.block_readahead_depth = std::min(*value, MAX_BLOCK_READAHEAD); + } + + if (auto value{args.GetArg("-blockreadaheadthreads")}) { + if (*value < 0) { + return util::Error{Untranslated(strprintf("-blockreadaheadthreads must be non-negative (got %d). Use 0 to disable block read-ahead.", *value))}; + } + opts.block_readahead_threads = std::min(*value, MAX_BLOCK_READAHEAD_THREADS); + } + if (auto max_size = args.GetIntArg("-maxsigcachesize")) { // 1. When supplied with a max_size of 0, both the signature cache and // script execution cache create the minimum possible cache (2 diff --git a/src/validation.cpp b/src/validation.cpp index a21e8bafbd..08c47ca6e4 100644 --- a/src/validation.cpp +++ b/src/validation.cpp @@ -29,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -1883,6 +1884,8 @@ Chainstate::Chainstate( m_assumeutxo(from_snapshot_blockhash ? Assumeutxo::UNVALIDATED : Assumeutxo::VALIDATED), m_from_snapshot_blockhash(from_snapshot_blockhash) {} +Chainstate::~Chainstate() = default; + fs::path Chainstate::StoragePath() const { fs::path path{m_chainman.m_options.datadir / "chainstate"}; @@ -1946,6 +1949,14 @@ void Chainstate::InitCoinsCache(size_t cache_size_bytes) assert(m_coins_views != nullptr); m_coinstip_cache_size_bytes = cache_size_bytes; m_coins_views->InitCache(m_chainman.m_options.prevoutfetch_threads_num); + + const int32_t readahead_depth{m_chainman.m_options.block_readahead_depth}; + const int32_t readahead_threads{m_chainman.m_options.block_readahead_threads}; + m_block_readahead = std::make_unique(m_blockman); + m_block_readahead->Start(readahead_depth, readahead_threads); + if (m_block_readahead->Enabled()) { + LogInfo("Block read-ahead buffering up to %d blocks on %d threads", readahead_depth, readahead_threads); + } } // Lock-free: depends on `m_cached_is_ibd`, which is latched by `UpdateIBDStatus()`. @@ -3035,6 +3046,10 @@ bool Chainstate::ConnectTip( assert(pindexNew->pprev == m_chain.Tip()); // Read block from disk. const auto time_1{SteadyClock::now()}; + if (m_block_readahead) { + auto pre_read{m_block_readahead->Take(pindexNew)}; + if (!block_to_connect) block_to_connect = std::move(pre_read); + } if (!block_to_connect) { std::shared_ptr pblockNew = std::make_shared(); if (!m_blockman.ReadBlock(*pblockNew, *pindexNew)) { @@ -3230,6 +3245,8 @@ bool Chainstate::ActivateBestChainStep(BlockValidationState& state, CBlockIndex& fBlocksDisconnected = true; } + if (fBlocksDisconnected && m_block_readahead) m_block_readahead->Clear(); + // Build list of new blocks to connect (in descending height order). std::vector vpindexToConnect; bool fContinue = true; @@ -3247,6 +3264,10 @@ bool Chainstate::ActivateBestChainStep(BlockValidationState& state, CBlockIndex& } nHeight = nTargetHeight; + if (m_block_readahead && vpindexToConnect.size() > 1) { + m_block_readahead->Prime(vpindexToConnect); + } + // Connect new blocks. for (CBlockIndex* pindexConnect : vpindexToConnect | std::views::reverse) { if (!ConnectTip(state, pindexConnect, pindexConnect == &index_most_work ? pblock : std::shared_ptr(), connected_blocks, disconnectpool)) { diff --git a/src/validation.h b/src/validation.h index d89ad3539e..986e0cec96 100644 --- a/src/validation.h +++ b/src/validation.h @@ -63,6 +63,7 @@ namespace kernel { struct ChainstateRole; } // namespace kernel namespace node { +class BlockReadAhead; class SnapshotMetadata; } // namespace node namespace Consensus { @@ -92,6 +93,9 @@ static constexpr int MAX_SCRIPTCHECK_THREADS{15}; /** Maximum number of dedicated threads allowed for prefetching block input prevouts */ static constexpr int32_t MAX_PREVOUTFETCH_THREADS{16}; +static constexpr int32_t MAX_BLOCK_READAHEAD{64}; +static constexpr int32_t MAX_BLOCK_READAHEAD_THREADS{16}; + /** Current sync state passed to tip changed callbacks. */ enum class SynchronizationState { INIT_REINDEX, @@ -567,6 +571,8 @@ protected: //! Manages the UTXO set, which is a reflection of the contents of `m_chain`. std::unique_ptr m_coins_views; + std::unique_ptr m_block_readahead; + //! Cached result of LookupBlockIndex(*m_from_snapshot_blockhash) mutable const CBlockIndex* m_cached_snapshot_base GUARDED_BY(::cs_main){nullptr}; @@ -591,6 +597,8 @@ public: ChainstateManager& chainman, std::optional from_snapshot_blockhash = std::nullopt); + ~Chainstate(); + //! Return path to chainstate leveldb directory. fs::path StoragePath() const;