From ca0816152d91c929b51b2f644fcb9a69797a49fa Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Thu, 10 May 2012 18:44:07 +0200 Subject: [PATCH] Use semaphores instead of condition variables --- src/net.cpp | 69 ++++++++++++++++++++++++++++++++++------------------------ src/net.h | 2 + src/util.h | 17 +------------ 3 files changed, 44 insertions(+), 44 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 92b4a31..88a3b43 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -35,7 +35,7 @@ void ThreadOpenAddedConnections2(void* parg); void ThreadMapPort2(void* parg); #endif void ThreadDNSAddressSeed2(void* parg); -bool OpenNetworkConnection(const CAddress& addrConnect); +bool OpenNetworkConnection(const CAddress& addrConnect, bool fUseGrant = true); @@ -64,10 +64,7 @@ map mapAlreadyAskedFor; set setservAddNodeAddresses; CCriticalSection cs_setservAddNodeAddresses; -static CWaitableCriticalSection csOutbound; -static int nOutbound = 0; -static CConditionVariable condOutbound; - +static CSemaphore *semOutbound = NULL; unsigned short GetListenPort() { @@ -368,10 +365,6 @@ CNode* ConnectNode(CAddress addrConnect, int64 nTimeout) LOCK(cs_vNodes); vNodes.push_back(pnode); } - { - WAITABLE_LOCK(csOutbound); - nOutbound++; - } pnode->nTimeConnected = GetTime(); return pnode; @@ -517,14 +510,9 @@ void ThreadSocketHandler2(void* parg) // remove from vNodes vNodes.erase(remove(vNodes.begin(), vNodes.end(), pnode), vNodes.end()); - if (!pnode->fInbound) - { - WAITABLE_LOCK(csOutbound); - nOutbound--; - - // Connection slot(s) were removed, notify connection creator(s) - NOTIFY(condOutbound); - } + if (pnode->fHasGrant) + semOutbound->post(); + pnode->fHasGrant = false; // close socket and cleanup pnode->CloseSocketDisconnect(); @@ -1201,7 +1189,7 @@ void ThreadOpenConnections2(void* parg) { CAddress addr(CService(strAddr, GetDefaultPort(), fAllowDNS)); if (addr.IsValid()) - OpenNetworkConnection(addr); + OpenNetworkConnection(addr, false); for (int i = 0; i < 10 && i < nLoop; i++) { Sleep(500); @@ -1222,13 +1210,9 @@ void ThreadOpenConnections2(void* parg) if (fShutdown) return; - // Limit outbound connections - int nMaxOutbound = min(MAX_OUTBOUND_CONNECTIONS, (int)GetArg("-maxconnections", 125)); + vnThreadsRunning[THREAD_OPENCONNECTIONS]--; - { - WAITABLE_LOCK(csOutbound); - WAIT(condOutbound, fShutdown || nOutbound < nMaxOutbound); - } + semOutbound->wait(); vnThreadsRunning[THREAD_OPENCONNECTIONS]++; if (fShutdown) return; @@ -1261,11 +1245,15 @@ void ThreadOpenConnections2(void* parg) // Only connect to one address per a.b.?.? range. // Do this here so we don't have to critsect vNodes inside mapAddresses critsect. + int nOutbound = 0; set > setConnected; { LOCK(cs_vNodes); - BOOST_FOREACH(CNode* pnode, vNodes) + BOOST_FOREACH(CNode* pnode, vNodes) { setConnected.insert(pnode->addr.GetGroup()); + if (!pnode->fInbound) + nOutbound++; + } } int64 nANow = GetAdjustedTime(); @@ -1296,6 +1284,8 @@ void ThreadOpenConnections2(void* parg) if (addrConnect.IsValid()) OpenNetworkConnection(addrConnect); + else + semOutbound->post(); } } @@ -1358,6 +1348,7 @@ void ThreadOpenAddedConnections2(void* parg) } BOOST_FOREACH(vector& vserv, vservConnectAddresses) { + semOutbound->wait(); OpenNetworkConnection(CAddress(*(vserv.begin()))); Sleep(500); if (fShutdown) @@ -1373,7 +1364,14 @@ void ThreadOpenAddedConnections2(void* parg) } } -bool OpenNetworkConnection(const CAddress& addrConnect) +bool static ReleaseGrant(bool fUseGrant) { + if (fUseGrant) + semOutbound->post(); + return false; +} + +// only call this function when semOutbound has been waited for +bool OpenNetworkConnection(const CAddress& addrConnect, bool fUseGrant) { // // Initiate outbound network connection @@ -1382,7 +1380,7 @@ bool OpenNetworkConnection(const CAddress& addrConnect) return false; if ((CNetAddr)addrConnect == (CNetAddr)addrLocalHost || !addrConnect.IsIPv4() || FindNode((CNetAddr)addrConnect) || CNode::IsBanned(addrConnect)) - return false; + return ReleaseGrant(fUseGrant); vnThreadsRunning[THREAD_OPENCONNECTIONS]--; CNode* pnode = ConnectNode(addrConnect); @@ -1390,7 +1388,13 @@ bool OpenNetworkConnection(const CAddress& addrConnect) if (fShutdown) return false; if (!pnode) - return false; + return ReleaseGrant(fUseGrant); + if (pnode->fHasGrant) { + // node already has connection grant, release the one that was passed to us + ReleaseGrant(fUseGrant); + } else { + pnode->fHasGrant = fUseGrant; + } pnode->fNetworkNode = true; return true; @@ -1567,6 +1571,12 @@ bool BindListenPort(string& strError) void StartNode(void* parg) { + if (semOutbound == NULL) { + // initialize semaphore + int nMaxOutbound = min(MAX_OUTBOUND_CONNECTIONS, (int)GetArg("-maxconnections", 125)); + semOutbound = new CSemaphore(nMaxOutbound); + } + #ifdef USE_UPNP #if USE_UPNP fUseUPnP = GetBoolArg("-upnp", true); @@ -1693,7 +1703,8 @@ bool StopNode() fShutdown = true; nTransactionsUpdated++; int64 nStart = GetTime(); - NOTIFY_ALL(condOutbound); + for (int i=0; ipost(); do { int nThreadsRunning = 0; diff --git a/src/net.h b/src/net.h index bad49a9..4cc82b2 100644 --- a/src/net.h +++ b/src/net.h @@ -127,6 +127,7 @@ public: bool fNetworkNode; bool fSuccessfullyConnected; bool fDisconnect; + bool fHasGrant; // whether to call semOutbound.post() at disconnect protected: int nRefCount; @@ -171,6 +172,7 @@ public: nVersion = 0; strSubVer = ""; fClient = false; // set by version message + fHasGrant = false; fInbound = fInboundIn; fNetworkNode = false; fSuccessfullyConnected = false; diff --git a/src/util.h b/src/util.h index 5f8d037..f90b95b 100644 --- a/src/util.h +++ b/src/util.h @@ -23,7 +23,7 @@ typedef int pid_t; /* define for windows compatiblity */ #include #include #include -#include +#include #include #include #include @@ -270,24 +270,10 @@ public: }; typedef CMutexLock CCriticalBlock; -typedef CMutexLock CWaitableCriticalBlock; -typedef boost::interprocess::interprocess_condition CConditionVariable; - -/** Wait for a given condition inside a WAITABLE_CRITICAL_BLOCK */ -#define WAIT(name,condition) \ - do { while(!(condition)) { (name).wait(waitablecriticalblock.GetLock()); } } while(0) - -/** Notify waiting threads that a condition may hold now */ -#define NOTIFY(name) \ - do { (name).notify_one(); } while(0) - -#define NOTIFY_ALL(name) \ - do { (name).notify_all(); } while(0) #define LOCK(cs) CCriticalBlock criticalblock(cs, #cs, __FILE__, __LINE__) #define LOCK2(cs1,cs2) CCriticalBlock criticalblock1(cs1, #cs1, __FILE__, __LINE__),criticalblock2(cs2, #cs2, __FILE__, __LINE__) #define TRY_LOCK(cs,name) CCriticalBlock name(cs, #cs, __FILE__, __LINE__, true) -#define WAITABLE_LOCK(cs) CWaitableCriticalBlock waitablecriticalblock(cs, #cs, __FILE__, __LINE__) #define ENTER_CRITICAL_SECTION(cs) \ { \ @@ -301,6 +287,7 @@ typedef boost::interprocess::interprocess_condition CConditionVariable; LeaveCritical(); \ } +typedef boost::interprocess::interprocess_semaphore CSemaphore; inline std::string i64tostr(int64 n) { -- 1.7.1