partial Merge #18638: net: Use mockable time for ping/pong, add tests

fa3365430c5fb57d7c0b5f2bce9fbbe290be93c3 net: Use mockable time for ping/pong, add tests (MarcoFalke)
faab4aaf2fa1153c6d76efc8113fa01b06943ece util: Add count_microseconds helper (MarcoFalke)

Pull request description:

  Switch `CNode::m_ping_start` and `CNetMessage::m_time` to mockable time, so that tests can be added.

  Mockable time is also type-safe, since it uses `std::chrono`

ACKs for top commit:
  jonatack:
    Code review re-ACK fa33654 re-read code, verified rebase per `git range-diff 4b5c919 fa94d6f fa33654`, previous tested ACKs still valid
  troygiorshev:
    ACK fa3365430c5fb57d7c0b5f2bce9fbbe290be93c3

Tree-SHA512: 7d632bd6019ce7c882029e71b667a61517e783af82755a85dd979ef09380934e172dec8b8f91d57b200a30a6e096aeaf01f19fee7f3aed0e0e871c72eb44d70e

src/net.cpp

@@ -691,15 +691,15 @@ void CNode::copyStats(CNodeStats &stats, const std::vector<bool> &m_asmap)
     // since pingtime does not update until the ping is complete, which might take a while.
     // So, if a ping is taking an unusually long time in flight,
     // the caller can immediately detect that this is happening.
