update deb changelog

add readwrite lock for forking

.github/workflows/endurance.yml

@@ -2,7 +2,7 @@ name: Endurance
 
 on:
   schedule:
-    - cron: '30 * * * *'
+    - cron: '0 * * * *'
 
 jobs:
 

pkg/deb/debian/keydb-tools.install

@@ -5,3 +5,4 @@ src/keydb-check-rdb	/usr/bin
 src/keydb-cli		/usr/bin
 src/keydb-server	/usr/bin
 src/keydb-sentinel	/usr/bin
+src/keydb-diagnostic-tool	/usr/bin

pkg/deb/debian_dh9/keydb-tools.install

@@ -5,3 +5,4 @@ src/keydb-check-aof	/usr/bin
 src/keydb-check-rdb	/usr/bin
 src/keydb-cli		/usr/bin
 src/keydb-sentinel	/usr/bin
+src/keydb-diagnostic-tool	/usr/bin

pkg/deb/master_changelog

@@ -1,3 +1,72 @@
+keydb (6:6.2.2-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * Acquire lock in module.cpp to fix module test break
+  * Fix usage of nanosleep() on Apple hardware
+  * Fix networking deadlock (#183)
+  * Remove redundant zfree() in keydb-benchmark
+  * Issues resolved: #378, #384
+
+-- Ben Schermel <ben@eqalpha.com>  Sat, 15 Jan 2022 20:00:37 +0000
+
+keydb (6:6.2.1-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * This release of KeyDB is at parity with Redis 6.2.6. In addition, the following changes were included:
+  * Systemd support for release packages 
+  * Fixes to buffer overflows in the `BITOP SHIFT` and `CRON` commands 
+  * Server times are now computed on a seperate thread to improve performance 
+  * Now enforces syslog identity and facility as soon as possible 
+  * Removed erroneous use of LOG_... flags when using syslog 
+  * Fixed erroneous `#endif` leading to build errors on some platforms 
+  * Fixed the incorrect counting of client connections 
+  * Issues resolved: #355, #370
+
+ -- Ben Schermel <ben@eqalpha.com>  Wed, 17 Nov 2021 20:00:37 +0000
+
+keydb (6:6.2.0-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * Removed unused command line options(-a, -d, -P, -r, -q, --csv, -l, -I, -e, --precision, --cluster, --enable-tracking)
+  * --ms option changed to --time, added --clients, --host, --port, and --threads options
+  * Added keydb-diagnostic-tool
+  * Improved test reliability
+  * Fixed memory leak in keydb-benchmark
+  * Fixed a number of race conditions
+  * Fixed OSX build issues
+  * Feature parity with Redis 6.2.3
+  * Fixed Issues: #323 #325 #276
+
+ -- Ben Schermel <ben@eqalpha.com>  Thu, 12 Aug 2021 20:00:37 +0000
+
+keydb (6:6.0.18-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * 6.0.18 - This release contains a number of improvements to stability and performance. 
+  * Additional tests were added with improvements to existing test reliability
+  * Merged Redis 6.0.10 into KeyDB
+  * Issues resolved in this release: #214, #222, #238, #257, #273, #276, #285
+
+ -- Ben Schermel <ben@eqalpha.com>  Fri, 26 Mar 2021 16:00:00 +0000
+
+keydb (6:6.0.16-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * 6.0.16 - This release contains significant performance improvements as well as fixes to the following issues: #234, #236, #233, #207, #229, #231
+  * Reduced memory consumption when Active Replication is not used
+  * Reduced CPU consumption
+  * Improved TLS performance
+  * replica-quorum config for multi-master.Intended to be used with "replica-serve-stale-data no", if at least N replicas are connected we will consider our master link up and serve data.
+  * Subkey expires performance improvement for big sets
+
+ -- Ben Schermel <ben@eqalpha.com>  Mon, 28 Sep 2020 16:00:00 +0000
+
+keydb (6:6.0.13-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * 6.0.13 New features added: KEYDB.MEXISTS, REPLICAOF REMOVE, multi-master-no-forward configuration option
+  * KEYDB.MEXISTS - return exactly which keys exist in the database not just the count
+  * REPLICAOF REMOVE - Remove specific masters in a multi master setup
+  * multi-master-no-forward configuration option - Reduce network traffic in mesh topology multimaster setups
+  * Redis 6.0.5 feature parity
+  * Fixed Issues: 194 Bad directive or wrong number of arguments error on 'lazyfree-lazy-user-del no'; 209 Timeout using modules; 210 Databases are not merged on multi-master sync; 211 rpm packages now signed; 212 rpm file permissions updated
+
+ -- Ben Schermel <ben@eqalpha.com>  Mon, 13 Jul 2020 8:00:00 +0000
+
 keydb (6:6.0.9-1distribution_placeholder) codename_placeholder; urgency=medium
 
   * 6.0.9 Addressing issues 187 - cpu lockup with subkey expire, 190 - missing sentinel binary in keydb-tools. 

pkg/rpm/generate_rpms.sh

@@ -25,7 +25,13 @@ mkdir -p $DIR/keydb_build/keydb_rpm/var/log/keydb
 
 # move binaries to bin
 rm $DIR/keydb_build/keydb_rpm/usr/bin/*
-cp $DIR/../../src/keydb-* $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-server $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-sentinel $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-cli $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-benchmark $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-check-aof $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-check-rdb $DIR/keydb_build/keydb_rpm/usr/bin/
+cp $DIR/../../src/keydb-diagnostic-tool $DIR/keydb_build/keydb_rpm/usr/bin/
 
 # update spec file with build info
 sed -i '2d' $DIR/keydb_build/keydb.spec

src/blocked.cpp

@@ -592,7 +592,6 @@ void handleClientsBlockedOnKeys(void) {
              * lookup, invalidating the first one.
              * See https://github.com/redis/redis/pull/6554. */
             serverTL->fixed_time_expire++;
-            updateCachedTime(0);
 
             /* Serve clients blocked on the key. */
             robj *o = lookupKeyWrite(rl->db,rl->key);

src/cluster.cpp

@@ -2447,7 +2447,10 @@ void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
     if (sdslen(link->sndbuf) == 0 && msglen != 0)
     {
         aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [link] {
-            connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
+            /* The connection could be timed out before this posted function executes (thanks to TCP keepalive).
+             * So check that the connection is still there before setting the write handler, otherwise you segfault */
+            if (link->conn != nullptr)
+                connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
         });
     }
 

src/module.cpp

@@ -365,11 +365,7 @@ typedef struct RedisModuleCommandFilter {
 static list *moduleCommandFilters;
 
 /* Module GIL Variables */
-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;
+static readWriteLock s_moduleGIL;
 thread_local bool g_fModuleThread = false;
 
 typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
@@ -4289,7 +4285,14 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
         if (!(flags & REDISMODULE_ARGV_NO_REPLICAS))
             call_flags |= CMD_CALL_PROPAGATE_REPL;
     }
+
+    {
+    aeAcquireLock();
+    std::unique_lock<fastlock> ul(c->lock);
     call(c,call_flags);
+    aeReleaseLock();
+    }
+
     g_pserver->replication_allowed = prev_replication_allowed;
 
     serverAssert((c->flags & CLIENT_BLOCKED) == 0);
@@ -5962,95 +5965,58 @@ 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_cAcquisitionsServer > 0;
+    return !fServerThread && aeThreadOwnsLock() && !g_fModuleThread && s_moduleGIL.hasReader();
 }
 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_cAcquisitionsServer;
+        s_moduleGIL.acquireRead();
     }
     else
     {
-        // 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++;
+        s_moduleGIL.acquireWrite(fExclusive);
     }
 }
 
 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)
     {
-        ++s_cAcquisitionsServer;
+        if (!s_moduleGIL.tryAcquireRead())
+            return 1;
     }
     else
     {
-        // only try to acquire the mutexModule in exclusive mode 
-        if (fExclusive){
-            if (!s_mutexModule.try_lock())
-                return 1;
-        }
-        ++s_cAcquisitionsModule;
-        fModuleGILWlocked++;
+        if (!s_moduleGIL.tryAcquireWrite(fExclusive))
+            return 1;
     }
     return 0;
 }
 
 void moduleReleaseGIL(int fServerThread, int fExclusive) {
-    std::unique_lock<std::mutex> lock(s_mutex);
-
     if (FModuleCallBackLock(fServerThread)) {
         return;
     }
-
+    
     if (fServerThread)
     {
-        --s_cAcquisitionsServer;
+        s_moduleGIL.releaseRead();
     }
     else
     {
-        // only try to release the mutexModule in exclusive mode
-        if (fExclusive)
-            s_mutexModule.unlock();
-        --s_cAcquisitionsModule;
-        fModuleGILWlocked--;
+        s_moduleGIL.releaseWrite(fExclusive);
     }
-    s_cv.notify_all();
 }
 
 int moduleGILAcquiredByModule(void) {
-    return fModuleGILWlocked > 0;
+    return s_moduleGIL.hasWriter();
 }
 
 

src/networking.cpp

@@ -1942,7 +1942,9 @@ void ProcessPendingAsyncWrites()
                 bool fResult = c->postFunction([](client *c) {
                     c->fPendingAsyncWriteHandler = false;
                     clientInstallWriteHandler(c);
+                    c->lock.unlock();
                     handleClientsWithPendingWrites(c->iel, g_pserver->aof_state);
+                    c->lock.lock();
                 }, false);
 
                 if (!fResult)
@@ -2461,6 +2463,8 @@ void processInputBuffer(client *c, int callFlags) {
 
         if (c->flags & CLIENT_EXECUTING_COMMAND) break;
 
+        if (c->flags & CLIENT_EXECUTING_COMMAND) break;
+
         /* Don't process input from the master while there is a busy script
          * condition on the replica. We want just to accumulate the replication
          * stream (instead of replying -BUSY like we do with other clients) and
@@ -3859,86 +3863,37 @@ 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. */
 
-    std::vector<client*> vecclients;
-    listIter li;
-    listNode *ln;
-    listRewind(g_pserver->clients, &li);
-
-    /* Update our cached time since it is used to create and update the last
-     * interaction time with clients and for other important things. */
-    updateCachedTime(0);
-
-    // 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);
+    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;
         }
-    }
-    
-    /* 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;
+        catch (...)
+        {
+            ProcessingEventsWhileBlocked = 0;
+            throw;
         }
-        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/readwritelock.h

@@ -0,0 +1,113 @@
+#pragma once
+#include <condition_variable>
+
+class readWriteLock {
+    std::mutex m_readLock;
+    std::recursive_mutex m_writeLock;
+    std::condition_variable m_cv;
+    int m_readCount = 0;
+    int m_writeCount = 0;
+    bool m_writeWaiting = false;
+public:
+    void acquireRead() {
+        std::unique_lock<std::mutex> rm(m_readLock);
+        while (m_writeCount > 0 || m_writeWaiting)
+            m_cv.wait(rm);
+        m_readCount++;
+    }
+    
+    bool tryAcquireRead() {
+        std::unique_lock<std::mutex> 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<std::mutex> 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) {
+        std::unique_lock<std::mutex> 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) {
+        std::unique_lock<std::mutex> 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<std::mutex> rm(m_readLock);
+        serverAssert(m_readCount > 0);
+        m_readCount--;
+        m_cv.notify_all();
+    }
+
+    void releaseWrite(bool exclusive = true) {
+        std::unique_lock<std::mutex> rm(m_readLock);
+        serverAssert(m_writeCount > 0);
+        if (exclusive)
+            m_writeLock.unlock();
+        m_writeCount--;
+        m_cv.notify_all();
+    }
+
+    void downgradeWrite(bool exclusive = true) {
+        std::unique_lock<std::mutex> 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() {
+        return m_readCount > 0;
+    }
+
+    bool hasWriter() {
+        return m_writeCount > 0;
+    }
+
+    bool writeWaiting() {
+        return m_writeWaiting;
+    }
+};

src/redis-benchmark.cpp

@@ -1984,7 +1984,6 @@ int main(int argc, const char **argv) {
     } while(config.loop);
     zfree(data);
 
-    zfree(data);
     if (config.redis_config != NULL) freeRedisConfig(config.redis_config);
 
     return 0;

src/server.cpp

@@ -63,9 +63,11 @@
 #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>
@@ -90,10 +92,16 @@ double R_Zero, R_PosInf, R_NegInf, R_Nan;
 /* Global vars */
 namespace GlobalHidden {
 struct redisServer server; /* Server global state */
+readWriteLock forkLock;
 }
 redisServer *g_pserver = &GlobalHidden::server;
