Skip to content

Commit

Permalink
sync: modernize CSemaphore / CSemaphoreGrant
Browse files Browse the repository at this point in the history
  • Loading branch information
@sipa
sipa committed Oct 2, 2023
1 parent c73cd42 commit 4d265d0
Show file tree
Hide file tree
Showing 4 changed files with 65 additions and 35 deletions.
23 changes: 11 additions & 12 deletions src/net.cpp
View file
Expand Up @@ -1862,7 +1862,7 @@ bool CConnman::AddConnection(const std::string& address, ConnectionType conn_typ
CSemaphoreGrant grant(*semOutbound, true);
if (!grant) return false;

OpenNetworkConnection(CAddress(), false, &grant, address.c_str(), conn_type, /*use_v2transport=*/false);
OpenNetworkConnection(CAddress(), false, std::move(grant), address.c_str(), conn_type, /*use_v2transport=*/false);
return true;
}

Expand Down Expand Up @@ -2294,9 +2294,9 @@ void CConnman::ProcessAddrFetch()
m_addr_fetches.pop_front();
}
CAddress addr;
CSemaphoreGrant grant(*semOutbound, true);
CSemaphoreGrant grant(*semOutbound, /*fTry=*/true);
if (grant) {
OpenNetworkConnection(addr, false, &grant, strDest.c_str(), ConnectionType::ADDR_FETCH, /*use_v2transport=*/false);
OpenNetworkConnection(addr, false, std::move(grant), strDest.c_str(), ConnectionType::ADDR_FETCH, /*use_v2transport=*/false);
}
}

Expand Down Expand Up @@ -2398,7 +2398,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect)
for (const std::string& strAddr : connect)
{
CAddress addr(CService(), NODE_NONE);
OpenNetworkConnection(addr, false, nullptr, strAddr.c_str(), ConnectionType::MANUAL, /*use_v2transport=*/false);
OpenNetworkConnection(addr, false, {}, strAddr.c_str(), ConnectionType::MANUAL, /*use_v2transport=*/false);
for (int i = 0; i < 10 && i < nLoop; i++)
{
if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
Expand Down Expand Up @@ -2703,7 +2703,7 @@ void CConnman::ThreadOpenConnections(const std::vector<std::string> connect)
const bool count_failures{((int)outbound_ipv46_peer_netgroups.size() + outbound_privacy_network_peers) >= std::min(nMaxConnections - 1, 2)};
// Use BIP324 transport when both us and them have NODE_V2_P2P set.
const bool use_v2transport(addrConnect.nServices & GetLocalServices() & NODE_P2P_V2);
OpenNetworkConnection(addrConnect, count_failures, &grant, /*strDest=*/nullptr, conn_type, use_v2transport);
OpenNetworkConnection(addrConnect, count_failures, std::move(grant), /*strDest=*/nullptr, conn_type, use_v2transport);
}
}
}
Expand Down Expand Up @@ -2785,16 +2785,16 @@ void CConnman::ThreadOpenAddedConnections()
bool tried = false;
for (const AddedNodeInfo& info : vInfo) {
if (!info.fConnected) {
if (!grant.TryAcquire()) {
if (!grant) {
// If we've used up our semaphore and need a new one, let's not wait here since while we are waiting
// the addednodeinfo state might change.
break;
}
tried = true;
CAddress addr(CService(), NODE_NONE);
OpenNetworkConnection(addr, false, &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;
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;
grant = CSemaphoreGrant(*semAddnode, /*fTry=*/true);
}
}
// Retry every 60 seconds if a connection was attempted, otherwise two seconds
Expand All @@ -2804,7 +2804,7 @@ void CConnman::ThreadOpenAddedConnections()
}

// if successful, this moves the passed grant to the constructed node
void CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant *grantOutbound, const char *pszDest, ConnectionType conn_type, bool use_v2transport)
void CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant&& grant_outbound, const char *pszDest, ConnectionType conn_type, bool use_v2transport)
{
AssertLockNotHeld(m_unused_i2p_sessions_mutex);
assert(conn_type != ConnectionType::INBOUND);
Expand All @@ -2830,8 +2830,7 @@ void CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFai

if (!pnode)
return;
if (grantOutbound)
grantOutbound->MoveTo(pnode->grantOutbound);
pnode->grantOutbound = std::move(grant_outbound);

m_msgproc->InitializeNode(*pnode, nLocalServices);
{
Expand Down
2 changes: 1 addition & 1 deletion src/net.h
View file
Expand Up @@ -1107,7 +1107,7 @@ class CConnman
bool GetNetworkActive() const { return fNetworkActive; };
bool GetUseAddrmanOutgoing() const { return m_use_addrman_outgoing; };
void SetNetworkActive(bool active);
void OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant* grantOutbound, const char* strDest, ConnectionType conn_type, bool use_v2transport) EXCLUSIVE_LOCKS_REQUIRED(!m_unused_i2p_sessions_mutex);
void OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant&& grant_outbound, const char* strDest, ConnectionType conn_type, bool use_v2transport) EXCLUSIVE_LOCKS_REQUIRED(!m_unused_i2p_sessions_mutex);
bool CheckIncomingNonce(uint64_t nonce);

