diff --git a/src/index/base.cpp b/src/index/base.cpp index 82259ac046a..f34d47b0b1e 100644 --- a/src/index/base.cpp +++ b/src/index/base.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -20,6 +21,7 @@ #include #include +#include #include #include #include @@ -149,7 +151,7 @@ static const CBlockIndex* NextSyncBlock(const CBlockIndex* pindex_prev, CChain& return chain.Next(chain.FindFork(pindex_prev)); } -bool BaseIndex::ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data) +std::any BaseIndex::ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data) { interfaces::BlockInfo block_info = kernel::MakeBlockInfo(pindex, block_data); @@ -158,7 +160,7 @@ bool BaseIndex::ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data if (!m_chainstate->m_blockman.ReadBlock(block, *pindex)) { FatalErrorf("Failed to read block %s from disk", pindex->GetBlockHash().ToString()); - return false; + return {}; } block_info.data = █ } @@ -168,87 +170,292 @@ bool BaseIndex::ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data if (pindex->nHeight > 0 && !m_chainstate->m_blockman.ReadBlockUndo(block_undo, *pindex)) { FatalErrorf("Failed to read undo block data %s from disk", pindex->GetBlockHash().ToString()); - return false; + return {}; } block_info.undo_data = &block_undo; } - if (!CustomAppend(block_info)) { - FatalErrorf("Failed to write block %s to index database", - pindex->GetBlockHash().ToString()); - return false; + const auto& any_obj = CustomProcessBlock(block_info); + if (!any_obj.has_value()) { + FatalErrorf("Failed to process block %s for index %s", + pindex->GetBlockHash().GetHex(), GetName()); + return {}; } - - return true; + return any_obj; } +std::vector BaseIndex::ProcessBlocks(bool process_in_order, const CBlockIndex* start, const CBlockIndex* end) +{ + std::vector results; + if (process_in_order) { + // When ordering is required, collect all block indexes from [end..start] in order + std::vector ordered_blocks; + for (const CBlockIndex* block = end; block && start->pprev != block; block = block->pprev) { + ordered_blocks.emplace_back(block); + } + + // And process blocks in forward order: from start to end + for (auto it = ordered_blocks.rbegin(); it != ordered_blocks.rend(); ++it) { + auto op_res = ProcessBlock(*it); + if (!op_res.has_value()) return {}; + results.emplace_back(std::move(op_res)); + } + return results; + } + + // If ordering is not required, process blocks directly from end to start + for (const CBlockIndex* block = end; block && start->pprev != block; block = block->pprev) { + auto op_res = ProcessBlock(block); + if (!op_res.has_value()) return {}; + results.emplace_back(std::move(op_res)); + } + + return results; +} + +struct Task { + int id; + const CBlockIndex* start_index; + const CBlockIndex* end_index; + std::vector result; + + Task(int task_id, const CBlockIndex* start, const CBlockIndex* end) + : id(task_id), start_index(start), end_index(end) {} + + // Disallow copy + Task(const Task&) = delete; + Task& operator=(const Task&) = delete; + Task(Task&&) noexcept = default; +}; + +// Context shared across the initial sync workers +struct SyncContext { + Mutex mutex_pending_tasks; + std::queue pending_tasks GUARDED_BY(mutex_pending_tasks); + + Mutex mutex_processed_tasks; + std::map processed_tasks GUARDED_BY(mutex_processed_tasks); + + std::atomic next_id_to_process{0}; +}; + +// Synchronizes the index with the active chain. +// +// If parallel sync is enabled, this method uses WorkersCount()+1 threads (including the current thread) +// to process block ranges concurrently. Each worker handles up to 'm_blocks_per_worker' blocks each time +// (this is called a "task"), which are processed via CustomProcessBlock calls. Results are stored in the +// SyncContext's 'processed_tasks' map so they can be sequentially post-processed later. +// +// After completing a task, workers opportunistically post-process completed tasks *in order* using +// CustomPostProcessBlocks. This continues until all blocks have been fully processed and committed. +// +// Reorgs are detected and handled before syncing begins, ensuring the index starts aligned with the active chain. void BaseIndex::Sync() { + if (m_synced) return; // we are synced, nothing to do + + // Before anything, verify we are in the active chain const CBlockIndex* pindex = m_best_block_index.load(); - if (!m_synced) { - auto last_log_time{NodeClock::now()}; - auto last_locator_write_time{last_log_time}; - while (true) { - if (m_interrupt) { - LogInfo("%s: m_interrupt set; exiting ThreadSync", GetName()); + const int tip_height = WITH_LOCK(cs_main, return m_chainstate->m_chain.Height()); + // Note: be careful, could return null if there is no more work to do or if pindex is not found (erased blocks dir). + const CBlockIndex* pindex_next = WITH_LOCK(cs_main, return NextSyncBlock(pindex, m_chainstate->m_chain)); + if (!pindex_next) { + m_synced = true; + return; + } - SetBestBlockIndex(pindex); - // No need to handle errors in Commit. If it fails, the error will be already be - // logged. The best way to recover is to continue, as index cannot be corrupted by - // a missed commit to disk for an advanced index state. - Commit(); - return; - } + // Handle potential reorgs; if the next block's parent doesn't match our current tip, + // rewind our index state to match the chain and resume from there. + if (pindex_next->pprev != pindex && !Rewind(pindex, pindex_next->pprev)) { + FatalErrorf("Failed to rewind index %s to a previous chain tip", GetName()); + return; + } - const CBlockIndex* pindex_next = WITH_LOCK(cs_main, return NextSyncBlock(pindex, m_chainstate->m_chain)); - // If pindex_next is null, it means pindex is the chain tip, so - // commit data indexed so far. - if (!pindex_next) { - SetBestBlockIndex(pindex); - // No need to handle errors in Commit. See rationale above. - Commit(); + // Compute tasks ranges + const int blocks_to_sync = tip_height - pindex_next->nHeight; + const int num_tasks = blocks_to_sync / m_blocks_per_worker; + const int remaining_blocks = blocks_to_sync % m_blocks_per_worker; + const bool process_in_order = !AllowParallelSync(); - // If pindex is still the chain tip after committing, exit the - // sync loop. It is important for cs_main to be locked while - // setting m_synced = true, otherwise a new block could be - // attached while m_synced is still false, and it would not be - // indexed. - LOCK(::cs_main); - pindex_next = NextSyncBlock(pindex, m_chainstate->m_chain); - if (!pindex_next) { - m_synced = true; - break; - } - } - if (pindex_next->pprev != pindex && !Rewind(pindex, pindex_next->pprev)) { - FatalErrorf("Failed to rewind %s to a previous chain tip", GetName()); - return; - } - pindex = pindex_next; + std::shared_ptr ctx = std::make_shared(); + { + LOCK2(ctx->mutex_pending_tasks, ::cs_main); + // Create fixed-size tasks + const CBlockIndex* it_start = pindex; + const CBlockIndex* it_end; + for (int id = 0; id < num_tasks; ++id) { + it_start = Assert(NextSyncBlock(it_start, m_chainstate->m_chain)); + it_end = m_chainstate->m_chain[it_start->nHeight + m_blocks_per_worker - 1]; + ctx->pending_tasks.emplace(id, it_start, it_end); + it_start = it_end; + } - - if (!ProcessBlock(pindex)) return; // error logged internally - - auto current_time{NodeClock::now()}; - if (current_time - last_log_time >= SYNC_LOG_INTERVAL) { - LogInfo("Syncing %s with block chain from height %d", GetName(), pindex->nHeight); - last_log_time = current_time; - } - - if (current_time - last_locator_write_time >= SYNC_LOCATOR_WRITE_INTERVAL) { - SetBestBlockIndex(pindex); - last_locator_write_time = current_time; - // No need to handle errors in Commit. See rationale above. - Commit(); - } + // Add final task with the remaining blocks, if any + if (remaining_blocks > 0) { + it_start = Assert(NextSyncBlock(it_start, m_chainstate->m_chain)); + it_end = m_chainstate->m_chain[it_start->nHeight + remaining_blocks]; + ctx->pending_tasks.emplace(/*task_id=*/num_tasks, it_start, it_end); } } - if (pindex) { - LogInfo("%s is enabled at height %d", GetName(), pindex->nHeight); - } else { - LogInfo("%s is enabled", GetName()); + // Returns nullopt only when there are no pending tasks + const auto& try_get_task = [](auto& ctx) -> std::optional { + LOCK(ctx->mutex_pending_tasks); + if (ctx->pending_tasks.empty()) return std::nullopt; + Task t = std::move(ctx->pending_tasks.front()); + ctx->pending_tasks.pop(); + return t; + }; + + enum class WorkerStatus { ABORT, PROCESSING, FINISHED }; + + const auto& func_worker = [this, try_get_task, process_in_order](auto& ctx) -> WorkerStatus { + if (m_interrupt) return WorkerStatus::FINISHED; + + // Try to obtain a task and process it + if (std::optional maybe_task = try_get_task(ctx)) { + Task task = std::move(*maybe_task); + task.result = ProcessBlocks(process_in_order, task.start_index, task.end_index); + if (task.result.empty()) { + // Empty result indicates an internal error (logged internally). + m_interrupt(); // notify other workers and abort. + return WorkerStatus::ABORT; + } + + LOCK(ctx->mutex_processed_tasks); + ctx->processed_tasks.emplace(task.id, std::move(task)); + } else { + // No pending tasks — might be finished + // If we still have processed task to consume, proceed to finalize them. + LOCK(ctx->mutex_processed_tasks); + if (ctx->processed_tasks.empty()) return WorkerStatus::FINISHED; + } + + // Post-process completed tasks opportunistically + std::vector to_process; + { + TRY_LOCK(ctx->mutex_processed_tasks, locked); + if (!locked) return WorkerStatus::PROCESSING; + + // Collect ready-to-process tasks in order + int next_id = ctx->next_id_to_process.load(); + while (true) { + auto it = ctx->processed_tasks.find(next_id); + if (it == ctx->processed_tasks.end()) break; + to_process.push_back(std::move(it->second)); + ctx->processed_tasks.erase(it); + ++next_id; + } + + // Nothing to process right now, keep processing + if (to_process.empty()) return WorkerStatus::PROCESSING; + } + + // Post-Process tasks + for (const Task& task : to_process) { + for (auto it = task.result.rbegin(); it != task.result.rend(); ++it) { + if (!CustomPostProcessBlocks(*it)) { // error logged internally + m_interrupt(); + FatalErrorf("Index %s: Failed to post process blocks", GetName()); + return WorkerStatus::ABORT; + } + } + // Update progress + SetBestBlockIndex(task.end_index); + ctx->next_id_to_process.store(task.id + 1); + } + + // Check if there's anything left to do + LOCK2(ctx->mutex_pending_tasks, ctx->mutex_processed_tasks); + if (ctx->pending_tasks.empty() && ctx->processed_tasks.empty()) { + return WorkerStatus::FINISHED; + } + + return WorkerStatus::PROCESSING; + }; + + // Process task in parallel if enabled + std::vector> workers_job; + if (m_thread_pool) { + for (size_t i = 0; i < m_thread_pool->WorkersCount(); ++i) { + workers_job.emplace_back(m_thread_pool->Submit([this, ctx, func_worker]() { + WorkerStatus status{WorkerStatus::PROCESSING}; + while (!m_synced && status == WorkerStatus::PROCESSING) { + status = func_worker(ctx); + if (m_interrupt) return; + } + })); + } } + + // Main index thread + // Active-wait: we process blocks in this thread too. + auto last_log_time{NodeClock::now()}; + auto last_locator_write_time{last_log_time}; + + while (!m_synced) { + const WorkerStatus status{func_worker(ctx)}; + if (m_interrupt || status == WorkerStatus::ABORT) { + m_interrupt(); + // Ensure all workers are interrupted before returning. + // This avoids accessing any local variable post-destruction. + for (const auto& job : workers_job) job.wait(); + return; + } + + if (status == WorkerStatus::FINISHED) { + // No more tasks to process; wait for all workers to finish their current tasks + for (const auto& job : workers_job) job.wait(); + // No need to handle errors in Commit. If it fails, the error will be already be + // logged. The best way to recover is to continue, as index cannot be corrupted by + // a missed commit to disk for an advanced index state. + Commit(); + + // Before finishing, check if any new blocks were connected while we were syncing. + // If so, process them synchronously before exiting. + // + // Note: it is important for cs_main to be locked while setting m_synced = true, + // otherwise a new block could be attached while m_synced is still false, and + // it would not be indexed. + LOCK2(ctx->mutex_pending_tasks, ::cs_main); + const CBlockIndex* curr_tip{m_best_block_index.load()}; + pindex_next = NextSyncBlock(curr_tip, m_chainstate->m_chain); + // If the next block is null, it means we are done! + if (!pindex_next) { + m_synced = true; + break; + } + + // New blocks arrived during sync. + // Handle potential reorgs; if the next block's parent doesn't match our tip, + // rewind index state to the correct chain, then resume. + if (pindex_next->pprev != curr_tip && !Rewind(curr_tip, pindex_next->pprev)) { + FatalErrorf("Failed to rewind index %s to a previous chain tip", GetName()); + return; + } + + // Queue the final range of blocks to process. + ctx->pending_tasks.emplace(ctx->next_id_to_process.load(), + /*start_index=*/pindex_next, + /*end_index=*/m_chainstate->m_chain.Tip()); + } + + auto current_time{NodeClock::now()}; + // Log progress every so often + if (current_time - last_log_time >= SYNC_LOG_INTERVAL) { + LogInfo("Syncing %s with block chain from height %d\n", + GetName(), m_best_block_index.load()->nHeight); + last_log_time = current_time; + } + + // Commit changes every so often + if (current_time - last_locator_write_time >= SYNC_LOCATOR_WRITE_INTERVAL) { + Commit(); // No need to handle errors in Commit. See rationale above. + last_locator_write_time = current_time; + } + } + + LogInfo("%s is enabled at height %d\n", GetName(), (m_best_block_index) ? m_best_block_index.load()->nHeight : 0); } bool BaseIndex::Commit() @@ -273,6 +480,7 @@ bool BaseIndex::Commit() bool BaseIndex::Rewind(const CBlockIndex* current_tip, const CBlockIndex* new_tip) { + assert(current_tip == m_best_block_index); assert(current_tip->GetAncestor(new_tip->nHeight) == new_tip); CBlock block; @@ -354,7 +562,7 @@ void BaseIndex::BlockConnected(ChainstateRole role, const std::shared_ptr #include +#include #include class CBlock; class CBlockIndex; class Chainstate; class ChainstateManager; +class ThreadPool; namespace interfaces { class Chain; } // namespace interfaces +/** Maximum number of threads a single thread pool instance can have */ +static constexpr int16_t MAX_INDEX_WORKERS_COUNT = 100; +/** Number of concurrent jobs during the initial sync process */ +static constexpr int16_t INDEX_WORKERS_COUNT = 0; +/** Number of tasks processed by each worker */ +static constexpr int16_t INDEX_WORK_PER_CHUNK = 1000; + struct IndexSummary { std::string name; bool synced{false}; @@ -80,6 +89,9 @@ private: std::thread m_thread_sync; CThreadInterrupt m_interrupt; + ThreadPool* m_thread_pool{nullptr}; + int m_blocks_per_worker{INDEX_WORK_PER_CHUNK}; + /// Write the current index state (eg. chain block locator and subclass-specific items) to disk. /// /// Recommendations for error handling: @@ -93,7 +105,8 @@ private: /// Loop over disconnected blocks and call CustomRemove. bool Rewind(const CBlockIndex* current_tip, const CBlockIndex* new_tip); - bool ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data = nullptr); + std::any ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data = nullptr); + std::vector ProcessBlocks(bool process_in_order, const CBlockIndex* start, const CBlockIndex* end); virtual bool AllowPrune() const = 0; @@ -130,6 +143,26 @@ protected: /// Update the internal best block index as well as the prune lock. void SetBestBlockIndex(const CBlockIndex* block); + /// If 'AllowParallelSync()' returns true, 'ProcessBlock()' will run concurrently in batches. + /// The 'std::any' result will be passed to 'CustomPostProcessBlocks()' so the index can process + /// async result batches in a synchronous fashion (if required). + [[nodiscard]] virtual std::any CustomProcessBlock(const interfaces::BlockInfo& block_info) { + // If parallel sync is enabled, the child class must implement this method. + if (AllowParallelSync()) return std::any(); + + // Default, synchronous write + if (!CustomAppend(block_info)) { + throw std::runtime_error(strprintf("%s: Failed to write block %s to index database", + __func__, block_info.hash.ToString())); + } + return true; + } + + /// 'CustomPostProcessBlocks()' is called in a synchronous manner after a batch of async 'ProcessBlock()' + /// calls have completed. + /// Here the index usually links and dump information that cannot be processed in an asynchronous fashion. + [[nodiscard]] virtual bool CustomPostProcessBlocks(const std::any& obj) { return true; }; + public: BaseIndex(std::unique_ptr chain, std::string name); /// Destructor interrupts sync thread if running and blocks until it exits. @@ -138,6 +171,8 @@ public: /// Get the name of the index for display in logs. const std::string& GetName() const LIFETIMEBOUND { return m_name; } + void SetThreadPool(ThreadPool& thread_pool) { m_thread_pool = &thread_pool; } + /// Blocks the current thread until the index is caught up to the current /// state of the block chain. This only blocks if the index has gotten in /// sync once and only needs to process blocks in the ValidationInterface @@ -164,6 +199,12 @@ public: /// Stops the instance from staying in sync with blockchain updates. void Stop(); + /// Number of blocks each worker thread will process at a time + void SetBlocksPerWorker(int count) { m_blocks_per_worker = count; } + + /// True if the child class allows concurrent sync. + virtual bool AllowParallelSync() { return false; } + /// Get a summary of the index and its state. IndexSummary GetSummary() const; }; diff --git a/src/index/blockfilterindex.cpp b/src/index/blockfilterindex.cpp index 2ccae3a221b..5b6ed8d2f4d 100644 --- a/src/index/blockfilterindex.cpp +++ b/src/index/blockfilterindex.cpp @@ -299,6 +299,23 @@ bool BlockFilterIndex::Write(const BlockFilter& filter, uint32_t block_height, c return true; } +std::any BlockFilterIndex::CustomProcessBlock(const interfaces::BlockInfo& block_info) +{ + return std::make_pair(BlockFilter(m_filter_type, *block_info.data, *block_info.undo_data), block_info.height); +} + +bool BlockFilterIndex::CustomPostProcessBlocks(const std::any& obj) +{ + const auto& [filter, height] = std::any_cast>(obj); + const uint256& header = filter.ComputeHeader(m_last_header); + if (!Write(filter, height, header)) { + LogError("Error writing filters, shutting down block filters index\n"); + return false; + } + m_last_header = header; + return true; +} + [[nodiscard]] static bool CopyHeightIndexToHashIndex(CDBIterator& db_it, CDBBatch& batch, const std::string& index_name, int height) { diff --git a/src/index/blockfilterindex.h b/src/index/blockfilterindex.h index 983cf32206a..3ad27102b24 100644 --- a/src/index/blockfilterindex.h +++ b/src/index/blockfilterindex.h @@ -64,6 +64,9 @@ protected: BaseIndex::DB& GetDB() const LIFETIMEBOUND override { return *m_db; } + std::any CustomProcessBlock(const interfaces::BlockInfo& block) override; + bool CustomPostProcessBlocks(const std::any& obj) override; + public: /** Constructs the index, which becomes available to be queried. */ explicit BlockFilterIndex(std::unique_ptr chain, BlockFilterType filter_type, @@ -71,6 +74,8 @@ public: BlockFilterType GetFilterType() const { return m_filter_type; } + bool AllowParallelSync() override { return true; } + /** Get a single filter by block. */ bool LookupFilter(const CBlockIndex* block_index, BlockFilter& filter_out) const; diff --git a/src/index/txindex.h b/src/index/txindex.h index f8236c92844..4ecb1d34184 100644 --- a/src/index/txindex.h +++ b/src/index/txindex.h @@ -29,6 +29,10 @@ protected: BaseIndex::DB& GetDB() const override; + std::any CustomProcessBlock(const interfaces::BlockInfo& block) override { + return CustomAppend(block); + } + public: /// Constructs the index, which becomes available to be queried. explicit TxIndex(std::unique_ptr chain, size_t n_cache_size, bool f_memory = false, bool f_wipe = false); @@ -36,6 +40,8 @@ public: // Destructor is declared because this class contains a unique_ptr to an incomplete type. virtual ~TxIndex() override; + bool AllowParallelSync() override { return true; } + /// Look up a transaction by hash. /// /// @param[in] tx_hash The hash of the transaction to be returned. diff --git a/src/init.cpp b/src/init.cpp index dadd56d2469..cf90730acb5 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -87,6 +87,7 @@ #include #include #include +#include #include #include #include @@ -359,10 +360,12 @@ void Shutdown(NodeContext& node) // Stop and delete all indexes only after flushing background callbacks. for (auto* index : node.indexes) index->Stop(); + if (node.m_index_threads) node.m_index_threads->Stop(); if (g_txindex) g_txindex.reset(); if (g_coin_stats_index) g_coin_stats_index.reset(); DestroyAllBlockFilterIndexes(); node.indexes.clear(); // all instances are nullptr now + if (node.m_index_threads) node.m_index_threads.reset(); // Any future callbacks will be dropped. This should absolutely be safe - if // missing a callback results in an unrecoverable situation, unclean shutdown @@ -530,6 +533,8 @@ void SetupServerArgs(ArgsManager& argsman, bool can_listen_ipc) strprintf("Maintain an index of compact filters by block (default: %s, values: %s).", DEFAULT_BLOCKFILTERINDEX, ListBlockFilterTypes()) + " If is not supplied or if = 1, indexes for all known types are enabled.", ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS); + argsman.AddArg("-indexworkers=", strprintf("Number of worker threads spawned for the initial index synchronization (default: %d). These threads are shared across all indexes. " + "Improves indexing speed on fast storage but may slow indexing on HDDs due to additional disk seeks.", INDEX_WORKERS_COUNT), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS); argsman.AddArg("-addnode=", strprintf("Add a node to connect to and attempt to keep the connection open (see the addnode RPC help for more info). This option can be specified multiple times to add multiple nodes; connections are limited to %u at a time and are counted separately from the -maxconnections limit.", MAX_ADDNODE_CONNECTIONS), ArgsManager::ALLOW_ANY | ArgsManager::NETWORK_ONLY, OptionsCategory::CONNECTION); argsman.AddArg("-asmap=", strprintf("Specify asn mapping used for bucketing of the peers (default: %s). Relative paths will be prefixed by the net-specific datadir location.", DEFAULT_ASMAP_FILENAME), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); @@ -2214,6 +2219,8 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info) bool StartIndexBackgroundSync(NodeContext& node) { + if (node.indexes.empty()) return true; + // Find the oldest block among all indexes. // This block is used to verify that we have the required blocks' data stored on disk, // starting from that point up to the current tip. @@ -2252,7 +2259,19 @@ bool StartIndexBackgroundSync(NodeContext& node) } } + if (node.args->IsArgSet("-indexworkers")) { + int index_workers = node.args->GetIntArg("-indexworkers", INDEX_WORKERS_COUNT); + if (index_workers < 0 || index_workers > MAX_INDEX_WORKERS_COUNT) return InitError(Untranslated(strprintf("Invalid -indexworkers arg. Must be a number between 0 and %d", MAX_INDEX_WORKERS_COUNT))); + + node.m_index_threads = std::make_unique("indexes"); + node.m_index_threads->Start(index_workers); + } + // Start threads - for (auto index : node.indexes) if (!index->StartBackgroundSync()) return false; + for (auto index : node.indexes) { + // Provide thread pool to indexes + if (node.m_index_threads && index->AllowParallelSync()) index->SetThreadPool(*node.m_index_threads); + if (!index->StartBackgroundSync()) return false; + } return true; } diff --git a/src/node/context.h b/src/node/context.h index debc1221206..e808f1950bc 100644 --- a/src/node/context.h +++ b/src/node/context.h @@ -5,6 +5,8 @@ #ifndef BITCOIN_NODE_CONTEXT_H #define BITCOIN_NODE_CONTEXT_H +#include + #include #include #include @@ -90,6 +92,7 @@ struct NodeContext { //! Manages all the node warnings std::unique_ptr warnings; std::thread background_init_thread; + std::unique_ptr m_index_threads; //! Declare default constructor and destructor that are not inline, so code //! instantiating the NodeContext struct doesn't need to #include class diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index b4c1ec69339..0433be83b48 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -106,6 +106,7 @@ add_executable(test_bitcoin system_ram_tests.cpp system_tests.cpp testnet4_miner_tests.cpp + threadpool_tests.cpp timeoffsets_tests.cpp torcontrol_tests.cpp transaction_tests.cpp diff --git a/src/test/blockfilter_index_tests.cpp b/src/test/blockfilter_index_tests.cpp index 224acb8b079..0499322eb40 100644 --- a/src/test/blockfilter_index_tests.cpp +++ b/src/test/blockfilter_index_tests.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include @@ -269,6 +270,50 @@ BOOST_FIXTURE_TEST_CASE(blockfilter_index_initial_sync, BuildChainTestingSetup) filter_index.Stop(); } +BOOST_FIXTURE_TEST_CASE(blockfilter_index_parallel_initial_sync, BuildChainTestingSetup) +{ + int tip_height = 100; // pre-mined blocks + const uint16_t MINE_BLOCKS = 650; + for (int round = 0; round < 2; round++) { // two rounds to test sync from genesis and from a higher block + // Generate blocks + mineBlocks(MINE_BLOCKS); + const CBlockIndex* tip = WITH_LOCK(::cs_main, return m_node.chainman->ActiveChain().Tip()); + BOOST_REQUIRE(tip->nHeight == MINE_BLOCKS + tip_height); + tip_height = tip->nHeight; + + // Init index + BlockFilterIndex filter_index(interfaces::MakeChain(m_node), BlockFilterType::BASIC, 1 << 20, /*f_memory=*/false); + BOOST_REQUIRE(filter_index.Init()); + + ThreadPool thread_pool("blockfilter"); + thread_pool.Start(2); + filter_index.SetThreadPool(thread_pool); + filter_index.SetBlocksPerWorker(200); + + // Start sync + BOOST_CHECK(!filter_index.BlockUntilSyncedToCurrentChain()); + filter_index.Sync(); + const auto& summary{filter_index.GetSummary()}; + BOOST_CHECK(summary.synced); + BOOST_CHECK_EQUAL(summary.best_block_height, tip_height); + + // Check that filter index has all blocks that were in the chain before it started. + { + uint256 last_header; + LOCK(cs_main); + const CBlockIndex* block_index; + for (block_index = m_node.chainman->ActiveChain().Genesis(); + block_index != nullptr; + block_index = m_node.chainman->ActiveChain().Next(block_index)) { + CheckFilterLookups(filter_index, block_index, last_header, m_node.chainman->m_blockman); + } + } + + filter_index.Interrupt(); + filter_index.Stop(); + } +} + BOOST_FIXTURE_TEST_CASE(blockfilter_index_init_destroy, BasicTestingSetup) { BlockFilterIndex* filter_index; @@ -306,81 +351,4 @@ BOOST_FIXTURE_TEST_CASE(blockfilter_index_init_destroy, BasicTestingSetup) BOOST_CHECK(filter_index == nullptr); } -class IndexReorgCrash : public BaseIndex -{ -private: - std::unique_ptr m_db; - std::shared_future m_blocker; - int m_blocking_height; - -public: - explicit IndexReorgCrash(std::unique_ptr chain, std::shared_future blocker, - int blocking_height) : BaseIndex(std::move(chain), "test index"), m_blocker(blocker), - m_blocking_height(blocking_height) - { - const fs::path path = gArgs.GetDataDirNet() / "index"; - fs::create_directories(path); - m_db = std::make_unique(path / "db", /*n_cache_size=*/0, /*f_memory=*/true, /*f_wipe=*/false); - } - - bool AllowPrune() const override { return false; } - BaseIndex::DB& GetDB() const override { return *m_db; } - - bool CustomAppend(const interfaces::BlockInfo& block) override - { - // Simulate a delay so new blocks can get connected during the initial sync - if (block.height == m_blocking_height) m_blocker.wait(); - - // Move mock time forward so the best index gets updated only when we are not at the blocking height - if (block.height == m_blocking_height - 1 || block.height > m_blocking_height) { - SetMockTime(GetTime() + 31s); - } - - return true; - } -}; - -BOOST_FIXTURE_TEST_CASE(index_reorg_crash, BuildChainTestingSetup) -{ - // Enable mock time - SetMockTime(GetTime()); - - std::promise promise; - std::shared_future blocker(promise.get_future()); - int blocking_height = WITH_LOCK(cs_main, return m_node.chainman->ActiveChain().Tip()->nHeight); - - IndexReorgCrash index(interfaces::MakeChain(m_node), blocker, blocking_height); - BOOST_REQUIRE(index.Init()); - BOOST_REQUIRE(index.StartBackgroundSync()); - - auto func_wait_until = [&](int height, std::chrono::milliseconds timeout) { - auto deadline = std::chrono::steady_clock::now() + timeout; - while (index.GetSummary().best_block_height < height) { - if (std::chrono::steady_clock::now() > deadline) { - BOOST_FAIL(strprintf("Timeout waiting for index height %d (current: %d)", height, index.GetSummary().best_block_height)); - return; - } - std::this_thread::sleep_for(100ms); - } - }; - - // Wait until the index is one block before the fork point - func_wait_until(blocking_height - 1, /*timeout=*/5s); - - // Create a fork to trigger the reorg - std::vector> fork; - const CBlockIndex* prev_tip = WITH_LOCK(cs_main, return m_node.chainman->ActiveChain().Tip()->pprev); - BOOST_REQUIRE(BuildChain(prev_tip, GetScriptForDestination(PKHash(GenerateRandomKey().GetPubKey())), 3, fork)); - - for (const auto& block : fork) { - BOOST_REQUIRE(m_node.chainman->ProcessNewBlock(block, /*force_processing=*/true, /*min_pow_checked=*/true, nullptr)); - } - - // Unblock the index thread so it can process the reorg - promise.set_value(); - // Wait for the index to reach the new tip - func_wait_until(blocking_height + 2, 5s); - index.Stop(); -} - BOOST_AUTO_TEST_SUITE_END() diff --git a/src/test/threadpool_tests.cpp b/src/test/threadpool_tests.cpp new file mode 100644 index 00000000000..e245fc3bc34 --- /dev/null +++ b/src/test/threadpool_tests.cpp @@ -0,0 +1,268 @@ +// Copyright (c) 2024-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. + +#include +#include + +#include + +BOOST_AUTO_TEST_SUITE(threadpool_tests) + +constexpr auto TIMEOUT_SECS = std::chrono::seconds(120); + +template +void WaitFor(std::vector>& futures, const std::string& context) +{ + for (size_t i = 0; i < futures.size(); ++i) { + if (futures[i].wait_for(TIMEOUT_SECS) != std::future_status::ready) { + throw std::runtime_error("Timeout waiting for: " + context + ", task index " + util::ToString(i)); + } + } +} + +BOOST_AUTO_TEST_CASE(threadpool_basic) +{ + // Test Cases + // 1) Submit tasks and verify completion. + // 2) Maintain all threads busy except one. + // 3) Wait for work to finish. + // 4) Wait for result object. + // 5) The task throws an exception, catch must be done in the consumer side. + // 6) Busy workers, help them by processing tasks from outside. + + const int NUM_WORKERS_DEFAULT = 3; + const std::string POOL_NAME = "test"; + + // Test case 1, submit tasks and verify completion. + { + int num_tasks = 50; + + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + std::atomic counter = 0; + + // Store futures to ensure completion before checking counter. + std::vector> futures; + futures.reserve(num_tasks); + + for (int i = 1; i <= num_tasks; i++) { + futures.emplace_back(threadPool.Submit([&counter, i]() { + counter.fetch_add(i); + })); + } + + // Wait for all tasks to finish + WaitFor(futures, /*context=*/"test1 task"); + int expected_value = (num_tasks * (num_tasks + 1)) / 2; // Gauss sum. + BOOST_CHECK_EQUAL(counter.load(), expected_value); + BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0); + } + + // Test case 2, maintain all threads busy except one. + { + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + // Single blocking future for all threads + std::promise blocker; + std::shared_future blocker_future(blocker.get_future()); + + // Use per-thread ready promises to ensure all blocked threads have started + std::vector> ready_promises(NUM_WORKERS_DEFAULT - 1); + std::vector> ready_futures; + ready_futures.reserve(NUM_WORKERS_DEFAULT - 1); + for (auto& p : ready_promises) ready_futures.emplace_back(p.get_future()); + + // Submit blocking task to all threads except one + std::vector> blocking_tasks; + blocking_tasks.reserve(NUM_WORKERS_DEFAULT - 1); + for (int i = 0; i < NUM_WORKERS_DEFAULT - 1; i++) { + std::promise& ready = ready_promises[i]; + blocking_tasks.emplace_back(threadPool.Submit([&ready, blocker_future]() { + ready.set_value(); + blocker_future.wait(); + })); + } + // Wait until all blocked threads are actually blocked + WaitFor(ready_futures, /*context=*/"test2 blocking tasks enabled"); + + // Now execute tasks on the single available worker + // and check that all the tasks are executed. + int num_tasks = 15; + std::atomic counter = 0; + + // Store futures to wait on + std::vector> futures; + futures.reserve(num_tasks); + for (int i = 0; i < num_tasks; i++) { + futures.emplace_back(threadPool.Submit([&counter]() { + counter.fetch_add(1); + })); + } + + WaitFor(futures, /*context=*/"test2 tasks"); + BOOST_CHECK_EQUAL(counter.load(), num_tasks); + + blocker.set_value(); + WaitFor(blocking_tasks, /*context=*/"test2 blocking tasks disabled"); + threadPool.Stop(); + BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0); + } + + // Test case 3, wait for work to finish. + { + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + std::atomic flag = false; + std::future future = threadPool.Submit([&flag]() { + std::this_thread::sleep_for(std::chrono::milliseconds{200}); + flag.store(true); + }); + future.wait(); + BOOST_CHECK(flag.load()); + } + + // Test case 4, obtain result object. + { + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + std::future future_bool = threadPool.Submit([]() { + return true; + }); + BOOST_CHECK(future_bool.get()); + + std::future future_str = threadPool.Submit([]() { + return std::string("true"); + }); + std::string result = future_str.get(); + BOOST_CHECK_EQUAL(result, "true"); + } + + // Test case 5, throw exception and catch it on the consumer side. + { + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + + int ROUNDS = 5; + std::string err_msg{"something wrong happened"}; + std::vector> futures; + futures.reserve(ROUNDS); + for (int i = 0; i < ROUNDS; i++) { + futures.emplace_back(threadPool.Submit([err_msg, i]() { + throw std::runtime_error(err_msg + util::ToString(i)); + })); + } + + for (int i = 0; i < ROUNDS; i++) { + try { + futures.at(i).get(); + BOOST_FAIL("Expected exception not thrown"); + } catch (const std::runtime_error& e) { + BOOST_CHECK_EQUAL(e.what(), err_msg + util::ToString(i)); + } + } + } + + // Test case 6, all workers are busy, help them by processing tasks from outside. + { + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + + std::promise blocker; + std::shared_future blocker_future(blocker.get_future()); + + // Submit blocking task + for (int i = 0; i < NUM_WORKERS_DEFAULT; i++) { + threadPool.Submit([blocker_future]() { + blocker_future.wait(); + }); + } + + // Now submit tasks and check that none of them are executed. + int num_tasks = 20; + std::atomic counter = 0; + for (int i = 0; i < num_tasks; i++) { + threadPool.Submit([&counter]() { + counter.fetch_add(1); + }); + } + std::this_thread::sleep_for(std::chrono::milliseconds{100}); + BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 20); + + // Now process manually + for (int i = 0; i < num_tasks; i++) { + threadPool.ProcessTask(); + } + BOOST_CHECK_EQUAL(counter.load(), num_tasks); + blocker.set_value(); + threadPool.Stop(); + } + + // Test case 7, recursive submission of tasks. + { + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + + std::promise signal; + threadPool.Submit([&]() { + threadPool.Submit([&]() { + signal.set_value(); + }); + }); + + signal.get_future().wait(); + threadPool.Stop(); + } + + // Test case 8, submit a task when all threads are busy and then stop the pool. + { + ThreadPool threadPool(POOL_NAME); + threadPool.Start(NUM_WORKERS_DEFAULT); + + std::promise blocker; + std::shared_future blocker_future(blocker.get_future()); + + // Per-thread ready promises to ensure all workers are actually blocked + std::vector> ready_promises(NUM_WORKERS_DEFAULT); + std::vector> ready_futures; + ready_futures.reserve(NUM_WORKERS_DEFAULT); + for (auto& p : ready_promises) ready_futures.emplace_back(p.get_future()); + + // Fill all workers with blocking tasks + for (int i = 0; i < NUM_WORKERS_DEFAULT; i++) { + std::promise& ready = ready_promises[i]; + threadPool.Submit([blocker_future, &ready]() { + ready.set_value(); + blocker_future.wait(); + }); + } + + // Wait until all threads are actually blocked + WaitFor(ready_futures, /*context=*/"test8 blocking tasks enabled"); + + // Submit an extra task that should execute once a worker is free + std::future future = threadPool.Submit([]() { return true; }); + + // At this point, all workers are blocked, and the extra task is queued + BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 1); + + // Wait a short moment before unblocking the threads to mimic a concurrent shutdown + std::thread thread_unblocker([&blocker]() { + std::this_thread::sleep_for(std::chrono::milliseconds{300}); + blocker.set_value(); + }); + + // Stop the pool while the workers are still blocked + threadPool.Stop(); + + // Expect the submitted task to complete + BOOST_CHECK(future.get()); + thread_unblocker.join(); + + // Pool should be stopped and no workers remaining + BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0); + } + +} + +BOOST_AUTO_TEST_SUITE_END() diff --git a/src/test/txindex_tests.cpp b/src/test/txindex_tests.cpp index e06735e186b..6b27fc7c510 100644 --- a/src/test/txindex_tests.cpp +++ b/src/test/txindex_tests.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include @@ -73,4 +74,45 @@ BOOST_FIXTURE_TEST_CASE(txindex_initial_sync, TestChain100Setup) txindex.Stop(); } +BOOST_FIXTURE_TEST_CASE(txindex_parallel_initial_sync, TestChain100Setup) +{ + int tip_height = 100; // pre-mined blocks + const uint16_t MINE_BLOCKS = 650; + for (int round = 0; round < 2; round++) { // two rounds to test sync from genesis and from a higher block + // Generate blocks + mineBlocks(MINE_BLOCKS); + const CBlockIndex* tip = WITH_LOCK(::cs_main, return m_node.chainman->ActiveChain().Tip()); + BOOST_REQUIRE(tip->nHeight == MINE_BLOCKS + tip_height); + tip_height = tip->nHeight; + + // Init and start index + TxIndex txindex(interfaces::MakeChain(m_node), 1 << 20, /*f_memory=*/false); + BOOST_REQUIRE(txindex.Init()); + ThreadPool thread_pool("txindex"); + thread_pool.Start(2); + txindex.SetThreadPool(thread_pool); + txindex.SetBlocksPerWorker(200); + + BOOST_CHECK(!txindex.BlockUntilSyncedToCurrentChain()); + txindex.Sync(); + const auto& summary{txindex.GetSummary()}; + BOOST_CHECK(summary.synced); + BOOST_CHECK_EQUAL(summary.best_block_height, tip_height); + + // Check that txindex has all txs that were in the chain before it started. + CTransactionRef tx_disk; + uint256 block_hash; + for (const auto& txn : m_coinbase_txns) { + if (!txindex.FindTx(txn->GetHash(), block_hash, tx_disk)) { + BOOST_ERROR("FindTx failed"); + } else if (tx_disk->GetHash() != txn->GetHash()) { + BOOST_ERROR("Read incorrect tx"); + } + } + + txindex.Interrupt(); + txindex.Stop(); + } +} + BOOST_AUTO_TEST_SUITE_END() diff --git a/src/util/threadpool.h b/src/util/threadpool.h new file mode 100644 index 00000000000..628efd48ea3 --- /dev/null +++ b/src/util/threadpool.h @@ -0,0 +1,191 @@ +// Copyright (c) 2024-present The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or https://www.opensource.org/licenses/mit-license.php. + +#ifndef BITCOIN_UTIL_THREADPOOL_H +#define BITCOIN_UTIL_THREADPOOL_H + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/** + * @brief Fixed-size thread pool for running arbitrary tasks concurrently. + * + * The thread pool maintains a set of worker threads that consume and execute + * tasks submitted through Submit(). Once started, tasks can be queued and + * processed asynchronously until Stop() is called. + * + * ### Thread-safety and lifecycle + * - `Start()` and `Stop()` must be called from a controller (non-worker) thread. + * Calling `Stop()` from a worker thread will deadlock, as it waits for all + * workers to join, including the current one. + * + * - `Submit()` can be called from any thread, including workers. It safely + * enqueues new work for execution as long as the pool has active workers. + * + * - `Stop()` prevents further task submission and wakes all worker threads. + * Workers finish processing all remaining queued tasks before exiting, + * guaranteeing that no caller waits forever on a pending future. + */ +class ThreadPool { + +private: + std::string m_name; + Mutex m_mutex; + std::queue> m_work_queue GUARDED_BY(m_mutex); + std::condition_variable m_cv; + // Note: m_interrupt must be modified while holding the same mutex used by threads waiting on the condition variable. + // This ensures threads blocked on m_cv reliably observe the change and proceed correctly without missing signals. + // Ref: https://en.cppreference.com/w/cpp/thread/condition_variable + bool m_interrupt GUARDED_BY(m_mutex){false}; + std::vector m_workers GUARDED_BY(m_mutex); + + void WorkerThread() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + { + WAIT_LOCK(m_mutex, wait_lock); + for (;;) { + std::function task; + { + // Wait only if needed; avoid sleeping when a new task was submitted while we were processing another one. + if (!m_interrupt && m_work_queue.empty()) { + // Block until the pool is interrupted or a task is available. + m_cv.wait(wait_lock,[&]() EXCLUSIVE_LOCKS_REQUIRED(m_mutex) { return m_interrupt || !m_work_queue.empty(); }); + } + + // If stopped and no work left, exit worker + if (m_interrupt && m_work_queue.empty()) { + return; + } + + task = std::move(m_work_queue.front()); + m_work_queue.pop(); + } + + { + // Execute the task without the lock + REVERSE_LOCK(wait_lock, m_mutex); + task(); + } + } + } + +public: + explicit ThreadPool(const std::string& name) : m_name(name) {} + + ~ThreadPool() + { + Stop(); // In case it hasn't been stopped. + } + + /** + * @brief Start worker threads. + * + * Creates and launches `num_workers` threads that begin executing tasks + * from the queue. If the pool is already started, throws. + * + * Must be called from a controller (non-worker) thread. + */ + void Start(int num_workers) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + { + LOCK(m_mutex); + if (!m_workers.empty()) throw std::runtime_error("Thread pool already started"); + m_interrupt = false; // Reset + + // Create workers + for (int i = 0; i < num_workers; i++) { + m_workers.emplace_back(&util::TraceThread, m_name + "_pool_" + util::ToString(i), [this] { WorkerThread(); }); + } + } + + /** + * @brief Stop all worker threads and wait for them to exit. + * + * Sets the interrupt flag, wakes all waiting workers, and joins them. + * Any remaining tasks in the queue will be processed before returning. + * + * Must be called from a controller (non-worker) thread. + */ + void Stop() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + { + // Notify workers and join them. + std::vector threads_to_join; + { + LOCK(m_mutex); + m_interrupt = true; + threads_to_join.swap(m_workers); + } + m_cv.notify_all(); + for (auto& worker : threads_to_join) worker.join(); + // m_interrupt is left true until next Start() + } + + /** + * @brief Submit a new task for asynchronous execution. + * + * Enqueues a callable to be executed by one of the worker threads. + * Returns a `std::future` that can be used to retrieve the task’s result. + */ + template EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + auto Submit(T task) -> std::future + { + using TaskType = std::packaged_task; + auto ptr_task = std::make_shared(std::move(task)); + std::future future = ptr_task->get_future(); + { + LOCK(m_mutex); + if (m_workers.empty() || m_interrupt) { + throw std::runtime_error("No active workers; cannot accept new tasks"); + } + m_work_queue.emplace([ptr_task]() { + (*ptr_task)(); + }); + } + m_cv.notify_one(); + return future; + } + + /** + * @brief Execute a single queued task synchronously. + * Removes one task from the queue and executes it on the calling thread. + */ + void ProcessTask() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + { + std::function task; + { + LOCK(m_mutex); + if (m_work_queue.empty()) return; + + // Pop the task + task = std::move(m_work_queue.front()); + m_work_queue.pop(); + } + task(); + } + + size_t WorkQueueSize() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + { + return WITH_LOCK(m_mutex, return m_work_queue.size()); + } + + size_t WorkersCount() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex) + { + return WITH_LOCK(m_mutex, return m_workers.size()); + } +}; + +#endif // BITCOIN_UTIL_THREADPOOL_H