use fastlock for readwrite lock

src/ae.cpp

@@ -53,7 +53,6 @@
 #include "zmalloc.h"
 #include "config.h"
 #include "serverassert.h"
-#include "readwritelock.h"
 
 #ifdef USE_MUTEX
 thread_local int cOwnLock = 0;
@@ -88,7 +87,6 @@ mutex_wrapper g_lock;
 #else
 fastlock g_lock("AE (global)");
 #endif
-readWriteLock g_forkLock("Fork (global)");
 thread_local aeEventLoop *g_eventLoopThisThread = NULL;
 
 /* Include the best multiplexing layer supported by this system.
@@ -153,11 +151,8 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
         case AE_ASYNC_OP::PostFunction:
             {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (cmd.fLock) {
-                g_forkLock.releaseRead();
+            if (cmd.fLock)
                 ulock.lock();
-                g_forkLock.acquireRead();
-            }
             ((aePostFunctionProc*)cmd.proc)(cmd.clientData);
             break;
             }
@@ -165,11 +160,8 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
         case AE_ASYNC_OP::PostCppFunction:
         {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (cmd.fLock) {
-                g_forkLock.releaseRead();
+            if (cmd.fLock)
                 ulock.lock();
-                g_forkLock.acquireRead();
-            }
             (*cmd.pfn)();
 
             delete cmd.pfn;
@@ -553,11 +545,7 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
             if (te->next)
                 te->next->prev = te->prev;
             if (te->finalizerProc) {
-                if (!ulock.owns_lock()) {
-                    g_forkLock.releaseRead();
-                    ulock.lock();
-                    g_forkLock.acquireRead();
-                }
+                if (!ulock.owns_lock()) ulock.lock();
                 te->finalizerProc(eventLoop, te->clientData);
                 now = getMonotonicUs();
             }
@@ -577,11 +565,7 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
         }
 
         if (te->when <= now) {
-            if (!ulock.owns_lock()) {
-                g_forkLock.releaseRead();
-                ulock.lock();
-                g_forkLock.acquireRead();
-            }
+            if (!ulock.owns_lock()) ulock.lock();
             int retval;
 
             id = te->id;
@@ -605,11 +589,8 @@ extern "C" void ProcessEventCore(aeEventLoop *eventLoop, aeFileEvent *fe, int ma
 {
 #define LOCK_IF_NECESSARY(fe, tsmask) \
     std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock); \
-    if (!(fe->mask & tsmask)) { \
-        g_forkLock.releaseRead(); \
-        ulock.lock(); \
-        g_forkLock.acquireRead(); \
-    }
+    if (!(fe->mask & tsmask)) \
+        ulock.lock()
 
     int fired = 0; /* Number of events fired for current fd. */
 
@@ -721,11 +702,8 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
 
         if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP) {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE)) {
-                g_forkLock.releaseRead();
+            if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE))
                 ulock.lock();
-                g_forkLock.acquireRead();
-            }
             eventLoop->beforesleep(eventLoop);
         }
 
@@ -736,11 +714,8 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
         /* After sleep callback. */
         if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP) {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
-            if (!(eventLoop->aftersleepFlags & AE_SLEEP_THREADSAFE)) {
-                g_forkLock.releaseRead();
+            if (!(eventLoop->aftersleepFlags & AE_SLEEP_THREADSAFE))
                 ulock.lock();
-                g_forkLock.acquireRead();
-            }
             eventLoop->aftersleep(eventLoop);
         }
 
@@ -814,21 +789,9 @@ void setAeLockSetThreadSpinWorker(spin_worker worker)
     tl_worker = worker;
 }
 
-void aeThreadOnline()
-{
-    g_forkLock.acquireRead();
-}
-
 void aeAcquireLock()
 {
-    g_forkLock.releaseRead();
     g_lock.lock(tl_worker);
-    g_forkLock.acquireRead();
-}
-
-void aeAcquireForkLock()
-{
-    g_forkLock.upgradeWrite();
 }
 
 int aeTryAcquireLock(int fWeak)
@@ -836,21 +799,11 @@ int aeTryAcquireLock(int fWeak)
     return g_lock.try_lock(!!fWeak);
 }
 
-void aeThreadOffline()
-{
-    g_forkLock.releaseRead();
-}
-
 void aeReleaseLock()
 {
     g_lock.unlock();
 }
 
