validation: prefetch blocks while connecting #36000

This is creating a new ThreadPool on each invocation of ActivateBestChain. This is what's causing the logs in #36000 (comment). The BlockFetcher or at least a shared pointer to a ThreadPool that can be passed to it should be owned by the Chainstate, so it can keep the fetcher threads alive throughout IBD.

This works fine for -reindex-chainstate, because it is one long ActivateBestChain call. But for IBD this gets called many times.

Yes, thanks @fanquake and @andrewtoth, I also just noticed that BlockFetcher was indeed recreated on every ActivateBestChain() call: I’ve pushed a fix that keeps it alive on Chainstate, and will remeasure IBD performance. Added you both as coauthors, thanks for the tests!

I’ve also reduced the default queue size from 4 to 2, since the results so far indicate that 4 isn’t worthwhile and single threaded for now for simplicity - proper HDD measurements may change this again later: <img width="235" height="201" alt="image" src="https://github.com/user-attachments/assets/3225e3ad-a924-46bb-89b4-6124cb4862b9" />

Also removed configurability to make the patch even simpler - we can add it back later if needed.

We can make this useful for reorgs here by checking if the pindexFork is not our current tip, then clearing the prefetch queue and start fetching from the fork index. We would then be fetching the sibling while we read the current tip synchronously in DisconnectTip.

Thanks, implemented. Before disconnecting the current tip, activation now clears followups for the previous candidate and fills the queue from the first sibling. This overlaps that read with DisconnectTip().

Should we just pass in height here, and internally start at height + 1? We are adding the + 1 to callsites, so this would probably be cleaner to handle it in here instead.

    void Prefetch(const CBlockIndex& index_most_work, int height) EXCLUSIVE_LOCKS_REQUIRED(::cs_main)

I think next_height is clearer here because the argument is the first height to enqueue: there's no "current" height here, it doesn't make sense to give an unrelated block height to the fetcher. But the names should be consolidated (pending vs queued vs followups vs provided vs saved), I kept renaming these and they don't match anymore.

I think this method might be more efficient as a sliding window rather than just adding all pending tasks only when the validation thread has emptied the queue. The current way has the worker thread go idle once it hits the max queue size and waits until validation catches up, whereas we can make sure the worker thread is always either working or bumping up against the max queue size.

        while (m_pending.size() < QUEUE_SIZE) {
            auto* next{index_most_work.GetAncestor(next_height + static_cast<int>(m_pending.size()))};
            if (!next || !Enqueue(*next)) break;
        }

Thanks, implemented. FillQueue() (renamed) now starts from the number of queued followups and fills every missing slot.

