Merge pull request #1260 from sipa/splitsync

gmaxwell · gmaxwell · commit c05271901a7a · 2012-05-11T18:05:49.000-07:00
Split synchronization mechanisms from util.{h,cpp}
diff --git a/bitcoin-qt.pro b/bitcoin-qt.pro
@@ -109,6 +109,7 @@ HEADERS += src/qt/bitcoingui.h \
     src/bignum.h \
     src/checkpoints.h \
     src/compat.h \
+    src/sync.h \
     src/util.h \
     src/uint256.h \
     src/serialize.h \
@@ -172,6 +173,7 @@ SOURCES += src/qt/bitcoin.cpp src/qt/bitcoingui.cpp \
     src/qt/editaddressdialog.cpp \
     src/qt/bitcoinaddressvalidator.cpp \
     src/version.cpp \
+    src/sync.cpp \
     src/util.cpp \
     src/netbase.cpp \
     src/key.cpp \
diff --git a/src/addrman.h b/src/addrman.h
@@ -7,6 +7,7 @@
 #include "netbase.h"
 #include "protocol.h"
 #include "util.h"
+#include "sync.h"
 
 
 #include <map>
diff --git a/src/keystore.h b/src/keystore.h
@@ -6,7 +6,7 @@
 #define BITCOIN_KEYSTORE_H
 
 #include "crypter.h"
-#include "util.h"
+#include "sync.h"
 #include "base58.h"
 
 class CScript;
diff --git a/src/main.h b/src/main.h
@@ -6,6 +6,7 @@
 #define BITCOIN_MAIN_H
 
 #include "bignum.h"
+#include "sync.h"
 #include "net.h"
 #include "script.h"
 
diff --git a/src/makefile.linux-mingw b/src/makefile.linux-mingw
@@ -61,6 +61,7 @@ OBJS= \
     obj/bitcoinrpc.o \
     obj/rpcdump.o \
     obj/script.o \
+    obj/sync.o \
     obj/util.o \
     obj/wallet.o \
     obj/walletdb.o \
diff --git a/src/makefile.mingw b/src/makefile.mingw
@@ -58,6 +58,7 @@ OBJS= \
     obj/bitcoinrpc.o \
     obj/rpcdump.o \
     obj/script.o \
+    obj/sync.o \
     obj/util.o \
     obj/wallet.o \
     obj/walletdb.o \
diff --git a/src/makefile.osx b/src/makefile.osx
@@ -85,6 +85,7 @@ OBJS= \
     obj/bitcoinrpc.o \
     obj/rpcdump.o \
     obj/script.o \
+    obj/sync.o \
     obj/util.o \
     obj/wallet.o \
     obj/walletdb.o \
diff --git a/src/makefile.unix b/src/makefile.unix
@@ -102,6 +102,7 @@ OBJS= \
     obj/bitcoinrpc.o \
     obj/rpcdump.o \
     obj/script.o \
+    obj/sync.o \
     obj/util.o \
     obj/wallet.o \
     obj/walletdb.o \
diff --git 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, const char *strDest = NULL, bool fOneShot = false);
+bool OpenNetworkConnection(const CAddress& addrConnect, CSemaphoreGrant *grantOutbound = NULL, const char *strDest = NULL, bool fOneShot = false);
 
 
 
@@ -66,10 +66,7 @@ CCriticalSection cs_vOneShots;
 set<CNetAddr> setservAddNodeAddresses;
 CCriticalSection cs_setservAddNodeAddresses;
 
-static CWaitableCriticalSection csOutbound;
-static int nOutbound = 0;
-static CConditionVariable condOutbound;
-
+static CSemaphore *semOutbound = NULL;
 
 void AddOneShot(string strDest)
 {
@@ -463,10 +460,6 @@ CNode* ConnectNode(CAddress addrConnect, const char *pszDest, int64 nTimeout)
             LOCK(cs_vNodes);
             vNodes.push_back(pnode);
         }
-        {
-            WAITABLE_LOCK(csOutbound);
-            nOutbound++;
-        }
 
         pnode->nTimeConnected = GetTime();
         return pnode;
@@ -612,14 +605,8 @@ 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);
-                        }
+                    // release outbound grant (if any)
+                    pnode->grantOutbound.Release();
 
                     // close socket and cleanup
                     pnode->CloseSocketDisconnect();
@@ -1295,8 +1282,11 @@ void static ProcessOneShot()
         vOneShots.pop_front();
     }
     CAddress addr;
-    if (!OpenNetworkConnection(addr, strDest.c_str(), true))
-        AddOneShot(strDest);
+    CSemaphoreGrant grant(*semOutbound, true);
+    if (grant) {
+        if (!OpenNetworkConnection(addr, &grant, strDest.c_str(), true))
+            AddOneShot(strDest);
+    }
 }
 
 void ThreadOpenConnections2(void* parg)
