Merge remote-tracking branch 'origin/unstable' into mm-psync

.github/workflows/ci.yml

@@ -30,7 +30,7 @@ jobs:
           ./runtest-moduleapi
         
   build-ubuntu-old:
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-16.04
     steps:
     - uses: actions/checkout@v1
     - name: make -j2

README.md

@@ -33,13 +33,19 @@ Because of this difference of opinion features which are right for KeyDB may not
 Additional Resources
 --------------------
 
-Check out KeyDB's [Docker Image](https://hub.docker.com/r/eqalpha/keydb)
+Try our docker container: https://hub.docker.com/r/eqalpha/keydb
 
-Join us on [Slack](https://docs.keydb.dev/slack/)
+Talk on Gitter: https://gitter.im/KeyDB
 
-Post to the [Community Forum](https://community.keydb.dev)
+Visit our Website: https://keydb.dev
 
-Learn more through KeyDB's [Documentation & Learning Center](https://docs.keydb.dev)
+See options for channel partners and support contracts: https://keydb.dev/support.html
+
+Learn with KeyDB’s official documentation site: https://docs.keydb.dev
+
+[Subscribe to the KeyDB mailing list](https://eqalpha.us20.list-manage.com/subscribe/post?u=978f486c2f95589b24591a9cc&id=4ab9220500)
+
+Management GUI: We recommend [FastoNoSQL](https://fastonosql.com/) which has official KeyDB support.
 
 
 Benchmarking KeyDB

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/module.cpp

@@ -365,7 +365,11 @@ typedef struct RedisModuleCommandFilter {
 static list *moduleCommandFilters;
 
 /* Module GIL Variables */
-static readWriteLock s_moduleGIL("Module GIL");
+static int s_cAcquisitionsServer = 0;
+static int s_cAcquisitionsModule = 0;
+static std::mutex s_mutex;
+static std::condition_variable s_cv;
+static std::recursive_mutex s_mutexModule;
 thread_local bool g_fModuleThread = false;
 
 typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
@@ -5965,58 +5969,95 @@ void RM_ThreadSafeContextUnlock(RedisModuleCtx *ctx) {
 //  as the server thread acquisition is sufficient.  If we did try to lock we would deadlock
 static bool FModuleCallBackLock(bool fServerThread)
 {
-    return !fServerThread && aeThreadOwnsLock() && !g_fModuleThread && s_moduleGIL.hasReader();
+    return !fServerThread && aeThreadOwnsLock() && !g_fModuleThread && s_cAcquisitionsServer > 0;
 }
 void moduleAcquireGIL(int fServerThread, int fExclusive) {
+    std::unique_lock<std::mutex> lock(s_mutex);
+    int *pcheck = fServerThread ? &s_cAcquisitionsModule : &s_cAcquisitionsServer;
+
     if (FModuleCallBackLock(fServerThread)) {
         return;
     }
 
+    while (*pcheck > 0)
+        s_cv.wait(lock);
+
     if (fServerThread)
     {
-        s_moduleGIL.acquireRead();
+        ++s_cAcquisitionsServer;
     }
     else
     {
-        s_moduleGIL.acquireWrite(fExclusive);
+        // only try to acquire the mutexModule in exclusive mode 
+        if (fExclusive){
+            // It is possible that another module thread holds the GIL (and s_mutexModule as a result). 
+            // When said thread goes to release the GIL, it will wait for s_mutex, which this thread owns. 
+            // This thread is however waiting for the GIL (and s_mutexModule) that the other thread owns.
+            // As a result, a deadlock has occured. 
+            // We release the lock on s_mutex and wait until we are able to safely acquire the GIL 
+            // in order to prevent this deadlock from occuring. 
+            while (!s_mutexModule.try_lock())
+                s_cv.wait(lock);     
+        }    
+        ++s_cAcquisitionsModule;
+        fModuleGILWlocked++;
     }
 }
 
 int moduleTryAcquireGIL(bool fServerThread, int fExclusive) {
+    std::unique_lock<std::mutex> lock(s_mutex, std::defer_lock);
+    if (!lock.try_lock())
+        return 1;
+    int *pcheck = fServerThread ? &s_cAcquisitionsModule : &s_cAcquisitionsServer;
+
     if (FModuleCallBackLock(fServerThread)) {
         return 0;
     }
 
+    if (*pcheck > 0)
+        return 1;
+
     if (fServerThread)
     {
-        if (!s_moduleGIL.tryAcquireRead())
-            return 1;
+        ++s_cAcquisitionsServer;
     }
     else
     {
-        if (!s_moduleGIL.tryAcquireWrite(fExclusive))
-            return 1;
+        // only try to acquire the mutexModule in exclusive mode 
+        if (fExclusive){
+            if (!s_mutexModule.try_lock())
+                return 1;
+        }
+        ++s_cAcquisitionsModule;
+        fModuleGILWlocked++;
     }
     return 0;
 }
 
 void moduleReleaseGIL(int fServerThread, int fExclusive) {
+    std::unique_lock<std::mutex> lock(s_mutex);
+
     if (FModuleCallBackLock(fServerThread)) {
         return;
     }
-    
+
     if (fServerThread)
     {
-        s_moduleGIL.releaseRead();
+        --s_cAcquisitionsServer;
     }
     else
     {
-        s_moduleGIL.releaseWrite(fExclusive);
+        // only try to release the mutexModule in exclusive mode
+        if (fExclusive)
+            s_mutexModule.unlock();
+        --s_cAcquisitionsModule;
+        fModuleGILWlocked--;
     }
+    s_cv.notify_all();
 }
 
 int moduleGILAcquiredByModule(void) {
-    return s_moduleGIL.hasWriter();
+    return fModuleGILWlocked > 0;
 }
 
 

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);
@@ -1907,6 +1907,19 @@ void ProcessPendingAsyncWrites()
         zfree(c->replyAsync);
         c->replyAsync = nullptr;
         c->fPendingAsyncWrite = FALSE;
+
+        // Now install the write event handler
+        int ae_flags = AE_WRITABLE|AE_WRITE_THREADSAFE;
+        /* For the fsync=always policy, we want that a given FD is never
+            * served for reading and writing in the same event loop iteration,
+            * so that in the middle of receiving the query, and serving it
+            * to the client, we'll call beforeSleep() that will do the
+            * actual fsync of AOF to disk. AE_BARRIER ensures that. */
+        if (g_pserver->aof_state == AOF_ON &&
+            g_pserver->aof_fsync == AOF_FSYNC_ALWAYS)
+        {
+            ae_flags |= AE_BARRIER;
+        }
         
         if (!((c->replstate == REPL_STATE_NONE ||
          (c->replstate == SLAVE_STATE_ONLINE && !c->repl_put_online_on_ack))))
@@ -3848,37 +3861,82 @@ int checkClientPauseTimeoutAndReturnIfPaused(void) {
  *
  * The function returns the total number of events processed. */
 void processEventsWhileBlocked(int iel) {
+    serverAssert(GlobalLocksAcquired());
+    int iterations = 4; /* See the function top-comment. */
 
-    int eventsCount = 0;
-    executeWithoutGlobalLock([&](){
-        int iterations = 4; /* See the function top-comment. */
-        try
-        {
-            ProcessingEventsWhileBlocked = 1;
-            while (iterations--) {
-                long long startval = g_pserver->events_processed_while_blocked;
-                long long ae_events = aeProcessEvents(g_pserver->rgthreadvar[iel].el,
-                    AE_FILE_EVENTS|AE_DONT_WAIT|
-                    AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);
-                /* Note that g_pserver->events_processed_while_blocked will also get
-                * incremeted by callbacks called by the event loop handlers. */
-                eventsCount += ae_events;
-                long long events = eventsCount - startval;
-                if (!events) break;
-            }
-            ProcessingEventsWhileBlocked = 0;
+    std::vector<client*> vecclients;
+    listIter li;
+    listNode *ln;
+    listRewind(g_pserver->clients, &li);
+
+    // All client locks must be acquired *after* the global lock is reacquired to prevent deadlocks
+    //  so unlock here, and save them for reacquisition later
+    while ((ln = listNext(&li)) != nullptr)
+    {
+        client *c = (client*)listNodeValue(ln);
+        if (c->lock.fOwnLock()) {
+            serverAssert(c->flags & CLIENT_PROTECTED);  // If the client is not protected we have no gurantee they won't be free'd in the event loop
+            c->lock.unlock();
+            vecclients.push_back(c);
         }
-        catch (...)
-        {
-            ProcessingEventsWhileBlocked = 0;
-            throw;
+    }
+    
+    /* Since we're about to release our lock we need to flush the repl backlog queue */
+    bool fReplBacklog = g_pserver->repl_batch_offStart >= 0;
+    if (fReplBacklog) {
+        flushReplBacklogToClients();
+        g_pserver->repl_batch_idxStart = -1;
+        g_pserver->repl_batch_offStart = -1;
+    }
+
+    long long eventsCount = 0;
+    aeReleaseLock();
+    serverAssert(!GlobalLocksAcquired());
+    try
+    {
+        ProcessingEventsWhileBlocked = 1;
+        while (iterations--) {
+            long long startval = g_pserver->events_processed_while_blocked;
+            long long ae_events = aeProcessEvents(g_pserver->rgthreadvar[iel].el,
+                AE_FILE_EVENTS|AE_DONT_WAIT|
+                AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);
+            /* Note that g_pserver->events_processed_while_blocked will also get
+            * incremeted by callbacks called by the event loop handlers. */
+            eventsCount += ae_events;
+            long long events = eventsCount - startval;
+            if (!events) break;
         }
-    });
+        ProcessingEventsWhileBlocked = 0;
+    }
+    catch (...)
+    {
+        // Caller expects us to be locked so fix and rethrow
+        ProcessingEventsWhileBlocked = 0;
+        AeLocker locker;
+        locker.arm(nullptr);
+        locker.release();
+        for (client *c : vecclients)
+            c->lock.lock();
+        throw;
+    }
+    
+    AeLocker locker;
+    locker.arm(nullptr);
+    locker.release();
 
     g_pserver->events_processed_while_blocked += eventsCount;
 
     whileBlockedCron();
 
+    // Restore it so the calling code is not confused
+    if (fReplBacklog) {
+        g_pserver->repl_batch_idxStart = g_pserver->repl_backlog_idx;
+        g_pserver->repl_batch_offStart = g_pserver->master_repl_offset;
+    }
+
+    for (client *c : vecclients)
+        c->lock.lock();
+
     // If a different thread processed the shutdown we need to abort the lua command or we will hang
     if (serverTL->el->stop)
         throw ShutdownException();

src/object.cpp

@@ -136,7 +136,7 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
  *
  * The current limit of 52 is chosen so that the biggest string object
  * we allocate as EMBSTR will still fit into the 64 byte arena of jemalloc. */
-size_t OBJ_ENCODING_EMBSTR_SIZE_LIMIT = 52;
+#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 52
 
 //static_assert((sizeof(redisObject)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT-8) == 64, "Max EMBSTR obj should be 64 bytes total");
 robj *createStringObject(const char *ptr, size_t len) {

src/pqsort.c

@@ -41,7 +41,6 @@
 
 #include <errno.h>
 #include <stdlib.h>
-#include <stdint.h>
 
 static inline char	*med3 (char *, char *, char *,
     int (*)(const void *, const void *));
@@ -63,7 +62,7 @@ static inline void	 swapfunc (char *, char *, size_t, int);
         } while (--i > 0);				\
 }
 
-#define SWAPINIT(a, es) swaptype = (uintptr_t)a % sizeof(long) || \
+#define SWAPINIT(a, es) swaptype = ((char *)a - (char *)0) % sizeof(long) || \
 	es % sizeof(long) ? 2 : es == sizeof(long)? 0 : 1;
 
 static inline void

src/readwritelock.h

@@ -1,96 +0,0 @@
-#pragma once
-#include <condition_variable>
-
-class readWriteLock {
-    fastlock m_readLock;
-    fastlock m_writeLock;
-    std::condition_variable_any m_cv;
-    int m_readCount = 0;
-    int m_writeCount = 0;
-    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)
-            m_cv.wait(rm);
-        m_readCount++;
-    }
-    
-    bool tryAcquireRead() {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (!rm.try_lock())
-            return false;
-        if (m_writeCount > 0 || m_writeWaiting)
-            return false;
-        m_readCount++;
-        return true;
-    }
-
-    void acquireWrite(bool exclusive = true) {
-        std::unique_lock<fastlock> rm(m_readLock);
-        m_writeWaiting = true;
-        while (m_readCount > 0)
-            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_writeWaiting = false;
-    }
-
-    void upgradeWrite(bool exclusive = true) {
-        releaseRead();
-        acquireWrite(exclusive);
-    }
-
-    bool tryAcquireWrite(bool exclusive = true) {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (!rm.try_lock())
-            return false;
-        if (m_readCount > 0)
-            return false;
-        if (exclusive)
-            if (!m_writeLock.try_lock())
-                return false;
-        m_writeCount++;
-        return true;
-    }
-
-    void releaseRead() {
-        std::unique_lock<fastlock> rm(m_readLock);
-        m_readCount--;
-        m_cv.notify_all();
-    }
-
-    void releaseWrite(bool exclusive = true) {
-        std::unique_lock<fastlock> rm(m_readLock);
-        serverAssert(m_writeCount > 0);
-        if (exclusive)
-            m_writeLock.unlock();
-        m_writeCount--;
-        m_cv.notify_all();
-    }
-
-    void downgradeWrite(bool exclusive = true) {
-        releaseWrite(exclusive);
-        acquireRead();
-    }
-
-    bool hasReader() {
-        return m_readCount > 0;
-    }
-
-    bool hasWriter() {
-        return m_writeCount > 0;
-    }
-
-    bool writeWaiting() {
-        return m_writeWaiting;
-    }
-};

src/server.cpp

@@ -62,11 +62,11 @@
 #include <sys/socket.h>
 #include <algorithm>
 #include <uuid/uuid.h>
+#include <mutex>
 #include <condition_variable>
 #include "aelocker.h"
 #include "motd.h"
 #include "t_nhash.h"
-#include "readwritelock.h"
 #ifdef __linux__
 #include <sys/prctl.h>
 #include <sys/mman.h>
@@ -95,8 +95,8 @@ struct redisServer server; /* Server global state */
 redisServer *g_pserver = &GlobalHidden::server;
 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,10 +2623,9 @@ 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);
     }
@@ -2650,9 +2649,8 @@ void afterSleep(struct aeEventLoop *eventLoop) {
        Don't check here that modules are enabled, rather use the result from beforeSleep
        Otherwise you may double acquire the GIL and cause deadlocks in the module */
     if (!ProcessingEventsWhileBlocked) {
-        if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
-        aeThreadOnline();
         wakeTimeThread();
+        if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
     }
 }
 
@@ -6220,63 +6218,6 @@ void closeChildUnusedResourceAfterFork() {
     cserver.pidfile = NULL;
 }
 
-void executeWithoutGlobalLock(std::function<void()> func) {
-    serverAssert(GlobalLocksAcquired());
-
-    std::vector<client*> vecclients;
-    listIter li;
-    listNode *ln;
-    listRewind(g_pserver->clients, &li);
-
-    // All client locks must be acquired *after* the global lock is reacquired to prevent deadlocks
-    //  so unlock here, and save them for reacquisition later
-    while ((ln = listNext(&li)) != nullptr)
-    {
-        client *c = (client*)listNodeValue(ln);
-        if (c->lock.fOwnLock()) {
-            serverAssert(c->flags & CLIENT_PROTECTED || c->flags & CLIENT_EXECUTING_COMMAND);  // If the client is not protected we have no gurantee they won't be free'd in the event loop
-            c->lock.unlock();
-            vecclients.push_back(c);
-        }
-    }
-    
-    /* Since we're about to release our lock we need to flush the repl backlog queue */
-    bool fReplBacklog = g_pserver->repl_batch_offStart >= 0;
-    if (fReplBacklog) {
-        flushReplBacklogToClients();
-        g_pserver->repl_batch_idxStart = -1;
-        g_pserver->repl_batch_offStart = -1;
-    }
-
-    aeReleaseLock();
-    serverAssert(!GlobalLocksAcquired());
-    try {
-        func();
-    }
-    catch (...) {
-        // Caller expects us to be locked so fix and rethrow
-        AeLocker locker;
-        locker.arm(nullptr);
-        locker.release();
-        for (client *c : vecclients)
-            c->lock.lock();
-        throw;
-    }
-    
-    AeLocker locker;
-    locker.arm(nullptr);
-    locker.release();
-
-    // Restore it so the calling code is not confused
-    if (fReplBacklog) {
-        g_pserver->repl_batch_idxStart = g_pserver->repl_backlog_idx;
-        g_pserver->repl_batch_offStart = g_pserver->master_repl_offset;
-    }
-
-    for (client *c : vecclients)
-        c->lock.lock();
-}
-
 /* purpose is one of CHILD_TYPE_ types */
 int redisFork(int purpose) {
     int childpid;
@@ -6288,9 +6229,7 @@ int redisFork(int purpose) {
 
         openChildInfoPipe();
     }
-    long long startWriteLock = ustime();
-    aeAcquireForkLock();
-    latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
+    
     if ((childpid = fork()) == 0) {
         /* Child */
         g_pserver->in_fork_child = purpose;
@@ -6299,7 +6238,6 @@ int redisFork(int purpose) {
         closeChildUnusedResourceAfterFork();
     } else {
         /* Parent */
-        aeReleaseForkLock();
         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 +6555,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--;
@@ -6630,34 +6566,20 @@ void *timeThreadMain(void*) {
     timespec delay;
     delay.tv_sec = 0;
     delay.tv_nsec = 100;
-    int cycle_count = 0;
-    aeThreadOnline();
     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();
                 time_thread_cv.wait(lock);
-                aeThreadOnline();
-                cycle_count = 0;
             }
         }
         updateCachedTime();
-        if (cycle_count == MAX_CYCLES_TO_HOLD_FORK_LOCK) {
-            aeThreadOffline();
-            aeThreadOnline();
-            cycle_count = 0;
-        }
 #if defined(__APPLE__)
         nanosleep(&delay, nullptr);
 #else
         clock_nanosleep(CLOCK_MONOTONIC, 0, &delay, NULL);
 #endif
-        cycle_count++;
     }
-    aeThreadOffline();
 }
 
 void *workerThreadMain(void *parg)
@@ -6675,7 +6597,6 @@ void *workerThreadMain(void *parg)
     }
 
     moduleAcquireGIL(true); // Normally afterSleep acquires this, but that won't be called on the first run
-    aeThreadOnline();
     aeEventLoop *el = g_pserver->rgthreadvar[iel].el;
     try
     {
@@ -6684,7 +6605,6 @@ void *workerThreadMain(void *parg)
     catch (ShutdownException)
     {
     }
-    aeThreadOffline();
     moduleReleaseGIL(true);
     serverAssert(!GlobalLocksAcquired());
     aeDeleteEventLoop(el);
@@ -6822,8 +6742,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 +6973,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

@@ -96,7 +96,6 @@ typedef long long ustime_t; /* microsecond time type. */
 #include "connection.h" /* Connection abstraction */
 #include "serverassert.h"
 #include "expire.h"
-#include "readwritelock.h"
 
 #define REDISMODULE_CORE 1
 #include "redismodule.h"    /* Redis modules API defines. */
@@ -734,9 +733,6 @@ typedef enum {
 #define REDISMODULE_AUX_BEFORE_RDB (1<<0)
 #define REDISMODULE_AUX_AFTER_RDB (1<<1)
 
-/* Number of cycles before time thread gives up fork lock */
-#define MAX_CYCLES_TO_HOLD_FORK_LOCK 10
-
 struct RedisModule;
 struct RedisModuleIO;
 struct RedisModuleDigest;
@@ -2561,7 +2557,6 @@ void sendChildInfo(childInfoType info_type, size_t keys, const char *pname);
 void receiveChildInfo(void);
 
 /* Fork helpers */
-void executeWithoutGlobalLock(std::function<void()> func);
 int redisFork(int type);
 int hasActiveChildProcess();
 void resetChildState();

src/storage.cpp

@@ -114,7 +114,6 @@ void pool_free(struct alloc_pool *ppool, void *pv)
     return;
 }
 
-extern size_t OBJ_ENCODING_EMBSTR_SIZE_LIMIT;
 #define EMBSTR_ROBJ_SIZE (sizeof(robj)+sizeof(struct sdshdr8)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT+1)
 struct alloc_pool poolobj;
 struct alloc_pool poolembstrobj;

src/zmalloc.cpp

@@ -243,7 +243,7 @@ void *ztryrealloc_usable(void *ptr, size_t size, size_t *usable) {
 #else
     realptr = (char*)ptr-PREFIX_SIZE;
     oldsize = *((size_t*)realptr);
-    newptr = realloc(realptr,size+PREFIX_SIZE, MALLOC_LOCAL);
+    newptr = realloc(realptr,size+PREFIX_SIZE);
     if (newptr == NULL) {
         if (usable) *usable = 0;
         return NULL;

tests/assets/minimal.conf

@@ -1,5 +1,4 @@
 # Minimal configuration for testing.
-bind 127.0.0.1
 always-show-logo yes
 daemonize no
 pidfile /var/run/keydb.pid

tests/cluster/run.tcl

@@ -13,9 +13,8 @@ set ::tlsdir "../../tls"
 proc main {} {
     parse_options
     spawn_instance redis $::redis_base_port $::instances_count {
-	"bind 127.0.0.1"
-	"cluster-enabled yes"
-	"appendonly yes"
+        "cluster-enabled yes"
+        "appendonly yes"
 	"testmode yes"
 	"server-threads 3"
     }

tests/modules/Makefile

@@ -32,7 +32,6 @@ TEST_MODULES = \
     auth.so \
     keyspace_events.so \
     blockedclient.so \
-    timers.so \
     getkeys.so \
     test_lazyfree.so \
     timer.so \