I originally waited for the queue to empty so adjacent reads would be submitted together (from the same file probably), which I thought might improve file-cache locality, but your measurements indicate that's not the case, so I'll remeasure. Instead of the while loop above (which might be infinite if m_pending doesn't get updated in the loop) I used a bounded loop.

What if we moved this out into a header in src/node/blockfetcher.h, and allowed passing the thread name in the constructor? Then with a simple commit on top (and shortening index thread names)

<details><summary>Index patch</summary>

diff --git a/src/index/base.cpp b/src/index/base.cpp
index 5820448bb7..4225796fc0 100644
--- a/src/index/base.cpp
+++ b/src/index/base.cpp
@@ -11,6 +11,7 @@
 #include <interfaces/types.h>
 #include <kernel/types.h>
 #include <node/abort.h>
+#include <node/blockfetcher.h>
 #include <node/blockstorage.h>
 #include <node/context.h>
 #include <node/database_args.h>
@@ -209,6 +210,7 @@ void BaseIndex::Sync()
 {
     const CBlockIndex* pindex = m_best_block_index.load();
     if (!m_synced) {
+        node::BlockFetcher fetcher{m_chainstate->m_blockman, m_thread_name};
         auto last_log_time{NodeClock::now()};
         auto last_locator_write_time{last_log_time};
         while (true) {
@@ -249,8 +251,14 @@ void BaseIndex::Sync()
             }
             pindex = pindex_next;
 
-
-            if (!ProcessBlock(pindex)) return; // error logged internally
+            std::shared_ptr<const CBlock> loaded;
+            WITH_LOCK(::cs_main, {
+                loaded = fetcher.Load(pindex->GetBlockHash());
+                if (const auto* tip{m_chainstate->m_chain.Tip()}) {
+                    fetcher.FillQueue(*tip, pindex->nHeight + 1);
+                }
+            });
+            if (!ProcessBlock(pindex, loaded.get())) return; // error logged internally
 
             auto current_time{NodeClock::now()};
             if (current_time - last_log_time >= SYNC_LOG_INTERVAL) {

</details>

we could have parallelism for all indexes. With this patch I measured speedups of 30% for txindex, 50% for blockfilterindex, and 6% for coinstatsindex. We could improve the latter two even more with an optional prefetch of the undo data as a follow-up.

cc @furszy what do you think of this approach to parallelizing the indexing code? It would stack easily with #34489 for txospenderindex and txindex; it should be as simple as increasing the threadcount and queue depth of the blockfetcher (since each append would no longer be writing so the bottleneck would just be reading/deserializing blocks).

Thanks, I like this approach. I moved BlockFetcher into src/node/blockfetcher.{h,cpp} in this push and added you as a co-author - although I initially forgot to add it to the Kernel build, and ReadBlock introduced a circular dependency. Since the extraction is useful independently of the index follow-up and makes the diff simpler, I also wired it into BaseIndex::Sync() locally to make sure the interface works. Your results make index prefetching worth doing, but I left the index wiring and configurable thread name out so we can finish the validation path in this PR first.

One detail for the follow-up is thread naming: ThreadPool adds a .xx suffix, so the current blkfltbscidx, coinstatsidx, and txospenderidx names would exceed the limit accepted by ThreadRename() before it adds b-. Since we just shortened these names in this release, maybe we should leave room for the worker suffix now, even if index prefetching remains a follow-up, to avoid renaming the threads again later (assuming each index will get their own ThreadPool, which would kind of defeat the purpose of handling shared resources).

And about the index parallelization work: this change is independent of batching index writes or parallelizing MuHash, so @furszy's ideas can still make sense on top. Before batching or adding more index concurrency, I think we should fix the concrete Stop()/Init() reader issue and account for pre-genesis sync and flush callbacks, stale-tip rewind after a disconnect, and blockfilter/coinstats crash-reorg recovery (I found several of these issues by repeatedly crashing during sync, and while some paths are theoretical and can wait, we should account for these states first).

There is also a separate follow-up for context-free validation: the prefetched block is owned by the worker until its future is consumed, so CheckBlock() can safely populate its memoization flags there, and I left a TODO at that point.

So I agree index block prefetching should be a follow-up, but before batching or parallelizing the indexes, I'd like to harden them to be more crash-safe.

I don't think there's a circular dependency. We don't need the ReadBlockFn (so we also don't need BlockFetcher to be wrapped in a std::unique_ptr):

<details><summary>Patch</summary>

diff --git a/src/node/blockfetcher.cpp b/src/node/blockfetcher.cpp
index cc8c65e182..0528622606 100644
--- a/src/node/blockfetcher.cpp
+++ b/src/node/blockfetcher.cpp
@@ -5,12 +5,11 @@
 #include <node/blockfetcher.h>
 
 #include <chain.h>
-#include <flatfile.h>
 #include <kernel/cs_main.h>
+#include <node/blockstorage.h>
 #include <primitives/block.h>
 #include <sync.h>
 #include <uint256.h>
-#include <util/expected.h>
 #include <util/threadpool.h>
 
 #include <cstddef>
@@ -31,9 +30,9 @@ std::shared_ptr<const CBlock> BlockFetcher::PopFollowup()
 bool BlockFetcher::Enqueue(const CBlockIndex& index)
 {
     if (m_pool.WorkersCount() == 0) m_pool.Start(WORKER_COUNT);
-    auto followup{m_pool.Submit([&read_block = m_read_block, hash = index.GetBlockHash(), pos = index.GetBlockPos()]() -> std::shared_ptr<const CBlock> {
+    auto followup{m_pool.Submit([&blockman = m_blockman, hash = index.GetBlockHash(), pos = index.GetBlockPos()]() -> std::shared_ptr<const CBlock> {
         auto block{std::make_shared<CBlock>()};
-        if (!read_block(*block, pos, hash)) return nullptr;
+        if (!blockman.ReadBlock(*block, pos, hash)) return nullptr;
         // TODO The block is owned by the worker until its future is consumed, so CheckBlock() may safely set its memoization flags.
         return block;
     })};
diff --git a/src/node/blockfetcher.h b/src/node/blockfetcher.h
index 88290a39af..40e489f328 100644
--- a/src/node/blockfetcher.h
+++ b/src/node/blockfetcher.h
@@ -5,33 +5,30 @@
 #ifndef BITCOIN_NODE_BLOCKFETCHER_H
 #define BITCOIN_NODE_BLOCKFETCHER_H
 
+#include <attributes.h>
 #include <kernel/cs_main.h>
 #include <sync.h>
 #include <util/threadpool.h>
 
 #include <cstdint>
 #include <deque>
-#include <functional>
 #include <future>
 #include <memory>
-#include <string>
-#include <utility>
 
 class CBlock;
 class CBlockIndex;
-struct FlatFilePos;
 class uint256;
 
 namespace node {
+class BlockManager;
+
 /** Supplies blocks to validation. Destruction waits for any queued reads. */
 class BlockFetcher
 {
-    using ReadBlockFn = std::function<bool(CBlock&, const FlatFilePos&, const uint256&)>;
-
     static constexpr uint32_t WORKER_COUNT{1};
     static constexpr uint32_t QUEUE_SIZE{2};
 
-    const ReadBlockFn m_read_block;
+    const BlockManager& m_blockman;
     ThreadPool m_pool{"blockread"};
     std::deque<std::future<std::shared_ptr<const CBlock>>> m_followups GUARDED_BY(::cs_main);
 
@@ -40,7 +37,7 @@ class BlockFetcher
     bool Enqueue(const CBlockIndex& index) EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
 
 public:
-    explicit BlockFetcher(ReadBlockFn read_block) : m_read_block{std::move(read_block)} {}
+    explicit BlockFetcher(const BlockManager& blockman LIFETIMEBOUND) : m_blockman{blockman} {}
 
     void Clear() EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
     std::shared_ptr<const CBlock> Load(const uint256& hash) EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
diff --git a/src/validation.cpp b/src/validation.cpp
index abb2a612be..be006fdd0d 100644
--- a/src/validation.cpp
+++ b/src/validation.cpp
@@ -29,7 +29,6 @@
 #include <kernel/types.h>
 #include <kernel/warning.h>
 #include <logging/timer.h>
-#include <node/blockfetcher.h>
 #include <node/blockstorage.h>
 #include <node/utxo_snapshot.h>
 #include <policy/ephemeral_policy.h>
@@ -1876,17 +1875,13 @@ Chainstate::Chainstate(
     BlockManager& blockman,
     ChainstateManager& chainman,
     std::optional<uint256> from_snapshot_blockhash)
-    : m_block_fetcher{std::make_unique<node::BlockFetcher>([blockman = &blockman](CBlock& block, const FlatFilePos& pos, const uint256& hash) {
-          return blockman->ReadBlock(block, pos, hash);
-      })},
+    : m_block_fetcher{blockman},
       m_mempool(mempool),
       m_blockman(blockman),
       m_chainman(chainman),
       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"};
@@ -3222,8 +3217,8 @@ bool Chainstate::ActivateBestChainStep(BlockValidationState& state, CBlockIndex&
     const CBlockIndex* pindexFork = m_chain.FindFork(index_most_work);
     const CBlockIndex* read_ahead_tip{provided_block ? index_most_work.pprev : &index_most_work}; // Avoid rereading the provided block
     if (pindexFork && pindexFork != pindexOldTip) {
-        m_block_fetcher->Clear();
-        if (read_ahead_tip) m_block_fetcher->FillQueue(*read_ahead_tip, pindexFork->nHeight + 1);
+        m_block_fetcher.Clear();
+        if (read_ahead_tip) m_block_fetcher.FillQueue(*read_ahead_tip, pindexFork->nHeight + 1);
     }
 
     // Disconnect active blocks which are no longer in the best chain.
@@ -3263,8 +3258,8 @@ bool Chainstate::ActivateBestChainStep(BlockValidationState& state, CBlockIndex&
 
         // Connect new blocks.
         for (CBlockIndex* pindexConnect : vpindexToConnect | std::views::reverse) {
-            auto block_to_connect{provided_block && pindexConnect == &index_most_work ? provided_block : m_block_fetcher->Load(pindexConnect->GetBlockHash())};
-            if (read_ahead_tip) m_block_fetcher->FillQueue(*read_ahead_tip, pindexConnect->nHeight + 1);
+            auto block_to_connect{provided_block && pindexConnect == &index_most_work ? provided_block : m_block_fetcher.Load(pindexConnect->GetBlockHash())};
+            if (read_ahead_tip) m_block_fetcher.FillQueue(*read_ahead_tip, pindexConnect->nHeight + 1);
             if (!ConnectTip(state, pindexConnect, std::move(block_to_connect), connected_blocks, disconnectpool)) {
                 if (state.IsInvalid()) {
                     // The block violates a consensus rule.
diff --git a/src/validation.h b/src/validation.h
index 2c67d1aa32..2793ccb92b 100644
--- a/src/validation.h
+++ b/src/validation.h
@@ -18,6 +18,7 @@
 #include <kernel/chainparams.h>
 #include <kernel/chainstatemanager_opts.h>
 #include <kernel/cs_main.h> // IWYU pragma: export
+#include <node/blockfetcher.h>
 #include <node/blockstorage.h>
 #include <policy/feerate.h>
 #include <policy/packages.h>
@@ -63,7 +64,6 @@ namespace kernel {
 struct ChainstateRole;
 } // namespace kernel
 namespace node {
-class BlockFetcher;
 class SnapshotMetadata;
 } // namespace node
 namespace Consensus {
@@ -562,7 +562,7 @@ protected:
     Mutex m_chainstate_mutex;
 
     //! Reads blocks ahead during chain activation.
-    std::unique_ptr<node::BlockFetcher> m_block_fetcher;
+    node::BlockFetcher m_block_fetcher;
 
     //! Optional mempool that is kept in sync with the chain.
     //! Only the active chainstate has a mempool.
@@ -594,7 +594,6 @@ public:
         node::BlockManager& blockman,
         ChainstateManager& chainman,
         std::optional<uint256> from_snapshot_blockhash = std::nullopt);
-    ~Chainstate();
 
     //! Return path to chainstate leveldb directory.
     fs::path StoragePath() const;

</details>

The full patch still introduces the exact cycle: node/blockfetcher -> node/blockstorage -> validation -> node/blockfetcher, can you please check it with lint-circular-dependencies.py?

Ahh I did not use the linter, I just compiled. Indeed, my patch fails lint. Thanks!

Does this warrant a release note? It is covered already by "performance improvements" in every release note.

I expect this to land in a future release, so for now I'd keep it

nit: we can remove the dates altogether. same for header file.

// Copyright (c) The Bitcoin Core developers

Not sure we should have this TODO here. I don't know if there is consensus for this yet. Changing the validation ordering is a different beast than just parallelizing reading a block from disk and deserializing.

Could we pass these as default parameters to the constructor, and call ThreadPool::Start on construction? That way we can inject different values. Why do we wait to start the threadpool later? IMO it's cleaner/more RAII to start in constructor.

We should call Clear here, since on a miss we have gotten out of sync of the fetching.

Since we have multiple worker threads now, maybe we can take advantage of the multi-Submit overload whenever we have an empty m_followups? That would awaken all threads at once. This would happen on every ActivateBestChainStep call, so could show some improvement?

Also, if we have multiple threads and queue depth, we might want to allow disabling this mechanism for low memory systems. Maybe worth it to just be an on/off toggle (that defaults to on)? Or maybe a configuration of threads only and the queue depth is 2*threads? Just trying to think of ways to make it simple instead of adding all kinds of configuration options.

The release notes are stale. There are now multiple threads.

This comment is not very helpful IMO. Why would anyone care about destruction waiting on queued reads? Something like this would be better.

/**
 * Reads and deserializes blocks in parallel starting from a requested block index.
 * Useful for improving performance when scanning blocks.
 **/

This comment is a bit cryptic.

    // If we have a provided block, we don't need to read it so we can stop reading ahead at the previous block.
    const CBlockIndex* read_ahead_tip{provided_block ? index_most_work.pprev : &index_most_work};