Merge bitcoin/bitcoin#30507: m_tx_download_mutex followups

7c29e55 m_tx_download_mutex followups (glozow)
e543c65 release m_tx_download_mutex before MakeAndPushMessage GETDATA (glozow)
bce5f37 [refactor] change ActiveTipChange to use CBlockIndex ref instead of ptr (glozow)
7cc5ac5 [doc] TxOrphanage is no longer thread-safe (glozow)
6f49548 [refactor] combine block vtx loops in BlockConnected (glozow)

Pull request description:

  Followup to #30111. Includes suggestions:
  - bitcoin/bitcoin#30111 (comment)
  - bitcoin/bitcoin#30111 (comment)
  - bitcoin/bitcoin#30111 (comment)
  - bitcoin/bitcoin#30111 (comment)
  - bitcoin/bitcoin#30111 (comment)

ACKs for top commit:
  instagibbs:
    reACK bitcoin/bitcoin@7c29e55
  theStack:
    re-ACK 7c29e55
  dergoegge:
    reACK 7c29e55

Tree-SHA512: 79a9002d74739367789bbc64bb1d431f4d43a25a7934231e55814c2cb6981c15ef2d8465544ae2a4fbd734d9bed6cc41b37a923938a88cb8fea139523c1e98da

Author: fanquake

src/net_processing.cpp

@@ -489,7 +489,7 @@ class PeerManagerImpl final : public PeerManager
                     CTxMemPool& pool, node::Warnings& warnings, Options opts);
 
     /** Overridden from CValidationInterface. */
-    void ActiveTipChange(const CBlockIndex* new_tip, bool) override
+    void ActiveTipChange(const CBlockIndex& new_tip, bool) override
         EXCLUSIVE_LOCKS_REQUIRED(!m_tx_download_mutex);
     void BlockConnected(ChainstateRole role, const std::shared_ptr<const CBlock>& pblock, const CBlockIndex* pindexConnected) override
         EXCLUSIVE_LOCKS_REQUIRED(!m_tx_download_mutex);
@@ -780,10 +780,8 @@ class PeerManagerImpl final : public PeerManager
      * - A txhash (txid or wtxid) in m_txrequest is not also in m_recent_rejects_reconsiderable.
      * - A txhash (txid or wtxid) in m_txrequest is not also in m_recent_confirmed_transactions.
      * - Each data structure's limits hold (m_orphanage max size, m_txrequest per-peer limits, etc).
-     *
-     * m_tx_download_mutex must be acquired before mempool.cs
      */
-    Mutex m_tx_download_mutex;
+    Mutex m_tx_download_mutex ACQUIRED_BEFORE(m_mempool.cs);
     TxRequestTracker m_txrequest GUARDED_BY(m_tx_download_mutex);
     std::unique_ptr<TxReconciliationTracker> m_txreconciliation;
 
@@ -2070,8 +2068,10 @@ void PeerManagerImpl::StartScheduledTasks(CScheduler& scheduler)
     scheduler.scheduleFromNow([&] { ReattemptInitialBroadcast(scheduler); }, delta);
 }
 
