index: Extract logic from TxIndex into reusable base class.

This commit is contained in:
Jim Posen 2018-05-15 14:47:37 -07:00
parent e5af5fc6fb
commit 61a1226d87
2 changed files with 80 additions and 53 deletions

View File

@ -28,11 +28,9 @@ static void FatalError(const char* fmt, const Args&... args)
StartShutdown(); StartShutdown();
} }
TxIndex::TxIndex(std::unique_ptr<TxIndexDB> db) : TxIndex::TxIndex(std::unique_ptr<TxIndexDB> db) : m_db(std::move(db)) {}
m_db(std::move(db)), m_synced(false), m_best_block_index(nullptr)
{}
TxIndex::~TxIndex() BaseIndex::~BaseIndex()
{ {
Interrupt(); Interrupt();
Stop(); Stop();
@ -49,11 +47,17 @@ bool TxIndex::Init()
return false; return false;
} }
return BaseIndex::Init();
}
bool BaseIndex::Init()
{
CBlockLocator locator; CBlockLocator locator;
if (!m_db->ReadBestBlock(locator)) { if (!GetDB().ReadBestBlock(locator)) {
locator.SetNull(); locator.SetNull();
} }
LOCK(cs_main);
m_best_block_index = FindForkInGlobalIndex(chainActive, locator); m_best_block_index = FindForkInGlobalIndex(chainActive, locator);
m_synced = m_best_block_index.load() == chainActive.Tip(); m_synced = m_best_block_index.load() == chainActive.Tip();
return true; return true;
@ -75,7 +79,7 @@ static const CBlockIndex* NextSyncBlock(const CBlockIndex* pindex_prev)
return chainActive.Next(chainActive.FindFork(pindex_prev)); return chainActive.Next(chainActive.FindFork(pindex_prev));
} }
void TxIndex::ThreadSync() void BaseIndex::ThreadSync()
{ {
const CBlockIndex* pindex = m_best_block_index.load(); const CBlockIndex* pindex = m_best_block_index.load();
if (!m_synced) { if (!m_synced) {
@ -145,16 +149,18 @@ 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) BaseIndexDB& TxIndex::GetDB() const { return *m_db; }
bool BaseIndex::WriteBestBlock(const CBlockIndex* block_index)
{ {
LOCK(cs_main); LOCK(cs_main);
if (!m_db->WriteBestBlock(chainActive.GetLocator(block_index))) { if (!GetDB().WriteBestBlock(chainActive.GetLocator(block_index))) {
return error("%s: Failed to write locator to disk", __func__); return error("%s: Failed to write locator to disk", __func__);
} }
return true; return true;
} }
void TxIndex::BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex, void BaseIndex::BlockConnected(const std::shared_ptr<const CBlock>& block, const CBlockIndex* pindex,
const std::vector<CTransactionRef>& txn_conflicted) const std::vector<CTransactionRef>& txn_conflicted)
{ {
if (!m_synced) { if (!m_synced) {
@ -192,7 +198,7 @@ void TxIndex::BlockConnected(const std::shared_ptr<const CBlock>& block, const C
} }
} }
void TxIndex::ChainStateFlushed(const CBlockLocator& locator) void BaseIndex::ChainStateFlushed(const CBlockLocator& locator)
{ {
if (!m_synced) { if (!m_synced) {
return; return;
@ -225,12 +231,12 @@ void TxIndex::ChainStateFlushed(const CBlockLocator& locator)
return; return;
} }
if (!m_db->WriteBestBlock(locator)) { if (!GetDB().WriteBestBlock(locator)) {
error("%s: Failed to write locator to disk", __func__); error("%s: Failed to write locator to disk", __func__);
} }
} }
bool TxIndex::BlockUntilSyncedToCurrentChain() bool BaseIndex::BlockUntilSyncedToCurrentChain()
{ {
AssertLockNotHeld(cs_main); AssertLockNotHeld(cs_main);
@ -282,12 +288,12 @@ bool TxIndex::FindTx(const uint256& tx_hash, uint256& block_hash, CTransactionRe
return true; return true;
} }
void TxIndex::Interrupt() void BaseIndex::Interrupt()
{ {
m_interrupt(); m_interrupt();
} }
void TxIndex::Start() void BaseIndex::Start()
{ {
// Need to register this ValidationInterface before running Init(), so that // Need to register this ValidationInterface before running Init(), so that
// callbacks are not missed if Init sets m_synced to true. // callbacks are not missed if Init sets m_synced to true.
@ -298,10 +304,10 @@ void TxIndex::Start()
} }
m_thread_sync = std::thread(&TraceThread<std::function<void()>>, "txindex", m_thread_sync = std::thread(&TraceThread<std::function<void()>>, "txindex",
std::bind(&TxIndex::ThreadSync, this)); std::bind(&BaseIndex::ThreadSync, this));
} }
void TxIndex::Stop() void BaseIndex::Stop()
{ {
UnregisterValidationInterface(this); UnregisterValidationInterface(this);

View File

@ -15,39 +15,31 @@
class CBlockIndex; class CBlockIndex;
/** /**
* TxIndex is used to look up transactions included in the blockchain by hash. * Base class for indices of blockchain data. This implements
* The index is written to a LevelDB database and records the filesystem * CValidationInterface and ensures blocks are indexed sequentially according
* location of each transaction by transaction hash. * to their position in the active chain.
*/ */
class TxIndex final : public CValidationInterface class BaseIndex : public CValidationInterface
{ {
private: private:
const std::unique_ptr<TxIndexDB> m_db;
/// Whether the index is in sync with the main chain. The flag is flipped /// 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 /// from false to true once, after which point this starts processing
/// ValidationInterface notifications to stay in sync. /// ValidationInterface notifications to stay in sync.
std::atomic<bool> m_synced; std::atomic<bool> m_synced{false};
/// The last block in the chain that the TxIndex is in sync with. /// The last block in the chain that the index is in sync with.
std::atomic<const CBlockIndex*> m_best_block_index; std::atomic<const CBlockIndex*> m_best_block_index{nullptr};
std::thread m_thread_sync; std::thread m_thread_sync;
CThreadInterrupt m_interrupt; CThreadInterrupt m_interrupt;
/// Initialize internal state from the database and block index. /// Sync the index with the block index starting from the current best block.
bool Init(); /// 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
/// Sync the tx index with the block index starting from the current best /// flag is set and the BlockConnected ValidationInterface callback takes
/// block. Intended to be run in its own thread, m_thread_sync, and can be /// over and the sync thread exits.
/// 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(); 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. /// Write the current chain block locator to the DB.
bool WriteBestBlock(const CBlockIndex* block_index); bool WriteBestBlock(const CBlockIndex* block_index);
@ -57,27 +49,25 @@ protected:
void ChainStateFlushed(const CBlockLocator& locator) 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 BaseIndexDB& GetDB() const = 0;
public: 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. /// Destructor interrupts sync thread if running and blocks until it exits.
~TxIndex(); virtual ~BaseIndex();
/// Blocks the current thread until the transaction index is caught up to /// Blocks the current thread until the index is caught up to the current
/// the current state of the block chain. This only blocks if the index has gotten in sync once /// state of the block chain. This only blocks if the index has gotten in
/// and only needs to process blocks in the ValidationInterface queue. If the index is catching /// sync once and only needs to process blocks in the ValidationInterface
/// up from far behind, this method does not block and immediately returns false. /// queue. If the index is catching up from far behind, this method does
/// not block and immediately returns false.
bool BlockUntilSyncedToCurrentChain(); bool BlockUntilSyncedToCurrentChain();
/// Look up a transaction by hash.
///
/// @param[in] tx_hash The hash of the transaction to be returned.
/// @param[out] block_hash The hash of the block the transaction is found in.
/// @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(); void Interrupt();
/// Start initializes the sync state and registers the instance as a /// Start initializes the sync state and registers the instance as a
@ -88,6 +78,37 @@ public:
void Stop(); void Stop();
}; };
/**
* 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 BaseIndex
{
private:
const std::unique_ptr<TxIndexDB> m_db;
protected:
/// Override base class init to migrate from old database.
bool Init() override;
bool WriteBlock(const CBlock& block, const CBlockIndex* pindex) override;
BaseIndexDB& GetDB() const override;
public:
/// Constructs the index, which becomes available to be queried.
explicit TxIndex(std::unique_ptr<TxIndexDB> db);
/// Look up a transaction by hash.
///
/// @param[in] tx_hash The hash of the transaction to be returned.
/// @param[out] block_hash The hash of the block the transaction is found in.
/// @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;
};
/// The global transaction index, used in GetTransaction. May be null. /// The global transaction index, used in GetTransaction. May be null.
extern std::unique_ptr<TxIndex> g_txindex; extern std::unique_ptr<TxIndex> g_txindex;