fuzz: make it possible to mock (fuzz) CThreadInterrupt

* Make the methods of `CThreadInterrupt` virtual and store a pointer to
  it in `CConnman`, thus making it possible to override with a mocked
  instance.
* Initialize `CConnman::m_interrupt_net` from the constructor, making it
  possible for callers to supply mocked version.
* Introduce `FuzzedThreadInterrupt` and `ConsumeThreadInterrupt()` and
  use them in `src/test/fuzz/connman.cpp` and `src/test/fuzz/i2p.cpp`.

This improves the CPU utilization of the `connman` fuzz test.

As a nice side effect, the `std::shared_ptr` used for
`CConnman::m_interrupt_net` resolves the possible lifetime issues with
it (see the removed comment for that variable).

Author: vasild

src/i2p.cpp

@@ -119,15 +119,15 @@ namespace sam {
 
 Session::Session(const fs::path& private_key_file,
                  const Proxy& control_host,
-                 CThreadInterrupt* interrupt)
+                 std::shared_ptr<CThreadInterrupt> interrupt)
     : m_private_key_file{private_key_file},
       m_control_host{control_host},
       m_interrupt{interrupt},
       m_transient{false}
 {
 }
 
-Session::Session(const Proxy& control_host, CThreadInterrupt* interrupt)
+Session::Session(const Proxy& control_host, std::shared_ptr<CThreadInterrupt> interrupt)
     : m_control_host{control_host},
       m_interrupt{interrupt},
       m_transient{true}
@@ -162,7 +162,7 @@ bool Session::Accept(Connection& conn)
     std::string errmsg;
     bool disconnect{false};
 
