[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:
Jim Posen 2017-12-08 10:42:31 -08:00
parent 34d68bf3a3
commit 94b4f8bbb9
2 changed files with 100 additions and 0 deletions

View file

@ -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();
}
} }

View file

@ -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;