p2p: Fill reconciliation sets (Erlay)

src/Makefile.bench.include

@@ -51,6 +51,7 @@ bench_bench_bitcoin_SOURCES = \
   bench/rpc_mempool.cpp \
   bench/streams_findbyte.cpp \
   bench/strencodings.cpp \
+  bench/txreconciliation.cpp \
   bench/util_time.cpp \
   bench/verify_script.cpp \
   bench/xor.cpp

src/bench/txreconciliation.cpp

@@ -0,0 +1,40 @@
+// Copyright (c) 2020-2022 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 <bench/bench.h>
+
+#include <net.h>
+#include <node/txreconciliation.h>
+
+
+/* Benchmarks */
+
+static void ShouldFanoutTo(benchmark::Bench& bench)
+{
+    FastRandomContext frc{/*fDeterministic=*/true};
+    CSipHasher hasher(frc.rand64(), frc.rand64());
+    TxReconciliationTracker tracker(TXRECONCILIATION_VERSION, hasher);
+    // Register 120 inbound peers
+    int num_peers{120};
+    for (NodeId peer = 0; peer < num_peers; peer++) {
+        tracker.PreRegisterPeer(peer);
+        tracker.RegisterPeer(peer, /*is_peer_inbound=*/true, 1, 1);
+    }
+    FastRandomContext rc{/*fDeterministic=*/true};
+
+    // The target function uses caching, so we want to mimic tx repetitions
+    // of the real-world behavior.
+    std::vector<Wtxid> txs;
+    for (size_t i = 0; i < 1000; i++) {
+        txs.push_back(Wtxid::FromUint256(rc.rand256()));
+    }
+
+    bench.run([&] {
+        for (NodeId peer = 0; peer < num_peers; ++peer) {
+            tracker.ShouldFanoutTo(txs[rand() % txs.size()], peer, /*inbounds_fanout_tx_relay=*/0, /*outbounds_fanout_tx_relay=*/0);
+        }
+    });
+}
+
+BENCHMARK(ShouldFanoutTo, benchmark::PriorityLevel::HIGH);

src/net_processing.cpp

@@ -175,6 +175,8 @@ static constexpr double MAX_ADDR_RATE_PER_SECOND{0.1};
 static constexpr size_t MAX_ADDR_PROCESSING_TOKEN_BUCKET{MAX_ADDR_TO_SEND};
 /** The compactblocks version we support. See BIP 152. */
 static constexpr uint64_t CMPCTBLOCKS_VERSION{2};
+/** Used to determine whether to use low-fanout flooding (or reconciliation) for a tx relay event. */
+static const uint64_t RANDOMIZER_ID_FANOUTTARGET = 0xbac89af818407b6aULL; // SHA256("fanouttarget")[0:8]
 
 // Internal stuff
 namespace {
@@ -1608,7 +1610,7 @@ void PeerManagerImpl::FinalizeNode(const CNode& node)
     }
     m_orphanage.EraseForPeer(nodeid);
     m_txrequest.DisconnectedPeer(nodeid);
-    if (m_txreconciliation) m_txreconciliation->ForgetPeer(nodeid);
+    if (m_txreconciliation) m_txreconciliation->ForgetPeer(nodeid, node.IsInboundConn());
     m_num_preferred_download_peers -= state->fPreferredDownload;
     m_peers_downloading_from -= (!state->vBlocksInFlight.empty());
     assert(m_peers_downloading_from >= 0);
@@ -1896,7 +1898,8 @@ PeerManagerImpl::PeerManagerImpl(CConnman& connman, AddrMan& addrman,
     // While Erlay support is incomplete, it must be enabled explicitly via -txreconciliation.
     // This argument can go away after Erlay support is complete.
     if (opts.reconcile_txs) {
-        m_txreconciliation = std::make_unique<TxReconciliationTracker>(TXRECONCILIATION_VERSION);
+        m_txreconciliation = std::make_unique<TxReconciliationTracker>(TXRECONCILIATION_VERSION,
+                                                                       m_connman.GetDeterministicRandomizer(RANDOMIZER_ID_FANOUTTARGET));
     }
 }
 
@@ -3620,7 +3623,7 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
                 // We could have optimistically pre-registered/registered the peer. In that case,
                 // we should forget about the reconciliation state here if this wasn't followed
                 // by WTXIDRELAY (since WTXIDRELAY can't be announced later).
-                m_txreconciliation->ForgetPeer(pfrom.GetId());
+                m_txreconciliation->ForgetPeer(pfrom.GetId(), pfrom.IsInboundConn());
             }
         }
 
@@ -3914,6 +3917,9 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
                 if (!fAlreadyHave && !m_chainman.IsInitialBlockDownload()) {
                     AddTxAnnouncement(pfrom, gtxid, current_time);
                 }