-    while (!*m_interrupt) {
+    while (!m_interrupt->interrupted()) {
         Sock::Event occurred;
         if (!conn.sock->Wait(MAX_WAIT_FOR_IO, Sock::RECV, &occurred)) {
             errmsg = "wait on socket failed";
@@ -205,7 +205,7 @@ bool Session::Accept(Connection& conn)
         return true;
     }
 
-    if (*m_interrupt) {
+    if (m_interrupt->interrupted()) {
         LogPrintLevel(BCLog::I2P, BCLog::Level::Debug, "Accept was interrupted\n");
     } else {
         LogPrintLevel(BCLog::I2P, BCLog::Level::Debug, "Error accepting%s: %s\n", disconnect ? " (will close the session)" : "", errmsg);

src/i2p.h

@@ -63,13 +63,11 @@ class Session
      * private key will be generated and saved into the file.
      * @param[in] control_host Location of the SAM proxy.
      * @param[in,out] interrupt If this is signaled then all operations are canceled as soon as
-     * possible and executing methods throw an exception. Notice: only a pointer to the
-     * `CThreadInterrupt` object is saved, so it must not be destroyed earlier than this
-     * `Session` object.
+     * possible and executing methods throw an exception.
      */
     Session(const fs::path& private_key_file,
             const Proxy& control_host,
-            CThreadInterrupt* interrupt);
+            std::shared_ptr<CThreadInterrupt> interrupt);
 
     /**
      * Construct a transient session which will generate its own I2P private key
@@ -78,11 +76,9 @@ class Session
      * the session will be lazily created later when first used.
      * @param[in] control_host Location of the SAM proxy.
      * @param[in,out] interrupt If this is signaled then all operations are canceled as soon as
-     * possible and executing methods throw an exception. Notice: only a pointer to the
-     * `CThreadInterrupt` object is saved, so it must not be destroyed earlier than this
-     * `Session` object.
+     * possible and executing methods throw an exception.
      */
-    Session(const Proxy& control_host, CThreadInterrupt* interrupt);
+    Session(const Proxy& control_host, std::shared_ptr<CThreadInterrupt> interrupt);
 
     /**
      * Destroy the session, closing the internally used sockets. The sockets that have been
@@ -235,7 +231,7 @@ class Session
     /**
      * Cease network activity when this is signaled.
      */
-    CThreadInterrupt* const m_interrupt;
+    const std::shared_ptr<CThreadInterrupt> m_interrupt;
 
     /**
      * Mutex protecting the members that can be concurrently accessed.

src/net.cpp

@@ -471,7 +471,7 @@ CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCo
                         LOCK(m_unused_i2p_sessions_mutex);
                         if (m_unused_i2p_sessions.empty()) {
                             i2p_transient_session =
-                                std::make_unique<i2p::sam::Session>(proxy, &interruptNet);
+                                std::make_unique<i2p::sam::Session>(proxy, m_interrupt_net);
                         } else {
                             i2p_transient_session.swap(m_unused_i2p_sessions.front());
                             m_unused_i2p_sessions.pop();
@@ -2094,7 +2094,7 @@ void CConnman::SocketHandler()
         // empty sets.
         events_per_sock = GenerateWaitSockets(snap.Nodes());
         if (events_per_sock.empty() || !events_per_sock.begin()->first->WaitMany(timeout, events_per_sock)) {
-            interruptNet.sleep_for(timeout);
+            m_interrupt_net->sleep_for(timeout);
         }
 
         // Service (send/receive) each of the already connected nodes.
@@ -2111,8 +2111,9 @@ void CConnman::SocketHandlerConnected(const std::vector<CNode*>& nodes,
     AssertLockNotHeld(m_total_bytes_sent_mutex);
 
     for (CNode* pnode : nodes) {
-        if (interruptNet)
+        if (m_interrupt_net->interrupted()) {
             return;
+        }
 
         //
         // Receive
@@ -2207,7 +2208,7 @@ void CConnman::SocketHandlerConnected(const std::vector<CNode*>& nodes,
 void CConnman::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock)
 {
     for (const ListenSocket& listen_socket : vhListenSocket) {
-        if (interruptNet) {
+        if (m_interrupt_net->interrupted()) {
             return;
         }
         const auto it = events_per_sock.find(listen_socket.sock);
@@ -2221,8 +2222,7 @@ void CConnman::ThreadSocketHandler()
 {
     AssertLockNotHeld(m_total_bytes_sent_mutex);
 
-    while (!interruptNet)
-    {
+    while (!m_interrupt_net->interrupted()) {
         DisconnectNodes();
         NotifyNumConnectionsChanged();
         SocketHandler();
@@ -2246,9 +2246,10 @@ void CConnman::ThreadDNSAddressSeed()
         auto start = NodeClock::now();
         constexpr std::chrono::seconds SEEDNODE_TIMEOUT = 30s;
         LogPrintf("-seednode enabled. Trying the provided seeds for %d seconds before defaulting to the dnsseeds.\n", SEEDNODE_TIMEOUT.count());
-        while (!interruptNet) {
-            if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
+        while (!m_interrupt_net->interrupted()) {
+            if (!m_interrupt_net->sleep_for(500ms)) {
                 return;
+            }
 
             // Abort if we have spent enough time without reaching our target.
             // Giving seed nodes 30 seconds so this does not become a race against fixedseeds (which triggers after 1 min)
@@ -2309,7 +2310,7 @@ void CConnman::ThreadDNSAddressSeed()
                         // early to see if we have enough peers and can stop
                         // this thread entirely freeing up its resources
                         std::chrono::seconds w = std::min(DNSSEEDS_DELAY_FEW_PEERS, to_wait);
-                        if (!interruptNet.sleep_for(w)) return;
+                        if (!m_interrupt_net->sleep_for(w)) return;
                         to_wait -= w;
 
                         if (GetFullOutboundConnCount() >= SEED_OUTBOUND_CONNECTION_THRESHOLD) {
@@ -2325,13 +2326,13 @@ void CConnman::ThreadDNSAddressSeed()
                 }
             }
 
-            if (interruptNet) return;
+            if (m_interrupt_net->interrupted()) return;
 
             // hold off on querying seeds if P2P network deactivated
             if (!fNetworkActive) {
                 LogPrintf("Waiting for network to be reactivated before querying DNS seeds.\n");
                 do {
-                    if (!interruptNet.sleep_for(std::chrono::seconds{1})) return;
+                    if (!m_interrupt_net->sleep_for(1s)) return;
                 } while (!fNetworkActive);
             }
 
@@ -2526,12 +2527,14 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, std
                 OpenNetworkConnection(addr, false, {}, strAddr.c_str(), ConnectionType::MANUAL, /*use_v2transport=*/use_v2transport);
                 for (int i = 0; i < 10 && i < nLoop; i++)
                 {
-                    if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
+                    if (!m_interrupt_net->sleep_for(500ms)) {
                         return;
+                    }
                 }
             }
-            if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
+            if (!m_interrupt_net->sleep_for(500ms)) {
                 return;
+            }
             PerformReconnections();
         }
     }
