diff --git a/src/validationinterface.cpp b/src/validationinterface.cpp index fb923e8c0..b696a6df4 100644 --- a/src/validationinterface.cpp +++ b/src/validationinterface.cpp @@ -1,252 +1,276 @@ // Copyright (c) 2009-2010 Satoshi Nakamoto // Copyright (c) 2009-2016 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 #include #include #include #include #include #include #include #include #include #include #include -#include - -struct ValidationInterfaceConnections { - boost::signals2::scoped_connection UpdatedBlockTip; - boost::signals2::scoped_connection TransactionAddedToMempool; - boost::signals2::scoped_connection BlockConnected; - boost::signals2::scoped_connection BlockDisconnected; - boost::signals2::scoped_connection TransactionRemovedFromMempool; - boost::signals2::scoped_connection ChainStateFlushed; - boost::signals2::scoped_connection BlockChecked; - boost::signals2::scoped_connection NewPoWValidBlock; -}; - +//! The MainSignalsInstance manages a list of shared_ptr +//! callbacks. +//! +//! A std::unordered_map is used to track what callbacks are currently +//! registered, and a std::list is to used to store the callbacks that are +//! currently registered as well as any callbacks that are just unregistered +//! and about to be deleted when they are done executing. struct MainSignalsInstance { - boost::signals2::signal - UpdatedBlockTip; - boost::signals2::signal - TransactionAddedToMempool; - boost::signals2::signal &, - const CBlockIndex *pindex, - const std::vector &)> - BlockConnected; - boost::signals2::signal &)> - BlockDisconnected; - boost::signals2::signal - TransactionRemovedFromMempool; - boost::signals2::signal ChainStateFlushed; - boost::signals2::signal - BlockChecked; - boost::signals2::signal &)> - NewPoWValidBlock; +private: + Mutex m_mutex; + //! List entries consist of a callback pointer and reference count. The + //! count is equal to the number of current executions of that entry, plus 1 + //! if it's registered. It cannot be 0 because that would imply it is + //! unregistered and also not being executed (so shouldn't exist). + struct ListEntry { + std::shared_ptr callbacks; + int count = 1; + }; + std::list m_list GUARDED_BY(m_mutex); + std::unordered_map::iterator> + m_map GUARDED_BY(m_mutex); +public: // We are not allowed to assume the scheduler only runs in one thread, // but must ensure all callbacks happen in-order, so we end up creating // our own queue here :( SingleThreadedSchedulerClient m_schedulerClient; - std::unordered_map - m_connMainSignals; explicit MainSignalsInstance(CScheduler *pscheduler) : m_schedulerClient(pscheduler) {} + + void Register(std::shared_ptr callbacks) { + LOCK(m_mutex); + auto inserted = m_map.emplace(callbacks.get(), m_list.end()); + if (inserted.second) { + inserted.first->second = m_list.emplace(m_list.end()); + } + inserted.first->second->callbacks = std::move(callbacks); + } + + void Unregister(CValidationInterface *callbacks) { + LOCK(m_mutex); + auto it = m_map.find(callbacks); + if (it != m_map.end()) { + if (!--it->second->count) { + m_list.erase(it->second); + } + m_map.erase(it); + } + } + + //! Clear unregisters every previously registered callback, erasing every + //! map entry. After this call, the list may still contain callbacks that + //! are currently executing, but it will be cleared when they are done + //! executing. + void Clear() { + LOCK(m_mutex); + for (auto it = m_list.begin(); it != m_list.end();) { + it = --it->count ? std::next(it) : m_list.erase(it); + } + m_map.clear(); + } + + template void Iterate(F &&f) { + WAIT_LOCK(m_mutex, lock); + for (auto it = m_list.begin(); it != m_list.end();) { + ++it->count; + { + REVERSE_LOCK(lock); + f(*it->callbacks); + } + it = --it->count ? std::next(it) : m_list.erase(it); + } + } }; static CMainSignals g_signals; void CMainSignals::RegisterBackgroundSignalScheduler(CScheduler &scheduler) { assert(!m_internals); m_internals.reset(new MainSignalsInstance(&scheduler)); } void CMainSignals::UnregisterBackgroundSignalScheduler() { m_internals.reset(nullptr); } void CMainSignals::FlushBackgroundCallbacks() { if (m_internals) { m_internals->m_schedulerClient.EmptyQueue(); } } size_t CMainSignals::CallbacksPending() { if (!m_internals) { return 0; } return m_internals->m_schedulerClient.CallbacksPending(); } CMainSignals &GetMainSignals() { return g_signals; } void RegisterSharedValidationInterface( std::shared_ptr pwalletIn) { // Each connection captures pwalletIn to ensure that each callback is // executed before pwalletIn is destroyed. For more details see #18338. - ValidationInterfaceConnections &conns = - g_signals.m_internals->m_connMainSignals[pwalletIn.get()]; - conns.UpdatedBlockTip = g_signals.m_internals->UpdatedBlockTip.connect( - std::bind(&CValidationInterface::UpdatedBlockTip, pwalletIn, - std::placeholders::_1, std::placeholders::_2, - std::placeholders::_3)); - conns.TransactionAddedToMempool = - g_signals.m_internals->TransactionAddedToMempool.connect( - std::bind(&CValidationInterface::TransactionAddedToMempool, - pwalletIn, std::placeholders::_1)); - conns.BlockConnected = g_signals.m_internals->BlockConnected.connect( - std::bind(&CValidationInterface::BlockConnected, pwalletIn, - std::placeholders::_1, std::placeholders::_2, - std::placeholders::_3)); - conns.BlockDisconnected = g_signals.m_internals->BlockDisconnected.connect( - std::bind(&CValidationInterface::BlockDisconnected, pwalletIn, - std::placeholders::_1)); - conns.TransactionRemovedFromMempool = - g_signals.m_internals->TransactionRemovedFromMempool.connect( - std::bind(&CValidationInterface::TransactionRemovedFromMempool, - pwalletIn, std::placeholders::_1)); - conns.ChainStateFlushed = g_signals.m_internals->ChainStateFlushed.connect( - std::bind(&CValidationInterface::ChainStateFlushed, pwalletIn, - std::placeholders::_1)); - conns.BlockChecked = g_signals.m_internals->BlockChecked.connect( - std::bind(&CValidationInterface::BlockChecked, pwalletIn, - std::placeholders::_1, std::placeholders::_2)); - conns.NewPoWValidBlock = g_signals.m_internals->NewPoWValidBlock.connect( - std::bind(&CValidationInterface::NewPoWValidBlock, pwalletIn, - std::placeholders::_1, std::placeholders::_2)); + g_signals.m_internals->Register(std::move(pwalletIn)); } void RegisterValidationInterface(CValidationInterface *callbacks) { // Create a shared_ptr with a no-op deleter - CValidationInterface lifecycle // is managed by the caller. RegisterSharedValidationInterface( {callbacks, [](CValidationInterface *) {}}); } void UnregisterSharedValidationInterface( std::shared_ptr callbacks) { UnregisterValidationInterface(callbacks.get()); } void UnregisterValidationInterface(CValidationInterface *pwalletIn) { if (g_signals.m_internals) { - g_signals.m_internals->m_connMainSignals.erase(pwalletIn); + g_signals.m_internals->Unregister(pwalletIn); } } void UnregisterAllValidationInterfaces() { if (!g_signals.m_internals) { return; } - g_signals.m_internals->m_connMainSignals.clear(); + g_signals.m_internals->Clear(); } void CallFunctionInValidationInterfaceQueue(std::function func) { g_signals.m_internals->m_schedulerClient.AddToProcessQueue(std::move(func)); } void SyncWithValidationInterfaceQueue() { AssertLockNotHeld(cs_main); // Block until the validation queue drains std::promise promise; CallFunctionInValidationInterfaceQueue([&promise] { promise.set_value(); }); promise.get_future().wait(); } // Use a macro instead of a function for conditional logging to prevent // evaluating arguments when logging is not enabled. // // NOTE: The lambda captures all local variables by value. #define ENQUEUE_AND_LOG_EVENT(event, fmt, name, ...) \ do { \ auto local_name = (name); \ LOG_EVENT("Enqueuing " fmt, local_name, __VA_ARGS__); \ m_internals->m_schedulerClient.AddToProcessQueue([=] { \ LOG_EVENT(fmt, local_name, __VA_ARGS__); \ event(); \ }); \ } while (0) #define LOG_EVENT(fmt, ...) LogPrint(BCLog::VALIDATION, fmt "\n", __VA_ARGS__) void CMainSignals::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) { // Dependencies exist that require UpdatedBlockTip events to be delivered in // the order in which the chain actually updates. One way to ensure this is // for the caller to invoke this signal in the same critical section where // the chain is updated auto event = [pindexNew, pindexFork, fInitialDownload, this] { - m_internals->UpdatedBlockTip(pindexNew, pindexFork, fInitialDownload); + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.UpdatedBlockTip(pindexNew, pindexFork, fInitialDownload); + }); }; ENQUEUE_AND_LOG_EVENT( event, "%s: new block hash=%s fork block hash=%s (in IBD=%s)", __func__, pindexNew->GetBlockHash().ToString(), pindexFork ? pindexFork->GetBlockHash().ToString() : "null", fInitialDownload); } void CMainSignals::TransactionAddedToMempool(const CTransactionRef &ptx) { - auto event = [ptx, this] { m_internals->TransactionAddedToMempool(ptx); }; + auto event = [ptx, this] { + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.TransactionAddedToMempool(ptx); + }); + }; ENQUEUE_AND_LOG_EVENT(event, "%s: txid=%s", __func__, ptx->GetHash().ToString()); } void CMainSignals::TransactionRemovedFromMempool(const CTransactionRef &ptx) { auto event = [ptx, this] { - m_internals->TransactionRemovedFromMempool(ptx); + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.TransactionRemovedFromMempool(ptx); + }); }; ENQUEUE_AND_LOG_EVENT(event, "%s: txid=%s", __func__, ptx->GetHash().ToString()); } void CMainSignals::BlockConnected( const std::shared_ptr &pblock, const CBlockIndex *pindex, const std::shared_ptr> &pvtxConflicted) { auto event = [pblock, pindex, pvtxConflicted, this] { - m_internals->BlockConnected(pblock, pindex, *pvtxConflicted); + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.BlockConnected(pblock, pindex, *pvtxConflicted); + }); }; ENQUEUE_AND_LOG_EVENT(event, "%s: block hash=%s block height=%d", __func__, pblock->GetHash().ToString(), pindex->nHeight); } void CMainSignals::BlockDisconnected( const std::shared_ptr &pblock) { // TODO: This function was refactored as part of an out-of-order backport // of https://github.com/bitcoin/bitcoin/pull/16688 - auto event = [pblock, this] { m_internals->BlockDisconnected(pblock); }; + auto event = [pblock, this] { + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.BlockDisconnected(pblock); + }); + }; ENQUEUE_AND_LOG_EVENT(event, "%s: block hash=%s", __func__, pblock->GetHash().ToString()); } void CMainSignals::ChainStateFlushed(const CBlockLocator &locator) { - auto event = [locator, this] { m_internals->ChainStateFlushed(locator); }; + auto event = [locator, this] { + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.ChainStateFlushed(locator); + }); + }; ENQUEUE_AND_LOG_EVENT(event, "%s: block hash=%s", __func__, locator.IsNull() ? "null" : locator.vHave.front().ToString()); } void CMainSignals::BlockChecked(const CBlock &block, const CValidationState &state) { LOG_EVENT("%s: block hash=%s state=%s", __func__, block.GetHash().ToString(), FormatStateMessage(state)); - m_internals->BlockChecked(block, state); + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.BlockChecked(block, state); + }); } void CMainSignals::NewPoWValidBlock( const CBlockIndex *pindex, const std::shared_ptr &block) { LOG_EVENT("%s: block hash=%s", __func__, block->GetHash().ToString()); - m_internals->NewPoWValidBlock(pindex, block); + m_internals->Iterate([&](CValidationInterface &callbacks) { + callbacks.NewPoWValidBlock(pindex, block); + }); } diff --git a/src/validationinterface.h b/src/validationinterface.h index f667edf79..84ac5b2d8 100644 --- a/src/validationinterface.h +++ b/src/validationinterface.h @@ -1,217 +1,214 @@ // Copyright (c) 2009-2010 Satoshi Nakamoto // Copyright (c) 2009-2016 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_VALIDATIONINTERFACE_H #define BITCOIN_VALIDATIONINTERFACE_H #include // CTransaction(Ref) #include #include #include extern RecursiveMutex cs_main; class CBlock; class CBlockIndex; struct CBlockLocator; class CBlockIndex; class CConnman; class CReserveScript; class CValidationInterface; class CValidationState; class uint256; class CScheduler; // These functions dispatch to one or all registered wallets /** Register a wallet to receive updates from core */ void RegisterValidationInterface(CValidationInterface *pwalletIn); /** Unregister a wallet from core */ void UnregisterValidationInterface(CValidationInterface *pwalletIn); /** Unregister all wallets from core */ void UnregisterAllValidationInterfaces(); // Alternate registration functions that release a shared_ptr after the last // notification is sent. These are useful for race-free cleanup, since // unregistration is nonblocking and can return before the last notification is // processed. void RegisterSharedValidationInterface( std::shared_ptr callbacks); void UnregisterSharedValidationInterface( std::shared_ptr callbacks); /** * Pushes a function to callback onto the notification queue, guaranteeing any * callbacks generated prior to now are finished when the function is called. * * Be very careful blocking on func to be called if any locks are held - * validation interface clients may not be able to make progress as they often * wait for things like cs_main, so blocking until func is called with cs_main * will result in a deadlock (that DEBUG_LOCKORDER will miss). */ void CallFunctionInValidationInterfaceQueue(std::function func); /** * This is a synonym for the following, which asserts certain locks are not * held: * std::promise promise; * CallFunctionInValidationInterfaceQueue([&promise] { * promise.set_value(); * }); * promise.get_future().wait(); */ void SyncWithValidationInterfaceQueue() LOCKS_EXCLUDED(cs_main); /** * Implement this to subscribe to events generated in validation * * Each CValidationInterface() subscriber will receive event callbacks * in the order in which the events were generated by validation. * Furthermore, each ValidationInterface() subscriber may assume that * callbacks effectively run in a single thread with single-threaded * memory consistency. That is, for a given ValidationInterface() * instantiation, each callback will complete before the next one is * invoked. This means, for example when a block is connected that the * UpdatedBlockTip() callback may depend on an operation performed in * the BlockConnected() callback without worrying about explicit * synchronization. No ordering should be assumed across * ValidationInterface() subscribers. */ class CValidationInterface { protected: /** * Protected destructor so that instances can only be deleted by derived * classes. If that restriction is no longer desired, this should be made * public and virtual. */ ~CValidationInterface() = default; /** * Notifies listeners when the block chain tip advances. * * When multiple blocks are connected at once, UpdatedBlockTip will be * called on the final tip but may not be called on every intermediate tip. * If the latter behavior is desired, subscribe to BlockConnected() instead. * * Called on a background thread. */ virtual void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) {} /** * Notifies listeners of a transaction having been added to mempool. * * Called on a background thread. */ virtual void TransactionAddedToMempool(const CTransactionRef &ptxn) {} /** * Notifies listeners of a transaction leaving mempool. * * This only fires for transactions which leave mempool because of expiry, * size limiting, reorg (changes in lock times/coinbase maturity), or * replacement. This does not include any transactions which are included * in BlockConnectedDisconnected either in block->vtx or in txnConflicted. * * Called on a background thread. */ virtual void TransactionRemovedFromMempool(const CTransactionRef &ptx) {} /** * Notifies listeners of a block being connected. * Provides a vector of transactions evicted from the mempool as a result. * * Called on a background thread. */ virtual void BlockConnected(const std::shared_ptr &block, const CBlockIndex *pindex, const std::vector &txnConflicted) {} /** * Notifies listeners of a block being disconnected * * Called on a background thread. */ virtual void BlockDisconnected(const std::shared_ptr &block) { } /** * Notifies listeners of the new active block chain on-disk. * * Prior to this callback, any updates are not guaranteed to persist on disk * (ie clients need to handle shutdown/restart safety by being able to * understand when some updates were lost due to unclean shutdown). * * When this callback is invoked, the validation changes done by any prior * callback are guaranteed to exist on disk and survive a restart, including * an unclean shutdown. * * Provides a locator describing the best chain, which is likely useful for * storing current state on disk in client DBs. * * Called on a background thread. */ virtual void ChainStateFlushed(const CBlockLocator &locator) {} /** * Notifies listeners of a block validation result. * If the provided CValidationState IsValid, the provided block * is guaranteed to be the current best block at the time the * callback was generated (not necessarily now) */ virtual void BlockChecked(const CBlock &, const CValidationState &) {} /** * Notifies listeners that a block which builds directly on our current tip * has been received and connected to the headers tree, though not validated * yet. */ virtual void NewPoWValidBlock(const CBlockIndex *pindex, const std::shared_ptr &block){}; - friend void ::RegisterSharedValidationInterface( - std::shared_ptr); - friend void ::UnregisterValidationInterface(CValidationInterface *); - friend void ::UnregisterAllValidationInterfaces(); + friend class CMainSignals; }; struct MainSignalsInstance; class CMainSignals { private: std::unique_ptr m_internals; friend void ::RegisterSharedValidationInterface( std::shared_ptr); friend void ::UnregisterValidationInterface(CValidationInterface *); friend void ::UnregisterAllValidationInterfaces(); friend void ::CallFunctionInValidationInterfaceQueue( std::function func); public: /** * Register a CScheduler to give callbacks which should run in the * background (may only be called once) */ void RegisterBackgroundSignalScheduler(CScheduler &scheduler); /** * Unregister a CScheduler to give callbacks which should run in the * background - these callbacks will now be dropped! */ void UnregisterBackgroundSignalScheduler(); /** Call any remaining callbacks on the calling thread */ void FlushBackgroundCallbacks(); size_t CallbacksPending(); void UpdatedBlockTip(const CBlockIndex *, const CBlockIndex *, bool fInitialDownload); void TransactionAddedToMempool(const CTransactionRef &); void TransactionRemovedFromMempool(const CTransactionRef &); void BlockConnected(const std::shared_ptr &, const CBlockIndex *pindex, const std::shared_ptr> &); void BlockDisconnected(const std::shared_ptr &); void ChainStateFlushed(const CBlockLocator &); void BlockChecked(const CBlock &, const CValidationState &); void NewPoWValidBlock(const CBlockIndex *, const std::shared_ptr &); }; CMainSignals &GetMainSignals(); #endif // BITCOIN_VALIDATIONINTERFACE_H