diff options
Diffstat (limited to 'src/index')
| -rw-r--r-- | src/index/base.cpp | 278 | ||||
| -rw-r--r-- | src/index/base.h | 98 | ||||
| -rw-r--r-- | src/index/txindex.cpp | 419 | ||||
| -rw-r--r-- | src/index/txindex.h | 74 |
4 files changed, 590 insertions, 279 deletions
diff --git a/src/index/base.cpp b/src/index/base.cpp new file mode 100644 index 0000000000..738166dc94 --- /dev/null +++ b/src/index/base.cpp @@ -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(); + } +} diff --git a/src/index/base.h b/src/index/base.h new file mode 100644 index 0000000000..04ee6e6cc2 --- /dev/null +++ b/src/index/base.h @@ -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 diff --git a/src/index/txindex.cpp b/src/index/txindex.cpp index 3ff16b7664..e106b9b420 100644 --- a/src/index/txindex.cpp +++ b/src/index/txindex.cpp @@ -2,258 +2,261 @@ // 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 -TxIndex::TxIndex(std::unique_ptr<TxIndexDB> db) : - m_db(std::move(db)), m_synced(false), m_best_block_index(nullptr) -{} + ADD_SERIALIZE_METHODS; -TxIndex::~TxIndex() -{ - Interrupt(); - Stop(); -} - -bool TxIndex::Init() -{ - LOCK(cs_main); - - // Attempt to migrate txindex from the old database to the new one. Even if - // chain_tip is null, the node could be reindexing and we still want to - // delete txindex records in the old database. - if (!m_db->MigrateData(*pblocktree, chainActive.GetLocator())) { - return false; + template <typename Stream, typename Operation> + inline void SerializationOp(Stream& s, Operation ser_action) { + READWRITEAS(CDiskBlockPos, *this); + READWRITE(VARINT(nTxOffset)); } - CBlockLocator locator; - if (!m_db->ReadBestBlock(locator)) { - locator.SetNull(); + CDiskTxPos(const CDiskBlockPos &blockIn, unsigned int nTxOffsetIn) : CDiskBlockPos(blockIn.nFile, blockIn.nPos), nTxOffset(nTxOffsetIn) { } - 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(); + CDiskTxPos() { + SetNull(); } - const CBlockIndex* pindex = chainActive.Next(pindex_prev); - if (pindex) { - return pindex; + void SetNull() { + CDiskBlockPos::SetNull(); + nTxOffset = 0; } - - return chainActive.Next(chainActive.FindFork(pindex_prev)); -} - -void TxIndex::ThreadSync() +}; + +/** + * 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 { - 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; - } +public: + explicit DB(size_t n_cache_size, bool f_memory = false, bool f_wipe = false); - { - 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; - } + /// 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; - 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; - } + /// Write a batch of transaction positions to the DB. + bool WriteTxs(const std::vector<std::pair<uint256, CDiskTxPos>>& v_pos); - if (last_locator_write_time + SYNC_LOCATOR_WRITE_INTERVAL < current_time) { - WriteBestBlock(pindex); - last_locator_write_time = current_time; - } + /// 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); +}; - 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; - } - } - } +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) +{} - if (pindex) { - LogPrintf("txindex is enabled at height %d\n", pindex->nHeight); - } else { - LogPrintf("txindex is enabled\n"); - } +bool TxIndex::DB::ReadTxPos(const uint256 &txid, CDiskTxPos& pos) const +{ + return Read(std::make_pair(DB_TXINDEX, txid), pos); } -bool TxIndex::WriteBlock(const CBlock& block, const CBlockIndex* pindex) +bool TxIndex::DB::WriteTxs(const std::vector<std::pair<uint256, CDiskTxPos>>& v_pos) { - CDiskTxPos pos(pindex->GetBlockPos(), GetSizeOfCompactSize(block.vtx.size())); - std::vector<std::pair<uint256, CDiskTxPos>> vPos; - vPos.reserve(block.vtx.size()); - for (const auto& tx : block.vtx) { - vPos.emplace_back(tx->GetHash(), pos); - pos.nTxOffset += ::GetSerializeSize(*tx, SER_DISK, CLIENT_VERSION); + CDBBatch batch(*this); + for (const auto& tuple : v_pos) { + batch.Write(std::make_pair(DB_TXINDEX, tuple.first), tuple.second); } - return m_db->WriteTxs(vPos); + return WriteBatch(batch); } -bool TxIndex::WriteBestBlock(const CBlockIndex* block_index) +/* + * 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) { - LOCK(cs_main); - if (!m_db->WriteBestBlock(chainActive.GetLocator(block_index))) { - return error("%s: Failed to write locator to disk", __func__); - } - return true; + // 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(); } -void TxIndex::BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex, - const std::vector<CTransactionRef>& txn_conflicted) +bool TxIndex::DB::MigrateData(CBlockTreeDB& block_tree_db, const CBlockLocator& best_locator) { - 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; + // 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__); } - } 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 (!block_tree_db.WriteFlag("txindex", false)) { + return error("%s: cannot write block index db flag", __func__); } } - if (WriteBlock(*block, pindex)) { - m_best_block_index = pindex; - } else { - FatalError("%s: Failed to write block %s to txindex", - __func__, pindex->GetBlockHash().ToString()); - return; + CBlockLocator locator; + if (!block_tree_db.Read(DB_TXINDEX_BLOCK, locator)) { + return true; } -} -void TxIndex::ChainStateFlushed(const CBlockLocator& locator) -{ - if (!m_synced) { - return; - } + 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; + } - const uint256& locator_tip_hash = locator.vHave.front(); - const CBlockIndex* locator_tip_index; - { - LOCK(cs_main); - locator_tip_index = LookupBlockIndex(locator_tip_hash); - } + if (!cursor->GetKey(key)) { + return error("%s: cannot get key from valid cursor", __func__); + } + if (key.first != DB_TXINDEX) { + break; + } - if (!locator_tip_index) { - FatalError("%s: First block (hash=%s) in locator was not found", - __func__, locator_tip_hash.ToString()); - return; + // 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; + } } - // 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 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); } - if (!m_db->WriteBestBlock(locator)) { - error("%s: Failed to write locator to disk", __func__); + 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; } -bool TxIndex::BlockUntilSyncedToCurrentChain() +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() { - AssertLockNotHeld(cs_main); + LOCK(cs_main); - if (!m_synced) { + // Attempt to migrate txindex from the old database to the new one. Even if + // chain_tip is null, the node could be reindexing and we still want to + // delete txindex records in the old database. + if (!m_db->MigrateData(*pblocktree, chainActive.GetLocator())) { 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; - } - } + return BaseIndex::Init(); +} - LogPrintf("%s: txindex is catching up on block notifications\n", __func__); - SyncWithValidationInterfaceQueue(); - return true; +bool TxIndex::WriteBlock(const CBlock& block, const CBlockIndex* pindex) +{ + CDiskTxPos pos(pindex->GetBlockPos(), GetSizeOfCompactSize(block.vtx.size())); + std::vector<std::pair<uint256, CDiskTxPos>> vPos; + vPos.reserve(block.vtx.size()); + for (const auto& tx : block.vtx) { + vPos.emplace_back(tx->GetHash(), pos); + pos.nTxOffset += ::GetSerializeSize(*tx, SER_DISK, CLIENT_VERSION); + } + return m_db->WriteTxs(vPos); } +BaseIndex::DB& TxIndex::GetDB() const { return *m_db; } + bool TxIndex::FindTx(const uint256& tx_hash, uint256& block_hash, CTransactionRef& tx) const { CDiskTxPos postx; @@ -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(); - } -} diff --git a/src/index/txindex.h b/src/index/txindex.h index 4937bd64e9..8202c3c951 100644 --- a/src/index/txindex.h +++ b/src/index/txindex.h @@ -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 { -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(); +protected: + class DB; - /// 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(); +private: + const std::unique_ptr<DB> m_db; - /// Write update index entries for a newly connected block. - bool WriteBlock(const CBlock& block, const CBlockIndex* pindex); +protected: + /// Override base class init to migrate from old database. + bool Init() override; - /// Write the current chain block locator to the DB. - bool WriteBestBlock(const CBlockIndex* block_index); + bool WriteBlock(const CBlock& block, const CBlockIndex* pindex) override; -protected: - void BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex, - const std::vector<CTransactionRef>& txn_conflicted) override; + BaseIndex::DB& GetDB() const override; - void ChainStateFlushed(const CBlockLocator& locator) 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); - - /// Destructor interrupts sync thread if running and blocks until it exits. - ~TxIndex(); + /// Constructs the index, which becomes available to be queried. + explicit TxIndex(size_t n_cache_size, bool f_memory = false, bool f_wipe = false); - /// 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. |