[index] TxIndex initial sync thread.
TxIndex starts up a background thread to get in sync with the block index before blocks are processed through the ValidationInterface.
This commit is contained in:
parent
34d68bf3a3
commit
94b4f8bbb9
2 changed files with 100 additions and 0 deletions
|
@ -2,6 +2,7 @@
|
||||||
// Distributed under the MIT software license, see the accompanying
|
// Distributed under the MIT software license, see the accompanying
|
||||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||||
|
|
||||||
|
#include <chainparams.h>
|
||||||
#include <index/txindex.h>
|
#include <index/txindex.h>
|
||||||
#include <init.h>
|
#include <init.h>
|
||||||
#include <tinyformat.h>
|
#include <tinyformat.h>
|
||||||
|
@ -10,6 +11,9 @@
|
||||||
#include <validation.h>
|
#include <validation.h>
|
||||||
#include <warnings.h>
|
#include <warnings.h>
|
||||||
|
|
||||||
|
constexpr int64_t SYNC_LOG_INTERVAL = 30; // seconds
|
||||||
|
constexpr int64_t SYNC_LOCATOR_WRITE_INTERVAL = 30; // seconds
|
||||||
|
|
||||||
template<typename... Args>
|
template<typename... Args>
|
||||||
static void FatalError(const char* fmt, const Args&... args)
|
static void FatalError(const char* fmt, const Args&... args)
|
||||||
{
|
{
|
||||||
|
@ -47,6 +51,75 @@ bool TxIndex::Init()
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static const CBlockIndex* NextSyncBlock(const CBlockIndex* pindex_prev)
|
||||||
|
{
|
||||||
|
AssertLockHeld(cs_main);
|
||||||
|
|
||||||
|
if (!pindex_prev) {
|
||||||
|
return chainActive.Genesis();
|
||||||
|
}
|
||||||
|
|
||||||
|
const CBlockIndex* pindex = chainActive.Next(pindex_prev);
|
||||||
|
if (pindex) {
|
||||||
|
return pindex;
|
||||||
|
}
|
||||||
|
|
||||||
|
return chainActive.Next(chainActive.FindFork(pindex_prev));
|
||||||
|
}
|
||||||
|
|
||||||
|
void TxIndex::ThreadSync()
|
||||||
|
{
|
||||||
|
const CBlockIndex* pindex = m_best_block_index.load();
|
||||||
|
if (!m_synced) {
|
||||||
|
auto& consensus_params = Params().GetConsensus();
|
||||||
|
|
||||||
|
int64_t last_log_time = 0;
|
||||||
|
int64_t last_locator_write_time = 0;
|
||||||
|
while (true) {
|
||||||
|
{
|
||||||
|
LOCK(cs_main);
|
||||||
|
const CBlockIndex* pindex_next = NextSyncBlock(pindex);
|
||||||
|
if (!pindex_next) {
|
||||||
|
WriteBestBlock(pindex);
|
||||||
|
m_best_block_index = pindex;
|
||||||
|
m_synced = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
pindex = pindex_next;
|
||||||
|
}
|
||||||
|
|
||||||
|
int64_t current_time = GetTime();
|
||||||
|
if (last_log_time + SYNC_LOG_INTERVAL < current_time) {
|
||||||
|
LogPrintf("Syncing txindex with block chain from height %d\n", pindex->nHeight);
|
||||||
|
last_log_time = current_time;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (last_locator_write_time + SYNC_LOCATOR_WRITE_INTERVAL < current_time) {
|
||||||
|
WriteBestBlock(pindex);
|
||||||
|
last_locator_write_time = current_time;
|
||||||
|
}
|
||||||
|
|
||||||
|
CBlock block;
|
||||||
|
if (!ReadBlockFromDisk(block, pindex, consensus_params)) {
|
||||||
|
FatalError("%s: Failed to read block %s from disk",
|
||||||
|
__func__, pindex->GetBlockHash().ToString());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (!WriteBlock(block, pindex)) {
|
||||||
|
FatalError("%s: Failed to write block %s to tx index database",
|
||||||
|
__func__, pindex->GetBlockHash().ToString());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (pindex) {
|
||||||
|
LogPrintf("txindex is enabled at height %d\n", pindex->nHeight);
|
||||||
|
} else {
|
||||||
|
LogPrintf("txindex is enabled\n");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
bool TxIndex::WriteBlock(const CBlock& block, const CBlockIndex* pindex)
|
bool TxIndex::WriteBlock(const CBlock& block, const CBlockIndex* pindex)
|
||||||
{
|
{
|
||||||
CDiskTxPos pos(pindex->GetBlockPos(), GetSizeOfCompactSize(block.vtx.size()));
|
CDiskTxPos pos(pindex->GetBlockPos(), GetSizeOfCompactSize(block.vtx.size()));
|
||||||
|
@ -59,6 +132,15 @@ bool TxIndex::WriteBlock(const CBlock& block, const CBlockIndex* pindex)
|
||||||
return m_db->WriteTxs(vPos);
|
return m_db->WriteTxs(vPos);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
bool TxIndex::WriteBestBlock(const CBlockIndex* block_index)
|
||||||
|
{
|
||||||
|
LOCK(cs_main);
|
||||||
|
if (!m_db->WriteBestBlock(chainActive.GetLocator(block_index))) {
|
||||||
|
return error("%s: Failed to write locator to disk", __func__);
|
||||||
|
}
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
void TxIndex::BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
|
void TxIndex::BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
|
||||||
const std::vector<CTransactionRef>& txn_conflicted)
|
const std::vector<CTransactionRef>& txn_conflicted)
|
||||||
{
|
{
|
||||||
|
@ -149,9 +231,16 @@ void TxIndex::Start()
|
||||||
FatalError("%s: txindex failed to initialize", __func__);
|
FatalError("%s: txindex failed to initialize", __func__);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
m_thread_sync = std::thread(&TraceThread<std::function<void()>>, "txindex",
|
||||||
|
std::bind(&TxIndex::ThreadSync, this));
|
||||||
}
|
}
|
||||||
|
|
||||||
void TxIndex::Stop()
|
void TxIndex::Stop()
|
||||||
{
|
{
|
||||||
UnregisterValidationInterface(this);
|
UnregisterValidationInterface(this);
|
||||||
|
|
||||||
|
if (m_thread_sync.joinable()) {
|
||||||
|
m_thread_sync.join();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -30,12 +30,23 @@ private:
|
||||||
/// The last block in the chain that the TxIndex is in sync with.
|
/// The last block in the chain that the TxIndex is in sync with.
|
||||||
std::atomic<const CBlockIndex*> m_best_block_index;
|
std::atomic<const CBlockIndex*> m_best_block_index;
|
||||||
|
|
||||||
|
std::thread m_thread_sync;
|
||||||
|
|
||||||
/// Initialize internal state from the database and block index.
|
/// Initialize internal state from the database and block index.
|
||||||
bool Init();
|
bool Init();
|
||||||
|
|
||||||
|
/// Sync the tx index with the block index starting from the current best
|
||||||
|
/// block. Intended to be run in its own thread, m_thread_sync. Once the
|
||||||
|
/// txindex gets in sync, the m_synced flag is set and the BlockConnected
|
||||||
|
/// ValidationInterface callback takes over and the sync thread exits.
|
||||||
|
void ThreadSync();
|
||||||
|
|
||||||
/// Write update index entries for a newly connected block.
|
/// Write update index entries for a newly connected block.
|
||||||
bool WriteBlock(const CBlock& block, const CBlockIndex* pindex);
|
bool WriteBlock(const CBlock& block, const CBlockIndex* pindex);
|
||||||
|
|
||||||
|
/// Write the current chain block locator to the DB.
|
||||||
|
bool WriteBestBlock(const CBlockIndex* block_index);
|
||||||
|
|
||||||
protected:
|
protected:
|
||||||
void BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
|
void BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
|
||||||
const std::vector<CTransactionRef>& txn_conflicted) override;
|
const std::vector<CTransactionRef>& txn_conflicted) override;
|
||||||
|
|
Loading…
Reference in a new issue