-void aeReleaseForkLock()
-{
-    g_forkLock.downgradeWrite();
-}
-
 int aeThreadOwnsLock()
 {
     return g_lock.fOwnLock();

src/ae.h

@@ -164,13 +164,9 @@ void aeSetDontWait(aeEventLoop *eventLoop, int noWait);
 void aeClosePipesForForkChild(aeEventLoop *eventLoop);
 
 void setAeLockSetThreadSpinWorker(spin_worker worker);
-void aeThreadOnline();
 void aeAcquireLock();
-void aeAcquireForkLock();
 int aeTryAcquireLock(int fWeak);
-void aeThreadOffline();
 void aeReleaseLock();
-void aeReleaseForkLock();
 int aeThreadOwnsLock();
 
 #ifdef __cplusplus

src/networking.cpp

@@ -1321,7 +1321,7 @@ void acceptOnThread(connection *conn, int flags, char *cip)
             szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
             memcpy(szT, cip, NET_IP_STR_LEN);
         }
-        int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT] {
+        int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT, fBootLoad] {
             connMarshalThread(conn);
             acceptCommonHandler(conn,flags,szT,ielTarget);
             rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);

src/readwritelock.h

@@ -10,7 +10,7 @@ class readWriteLock {
     bool m_writeWaiting = false;
 public:
     readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}
