phantom/bitcoin
1
0
Fork 0
mirror of https://github.com/bitcoin/bitcoin.git synced 2025-01-09 19:37:27 -03:00
Code Issues Projects Releases Packages Wiki Activity Actions

fuzz: make it possible to mock (fuzz) CThreadInterrupt

Browse source 
* 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).
This commit is contained in:
Vasil Dimov 2024-11-06 11:12:35 +01:00
parent 4deb5e2121
commit 33d6a80c92
No known key found for this signature in database
GPG key ID: 54DF06F64B55CBBF
12 changed files with 154 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
src/i2p.cpp
View file

@ -118,7 +118,7 @@ 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},
@ -126,7 +126,7 @@ Session::Session(const fs::path& private_key_file,
{
}
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}
@ -161,7 +161,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";
@ -204,7 +204,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);

14
src/i2p.h
View file

@ -63,13 +63,11 @@ public:
* 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 @@ public:
* 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 @@ private:
/**
* 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.

70
src/net.cpp
View file

@ -474,7 +474,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();
@ -2046,7 +2046,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.
@ -2063,8 +2063,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
@ -2155,7 +2156,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);
@ -2169,8 +2170,7 @@ void CConnman::ThreadSocketHandler()
{
AssertLockNotHeld(m_total_bytes_sent_mutex);
while (!interruptNet)
{
while (!m_interrupt_net->interrupted()) {
DisconnectNodes();
NotifyNumConnectionsChanged();
SocketHandler();
@ -2194,9 +2194,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)
@ -2257,7 +2258,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) {
@ -2273,13 +2274,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);
}
@ -2474,12 +2475,14 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, Spa
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();
}
}
@ -2503,8 +2506,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, Spa
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)};
@ -2519,14 +2521,16 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, Spa
ProcessAddrFetch();
if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
if (!m_interrupt_net->sleep_for(500ms)) {
return;
}
PerformReconnections();
CSemaphoreGrant 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()) {
@ -2700,8 +2704,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, Spa
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();
@ -2803,7 +2806,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, Spa
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());
@ -2914,14 +2917,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 = CSemaphoreGrant(*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;
}
}
}
@ -2934,7 +2938,7 @@ void CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFai
//
// Initiate outbound network connection
//
if (interruptNet) {
if (m_interrupt_net->interrupted()) {
return;
}
if (!fNetworkActive) {
@ -3015,13 +3019,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()) {
@ -3144,12 +3148,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);
@ -3245,7 +3255,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)
@ -3282,7 +3292,7 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
// Start threads
//
assert(m_msgproc);
interruptNet.reset();
m_interrupt_net->reset();
flagInterruptMsgProc = false;
{
@ -3358,7 +3368,7 @@ void CConnman::Interrupt()
}
condMsgProc.notify_all();
interruptNet();
(*m_interrupt_net)();
g_socks5_interrupt();
if (semOutbound) {

15
src/net.h
View file

@ -1101,8 +1101,13 @@ public:
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();
@ -1526,11 +1531,9 @@ private:
/**
* 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.

4
src/test/fuzz/connman.cpp
View file

@ -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>
@ -69,7 +70,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;

9
src/test/fuzz/i2p.cpp
View file

@ -10,6 +10,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/fs_helpers.h>
#include <util/threadinterrupt.h>
@ -34,15 +35,15 @@ FUZZ_TARGET(i2p, .init = initialize_i2p)
const fs::path private_key_path = gArgs.GetDataDirNet() / "fuzzed_i2p_private_key";
const CService addr{in6_addr(IN6ADDR_LOOPBACK_INIT), 7656};
const Proxy sam_proxy{addr, false};
CThreadInterrupt interrupt;
auto interrupt{ConsumeThreadInterrupt(fuzzed_data_provider)};
i2p::sam::Session session{private_key_path, sam_proxy, &interrupt};
i2p::sam::Session session{private_key_path, sam_proxy, interrupt};
i2p::Connection conn;
if (session.Listen(conn)) {
if (session.Accept(conn)) {
try {
(void)conn.sock->RecvUntilTerminator('\n', 10ms, interrupt, i2p::sam::MAX_MSG_SIZE);
(void)conn.sock->RecvUntilTerminator('\n', 10ms, *interrupt, i2p::sam::MAX_MSG_SIZE);
} catch (const std::runtime_error&) {
}
}
@ -52,7 +53,7 @@ FUZZ_TARGET(i2p, .init = initialize_i2p)
if (session.Connect(CService{}, conn, proxy_error)) {
try {
conn.sock->SendComplete("verack\n", 10ms, interrupt);
conn.sock->SendComplete("verack\n", 10ms, *interrupt);
} catch (const std::runtime_error&) {
}
}

1
src/test/fuzz/util/CMakeLists.txt
View file

@ -6,6 +6,7 @@ add_library(test_fuzz STATIC EXCLUDE_FROM_ALL
descriptor.cpp
mempool.cpp
net.cpp
threadinterrupt.cpp
../fuzz.cpp
../util.cpp
)

22
src/test/fuzz/util/threadinterrupt.cpp Normal file
View file

@ -0,0 +1,22 @@
// Copyright (c) 2024-present The Bitcoin Core developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
#include <test/fuzz/util.h>
#include <test/fuzz/util/threadinterrupt.h>
FuzzedThreadInterrupt::FuzzedThreadInterrupt(FuzzedDataProvider& fuzzed_data_provider)
: m_fuzzed_data_provider{fuzzed_data_provider}
{
}
bool FuzzedThreadInterrupt::interrupted() const
{
return m_fuzzed_data_provider.ConsumeBool();
}
bool FuzzedThreadInterrupt::sleep_for(Clock::duration)
{
SetMockTime(ConsumeTime(m_fuzzed_data_provider)); // Time could go backwards.
return m_fuzzed_data_provider.ConsumeBool();
}

33
src/test/fuzz/util/threadinterrupt.h Normal file
View file

@ -0,0 +1,33 @@
// Copyright (c) 2024-present 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_TEST_FUZZ_UTIL_THREADINTERRUPT_H
#define BITCOIN_TEST_FUZZ_UTIL_THREADINTERRUPT_H
#include <test/fuzz/FuzzedDataProvider.h>
#include <util/threadinterrupt.h>
#include <memory>
/**
* Mocked CThreadInterrupt that returns "randomly" whether it is interrupted and never sleeps.
*/
class FuzzedThreadInterrupt : public CThreadInterrupt
{
public:
explicit FuzzedThreadInterrupt(FuzzedDataProvider& fuzzed_data_provider);
virtual bool interrupted() const override;
virtual bool sleep_for(Clock::duration) override;
private:
FuzzedDataProvider& m_fuzzed_data_provider;
};
[[nodiscard]] inline std::shared_ptr<CThreadInterrupt> ConsumeThreadInterrupt(FuzzedDataProvider& fuzzed_data_provider)
{
return std::make_shared<FuzzedThreadInterrupt>(fuzzed_data_provider);
}
#endif // BITCOIN_TEST_FUZZ_UTIL_THREADINTERRUPT_H

