Merge #21236: net processing: Extract `addr` send functionality into MaybeSendAddr()

935d4889228e7e361c8b0020761fa0e08a55fb48 [net processing] Refactor MaybeSendAddr() (John Newbery)
01a79ff924b11f91796d4aa63c571897b047ac7d [net processing] Fix overindentation in MaybeSendAddr() (John Newbery)
38c0be5da3af17208b165e73cee7612d3670b038 [net processing] Refactor MaybeSendAddr() - early exits (John Newbery)
c87423c58b5165de835a49bebd566538a70c07ab [net processing] Change MaybeSendAddr() to take a reference (John Newbery)
ad719297f2ecdd2394eff668b3be7070bc9cb3e2 [net processing] Extract `addr` send functionality into MaybeSendAddr() (John Newbery)
4ad4abcf07efefafd439b28679dff8d6bbf62943 [net] Change addr send times fields to be guarded by new mutex (John Newbery)
c02fa47baa517e17b5c43bde3902b1e410c1b93f [net processing] Only call GetTime() once in SendMessages() (John Newbery)

Pull request description:

  This continues the work of moving application layer data into net_processing. It refactors `addr` send functionality into its own function `MaybeSendAddr()` and flattens/simplifies the code. Isolating and simplifying the addr handling code makes subsequent changes (which will move addr data and logic into net processing) easier to review.

  This is a pure refactor. There are no functional changes.

  For motivation of the project, see #19398.

ACKs for top commit:
  sipa:
    utACK 935d4889228e7e361c8b0020761fa0e08a55fb48
  hebasto:
    ACK 935d4889228e7e361c8b0020761fa0e08a55fb48, I have reviewed the code and it looks OK, I agree it can be merged.
  MarcoFalke:
    review ACK 935d4889228e7e361c8b0020761fa0e08a55fb48 🐑

Tree-SHA512: 4e9dc84603147e74f479a211b42bcf315bdf5d14c21c08cf0b17d6c252775b90b012f0e0d834f1a607ed63c7ed5c63d5cf49b134344e7b64a1695bfcff111c92

2 files changed, 87 insertions, 75 deletions

diff --git a/src/net.h b/src/net.h
index add48b11a4..cae22969ce 100644
--- a/src/net.h
+++ b/src/net.h
@@ -549,8 +549,9 @@ public:
     std::vector<CAddress> vAddrToSend;
     std::unique_ptr<CRollingBloomFilter> m_addr_known{nullptr};
     bool fGetAddr{false};
-    std::chrono::microseconds m_next_addr_send GUARDED_BY(cs_sendProcessing){0};
-    std::chrono::microseconds m_next_local_addr_send GUARDED_BY(cs_sendProcessing){0};
+    Mutex m_addr_send_times_mutex;
+    std::chrono::microseconds m_next_addr_send GUARDED_BY(m_addr_send_times_mutex){0};
+    std::chrono::microseconds m_next_local_addr_send GUARDED_BY(m_addr_send_times_mutex){0};
 
     struct TxRelay {
         mutable RecursiveMutex cs_filter;
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
index e4054968c0..14b628acf0 100644
--- a/src/net_processing.cpp
+++ b/src/net_processing.cpp
@@ -33,6 +33,7 @@
 #include <util/system.h>
 #include <validation.h>
 
+#include <algorithm>
 #include <memory>
 #include <optional>
 #include <typeinfo>
@@ -317,8 +318,13 @@ private:
     void PushNodeVersion(CNode& pnode, int64_t nTime);
 
     /** Send a ping message every PING_INTERVAL or if requested via RPC. May
-     *  mark the peer to be disconnected if a ping has timed out. */
-    void MaybeSendPing(CNode& node_to, Peer& peer);
+     *  mark the peer to be disconnected if a ping has timed out.
+     *  We use mockable time for ping timeouts, so setmocktime may cause pings
+     *  to time out. */
+    void MaybeSendPing(CNode& node_to, Peer& peer, std::chrono::microseconds now);
+
+    /** Send `addr` messages on a regular schedule. */
+    void MaybeSendAddr(CNode& node, std::chrono::microseconds current_time);
 
     const CChainParams& m_chainparams;
     CConnman& m_connman;
@@ -4100,12 +4106,8 @@ void PeerManagerImpl::CheckForStaleTipAndEvictPeers()
     }
 }
 