-    int64_t nPingUsecWait = 0;
-    if ((0 != nPingNonceSent) && (0 != nPingUsecStart)) {
-        nPingUsecWait = GetTimeMicros() - nPingUsecStart;
+    std::chrono::microseconds ping_wait{0};
+    if ((0 != nPingNonceSent) && (0 != m_ping_start.load().count())) {
+        ping_wait = GetTime<std::chrono::microseconds>() - m_ping_start.load();
     }
 
     // Raw ping time is in microseconds, but show it to user as whole seconds (Dash users should be well used to small numbers with many decimal places by now :)
     stats.m_ping_usec = nPingUsecTime;
     stats.m_min_ping_usec  = nMinPingUsecTime;
-    stats.m_ping_wait_usec = nPingUsecWait;
+    stats.m_ping_wait_usec = count_microseconds(ping_wait);
 
     // Leave string empty if addrLocal invalid (not filled in yet)
     CService addrLocalUnlocked = GetAddrLocal();
@@ -718,9 +718,10 @@ void CNode::copyStats(CNodeStats &stats, const std::vector<bool> &m_asmap)
 bool CNode::ReceiveMsgBytes(Span<const uint8_t> msg_bytes, bool& complete)
 {
     complete = false;
-    int64_t nTimeMicros = GetTimeMicros();
+    // TODO: use mocktime here after bitcoin#19499 is backported
+    const auto time = std::chrono::microseconds(GetTimeMicros());
     LOCK(cs_vRecv);
-    nLastRecv = nTimeMicros / 1000000;
+    nLastRecv = std::chrono::duration_cast<std::chrono::seconds>(time).count();
     nRecvBytes += msg_bytes.size();
     while (msg_bytes.size() > 0) {
         // absorb network data
@@ -733,7 +734,7 @@ bool CNode::ReceiveMsgBytes(Span<const uint8_t> msg_bytes, bool& complete)
         if (m_deserializer->Complete()) {
             // decompose a transport agnostic CNetMessage from the deserializer
             uint32_t out_err_raw_size{0};
-            std::optional<CNetMessage> result{m_deserializer->GetMessage(nTimeMicros, out_err_raw_size)};
+            std::optional<CNetMessage> result{m_deserializer->GetMessage(time, out_err_raw_size)};
             if (!result) {
                 // Message deserialization failed.  Drop the message but don't disconnect the peer.
                 // store the size of the corrupt message
@@ -825,7 +826,7 @@ const uint256& V1TransportDeserializer::GetMessageHash() const
     return data_hash;
 }
 
-std::optional<CNetMessage> V1TransportDeserializer::GetMessage(int64_t time, uint32_t& out_err_raw_size)
+std::optional<CNetMessage> V1TransportDeserializer::GetMessage(const std::chrono::microseconds time, uint32_t& out_err_raw_size)
 {
     // decompose a single CNetMessage from the TransportDeserializer
     std::optional<CNetMessage> msg(std::move(vRecv));
@@ -1514,11 +1515,11 @@ bool CConnman::InactivityCheck(const CNode& node) const
         return true;
     }
 
-    if (node.nPingNonceSent && node.nPingUsecStart.load() + TIMEOUT_INTERVAL * 1000000 < GetTimeMicros()) {
+    if (node.nPingNonceSent && node.m_ping_start.load() + std::chrono::seconds{TIMEOUT_INTERVAL} < GetTime<std::chrono::microseconds>()) {
         // We use mockable time for ping timeouts. This means that setmocktime
         // may cause pings to time out for peers that have been connected for
         // longer than m_peer_connect_timeout.
-        LogPrintf("ping timeout: %fs\n", 0.000001 * (GetTimeMicros() - node.nPingUsecStart));
+        LogPrintf("ping timeout: %fs\n", 0.000001 * count_microseconds(GetTime<std::chrono::microseconds>() - node.m_ping_start.load()));
         return true;
     }
 

src/net.h

@@ -999,10 +999,10 @@ public:
  */
 class CNetMessage {
 public:
-    CDataStream m_recv;                  // received message data
-    int64_t m_time = 0;                  // time (in microseconds) of message receipt.
-    uint32_t m_message_size = 0;         // size of the payload
-    uint32_t m_raw_message_size = 0;     // used wire size of the message (including header/checksum)
+    CDataStream m_recv;                  //!< received message data
+    std::chrono::microseconds m_time{0}; //!< time of message receipt
+    uint32_t m_message_size{0};     //!< size of the payload
+    uint32_t m_raw_message_size{0}; //!< used wire size of the message (including header/checksum)
     std::string m_command;
 
     CNetMessage(CDataStream&& recv_in) : m_recv(std::move(recv_in)) {}
@@ -1026,7 +1026,7 @@ public:
     /** read and deserialize data, advances msg_bytes data pointer */
     virtual int Read(Span<const uint8_t>& msg_bytes) = 0;
     // decomposes a message from the context
-    virtual std::optional<CNetMessage> GetMessage(int64_t time, uint32_t& out_err) = 0;
+    virtual std::optional<CNetMessage> GetMessage(std::chrono::microseconds time, uint32_t& out_err) = 0;
     virtual ~TransportDeserializer() {}
 };
 
@@ -1090,7 +1090,7 @@ public:
         }
         return ret;
     }
-    std::optional<CNetMessage> GetMessage(int64_t time, uint32_t& out_err_raw_size) override;
+    std::optional<CNetMessage> GetMessage(std::chrono::microseconds time, uint32_t& out_err_raw_size) override;
 };
 
 /** The TransportSerializer prepares messages for the network transport
@@ -1340,8 +1340,8 @@ public:
     // Ping time measurement:
     // The pong reply we're expecting, or 0 if no pong expected.
     std::atomic<uint64_t> nPingNonceSent{0};
-    // Time (in usec) the last ping was sent, or 0 if no ping was ever sent.
-    std::atomic<int64_t> nPingUsecStart{0};
+    /** When the last ping was sent, or 0 if no ping was ever sent */
+    std::atomic<std::chrono::microseconds> m_ping_start{std::chrono::microseconds{0}};
     // Last measured round-trip time.
     std::atomic<int64_t> nPingUsecTime{0};
     // Best measured round-trip time.

src/net_processing.cpp

@@ -114,8 +114,8 @@ static constexpr int STALE_RELAY_AGE_LIMIT = 30 * 24 * 60 * 60;
 /// Age after which a block is considered historical for purposes of rate
 /// limiting block relay. Set to one week, denominated in seconds.
 static constexpr int HISTORICAL_BLOCK_AGE = 7 * 24 * 60 * 60;
-/** Time between pings automatically sent out for latency probing and keepalive (in seconds). */
-static const int PING_INTERVAL = 2 * 60;
+/** Time between pings automatically sent out for latency probing and keepalive */
+static constexpr std::chrono::minutes PING_INTERVAL{2};
 /** The maximum number of entries in a locator */
 static const unsigned int MAX_LOCATOR_SZ = 101;
 /** Number of blocks that can be requested at any given time from a single peer. */
@@ -262,7 +262,7 @@ public:
     void SetBestHeight(int height) override { m_best_height = height; };
     void Misbehaving(const NodeId pnode, const int howmuch, const std::string& message = "") override;
     void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv,
-                        int64_t nTimeReceived, const std::atomic<bool>& interruptMsgProc) override;
+                        const std::chrono::microseconds time_received, const std::atomic<bool>& interruptMsgProc) override;
     bool IsBanned(NodeId pnode) override EXCLUSIVE_LOCKS_REQUIRED(cs_main);
 
 private:
