Merge bitcoin#30111: locks: introduce mutex for tx download, flush rejection filters once per tip change

c85acce [refactor] delete EraseTxNoLock, just use EraseTx (glozow)
6ff8406 remove obsoleted TxOrphanage::m_mutex (glozow)
61745c7 lock m_recent_confirmed_transactions using m_tx_download_mutex (glozow)
723ea0f remove obsoleted hashRecentRejectsChainTip (glozow)
18a4355 update recent_rejects filters on ActiveTipChange (glozow)
36f170d add ValidationInterface::ActiveTipChange (glozow)
3eb1307 guard TxRequest and rejection caches with new mutex (glozow)

Pull request description:

  See bitcoin#27463 for full project tracking.

  This contains the first few commits of bitcoin#30110, which require some thinking about thread safety in review.
  - Introduce a new `m_tx_download_mutex` which guards the transaction download data structures including `m_txrequest`, the rolling bloom filters, and `m_orphanage`. Later this should become the mutex guarding `TxDownloadManager`.
    - `m_txrequest` doesn't need to be guarded using `cs_main` anymore
    - `m_recent_confirmed_transactions` doesn't need its own lock anymore
    - `m_orphanage` doesn't need its own lock anymore
  - Adds a new `ValidationInterface` event, `ActiveTipChanged`, which is a synchronous callback whenever the tip of the active chainstate changes.
  - Flush `m_recent_rejects` and `m_recent_rejects_reconsiderable` on `ActiveTipChanged` just once instead of checking the tip every time `AlreadyHaveTx` is called. This should speed up calls to that function (no longer comparing a block hash each time) and removes the need to lock `cs_main` every time it is called.

  Motivation:
  - These data structures need synchronization. While we are holding `m_tx_download_mutex`, these should hold:
    - a tx hash in `m_txrequest` is not also in `m_orphanage`
    - a tx hash in `m_txrequest` is not also in `m_recent_rejects` or `m_recent_confirmed_transactions`
    - In the future, orphan resolution tracking should also be synchronized. If a tx has an entry in the orphan resolution tracker, it is also in `m_orphanage`, and not in `m_txrequest`, etc.
  - Currently, `cs_main` is used to e.g. sync accesses to `m_txrequest`. We should not broaden the scope of things it locks.
  - Currently, we need to know the current chainstate every time we call `AlreadyHaveTx` so we can decide whether we should update it. Every call compares the current tip hash with `hashRecentRejectsChainTip`. It is more efficient to have a validation interface callback that updates the rejection filters whenever the chain tip changes.

ACKs for top commit:
  instagibbs:
    reACK c85acce
  dergoegge:
    Code review ACK c85acce
  theStack:
    Light code-review ACK c85acce
  hebasto:
    ACK c85acce, I have reviewed the code and it looks OK.

Tree-SHA512: c3bd524b5de1cafc9a10770dadb484cc479d6d4c687d80dd0f176d339fd95f73b85cb44cb3b6b464d38a52e20feda00aa2a1da5a73339e31831687e4bd0aa0c5

Author: fanquake

src/net_processing.cpp