@@ -1312,7 +1302,7 @@ void ThreadOpenConnections2(void* parg)
             BOOST_FOREACH(string strAddr, mapMultiArgs["-connect"])
             {
                 CAddress addr;
-                OpenNetworkConnection(addr, strAddr.c_str());
+                OpenNetworkConnection(addr, NULL, strAddr.c_str());
                 for (int i = 0; i < 10 && i < nLoop; i++)
                 {
                     Sleep(500);
@@ -1335,13 +1325,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);
-        }
+        CSemaphoreGrant grant(*semOutbound);
         vnThreadsRunning[THREAD_OPENCONNECTIONS]++;
         if (fShutdown)
             return;
@@ -1374,11 +1360,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<vector<unsigned char> > 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();
@@ -1408,7 +1398,7 @@ void ThreadOpenConnections2(void* parg)
         }
 
         if (addrConnect.IsValid())
-            OpenNetworkConnection(addrConnect);
+            OpenNetworkConnection(addrConnect, &grant);
     }
 }
 
@@ -1442,7 +1432,8 @@ void ThreadOpenAddedConnections2(void* parg)
         while(!fShutdown) {
             BOOST_FOREACH(string& strAddNode, mapMultiArgs["-addnode"]) {
                 CAddress addr;
-                OpenNetworkConnection(addr, strAddNode.c_str());
+                CSemaphoreGrant grant(*semOutbound);
+                OpenNetworkConnection(addr, &grant, strAddNode.c_str());
                 Sleep(500);
             }
             vnThreadsRunning[THREAD_ADDEDCONNECTIONS]--;
@@ -1485,7 +1476,8 @@ void ThreadOpenAddedConnections2(void* parg)
         }
         BOOST_FOREACH(vector<CService>& vserv, vservConnectAddresses)
         {
-            OpenNetworkConnection(CAddress(*(vserv.begin())));
+            CSemaphoreGrant grant(*semOutbound);
+            OpenNetworkConnection(CAddress(*(vserv.begin())), &grant);
             Sleep(500);
             if (fShutdown)
                 return;
@@ -1500,7 +1492,8 @@ void ThreadOpenAddedConnections2(void* parg)
     }
 }
 
