Merge #13243: Make reusable base class for auxiliary indices
ec3073a274index: Move index DBs into index/ directory. (Jim Posen)89eddcd365index: Remove TxIndexDB from public interface of TxIndex. (Jim Posen)2318affd27MOVEONLY: Move BaseIndex to its own file. (Jim Posen)f376a49241index: Generalize logged statements in BaseIndex. (Jim Posen)61a1226d87index: Extract logic from TxIndex into reusable base class. (Jim Posen)e5af5fc6fbdb: Make reusable base class for index databases. (Jim Posen)9b0ec1a7f9db: Remove obsolete methods from CBlockTreeDB. (Jim Posen) Pull request description: This refactors most of the logic in TxIndex into a reusable base class for other indices. There are two commits moving code between files, which may be be more easily reviewed using `git diff --color-moved` (https://blog.github.com/2018-04-05-git-217-released/). The motivation for this is to support BIP 157 by indexing block filters. <!-- Reviewable:start --> --- This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/bitcoin/bitcoin/13243) <!-- Reviewable:end --> Tree-SHA512: 0857f04df2aa920178dab2eb8e57984d8eb4d5010deca9971190358479e05b6672ccca2a08af0a7ac9fe02afb947be84cf35a3693204d0667263c6add2959cbf
This commit is contained in:
Wladimir J. van der Laan
commit
ea263e1eb0
|
|
@ -107,6 +107,7 @@ BITCOIN_CORE_H = \
|
|||
fs.h \
|
||||
httprpc.h \
|
||||
httpserver.h \
|
||||
index/base.h \
|
||||
index/txindex.h \
|
||||
indirectmap.h \
|
||||
init.h \
|
||||
|
|
@ -208,6 +209,7 @@ libbitcoin_server_a_SOURCES = \
|
|||
consensus/tx_verify.cpp \
|
||||
httprpc.cpp \
|
||||
httpserver.cpp \
|
||||
index/base.cpp \
|
||||
index/txindex.cpp \
|
||||
init.cpp \
|
||||
dbwrapper.cpp \
|
||||
|
|
|
|||
278
src/index/base.cpp
Normal file
278
src/index/base.cpp
Normal file
|
|
@ -0,0 +1,278 @@
|
|||
// Copyright (c) 2017-2018 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 <chainparams.h>
|
||||
#include <index/base.h>
|
||||
#include <init.h>
|
||||
#include <tinyformat.h>
|
||||
#include <ui_interface.h>
|
||||
#include <util.h>
|
||||
#include <validation.h>
|
||||
#include <warnings.h>
|
||||
|
||||
constexpr char DB_BEST_BLOCK = 'B';
|
||||
|
||||
constexpr int64_t SYNC_LOG_INTERVAL = 30; // seconds
|
||||
constexpr int64_t SYNC_LOCATOR_WRITE_INTERVAL = 30; // seconds
|
||||
|
||||
template<typename... Args>
|
||||
static void FatalError(const char* fmt, const Args&... args)
|
||||
{
|
||||
std::string strMessage = tfm::format(fmt, args...);
|
||||
SetMiscWarning(strMessage);
|
||||
LogPrintf("*** %s\n", strMessage);
|
||||
uiInterface.ThreadSafeMessageBox(
|
||||
"Error: A fatal internal error occurred, see debug.log for details",
|
||||
"", CClientUIInterface::MSG_ERROR);
|
||||
StartShutdown();
|
||||
}
|
||||
|
||||
BaseIndex::DB::DB(const fs::path& path, size_t n_cache_size, bool f_memory, bool f_wipe, bool f_obfuscate) :
|
||||
CDBWrapper(path, n_cache_size, f_memory, f_wipe, f_obfuscate)
|
||||
{}
|
||||
|
||||
bool BaseIndex::DB::ReadBestBlock(CBlockLocator& locator) const
|
||||
{
|
||||
bool success = Read(DB_BEST_BLOCK, locator);
|
||||
if (!success) {
|
||||
locator.SetNull();
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
bool BaseIndex::DB::WriteBestBlock(const CBlockLocator& locator)
|
||||
{
|
||||
return Write(DB_BEST_BLOCK, locator);
|
||||
}
|
||||
|
||||
BaseIndex::~BaseIndex()
|
||||
{
|
||||
Interrupt();
|
||||
Stop();
|
||||
}
|
||||
|
||||
bool BaseIndex::Init()
|
||||
{
|
||||
CBlockLocator locator;
|
||||
if (!GetDB().ReadBestBlock(locator)) {
|
||||
locator.SetNull();
|
||||
}
|
||||
|
||||
LOCK(cs_main);
|
||||
m_best_block_index = FindForkInGlobalIndex(chainActive, locator);
|
||||
m_synced = m_best_block_index.load() == chainActive.Tip();
|
||||
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 BaseIndex::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) {
|
||||
if (m_interrupt) {
|
||||
WriteBestBlock(pindex);
|
||||
return;
|
||||
}
|
||||
|
||||
{
|
||||
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 %s with block chain from height %d\n",
|
||||
GetName(), 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 index database",
|
||||
__func__, pindex->GetBlockHash().ToString());
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (pindex) {
|
||||
LogPrintf("%s is enabled at height %d\n", GetName(), pindex->nHeight);
|
||||
} else {
|
||||
LogPrintf("%s is enabled\n", GetName());
|
||||
}
|
||||
}
|
||||
|
||||
bool BaseIndex::WriteBestBlock(const CBlockIndex* block_index)
|
||||
{
|
||||
LOCK(cs_main);
|
||||
if (!GetDB().WriteBestBlock(chainActive.GetLocator(block_index))) {
|
||||
return error("%s: Failed to write locator to disk", __func__);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
void BaseIndex::BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
|
||||
const std::vector<CTransactionRef>& txn_conflicted)
|
||||
{
|
||||
if (!m_synced) {
|
||||
return;
|
||||
}
|
||||
|
||||
const CBlockIndex* best_block_index = m_best_block_index.load();
|
||||
if (!best_block_index) {
|
||||
if (pindex->nHeight != 0) {
|
||||
FatalError("%s: First block connected is not the genesis block (height=%d)",
|
||||
__func__, pindex->nHeight);
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
// Ensure block connects to an ancestor of the current best block. This should be the case
|
||||
// most of the time, but may not be immediately after the sync thread catches up and sets
|
||||
// m_synced. Consider the case where there is a reorg and the blocks on the stale branch are
|
||||
// in the ValidationInterface queue backlog even after the sync thread has caught up to the
|
||||
// new chain tip. In this unlikely event, log a warning and let the queue clear.
|
||||
if (best_block_index->GetAncestor(pindex->nHeight - 1) != pindex->pprev) {
|
||||
LogPrintf("%s: WARNING: Block %s does not connect to an ancestor of " /* Continued */
|
||||
"known best chain (tip=%s); not updating index\n",
|
||||
__func__, pindex->GetBlockHash().ToString(),
|
||||
best_block_index->GetBlockHash().ToString());
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (WriteBlock(*block, pindex)) {
|
||||
m_best_block_index = pindex;
|
||||
} else {
|
||||
FatalError("%s: Failed to write block %s to index",
|
||||
__func__, pindex->GetBlockHash().ToString());
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
void BaseIndex::ChainStateFlushed(const CBlockLocator& locator)
|
||||
{
|
||||
if (!m_synced) {
|
||||
return;
|
||||
}
|
||||
|
||||
const uint256& locator_tip_hash = locator.vHave.front();
|
||||
const CBlockIndex* locator_tip_index;
|
||||
{
|
||||
LOCK(cs_main);
|
||||
locator_tip_index = LookupBlockIndex(locator_tip_hash);
|
||||
}
|
||||
|
||||
if (!locator_tip_index) {
|
||||
FatalError("%s: First block (hash=%s) in locator was not found",
|
||||
__func__, locator_tip_hash.ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
// This checks that ChainStateFlushed callbacks are received after BlockConnected. The check may fail
|
||||
// immediately after the sync thread catches up and sets m_synced. Consider the case where
|
||||
// there is a reorg and the blocks on the stale branch are in the ValidationInterface queue
|
||||
// backlog even after the sync thread has caught up to the new chain tip. In this unlikely
|
||||
// event, log a warning and let the queue clear.
|
||||
const CBlockIndex* best_block_index = m_best_block_index.load();
|
||||
if (best_block_index->GetAncestor(locator_tip_index->nHeight) != locator_tip_index) {
|
||||
LogPrintf("%s: WARNING: Locator contains block (hash=%s) not on known best " /* Continued */
|
||||
"chain (tip=%s); not writing index locator\n",
|
||||
__func__, locator_tip_hash.ToString(),
|
||||
best_block_index->GetBlockHash().ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
if (!GetDB().WriteBestBlock(locator)) {
|
||||
error("%s: Failed to write locator to disk", __func__);
|
||||
}
|
||||
}
|
||||
|
||||
bool BaseIndex::BlockUntilSyncedToCurrentChain()
|
||||
{
|
||||
AssertLockNotHeld(cs_main);
|
||||
|
||||
if (!m_synced) {
|
||||
return false;
|
||||
}
|
||||
|
||||
{
|
||||
// Skip the queue-draining stuff if we know we're caught up with
|
||||
// chainActive.Tip().
|
||||
LOCK(cs_main);
|
||||
const CBlockIndex* chain_tip = chainActive.Tip();
|
||||
const CBlockIndex* best_block_index = m_best_block_index.load();
|
||||
if (best_block_index->GetAncestor(chain_tip->nHeight) == chain_tip) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
LogPrintf("%s: %s is catching up on block notifications\n", __func__, GetName());
|
||||
SyncWithValidationInterfaceQueue();
|
||||
return true;
|
||||
}
|
||||
|
||||
void BaseIndex::Interrupt()
|
||||
{
|
||||
m_interrupt();
|
||||
}
|
||||
|
||||
void BaseIndex::Start()
|
||||
{
|
||||
// Need to register this ValidationInterface before running Init(), so that
|
||||
// callbacks are not missed if Init sets m_synced to true.
|
||||
RegisterValidationInterface(this);
|
||||
if (!Init()) {
|
||||
FatalError("%s: %s failed to initialize", __func__, GetName());
|
||||
return;
|
||||
}
|
||||
|
||||
m_thread_sync = std::thread(&TraceThread<std::function<void()>>, GetName(),
|
||||
std::bind(&BaseIndex::ThreadSync, this));
|
||||
}
|
||||
|
||||
void BaseIndex::Stop()
|
||||
{
|
||||
UnregisterValidationInterface(this);
|
||||
|
||||
if (m_thread_sync.joinable()) {
|
||||
m_thread_sync.join();
|
||||
}
|
||||
}
|
||||
98
src/index/base.h
Normal file
98
src/index/base.h
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
// Copyright (c) 2017-2018 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#ifndef BITCOIN_INDEX_BASE_H
|
||||
#define BITCOIN_INDEX_BASE_H
|
||||
|
||||
#include <dbwrapper.h>
|
||||
#include <primitives/block.h>
|
||||
#include <primitives/transaction.h>
|
||||
#include <threadinterrupt.h>
|
||||
#include <uint256.h>
|
||||
#include <validationinterface.h>
|
||||
|
||||
class CBlockIndex;
|
||||
|
||||
/**
|
||||
* Base class for indices of blockchain data. This implements
|
||||
* CValidationInterface and ensures blocks are indexed sequentially according
|
||||
* to their position in the active chain.
|
||||
*/
|
||||
class BaseIndex : public CValidationInterface
|
||||
{
|
||||
protected:
|
||||
class DB : public CDBWrapper
|
||||
{
|
||||
public:
|
||||
DB(const fs::path& path, size_t n_cache_size,
|
||||
bool f_memory = false, bool f_wipe = false, bool f_obfuscate = false);
|
||||
|
||||
/// Read block locator of the chain that the txindex is in sync with.
|
||||
bool ReadBestBlock(CBlockLocator& locator) const;
|
||||
|
||||
/// Write block locator of the chain that the txindex is in sync with.
|
||||
bool WriteBestBlock(const CBlockLocator& locator);
|
||||
};
|
||||
|
||||
private:
|
||||
/// Whether the index is in sync with the main chain. The flag is flipped
|
||||
/// from false to true once, after which point this starts processing
|
||||
/// ValidationInterface notifications to stay in sync.
|
||||
std::atomic<bool> m_synced{false};
|
||||
|
||||
/// The last block in the chain that the index is in sync with.
|
||||
std::atomic<const CBlockIndex*> m_best_block_index{nullptr};
|
||||
|
||||
std::thread m_thread_sync;
|
||||
CThreadInterrupt m_interrupt;
|
||||
|
||||
/// Sync the index with the block index starting from the current best block.
|
||||
/// Intended to be run in its own thread, m_thread_sync, and can be
|
||||
/// interrupted with m_interrupt. Once the index gets in sync, the m_synced
|
||||
/// flag is set and the BlockConnected ValidationInterface callback takes
|
||||
/// over and the sync thread exits.
|
||||
void ThreadSync();
|
||||
|
||||
/// Write the current chain block locator to the DB.
|
||||
bool WriteBestBlock(const CBlockIndex* block_index);
|
||||
|
||||
protected:
|
||||
void BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
|
||||
const std::vector<CTransactionRef>& txn_conflicted) override;
|
||||
|
||||
void ChainStateFlushed(const CBlockLocator& locator) override;
|
||||
|
||||
/// Initialize internal state from the database and block index.
|
||||
virtual bool Init();
|
||||
|
||||
/// Write update index entries for a newly connected block.
|
||||
virtual bool WriteBlock(const CBlock& block, const CBlockIndex* pindex) { return true; }
|
||||
|
||||
virtual DB& GetDB() const = 0;
|
||||
|
||||
/// Get the name of the index for display in logs.
|
||||
virtual const char* GetName() const = 0;
|
||||
|
||||
public:
|
||||
/// Destructor interrupts sync thread if running and blocks until it exits.
|
||||
virtual ~BaseIndex();
|
||||
|
||||
/// 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
|
||||
/// queue. If the index is catching up from far behind, this method does
|
||||
/// not block and immediately returns false.
|
||||
bool BlockUntilSyncedToCurrentChain();
|
||||
|
||||
void Interrupt();
|
||||
|
||||
/// Start initializes the sync state and registers the instance as a
|
||||
/// ValidationInterface so that it stays in sync with blockchain updates.
|
||||
void Start();
|
||||
|
||||
/// Stops the instance from staying in sync with blockchain updates.
|
||||
void Stop();
|
||||
};
|
||||
|
||||
#endif // BITCOIN_INDEX_BASE_H
|
||||
|
|
@ -2,42 +2,233 @@
|
|||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#include <chainparams.h>
|
||||
#include <index/txindex.h>
|
||||
#include <init.h>
|
||||
#include <tinyformat.h>
|
||||
#include <ui_interface.h>
|
||||
#include <util.h>
|
||||
#include <validation.h>
|
||||
#include <warnings.h>
|
||||
|
||||
constexpr int64_t SYNC_LOG_INTERVAL = 30; // seconds
|
||||
constexpr int64_t SYNC_LOCATOR_WRITE_INTERVAL = 30; // seconds
|
||||
#include <boost/thread.hpp>
|
||||
|
||||
constexpr char DB_BEST_BLOCK = 'B';
|
||||
constexpr char DB_TXINDEX = 't';
|
||||
constexpr char DB_TXINDEX_BLOCK = 'T';
|
||||
|
||||
std::unique_ptr<TxIndex> g_txindex;
|
||||
|
||||
template<typename... Args>
|
||||
static void FatalError(const char* fmt, const Args&... args)
|
||||
struct CDiskTxPos : public CDiskBlockPos
|
||||
{
|
||||
std::string strMessage = tfm::format(fmt, args...);
|
||||
SetMiscWarning(strMessage);
|
||||
LogPrintf("*** %s\n", strMessage);
|
||||
uiInterface.ThreadSafeMessageBox(
|
||||
"Error: A fatal internal error occurred, see debug.log for details",
|
||||
"", CClientUIInterface::MSG_ERROR);
|
||||
StartShutdown();
|
||||
unsigned int nTxOffset; // after header
|
||||
|
||||
ADD_SERIALIZE_METHODS;
|
||||
|
||||
template <typename Stream, typename Operation>
|
||||
inline void SerializationOp(Stream& s, Operation ser_action) {
|
||||
READWRITEAS(CDiskBlockPos, *this);
|
||||
READWRITE(VARINT(nTxOffset));
|
||||
}
|
||||
|
||||
TxIndex::TxIndex(std::unique_ptr<TxIndexDB> db) :
|
||||
m_db(std::move(db)), m_synced(false), m_best_block_index(nullptr)
|
||||
CDiskTxPos(const CDiskBlockPos &blockIn, unsigned int nTxOffsetIn) : CDiskBlockPos(blockIn.nFile, blockIn.nPos), nTxOffset(nTxOffsetIn) {
|
||||
}
|
||||
|
||||
CDiskTxPos() {
|
||||
SetNull();
|
||||
}
|
||||
|
||||
void SetNull() {
|
||||
CDiskBlockPos::SetNull();
|
||||
nTxOffset = 0;
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Access to the txindex database (indexes/txindex/)
|
||||
*
|
||||
* The database stores a block locator of the chain the database is synced to
|
||||
* so that the TxIndex can efficiently determine the point it last stopped at.
|
||||
* A locator is used instead of a simple hash of the chain tip because blocks
|
||||
* and block index entries may not be flushed to disk until after this database
|
||||
* is updated.
|
||||
*/
|
||||
class TxIndex::DB : public BaseIndex::DB
|
||||
{
|
||||
public:
|
||||
explicit DB(size_t n_cache_size, bool f_memory = false, bool f_wipe = false);
|
||||
|
||||
/// Read the disk location of the transaction data with the given hash. Returns false if the
|
||||
/// transaction hash is not indexed.
|
||||
bool ReadTxPos(const uint256& txid, CDiskTxPos& pos) const;
|
||||
|
||||
/// Write a batch of transaction positions to the DB.
|
||||
bool WriteTxs(const std::vector<std::pair<uint256, CDiskTxPos>>& v_pos);
|
||||
|
||||
/// Migrate txindex data from the block tree DB, where it may be for older nodes that have not
|
||||
/// been upgraded yet to the new database.
|
||||
bool MigrateData(CBlockTreeDB& block_tree_db, const CBlockLocator& best_locator);
|
||||
};
|
||||
|
||||
TxIndex::DB::DB(size_t n_cache_size, bool f_memory, bool f_wipe) :
|
||||
BaseIndex::DB(GetDataDir() / "indexes" / "txindex", n_cache_size, f_memory, f_wipe)
|
||||
{}
|
||||
|
||||
TxIndex::~TxIndex()
|
||||
bool TxIndex::DB::ReadTxPos(const uint256 &txid, CDiskTxPos& pos) const
|
||||
{
|
||||
Interrupt();
|
||||
Stop();
|
||||
return Read(std::make_pair(DB_TXINDEX, txid), pos);
|
||||
}
|
||||
|
||||
bool TxIndex::DB::WriteTxs(const std::vector<std::pair<uint256, CDiskTxPos>>& v_pos)
|
||||
{
|
||||
CDBBatch batch(*this);
|
||||
for (const auto& tuple : v_pos) {
|
||||
batch.Write(std::make_pair(DB_TXINDEX, tuple.first), tuple.second);
|
||||
}
|
||||
return WriteBatch(batch);
|
||||
}
|
||||
|
||||
/*
|
||||
* Safely persist a transfer of data from the old txindex database to the new one, and compact the
|
||||
* range of keys updated. This is used internally by MigrateData.
|
||||
*/
|
||||
static void WriteTxIndexMigrationBatches(CDBWrapper& newdb, CDBWrapper& olddb,
|
||||
CDBBatch& batch_newdb, CDBBatch& batch_olddb,
|
||||
const std::pair<unsigned char, uint256>& begin_key,
|
||||
const std::pair<unsigned char, uint256>& end_key)
|
||||
{
|
||||
// Sync new DB changes to disk before deleting from old DB.
|
||||
newdb.WriteBatch(batch_newdb, /*fSync=*/ true);
|
||||
olddb.WriteBatch(batch_olddb);
|
||||
olddb.CompactRange(begin_key, end_key);
|
||||
|
||||
batch_newdb.Clear();
|
||||
batch_olddb.Clear();
|
||||
}
|
||||
|
||||
bool TxIndex::DB::MigrateData(CBlockTreeDB& block_tree_db, const CBlockLocator& best_locator)
|
||||
{
|
||||
// The prior implementation of txindex was always in sync with block index
|
||||
// and presence was indicated with a boolean DB flag. If the flag is set,
|
||||
// this means the txindex from a previous version is valid and in sync with
|
||||
// the chain tip. The first step of the migration is to unset the flag and
|
||||
// write the chain hash to a separate key, DB_TXINDEX_BLOCK. After that, the
|
||||
// index entries are copied over in batches to the new database. Finally,
|
||||
// DB_TXINDEX_BLOCK is erased from the old database and the block hash is
|
||||
// written to the new database.
|
||||
//
|
||||
// Unsetting the boolean flag ensures that if the node is downgraded to a
|
||||
// previous version, it will not see a corrupted, partially migrated index
|
||||
// -- it will see that the txindex is disabled. When the node is upgraded
|
||||
// again, the migration will pick up where it left off and sync to the block
|
||||
// with hash DB_TXINDEX_BLOCK.
|
||||
bool f_legacy_flag = false;
|
||||
block_tree_db.ReadFlag("txindex", f_legacy_flag);
|
||||
if (f_legacy_flag) {
|
||||
if (!block_tree_db.Write(DB_TXINDEX_BLOCK, best_locator)) {
|
||||
return error("%s: cannot write block indicator", __func__);
|
||||
}
|
||||
if (!block_tree_db.WriteFlag("txindex", false)) {
|
||||
return error("%s: cannot write block index db flag", __func__);
|
||||
}
|
||||
}
|
||||
|
||||
CBlockLocator locator;
|
||||
if (!block_tree_db.Read(DB_TXINDEX_BLOCK, locator)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
int64_t count = 0;
|
||||
LogPrintf("Upgrading txindex database... [0%%]\n");
|
||||
uiInterface.ShowProgress(_("Upgrading txindex database"), 0, true);
|
||||
int report_done = 0;
|
||||
const size_t batch_size = 1 << 24; // 16 MiB
|
||||
|
||||
CDBBatch batch_newdb(*this);
|
||||
CDBBatch batch_olddb(block_tree_db);
|
||||
|
||||
std::pair<unsigned char, uint256> key;
|
||||
std::pair<unsigned char, uint256> begin_key{DB_TXINDEX, uint256()};
|
||||
std::pair<unsigned char, uint256> prev_key = begin_key;
|
||||
|
||||
bool interrupted = false;
|
||||
std::unique_ptr<CDBIterator> cursor(block_tree_db.NewIterator());
|
||||
for (cursor->Seek(begin_key); cursor->Valid(); cursor->Next()) {
|
||||
boost::this_thread::interruption_point();
|
||||
if (ShutdownRequested()) {
|
||||
interrupted = true;
|
||||
break;
|
||||
}
|
||||
|
||||
if (!cursor->GetKey(key)) {
|
||||
return error("%s: cannot get key from valid cursor", __func__);
|
||||
}
|
||||
if (key.first != DB_TXINDEX) {
|
||||
break;
|
||||
}
|
||||
|
||||
// Log progress every 10%.
|
||||
if (++count % 256 == 0) {
|
||||
// Since txids are uniformly random and traversed in increasing order, the high 16 bits
|
||||
// of the hash can be used to estimate the current progress.
|
||||
const uint256& txid = key.second;
|
||||
uint32_t high_nibble =
|
||||
(static_cast<uint32_t>(*(txid.begin() + 0)) << 8) +
|
||||
(static_cast<uint32_t>(*(txid.begin() + 1)) << 0);
|
||||
int percentage_done = (int)(high_nibble * 100.0 / 65536.0 + 0.5);
|
||||
|
||||
uiInterface.ShowProgress(_("Upgrading txindex database"), percentage_done, true);
|
||||
if (report_done < percentage_done/10) {
|
||||
LogPrintf("Upgrading txindex database... [%d%%]\n", percentage_done);
|
||||
report_done = percentage_done/10;
|
||||
}
|
||||
}
|
||||
|
||||
CDiskTxPos value;
|
||||
if (!cursor->GetValue(value)) {
|
||||
return error("%s: cannot parse txindex record", __func__);
|
||||
}
|
||||
batch_newdb.Write(key, value);
|
||||
batch_olddb.Erase(key);
|
||||
|
||||
if (batch_newdb.SizeEstimate() > batch_size || batch_olddb.SizeEstimate() > batch_size) {
|
||||
// NOTE: it's OK to delete the key pointed at by the current DB cursor while iterating
|
||||
// because LevelDB iterators are guaranteed to provide a consistent view of the
|
||||
// underlying data, like a lightweight snapshot.
|
||||
WriteTxIndexMigrationBatches(*this, block_tree_db,
|
||||
batch_newdb, batch_olddb,
|
||||
prev_key, key);
|
||||
prev_key = key;
|
||||
}
|
||||
}
|
||||
|
||||
// If these final DB batches complete the migration, write the best block
|
||||
// hash marker to the new database and delete from the old one. This signals
|
||||
// that the former is fully caught up to that point in the blockchain and
|
||||
// that all txindex entries have been removed from the latter.
|
||||
if (!interrupted) {
|
||||
batch_olddb.Erase(DB_TXINDEX_BLOCK);
|
||||
batch_newdb.Write(DB_BEST_BLOCK, locator);
|
||||
}
|
||||
|
||||
WriteTxIndexMigrationBatches(*this, block_tree_db,
|
||||
batch_newdb, batch_olddb,
|
||||
begin_key, key);
|
||||
|
||||
if (interrupted) {
|
||||
LogPrintf("[CANCELLED].\n");
|
||||
return false;
|
||||
}
|
||||
|
||||
uiInterface.ShowProgress("", 100, false);
|
||||
|
||||
LogPrintf("[DONE].\n");
|
||||
return true;
|
||||
}
|
||||
|
||||
TxIndex::TxIndex(size_t n_cache_size, bool f_memory, bool f_wipe)
|
||||
: m_db(MakeUnique<TxIndex::DB>(n_cache_size, f_memory, f_wipe))
|
||||
{}
|
||||
|
||||
TxIndex::~TxIndex() {}
|
||||
|
||||
bool TxIndex::Init()
|
||||
{
|
||||
LOCK(cs_main);
|
||||
|
|
@ -49,88 +240,7 @@ bool TxIndex::Init()
|
|||
return false;
|
||||
}
|
||||
|
||||
CBlockLocator locator;
|
||||
if (!m_db->ReadBestBlock(locator)) {
|
||||
locator.SetNull();
|
||||
}
|
||||
|
||||
m_best_block_index = FindForkInGlobalIndex(chainActive, locator);
|
||||
m_synced = m_best_block_index.load() == chainActive.Tip();
|
||||
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) {
|
||||
if (m_interrupt) {
|
||||
WriteBestBlock(pindex);
|
||||
return;
|
||||
}
|
||||
|
||||
{
|
||||
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");
|
||||
}
|
||||
return BaseIndex::Init();
|
||||
}
|
||||
|
||||
bool TxIndex::WriteBlock(const CBlock& block, const CBlockIndex* pindex)
|
||||
|
|
@ -145,114 +255,7 @@ bool TxIndex::WriteBlock(const CBlock& block, const CBlockIndex* pindex)
|
|||
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,
|
||||
const std::vector<CTransactionRef>& txn_conflicted)
|
||||
{
|
||||
if (!m_synced) {
|
||||
return;
|
||||
}
|
||||
|
||||
const CBlockIndex* best_block_index = m_best_block_index.load();
|
||||
if (!best_block_index) {
|
||||
if (pindex->nHeight != 0) {
|
||||
FatalError("%s: First block connected is not the genesis block (height=%d)",
|
||||
__func__, pindex->nHeight);
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
// Ensure block connects to an ancestor of the current best block. This should be the case
|
||||
// most of the time, but may not be immediately after the sync thread catches up and sets
|
||||
// m_synced. Consider the case where there is a reorg and the blocks on the stale branch are
|
||||
// in the ValidationInterface queue backlog even after the sync thread has caught up to the
|
||||
// new chain tip. In this unlikely event, log a warning and let the queue clear.
|
||||
if (best_block_index->GetAncestor(pindex->nHeight - 1) != pindex->pprev) {
|
||||
LogPrintf("%s: WARNING: Block %s does not connect to an ancestor of " /* Continued */
|
||||
"known best chain (tip=%s); not updating txindex\n",
|
||||
__func__, pindex->GetBlockHash().ToString(),
|
||||
best_block_index->GetBlockHash().ToString());
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (WriteBlock(*block, pindex)) {
|
||||
m_best_block_index = pindex;
|
||||
} else {
|
||||
FatalError("%s: Failed to write block %s to txindex",
|
||||
__func__, pindex->GetBlockHash().ToString());
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
void TxIndex::ChainStateFlushed(const CBlockLocator& locator)
|
||||
{
|
||||
if (!m_synced) {
|
||||
return;
|
||||
}
|
||||
|
||||
const uint256& locator_tip_hash = locator.vHave.front();
|
||||
const CBlockIndex* locator_tip_index;
|
||||
{
|
||||
LOCK(cs_main);
|
||||
locator_tip_index = LookupBlockIndex(locator_tip_hash);
|
||||
}
|
||||
|
||||
if (!locator_tip_index) {
|
||||
FatalError("%s: First block (hash=%s) in locator was not found",
|
||||
__func__, locator_tip_hash.ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
// This checks that ChainStateFlushed callbacks are received after BlockConnected. The check may fail
|
||||
// immediately after the sync thread catches up and sets m_synced. Consider the case where
|
||||
// there is a reorg and the blocks on the stale branch are in the ValidationInterface queue
|
||||
// backlog even after the sync thread has caught up to the new chain tip. In this unlikely
|
||||
// event, log a warning and let the queue clear.
|
||||
const CBlockIndex* best_block_index = m_best_block_index.load();
|
||||
if (best_block_index->GetAncestor(locator_tip_index->nHeight) != locator_tip_index) {
|
||||
LogPrintf("%s: WARNING: Locator contains block (hash=%s) not on known best " /* Continued */
|
||||
"chain (tip=%s); not writing txindex locator\n",
|
||||
__func__, locator_tip_hash.ToString(),
|
||||
best_block_index->GetBlockHash().ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
if (!m_db->WriteBestBlock(locator)) {
|
||||
error("%s: Failed to write locator to disk", __func__);
|
||||
}
|
||||
}
|
||||
|
||||
bool TxIndex::BlockUntilSyncedToCurrentChain()
|
||||
{
|
||||
AssertLockNotHeld(cs_main);
|
||||
|
||||
if (!m_synced) {
|
||||
return false;
|
||||
}
|
||||
|
||||
{
|
||||
// Skip the queue-draining stuff if we know we're caught up with
|
||||
// chainActive.Tip().
|
||||
LOCK(cs_main);
|
||||
const CBlockIndex* chain_tip = chainActive.Tip();
|
||||
const CBlockIndex* best_block_index = m_best_block_index.load();
|
||||
if (best_block_index->GetAncestor(chain_tip->nHeight) == chain_tip) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
LogPrintf("%s: txindex is catching up on block notifications\n", __func__);
|
||||
SyncWithValidationInterfaceQueue();
|
||||
return true;
|
||||
}
|
||||
BaseIndex::DB& TxIndex::GetDB() const { return *m_db; }
|
||||
|
||||
bool TxIndex::FindTx(const uint256& tx_hash, uint256& block_hash, CTransactionRef& tx) const
|
||||
{
|
||||
|
|
@ -281,31 +284,3 @@ bool TxIndex::FindTx(const uint256& tx_hash, uint256& block_hash, CTransactionRe
|
|||
block_hash = header.GetHash();
|
||||
return true;
|
||||
}
|
||||
|
||||
void TxIndex::Interrupt()
|
||||
{
|
||||
m_interrupt();
|
||||
}
|
||||
|
||||
void TxIndex::Start()
|
||||
{
|
||||
// Need to register this ValidationInterface before running Init(), so that
|
||||
// callbacks are not missed if Init sets m_synced to true.
|
||||
RegisterValidationInterface(this);
|
||||
if (!Init()) {
|
||||
FatalError("%s: txindex failed to initialize", __func__);
|
||||
return;
|
||||
}
|
||||
|
||||
m_thread_sync = std::thread(&TraceThread<std::function<void()>>, "txindex",
|
||||
std::bind(&TxIndex::ThreadSync, this));
|
||||
}
|
||||
|
||||
void TxIndex::Stop()
|
||||
{
|
||||
UnregisterValidationInterface(this);
|
||||
|
||||
if (m_thread_sync.joinable()) {
|
||||
m_thread_sync.join();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,70 +5,39 @@
|
|||
#ifndef BITCOIN_INDEX_TXINDEX_H
|
||||
#define BITCOIN_INDEX_TXINDEX_H
|
||||
|
||||
#include <primitives/block.h>
|
||||
#include <primitives/transaction.h>
|
||||
#include <threadinterrupt.h>
|
||||
#include <chain.h>
|
||||
#include <index/base.h>
|
||||
#include <txdb.h>
|
||||
#include <uint256.h>
|
||||
#include <validationinterface.h>
|
||||
|
||||
class CBlockIndex;
|
||||
|
||||
/**
|
||||
* TxIndex is used to look up transactions included in the blockchain by hash.
|
||||
* The index is written to a LevelDB database and records the filesystem
|
||||
* location of each transaction by transaction hash.
|
||||
*/
|
||||
class TxIndex final : public CValidationInterface
|
||||
class TxIndex final : public BaseIndex
|
||||
{
|
||||
protected:
|
||||
class DB;
|
||||
|
||||
private:
|
||||
const std::unique_ptr<TxIndexDB> m_db;
|
||||
|
||||
/// Whether the index is in sync with the main chain. The flag is flipped
|
||||
/// from false to true once, after which point this starts processing
|
||||
/// ValidationInterface notifications to stay in sync.
|
||||
std::atomic<bool> m_synced;
|
||||
|
||||
/// The last block in the chain that the TxIndex is in sync with.
|
||||
std::atomic<const CBlockIndex*> m_best_block_index;
|
||||
|
||||
std::thread m_thread_sync;
|
||||
CThreadInterrupt m_interrupt;
|
||||
|
||||
/// Initialize internal state from the database and block index.
|
||||
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, and can be
|
||||
/// interrupted with m_interrupt. 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.
|
||||
bool WriteBlock(const CBlock& block, const CBlockIndex* pindex);
|
||||
|
||||
/// Write the current chain block locator to the DB.
|
||||
bool WriteBestBlock(const CBlockIndex* block_index);
|
||||
const std::unique_ptr<DB> m_db;
|
||||
|
||||
protected:
|
||||
void BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
|
||||
const std::vector<CTransactionRef>& txn_conflicted) override;
|
||||
/// Override base class init to migrate from old database.
|
||||
bool Init() override;
|
||||
|
||||
void ChainStateFlushed(const CBlockLocator& locator) override;
|
||||
bool WriteBlock(const CBlock& block, const CBlockIndex* pindex) override;
|
||||
|
||||
BaseIndex::DB& GetDB() const override;
|
||||
|
||||
const char* GetName() const override { return "txindex"; }
|
||||
|
||||
public:
|
||||
/// Constructs the TxIndex, which becomes available to be queried.
|
||||
explicit TxIndex(std::unique_ptr<TxIndexDB> db);
|
||||
/// Constructs the index, which becomes available to be queried.
|
||||
explicit TxIndex(size_t n_cache_size, bool f_memory = false, bool f_wipe = false);
|
||||
|
||||
/// Destructor interrupts sync thread if running and blocks until it exits.
|
||||
~TxIndex();
|
||||
|
||||
/// Blocks the current thread until the transaction 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 queue. If the index is catching
|
||||
/// up from far behind, this method does not block and immediately returns false.
|
||||
bool BlockUntilSyncedToCurrentChain();
|
||||
// Destructor is declared because this class contains a unique_ptr to an incomplete type.
|
||||
virtual ~TxIndex() override;
|
||||
|
||||
/// Look up a transaction by hash.
|
||||
///
|
||||
|
|
@ -77,15 +46,6 @@ public:
|
|||
/// @param[out] tx The transaction itself.
|
||||
/// @return true if transaction is found, false otherwise
|
||||
bool FindTx(const uint256& tx_hash, uint256& block_hash, CTransactionRef& tx) const;
|
||||
|
||||
void Interrupt();
|
||||
|
||||
/// Start initializes the sync state and registers the instance as a
|
||||
/// ValidationInterface so that it stays in sync with blockchain updates.
|
||||
void Start();
|
||||
|
||||
/// Stops the instance from staying in sync with blockchain updates.
|
||||
void Stop();
|
||||
};
|
||||
|
||||
/// The global transaction index, used in GetTransaction. May be null.
|
||||
|
|
|
|||
|
|
@ -1606,8 +1606,7 @@ bool AppInitMain()
|
|||
|
||||
// ********************************************************* Step 8: start indexers
|
||||
if (gArgs.GetBoolArg("-txindex", DEFAULT_TXINDEX)) {
|
||||
auto txindex_db = MakeUnique<TxIndexDB>(nTxIndexCache, false, fReindex);
|
||||
g_txindex = MakeUnique<TxIndex>(std::move(txindex_db));
|
||||
g_txindex = MakeUnique<TxIndex>(nTxIndexCache, false, fReindex);
|
||||
g_txindex->Start();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ BOOST_AUTO_TEST_SUITE(txindex_tests)
|
|||
|
||||
BOOST_FIXTURE_TEST_CASE(txindex_initial_sync, TestChain100Setup)
|
||||
{
|
||||
TxIndex txindex(MakeUnique<TxIndexDB>(1 << 20, true));
|
||||
TxIndex txindex(1 << 20, true);
|
||||
|
||||
CTransactionRef tx_disk;
|
||||
uint256 block_hash;
|
||||
|
|
|
|||
183
src/txdb.cpp
183
src/txdb.cpp
|
|
@ -21,8 +21,6 @@
|
|||
static const char DB_COIN = 'C';
|
||||
static const char DB_COINS = 'c';
|
||||
static const char DB_BLOCK_FILES = 'f';
|
||||
static const char DB_TXINDEX = 't';
|
||||
static const char DB_TXINDEX_BLOCK = 'T';
|
||||
static const char DB_BLOCK_INDEX = 'b';
|
||||
|
||||
static const char DB_BEST_BLOCK = 'B';
|
||||
|
|
@ -237,17 +235,6 @@ bool CBlockTreeDB::WriteBatchSync(const std::vector<std::pair<int, const CBlockF
|
|||
return WriteBatch(batch, true);
|
||||
}
|
||||
|
||||
bool CBlockTreeDB::ReadTxIndex(const uint256 &txid, CDiskTxPos &pos) {
|
||||
return Read(std::make_pair(DB_TXINDEX, txid), pos);
|
||||
}
|
||||
|
||||
bool CBlockTreeDB::WriteTxIndex(const std::vector<std::pair<uint256, CDiskTxPos> >&vect) {
|
||||
CDBBatch batch(*this);
|
||||
for (std::vector<std::pair<uint256,CDiskTxPos> >::const_iterator it=vect.begin(); it!=vect.end(); it++)
|
||||
batch.Write(std::make_pair(DB_TXINDEX, it->first), it->second);
|
||||
return WriteBatch(batch);
|
||||
}
|
||||
|
||||
bool CBlockTreeDB::WriteFlag(const std::string &name, bool fValue) {
|
||||
return Write(std::make_pair(DB_FLAG, name), fValue ? '1' : '0');
|
||||
}
|
||||
|
|
@ -425,173 +412,3 @@ bool CCoinsViewDB::Upgrade() {
|
|||
LogPrintf("[%s].\n", ShutdownRequested() ? "CANCELLED" : "DONE");
|
||||
return !ShutdownRequested();
|
||||
}
|
||||
|
||||
TxIndexDB::TxIndexDB(size_t n_cache_size, bool f_memory, bool f_wipe) :
|
||||
CDBWrapper(GetDataDir() / "indexes" / "txindex", n_cache_size, f_memory, f_wipe)
|
||||
{}
|
||||
|
||||
bool TxIndexDB::ReadTxPos(const uint256 &txid, CDiskTxPos& pos) const
|
||||
{
|
||||
return Read(std::make_pair(DB_TXINDEX, txid), pos);
|
||||
}
|
||||
|
||||
bool TxIndexDB::WriteTxs(const std::vector<std::pair<uint256, CDiskTxPos>>& v_pos)
|
||||
{
|
||||
CDBBatch batch(*this);
|
||||
for (const auto& tuple : v_pos) {
|
||||
batch.Write(std::make_pair(DB_TXINDEX, tuple.first), tuple.second);
|
||||
}
|
||||
return WriteBatch(batch);
|
||||
}
|
||||
|
||||
bool TxIndexDB::ReadBestBlock(CBlockLocator& locator) const
|
||||
{
|
||||
bool success = Read(DB_BEST_BLOCK, locator);
|
||||
if (!success) {
|
||||
locator.SetNull();
|
||||
}
|
||||
return success;
|
||||
}
|
||||
|
||||
bool TxIndexDB::WriteBestBlock(const CBlockLocator& locator)
|
||||
{
|
||||
return Write(DB_BEST_BLOCK, locator);
|
||||
}
|
||||
|
||||
/*
|
||||
* Safely persist a transfer of data from the old txindex database to the new one, and compact the
|
||||
* range of keys updated. This is used internally by MigrateData.
|
||||
*/
|
||||
static void WriteTxIndexMigrationBatches(TxIndexDB& newdb, CBlockTreeDB& olddb,
|
||||
CDBBatch& batch_newdb, CDBBatch& batch_olddb,
|
||||
const std::pair<unsigned char, uint256>& begin_key,
|
||||
const std::pair<unsigned char, uint256>& end_key)
|
||||
{
|
||||
// Sync new DB changes to disk before deleting from old DB.
|
||||
newdb.WriteBatch(batch_newdb, /*fSync=*/ true);
|
||||
olddb.WriteBatch(batch_olddb);
|
||||
olddb.CompactRange(begin_key, end_key);
|
||||
|
||||
batch_newdb.Clear();
|
||||
batch_olddb.Clear();
|
||||
}
|
||||
|
||||
bool TxIndexDB::MigrateData(CBlockTreeDB& block_tree_db, const CBlockLocator& best_locator)
|
||||
{
|
||||
// The prior implementation of txindex was always in sync with block index
|
||||
// and presence was indicated with a boolean DB flag. If the flag is set,
|
||||
// this means the txindex from a previous version is valid and in sync with
|
||||
// the chain tip. The first step of the migration is to unset the flag and
|
||||
// write the chain hash to a separate key, DB_TXINDEX_BLOCK. After that, the
|
||||
// index entries are copied over in batches to the new database. Finally,
|
||||
// DB_TXINDEX_BLOCK is erased from the old database and the block hash is
|
||||
// written to the new database.
|
||||
//
|
||||
// Unsetting the boolean flag ensures that if the node is downgraded to a
|
||||
// previous version, it will not see a corrupted, partially migrated index
|
||||
// -- it will see that the txindex is disabled. When the node is upgraded
|
||||
// again, the migration will pick up where it left off and sync to the block
|
||||
// with hash DB_TXINDEX_BLOCK.
|
||||
bool f_legacy_flag = false;
|
||||
block_tree_db.ReadFlag("txindex", f_legacy_flag);
|
||||
if (f_legacy_flag) {
|
||||
if (!block_tree_db.Write(DB_TXINDEX_BLOCK, best_locator)) {
|
||||
return error("%s: cannot write block indicator", __func__);
|
||||
}
|
||||
if (!block_tree_db.WriteFlag("txindex", false)) {
|
||||
return error("%s: cannot write block index db flag", __func__);
|
||||
}
|
||||
}
|
||||
|
||||
CBlockLocator locator;
|
||||
if (!block_tree_db.Read(DB_TXINDEX_BLOCK, locator)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
int64_t count = 0;
|
||||
LogPrintf("Upgrading txindex database... [0%%]\n");
|
||||
uiInterface.ShowProgress(_("Upgrading txindex database"), 0, true);
|
||||
int report_done = 0;
|
||||
const size_t batch_size = 1 << 24; // 16 MiB
|
||||
|
||||
CDBBatch batch_newdb(*this);
|
||||
CDBBatch batch_olddb(block_tree_db);
|
||||
|
||||
std::pair<unsigned char, uint256> key;
|
||||
std::pair<unsigned char, uint256> begin_key{DB_TXINDEX, uint256()};
|
||||
std::pair<unsigned char, uint256> prev_key = begin_key;
|
||||
|
||||
bool interrupted = false;
|
||||
std::unique_ptr<CDBIterator> cursor(block_tree_db.NewIterator());
|
||||
for (cursor->Seek(begin_key); cursor->Valid(); cursor->Next()) {
|
||||
boost::this_thread::interruption_point();
|
||||
if (ShutdownRequested()) {
|
||||
interrupted = true;
|
||||
break;
|
||||
}
|
||||
|
||||
if (!cursor->GetKey(key)) {
|
||||
return error("%s: cannot get key from valid cursor", __func__);
|
||||
}
|
||||
if (key.first != DB_TXINDEX) {
|
||||
break;
|
||||
}
|
||||
|
||||
// Log progress every 10%.
|
||||
if (++count % 256 == 0) {
|
||||
// Since txids are uniformly random and traversed in increasing order, the high 16 bits
|
||||
// of the hash can be used to estimate the current progress.
|
||||
const uint256& txid = key.second;
|
||||
uint32_t high_nibble =
|
||||
(static_cast<uint32_t>(*(txid.begin() + 0)) << 8) +
|
||||
(static_cast<uint32_t>(*(txid.begin() + 1)) << 0);
|
||||
int percentage_done = (int)(high_nibble * 100.0 / 65536.0 + 0.5);
|
||||
|
||||
uiInterface.ShowProgress(_("Upgrading txindex database"), percentage_done, true);
|
||||
if (report_done < percentage_done/10) {
|
||||
LogPrintf("Upgrading txindex database... [%d%%]\n", percentage_done);
|
||||
report_done = percentage_done/10;
|
||||
}
|
||||
}
|
||||
|
||||
CDiskTxPos value;
|
||||
if (!cursor->GetValue(value)) {
|
||||
return error("%s: cannot parse txindex record", __func__);
|
||||
}
|
||||
batch_newdb.Write(key, value);
|
||||
batch_olddb.Erase(key);
|
||||
|
||||
if (batch_newdb.SizeEstimate() > batch_size || batch_olddb.SizeEstimate() > batch_size) {
|
||||
// NOTE: it's OK to delete the key pointed at by the current DB cursor while iterating
|
||||
// because LevelDB iterators are guaranteed to provide a consistent view of the
|
||||
// underlying data, like a lightweight snapshot.
|
||||
WriteTxIndexMigrationBatches(*this, block_tree_db,
|
||||
batch_newdb, batch_olddb,
|
||||
prev_key, key);
|
||||
prev_key = key;
|
||||
}
|
||||
}
|
||||
|
||||
// If these final DB batches complete the migration, write the best block
|
||||
// hash marker to the new database and delete from the old one. This signals
|
||||
// that the former is fully caught up to that point in the blockchain and
|
||||
// that all txindex entries have been removed from the latter.
|
||||
if (!interrupted) {
|
||||
batch_olddb.Erase(DB_TXINDEX_BLOCK);
|
||||
batch_newdb.Write(DB_BEST_BLOCK, locator);
|
||||
}
|
||||
|
||||
WriteTxIndexMigrationBatches(*this, block_tree_db,
|
||||
batch_newdb, batch_olddb,
|
||||
begin_key, key);
|
||||
|
||||
if (interrupted) {
|
||||
LogPrintf("[CANCELLED].\n");
|
||||
return false;
|
||||
}
|
||||
|
||||
uiInterface.ShowProgress("", 100, false);
|
||||
|
||||
LogPrintf("[DONE].\n");
|
||||
return true;
|
||||
}
|
||||
|
|
|
|||
59
src/txdb.h
59
src/txdb.h
|
|
@ -40,31 +40,6 @@ static const int64_t nMaxTxIndexCache = 1024;
|
|||
//! Max memory allocated to coin DB specific cache (MiB)
|
||||
static const int64_t nMaxCoinsDBCache = 8;
|
||||
|
||||
struct CDiskTxPos : public CDiskBlockPos
|
||||
{
|
||||
unsigned int nTxOffset; // after header
|
||||
|
||||
ADD_SERIALIZE_METHODS;
|
||||
|
||||
template <typename Stream, typename Operation>
|
||||
inline void SerializationOp(Stream& s, Operation ser_action) {
|
||||
READWRITEAS(CDiskBlockPos, *this);
|
||||
READWRITE(VARINT(nTxOffset));
|
||||
}
|
||||
|
||||
CDiskTxPos(const CDiskBlockPos &blockIn, unsigned int nTxOffsetIn) : CDiskBlockPos(blockIn.nFile, blockIn.nPos), nTxOffset(nTxOffsetIn) {
|
||||
}
|
||||
|
||||
CDiskTxPos() {
|
||||
SetNull();
|
||||
}
|
||||
|
||||
void SetNull() {
|
||||
CDiskBlockPos::SetNull();
|
||||
nTxOffset = 0;
|
||||
}
|
||||
};
|
||||
|
||||
/** CCoinsView backed by the coin database (chainstate/) */
|
||||
class CCoinsViewDB final : public CCoinsView
|
||||
{
|
||||
|
|
@ -118,43 +93,9 @@ public:
|
|||
bool ReadLastBlockFile(int &nFile);
|
||||
bool WriteReindexing(bool fReindexing);
|
||||
bool ReadReindexing(bool &fReindexing);
|
||||
bool ReadTxIndex(const uint256 &txid, CDiskTxPos &pos);
|
||||
bool WriteTxIndex(const std::vector<std::pair<uint256, CDiskTxPos> > &vect);
|
||||
bool WriteFlag(const std::string &name, bool fValue);
|
||||
bool ReadFlag(const std::string &name, bool &fValue);
|
||||
bool LoadBlockIndexGuts(const Consensus::Params& consensusParams, std::function<CBlockIndex*(const uint256&)> insertBlockIndex);
|
||||
};
|
||||
|
||||
/**
|
||||
* Access to the txindex database (indexes/txindex/)
|
||||
*
|
||||
* The database stores a block locator of the chain the database is synced to
|
||||
* so that the TxIndex can efficiently determine the point it last stopped at.
|
||||
* A locator is used instead of a simple hash of the chain tip because blocks
|
||||
* and block index entries may not be flushed to disk until after this database
|
||||
* is updated.
|
||||
*/
|
||||
class TxIndexDB : public CDBWrapper
|
||||
{
|
||||
public:
|
||||
explicit TxIndexDB(size_t n_cache_size, bool f_memory = false, bool f_wipe = false);
|
||||
|
||||
/// Read the disk location of the transaction data with the given hash. Returns false if the
|
||||
/// transaction hash is not indexed.
|
||||
bool ReadTxPos(const uint256& txid, CDiskTxPos& pos) const;
|
||||
|
||||
/// Write a batch of transaction positions to the DB.
|
||||
bool WriteTxs(const std::vector<std::pair<uint256, CDiskTxPos>>& v_pos);
|
||||
|
||||
/// Read block locator of the chain that the txindex is in sync with.
|
||||
bool ReadBestBlock(CBlockLocator& locator) const;
|
||||
|
||||
/// Write block locator of the chain that the txindex is in sync with.
|
||||
bool WriteBestBlock(const CBlockLocator& locator);
|
||||
|
||||
/// Migrate txindex data from the block tree DB, where it may be for older nodes that have not
|
||||
/// been upgraded yet to the new database.
|
||||
bool MigrateData(CBlockTreeDB& block_tree_db, const CBlockLocator& best_locator);
|
||||
};
|
||||
|
||||
#endif // BITCOIN_TXDB_H
|
||||
|
|
|
|||
Loading…
Reference in a new issue