-void PeerManagerImpl::MaybeSendPing(CNode& node_to, Peer& peer)
+void PeerManagerImpl::MaybeSendPing(CNode& node_to, Peer& peer, std::chrono::microseconds now)
 {
-    // Use mockable time for ping timeouts.
-    // This means that setmocktime may cause pings to time out.
-    auto now = GetTime<std::chrono::microseconds>();
-
     if (m_connman.RunInactivityChecks(node_to) && peer.m_ping_nonce_sent &&
         now > peer.m_ping_start.load() + std::chrono::seconds{TIMEOUT_INTERVAL}) {
         LogPrint(BCLog::NET, "ping timeout: %fs peer=%d\n", 0.000001 * count_microseconds(now - peer.m_ping_start.load()), peer.m_id);
@@ -4144,6 +4146,75 @@ void PeerManagerImpl::MaybeSendPing(CNode& node_to, Peer& peer)
     }
 }
 
+void PeerManagerImpl::MaybeSendAddr(CNode& node, std::chrono::microseconds current_time)
+{
+    // Nothing to do for non-address-relay peers
+    if (!node.RelayAddrsWithConn()) return;
+
+    assert(node.m_addr_known);
+
+    LOCK(node.m_addr_send_times_mutex);
+    // Periodically advertise our local address to the peer.
+    if (fListen && !m_chainman.ActiveChainstate().IsInitialBlockDownload() &&
+        node.m_next_local_addr_send < current_time) {
+        // If we've sent before, clear the bloom filter for the peer, so that our
+        // self-announcement will actually go out.
+        // This might be unnecessary if the bloom filter has already rolled
+        // over since our last self-announcement, but there is only a small
+        // bandwidth cost that we can incur by doing this (which happens
+        // once a day on average).
+        if (node.m_next_local_addr_send != 0us) {
+            node.m_addr_known->reset();
+        }
+        if (std::optional<CAddress> local_addr = GetLocalAddrForPeer(&node)) {
+            FastRandomContext insecure_rand;
+            node.PushAddress(*local_addr, insecure_rand);
+        }
+        node.m_next_local_addr_send = PoissonNextSend(current_time, AVG_LOCAL_ADDRESS_BROADCAST_INTERVAL);
+    }
+
+    // We sent an `addr` message to this peer recently. Nothing more to do.
+    if (current_time <= node.m_next_addr_send) return;
+
+    node.m_next_addr_send = PoissonNextSend(current_time, AVG_ADDRESS_BROADCAST_INTERVAL);
+
+    if (!Assume(node.vAddrToSend.size() <= MAX_ADDR_TO_SEND)) {
+        // Should be impossible since we always check size before adding to
+        // vAddrToSend. Recover by trimming the vector.
+        node.vAddrToSend.resize(MAX_ADDR_TO_SEND);
+    }
+
+    // Remove addr records that the peer already knows about, and add new
+    // addrs to the m_addr_known filter on the same pass.
+    auto addr_already_known = [&node](const CAddress& addr) {
+        bool ret = node.m_addr_known->contains(addr.GetKey());
+        if (!ret) node.m_addr_known->insert(addr.GetKey());
+        return ret;
+    };
+    node.vAddrToSend.erase(std::remove_if(node.vAddrToSend.begin(), node.vAddrToSend.end(), addr_already_known),
+                           node.vAddrToSend.end());
+
+    // No addr messages to send
+    if (node.vAddrToSend.empty()) return;
+
+    const char* msg_type;
+    int make_flags;
+    if (node.m_wants_addrv2) {
+        msg_type = NetMsgType::ADDRV2;
+        make_flags = ADDRV2_FORMAT;
+    } else {
+        msg_type = NetMsgType::ADDR;
+        make_flags = 0;
+    }
+    m_connman.PushMessage(&node, CNetMsgMaker(node.GetCommonVersion()).Make(make_flags, msg_type, node.vAddrToSend));
+    node.vAddrToSend.clear();
+
+    // we only send the big addr message once
+    if (node.vAddrToSend.capacity() > 40) {
+        node.vAddrToSend.shrink_to_fit();
+    }
+}
+
 namespace {
 class CompareInvMempoolOrder
 {
@@ -4182,79 +4253,20 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
     // If we get here, the outgoing message serialization version is set and can't change.
     const CNetMsgMaker msgMaker(pto->GetCommonVersion());
 
-    MaybeSendPing(*pto, *peer);
+    const auto current_time = GetTime<std::chrono::microseconds>();
+
+    MaybeSendPing(*pto, *peer, current_time);
 
     // MaybeSendPing may have marked peer for disconnection
     if (pto->fDisconnect) return true;
 
+    MaybeSendAddr(*pto, current_time);
+
     {
         LOCK(cs_main);
 
         CNodeState &state = *State(pto->GetId());
 
-        // Address refresh broadcast
-        auto current_time = GetTime<std::chrono::microseconds>();
-
-        if (fListen && pto->RelayAddrsWithConn() &&
-            !m_chainman.ActiveChainstate().IsInitialBlockDownload() &&
-            pto->m_next_local_addr_send < current_time) {
-            // If we've sent before, clear the bloom filter for the peer, so that our
-            // self-announcement will actually go out.
-            // This might be unnecessary if the bloom filter has already rolled
-            // over since our last self-announcement, but there is only a small
-            // bandwidth cost that we can incur by doing this (which happens
-            // once a day on average).
-            if (pto->m_next_local_addr_send != 0us) {
-                pto->m_addr_known->reset();
-            }
-            if (std::optional<CAddress> local_addr = GetLocalAddrForPeer(pto)) {
-                FastRandomContext insecure_rand;
-                pto->PushAddress(*local_addr, insecure_rand);
-            }
-            pto->m_next_local_addr_send = PoissonNextSend(current_time, AVG_LOCAL_ADDRESS_BROADCAST_INTERVAL);
-        }
-
-        //
-        // Message: addr
-        //
-        if (pto->RelayAddrsWithConn() && pto->m_next_addr_send < current_time) {
-            pto->m_next_addr_send = PoissonNextSend(current_time, AVG_ADDRESS_BROADCAST_INTERVAL);
-            std::vector<CAddress> vAddr;
-            vAddr.reserve(pto->vAddrToSend.size());
-            assert(pto->m_addr_known);
-
-            const char* msg_type;
-            int make_flags;
-            if (pto->m_wants_addrv2) {
-                msg_type = NetMsgType::ADDRV2;
-                make_flags = ADDRV2_FORMAT;
-            } else {
-                msg_type = NetMsgType::ADDR;
-                make_flags = 0;
-            }
-
-            for (const CAddress& addr : pto->vAddrToSend)
-            {
-                if (!pto->m_addr_known->contains(addr.GetKey()))
-                {
-                    pto->m_addr_known->insert(addr.GetKey());
-                    vAddr.push_back(addr);
-                    // receiver rejects addr messages larger than MAX_ADDR_TO_SEND
-                    if (vAddr.size() >= MAX_ADDR_TO_SEND)
-                    {
-                        m_connman.PushMessage(pto, msgMaker.Make(make_flags, msg_type, vAddr));
-                        vAddr.clear();
-                    }
-                }
-            }
-            pto->vAddrToSend.clear();
-            if (!vAddr.empty())
-                m_connman.PushMessage(pto, msgMaker.Make(make_flags, msg_type, vAddr));
-            // we only send the big addr message once
-            if (pto->vAddrToSend.capacity() > 40)
-                pto->vAddrToSend.shrink_to_fit();
-        }
-
         // Start block sync
         if (pindexBestHeader == nullptr)
             pindexBestHeader = m_chainman.ActiveChain().Tip();
@@ -4489,7 +4501,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
                             vInv.clear();
                         }
                     }
-                    pto->m_tx_relay->m_last_mempool_req = GetTime<std::chrono::seconds>();
+                    pto->m_tx_relay->m_last_mempool_req = std::chrono::duration_cast<std::chrono::seconds>(current_time);
                 }
 
                 // Determine transactions to relay
@@ -4577,7 +4589,6 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
             m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv));
 
         // Detect whether we're stalling
-        current_time = GetTime<std::chrono::microseconds>();
         if (state.m_stalling_since.count() && state.m_stalling_since < current_time - BLOCK_STALLING_TIMEOUT) {
             // Stalling only triggers when the block download window cannot move. During normal steady state,
             // the download window should be much larger than the to-be-downloaded set of blocks, so disconnection