// alias for thread safety annotations only, not defined
Expand Down
2 changes: 1 addition & 1 deletion src/rpc/net.cpp
View file
Expand Up @@ -317,7 +317,7 @@ static RPCHelpMan addnode()
if (command == "onetry")
{
CAddress addr;
connman.OpenNetworkConnection(addr, /*fCountFailure=*/false, /*grantOutbound=*/nullptr, node_arg.c_str(), ConnectionType::MANUAL, use_v2transport);
connman.OpenNetworkConnection(addr, /*fCountFailure=*/false, /*grant_outbound=*/{}, node_arg.c_str(), ConnectionType::MANUAL, use_v2transport);
return UniValue::VNULL;
}

Expand Down
73 changes: 52 additions & 21 deletions src/sync.h
View file
Expand Up @@ -301,6 +301,10 @@ inline MutexType* MaybeCheckNotHeld(MutexType* m) LOCKS_EXCLUDED(m) LOCK_RETURNE
//! gcc and the -Wreturn-stack-address flag in clang, both enabled by default.
#define WITH_LOCK(cs, code) (MaybeCheckNotHeld(cs), [&]() -> decltype(auto) { LOCK(cs); code; }())

/** An implementation of a semaphore.
*
* See https://en.wikipedia.org/wiki/Semaphore_(programming)
*/
class CSemaphore
{
private:
Expand All @@ -309,25 +313,33 @@ class CSemaphore
int value;

public:
explicit CSemaphore(int init) : value(init) {}
explicit CSemaphore(int init) noexcept : value(init) {}

void wait()
// Disallow default construct, copy, move.
CSemaphore() = delete;
CSemaphore(const CSemaphore&) = delete;
CSemaphore(CSemaphore&&) = delete;
CSemaphore& operator=(const CSemaphore&) = delete;
CSemaphore& operator=(CSemaphore&&) = delete;

void wait() noexcept
{
std::unique_lock<std::mutex> lock(mutex);
condition.wait(lock, [&]() { return value >= 1; });
value--;
}

bool try_wait()
bool try_wait() noexcept
{
std::lock_guard<std::mutex> lock(mutex);
if (value < 1)
if (value < 1) {
return false;
}
value--;
return true;
}

void post()
void post() noexcept
{
{
std::lock_guard<std::mutex> lock(mutex);
Expand All @@ -345,53 +357,72 @@ class CSemaphoreGrant
bool fHaveGrant;

public:
void Acquire()
void Acquire() noexcept
{
if (fHaveGrant)
if (fHaveGrant) {
return;
}
sem->wait();
fHaveGrant = true;
}

void Release()
void Release() noexcept
{
if (!fHaveGrant)
if (!fHaveGrant) {
return;
}
sem->post();
fHaveGrant = false;
}

bool TryAcquire()
bool TryAcquire() noexcept
{
if (!fHaveGrant && sem->try_wait())
if (!fHaveGrant && sem->try_wait()) {
fHaveGrant = true;
}
return fHaveGrant;
}

void MoveTo(CSemaphoreGrant& grant)
// Disallow copy.
CSemaphoreGrant(const CSemaphoreGrant&) = delete;
CSemaphoreGrant& operator=(const CSemaphoreGrant&) = delete;

// Allow move.
CSemaphoreGrant(CSemaphoreGrant&& other) noexcept
{
sem = other.sem;
fHaveGrant = other.fHaveGrant;
other.fHaveGrant = false;
other.sem = nullptr;
}

CSemaphoreGrant& operator=(CSemaphoreGrant&& other) noexcept
{
grant.Release();
grant.sem = sem;
grant.fHaveGrant = fHaveGrant;
fHaveGrant = false;
Release();
sem = other.sem;
fHaveGrant = other.fHaveGrant;
other.fHaveGrant = false;
other.sem = nullptr;
return *this;
}

CSemaphoreGrant() : sem(nullptr), fHaveGrant(false) {}
CSemaphoreGrant() noexcept : sem(nullptr), fHaveGrant(false) {}

explicit CSemaphoreGrant(CSemaphore& sema, bool fTry = false) : sem(&sema), fHaveGrant(false)
explicit CSemaphoreGrant(CSemaphore& sema, bool fTry = false) noexcept : sem(&sema), fHaveGrant(false)
{
if (fTry)
if (fTry) {
TryAcquire();
else
} else {
Acquire();
}
}

~CSemaphoreGrant()
{
Release();
}

operator bool() const
explicit operator bool() const noexcept
{
return fHaveGrant;
}
Expand Down

0 comments on commit 4d265d0

Please sign in to comment.