@@ -2555,8 +2558,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, std
         LogPrintf("Fixed seeds are disabled\n");
     }
 
-    while (!interruptNet)
-    {
+    while (!m_interrupt_net->interrupted()) {
         if (add_addr_fetch) {
             add_addr_fetch = false;
             const auto& seed{SpanPopBack(seed_nodes)};
@@ -2571,14 +2573,16 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, std
 
         ProcessAddrFetch();
 
-        if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
+        if (!m_interrupt_net->sleep_for(500ms)) {
             return;
+        }
 
         PerformReconnections();
 
         CountingSemaphoreGrant<> grant(*semOutbound);
-        if (interruptNet)
+        if (m_interrupt_net->interrupted()) {
             return;
+        }
 
         const std::unordered_set<Network> fixed_seed_networks{GetReachableEmptyNetworks()};
         if (add_fixed_seeds && !fixed_seed_networks.empty()) {
@@ -2752,8 +2756,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, std
         int nTries = 0;
         const auto reachable_nets{g_reachable_nets.All()};
 
-        while (!interruptNet)
-        {
+        while (!m_interrupt_net->interrupted()) {
             if (anchor && !m_anchors.empty()) {
                 const CAddress addr = m_anchors.back();
                 m_anchors.pop_back();
@@ -2855,7 +2858,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, std
         if (addrConnect.IsValid()) {
             if (fFeeler) {
                 // Add small amount of random noise before connection to avoid synchronization.
-                if (!interruptNet.sleep_for(rng.rand_uniform_duration<CThreadInterrupt::Clock>(FEELER_SLEEP_WINDOW))) {
+                if (!m_interrupt_net->sleep_for(rng.rand_uniform_duration<CThreadInterrupt::Clock>(FEELER_SLEEP_WINDOW))) {
                     return;
                 }
                 LogDebug(BCLog::NET, "Making feeler connection to %s\n", addrConnect.ToStringAddrPort());
@@ -2966,14 +2969,15 @@ void CConnman::ThreadOpenAddedConnections()
             tried = true;
             CAddress addr(CService(), NODE_NONE);
             OpenNetworkConnection(addr, false, std::move(grant), info.m_params.m_added_node.c_str(), ConnectionType::MANUAL, info.m_params.m_use_v2transport);
-            if (!interruptNet.sleep_for(std::chrono::milliseconds(500))) return;
+            if (!m_interrupt_net->sleep_for(500ms)) return;
             grant = CountingSemaphoreGrant<>(*semAddnode, /*fTry=*/true);
         }
         // See if any reconnections are desired.
         PerformReconnections();
         // Retry every 60 seconds if a connection was attempted, otherwise two seconds
-        if (!interruptNet.sleep_for(std::chrono::seconds(tried ? 60 : 2)))
+        if (!m_interrupt_net->sleep_for(tried ? 60s : 2s)) {
             return;
+        }
     }
 }
 
@@ -2986,7 +2990,7 @@ void CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFai
     //
     // Initiate outbound network connection
     //
-    if (interruptNet) {
+    if (m_interrupt_net->interrupted()) {
         return;
     }
     if (!fNetworkActive) {
@@ -3074,13 +3078,13 @@ void CConnman::ThreadI2PAcceptIncoming()
     i2p::Connection conn;
 
     auto SleepOnFailure = [&]() {
-        interruptNet.sleep_for(err_wait);
+        m_interrupt_net->sleep_for(err_wait);
         if (err_wait < err_wait_cap) {
             err_wait += 1s;
         }
     };
 
-    while (!interruptNet) {
+    while (!m_interrupt_net->interrupted()) {
 
         if (!m_i2p_sam_session->Listen(conn)) {
             if (advertising_listen_addr && conn.me.IsValid()) {
@@ -3202,12 +3206,18 @@ void CConnman::SetNetworkActive(bool active)
     }
 }
 
-CConnman::CConnman(uint64_t nSeed0In, uint64_t nSeed1In, AddrMan& addrman_in,
-                   const NetGroupManager& netgroupman, const CChainParams& params, bool network_active)
+CConnman::CConnman(uint64_t nSeed0In,
+                   uint64_t nSeed1In,
+                   AddrMan& addrman_in,
+                   const NetGroupManager& netgroupman,
+                   const CChainParams& params,
+                   bool network_active,
+                   std::shared_ptr<CThreadInterrupt> interrupt_net)
     : addrman(addrman_in)
     , m_netgroupman{netgroupman}
     , nSeed0(nSeed0In)
     , nSeed1(nSeed1In)
+    , m_interrupt_net{interrupt_net}
     , m_params(params)
 {
     SetTryNewOutboundPeer(false);
@@ -3303,7 +3313,7 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
     Proxy i2p_sam;
     if (GetProxy(NET_I2P, i2p_sam) && connOptions.m_i2p_accept_incoming) {
         m_i2p_sam_session = std::make_unique<i2p::sam::Session>(gArgs.GetDataDirNet() / "i2p_private_key",
-                                                                i2p_sam, &interruptNet);
+                                                                i2p_sam, m_interrupt_net);
     }
 
     // Randomize the order in which we may query seednode to potentially prevent connecting to the same one every restart (and signal that we have restarted)
@@ -3340,7 +3350,7 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
     // Start threads
     //
     assert(m_msgproc);
-    interruptNet.reset();
+    m_interrupt_net->reset();
     flagInterruptMsgProc = false;
 
     {
@@ -3416,7 +3426,7 @@ void CConnman::Interrupt()
     }
     condMsgProc.notify_all();
 
-    interruptNet();
+    (*m_interrupt_net)();
     g_socks5_interrupt();
 
     if (semOutbound) {

src/net.h

@@ -1118,8 +1118,13 @@ class CConnman
         whitelist_relay = connOptions.whitelist_relay;
     }
 
-    CConnman(uint64_t seed0, uint64_t seed1, AddrMan& addrman, const NetGroupManager& netgroupman,
-             const CChainParams& params, bool network_active = true);
+    CConnman(uint64_t seed0,
+             uint64_t seed1,
+             AddrMan& addrman,
+             const NetGroupManager& netgroupman,
+             const CChainParams& params,
+             bool network_active = true,
+             std::shared_ptr<CThreadInterrupt> interrupt_net = std::make_shared<CThreadInterrupt>());
 
     ~CConnman();
 
@@ -1543,11 +1548,9 @@ class CConnman
 
     /**
      * This is signaled when network activity should cease.
-     * A pointer to it is saved in `m_i2p_sam_session`, so make sure that
-     * the lifetime of `interruptNet` is not shorter than
-     * the lifetime of `m_i2p_sam_session`.
+     * A copy of this is saved in `m_i2p_sam_session`.
      */
-    CThreadInterrupt interruptNet;
+    const std::shared_ptr<CThreadInterrupt> m_interrupt_net;
 
     /**
      * I2P SAM session.

src/test/fuzz/connman.cpp

@@ -13,6 +13,7 @@
 #include <test/fuzz/fuzz.h>
 #include <test/fuzz/util.h>
 #include <test/fuzz/util/net.h>
+#include <test/fuzz/util/threadinterrupt.h>
 #include <test/util/setup_common.h>
 #include <util/translation.h>
 
@@ -71,7 +72,8 @@ FUZZ_TARGET(connman, .init = initialize_connman)
                      addr_man,
                      netgroupman,
                      Params(),
-                     fuzzed_data_provider.ConsumeBool()};
+                     fuzzed_data_provider.ConsumeBool(),
+                     ConsumeThreadInterrupt(fuzzed_data_provider)};
 
     const uint64_t max_outbound_limit{fuzzed_data_provider.ConsumeIntegral<uint64_t>()};
     CConnman::Options options;