12
src/test/i2p_tests.cpp
View file

@ -50,10 +50,10 @@ BOOST_AUTO_TEST_CASE(unlimited_recv)
return std::make_unique<StaticContentsSock>(std::string(i2p::sam::MAX_MSG_SIZE + 1, 'a'));
};
CThreadInterrupt interrupt;
auto interrupt{std::make_shared<CThreadInterrupt>()};
const std::optional<CService> addr{Lookup("127.0.0.1", 9000, false)};
const Proxy sam_proxy(addr.value(), false);
i2p::sam::Session session(gArgs.GetDataDirNet() / "test_i2p_private_key", sam_proxy, &interrupt);
i2p::sam::Session session(gArgs.GetDataDirNet() / "test_i2p_private_key", sam_proxy, interrupt);
{
ASSERT_DEBUG_LOG("Creating persistent SAM session");
@ -112,12 +112,12 @@ BOOST_AUTO_TEST_CASE(listen_ok_accept_fail)
// clang-format on
};
CThreadInterrupt interrupt;
auto interrupt{std::make_shared<CThreadInterrupt>()};
const CService addr{in6_addr(IN6ADDR_LOOPBACK_INIT), /*port=*/7656};
const Proxy sam_proxy(addr, false);
i2p::sam::Session session(gArgs.GetDataDirNet() / "test_i2p_private_key",
sam_proxy,
&interrupt);
interrupt);
i2p::Connection conn;
for (size_t i = 0; i < 5; ++i) {
@ -155,10 +155,10 @@ BOOST_AUTO_TEST_CASE(damaged_private_key)
"391 bytes"}}) {
BOOST_REQUIRE(WriteBinaryFile(i2p_private_key_file, file_contents));
CThreadInterrupt interrupt;
auto interrupt{std::make_shared<CThreadInterrupt>()};
const CService addr{in6_addr(IN6ADDR_LOOPBACK_INIT), /*port=*/7656};
const Proxy sam_proxy{addr, false};
i2p::sam::Session session(i2p_private_key_file, sam_proxy, &interrupt);
i2p::sam::Session session(i2p_private_key_file, sam_proxy, interrupt);
{
ASSERT_DEBUG_LOG("Creating persistent SAM session");

7
src/util/threadinterrupt.cpp
View file

@ -9,11 +9,16 @@
CThreadInterrupt::CThreadInterrupt() : flag(false) {}
CThreadInterrupt::operator bool() const
bool CThreadInterrupt::interrupted() const
{
return flag.load(std::memory_order_acquire);
}
CThreadInterrupt::operator bool() const
{
return interrupted();
}
void CThreadInterrupt::reset()
{
flag.store(false, std::memory_order_release);

24
src/util/threadinterrupt.h
View file

@ -27,11 +27,27 @@ class CThreadInterrupt
{
public:
using Clock = std::chrono::steady_clock;
CThreadInterrupt();
explicit operator bool() const;
void operator()() EXCLUSIVE_LOCKS_REQUIRED(!mut);
void reset();
bool sleep_for(Clock::duration rel_time) EXCLUSIVE_LOCKS_REQUIRED(!mut);
virtual ~CThreadInterrupt() = default;
/// Return true if `operator()()` has been called.
virtual bool interrupted() const;
/// An alias for `interrupted()`.
virtual explicit operator bool() const;
/// Interrupt any sleeps. After this `interrupted()` will return `true`.
virtual void operator()() EXCLUSIVE_LOCKS_REQUIRED(!mut);
/// Reset to an non-interrupted state.
virtual void reset();
/// Sleep for the given duration.
/// @retval true The time passed.
/// @retval false The sleep was interrupted.
virtual bool sleep_for(Clock::duration rel_time) EXCLUSIVE_LOCKS_REQUIRED(!mut);
private:
std::condition_variable cond;