-
+    
     void acquireRead() {
         std::unique_lock<fastlock> rm(m_readLock);
         while (m_writeCount > 0 || m_writeWaiting)
@@ -45,8 +45,20 @@ public:
     }
 
     void upgradeWrite(bool exclusive = true) {
-        releaseRead();
-        acquireWrite(exclusive);
+        std::unique_lock<fastlock> rm(m_readLock);
+        m_writeWaiting = true;
+        while (m_readCount > 1)
+            m_cv.wait(rm);
+        if (exclusive) {
+            /* Another thread might have the write lock while we have the read lock
+               but won't be able to release it until they can acquire the read lock
+               so release the read lock and try again instead of waiting to avoid deadlock */
+            while(!m_writeLock.try_lock())
+                m_cv.wait(rm);
+        }
+        m_writeCount++;
+        m_readCount--;
+        m_writeWaiting = false;
     }
 
     bool tryAcquireWrite(bool exclusive = true) {
@@ -64,6 +76,7 @@ public:
 
     void releaseRead() {
         std::unique_lock<fastlock> rm(m_readLock);
+        serverAssert(m_readCount > 0);
         m_readCount--;
         m_cv.notify_all();
     }
@@ -78,8 +91,14 @@ public:
     }
 
     void downgradeWrite(bool exclusive = true) {
-        releaseWrite(exclusive);
-        acquireRead();
+        std::unique_lock<fastlock> rm(m_readLock);
+        serverAssert(m_writeCount > 0);
+        if (exclusive)
+            m_writeLock.unlock();
+        m_writeCount--;
+        while (m_writeCount > 0 || m_writeWaiting)
+            m_cv.wait(rm);
+        m_readCount++;
     }
 
     bool hasReader() {

src/server.cpp

@@ -62,6 +62,7 @@
 #include <sys/socket.h>
 #include <algorithm>
 #include <uuid/uuid.h>
+#include <mutex>
 #include <condition_variable>
 #include "aelocker.h"
 #include "motd.h"
@@ -91,12 +92,14 @@ double R_Zero, R_PosInf, R_NegInf, R_Nan;
 /* Global vars */
 namespace GlobalHidden {
 struct redisServer server; /* Server global state */
+readWriteLock forkLock("Fork (global)");
 }
 redisServer *g_pserver = &GlobalHidden::server;
+readWriteLock *g_forkLock = &GlobalHidden::forkLock;
 struct redisServerConst cserver;
 __thread struct redisServerThreadVars *serverTL = NULL;   // thread local server vars
-fastlock time_thread_lock("Time thread lock");
-std::condition_variable_any time_thread_cv;
+std::mutex time_thread_mutex;
+std::condition_variable time_thread_cv;
 int sleeping_threads = 0;
 void wakeTimeThread();
 
@@ -2623,14 +2626,14 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     if (!fSentReplies)
         handleClientsWithPendingWrites(iel, aof_state);
 
-    aeThreadOffline();
     // Scope lock_guard
     {
-        std::unique_lock<fastlock> lock(time_thread_lock);
+        std::lock_guard<std::mutex> lock(time_thread_mutex);
         sleeping_threads++;
         serverAssert(sleeping_threads <= cserver.cthreads);
     }
-    
+
+    g_forkLock->releaseRead();    
     /* Determine whether the modules are enabled before sleeping, and use that result
        both here, and after wakeup to avoid double acquire or release of the GIL */
     serverTL->modulesEnabledThisAeLoop = !!moduleCount();
@@ -2651,7 +2654,7 @@ void afterSleep(struct aeEventLoop *eventLoop) {
        Otherwise you may double acquire the GIL and cause deadlocks in the module */
     if (!ProcessingEventsWhileBlocked) {
         if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
-        aeThreadOnline();
+        g_forkLock->acquireRead();
         wakeTimeThread();
     }
 }
@@ -6289,7 +6292,8 @@ int redisFork(int purpose) {
         openChildInfoPipe();
     }
     long long startWriteLock = ustime();
-    aeAcquireForkLock();
+    g_forkLock->releaseRead();
+    executeWithoutGlobalLock([](){ g_forkLock->acquireWrite(); });
     latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
     if ((childpid = fork()) == 0) {
         /* Child */
@@ -6299,7 +6303,8 @@ int redisFork(int purpose) {
         closeChildUnusedResourceAfterFork();
     } else {
         /* Parent */
-        aeReleaseForkLock();
+        g_forkLock->releaseWrite();
+        g_forkLock->acquireRead();
         g_pserver->stat_total_forks++;
         g_pserver->stat_fork_time = ustime()-start;
         g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
@@ -6617,9 +6622,7 @@ void OnTerminate()
 
 void wakeTimeThread() {
     updateCachedTime();
-    aeThreadOffline();
-    std::unique_lock<fastlock> lock(time_thread_lock);
-    aeThreadOnline();
+    std::lock_guard<std::mutex> lock(time_thread_mutex);
     if (sleeping_threads >= cserver.cthreads)
         time_thread_cv.notify_one();
     sleeping_threads--;
@@ -6631,23 +6634,21 @@ void *timeThreadMain(void*) {
     delay.tv_sec = 0;
     delay.tv_nsec = 100;
     int cycle_count = 0;
-    aeThreadOnline();
+    g_forkLock->acquireRead();
     while (true) {
         {
-            aeThreadOffline();
-            std::unique_lock<fastlock> lock(time_thread_lock);
-            aeThreadOnline();
+            std::unique_lock<std::mutex> lock(time_thread_mutex);
             if (sleeping_threads >= cserver.cthreads) {
-                aeThreadOffline();
+                g_forkLock->releaseRead();
                 time_thread_cv.wait(lock);
-                aeThreadOnline();
+                g_forkLock->acquireRead();
                 cycle_count = 0;
             }
         }
         updateCachedTime();
         if (cycle_count == MAX_CYCLES_TO_HOLD_FORK_LOCK) {
-            aeThreadOffline();
-            aeThreadOnline();
+            g_forkLock->releaseRead();
+            g_forkLock->acquireRead();
             cycle_count = 0;
         }
 #if defined(__APPLE__)
@@ -6657,7 +6658,7 @@ void *timeThreadMain(void*) {
 #endif
         cycle_count++;
     }
-    aeThreadOffline();
+    g_forkLock->releaseRead();
 }
 
 void *workerThreadMain(void *parg)
@@ -6675,7 +6676,7 @@ void *workerThreadMain(void *parg)
     }
 
     moduleAcquireGIL(true); // Normally afterSleep acquires this, but that won't be called on the first run
-    aeThreadOnline();
+    g_forkLock->acquireRead();
     aeEventLoop *el = g_pserver->rgthreadvar[iel].el;
     try
     {
@@ -6684,7 +6685,7 @@ void *workerThreadMain(void *parg)
     catch (ShutdownException)
     {
     }
-    aeThreadOffline();
+    g_forkLock->releaseRead();
     moduleReleaseGIL(true);
     serverAssert(!GlobalLocksAcquired());
     aeDeleteEventLoop(el);
@@ -6822,8 +6823,8 @@ int main(int argc, char **argv) {
     g_pserver->sentinel_mode = checkForSentinelMode(argc,argv);
     initServerConfig();
     serverTL = &g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN];
-    aeThreadOnline();
     aeAcquireLock();    // We own the lock on boot
+
     ACLInit(); /* The ACL subsystem must be initialized ASAP because the
                   basic networking code and client creation depends on it. */
     moduleInitModulesSystem();
@@ -7053,7 +7054,6 @@ int main(int argc, char **argv) {
 
     redisSetCpuAffinity(g_pserver->server_cpulist);
     aeReleaseLock();    //Finally we can dump the lock
-    aeThreadOffline();
     moduleReleaseGIL(true);
     
     setOOMScoreAdj(-1);

src/server.h

@@ -2177,6 +2177,7 @@ typedef struct {
 
 //extern struct redisServer server;
 extern redisServer *g_pserver;
+extern readWriteLock *g_forkLock;
 extern struct redisServerConst cserver;
 extern __thread struct redisServerThreadVars *serverTL;   // thread local server vars
 extern struct sharedObjectsStruct shared;