Skip to content

Commit

Permalink
merge bitcoin#24157: Replace RecursiveMutex cs_totalBytesSent with Mu…
Browse files Browse the repository at this point in the history 
…tex and rename it
  • Loading branch information
@kwvg
kwvg committed May 9, 2024
1 parent de4b4bf commit a1f005e
Show file tree
Hide file tree
Showing 2 changed files with 55 additions and 27 deletions.
36 changes: 29 additions & 7 deletions src/net.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1806,6 +1806,8 @@ void CConnman::SocketEvents(const std::vector<CNode*>& nodes,

void CConnman::SocketHandler(CMasternodeSync& mn_sync)
{
AssertLockNotHeld(m_total_bytes_sent_mutex);

std::set<SOCKET> recv_set;
std::set<SOCKET> send_set;
std::set<SOCKET> error_set;
Expand Down Expand Up @@ -1865,6 +1867,8 @@ void CConnman::SocketHandlerConnected(const std::set<SOCKET>& recv_set,
const std::set<SOCKET>& send_set,
const std::set<SOCKET>& error_set)
{
AssertLockNotHeld(m_total_bytes_sent_mutex);

if (interruptNet) return;

std::vector<CNode*> vErrorNodes;
Expand Down Expand Up @@ -2098,6 +2102,8 @@ size_t CConnman::SocketRecvData(CNode *pnode)

void CConnman::ThreadSocketHandler(CMasternodeSync& mn_sync)
{
AssertLockNotHeld(m_total_bytes_sent_mutex);

int64_t nLastCleanupNodes = 0;

while (!interruptNet)
Expand Down Expand Up @@ -3293,6 +3299,7 @@ bool CConnman::InitBinds(
bool CConnman::Start(CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman, CMasternodeSync& mn_sync,
CScheduler& scheduler, const Options& connOptions)
{
AssertLockNotHeld(m_total_bytes_sent_mutex);
Init(connOptions);

#ifdef USE_KQUEUE
Expand Down Expand Up @@ -3954,7 +3961,9 @@ void CConnman::RecordBytesRecv(uint64_t bytes)

void CConnman::RecordBytesSent(uint64_t bytes)
{
LOCK(cs_totalBytesSent);
AssertLockNotHeld(m_total_bytes_sent_mutex);
LOCK(m_total_bytes_sent_mutex);

nTotalBytesSent += bytes;
statsClient.count("bandwidth.bytesSent", bytes, 0.01f);
statsClient.gauge("bandwidth.totalBytesSent", nTotalBytesSent, 0.01f);
Expand All @@ -3972,7 +3981,8 @@ void CConnman::RecordBytesSent(uint64_t bytes)

uint64_t CConnman::GetMaxOutboundTarget() const
{
LOCK(cs_totalBytesSent);
AssertLockNotHeld(m_total_bytes_sent_mutex);
LOCK(m_total_bytes_sent_mutex);
return nMaxOutboundLimit;
}

Expand All @@ -3983,7 +3993,15 @@ std::chrono::seconds CConnman::GetMaxOutboundTimeframe() const

std::chrono::seconds CConnman::GetMaxOutboundTimeLeftInCycle() const
{
LOCK(cs_totalBytesSent);
AssertLockNotHeld(m_total_bytes_sent_mutex);
LOCK(m_total_bytes_sent_mutex);
return GetMaxOutboundTimeLeftInCycle_();
}

std::chrono::seconds CConnman::GetMaxOutboundTimeLeftInCycle_() const
{
AssertLockHeld(m_total_bytes_sent_mutex);

if (nMaxOutboundLimit == 0)
return 0s;

Expand All @@ -3997,14 +4015,15 @@ std::chrono::seconds CConnman::GetMaxOutboundTimeLeftInCycle() const

bool CConnman::OutboundTargetReached(bool historicalBlockServingLimit) const
{
LOCK(cs_totalBytesSent);
AssertLockNotHeld(m_total_bytes_sent_mutex);
LOCK(m_total_bytes_sent_mutex);
if (nMaxOutboundLimit == 0)
return false;

if (historicalBlockServingLimit)
{
// keep a large enough buffer to at least relay each block once
const std::chrono::seconds timeLeftInCycle = GetMaxOutboundTimeLeftInCycle();
const std::chrono::seconds timeLeftInCycle = GetMaxOutboundTimeLeftInCycle_();
const uint64_t buffer = timeLeftInCycle / std::chrono::minutes{10} * MaxBlockSize(fDIP0001ActiveAtTip);
if (buffer >= nMaxOutboundLimit || nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit - buffer)
return true;
Expand All @@ -4017,7 +4036,8 @@ bool CConnman::OutboundTargetReached(bool historicalBlockServingLimit) const

uint64_t CConnman::GetOutboundTargetBytesLeft() const
{
LOCK(cs_totalBytesSent);
AssertLockNotHeld(m_total_bytes_sent_mutex);
LOCK(m_total_bytes_sent_mutex);
if (nMaxOutboundLimit == 0)
return 0;

Expand All @@ -4031,7 +4051,8 @@ uint64_t CConnman::GetTotalBytesRecv() const

uint64_t CConnman::GetTotalBytesSent() const
{
LOCK(cs_totalBytesSent);
AssertLockNotHeld(m_total_bytes_sent_mutex);
LOCK(m_total_bytes_sent_mutex);
return nTotalBytesSent;
}

Expand Down Expand Up @@ -4083,6 +4104,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
{
AssertLockNotHeld(m_total_bytes_sent_mutex);
size_t nMessageSize = msg.data.size();
LogPrint(BCLog::NET, "sending %s (%d bytes) peer=%d\n", SanitizeString(msg.m_type), nMessageSize, pnode->GetId());
if (gArgs.GetBoolArg("-capturemessages", false)) {
Expand Down
46 changes: 26 additions & 20 deletions src/net.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -857,7 +857,10 @@ friend class CNode;
bool m_i2p_accept_incoming;
};

void Init(const Options& connOptions) {
void Init(const Options& connOptions) EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex)
{
AssertLockNotHeld(m_total_bytes_sent_mutex);

nLocalServices = connOptions.nLocalServices;
nMaxConnections = connOptions.nMaxConnections;
m_max_outbound_full_relay = std::min(connOptions.m_max_outbound_full_relay, connOptions.nMaxConnections);
Expand All @@ -873,7 +876,7 @@ friend class CNode;
nReceiveFloodSize = connOptions.nReceiveFloodSize;
m_peer_connect_timeout = std::chrono::seconds{connOptions.m_peer_connect_timeout};
{
LOCK(cs_totalBytesSent);
LOCK(m_total_bytes_sent_mutex);
nMaxOutboundLimit = connOptions.nMaxOutboundLimit;
}
vWhitelistedRange = connOptions.vWhitelistedRange;
Expand All @@ -888,7 +891,7 @@ friend class CNode;
CConnman(uint64_t seed0, uint64_t seed1, CAddrMan& addrman, bool network_active = true);
~CConnman();
bool Start(CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman, CMasternodeSync& mn_sync,
CScheduler& scheduler, const Options& options);
CScheduler& scheduler, const Options& options) EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

void StopThreads();
void StopNodes();
Expand Down Expand Up @@ -956,7 +959,7 @@ friend class CNode;

bool IsMasternodeOrDisconnectRequested(const CService& addr);

void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
void PushMessage(CNode* pnode, CSerializedNetMsg&& msg) EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

template<typename Condition, typename Callable>
bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand Down Expand Up @@ -1141,24 +1144,22 @@ friend class CNode;
//! that peer during `net_processing.cpp:PushNodeVersion()`.
ServiceFlags GetLocalServices() const;

uint64_t GetMaxOutboundTarget() const;
uint64_t GetMaxOutboundTarget() const EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);
std::chrono::seconds GetMaxOutboundTimeframe() const;

//! check if the outbound target is reached
//! if param historicalBlockServingLimit is set true, the function will
//! response true if the limit for serving historical blocks has been reached
bool OutboundTargetReached(bool historicalBlockServingLimit) const;
bool OutboundTargetReached(bool historicalBlockServingLimit) const EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

//! response the bytes left in the current max outbound cycle
//! in case of no limit, it will always response 0
uint64_t GetOutboundTargetBytesLeft() const;
uint64_t GetOutboundTargetBytesLeft() const EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

//! returns the time left in the current max outbound cycle
//! in case of no limit, it will always return 0
std::chrono::seconds GetMaxOutboundTimeLeftInCycle() const;
std::chrono::seconds GetMaxOutboundTimeLeftInCycle() const EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

uint64_t GetTotalBytesRecv() const;
uint64_t GetTotalBytesSent() const;
uint64_t GetTotalBytesSent() const EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

/** Get a unique deterministic randomizer. */
CSipHasher GetDeterministicRandomizer(uint64_t id) const;
Expand Down Expand Up @@ -1209,6 +1210,10 @@ friend class CNode;
NetPermissionFlags m_permissions;
};

//! returns the time left in the current max outbound cycle
//! in case of no limit, it will always return 0
std::chrono::seconds GetMaxOutboundTimeLeftInCycle_() const EXCLUSIVE_LOCKS_REQUIRED(m_total_bytes_sent_mutex);

bool BindListenPort(const CService& bindAddr, bilingual_str& strError, NetPermissionFlags permissions);
bool Bind(const CService& addr, unsigned int flags, NetPermissionFlags permissions);
bool InitBinds(
Expand Down Expand Up @@ -1300,7 +1305,7 @@ friend class CNode;
/**
* Check connected and listening sockets for IO readiness and process them accordingly.
*/
void SocketHandler(CMasternodeSync& mn_sync);
void SocketHandler(CMasternodeSync& mn_sync) EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

/**
* Do the read/write for connected sockets that are ready for IO.
Expand All @@ -1310,15 +1315,16 @@ friend class CNode;
*/
void SocketHandlerConnected(const std::set<SOCKET>& recv_set,
const std::set<SOCKET>& send_set,
const std::set<SOCKET>& error_set);
const std::set<SOCKET>& error_set)
EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

/**
* Accept incoming connections, one from each read-ready listening socket.
* @param[in] recv_set Sockets that are ready for read.
*/
void SocketHandlerListening(const std::set<SOCKET>& recv_set, CMasternodeSync& mn_sync);

void ThreadSocketHandler(CMasternodeSync& mn_sync);
void ThreadSocketHandler(CMasternodeSync& mn_sync) EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);
void ThreadDNSAddressSeed();
void ThreadOpenMasternodeConnections(CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman,
CMasternodeSync& mn_sync);
Expand Down Expand Up @@ -1350,7 +1356,7 @@ friend class CNode;

// Network stats
void RecordBytesRecv(uint64_t bytes);
void RecordBytesSent(uint64_t bytes);
void RecordBytesSent(uint64_t bytes) EXCLUSIVE_LOCKS_REQUIRED(!m_total_bytes_sent_mutex);

/**
* Return vector of current BLOCK_RELAY peers.
Expand All @@ -1364,14 +1370,14 @@ friend class CNode;
void UnregisterEvents(CNode* pnode);

// Network usage totals
mutable RecursiveMutex cs_totalBytesSent;
mutable Mutex m_total_bytes_sent_mutex;
std::atomic<uint64_t> nTotalBytesRecv{0};
uint64_t nTotalBytesSent GUARDED_BY(cs_totalBytesSent) {0};
uint64_t nTotalBytesSent GUARDED_BY(m_total_bytes_sent_mutex) {0};

// outbound limit & stats
uint64_t nMaxOutboundTotalBytesSentInCycle GUARDED_BY(cs_totalBytesSent) {0};
std::chrono::seconds nMaxOutboundCycleStartTime GUARDED_BY(cs_totalBytesSent) {0};
uint64_t nMaxOutboundLimit GUARDED_BY(cs_totalBytesSent);
uint64_t nMaxOutboundTotalBytesSentInCycle GUARDED_BY(m_total_bytes_sent_mutex) {0};
std::chrono::seconds nMaxOutboundCycleStartTime GUARDED_BY(m_total_bytes_sent_mutex) {0};
uint64_t nMaxOutboundLimit GUARDED_BY(m_total_bytes_sent_mutex);

// P2P timeout in seconds
std::chrono::seconds m_peer_connect_timeout;
Expand Down

0 comments on commit a1f005e

Please sign in to comment.