+                if (m_txreconciliation && gtxid.IsWtxid()) {
+                    m_txreconciliation->TryRemovingFromSet(pfrom.GetId(), Wtxid::FromUint256(inv.hash));
+                }
             } else {
                 LogPrint(BCLog::NET, "Unknown inv type \"%s\" received from peer=%d\n", inv.ToString(), pfrom.GetId());
             }
@@ -4200,6 +4206,8 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
         m_txrequest.ReceivedResponse(pfrom.GetId(), txid);
         if (tx.HasWitness()) m_txrequest.ReceivedResponse(pfrom.GetId(), wtxid);
 
+        if (m_txreconciliation) m_txreconciliation->TryRemovingFromSet(pfrom.GetId(), Wtxid::FromUint256(wtxid));
+
         // We do the AlreadyHaveTx() check using wtxid, rather than txid - in the
         // absence of witness malleation, this is strictly better, because the
         // recent rejects filter may contain the wtxid but rarely contains
@@ -5751,6 +5759,8 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
         }
 
         if (auto tx_relay = peer->GetTxRelay(); tx_relay != nullptr) {
+                // Lock way before it's used to maintain lock ordering.
+                LOCK2(m_mempool.cs, m_peer_mutex);
                 LOCK(tx_relay->m_tx_inventory_mutex);
                 // Check whether periodic sends should happen
                 bool fSendTrickle = pto->HasPermission(NetPermissionFlags::NoBan);
@@ -5818,6 +5828,29 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
                     // No reason to drain out at many times the network's capacity,
                     // especially since we have many peers and some will draw much shorter delays.
                     unsigned int nRelayedTransactions = 0;
+
+                    size_t inbounds_fanout_tx_relay = 0, outbounds_fanout_tx_relay = 0;
+                    const bool reconciles_txs = m_txreconciliation && m_txreconciliation->IsPeerRegistered(pto->GetId());
+                    if (reconciles_txs) {
+                        for (const auto& [cur_peer_id, cur_peer] : m_peer_map) {
+                            // Skip the source of the transaction.
+                            if (cur_peer_id == pto->GetId()) continue;
+                            const auto cur_state{State(cur_peer_id)};
+                            if (!cur_state) continue;
+                            if (auto peer_tx_relay = cur_peer->GetTxRelay()) {
+                                LOCK(peer_tx_relay->m_bloom_filter_mutex);
+                                // When we consider to which (and how many) Erlay peers
+                                // we should fanout a tx, we must know to how
+                                // many peers we would certainly announce this tx
+                                // (non-Erlay peers).
+                                if (peer_tx_relay->m_relay_txs && !m_txreconciliation->IsPeerRegistered(cur_peer_id)) {
+                                    inbounds_fanout_tx_relay += cur_state->m_is_inbound;
+                                    outbounds_fanout_tx_relay += !cur_state->m_is_inbound;
+                                }
+                            }
+                        }
+                    }
+
                     LOCK(tx_relay->m_bloom_filter_mutex);
                     size_t broadcast_max{INVENTORY_BROADCAST_TARGET + (tx_relay->m_tx_inventory_to_send.size()/1000)*5};
                     broadcast_max = std::min<size_t>(INVENTORY_BROADCAST_MAX, broadcast_max);
@@ -5845,7 +5878,32 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
                         }
                         if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*txinfo.tx)) continue;
                         // Send
-                        vInv.push_back(inv);
+                        bool fanout = true;
+                        const auto wtxid = txinfo.tx->GetWitnessHash();
+                        if (reconciles_txs) {
+                            auto txiter = m_mempool.GetIter(txinfo.tx->GetHash());
+                            if (txiter) {
+                                // If a transaction has in-mempool children, always fanout it.
+                                // Until package relay is implemented, this is needed to avoid
+                                // breaking parent+child relay expectations in some cases.
+                                //
+                                // Potentially reconciling parent+child would mean that for every
+                                // child we need to check if any of the parents is currently
+                                // reconciled so that the child isn't fanouted ahead. But then
+                                // it gets tricky when reconciliation sets are full: a) the child
+                                // can't just be added; b) removing parents from reconciliation
+                                // sets for this one child is not good either.
+                                if ((*txiter)->GetCountWithDescendants() <= 1) {
+                                    fanout = m_txreconciliation->ShouldFanoutTo(wtxid, pto->GetId(),
+                                                                                inbounds_fanout_tx_relay, outbounds_fanout_tx_relay);
+                                }
+                            }
+                        }
+
+                        if (fanout || !m_txreconciliation->AddToSet(pto->GetId(), wtxid)) {
+                            vInv.push_back(inv);
+                        }
+
                         nRelayedTransactions++;
                         if (vInv.size() == MAX_INV_SZ) {
                             m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv));
@@ -5855,7 +5913,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
                     }
 
                     // Ensure we'll respond to GETDATA requests for anything we've just announced
-                    LOCK(m_mempool.cs);
                     tx_relay->m_last_inv_sequence = m_mempool.GetSequence();
                 }
         }