-void PeerManagerImpl::ActiveTipChange(const CBlockIndex* new_tip, bool is_ibd)
+void PeerManagerImpl::ActiveTipChange(const CBlockIndex& new_tip, bool is_ibd)
 {
+    // Ensure mempool mutex was released, otherwise deadlock may occur if another thread holding
+    // m_tx_download_mutex waits on the mempool mutex.
     AssertLockNotHeld(m_mempool.cs);
     AssertLockNotHeld(m_tx_download_mutex);
 
@@ -2123,8 +2123,6 @@ void PeerManagerImpl::BlockConnected(
         if (ptx->HasWitness()) {
             m_recent_confirmed_transactions.insert(ptx->GetWitnessHash().ToUint256());
         }
-    }
-    for (const auto& ptx : pblock->vtx) {
         m_txrequest.ForgetTxHash(ptx->GetHash());
         m_txrequest.ForgetTxHash(ptx->GetWitnessHash());
     }
@@ -5336,6 +5334,7 @@ bool PeerManagerImpl::MaybeDiscourageAndDisconnect(CNode& pnode, Peer& peer)
 
 bool PeerManagerImpl::ProcessMessages(CNode* pfrom, std::atomic<bool>& interruptMsgProc)
 {
+    AssertLockNotHeld(m_tx_download_mutex);
     AssertLockHeld(g_msgproc_mutex);
 
     PeerRef peer = GetPeerRef(pfrom->GetId());
@@ -5827,6 +5826,7 @@ bool PeerManagerImpl::SetupAddressRelay(const CNode& node, Peer& peer)
 
 bool PeerManagerImpl::SendMessages(CNode* pto)
 {
+    AssertLockNotHeld(m_tx_download_mutex);
     AssertLockHeld(g_msgproc_mutex);
 
     PeerRef peer = GetPeerRef(pto->GetId());
@@ -6297,32 +6297,33 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
         //
         // Message: getdata (transactions)
         //
-        LOCK(m_tx_download_mutex);
-        std::vector<std::pair<NodeId, GenTxid>> expired;
-        auto requestable = m_txrequest.GetRequestable(pto->GetId(), current_time, &expired);
-        for (const auto& entry : expired) {
-            LogPrint(BCLog::NET, "timeout of inflight %s %s from peer=%d\n", entry.second.IsWtxid() ? "wtx" : "tx",
-                entry.second.GetHash().ToString(), entry.first);
-        }
-        for (const GenTxid& gtxid : requestable) {
-            // Exclude m_recent_rejects_reconsiderable: we may be requesting a missing parent
-            // that was previously rejected for being too low feerate.
-            if (!AlreadyHaveTx(gtxid, /*include_reconsiderable=*/false)) {
-                LogPrint(BCLog::NET, "Requesting %s %s peer=%d\n", gtxid.IsWtxid() ? "wtx" : "tx",
-                    gtxid.GetHash().ToString(), pto->GetId());
-                vGetData.emplace_back(gtxid.IsWtxid() ? MSG_WTX : (MSG_TX | GetFetchFlags(*peer)), gtxid.GetHash());
-                if (vGetData.size() >= MAX_GETDATA_SZ) {
-                    MakeAndPushMessage(*pto, NetMsgType::GETDATA, vGetData);
-                    vGetData.clear();
+        {
+            LOCK(m_tx_download_mutex);
+            std::vector<std::pair<NodeId, GenTxid>> expired;
+            auto requestable = m_txrequest.GetRequestable(pto->GetId(), current_time, &expired);
+            for (const auto& entry : expired) {
+                LogPrint(BCLog::NET, "timeout of inflight %s %s from peer=%d\n", entry.second.IsWtxid() ? "wtx" : "tx",
+                    entry.second.GetHash().ToString(), entry.first);
+            }
+            for (const GenTxid& gtxid : requestable) {
+                // Exclude m_recent_rejects_reconsiderable: we may be requesting a missing parent
+                // that was previously rejected for being too low feerate.
+                if (!AlreadyHaveTx(gtxid, /*include_reconsiderable=*/false)) {
+                    LogPrint(BCLog::NET, "Requesting %s %s peer=%d\n", gtxid.IsWtxid() ? "wtx" : "tx",
+                        gtxid.GetHash().ToString(), pto->GetId());
+                    vGetData.emplace_back(gtxid.IsWtxid() ? MSG_WTX : (MSG_TX | GetFetchFlags(*peer)), gtxid.GetHash());
+                    if (vGetData.size() >= MAX_GETDATA_SZ) {
+                        MakeAndPushMessage(*pto, NetMsgType::GETDATA, vGetData);
+                        vGetData.clear();
+                    }
+                    m_txrequest.RequestedTx(pto->GetId(), gtxid.GetHash(), current_time + GETDATA_TX_INTERVAL);
+                } else {
+                    // We have already seen this transaction, no need to download. This is just a belt-and-suspenders, as
+                    // this should already be called whenever a transaction becomes AlreadyHaveTx().
+                    m_txrequest.ForgetTxHash(gtxid.GetHash());
                 }
-                m_txrequest.RequestedTx(pto->GetId(), gtxid.GetHash(), current_time + GETDATA_TX_INTERVAL);
-            } else {
-                // We have already seen this transaction, no need to download. This is just a belt-and-suspenders, as
-                // this should already be called whenever a transaction becomes AlreadyHaveTx().
-                m_txrequest.ForgetTxHash(gtxid.GetHash());
             }
-        }
-
+        } // release m_tx_download_mutex
 
         if (!vGetData.empty())
             MakeAndPushMessage(*pto, NetMsgType::GETDATA, vGetData);

src/txorphanage.h

@@ -18,6 +18,7 @@
  * Since we cannot distinguish orphans from bad transactions with
  * non-existent inputs, we heavily limit the number of orphans
  * we keep and the duration we keep them for.
+ * Not thread-safe. Requires external synchronization.
  */
 class TxOrphanage {
 public:

src/validation.cpp

@@ -3553,7 +3553,7 @@ bool Chainstate::ActivateBestChain(BlockValidationState& state, std::shared_ptr<
             } // 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());
+                m_chainman.m_options.signals->ActiveTipChange(*Assert(pindexNewTip), m_chainman.IsInitialBlockDownload());
             }
         } // release cs_main
         // When we reach this point, we switched to a new tip (stored in pindexNewTip).
@@ -3778,7 +3778,7 @@ bool Chainstate::InvalidateBlock(BlockValidationState& state, CBlockIndex* pinde
         // 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());
+            m_chainman.m_options.signals->ActiveTipChange(*Assert(m_chain.Tip()), m_chainman.IsInitialBlockDownload());
         }
     }
     return true;

src/validationinterface.cpp

@@ -183,9 +183,9 @@ void ValidationSignals::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlo
                           fInitialDownload);
 }
 
-void ValidationSignals::ActiveTipChange(const CBlockIndex *new_tip, bool is_ibd)
+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);
+    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); });
 }
 

src/validationinterface.h

@@ -64,7 +64,7 @@ class CValidationInterface {
     /**
      * Notifies listeners any time the block chain tip changes, synchronously.
      */
-    virtual void ActiveTipChange(const CBlockIndex* new_tip, bool is_ibd) {};
+    virtual void ActiveTipChange(const CBlockIndex& new_tip, bool is_ibd) {};
     /**
      * Notifies listeners of a transaction having been added to mempool.
      *
@@ -218,7 +218,7 @@ class ValidationSignals {
     void SyncWithValidationInterfaceQueue() LOCKS_EXCLUDED(cs_main);
 
     void UpdatedBlockTip(const CBlockIndex *, const CBlockIndex *, bool fInitialDownload);
-    void ActiveTipChange(const CBlockIndex*, bool);
+    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);