@@ -21,15 +21,13 @@ BOOST_FIXTURE_TEST_SUITE(orphanage_tests, TestingSetup)
 class TxOrphanageTest : public TxOrphanage
 {
 public:
-    inline size_t CountOrphans() const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+    inline size_t CountOrphans() const
     {
-        LOCK(m_mutex);
         return m_orphans.size();
     }
 
-    CTransactionRef RandomOrphan() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+    CTransactionRef RandomOrphan()
     {
-        LOCK(m_mutex);
         std::map<Wtxid, OrphanTx>::iterator it;
         it = m_orphans.lower_bound(Wtxid::FromUint256(InsecureRand256()));
         if (it == m_orphans.end())

src/test/orphanage_tests.cpp

@@ -20,8 +20,6 @@ static constexpr auto ORPHAN_TX_EXPIRE_INTERVAL{5min};
 
 bool TxOrphanage::AddTx(const CTransactionRef& tx, NodeId peer)
 {
-    LOCK(m_mutex);
-
     const Txid& hash = tx->GetHash();
     const Wtxid& wtxid = tx->GetWitnessHash();
     if (m_orphans.count(wtxid))
@@ -55,13 +53,6 @@ bool TxOrphanage::AddTx(const CTransactionRef& tx, NodeId peer)
 
 int TxOrphanage::EraseTx(const Wtxid& wtxid)
 {
-    LOCK(m_mutex);
-    return EraseTxNoLock(wtxid);
-}
-
-int TxOrphanage::EraseTxNoLock(const Wtxid& wtxid)
-{
-    AssertLockHeld(m_mutex);
     std::map<Wtxid, OrphanTx>::iterator it = m_orphans.find(wtxid);
     if (it == m_orphans.end())
         return 0;
@@ -97,8 +88,6 @@ int TxOrphanage::EraseTxNoLock(const Wtxid& wtxid)
 
 void TxOrphanage::EraseForPeer(NodeId peer)
 {
-    LOCK(m_mutex);
-
     m_peer_work_set.erase(peer);
 
     int nErased = 0;
@@ -108,16 +97,14 @@ void TxOrphanage::EraseForPeer(NodeId peer)
         // increment to avoid iterator becoming invalid after erasure
         const auto& [wtxid, orphan] = *iter++;
         if (orphan.fromPeer == peer) {
-            nErased += EraseTxNoLock(wtxid);
+            nErased += EraseTx(wtxid);
         }
     }
     if (nErased > 0) LogPrint(BCLog::TXPACKAGES, "Erased %d orphan transaction(s) from peer=%d\n", nErased, peer);
 }
 
 void TxOrphanage::LimitOrphans(unsigned int max_orphans, FastRandomContext& rng)
 {
-    LOCK(m_mutex);
-
     unsigned int nEvicted = 0;
     auto nNow{Now<NodeSeconds>()};
     if (m_next_sweep <= nNow) {
@@ -129,7 +116,7 @@ void TxOrphanage::LimitOrphans(unsigned int max_orphans, FastRandomContext& rng)
         {
             std::map<Wtxid, OrphanTx>::iterator maybeErase = iter++;
             if (maybeErase->second.nTimeExpire <= nNow) {
-                nErased += EraseTxNoLock(maybeErase->second.tx->GetWitnessHash());
+                nErased += EraseTx(maybeErase->second.tx->GetWitnessHash());
             } else {
                 nMinExpTime = std::min(maybeErase->second.nTimeExpire, nMinExpTime);
             }
@@ -142,17 +129,14 @@ void TxOrphanage::LimitOrphans(unsigned int max_orphans, FastRandomContext& rng)
     {
         // Evict a random orphan:
         size_t randompos = rng.randrange(m_orphan_list.size());
-        EraseTxNoLock(m_orphan_list[randompos]->second.tx->GetWitnessHash());
+        EraseTx(m_orphan_list[randompos]->second.tx->GetWitnessHash());
         ++nEvicted;
     }
     if (nEvicted > 0) LogPrint(BCLog::TXPACKAGES, "orphanage overflow, removed %u tx\n", nEvicted);
 }
 
 void TxOrphanage::AddChildrenToWorkSet(const CTransaction& tx)
 {
-    LOCK(m_mutex);
-
-
     for (unsigned int i = 0; i < tx.vout.size(); i++) {
         const auto it_by_prev = m_outpoint_to_orphan_it.find(COutPoint(tx.GetHash(), i));
         if (it_by_prev != m_outpoint_to_orphan_it.end()) {
@@ -171,14 +155,11 @@ void TxOrphanage::AddChildrenToWorkSet(const CTransaction& tx)
 
 bool TxOrphanage::HaveTx(const Wtxid& wtxid) const
 {
-    LOCK(m_mutex);
     return m_orphans.count(wtxid);
 }
 
 CTransactionRef TxOrphanage::GetTxToReconsider(NodeId peer)
 {
-    LOCK(m_mutex);
-
     auto work_set_it = m_peer_work_set.find(peer);
     if (work_set_it != m_peer_work_set.end()) {
         auto& work_set = work_set_it->second;
@@ -197,8 +178,6 @@ CTransactionRef TxOrphanage::GetTxToReconsider(NodeId peer)
 
 bool TxOrphanage::HaveTxToReconsider(NodeId peer)
 {
-    LOCK(m_mutex);
-
     auto work_set_it = m_peer_work_set.find(peer);
     if (work_set_it != m_peer_work_set.end()) {
         auto& work_set = work_set_it->second;
@@ -209,8 +188,6 @@ bool TxOrphanage::HaveTxToReconsider(NodeId peer)
 
 void TxOrphanage::EraseForBlock(const CBlock& block)
 {
-    LOCK(m_mutex);
-
     std::vector<Wtxid> vOrphanErase;
 
     for (const CTransactionRef& ptx : block.vtx) {
@@ -231,16 +208,14 @@ void TxOrphanage::EraseForBlock(const CBlock& block)
     if (vOrphanErase.size()) {
         int nErased = 0;
         for (const auto& orphanHash : vOrphanErase) {
-            nErased += EraseTxNoLock(orphanHash);
+            nErased += EraseTx(orphanHash);
         }
         LogPrint(BCLog::TXPACKAGES, "Erased %d orphan transaction(s) included or conflicted by block\n", nErased);
     }
 }
 
 std::vector<CTransactionRef> TxOrphanage::GetChildrenFromSamePeer(const CTransactionRef& parent, NodeId nodeid) const
 {
-    LOCK(m_mutex);
-
     // First construct a vector of iterators to ensure we do not return duplicates of the same tx
     // and so we can sort by nTimeExpire.
     std::vector<OrphanMap::iterator> iters;
@@ -281,8 +256,6 @@ std::vector<CTransactionRef> TxOrphanage::GetChildrenFromSamePeer(const CTransac
 
 std::vector<std::pair<CTransactionRef, NodeId>> TxOrphanage::GetChildrenFromDifferentPeer(const CTransactionRef& parent, NodeId nodeid) const
 {
-    LOCK(m_mutex);
-
     // First construct vector of iterators to ensure we do not return duplicates of the same tx.
     std::vector<OrphanMap::iterator> iters;
 

src/txorphanage.cpp

@@ -22,55 +22,51 @@
 class TxOrphanage {
 public:
     /** Add a new orphan transaction */
-    bool AddTx(const CTransactionRef& tx, NodeId peer) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    bool AddTx(const CTransactionRef& tx, NodeId peer);
 
     /** Check if we already have an orphan transaction (by wtxid only) */
-    bool HaveTx(const Wtxid& wtxid) const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    bool HaveTx(const Wtxid& wtxid) const;
 
     /** Extract a transaction from a peer's work set
      *  Returns nullptr if there are no transactions to work on.
      *  Otherwise returns the transaction reference, and removes
      *  it from the work set.
      */
-    CTransactionRef GetTxToReconsider(NodeId peer) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    CTransactionRef GetTxToReconsider(NodeId peer);
 
     /** Erase an orphan by wtxid */
-    int EraseTx(const Wtxid& wtxid) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    int EraseTx(const Wtxid& wtxid);
 
     /** Erase all orphans announced by a peer (eg, after that peer disconnects) */
-    void EraseForPeer(NodeId peer) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    void EraseForPeer(NodeId peer);
 
     /** Erase all orphans included in or invalidated by a new block */
-    void EraseForBlock(const CBlock& block) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    void EraseForBlock(const CBlock& block);
 
     /** Limit the orphanage to the given maximum */
-    void LimitOrphans(unsigned int max_orphans, FastRandomContext& rng) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    void LimitOrphans(unsigned int max_orphans, FastRandomContext& rng);
 
     /** Add any orphans that list a particular tx as a parent into the from peer's work set */
-    void AddChildrenToWorkSet(const CTransaction& tx) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);;
+    void AddChildrenToWorkSet(const CTransaction& tx);
 
     /** Does this peer have any work to do? */
-    bool HaveTxToReconsider(NodeId peer) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);;
+    bool HaveTxToReconsider(NodeId peer);
 
     /** Get all children that spend from this tx and were received from nodeid. Sorted from most
      * recent to least recent. */
-    std::vector<CTransactionRef> GetChildrenFromSamePeer(const CTransactionRef& parent, NodeId nodeid) const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    std::vector<CTransactionRef> GetChildrenFromSamePeer(const CTransactionRef& parent, NodeId nodeid) const;
 
     /** Get all children that spend from this tx but were not received from nodeid. Also return
      * which peer provided each tx. */
-    std::vector<std::pair<CTransactionRef, NodeId>> GetChildrenFromDifferentPeer(const CTransactionRef& parent, NodeId nodeid) const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
+    std::vector<std::pair<CTransactionRef, NodeId>> GetChildrenFromDifferentPeer(const CTransactionRef& parent, NodeId nodeid) const;
 
     /** Return how many entries exist in the orphange */
-    size_t Size() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
+    size_t Size()
     {
-        LOCK(m_mutex);
         return m_orphans.size();
     }
 
 protected:
-    /** Guards orphan transactions */
-    mutable Mutex m_mutex;
-
     struct OrphanTx {
         CTransactionRef tx;
         NodeId fromPeer;
@@ -80,10 +76,10 @@ class TxOrphanage {
 
     /** Map from wtxid to orphan transaction record. Limited by
      *  -maxorphantx/DEFAULT_MAX_ORPHAN_TRANSACTIONS */
-    std::map<Wtxid, OrphanTx> m_orphans GUARDED_BY(m_mutex);
+    std::map<Wtxid, OrphanTx> m_orphans;
 
     /** Which peer provided the orphans that need to be reconsidered */
-    std::map<NodeId, std::set<Wtxid>> m_peer_work_set GUARDED_BY(m_mutex);
+    std::map<NodeId, std::set<Wtxid>> m_peer_work_set;
 
     using OrphanMap = decltype(m_orphans);
 
@@ -98,16 +94,13 @@ class TxOrphanage {
 
     /** Index from the parents' COutPoint into the m_orphans. Used
      *  to remove orphan transactions from the m_orphans */
-    std::map<COutPoint, std::set<OrphanMap::iterator, IteratorComparator>> m_outpoint_to_orphan_it GUARDED_BY(m_mutex);
+    std::map<COutPoint, std::set<OrphanMap::iterator, IteratorComparator>> m_outpoint_to_orphan_it;
 
     /** Orphan transactions in vector for quick random eviction */
-    std::vector<OrphanMap::iterator> m_orphan_list GUARDED_BY(m_mutex);
-
-    /** Erase an orphan by wtxid */
-    int EraseTxNoLock(const Wtxid& wtxid) EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
+    std::vector<OrphanMap::iterator> m_orphan_list;
 
     /** Timestamp for the next scheduled sweep of expired orphans */
-    NodeSeconds m_next_sweep GUARDED_BY(m_mutex){0s};
+    NodeSeconds m_next_sweep{0s};
 };
 
 #endif // BITCOIN_TXORPHANAGE_H

src/txorphanage.h

@@ -3473,6 +3473,7 @@ bool Chainstate::ActivateBestChain(BlockValidationState& state, std::shared_ptr<
 
         {
             LOCK(cs_main);
+            {
             // Lock transaction pool for at least as long as it takes for connectTrace to be consumed
             LOCK(MempoolMutex());
             const bool was_in_ibd = m_chainman.IsInitialBlockDownload();
@@ -3549,7 +3550,12 @@ bool Chainstate::ActivateBestChain(BlockValidationState& state, std::shared_ptr<
                     break;
                 }
             }
-        }
+            } // release MempoolMutex
+            // Notify external listeners about the new tip, even if pindexFork == pindexNewTip.
+            if (m_chainman.m_options.signals && this == &m_chainman.ActiveChainstate()) {
+                m_chainman.m_options.signals->ActiveTipChange(pindexNewTip, m_chainman.IsInitialBlockDownload());
+            }
+        } // release cs_main
         // When we reach this point, we switched to a new tip (stored in pindexNewTip).
 
         if (exited_ibd) {
@@ -3768,6 +3774,12 @@ bool Chainstate::InvalidateBlock(BlockValidationState& state, CBlockIndex* pinde
         // distinguish user-initiated invalidateblock changes from other
         // changes.
         (void)m_chainman.GetNotifications().blockTip(GetSynchronizationState(m_chainman.IsInitialBlockDownload(), m_chainman.m_blockman.m_blockfiles_indexed), *to_mark_failed->pprev);
+
+        // Fire ActiveTipChange now for the current chain tip to make sure clients are notified.
+        // ActivateBestChain may call this as well, but not necessarily.
+        if (m_chainman.m_options.signals) {
+            m_chainman.m_options.signals->ActiveTipChange(m_chain.Tip(), m_chainman.IsInitialBlockDownload());
+        }
     }
     return true;
 }

src/validation.cpp

@@ -183,6 +183,12 @@ void ValidationSignals::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlo
                           fInitialDownload);
 }
 
+void ValidationSignals::ActiveTipChange(const CBlockIndex *new_tip, bool is_ibd)
+{
+    LOG_EVENT("%s: new block hash=%s block height=%d", __func__, new_tip->GetBlockHash().ToString(), new_tip->nHeight);
+    m_internals->Iterate([&](CValidationInterface& callbacks) { callbacks.ActiveTipChange(new_tip, is_ibd); });
+}
+
 void ValidationSignals::TransactionAddedToMempool(const NewMempoolTransactionInfo& tx, uint64_t mempool_sequence)
 {
     auto event = [tx, mempool_sequence, this] {

src/validationinterface.cpp

@@ -61,6 +61,10 @@ class CValidationInterface {
      * Called on a background thread. Only called for the active chainstate.
      */
     virtual void UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) {}
+    /**
+     * Notifies listeners any time the block chain tip changes, synchronously.
+     */
+    virtual void ActiveTipChange(const CBlockIndex* new_tip, bool is_ibd) {};
     /**
      * Notifies listeners of a transaction having been added to mempool.
      *
@@ -214,6 +218,7 @@ class ValidationSignals {
     void SyncWithValidationInterfaceQueue() LOCKS_EXCLUDED(cs_main);
 
     void UpdatedBlockTip(const CBlockIndex *, const CBlockIndex *, bool fInitialDownload);
+    void ActiveTipChange(const CBlockIndex*, bool);
     void TransactionAddedToMempool(const NewMempoolTransactionInfo&, uint64_t mempool_sequence);
     void TransactionRemovedFromMempool(const CTransactionRef&, MemPoolRemovalReason, uint64_t mempool_sequence);
     void MempoolTransactionsRemovedForBlock(const std::vector<RemovedMempoolTransactionInfo>&, unsigned int nBlockHeight);