+readWriteLock *g_forkLock = &GlobalHidden::forkLock;
 struct redisServerConst cserver;
 __thread struct redisServerThreadVars *serverTL = NULL;   // thread local server vars
+std::mutex time_thread_mutex;
+std::condition_variable time_thread_cv;
+int sleeping_threads = 0;
+void wakeTimeThread();
 
 /* Our command table.
  *
@@ -2007,7 +2015,7 @@ void databasesCron(bool fMainThread) {
  * info or not using the 'update_daylight_info' argument. Normally we update
  * such info only when calling this function from serverCron() but not when
  * calling it from call(). */
-void updateCachedTime(int update_daylight_info) {
+void updateCachedTime() {
     long long t = ustime();
     __atomic_store(&g_pserver->ustime, &t, __ATOMIC_RELAXED);
     t /= 1000;
@@ -2020,12 +2028,10 @@ void updateCachedTime(int update_daylight_info) {
      * context is safe since we will never fork() while here, in the main
      * thread. The logging function will call a thread safe version of
      * localtime that has no locks. */
-    if (update_daylight_info) {
-        struct tm tm;
-        time_t ut = g_pserver->unixtime;
-        localtime_r(&ut,&tm);
-        __atomic_store(&g_pserver->daylight_active, &tm.tm_isdst, __ATOMIC_RELAXED);
-    }
+    struct tm tm;
+    time_t ut = g_pserver->unixtime;
+    localtime_r(&ut,&tm);
+    __atomic_store(&g_pserver->daylight_active, &tm.tm_isdst, __ATOMIC_RELAXED);
 }
 
 void checkChildrenDone(void) {
@@ -2152,9 +2158,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
      * handler if we don't return here fast enough. */
     if (g_pserver->watchdog_period) watchdogScheduleSignal(g_pserver->watchdog_period);
 
-    /* Update the time cache. */
-    updateCachedTime(1);
-
     g_pserver->hz = g_pserver->config_hz;
     /* Adapt the g_pserver->hz value to the number of configured clients. If we have
      * many clients, we want to call serverCron() with an higher frequency. */
@@ -2412,7 +2415,6 @@ int serverCronLite(struct aeEventLoop *eventLoop, long long id, void *clientData
 
 void blockingOperationStarts() {
     if(!g_pserver->blocking_op_nesting++){
-        updateCachedTime(0);
         g_pserver->blocked_last_cron = g_pserver->mstime;
     }
 }
@@ -2623,6 +2625,14 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     locker.disarm();
     if (!fSentReplies)
         handleClientsWithPendingWrites(iel, aof_state);
+
+    // Scope lock_guard
+    {
+        std::lock_guard<std::mutex> lock(time_thread_mutex);
+        sleeping_threads++;
+        serverAssert(sleeping_threads <= cserver.cthreads);
+    }
+    
     /* 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();
@@ -2643,6 +2653,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*/);
+        wakeTimeThread();
     }
 }
 
@@ -2839,7 +2850,7 @@ void initMasterInfo(redisMaster *master)
 void initServerConfig(void) {
     int j;
 
-    updateCachedTime(true);
+    updateCachedTime();
     getRandomHexChars(g_pserver->runid,CONFIG_RUN_ID_SIZE);
     g_pserver->runid[CONFIG_RUN_ID_SIZE] = '\0';
     changeReplicationId();
@@ -4006,7 +4017,6 @@ void call(client *c, int flags) {
     prev_err_count = g_pserver->stat_total_error_replies;
     g_pserver->fixed_time_expire++;
 
-    updateCachedTime(0);
     incrementMvccTstamp();
     elapsedStart(&call_timer);
     try {
@@ -6211,6 +6221,63 @@ 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;
@@ -6222,7 +6289,9 @@ int redisFork(int purpose) {
 
         openChildInfoPipe();
     }
-    
+    long long startWriteLock = ustime();
+    g_forkLock->acquireWrite();
+    latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
     if ((childpid = fork()) == 0) {
         /* Child */
         g_pserver->in_fork_child = purpose;
@@ -6231,6 +6300,7 @@ int redisFork(int purpose) {
         closeChildUnusedResourceAfterFork();
     } else {
         /* Parent */
+        g_forkLock->releaseWrite();
         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. */
@@ -6534,6 +6604,47 @@ void OnTerminate()
     serverPanic("std::teminate() called");
 }
 
+void wakeTimeThread() {
+    updateCachedTime();
+    std::lock_guard<std::mutex> lock(time_thread_mutex);
+    if (sleeping_threads >= cserver.cthreads)
+        time_thread_cv.notify_one();
+    sleeping_threads--;
+    serverAssert(sleeping_threads >= 0);
+}
+
+void *timeThreadMain(void*) {
+    timespec delay;
+    delay.tv_sec = 0;
+    delay.tv_nsec = 100;
+    int cycle_count = 0;
+    g_forkLock->acquireRead();
+    while (true) {
+        {
+            std::unique_lock<std::mutex> lock(time_thread_mutex);
+            if (sleeping_threads >= cserver.cthreads) {
+                g_forkLock->releaseRead();
+                time_thread_cv.wait(lock);
+                g_forkLock->acquireRead();
+                cycle_count = 0;
+            }
+        }
+        updateCachedTime();
+        if (cycle_count == MAX_CYCLES_TO_HOLD_FORK_LOCK) {
+            g_forkLock->releaseRead();
+            g_forkLock->acquireRead();
+            cycle_count = 0;
+        }
+#if defined(__APPLE__)
+        nanosleep(&delay, nullptr);
+#else
+        clock_nanosleep(CLOCK_MONOTONIC, 0, &delay, NULL);
+#endif
+        cycle_count++;
+    }
+    g_forkLock->releaseRead();
+}
+
 void *workerThreadMain(void *parg)
 {
     int iel = (int)((int64_t)parg);
@@ -6930,6 +7041,8 @@ int main(int argc, char **argv) {
     setOOMScoreAdj(-1);
     serverAssert(cserver.cthreads > 0 && cserver.cthreads <= MAX_EVENT_LOOPS);
 
+    pthread_create(&cserver.time_thread_id, nullptr, timeThreadMain, nullptr);
+
     pthread_attr_t tattr;
     pthread_attr_init(&tattr);
     pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB

src/server.h

@@ -96,6 +96,7 @@ 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. */
@@ -733,6 +734,9 @@ 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;
@@ -1536,6 +1540,7 @@ struct redisServerConst {
     pid_t pid;                  /* Main process pid. */
     time_t stat_starttime;          /* Server start time */
     pthread_t main_thread_id;         /* Main thread id */
+    pthread_t time_thread_id;
     char *configfile;           /* Absolute config file path, or NULL */
     char *executable;           /* Absolute executable file path. */
     char **exec_argv;           /* Executable argv vector (copy). */
@@ -2112,6 +2117,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;
@@ -2495,6 +2501,7 @@ 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();
@@ -2637,7 +2644,7 @@ void resetErrorTableStats(void);
 void adjustOpenFilesLimit(void);
 void incrementErrorCount(const char *fullerr, size_t namelen);
 void closeListeningSockets(int unlink_unix_socket);
-void updateCachedTime(int update_daylight_info);
+void updateCachedTime(void);
 void resetServerStats(void);
 void activeDefragCycle(void);
 unsigned int getLRUClock(void);

src/version.h

@@ -1,4 +1,4 @@
-#define KEYDB_REAL_VERSION "255.255.255"
-#define KEYDB_VERSION_NUM 0x00ffffff
+#define KEYDB_REAL_VERSION "6.2.2"
+#define KEYDB_VERSION_NUM 0x00060202
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config
 

tests/unit/scripting.tcl

@@ -485,6 +485,21 @@ start_server {tags {"scripting"}} {
     }
 
 
+    test {EVAL with pipelined command (No crash)} {
+        r flushall
+        r config set lua-time-limit 1
+        set rd [redis_deferring_client]
+        $rd eval {for i=1,1000000 do redis.call('set', i, 'sdfdsfd') end} 0
+        $rd set testkey foo
+        $rd get testkey
+        after 1200
+        catch {r echo "foo"} err
+        assert_match {BUSY*} $err
+        $rd read
+        $rd close
+    }
+
+
     test {EVAL timeout from AOF} {
         # generate a long running script that is propagated to the AOF as script
         # make sure that the script times out during loading