@@ -2857,7 +2857,7 @@ void PeerManagerImpl::ProcessMessage(
     CNode& pfrom,
     const std::string& msg_type,
     CDataStream& vRecv,
-    int64_t nTimeReceived,
+    const std::chrono::microseconds time_received,
     const std::atomic<bool>& interruptMsgProc)
 {
     LogPrint(BCLog::NET, "received: %s (%u bytes) peer=%d\n", SanitizeString(msg_type), vRecv.size(), pfrom.GetId());
@@ -3923,7 +3923,7 @@ void PeerManagerImpl::ProcessMessage(
         } // cs_main
 
         if (fProcessBLOCKTXN)
-            return ProcessMessage(pfrom, NetMsgType::BLOCKTXN, blockTxnMsg, nTimeReceived, interruptMsgProc);
+            return ProcessMessage(pfrom, NetMsgType::BLOCKTXN, blockTxnMsg, time_received, interruptMsgProc);
 
         if (fRevertToHeaderProcessing) {
             // Headers received from HB compact block peers are permitted to be
@@ -4182,7 +4182,7 @@ void PeerManagerImpl::ProcessMessage(
     }
 
     if (msg_type == NetMsgType::PONG) {
-        int64_t pingUsecEnd = nTimeReceived;
+        const auto ping_end = time_received;
         uint64_t nonce = 0;
         size_t nAvail = vRecv.in_avail();
         bool bPingFinished = false;
@@ -4196,11 +4196,11 @@ void PeerManagerImpl::ProcessMessage(
                 if (nonce == pfrom.nPingNonceSent) {
                     // Matching pong received, this ping is no longer outstanding
                     bPingFinished = true;
-                    int64_t pingUsecTime = pingUsecEnd - pfrom.nPingUsecStart;
-                    if (pingUsecTime >= 0) {
+                    const auto ping_time = ping_end - pfrom.m_ping_start.load();
+                    if (ping_time.count() >= 0) {
                         // Successful ping time measurement, replace previous
-                        pfrom.nPingUsecTime = pingUsecTime;
-                        pfrom.nMinPingUsecTime = std::min(pfrom.nMinPingUsecTime.load(), pingUsecTime);
+                        pfrom.nPingUsecTime = count_microseconds(ping_time);
+                        pfrom.nMinPingUsecTime = std::min(pfrom.nMinPingUsecTime.load(), count_microseconds(ping_time));
                     } else {
                         // This should never happen
                         sProblem = "Timing mishap";
@@ -4778,7 +4778,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
         // RPC ping request by user
         pingSend = true;
     }
-    if (pto->nPingNonceSent == 0 && pto->nPingUsecStart + PING_INTERVAL * 1000000 < GetTimeMicros()) {
+    if (pto->nPingNonceSent == 0 && pto->m_ping_start.load() + PING_INTERVAL < GetTime<std::chrono::microseconds>()) {
         // Ping automatically sent as a latency probe & keepalive.
         pingSend = true;
     }
@@ -4788,7 +4788,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
             GetRandBytes((unsigned char*)&nonce, sizeof(nonce));
         }
         pto->fPingQueued = false;
-        pto->nPingUsecStart = GetTimeMicros();
+        pto->m_ping_start = GetTime<std::chrono::microseconds>();
         pto->nPingNonceSent = nonce;
         m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::PING, nonce));
     }

src/net_processing.h

@@ -72,7 +72,7 @@ public:
 
     /** Process a single message from a peer. Public for fuzz testing */
     virtual void ProcessMessage(CNode& pfrom, const std::string& msg_type, CDataStream& vRecv,
-                                int64_t nTimeReceived, const std::atomic<bool>& interruptMsgProc) = 0;
+                                const std::chrono::microseconds time_received, const std::atomic<bool>& interruptMsgProc) = 0;
 
     virtual bool IsBanned(NodeId pnode) = 0;
 

src/test/fuzz/p2p_transport_serialization.cpp

@@ -67,7 +67,7 @@ FUZZ_TARGET_INIT(p2p_transport_serialization, initialize_p2p_transport_serializa
             break;
         }
         if (deserializer.Complete()) {
-            const int64_t m_time = std::numeric_limits<int64_t>::max();
+            const std::chrono::microseconds m_time{std::numeric_limits<int64_t>::max()};
             uint32_t out_err_raw_size{0};
             std::optional<CNetMessage> result{deserializer.GetMessage(m_time, out_err_raw_size)};
             if (result) {

src/test/fuzz/process_message.cpp

@@ -98,7 +98,7 @@ void fuzz_target(FuzzBufferType buffer, const std::string& LIMIT_TO_MESSAGE_TYPE
     // fuzzed_data_provider is fully consumed after this call, don't use it
     CDataStream random_bytes_data_stream{fuzzed_data_provider.ConsumeRemainingBytes<unsigned char>(), SER_NETWORK, PROTOCOL_VERSION};
     try {
-        g_setup->m_node.peerman->ProcessMessage(p2p_node, random_message_type, random_bytes_data_stream, GetTimeMillis(), std::atomic<bool>{false});
+        g_setup->m_node.peerman->ProcessMessage(p2p_node, random_message_type, random_bytes_data_stream, GetTime<std::chrono::microseconds>(), std::atomic<bool>{false});
     } catch (const std::ios_base::failure& e) {
     }
     {

test/functional/feature_bip68_sequence.py

@@ -28,7 +28,10 @@ class BIP68Test(BitcoinTestFramework):
     def set_test_params(self):
         self.num_nodes = 2
         self.extra_args = [
-            ["-acceptnonstdtxn=1"],
+            [
+                "-acceptnonstdtxn=1",
+                "-peertimeout=9999",  # bump because mocktime might cause a disconnect otherwise
+            ],
             ["-acceptnonstdtxn=0"],
         ]
 

test/functional/feature_maxuploadtarget.py

@@ -35,7 +35,13 @@ class MaxUploadTest(BitcoinTestFramework):
     def set_test_params(self):
         self.setup_clean_chain = True
         self.num_nodes = 1
-        self.extra_args = [["-maxuploadtarget=200", "-blockmaxsize=999000", "-maxtipage="+str(2*60*60*24*7), "-acceptnonstdtxn=1"]]
+        self.extra_args = [[
+            "-maxuploadtarget=200",
+            "-blockmaxsize=999000",
+            "-peertimeout=9999",  # bump because mocktime might cause a disconnect otherwise
+            "-maxtipage="+str(2*60*60*24*7),
+            "-acceptnonstdtxn=1"
+        ]]
         self.supports_cli = False
 
         # Cache for utxos, as the listunspent may take a long time later in the test

test/functional/p2p_ping.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+# Copyright (c) 2020 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 ping message
+"""
+
+import time
+
+from test_framework.messages import (
+    msg_pong,
+)
+from test_framework.p2p import P2PInterface, wait_until
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import assert_equal
+
+PING_INTERVAL = 2 * 60
+
+
+class msg_pong_corrupt(msg_pong):
+    def serialize(self):
+        return b""
+
+
+class NodePongAdd1(P2PInterface):
+    def on_ping(self, message):
+        self.send_message(msg_pong(message.nonce + 1))
+
+
+class NodeNoPong(P2PInterface):
+    def on_ping(self, message):
+        pass
+
+
+class PingPongTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.setup_clean_chain = True
+        self.num_nodes = 1
+        self.extra_args = [['-peertimeout=3']]
+
+    def check_peer_info(self, *, pingtime, minping, pingwait):
+        stats = self.nodes[0].getpeerinfo()[0]
+        assert_equal(stats.pop('pingtime', None), pingtime)
+        assert_equal(stats.pop('minping', None), minping)
+        assert_equal(stats.pop('pingwait', None), pingwait)
+
+    def mock_forward(self, delta):
+        self.mock_time += delta
+        self.nodes[0].setmocktime(self.mock_time)
+
+    def run_test(self):
+        self.mock_time = int(time.time())
+        self.mock_forward(0)
+
+        self.log.info('Check that ping is sent after connection is established')
+        no_pong_node = self.nodes[0].add_p2p_connection(NodeNoPong())
+        self.mock_forward(3)
+        assert no_pong_node.last_message.pop('ping').nonce != 0
+        self.check_peer_info(pingtime=None, minping=None, pingwait=3)
+
+        self.log.info('Reply without nonce cancels ping')
+        with self.nodes[0].assert_debug_log(['pong peer=0: Short payload']):
+            no_pong_node.send_and_ping(msg_pong_corrupt())
+        self.check_peer_info(pingtime=None, minping=None, pingwait=None)
+
+        self.log.info('Reply without ping')
+        with self.nodes[0].assert_debug_log([
+                'pong peer=0: Unsolicited pong without ping, 0 expected, 0 received, 8 bytes',
+        ]):
+            no_pong_node.send_and_ping(msg_pong())
+        self.check_peer_info(pingtime=None, minping=None, pingwait=None)
+
+        self.log.info('Reply with wrong nonce does not cancel ping')
+        assert 'ping' not in no_pong_node.last_message
+        with self.nodes[0].assert_debug_log(['pong peer=0: Nonce mismatch']):
+            # mock time PING_INTERVAL ahead to trigger node into sending a ping
+            self.mock_forward(PING_INTERVAL + 1)
+            wait_until(lambda: 'ping' in no_pong_node.last_message)
+            self.mock_forward(9)
+            # Send the wrong pong
+            no_pong_node.send_and_ping(msg_pong(no_pong_node.last_message.pop('ping').nonce - 1))
+        self.check_peer_info(pingtime=None, minping=None, pingwait=9)
+
+        self.log.info('Reply with zero nonce does cancel ping')
+        with self.nodes[0].assert_debug_log(['pong peer=0: Nonce zero']):
+            no_pong_node.send_and_ping(msg_pong(0))
+        self.check_peer_info(pingtime=None, minping=None, pingwait=None)
+
+        self.log.info('Check that ping is properly reported on RPC')
+        assert 'ping' not in no_pong_node.last_message
+        # mock time PING_INTERVAL ahead to trigger node into sending a ping
+        self.mock_forward(PING_INTERVAL + 1)
+        wait_until(lambda: 'ping' in no_pong_node.last_message)
+        ping_delay = 29
+        self.mock_forward(ping_delay)
+        wait_until(lambda: 'ping' in no_pong_node.last_message)
+        no_pong_node.send_and_ping(msg_pong(no_pong_node.last_message.pop('ping').nonce))
+        # TODO this check doesn't work due to partial 18638
+        # re-enable it after #19499 is done
+        # self.check_peer_info(pingtime=ping_delay, minping=ping_delay, pingwait=None)
+
+        self.log.info('Check that minping is decreased after a fast roundtrip')
+        # mock time PING_INTERVAL ahead to trigger node into sending a ping
+        self.mock_forward(PING_INTERVAL + 1)
+        wait_until(lambda: 'ping' in no_pong_node.last_message)
+        ping_delay = 9
+        self.mock_forward(ping_delay)
+        wait_until(lambda: 'ping' in no_pong_node.last_message)
+        no_pong_node.send_and_ping(msg_pong(no_pong_node.last_message.pop('ping').nonce))
+        # TODO this check doesn't work due to partial 18638
+        # re-enable it after #19499 is done
+        # self.check_peer_info(pingtime=ping_delay, minping=ping_delay, pingwait=None)
+
+        self.log.info('Check that peer is disconnected after ping timeout')
+        assert 'ping' not in no_pong_node.last_message
+        self.nodes[0].ping()
+        wait_until(lambda: 'ping' in no_pong_node.last_message)
+        with self.nodes[0].assert_debug_log(['ping timeout: 1201.000000s']):
+            self.mock_forward(20 * 60 + 1)
+            time.sleep(4)  # peertimeout + 1
+
+
+if __name__ == '__main__':
+    PingPongTest().main()

test/functional/test_runner.py

@@ -272,6 +272,7 @@ BASE_SCRIPTS = [
     'mempool_compatibility.py',
     'rpc_deriveaddresses.py',
     'rpc_deriveaddresses.py --usecli',
+    'p2p_ping.py',
     'rpc_scantxoutset.py',
     'feature_logging.py',
     'feature_coinstatsindex.py',