github
/
bitcoin
mirror of https://github.com/bitcoin/bitcoin
1
Fork 0
Code Releases Activity

Merge 57e37aecfd into a46065e36c

This commit is contained in:
Matias Furszyfer 2024-04-29 04:33:17 +02:00 committed by GitHub
parent a46065e36c 57e37aecfd
commit 8bdca526d4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 579 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/Makefile.am
View File

@ -324,6 +324,7 @@ BITCOIN_CORE_H = \
util/syserror.h \
util/task_runner.h \
util/thread.h \
util/threadpool.h \
util/threadinterrupt.h \
util/threadnames.h \
util/time.h \

1
src/Makefile.test.include
View File

@ -153,6 +153,7 @@ BITCOIN_TESTS =\
test/streams_tests.cpp \
test/sync_tests.cpp \
test/system_tests.cpp \
test/threadpool_tests.cpp \
test/timedata_tests.cpp \
test/torcontrol_tests.cpp \
test/transaction_tests.cpp \

110
src/index/base.cpp
View File

@ -13,9 +13,11 @@
#include <node/context.h>
#include <node/database_args.h>
#include <node/interface_ui.h>
#include <util/threadpool.h>
#include <tinyformat.h>
#include <util/thread.h>
#include <util/translation.h>
#include <undo.h>
#include <validation.h> // For g_chainman
#include <warnings.h>
@ -141,10 +143,53 @@ static const CBlockIndex* NextSyncBlock(const CBlockIndex* pindex_prev, CChain&
return chain.Next(chain.FindFork(pindex_prev));
}
std::any BaseIndex::ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data)
{
interfaces::BlockInfo block_info = kernel::MakeBlockInfo(pindex, block_data);
CBlock block;
if (!block_data) { // disk lookup if block data wasn't provided
if (!m_chainstate->m_blockman.ReadBlockFromDisk(block, *pindex)) {
FatalErrorf("%s: Failed to read block %s from disk",
__func__, pindex->GetBlockHash().ToString());
return false;
}
block_info.data = &block;
}
CBlockUndo block_undo;
if (RequiresBlockUndoData()) {
if (pindex->nHeight > 0 && !m_chainstate->m_blockman.UndoReadFromDisk(block_undo, *pindex)) {
FatalErrorf("%s: Failed to read undo block data %s from disk",
__func__, pindex->GetBlockHash().ToString());
return false;
}
block_info.undo_data = &block_undo;
}
const auto& any_obj = CustomProcessBlock(block_info);
if (!any_obj.has_value()) {
throw std::runtime_error(strprintf("%s: Failed to process block %s for index %s", __func__, pindex->GetBlockHash().GetHex(), GetName()));
}
return any_obj;
}
std::vector<std::any> BaseIndex::ProcessBlocks(const CBlockIndex* start, const CBlockIndex* end)
{
std::vector<std::any> objs;
do {
objs.emplace_back(ProcessBlock(end));
end = end->pprev;
} while (end && start->pprev != end);
return objs;
}
void BaseIndex::Sync()
{
const CBlockIndex* pindex = m_best_block_index.load();
if (!m_synced) {
bool parallel_sync_enabled = AllowParallelSync() && m_thread_pool;
std::chrono::steady_clock::time_point last_log_time{0s};
std::chrono::steady_clock::time_point last_locator_write_time{0s};
while (true) {
@ -183,24 +228,53 @@ void BaseIndex::Sync()
FatalErrorf("%s: Failed to rewind index %s to a previous chain tip", __func__, GetName());
return;
}
pindex = pindex_next;
// By default, parallel sync disabled
int work_chunk = 1;
int workers_count = 0;
std::vector<std::future<std::vector<std::any>>> futures;
const CBlockIndex* it_start = pindex_next;
CBlock block;
interfaces::BlockInfo block_info = kernel::MakeBlockInfo(pindex);
if (!m_chainstate->m_blockman.ReadBlockFromDisk(block, *pindex)) {
FatalErrorf("%s: Failed to read block %s from disk",
__func__, pindex->GetBlockHash().ToString());
return;
} else {
block_info.data = &block;
}
if (!CustomAppend(block_info)) {
FatalErrorf("%s: Failed to write block %s to index database",
__func__, pindex->GetBlockHash().ToString());
return;
if (parallel_sync_enabled) {
const int max_blocks_to_sync = m_tasks_per_worker * m_thread_pool->WorkersCount() + m_tasks_per_worker; // extra 'm_tasks_per_worker' due the active-wait.
const int tip_height = WITH_LOCK(cs_main, return m_chainstate->m_chain.Height());
const int remaining_blocks = tip_height - pindex_next->nHeight;
work_chunk = remaining_blocks > max_blocks_to_sync ? m_tasks_per_worker : remaining_blocks / (m_thread_pool->WorkersCount() + 1);
workers_count = m_thread_pool->WorkersCount();
if (work_chunk == 0) { // disable parallel sync if we are close to the tip
workers_count = 0;
work_chunk = 1;
}
for (int i = 0; i < workers_count; i++) {
const CBlockIndex* it_end = WITH_LOCK(::cs_main, return m_chainstate->m_chain[it_start->nHeight + work_chunk - 1]);
// Async process
futures.emplace_back(m_thread_pool->Submit(std::bind(&BaseIndex::ProcessBlocks, this, it_start, it_end)));
// Update iterator
it_start = WITH_LOCK(::cs_main, return NextSyncBlock(it_end, m_chainstate->m_chain));
}
}
// If we have only one block to process, run it directly.
// Otherwise, this is an active-wait, so we also process blocks in this thread until all workers finish.
const CBlockIndex* it_end = work_chunk == 1 ? it_start : WITH_LOCK(::cs_main, return m_chainstate->m_chain[it_start->nHeight + work_chunk - 1]);
std::packaged_task<std::vector<std::any>()> task(std::bind(&BaseIndex::ProcessBlocks, this, it_start, it_end));
futures.emplace_back(task.get_future());
task();
// Process blocks in-order
for (auto& future : futures) {
const auto& objs = future.get();
for (auto it = objs.rbegin(); it != objs.rend();) {
if (!CustomPostProcessBlocks(*it)) break; // error logged internally
it++;
}
}
// Keep moving
pindex = it_end;
// Commit changes
auto current_time{std::chrono::steady_clock::now()};
if (last_log_time + SYNC_LOG_INTERVAL < current_time) {
LogPrintf("Syncing %s with block chain from height %d\n",
@ -311,17 +385,13 @@ void BaseIndex::BlockConnected(ChainstateRole role, const std::shared_ptr<const
return;
}
}
interfaces::BlockInfo block_info = kernel::MakeBlockInfo(pindex, block.get());
if (CustomAppend(block_info)) {
if (CustomPostProcessBlocks(ProcessBlock(pindex))) {
// Setting the best block index is intentionally the last step of this
// function, so BlockUntilSyncedToCurrentChain callers waiting for the
// best block index to be updated can rely on the block being fully
// processed, and the index object being safe to delete.
SetBestBlockIndex(pindex);
} else {
FatalErrorf("%s: Failed to write block %s to index",
__func__, pindex->GetBlockHash().ToString());
return;
}
}

46
src/index/base.h
View File

@ -10,15 +10,25 @@
#include <util/threadinterrupt.h>
#include <validationinterface.h>
#include <any>
#include <string>
class CBlock;
class CBlockIndex;
class Chainstate;
class ChainstateManager;
class ThreadPool;
namespace interfaces {
class Chain;
} // namespace interfaces
namespace Consensus {
struct Params;
}
/** 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;
@ -78,6 +88,9 @@ private:
std::thread m_thread_sync;
CThreadInterrupt m_interrupt;
std::shared_ptr<ThreadPool> m_thread_pool;
uint16_t m_tasks_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:
@ -91,6 +104,9 @@ private:
/// Loop over disconnected blocks and call CustomRewind.
bool Rewind(const CBlockIndex* current_tip, const CBlockIndex* new_tip);
std::any ProcessBlock(const CBlockIndex* pindex, const CBlock* block_data = nullptr);
std::vector<std::any> ProcessBlocks(const CBlockIndex* start, const CBlockIndex* end);
virtual bool AllowPrune() const = 0;
template <typename... Args>
@ -119,11 +135,34 @@ protected:
/// be an ancestor of the current best block.
[[nodiscard]] virtual bool CustomRewind(const interfaces::BlockKey& current_tip, const interfaces::BlockKey& new_tip) { return true; }
/// Whether the child class requires to receive block undo data inside 'CustomAppend'.
virtual bool RequiresBlockUndoData() const { return false; }
virtual DB& GetDB() const = 0;
/// Update the internal best block index as well as the prune lock.
void SetBestBlockIndex(const CBlockIndex* block);
/// If 'AllowParallelSync()' retrieves true, 'ProcessBlock()' will run concurrently in batches.
/// The 'std::any' result will be passed to 'PostProcessBlocks()' 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;
}
/// 'PostProcessBlocks()' 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<interfaces::Chain> chain, std::string name);
/// Destructor interrupts sync thread if running and blocks until it exits.
@ -132,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(const std::shared_ptr<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
@ -158,6 +199,11 @@ public:
/// Stops the instance from staying in sync with blockchain updates.
void Stop();
void SetTasksPerWorker(int16_t count) { m_tasks_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;
};

30
src/index/blockfilterindex.cpp
View File

@ -250,19 +250,8 @@ std::optional<uint256> BlockFilterIndex::ReadFilterHeader(int height, const uint
bool BlockFilterIndex::CustomAppend(const interfaces::BlockInfo& block)
{
CBlockUndo block_undo;
if (block.height > 0) {
// pindex variable gives indexing code access to node internals. It
// will be removed in upcoming commit
const CBlockIndex* pindex = WITH_LOCK(cs_main, return m_chainstate->m_blockman.LookupBlockIndex(block.hash));
if (!m_chainstate->m_blockman.UndoReadFromDisk(block_undo, *pindex)) {
return false;
}
}
const CBlockUndo& block_undo = block.height > 0 ? *Assert(block.undo_data) : CBlockUndo();
BlockFilter filter(m_filter_type, *Assert(block.data), block_undo);
const uint256& header = filter.ComputeHeader(m_last_header);
bool res = Write(filter, block.height, header);
if (res) m_last_header = header; // update last header
@ -288,6 +277,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(BlockFilterType::BASIC, *block_info.data, *block_info.undo_data), block_info.height);
}
bool BlockFilterIndex::CustomPostProcessBlocks(const std::any& obj)
{
const auto& [filter, height] = std::any_cast<std::pair<BlockFilter, int>>(obj);
const uint256& header = filter.ComputeHeader(m_last_header);
if (!Write(filter, height, header)) {
LogError("Error writings 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 start_height, int stop_height)

7
src/index/blockfilterindex.h
View File

@ -60,8 +60,13 @@ protected:
bool CustomRewind(const interfaces::BlockKey& current_tip, const interfaces::BlockKey& new_tip) override;
bool RequiresBlockUndoData() const override { return true; }
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<interfaces::Chain> chain, BlockFilterType filter_type,
@ -69,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;

9
src/index/coinstatsindex.cpp
View File

@ -120,12 +120,7 @@ bool CoinStatsIndex::CustomAppend(const interfaces::BlockInfo& block)
// Ignore genesis block
if (block.height > 0) {
// pindex variable gives indexing code access to node internals. It
// will be removed in upcoming commit
const CBlockIndex* pindex = WITH_LOCK(cs_main, return m_chainstate->m_blockman.LookupBlockIndex(block.hash));
if (!m_chainstate->m_blockman.UndoReadFromDisk(block_undo, *pindex)) {
return false;
}
block_undo = *Assert(block.undo_data);
std::pair<uint256, DBVal> read_out;
if (!m_db->Read(DBHeightKey(block.height - 1), read_out)) {
@ -150,7 +145,7 @@ bool CoinStatsIndex::CustomAppend(const interfaces::BlockInfo& block)
const auto& tx{block.data->vtx.at(i)};
// Skip duplicate txid coinbase transactions (BIP30).
if (IsBIP30Unspendable(*pindex) && tx->IsCoinBase()) {
if (IsBIP30Unspendable(block.hash, block.height) && tx->IsCoinBase()) {
m_total_unspendable_amount += block_subsidy;
m_total_unspendables_bip30 += block_subsidy;
continue;

2
src/index/coinstatsindex.h
View File

@ -51,6 +51,8 @@ protected:
bool CustomRewind(const interfaces::BlockKey& current_tip, const interfaces::BlockKey& new_tip) override;
bool RequiresBlockUndoData() const override { return true; }
BaseIndex::DB& GetDB() const override { return *m_db; }
public:

8
src/index/txindex.h
View File

@ -27,8 +27,14 @@ private:
protected:
bool CustomAppend(const interfaces::BlockInfo& block) override;
bool RequiresBlockUndoData() const override { return false; }
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<interfaces::Chain> chain, size_t n_cache_size, bool f_memory = false, bool f_wipe = false);
@ -36,6 +42,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.

19
src/init.cpp
View File

@ -83,6 +83,7 @@
#include <util/syserror.h>
#include <util/thread.h>
#include <util/threadnames.h>
#include <util/threadpool.h>
#include <util/time.h>
#include <util/translation.h>
#include <validation.h>
@ -505,6 +506,7 @@ void SetupServerArgs(ArgsManager& argsman)
strprintf("Maintain an index of compact filters by block (default: %s, values: %s).", DEFAULT_BLOCKFILTERINDEX, ListBlockFilterTypes()) +
" If <type> is not supplied or if <type> = 1, indexes for all known types are enabled.",
ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS);
argsman.AddArg("-indexworkers=<n>", strprintf("Number of worker threads spawned for the indexes initial sync process (default: %d).", INDEX_WORKERS_COUNT), ArgsManager::ALLOW_ANY, OptionsCategory::OPTIONS);
argsman.AddArg("-addnode=<ip>", 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=<file>", 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);
@ -1990,6 +1992,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.
@ -2028,7 +2032,20 @@ bool StartIndexBackgroundSync(NodeContext& node)
}
}
std::shared_ptr<ThreadPool> thread_pool;
if (node.args->IsArgSet("-indexworkers")) {
int index_workers = node.args->GetIntArg("-indexworkers", INDEX_WORKERS_COUNT);
if (index_workers < 0 || index_workers > 100) return InitError(_("Invalid -indexworkers arg"));
thread_pool = std::make_shared<ThreadPool>();
thread_pool->Start(index_workers);
}
// Start threads
for (auto index : node.indexes) if (!index->StartBackgroundSync()) return false;
for (auto index : node.indexes) {
if (index->AllowParallelSync()) index->SetThreadPool(thread_pool);
if (!index->StartBackgroundSync()) return false;
}
return true;
}

1
src/sync.h
View File

@ -299,6 +299,7 @@ inline MutexType* MaybeCheckNotHeld(MutexType* m) LOCKS_EXCLUDED(m) LOCK_RETURNE
//! The above is detectable at compile-time with the -Wreturn-local-addr flag in
//! gcc and the -Wreturn-stack-address flag in clang, both enabled by default.
#define WITH_LOCK(cs, code) (MaybeCheckNotHeld(cs), [&]() -> decltype(auto) { LOCK(cs); code; }())
#define WITH_REVERSE_LOCK(cs, code) ([&]() -> decltype(auto) { REVERSE_LOCK(cs); code; }())
/** An implementation of a semaphore.
*

45
src/test/blockfilter_index_tests.cpp
View File

@ -14,6 +14,7 @@
#include <test/util/blockfilter.h>
#include <test/util/index.h>
#include <test/util/setup_common.h>
#include <util/threadpool.h>
#include <validation.h>
#include <boost/test/unit_test.hpp>
@ -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());
std::shared_ptr<ThreadPool> thread_pool = std::make_shared<ThreadPool>();
thread_pool->Start(2);
filter_index.SetThreadPool(thread_pool);
filter_index.SetTasksPerWorker(200);
// Start sync
BOOST_CHECK(!filter_index.BlockUntilSyncedToCurrentChain());
BOOST_REQUIRE(filter_index.StartBackgroundSync());
// Allow filter index to catch up with the block index.
IndexWaitSynced(filter_index, *Assert(m_node.shutdown));
// 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;

164
src/test/threadpool_tests.cpp Normal file
View File

@ -0,0 +1,164 @@
// 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 <util/threadpool.h>
#include <boost/test/unit_test.hpp>
BOOST_AUTO_TEST_SUITE(threadpool_tests)
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;
// Test case 1, submit tasks and verify completion.
{
int num_tasks = 50;
ThreadPool threadPool;
threadPool.Start(NUM_WORKERS_DEFAULT);
std::atomic<int> counter = 0;
for (int i = 1; i < num_tasks + 1; i++) {
threadPool.Submit([&counter, i]() {
counter += i;
});
}
while (threadPool.WorkQueueSize() != 0) {
std::this_thread::sleep_for(std::chrono::milliseconds{2});
}
int expected_value = (num_tasks * (num_tasks + 1)) / 2; // Gauss sum.
BOOST_CHECK_EQUAL(counter, expected_value);
}
// Test case 2, maintain all threads busy except one.
{
ThreadPool threadPool;
threadPool.Start(NUM_WORKERS_DEFAULT);
std::atomic<bool> stop = false;
for (int i=0; i < NUM_WORKERS_DEFAULT - 1; i++) {
threadPool.Submit([&stop]() {
while (!stop) {
std::this_thread::sleep_for(std::chrono::milliseconds{10});
}
});
}
// Now execute tasks on the single available worker
// and check that all the tasks are executed.
int num_tasks = 15;
std::atomic<int> counter = 0;
for (int i=0; i < num_tasks; i++) {
threadPool.Submit([&counter]() {
counter++;
});
}
while (threadPool.WorkQueueSize() != 0) {
std::this_thread::sleep_for(std::chrono::milliseconds{2});
}
BOOST_CHECK_EQUAL(counter, num_tasks);
stop = true;
threadPool.Stop();
}
// Test case 3, wait for work to finish.
{
ThreadPool threadPool;
threadPool.Start(NUM_WORKERS_DEFAULT);
std::atomic<bool> flag = false;
std::future<void> future = threadPool.Submit([&flag]() {
std::this_thread::sleep_for(std::chrono::milliseconds{200});
flag = true;
});
future.get();
BOOST_CHECK(flag.load());
}
// Test case 4, wait for result object.
{
ThreadPool threadPool;
threadPool.Start(NUM_WORKERS_DEFAULT);
std::future<bool> future_bool = threadPool.Submit([]() {
return true;
});
BOOST_CHECK(future_bool.get());
std::future<std::string> future_str = threadPool.Submit([]() {
return std::string("true");
});
BOOST_CHECK_EQUAL(future_str.get(), "true");
}
// Test case 5
// Throw exception and catch it on the consumer side.
{
ThreadPool threadPool;
threadPool.Start(NUM_WORKERS_DEFAULT);
int ROUNDS = 5;
std::string err_msg{"something wrong happened"};
std::vector<std::future<void>> 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 + ToString(i));
}));
}
for (int i = 0; i < ROUNDS; i++) {
try {
futures.at(i).get();
BOOST_ASSERT_MSG(false, "Error: did not throw an exception");
} catch (const std::runtime_error& e) {
BOOST_CHECK_EQUAL(e.what(), err_msg + ToString(i));
}
}
}
// Test case 6
// All workers are busy, help them by processing tasks from outside
{
ThreadPool threadPool;
threadPool.Start(NUM_WORKERS_DEFAULT);
std::atomic<bool> stop = false;
for (int i=0; i < NUM_WORKERS_DEFAULT; i++) {
threadPool.Submit([&stop]() {
while (!stop) {
std::this_thread::sleep_for(std::chrono::milliseconds{10});
}
});
}
// Now submit tasks and check that none of them are executed.
int num_tasks = 20;
std::atomic<int> counter = 0;
for (int i=0; i < num_tasks; i++) {
threadPool.Submit([&counter]() {
counter++;
});
}
std::this_thread::sleep_for(std::chrono::milliseconds{100});
BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 20);
// Now process them manually
for (int i=0; i<num_tasks; i++) {
threadPool.ProcessTask();
}
BOOST_CHECK_EQUAL(counter, num_tasks);
stop = true;
threadPool.Stop();
}
}
BOOST_AUTO_TEST_SUITE_END()

42
src/test/txindex_tests.cpp
View File

@ -8,6 +8,7 @@
#include <interfaces/chain.h>
#include <test/util/index.h>
#include <test/util/setup_common.h>
#include <util/threadpool.h>
#include <validation.h>
#include <boost/test/unit_test.hpp>
@ -77,4 +78,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());
std::shared_ptr<ThreadPool> thread_pool = std::make_shared<ThreadPool>();
thread_pool->Start(2);
txindex.SetThreadPool(thread_pool);
txindex.SetTasksPerWorker(200);
BOOST_CHECK(!txindex.BlockUntilSyncedToCurrentChain());
BOOST_REQUIRE(txindex.StartBackgroundSync());
// Allow tx index to catch up with the block index.
IndexWaitSynced(txindex, *Assert(m_node.shutdown));
// 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()

130
src/util/threadpool.h Normal file
View File

@ -0,0 +1,130 @@
// 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 <sync.h>
#include <util/string.h>
#include <util/thread.h>
#include <util/threadinterrupt.h>
#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <cstddef>
#include <functional>
#include <future>
#include <memory>
#include <stdexcept>
#include <utility>
#include <queue>
#include <thread>
#include <vector>
class ThreadPool {
private:
Mutex cs_work_queue;
std::queue<std::function<void()>> m_work_queue GUARDED_BY(cs_work_queue);
std::condition_variable m_condition;
CThreadInterrupt m_interrupt;
std::vector<std::thread> m_workers;
void WorkerThread() EXCLUSIVE_LOCKS_REQUIRED(!cs_work_queue)
{
WAIT_LOCK(cs_work_queue, wait_lock);
while (!m_interrupt) {
std::function<void()> task;
{
// Wait for the task or until the stop flag is set
m_condition.wait(wait_lock,[&]() EXCLUSIVE_LOCKS_REQUIRED(cs_work_queue) { return m_interrupt || !m_work_queue.empty(); });
// If stopped, exit worker.
if (m_interrupt && m_work_queue.empty()) {
return;
}
// Pop the task
task = std::move(m_work_queue.front());
m_work_queue.pop();
}
// Execute the task without the lock
WITH_REVERSE_LOCK(wait_lock, task());
}
}
public:
ThreadPool() {}
~ThreadPool()
{
Stop(); // In case it hasn't been stopped.
}
void Start(int num_workers)
{
if (!m_workers.empty()) throw std::runtime_error("Thread pool already started");
// Create the workers
for (int i = 0; i < num_workers; i++) {
m_workers.emplace_back(&util::TraceThread, "threadpool_worker_" + ToString(i), [this] { WorkerThread(); });
}
}
void Stop()
{
// Notify workers and join them.
m_interrupt();
m_condition.notify_all();
for (auto& worker : m_workers) {
worker.join();
}
m_workers.clear();
m_interrupt.reset();
}
template<class T> EXCLUSIVE_LOCKS_REQUIRED(!cs_work_queue)
auto Submit(T task) -> std::future<decltype(task())>
{
auto ptr_task = std::make_shared<std::packaged_task<decltype(task()) ()>>(std::move(task));
std::future<decltype(task())> future = ptr_task->get_future();
{
LOCK(cs_work_queue);
m_work_queue.emplace([=]() {
(*ptr_task)();
});
}
m_condition.notify_one();
return future;
}
// Synchronous processing
void ProcessTask() EXCLUSIVE_LOCKS_REQUIRED(!cs_work_queue)
{
std::function<void()> task;
{
LOCK(cs_work_queue);
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(!cs_work_queue)
{
return WITH_LOCK(cs_work_queue, return m_work_queue.size());
}
size_t WorkersCount() const
{
return m_workers.size();
}
};
#endif // BITCOIN_UTIL_THREADPOOL_H

6
src/validation.cpp
View File

@ -6086,10 +6086,10 @@ bool IsBIP30Repeat(const CBlockIndex& block_index)
(block_index.nHeight==91880 && block_index.GetBlockHash() == uint256S("0x00000000000743f190a18c5577a3c2d2a1f610ae9601ac046a38084ccb7cd721"));
}
bool IsBIP30Unspendable(const CBlockIndex& block_index)
bool IsBIP30Unspendable(const uint256& block_hash, int block_height)
{
return (block_index.nHeight==91722 && block_index.GetBlockHash() == uint256S("0x00000000000271a2dc26e7667f8419f2e15416dc6955e5a6c6cdf3f2574dd08e")) ||
(block_index.nHeight==91812 && block_index.GetBlockHash() == uint256S("0x00000000000af0aed4792b1acee3d966af36cf5def14935db8de83d6f9306f2f"));
return (block_height==91722 && block_hash == uint256S("0x00000000000271a2dc26e7667f8419f2e15416dc6955e5a6c6cdf3f2574dd08e")) ||
(block_height==91812 && block_hash == uint256S("0x00000000000af0aed4792b1acee3d966af36cf5def14935db8de83d6f9306f2f"));
}
static fs::path GetSnapshotCoinsDBPath(Chainstate& cs) EXCLUSIVE_LOCKS_REQUIRED(::cs_main)

2
src/validation.h
View File

@ -1300,6 +1300,6 @@ bool DeploymentEnabled(const ChainstateManager& chainman, DEP dep)
bool IsBIP30Repeat(const CBlockIndex& block_index);
/** Identifies blocks which coinbase output was subsequently overwritten in the UTXO set (see BIP30) */
bool IsBIP30Unspendable(const CBlockIndex& block_index);
bool IsBIP30Unspendable(const uint256& block_hash, int block_height);
#endif // BITCOIN_VALIDATION_H