wallet: check blockfilters in parallel

This commit implements parallel block filter checking using the wallet
threadpool. The main thread reads block hashes and queues them while
worker threads check filters in parallel.

Synchronization:
- Operations requiring cs_wallet (GetLastBlockHeight, SyncTransaction)
  remain on the main thread since cs_wallet is a RecursiveMutex and
  ScanForWalletTransactions is called from AttachChain which locks cs_wallet
- Main thread uses ThreadPool::ProcessTask() to join workers when the
  block queue is full, avoiding busy-waiting

Batching:
- Up to m_max_blockqueue_size (1000) blocks are queued for filtering
- When queue is full, main loop processes filtered blocks before reading more
Thread safety:
- All futures (at most `workers_count`)  are waited on before returning to
  avoid data races on `FastWalletRescanFilter::m_filter_set`.

Benchmarks show considerable improvement (approx 5x with 16 threads).

Author: Eunovo

src/wallet/scan.cpp

@@ -8,6 +8,9 @@
 #include <wallet/scan.h>
 #include <wallet/wallet.h>
 
+#include <algorithm>
+#include <future>
+
 using interfaces::FoundBlock;
 
 namespace wallet {
@@ -130,30 +133,100 @@ std::optional<std::pair<uint256, int>> ChainScanner::ReadNextBlock(ScanResult& r
     return std::make_pair(block_hash, block_height);
 }
 
-std::optional<std::pair<size_t, size_t>> ChainScanner::ReadNextBlocks(const std::unique_ptr<FastWalletRescanFilter>& filter, ScanResult& result) {
-    auto next_block = ReadNextBlock(result);
-    if (next_block) m_blocks.emplace_back(*next_block);
-    else return std::nullopt;
+enum FilterRes {
+    FILTER_NO_MATCH,
+    FILTER_MATCH,
+    FILTER_NO_FILTER
+};
 
+std::optional<std::pair<size_t, size_t>> ChainScanner::ReadNextBlocks(const std::unique_ptr<FastWalletRescanFilter>& filter, ScanResult& result) {
     if (!filter) {
-        // Slow scan: scan all blocks.
-        return std::make_pair<size_t, size_t>(0, 1);
+        // Slow scan: no filter, scan all blocks
+        auto next_block = ReadNextBlock(result);
+        if (next_block) m_blocks.emplace_back(*next_block);
+        return std::make_pair<size_t, size_t>(0, m_blocks.size());
     }
-
     filter->UpdateIfNeeded();
+    auto* thread_pool = m_wallet.m_thread_pool;
+    // ThreadPool pointer should never be null here
+    // during normal operation because it should
+    // have been passed from the WalletContext to
+    // the CWallet constructor
+    assert(thread_pool != nullptr);
+    std::optional<std::pair<size_t, size_t>> range;
+    size_t workers_count = thread_pool->WorkersCount();
+    size_t i = 0;
+    size_t completed = 0;
+    std::vector<std::future<FilterRes>> futures;
+    futures.reserve(workers_count);
+
+    while ((m_next_block && completed < m_max_blockqueue_size) || i < m_blocks.size() || !futures.empty()) {
+        const bool result_ready = futures.size() > 0 && futures[0].wait_for(std::chrono::seconds::zero()) == std::future_status::ready;
+        if (result_ready) {
+            const auto& scan_res{futures[0].get()};
+            completed++;
+            futures.erase(futures.begin());
+            if (scan_res == FILTER_NO_MATCH) {
+                if (range.has_value()) {
+                    break;
+                }
+                continue;
+            }
+            int current_block_index = completed - 1;
+            if (scan_res == FilterRes::FILTER_MATCH) {
+                LogDebug(BCLog::SCAN, "Fast rescan: inspect block %d [%s] (filter matched)\n", m_blocks[current_block_index].second, m_blocks[current_block_index].first.ToString());
+            } else { // FILTER_NO_FILTER
+                LogDebug(BCLog::SCAN, "Fast rescan: inspect block %d [%s] (WARNING: block filter not found!)\n", m_blocks[current_block_index].second, m_blocks[current_block_index].first.ToString());
+            }
 
-    const auto& [block_hash, block_height] = m_blocks[0];
-    auto matches_block{filter->MatchesBlock(block_hash)};
+            if (!range.has_value()) range = std::make_pair(current_block_index, current_block_index + 1);
+            else range->second = current_block_index + 1;
+        }
+
+        // Submit jobs to the threadpool in batches of at most `workers_count` size.
+        // This prevents over-submission: if we queued all jobs upfront and the filtered
+        // block range is smaller than expected, worker threads would process blocks
+        // that get discarded, wasting CPU cycles.
+        const size_t job_gap = workers_count - futures.size();
+        if (job_gap > 0 && i < m_blocks.size()) {
+            for (size_t j = 0; j < job_gap && i < m_blocks.size(); ++j, ++i) {
+                auto block = m_blocks[i];
+                futures.emplace_back(*thread_pool->Submit([&filter, block = std::move(block)]() {
+                    const auto matches_block{filter->MatchesBlock(block.first)};
+                    if (matches_block.has_value()) {
+                        if (*matches_block) {
+                            return FilterRes::FILTER_MATCH;
+                        } else {
+                            return FilterRes::FILTER_NO_MATCH;
+                        }
+                    } else {
+                        return FilterRes::FILTER_NO_FILTER;
+                    }
+                }));
+            }
+        }
+
+        // If m_max_blockqueue_size blocks have been filtered,
+        // stop reading more blocks for now, to give the
+        // main scanning loop a chance to update progress
+        // and erase some blocks from the queue.
+        if (m_next_block && completed < m_max_blockqueue_size) {
+            auto next_block = ReadNextBlock(result);
+            if (next_block) m_blocks.emplace_back(*next_block);
+        }
+        else if (!futures.empty()) {
+            // Join work processing instead of waiting idly.
+            thread_pool->ProcessTask();
+        }
+    }
 
-    if (matches_block.has_value() && *matches_block) {
-        LogDebug(BCLog::SCAN, "Fast rescan: inspect block %d [%s] (filter matched)\n", block_height, block_hash.ToString());
-        return std::make_pair<size_t, size_t>(0, 1);
-    } else if (!matches_block.has_value()) {
-        LogDebug(BCLog::SCAN, "Fast rescan: inspect block %d [%s] (WARNING: block filter not found!)\n", block_height, block_hash.ToString());
-        return std::make_pair<size_t, size_t>(0, 1);
+    for (auto& fut : futures) {
+        // Wait for all remaining futures to complete
+        // to avoid data race on FastWalletRescanFilter::m_filter_set
+        fut.wait();
     }
 
-    return std::nullopt;
+    return range;
 }
 
 void ChainScanner::UpdateProgress(int block_height) {

src/wallet/wallet.h

@@ -408,7 +408,7 @@ class CWallet final : public WalletStorage, public interfaces::Chain::Notificati
     std::unique_ptr<WalletDatabase> m_database;
 
     /** Thread pool for wallet operations. */
-    [[maybe_unused]] ThreadPool* m_thread_pool;
+    ThreadPool* m_thread_pool;
 
     /**
      * The following is used to keep track of how far behind the wallet is