-bool OpenNetworkConnection(const CAddress& addrConnect, const char *strDest, bool fOneShot)
+// if succesful, this moves the passed grant to the constructed node
+bool OpenNetworkConnection(const CAddress& addrConnect, CSemaphoreGrant *grantOutbound, const char *strDest, bool fOneShot)
 {
     //
     // Initiate outbound network connection
@@ -1522,6 +1515,8 @@ bool OpenNetworkConnection(const CAddress& addrConnect, const char *strDest, boo
         return false;
     if (!pnode)
         return false;
+    if (grantOutbound)
+        grantOutbound->MoveTo(pnode->grantOutbound);
     pnode->fNetworkNode = true;
     if (fOneShot)
         pnode->fOneShot = true;
@@ -1770,6 +1765,12 @@ void StartNode(void* parg)
 #endif
 #endif
 
+    if (semOutbound == NULL) {
+        // initialize semaphore
+        int nMaxOutbound = min(MAX_OUTBOUND_CONNECTIONS, (int)GetArg("-maxconnections", 125));
+        semOutbound = new CSemaphore(nMaxOutbound);
+    }
+
     if (pnodeLocalHost == NULL)
         pnodeLocalHost = new CNode(INVALID_SOCKET, CAddress(CService("127.0.0.1", 0), nLocalServices));
 
@@ -1823,7 +1824,8 @@ bool StopNode()
     fShutdown = true;
     nTransactionsUpdated++;
     int64 nStart = GetTime();
-    NOTIFY_ALL(condOutbound);
+    for (int i=0; i<MAX_OUTBOUND_CONNECTIONS; i++)
+         semOutbound->post();
     do
     {
         int nThreadsRunning = 0;
diff --git a/src/net.h b/src/net.h
@@ -148,6 +148,7 @@ class CNode
     bool fNetworkNode;
     bool fSuccessfullyConnected;
     bool fDisconnect;
+    CSemaphoreGrant grantOutbound;
 protected:
     int nRefCount;
 
diff --git a/src/qt/messagepage.cpp b/src/qt/messagepage.cpp
@@ -10,7 +10,6 @@
 #include "main.h"
 #include "wallet.h"
 #include "init.h"
-#include "util.h"
 
 #include "messagepage.h"
 #include "ui_messagepage.h"
diff --git a/src/qt/qtipcserver.cpp b/src/qt/qtipcserver.cpp
@@ -8,7 +8,6 @@
 #include <boost/date_time/posix_time/posix_time.hpp>
 
 #include "ui_interface.h"
-#include "util.h"
 #include "qtipcserver.h"
 
 using namespace boost::interprocess;
diff --git a/src/sync.cpp b/src/sync.cpp
@@ -0,0 +1,119 @@
+// Copyright (c) 2011-2012 The Bitcoin developers
+// Distributed under the MIT/X11 software license, see the accompanying
+// file license.txt or http://www.opensource.org/licenses/mit-license.php.
+
+#include "sync.h"
+
+
+
+#ifdef DEBUG_LOCKORDER
+//
+// Early deadlock detection.
+// Problem being solved:
+//    Thread 1 locks  A, then B, then C
+//    Thread 2 locks  D, then C, then A
+//     --> may result in deadlock between the two threads, depending on when they run.
+// Solution implemented here:
+// Keep track of pairs of locks: (A before B), (A before C), etc.
+// Complain if any thread trys to lock in a different order.
+//
+
+struct CLockLocation
+{
+    CLockLocation(const char* pszName, const char* pszFile, int nLine)
+    {
+        mutexName = pszName;
+        sourceFile = pszFile;
+        sourceLine = nLine;
+    }
+
+    std::string ToString() const
+    {
+        return mutexName+"  "+sourceFile+":"+itostr(sourceLine);
+    }
+
+private:
+    std::string mutexName;
+    std::string sourceFile;
+    int sourceLine;
+};
+
+typedef std::vector< std::pair<void*, CLockLocation> > LockStack;
+
+static boost::interprocess::interprocess_mutex dd_mutex;
+static std::map<std::pair<void*, void*>, LockStack> lockorders;
+static boost::thread_specific_ptr<LockStack> lockstack;
+
+
+static void potential_deadlock_detected(const std::pair<void*, void*>& mismatch, const LockStack& s1, const LockStack& s2)
+{
+    printf("POTENTIAL DEADLOCK DETECTED\n");
+    printf("Previous lock order was:\n");
+    BOOST_FOREACH(const PAIRTYPE(void*, CLockLocation)& i, s2)
+    {
+        if (i.first == mismatch.first) printf(" (1)");
+        if (i.first == mismatch.second) printf(" (2)");
+        printf(" %s\n", i.second.ToString().c_str());
+    }
+    printf("Current lock order is:\n");
+    BOOST_FOREACH(const PAIRTYPE(void*, CLockLocation)& i, s1)
+    {
+        if (i.first == mismatch.first) printf(" (1)");
+        if (i.first == mismatch.second) printf(" (2)");
+        printf(" %s\n", i.second.ToString().c_str());
+    }
+}
+
+static void push_lock(void* c, const CLockLocation& locklocation, bool fTry)
+{
+    bool fOrderOK = true;
+    if (lockstack.get() == NULL)
+        lockstack.reset(new LockStack);
+
+    if (fDebug) printf("Locking: %s\n", locklocation.ToString().c_str());
+    dd_mutex.lock();
+
+    (*lockstack).push_back(std::make_pair(c, locklocation));
+
+    if (!fTry) BOOST_FOREACH(const PAIRTYPE(void*, CLockLocation)& i, (*lockstack))
+    {
+        if (i.first == c) break;
+
+        std::pair<void*, void*> p1 = std::make_pair(i.first, c);
+        if (lockorders.count(p1))
+            continue;
+        lockorders[p1] = (*lockstack);
+
+        std::pair<void*, void*> p2 = std::make_pair(c, i.first);
+        if (lockorders.count(p2))
+        {
+            potential_deadlock_detected(p1, lockorders[p2], lockorders[p1]);
+            break;
+        }
+    }
+    dd_mutex.unlock();
+}
+
+static void pop_lock()
+{
+    if (fDebug) 
+    {
+        const CLockLocation& locklocation = (*lockstack).rbegin()->second;
+        printf("Unlocked: %s\n", locklocation.ToString().c_str());
+    }
+    dd_mutex.lock();
+    (*lockstack).pop_back();
+    dd_mutex.unlock();
+}
+
+void EnterCritical(const char* pszName, const char* pszFile, int nLine, void* cs, bool fTry)
+{
+    push_lock(cs, CLockLocation(pszName, pszFile, nLine), fTry);
+}
+
+void LeaveCritical()
+{
+    pop_lock();
+}
+
+#endif /* DEBUG_LOCKORDER */
diff --git a/src/sync.h b/src/sync.h
diff --git a/src/util.cpp b/src/util.cpp
diff --git a/src/util.h b/src/util.h
