diff --git a/doc/release-notes-34628.md b/doc/release-notes-34628.md new file mode 100644 index 00000000000..884ac0734d2 --- /dev/null +++ b/doc/release-notes-34628.md @@ -0,0 +1,13 @@ +P2P and network changes +----------------------- + +- To reduce memory and CPU usage during periods of high transaction + volume, rate-limiting of outgoing transaction relay has been changed + to use a global backlog instead of being done on a per-peer basis. The + default rate-limit remains as 14 tx/s (boosted by 2.5x for outbound + peers), though this can be changed via the `-txsendrate` configuration + option. An additional bandwidth rate-limit has also been introduced + at 12MB of transactions per 10 minutes, with a high burst rate. The + size of the global backlog and the token bucket values for the rate + limits can be queried via the `getnetworkinfo` RPC. (#34628) + diff --git a/src/init.cpp b/src/init.cpp index 9fbf686383c..e7a9d59b177 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -716,6 +716,10 @@ void SetupServerArgs(ArgsManager& argsman, bool can_listen_ipc) OptionsCategory::NODE_RELAY); argsman.AddArg("-minrelaytxfee=", strprintf("Fees (in %s/kvB) smaller than this are considered zero fee for relaying, mining and transaction creation (default: %s)", CURRENCY_UNIT, FormatMoney(DEFAULT_MIN_RELAY_TX_FEE)), ArgsManager::ALLOW_ANY, OptionsCategory::NODE_RELAY); + argsman.AddArg("-txsendrate=", + strprintf("Set the maximum ongoing rate for sending transactions to (inbound) peers (default: %u tx/s)", + DEFAULT_TX_SEND_RATE), + ArgsManager::ALLOW_ANY | ArgsManager::DEBUG_ONLY, OptionsCategory::NODE_RELAY); argsman.AddArg("-privatebroadcast", strprintf( "Broadcast transactions submitted via sendrawtransaction RPC using short-lived " diff --git a/src/net_processing.cpp b/src/net_processing.cpp index dfbeac0706d..03e5c90b491 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -59,6 +59,7 @@ #include #include #include +#include #include #include @@ -167,15 +168,16 @@ static constexpr auto INBOUND_INVENTORY_BROADCAST_INTERVAL{5s}; * Use a smaller delay as there is less privacy concern for them. * Blocks and peers with NetPermissionFlags::NoBan permission bypass this. */ static constexpr auto OUTBOUND_INVENTORY_BROADCAST_INTERVAL{2s}; -/** Maximum rate of inventory items to send per second. - * Limits the impact of low-fee transaction floods. */ -static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND{14}; -/** Target number of tx inventory items to send per transmission. */ -static constexpr unsigned int INVENTORY_BROADCAST_TARGET = INVENTORY_BROADCAST_PER_SECOND * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL); -/** Maximum number of inventory items to send per transmission. */ -static constexpr unsigned int INVENTORY_BROADCAST_MAX = 1000; -static_assert(INVENTORY_BROADCAST_MAX >= INVENTORY_BROADCAST_TARGET, "INVENTORY_BROADCAST_MAX too low"); -static_assert(INVENTORY_BROADCAST_MAX <= node::MAX_PEER_TX_ANNOUNCEMENTS, "INVENTORY_BROADCAST_MAX too high"); +/** Multiplier for the inventory bucket rate for outbounds */ +static constexpr double OUTBOUND_INVENTORY_BUCKET_MULTIPLIER{Ticks(INBOUND_INVENTORY_BROADCAST_INTERVAL) / Ticks(OUTBOUND_INVENTORY_BROADCAST_INTERVAL)}; +/** Delay between checking inventory bucket and backlog */ +static constexpr auto INVENTORY_BUCKET_CHECK_DELAY{100ms}; +/** Empty backlog target capacity */ +static constexpr size_t INVENTORY_BUCKET_BACKLOG_CAPACITY{300}; +/** Delay between inventory bucket backlog heartbeat log entries */ +static constexpr auto INVENTORY_BUCKET_BACKLOG_HEARTBEAT{2000ms}; +/** Minimum backlog to trigger heartbeat log entries */ +static constexpr size_t INVENTORY_BUCKET_BACKLOG_HEARTBEAT_MIN{100}; /** Average delay between feefilter broadcasts in seconds. */ static constexpr auto AVG_FEEFILTER_BROADCAST_INTERVAL{10min}; /** Maximum feefilter broadcast delay after significant change. */ @@ -299,11 +301,11 @@ struct Peer { * us or we have announced to the peer. We use this to avoid announcing * the same (w)txid to a peer that already has the transaction. */ CRollingBloomFilter m_tx_inventory_known_filter GUARDED_BY(m_tx_inventory_mutex){50000, 0.000001}; - /** Set of wtxids we still have to announce. For non-wtxid-relay peers, + /** Vector of wtxids we still have to announce. For non-wtxid-relay peers, * we retrieve the txid from the corresponding mempool transaction when * constructing the `inv` message. We use the mempool to sort transactions * in dependency order before relay, so this does not have to be sorted. */ - std::set m_tx_inventory_to_send GUARDED_BY(m_tx_inventory_mutex); + std::vector m_tx_inventory_to_send GUARDED_BY(m_tx_inventory_mutex); /** Whether the peer has requested us to send our complete mempool. Only * permitted if the peer has NetPermissionFlags::Mempool or we advertise * NODE_BLOOM. See BIP35. */ @@ -498,6 +500,68 @@ struct CNodeState { int64_t m_last_block_announcement{0}; }; +struct InvToSendBucket { + const double count_floor{0}; + std::vector backlog; + util::TokenBucket size_bucket; + util::TokenBucket count_bucket; + + /* Initialization rationale: + * + * Count bucket: Fills at rate*mult, total/initial capacity of 30s with mult=1 + * Size bucket: Fills at 12MB every 600s, times mult so expected to be 6 times + * the rate at which blocks can confirm transactions, but at least 3 times that in + * the worst case. High limit to avoid triggering even with large spikes, but a + * modest initial value to ensure that frequent node restarts don't raise the limit + * too much. + * Count floor: In order to avoid sorting the global backlog too often, we ensure + * that we always remove at least an average INV message's number of transactions + * each time we do work. (Or 50kB if the size bucket is the limiting factor) + */ + + static constexpr double SIZE_INIT{12'000'000}; // 12 MB initially + static constexpr double SIZE_CAP{50'000'000}; // 50 MB maximum + static constexpr double SIZE_REFILL{20'000}; // 20kB/s = 12MB/600s + + static constexpr double INBOUND_COUNT_SECONDS{30}; // cap/initial at 30s/mult worth of txs + + InvToSendBucket(unsigned int rate, double mult) + : count_floor{-1.0 * rate * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL)}, + size_bucket(/*rate=*/SIZE_REFILL * mult, /*value=*/SIZE_INIT, /*cap=*/SIZE_CAP), + count_bucket(/*rate=*/rate * mult, /*value=*/rate * INBOUND_COUNT_SECONDS, /*cap=*/rate * INBOUND_COUNT_SECONDS) + { + } + + bool avail() const + { + return !backlog.empty() && size_bucket.value() > 0 && count_bucket.value() > 0; + } + + void increment(NodeClock::time_point now) + { + size_bucket.increment(now); + count_bucket.increment(now); + } + + std::vector TakeForProcessing(CTxMemPool& mempool) EXCLUSIVE_LOCKS_REQUIRED(mempool.cs); + + bool decrement(double size) + { + bool size_ok = size_bucket.decrement(size, /*floor=*/-50e3); + bool count_ok = count_bucket.decrement(1, /*floor=*/count_floor); + return size_ok && count_ok; + } + + PeerManagerInfo::InvBucketInfo info() const + { + return { + .backlog_count = backlog.size(), + .count_bucket = count_bucket.value(), + .size_bucket = size_bucket.value(), + }; + } +}; + class PeerManagerImpl final : public PeerManager { public: @@ -524,9 +588,9 @@ public: void FinalizeNode(const CNode& node) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_headers_presync_mutex, !m_tx_download_mutex); bool HasAllDesirableServiceFlags(ServiceFlags services) const override; bool ProcessMessages(CNode& node, std::atomic& interrupt) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); bool SendMessages(CNode& node) override - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); /** Implement PeerManager */ void StartScheduledTasks(CScheduler& scheduler) override; @@ -535,11 +599,11 @@ public: EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); bool GetNodeStateStats(NodeId nodeid, CNodeStateStats& stats) const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); std::vector GetOrphanTransactions() override EXCLUSIVE_LOCKS_REQUIRED(!m_tx_download_mutex); - PeerManagerInfo GetInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + PeerManagerInfo GetInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); std::vector GetPrivateBroadcastInfo() const override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); std::vector AbortPrivateBroadcast(const uint256& id) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SendPings() override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); - void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void InitiateTxBroadcastToAll(const Wtxid& wtxid) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); node::TransactionError InitiateTxBroadcastPrivate(const CTransactionRef& tx) override EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); void SetBestBlock(int height, std::chrono::seconds time) override { @@ -553,7 +617,7 @@ public: private: void ProcessMessage(Peer& peer, CNode& pfrom, const std::string& msg_type, DataStream& vRecv, NodeClock::time_point time_received, const std::atomic& interruptMsgProc) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_most_recent_block_mutex, !m_headers_presync_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); /** Consider evicting an outbound peer based on the amount of time they've been behind our tip */ void ConsiderEviction(CNode& pto, Peer& peer, std::chrono::seconds time_in_seconds) EXCLUSIVE_LOCKS_REQUIRED(cs_main, g_msgproc_mutex); @@ -562,7 +626,7 @@ private: void EvictExtraOutboundPeers(NodeClock::time_point now) EXCLUSIVE_LOCKS_REQUIRED(cs_main); /** Retrieve unbroadcast transactions from the mempool and reattempt sending to peers */ - void ReattemptInitialBroadcast(CScheduler& scheduler) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex); + void ReattemptInitialBroadcast(CScheduler& scheduler) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); /** Rebroadcast stale private transactions (already broadcast but not received back from the network). */ void ReattemptPrivateBroadcast(CScheduler& scheduler); @@ -634,13 +698,13 @@ private: /** Handle a transaction whose result was MempoolAcceptResult::ResultType::VALID. * Updates m_txrequest, m_orphanage, and vExtraTxnForCompact. Also queues the tx for relay. */ void ProcessValidTx(NodeId nodeid, const CTransactionRef& tx, const std::list& replaced_transactions) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex, !m_inv_to_send_mutex); /** Handle the results of package validation: calls ProcessValidTx and ProcessInvalidTx for * individual transactions, and caches rejection for the package as a group. */ void ProcessPackageResult(const node::PackageToValidate& package_to_validate, const PackageMempoolAcceptResult& package_result) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, m_tx_download_mutex, !m_inv_to_send_mutex); /** * Reconsider orphan transactions after a parent has been accepted to the mempool. @@ -654,7 +718,7 @@ private: * will be empty. */ bool ProcessOrphanTx(Peer& peer) - EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, !m_tx_download_mutex); + EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, g_msgproc_mutex, !m_tx_download_mutex, !m_inv_to_send_mutex); /** Process a single headers message from a peer. * @@ -1119,6 +1183,14 @@ private: /// The transactions to be broadcast privately. PrivateBroadcast m_tx_for_private_broadcast; + + mutable Mutex m_inv_to_send_mutex ACQUIRED_BEFORE(m_mempool.cs); + InvToSendBucket m_inbound_inv_bucket GUARDED_BY(m_inv_to_send_mutex); + InvToSendBucket m_outbound_inv_bucket GUARDED_BY(m_inv_to_send_mutex); + std::atomic m_next_inv_bucket_check{NodeClock::time_point::min()}; + std::optional m_next_inv_bucket_heartbeat GUARDED_BY(m_inv_to_send_mutex); + + void ProcessInvBacklog(NodeClock::time_point now, bool backlog_bumped=false) EXCLUSIVE_LOCKS_REQUIRED(!m_peer_mutex, !m_inv_to_send_mutex); }; const CNodeState* PeerManagerImpl::State(NodeId pnode) const @@ -1655,7 +1727,7 @@ void PeerManagerImpl::ReattemptInitialBroadcast(CScheduler& scheduler) CTransactionRef tx = m_mempool.get(txid); if (tx != nullptr) { - InitiateTxBroadcastToAll(txid, tx->GetWitnessHash()); + InitiateTxBroadcastToAll(tx->GetWitnessHash()); } else { m_mempool.RemoveUnbroadcastTx(txid, true); } @@ -1887,10 +1959,14 @@ std::vector PeerManagerImpl::GetOrphanTransaction PeerManagerInfo PeerManagerImpl::GetInfo() const { + LOCK(m_inv_to_send_mutex); return PeerManagerInfo{ .median_outbound_time_offset = m_outbound_time_offsets.Median(), .ignores_incoming_txs = m_opts.ignore_incoming_txs, .private_broadcast = m_opts.private_broadcast, + .tx_send_rate = m_opts.tx_send_rate, + .inbound_bucket = m_inbound_inv_bucket.info(), + .outbound_bucket = m_outbound_inv_bucket.info(), }; } @@ -2059,7 +2135,9 @@ PeerManagerImpl::PeerManagerImpl(CConnman& connman, AddrMan& addrman, m_mempool(pool), m_txdownloadman(node::TxDownloadOptions{pool, m_rng, opts.deterministic_rng}), m_warnings{warnings}, - m_opts{opts} + m_opts{opts}, + m_inbound_inv_bucket(/*rate=*/m_opts.tx_send_rate, /*mult=*/1.0), + m_outbound_inv_bucket(/*rate=*/m_opts.tx_send_rate, /*mult=*/OUTBOUND_INVENTORY_BUCKET_MULTIPLIER) { // While Erlay support is incomplete, it must be enabled explicitly via -txreconciliation. // This argument can go away after Erlay support is complete. @@ -2286,28 +2364,126 @@ void PeerManagerImpl::SendPings() for(auto& it : m_peer_map) it.second->m_ping_queued = true; } -void PeerManagerImpl::InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) +std::vector InvToSendBucket::TakeForProcessing(CTxMemPool& mempool) { - for (const PeerRef& peer_ref : GetAllPeers()) { - if (!peer_ref) continue; - Peer& peer{*peer_ref}; + AssertLockHeld(mempool.cs); - auto tx_relay = peer.GetTxRelay(); - if (!tx_relay) continue; + size_t n_to_take = static_cast(std::max(count_bucket.value() - count_floor, 0)); - LOCK(tx_relay->m_tx_inventory_mutex); - // Only queue transactions for announcement once the version handshake - // is completed. The time of arrival for these transactions is - // otherwise at risk of leaking to a spy, if the spy is able to - // distinguish transactions received during the handshake from the rest - // in the announcement. - if (tx_relay->m_next_inv_send_time == 0s) continue; + std::vector best; - const uint256& hash{peer.m_wtxid_relay ? wtxid.ToUint256() : txid.ToUint256()}; - if (!tx_relay->m_tx_inventory_known_filter.contains(hash)) { - tx_relay->m_tx_inventory_to_send.insert(wtxid); + auto itervec = mempool.ExtractBestByMiningScoreWithTopology(backlog, n_to_take); + bool tokens_left = true; + for (auto txiter : itervec) { + auto& wtxid = txiter->GetTx().GetWitnessHash(); + if (tokens_left) { + best.push_back(wtxid); + if (!decrement(txiter->GetTx().ComputeTotalSize())) { + tokens_left = false; + } + } else { + backlog.push_back(wtxid); } } + + // if the backlog is now empty, consider shrinking it if it's oversized + if (backlog.empty() && backlog.capacity() > INVENTORY_BUCKET_BACKLOG_CAPACITY) { + std::vector dummy; + dummy.reserve(INVENTORY_BUCKET_BACKLOG_CAPACITY); + dummy.swap(backlog); + } + + return best; +} + +void PeerManagerImpl::ProcessInvBacklog(NodeClock::time_point now, bool backlog_bumped) +{ + // Don't run the body of this function unless it's been a little + // while since the last run, or we just added a new tx to the backlog. + if (!backlog_bumped && now <= m_next_inv_bucket_check.load()) return; + m_next_inv_bucket_check = now + INVENTORY_BUCKET_CHECK_DELAY; + + LOCK(m_inv_to_send_mutex); + m_inbound_inv_bucket.increment(now); + m_outbound_inv_bucket.increment(now); + + // Regular heartbeat logging when there's a backlog + if (!m_next_inv_bucket_heartbeat.has_value()) { + if (m_inbound_inv_bucket.backlog.size() >= INVENTORY_BUCKET_BACKLOG_HEARTBEAT_MIN || m_outbound_inv_bucket.backlog.size() >= INVENTORY_BUCKET_BACKLOG_HEARTBEAT_MIN) { + m_next_inv_bucket_heartbeat = now; + } + } + if (m_next_inv_bucket_heartbeat.has_value() && now >= *m_next_inv_bucket_heartbeat) { + LogDebug(BCLog::NET, "Transaction rate-limiting backlog inbound=%d itok=%.1f isz=%.1f outbound=%d otok=%.1f osz=%.1f", + m_inbound_inv_bucket.backlog.size(), + m_inbound_inv_bucket.count_bucket.value(), + m_inbound_inv_bucket.size_bucket.value(), + m_outbound_inv_bucket.backlog.size(), + m_outbound_inv_bucket.count_bucket.value(), + m_outbound_inv_bucket.size_bucket.value()); + if (m_inbound_inv_bucket.backlog.empty() && m_outbound_inv_bucket.backlog.empty()) { + m_next_inv_bucket_heartbeat = std::nullopt; + } else { + m_next_inv_bucket_heartbeat = now + INVENTORY_BUCKET_BACKLOG_HEARTBEAT; + } + } + + // Early exit to skip pointlessly touching mempool lock + bool in_avail = m_inbound_inv_bucket.avail(); + bool out_avail = m_outbound_inv_bucket.avail(); + if (!in_avail && !out_avail) return; + + std::vector for_inbound; + std::vector for_outbound; + + { + LOCK(m_mempool.cs); + if (in_avail) for_inbound = m_inbound_inv_bucket.TakeForProcessing(m_mempool); + if (out_avail) for_outbound = m_outbound_inv_bucket.TakeForProcessing(m_mempool); + } + + if (!for_inbound.empty() || !for_outbound.empty()) { + bool any_inbound_connected = false; + bool any_outbound_connected = false; + for (const PeerRef& peer_ref : GetAllPeers()) { + if (!peer_ref) continue; + Peer& peer{*peer_ref}; + auto tx_relay = peer.GetTxRelay(); + if (!tx_relay) continue; + + LOCK(tx_relay->m_tx_inventory_mutex); + // Only queue transactions for announcement once the version handshake + // is completed. The time of arrival for these transactions is + // otherwise at risk of leaking to a spy, if the spy is able to + // distinguish transactions received during the handshake from the rest + // in the announcement. + if (tx_relay->m_next_inv_send_time == 0s) continue; + if (peer.m_is_inbound) { + any_inbound_connected = true; + } else { + any_outbound_connected = true; + } + for (auto& i : (peer.m_is_inbound ? for_inbound : for_outbound)) { + tx_relay->m_tx_inventory_to_send.push_back(i); + } + } + + // if the node has no in/outbound connections, clear the corresponding backlog entirely + // this reduces wasted memory, and avoids having the bucket artificially empty for when + // future peers do connect. + if (!any_inbound_connected) m_inbound_inv_bucket.backlog.clear(); + if (!any_outbound_connected) m_outbound_inv_bucket.backlog.clear(); + } +} + +void PeerManagerImpl::InitiateTxBroadcastToAll(const Wtxid& wtxid) +{ + { + LOCK(m_inv_to_send_mutex); + m_inbound_inv_bucket.backlog.push_back(wtxid); + m_outbound_inv_bucket.backlog.push_back(wtxid); + } + ProcessInvBacklog(NodeClock::now(), /*backlog_bumped=*/true); } node::TransactionError PeerManagerImpl::InitiateTxBroadcastPrivate(const CTransactionRef& tx) @@ -3212,7 +3388,7 @@ void PeerManagerImpl::ProcessValidTx(NodeId nodeid, const CTransactionRef& tx, c tx->GetWitnessHash().ToString(), m_mempool.size(), m_mempool.DynamicMemoryUsage() / 1000); - InitiateTxBroadcastToAll(tx->GetHash(), tx->GetWitnessHash()); + InitiateTxBroadcastToAll(tx->GetWitnessHash()); for (const CTransactionRef& removedTx : replaced_transactions) { AddToCompactExtraTransactions(removedTx); @@ -4537,7 +4713,7 @@ void PeerManagerImpl::ProcessMessage(Peer& peer, CNode& pfrom, const std::string } else { LogInfo("Force relaying tx %s (wtxid=%s) from peer=%d\n", txid.ToString(), wtxid.ToString(), pfrom.GetId()); - InitiateTxBroadcastToAll(txid, wtxid); + InitiateTxBroadcastToAll(wtxid); } } @@ -5662,22 +5838,6 @@ void PeerManagerImpl::MaybeSendFeefilter(CNode& pto, Peer& peer, std::chrono::mi } } -namespace { -class CompareInvMempoolOrder -{ - const CTxMemPool* m_mempool; -public: - explicit CompareInvMempoolOrder(CTxMemPool* mempool) : m_mempool{mempool} {} - - bool operator()(std::set::iterator a, std::set::iterator b) - { - /* As std::make_heap produces a max-heap, we want the entries with the - * higher mining score to sort later. */ - return m_mempool->CompareMiningScoreWithTopology(*b, *a); - } -}; -} // namespace - bool PeerManagerImpl::RejectIncomingTxs(const CNode& peer) const { // block-relay-only peers may never send txs to us @@ -5911,6 +6071,8 @@ bool PeerManagerImpl::SendMessages(CNode& node) MaybeSendSendHeaders(node, peer); + ProcessInvBacklog(now); + { LOCK(cs_main); @@ -6110,7 +6272,7 @@ bool PeerManagerImpl::SendMessages(CNode& node) std::vector vInv; { LOCK(peer.m_block_inv_mutex); - vInv.reserve(std::max(peer.m_blocks_for_inv_relay.size(), INVENTORY_BROADCAST_TARGET)); + vInv.reserve(peer.m_blocks_for_inv_relay.size()); // Add blocks for (const uint256& hash : peer.m_blocks_for_inv_relay) { @@ -6145,9 +6307,16 @@ bool PeerManagerImpl::SendMessages(CNode& node) // Respond to BIP35 mempool requests if (fSendTrickle && tx_relay->m_send_mempool) { auto vtxinfo = m_mempool.infoAll(); + + // Ensure we'll respond to GETDATA requests for anything we're about to announce + tx_relay->m_last_inv_sequence = WITH_LOCK(m_mempool.cs, return m_mempool.GetSequence()); + tx_relay->m_send_mempool = false; const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; + // we'll send everything in the mempool momentarily, so this is redundant + tx_relay->m_tx_inventory_to_send.clear(); + LOCK(tx_relay->m_bloom_filter_mutex); for (const auto& txinfo : vtxinfo) { @@ -6156,7 +6325,6 @@ bool PeerManagerImpl::SendMessages(CNode& node) const auto inv = peer.m_wtxid_relay ? CInv{MSG_WTX, wtxid.ToUint256()} : CInv{MSG_TX, txid.ToUint256()}; - tx_relay->m_tx_inventory_to_send.erase(wtxid); // Don't send transactions that peers will not put into their mempool if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) { @@ -6176,64 +6344,55 @@ bool PeerManagerImpl::SendMessages(CNode& node) // Determine transactions to relay if (fSendTrickle) { - // Produce a vector with all candidates for sending - std::vector::iterator> vInvTx; - vInvTx.reserve(tx_relay->m_tx_inventory_to_send.size()); - for (std::set::iterator it = tx_relay->m_tx_inventory_to_send.begin(); it != tx_relay->m_tx_inventory_to_send.end(); it++) { - vInvTx.push_back(it); - } - const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; // Topologically and fee-rate sort the inventory we send for privacy and priority reasons. - // A heap is used so that not all items need sorting if only a few are being sent. - CompareInvMempoolOrder compareInvMempoolOrder(&m_mempool); - std::make_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); - // 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; - 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(INVENTORY_BROADCAST_MAX, broadcast_max); - while (!vInvTx.empty() && nRelayedTransactions < broadcast_max) { - // Fetch the top element from the heap - std::pop_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); - std::set::iterator it = vInvTx.back(); - vInvTx.pop_back(); - auto wtxid = *it; - // Remove it from the to-be-sent set - tx_relay->m_tx_inventory_to_send.erase(it); - // Not in the mempool anymore? don't bother sending it. - auto txinfo = m_mempool.info(wtxid); - if (!txinfo.tx) { - continue; + // (sorted from higher priority to lowest, skipping low fee) + const CFeeRate filterrate{tx_relay->m_fee_filter_received.load()}; + + auto inv_tx = [&]() EXCLUSIVE_LOCKS_REQUIRED(tx_relay->m_tx_inventory_mutex) { + auto& invs = tx_relay->m_tx_inventory_to_send; + std::vector res; + + if (invs.size() == 0) return res; + + // if previous allocations were excessive, shrink to the current size + if (invs.capacity() > 2 * invs.size()) invs.shrink_to_fit(); + + LOCK(m_mempool.cs); + auto txiters = m_mempool.ExtractBestByMiningScoreWithTopology(invs, invs.size()); + res.reserve(txiters.size()); + for (auto txiter : txiters) { + if (txiter->GetFee() < filterrate.GetFee(txiter->GetTxSize())) { + continue; // higher feerate CPFP txs may follow, so just skip, don't stop + } + res.push_back(txiter->GetSharedTx()); } + // Ensure we'll respond to GETDATA requests for anything we're about to announce + tx_relay->m_last_inv_sequence = m_mempool.GetSequence(); + return res; + }(); + + LOCK(tx_relay->m_bloom_filter_mutex); + vInv.reserve(std::min(MAX_INV_SZ, vInv.size() + inv_tx.size())); + for (auto& tx : inv_tx) { // `TxRelay::m_tx_inventory_known_filter` contains either txids or wtxids // depending on whether our peer supports wtxid-relay. Therefore, first // construct the inv and then use its hash for the filter check. const auto inv = peer.m_wtxid_relay ? - CInv{MSG_WTX, wtxid.ToUint256()} : - CInv{MSG_TX, txinfo.tx->GetHash().ToUint256()}; + CInv{MSG_WTX, tx->GetWitnessHash().ToUint256()} : + CInv{MSG_TX, tx->GetHash().ToUint256()}; // Check if not in the filter already if (tx_relay->m_tx_inventory_known_filter.contains(inv.hash)) { continue; } - // Peer told you to not send transactions at that feerate? Don't bother sending it. - if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) { - continue; - } - if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*txinfo.tx)) continue; + if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*tx)) continue; // Send vInv.push_back(inv); - nRelayedTransactions++; if (vInv.size() == MAX_INV_SZ) { MakeAndPushMessage(node, NetMsgType::INV, vInv); vInv.clear(); } tx_relay->m_tx_inventory_known_filter.insert(inv.hash); } - - // 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(); } } if (!vInv.empty()) diff --git a/src/net_processing.h b/src/net_processing.h index f29adc0157f..a381a6d80bc 100644 --- a/src/net_processing.h +++ b/src/net_processing.h @@ -42,6 +42,8 @@ static constexpr bool DEFAULT_TXRECONCILIATION_ENABLE{false}; /** Default number of non-mempool transactions to keep around for block reconstruction. Includes orphan, replaced, and rejected transactions. */ static const uint32_t DEFAULT_BLOCK_RECONSTRUCTION_EXTRA_TXN{100}; +/** Default maximum per-second rate for sending transaction inventory to peers. */ +static constexpr unsigned int DEFAULT_TX_SEND_RATE{14}; static const bool DEFAULT_PEERBLOOMFILTERS = false; static const bool DEFAULT_PEERBLOCKFILTERS = false; /** Maximum number of outstanding CMPCTBLOCK requests for the same block. */ @@ -70,9 +72,18 @@ struct CNodeStateStats { }; struct PeerManagerInfo { + struct InvBucketInfo { + size_t backlog_count{0}; + double count_bucket{0}; + double size_bucket{0}; + }; + std::chrono::seconds median_outbound_time_offset{0s}; bool ignores_incoming_txs{false}; bool private_broadcast{DEFAULT_PRIVATE_BROADCAST}; + unsigned int tx_send_rate{0}; + InvBucketInfo inbound_bucket; + InvBucketInfo outbound_bucket; }; class PeerManager : public CValidationInterface, public NetEventsInterface @@ -96,6 +107,8 @@ public: uint32_t max_headers_result{MAX_HEADERS_RESULTS}; //! Whether private broadcast is used for sending transactions. bool private_broadcast{DEFAULT_PRIVATE_BROADCAST}; + //! Maximum per-second rate for sending transaction inventory to peers. + unsigned int tx_send_rate{DEFAULT_TX_SEND_RATE}; }; static std::unique_ptr make(CConnman& connman, AddrMan& addrman, @@ -139,11 +152,11 @@ public: /** * Initiate a transaction broadcast to eligible peers. - * Queue the witness transaction id to `Peer::TxRelay::m_tx_inventory_to_send` - * for each peer. Later, depending on `Peer::TxRelay::m_next_inv_send_time` and if + * Queue the witness transaction id to the inbound and outbound inv backlogs. + * Later, depending on `-txsendrate`, `Peer::TxRelay::m_next_inv_send_time` and if * the transaction is in the mempool, an `INV` about it may be sent to the peer. */ - virtual void InitiateTxBroadcastToAll(const Txid& txid, const Wtxid& wtxid) = 0; + virtual void InitiateTxBroadcastToAll(const Wtxid& wtxid) = 0; /** * Initiate a private transaction broadcast. This is done diff --git a/src/node/peerman_args.cpp b/src/node/peerman_args.cpp index 9745d69d5ae..47413af0ce5 100644 --- a/src/node/peerman_args.cpp +++ b/src/node/peerman_args.cpp @@ -24,6 +24,10 @@ void ApplyArgsManOptions(const ArgsManager& argsman, PeerManager::Options& optio if (auto value{argsman.GetBoolArg("-blocksonly")}) options.ignore_incoming_txs = *value; + if (auto value{argsman.GetIntArg("-txsendrate")}) { + options.tx_send_rate = uint32_t(std::clamp(*value, 1, 1000)); + } + if (auto value{argsman.GetBoolArg("-privatebroadcast")}) options.private_broadcast = *value; } diff --git a/src/node/transaction.cpp b/src/node/transaction.cpp index e7877c69855..d331ae05aa6 100644 --- a/src/node/transaction.cpp +++ b/src/node/transaction.cpp @@ -130,7 +130,7 @@ TransactionError BroadcastTransaction(NodeContext& node, case TxBroadcast::MEMPOOL_NO_BROADCAST: break; case TxBroadcast::MEMPOOL_AND_BROADCAST_TO_ALL: - node.peerman->InitiateTxBroadcastToAll(txid, wtxid); + node.peerman->InitiateTxBroadcastToAll(wtxid); break; case TxBroadcast::NO_MEMPOOL_PRIVATE_BROADCAST: return node.peerman->InitiateTxBroadcastPrivate(tx); diff --git a/src/node/txdownloadman.h b/src/node/txdownloadman.h index 2cc1ec2c279..bef1d162d22 100644 --- a/src/node/txdownloadman.h +++ b/src/node/txdownloadman.h @@ -25,7 +25,7 @@ class TxDownloadManagerImpl; static constexpr int32_t MAX_PEER_TX_REQUEST_IN_FLIGHT = 100; /** Maximum number of transactions to consider for requesting, per peer. It provides a reasonable DoS limit to * per-peer memory usage spent on announcements, while covering peers continuously sending INVs at the maximum - * rate (by our own policy, see INVENTORY_BROADCAST_PER_SECOND) for several minutes, while not receiving + * rate (by our own policy, see DEFAULT_TX_SEND_RATE) for several minutes, while not receiving * the actual transaction (from any peer) in response to requests for them. */ static constexpr int32_t MAX_PEER_TX_ANNOUNCEMENTS = 5000; /** How long to delay requesting transactions via txids, if we have wtxid-relaying peers */ diff --git a/src/rpc/net.cpp b/src/rpc/net.cpp index ba1080ed11f..2d20ff16106 100644 --- a/src/rpc/net.cpp +++ b/src/rpc/net.cpp @@ -659,6 +659,16 @@ static RPCMethod getnetworkinfo() }}, {RPCResult::Type::BOOL, "localrelay", "true if transaction relay is requested from peers"}, {RPCResult::Type::NUM, "timeoffset", "the time offset"}, + {RPCResult::Type::NUM, "tx_send_rate", "configured target for maximum number of transactions per second to send to inbound peers"}, + {RPCResult::Type::OBJ_DYN, "inv_buckets", "", { + {RPCResult::Type::OBJ, "inbound/outbound", "connection direction", + { + {RPCResult::Type::NUM, "backlog", "number of queued txs to announce"}, + {RPCResult::Type::NUM, "count_tok", "tokens available to be consumed per-transaction"}, + {RPCResult::Type::NUM, "size_tok", "tokens available to be consumed per-byte"}, + } + } + }}, {RPCResult::Type::NUM, "connections", "the total number of connections"}, {RPCResult::Type::NUM, "connections_in", "the number of inbound connections"}, {RPCResult::Type::NUM, "connections_out", "the number of outbound connections"}, @@ -716,6 +726,18 @@ static RPCMethod getnetworkinfo() auto peerman_info{node.peerman->GetInfo()}; obj.pushKV("localrelay", !peerman_info.ignores_incoming_txs); obj.pushKV("timeoffset", Ticks(peerman_info.median_outbound_time_offset)); + obj.pushKV("tx_send_rate", peerman_info.tx_send_rate); + auto buckjson = [&](const auto& buckinfo) { + UniValue b{UniValue::VOBJ}; + b.pushKV("backlog", buckinfo.backlog_count); + b.pushKV("count_tok", buckinfo.count_bucket); + b.pushKV("size_tok", buckinfo.size_bucket); + return b; + }; + UniValue invbuckets{UniValue::VOBJ}; + invbuckets.pushKV("inbound", buckjson(peerman_info.inbound_bucket)); + invbuckets.pushKV("outbound", buckjson(peerman_info.outbound_bucket)); + obj.pushKV("inv_buckets", invbuckets); } if (node.connman) { obj.pushKV("networkactive", node.connman->GetNetworkActive()); diff --git a/src/test/util_tests.cpp b/src/test/util_tests.cpp index 6eddd3627c5..4dbe9b456a1 100644 --- a/src/test/util_tests.cpp +++ b/src/test/util_tests.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include #include @@ -1929,4 +1930,135 @@ BOOST_AUTO_TEST_CASE(gib_string_literal_test) BOOST_CHECK_EQUAL(32_GiB, 32768_MiB); } +BOOST_AUTO_TEST_CASE(token_bucket_initial_value) +{ + // Initial value is clamped to cap + util::TokenBucket b1(/*rate=*/1, /*value=*/100, /*cap=*/10); + BOOST_CHECK_EQUAL(b1.value(), 10); + + // Initial value below cap is kept as-is + util::TokenBucket b2(/*rate=*/1, /*value=*/5, /*cap=*/10); + BOOST_CHECK_EQUAL(b2.value(), 5); +} + +BOOST_AUTO_TEST_CASE(token_bucket_first_increment) +{ + // First increment establishes the time baseline but does not refill + util::TokenBucket b(/*rate=*/100, /*value=*/0, /*cap=*/1000); + b.increment(NodeClock::time_point{10s}); + BOOST_CHECK_EQUAL(b.value(), 0); + + // Second increment refills based on elapsed time + b.increment(NodeClock::time_point{15s}); + BOOST_CHECK_EQUAL(b.value(), 500); // 100/s * 5s +} + +BOOST_AUTO_TEST_CASE(token_bucket_refill_caps) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/90, /*cap=*/100); + b.increment(NodeClock::time_point{1s}); + b.increment(NodeClock::time_point{100s}); // would add 990, but cap is 100 + BOOST_CHECK_EQUAL(b.value(), 100); +} + +BOOST_AUTO_TEST_CASE(token_bucket_time_backwards) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/50, /*cap=*/200); + b.increment(NodeClock::time_point{10s}); + b.increment(NodeClock::time_point{5s}); // backwards, no change + BOOST_CHECK_EQUAL(b.value(), 50); + b.increment(NodeClock::time_point{15s}); // forwards takes backwards into account + BOOST_CHECK_EQUAL(b.value(), 150); +} + +BOOST_AUTO_TEST_CASE(token_bucket_decrement_no_debt) +{ + // Default debt=0: returns false at exactly 0 + util::TokenBucket b(/*rate=*/1, /*value=*/3, /*cap=*/10); + BOOST_CHECK(b.decrement(1)); // 3 -> 2 + BOOST_CHECK(b.decrement(1)); // 2 -> 1 + BOOST_CHECK(!b.decrement(1)); // 1 -> 0, at floor + BOOST_CHECK_EQUAL(b.value(), 0); + BOOST_CHECK(!b.decrement(1)); // 0 -> -1, despite being at floor + BOOST_CHECK_EQUAL(b.value(), -1); +} + +BOOST_AUTO_TEST_CASE(token_bucket_decrement_with_debt) +{ + util::TokenBucket b(/*rate=*/1, /*value=*/2, /*cap=*/10); + BOOST_CHECK(b.decrement(1, -3)); // 2 -> 1 + BOOST_CHECK(b.decrement(1, -3)); // 1 -> 0 + BOOST_CHECK(b.decrement(1, -3)); // 0 -> -1, still above -3 + BOOST_CHECK(b.decrement(1, -3)); // -1 -> -2, still above -3 + BOOST_CHECK(!b.decrement(1, -3)); // -2 -> -3, at floor + BOOST_CHECK_EQUAL(b.value(), -3); +} + +BOOST_AUTO_TEST_CASE(token_bucket_drain_and_refill) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/20, /*cap=*/100); + b.decrement(20); // drain to 0 + BOOST_CHECK_EQUAL(b.value(), 0); + + b.increment(NodeClock::time_point{1s}); + b.increment(NodeClock::time_point{4s}); // +30 + BOOST_CHECK_EQUAL(b.value(), 30); +} + + +BOOST_AUTO_TEST_CASE(token_bucket_first_increment_at_epoch) +{ + // The first increment establishes the baseline (no refill) even when it + // lands exactly on the clock epoch; later increments then refill normally. + util::TokenBucket b(/*rate=*/100, /*value=*/0, /*cap=*/1000); + b.increment(NodeClock::time_point{0s}); + BOOST_CHECK_EQUAL(b.value(), 0); + b.increment(NodeClock::time_point{5s}); + BOOST_CHECK_EQUAL(b.value(), 500); // 100/s * 5s +} + +BOOST_AUTO_TEST_CASE(token_bucket_at_cap_advances_baseline) +{ + util::TokenBucket b(/*rate=*/10, /*value=*/100, /*cap=*/100); + BOOST_CHECK_EQUAL(b.value(), 100); // already at cap + b.increment(NodeClock::time_point{1s}); // baseline established at 1s + b.increment(NodeClock::time_point{100s}); // 99s spent at the cap; baseline -> 100s + BOOST_CHECK_EQUAL(b.value(), 100); + + b.decrement(100); // drain to 0 + BOOST_CHECK_EQUAL(b.value(), 0); + + // refill doesn't "bank" the extra 99s we were at cap + b.increment(NodeClock::time_point{101s}); + BOOST_CHECK_EQUAL(b.value(), 10); + + // And when real time genuinely elapses, a single increment refills straight + // back to the cap immediately. + b.increment(NodeClock::time_point{200s}); // 99s elapsed -> +990, clamped to cap + BOOST_CHECK_EQUAL(b.value(), 100); +} + +BOOST_AUTO_TEST_CASE(token_bucket_fractional_refill) +{ + // Sub-second elapsed time accumulates fractional tokens via double math. + util::TokenBucket b(/*rate=*/10, /*value=*/0, /*cap=*/100); + b.increment(NodeClock::time_point{1s}); + b.increment(NodeClock::time_point{1250ms}); // 10/s * 0.25s = 2.5 + BOOST_CHECK_EQUAL(b.value(), 2.5); +} + +BOOST_AUTO_TEST_CASE(token_bucket_refill_from_debt) +{ + // Refilling from a negative (debt) balance accrues normally and still + // clamps to the cap rather than to debt + increment. + util::TokenBucket b(/*rate=*/10, /*value=*/0, /*cap=*/100); + BOOST_CHECK(!b.decrement(50)); // -> -50, below floor 0 + BOOST_CHECK_EQUAL(b.value(), -50); + b.increment(NodeClock::time_point{1s}); // baseline + b.increment(NodeClock::time_point{4s}); // +30 -> -20 + BOOST_CHECK_EQUAL(b.value(), -20); + b.increment(NodeClock::time_point{100s}); // +960 but clamped to cap + BOOST_CHECK_EQUAL(b.value(), 100); +} + BOOST_AUTO_TEST_SUITE_END() diff --git a/src/txmempool.cpp b/src/txmempool.cpp index 86cbc1adc43..e5ec3b32cab 100644 --- a/src/txmempool.cpp +++ b/src/txmempool.cpp @@ -458,7 +458,7 @@ void CTxMemPool::check(const CCoinsViewCache& active_coins_tip, int64_t spendhei assert(diagram.size() <= score_with_topo.size() + 1); assert(diagram.size() >= 1); - std::optional last_wtxid = std::nullopt; + std::optional last_iter = std::nullopt; auto diagram_iter = diagram.cbegin(); for (const auto& it : score_with_topo) { @@ -480,11 +480,10 @@ void CTxMemPool::check(const CCoinsViewCache& active_coins_tip, int64_t spendhei innerUsage += it->DynamicMemoryUsage(); const CTransaction& tx = it->GetTx(); - // CompareMiningScoreWithTopology should agree with GetSortedScoreWithTopology() - if (last_wtxid) { - assert(CompareMiningScoreWithTopology(*last_wtxid, tx.GetWitnessHash())); + if (last_iter) { + assert(m_txgraph->CompareMainOrder(**last_iter, *it) < 0); } - last_wtxid = tx.GetWitnessHash(); + last_iter = it; std::set setParentCheck; std::set setParentsStored; @@ -553,20 +552,62 @@ void CTxMemPool::check(const CCoinsViewCache& active_coins_tip, int64_t spendhei assert(innerUsage == cachedInnerUsage); } -bool CTxMemPool::CompareMiningScoreWithTopology(const Wtxid& hasha, const Wtxid& hashb) const +std::vector CTxMemPool::ExtractBestByMiningScoreWithTopology(std::vector& wtxids, size_t n_to_sort) const { - /* Return `true` if hasha should be considered sooner than hashb, namely when: - * a is not in the mempool but b is, or - * both are in the mempool but a is sorted before b in the total mempool ordering - * (which takes dependencies and (chunk) feerates into account). + /* This function takes a vector of `wtxids`, and returns the + * best mempool entries corresponding to those `wtxids` (by mining + * score/topology). It updates the input `wtxids` so that multiple + * calls with the same vector will drain that vector to empty. + * + * It operates under the following constraints: + * - wtxids that do not correspond to a mempool entry are dropped + * - the return vector contains no duplicates, either with itself + * or with the updated `wtxids` input. + * - the return vector will have `n_to_sort` entries (or `wtxids` + will become empty). + * - the `wtxids` vector will be reduced by at least `n_to_sort` + * entries (or will become empty). */ - LOCK(cs); - auto j{GetIter(hashb)}; - if (!j.has_value()) return false; - auto i{GetIter(hasha)}; - if (!i.has_value()) return true; - return m_txgraph->CompareMainOrder(*i.value(), *j.value()) < 0; + auto cmp = [&](const auto& a, const auto& b) EXCLUSIVE_LOCKS_REQUIRED(cs) noexcept { return m_txgraph->CompareMainOrder(*a, *b) < 0; }; + + std::vector res; + + n_to_sort = std::min(wtxids.size(), n_to_sort); + if (n_to_sort > 0) { + res.reserve(wtxids.size()); + std::sort(wtxids.begin(), wtxids.end()); + for (auto it = wtxids.begin(); it != wtxids.end(); ++it) { + // skip duplicates + auto itnext = it + 1; + if (itnext != wtxids.end() && *it == *itnext) continue; + + if (auto i{GetIter(*it)}; i.has_value()) { + res.push_back(i.value()); + } + } + wtxids.clear(); + + if (!res.empty()) { + auto begin = res.begin(); + auto end = res.end(); + auto middle = end; + if (n_to_sort >= res.size()) { + // use regular sort when sorting everything + std::sort(begin, end, cmp); + } else { + middle = begin + n_to_sort; + std::partial_sort(begin, middle, end, cmp); + } + auto it = middle; + while (it != end) { + wtxids.push_back((*it)->GetTx().GetWitnessHash()); + ++it; + } + res.erase(middle, end); + } + } + return res; } std::vector CTxMemPool::GetSortedScoreWithTopology() const diff --git a/src/txmempool.h b/src/txmempool.h index 48082cbe82f..eb7e44a8c68 100644 --- a/src/txmempool.h +++ b/src/txmempool.h @@ -331,7 +331,19 @@ public: void removeForReorg(CChain& chain, std::function filter_final_and_mature) EXCLUSIVE_LOCKS_REQUIRED(cs, cs_main); void removeForBlock(const std::vector& vtx, unsigned int nBlockHeight) EXCLUSIVE_LOCKS_REQUIRED(cs); - bool CompareMiningScoreWithTopology(const Wtxid& hasha, const Wtxid& hashb) const; + /** Look up wtxids in the mempool and (partially) sort by mining score. + * + * The @p n_to_sort best entries are removed from @p wtxids and their + * corresponding txiter entries are returned. In addition wtxids + * that are duplicates or were not found in the mempool are silently + * dropped from @p wtxids. The returned vector is ordered from best + * to worst (by CompareMainOrder). Entries remaining in @p wtxids + * are in unspecified order. + * + * Note that the returned `txiter` values may become invalidated once + * mempool.cs is released. + */ + std::vector ExtractBestByMiningScoreWithTopology(std::vector& wtxids, size_t n_to_sort) const EXCLUSIVE_LOCKS_REQUIRED(cs); bool isSpent(const COutPoint& outpoint) const; unsigned int GetTransactionsUpdated() const; void AddTransactionsUpdated(unsigned int n); diff --git a/src/util/tokenbucket.h b/src/util/tokenbucket.h new file mode 100644 index 00000000000..ae8e8b53ad9 --- /dev/null +++ b/src/util/tokenbucket.h @@ -0,0 +1,69 @@ +// Copyright (c) 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_UTIL_TOKENBUCKET_H +#define BITCOIN_UTIL_TOKENBUCKET_H + +#include + +namespace util { + +/** A token bucket rate limiter. + * + * Tokens are added at a steady rate (m_rate per second) up to a capacity + * cap (m_cap). Tokens are removed by calling decrement(), which returns + * false if the bucket is emptied. + * + * Typical usage: + * bucket.increment(now); // refill based on elapsed time + * if (bucket.value() >= 1) bucket.decrement(1); // consume a token + */ +template +class TokenBucket +{ +public: + using clock = Clock; + using time_point = typename Clock::time_point; + using duration = typename Clock::duration; + + const double m_rate{1}; //!< Tokens added per second + const double m_cap{0}; //!< Maximum token balance + + /** @param rate Tokens added per second. + * @param value Initial token balance (clamped to cap). + * @param cap Maximum token balance. */ + TokenBucket(double rate, double value, double cap) : m_rate{rate}, m_cap{cap}, m_value{std::min(value, cap)} {} + + /** Refill tokens based on elapsed time since last call. No refill + * occurs on the first call (establishes the time baseline). */ + void increment(const time_point& now) + { + if (now > m_last_updated) { + if (m_value < m_cap && m_last_updated > MIN_TIME) { + double inc = m_rate * std::chrono::duration_cast(now - m_last_updated).count(); + m_value = std::min(m_cap, m_value + inc); + } + } + m_last_updated = now; + } + + /** Consume n tokens. Returns false if the balance dropped to/below the given floor. */ + bool decrement(double n = 1.0, double floor = 0.0) + { + m_value -= n; + return (m_value > floor); + } + + /** Current token balance. */ + double value() const { return m_value; } + +private: + static constexpr time_point MIN_TIME{time_point::min()}; + time_point m_last_updated{MIN_TIME}; + double m_value{0}; +}; + +} // namespace util + +#endif // BITCOIN_UTIL_TOKENBUCKET_H diff --git a/test/functional/mempool_limit.py b/test/functional/mempool_limit.py index 4770e155de8..2e554bb80c5 100755 --- a/test/functional/mempool_limit.py +++ b/test/functional/mempool_limit.py @@ -5,6 +5,7 @@ """Test mempool limiting together/eviction with the wallet.""" from decimal import Decimal +import time from test_framework.mempool_util import ( fill_mempool, @@ -205,7 +206,9 @@ class MempoolLimitTest(BitcoinTestFramework): self.log.info('Check that mempoolminfee is minrelaytxfee') assert_equal(node.getmempoolinfo()['minrelaytxfee'], node.getmempoolinfo()["mempoolminfee"]) + node.setmocktime(int(time.time())-3600) fill_mempool(self, node) + node.setmocktime(0) # bump time forward so the rate limit buckets refresh and don't block broadcast # Deliberately try to create a tx with a fee less than the minimum mempool fee to assert that it does not get added to the mempool self.log.info('Create a mempool tx that will not pass mempoolminfee') diff --git a/test/functional/p2p_tx_relay_rate_limit.py b/test/functional/p2p_tx_relay_rate_limit.py new file mode 100755 index 00000000000..9c61244412b --- /dev/null +++ b/test/functional/p2p_tx_relay_rate_limit.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +# Copyright (c) The Bitcoin Core developers +# Distributed under the MIT software license, see the accompanying +# file COPYING or http://www.opensource.org/licenses/mit-license.php. +"""Test transaction relay rate limiting via token buckets. + +With -txsendrate=R, the inbound count bucket has capacity R*30. A broadcast +transaction is relayed immediately while the bucket has tokens; once it is +exhausted the excess transactions queue in a global backlog and drain as the +bucket refills (R tokens/second as mocktime advances). +""" + +from decimal import Decimal +import time + +from test_framework.blocktools import COINBASE_MATURITY +from test_framework.p2p import P2PTxInvStore +from test_framework.test_framework import BitcoinTestFramework +from test_framework.util import assert_equal +from test_framework.wallet import MiniWallet + +SEND_RATE = 2 # -txsendrate value +BUCKET_CAP = SEND_RATE * 30 # count bucket capacity (60) +NUM_TXS = 80 # total transactions to submit + + +class TxRelayRateLimitTest(BitcoinTestFramework): + def set_test_params(self): + self.num_nodes = 1 + self.extra_args = [[f'-txsendrate={SEND_RATE}']] + + def inbound_backlog(self, node): + return node.getnetworkinfo()['inv_buckets']['inbound']['backlog'] + + def run_test(self): + node = self.nodes[0] + wallet = MiniWallet(node) + + node.setmocktime(int(time.time())) + + # Mine enough blocks for mature coinbase UTXOs. + self.generate(wallet, COINBASE_MATURITY + NUM_TXS + 50) + + # Connect an inbound peer (negotiates wtxid relay by default) + peer = node.add_p2p_connection(P2PTxInvStore()) + + # Advance time so the peer's trickle timer initializes + node.bumpmocktime(10) + peer.sync_with_ping() + assert_equal(len(peer.get_invs()), 0) + + # Verify the configured send rate + assert_equal(node.getnetworkinfo()['tx_send_rate'], SEND_RATE) + + self.test_rate_limit_and_rbf(node, wallet, peer) + + def test_rate_limit_and_rbf(self, node, wallet, peer): + self.log.info(f"Submitting {NUM_TXS} transactions at frozen time (bucket capacity {BUCKET_CAP})") + + # Prepare an RBF pair: original and replacement spending the same UTXO. + # Both are created upfront so we can submit the replacement later. + rbf_utxo = wallet.get_utxo() + tx_rbf_orig = wallet.create_self_transfer(utxo_to_spend=rbf_utxo) + tx_rbf_repl = wallet.create_self_transfer(utxo_to_spend=rbf_utxo, fee_rate=Decimal("0.009")) + + # Submit NUM_TXS transactions at frozen time. Each broadcast is relayed + # immediately while the count bucket has tokens, so the first BUCKET_CAP + # are handed straight to the peer and the rest queue in the backlog. The + # RBF original is placed in the backlogged tail. + RBF_INDEX = NUM_TXS - 5 + for i in range(NUM_TXS): + if i == RBF_INDEX: + node.sendrawtransaction(tx_rbf_orig['hex']) + else: + wallet.send_self_transfer(from_node=node) + + # The excess beyond the bucket capacity is backlogged. Nothing is + # announced yet -- the per-peer trickle timer hasn't fired (frozen time). + self.log.info(f"Backlog after burst: {self.inbound_backlog(node)}") + assert_equal(self.inbound_backlog(node), NUM_TXS - BUCKET_CAP) + assert_equal(len(peer.get_invs()), 0) + + # RBF the backlogged original while time is still frozen, so the + # replacement also queues in the backlog (the bucket is exhausted). The + # original's wtxid stays in the backlog vector for now but is dropped + # when the backlog is processed, since it is no longer in the mempool. + self.log.info("RBF'ing a backlogged transaction") + node.sendrawtransaction(tx_rbf_repl['hex']) + assert_equal(self.inbound_backlog(node), NUM_TXS - BUCKET_CAP + 1) + + # Advance time so the bucket refills and the backlog drains. Loop until + # the backlog is empty and every surviving tx has trickled out. + self.log.info("Advancing time to drain the backlog") + for _ in range(30): + if self.inbound_backlog(node) == 0 and len(peer.get_invs()) == NUM_TXS: + break + node.bumpmocktime(4) + peer.sync_with_ping() + + announced = set(peer.get_invs()) + self.log.info(f"Total announced: {len(announced)}") + + # Every surviving transaction is announced: the burst minus the dropped + # RBF original, plus the replacement. + assert_equal(self.inbound_backlog(node), 0) + assert_equal(len(announced), NUM_TXS) + assert int(tx_rbf_orig['wtxid'], 16) not in announced + assert int(tx_rbf_repl['wtxid'], 16) in announced + + self.log.info("Rate limiting and RBF backlog cleanup test passed") + + +if __name__ == '__main__': + TxRelayRateLimitTest(__file__).main() diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py index 5ad5bbbde91..b1adce6cb1a 100755 --- a/test/functional/test_runner.py +++ b/test/functional/test_runner.py @@ -275,6 +275,7 @@ BASE_SCRIPTS = [ 'wallet_importprunedfunds.py', 'p2p_leak_tx.py --v1transport', 'p2p_leak_tx.py --v2transport', + 'p2p_tx_relay_rate_limit.py', 'p2p_eviction.py', 'p2p_outbound_eviction.py', 'p2p_ibd_stalling.py --v1transport',