update deb changelog

update release packaging

.github/workflows/endurance.yml

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

pkg/deb/master_changelog

@@ -1,3 +1,62 @@
+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. 

src/aof.cpp

@@ -879,7 +879,7 @@ int loadAppendOnlyFile(char *filename) {
     } else {
         /* RDB preamble. Pass loading the RDB functions. */
         rio rdb;
-        rdbSaveInfo rsi;
+        rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
 
         serverLog(LL_NOTICE,"Reading RDB preamble from AOF file...");
         if (fseek(fp,0,SEEK_SET) == -1) goto readerr;

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);
@@ -4285,14 +4289,7 @@ 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);
@@ -5965,58 +5962,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

@@ -1942,9 +1942,7 @@ 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)
@@ -2463,6 +2461,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
@@ -3861,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/rdb.cpp

@@ -1205,40 +1205,6 @@ int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
             == -1) return -1;
         if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",g_pserver->master_repl_offset)
             == -1) return -1;
-        if (g_pserver->fActiveReplica && listLength(g_pserver->masters) > 0) {
-            sdsstring val = sdsstring(sdsempty());
-            listNode *ln;
-            listIter li;
-            redisMaster* mi;
-            listRewind(g_pserver->masters,&li);
-            while ((ln = listNext(&li)) != NULL) {
-                mi = (redisMaster*)listNodeValue(ln);
-                if (!mi->master) {
-                    // If master client is not available, use info from master struct - better than nothing
-                    serverLog(LL_NOTICE, "saving master %s", mi->master_replid);
-                    if (mi->master_replid[0] == 0) {
-                        // if replid is null, there's no reason to save it
-                        continue;
-                    }
-                    val = val.catfmt("%s:%I:%s:%i;", mi->master_replid,
-                        mi->master_initial_offset,
-                        mi->masterhost,
-                        mi->masterport);
-                }
-                else {
-                    serverLog(LL_NOTICE, "saving master %s", mi->master->replid);
-                    if (mi->master->replid[0] == 0) {
-                        // if replid is null, there's no reason to save it
-                        continue;
-                    }
-                    val = val.catfmt("%s:%I:%s:%i;", mi->master->replid,
-                        mi->master->reploff,
-                        mi->masterhost,
-                        mi->masterport);
-                }
-            }
-            if (rdbSaveAuxFieldStrStr(rdb, "repl-masters",val.get()) == -1) return -1;
-        }
     }
     if (rdbSaveAuxFieldStrInt(rdb,"aof-preamble",aof_preamble) == -1) return -1;
     return 1;
@@ -2687,23 +2653,6 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
                     memcpy(rsi->repl_id,ptrFromObj(auxval),CONFIG_RUN_ID_SIZE+1);
                     rsi->repl_id_is_set = 1;
                 }
-            } else if (!strcasecmp(szFromObj(auxkey),"repl-masters")) {
-                if (rsi) {
-                    struct redisMaster mi;
-                    char *masters = szFromObj(auxval);
-                    char *entry = strtok(masters, ":");
-                    while (entry != NULL) {
-                        memcpy(mi.master_replid, entry, sizeof(mi.master_replid));
-                        entry = strtok(NULL, ":");
-                        mi.master_initial_offset = atoi(entry);
-                        entry = strtok(NULL, ":");
-                        mi.masterhost = entry;
-                        entry = strtok(NULL, ";");
-                        mi.masterport = atoi(entry);
-                        entry = strtok(NULL, ":");
-                        rsi->addMaster(mi);
-                    }
-                }
             } else if (!strcasecmp(szFromObj(auxkey),"repl-offset")) {
                 if (rsi) rsi->repl_offset = strtoll(szFromObj(auxval),NULL,10);
             } else if (!strcasecmp(szFromObj(auxkey),"lua")) {
@@ -2868,11 +2817,11 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
             bool fExpiredKey = iAmMaster() && !(rdbflags&RDBFLAGS_AOF_PREAMBLE) && expiretime != -1 && expiretime < now;
         
             if (fStaleMvccKey || fExpiredKey) {
-                if (fStaleMvccKey && !fExpiredKey && rsi != nullptr && rsi->masters != nullptr && rsi->masters[0].staleKeyMap != nullptr && lookupKeyRead(db, &keyobj) == nullptr) {
+                if (fStaleMvccKey && !fExpiredKey && rsi != nullptr && rsi->mi != nullptr && rsi->mi->staleKeyMap != nullptr && lookupKeyRead(db, &keyobj) == nullptr) {
                     // We have a key that we've already deleted and is not back in our database.
                     //  We'll need to inform the sending master of the delete if it is also a replica of us
                     robj_sharedptr objKeyDup(createStringObject(key, sdslen(key)));
-                    rsi->masters[0].staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);                
+                    rsi->mi->staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);                
                 }
                 fLastKeyExpired = true;
                 sdsfree(key);
@@ -3297,8 +3246,8 @@ void bgsaveCommand(client *c) {
  * is returned, and the RDB saving will not persist any replication related
  * information. */
 rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi) {
-    rdbSaveInfo rsi_init;
-    *rsi = std::move(rsi_init);
+    rdbSaveInfo rsi_init = RDB_SAVE_INFO_INIT;
+    *rsi = rsi_init;
 
     /* If the instance is a master, we can populate the replication info
      * only when repl_backlog is not NULL. If the repl_backlog is NULL,

src/readwritelock.h

@@ -1,115 +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) {
-        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) {
-        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);
-        serverAssert(m_readCount > 0);
-        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) {
-        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() {
-        return m_readCount > 0;
-    }
-
-    bool hasWriter() {
-        return m_writeCount > 0;
-    }
-
-    bool writeWaiting() {
-        return m_writeWaiting;
-    }
-};

src/redis-benchmark.cpp

@@ -1984,6 +1984,7 @@ 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/replication.cpp

@@ -1987,7 +1987,7 @@ void readSyncBulkPayload(connection *conn) {
     ssize_t nread, readlen, nwritten;
     int use_diskless_load = useDisklessLoad();
     dbBackup *diskless_load_backup = NULL;
-    rdbSaveInfo rsi;
+    rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
     int empty_db_flags = g_pserver->repl_slave_lazy_flush ? EMPTYDB_ASYNC :
                                                         EMPTYDB_NO_FLAGS;
     off_t left;
@@ -2308,7 +2308,7 @@ void readSyncBulkPayload(connection *conn) {
                 mi->staleKeyMap->clear();
             else
                 mi->staleKeyMap = new (MALLOC_LOCAL) std::map<int, std::vector<robj_sharedptr>>();
-            rsi.addMaster(*mi);
+            rsi.mi = mi;
         }
         if (rdbLoadFile(rdb_filename,&rsi,RDBFLAGS_REPLICATION) != C_OK) {
             serverLog(LL_WARNING,
@@ -2359,7 +2359,7 @@ void readSyncBulkPayload(connection *conn) {
     {
         mergeReplicationId(mi->master->replid);
     }
-    else if (!g_pserver->fActiveReplica)
+    else
     {
         /* After a full resynchroniziation we use the replication ID and
         * offset of the master. The secondary ID / offset are cleared since
@@ -2555,7 +2555,7 @@ int slaveTryPartialResynchronization(redisMaster *mi, connection *conn, int read
          * client structure representing the master into g_pserver->master. */
         mi->master_initial_offset = -1;
 
-        if (mi->cached_master) {
+        if (mi->cached_master && !g_pserver->fActiveReplica) {
             psync_replid = mi->cached_master->replid;
             snprintf(psync_offset,sizeof(psync_offset),"%lld", mi->cached_master->reploff+1);
             serverLog(LL_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_replid, psync_offset);
@@ -2654,15 +2654,14 @@ int slaveTryPartialResynchronization(redisMaster *mi, connection *conn, int read
                     sizeof(g_pserver->replid2));
                 g_pserver->second_replid_offset = g_pserver->master_repl_offset+1;
 
-                if (!g_pserver->fActiveReplica) {
-                    /* Update the cached master ID and our own primary ID to the
-                     * new one. */
-                    memcpy(g_pserver->replid,sznew,sizeof(g_pserver->replid));
-                    memcpy(mi->cached_master->replid,sznew,sizeof(g_pserver->replid));
+                /* Update the cached master ID and our own primary ID to the
+                 * new one. */
+                memcpy(g_pserver->replid,sznew,sizeof(g_pserver->replid));
+                memcpy(mi->cached_master->replid,sznew,sizeof(g_pserver->replid));
 
-                    /* Disconnect all the sub-slaves: they need to be notified. */
+                /* Disconnect all the sub-slaves: they need to be notified. */
+                if (!g_pserver->fActiveReplica)
                     disconnectSlaves();
-                }
             }
         }
 
@@ -3023,6 +3022,18 @@ retry_connect:
         disconnectSlavesExcept(mi->master_uuid); /* Force our slaves to resync with us as well. */
         freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
     }
+    else
+    {
+        if (listLength(g_pserver->slaves))
+        {
+            changeReplicationId();
+            clearReplicationId2();
+        }
+        else
+        {
+            freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
+        }
+    }
 
     /* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC
      * and the g_pserver->master_replid and master_initial_offset are
@@ -3668,26 +3679,6 @@ void replicationCacheMasterUsingMyself(redisMaster *mi) {
     mi->master = NULL;
 }
 
-/* This function is called when reloading master info from an RDB in Active Replica mode.
- * It creates a cached master client using the info contained in the redisMaster struct.
- *
- * Assumes that the passed struct contains valid master info. */
-void replicationCacheMasterUsingMaster(redisMaster *mi) {
-    if (mi->cached_master) {
-        freeClient(mi->cached_master);
-    }
-
-    replicationCreateMasterClient(mi, NULL, -1);
-    std::lock_guard<decltype(mi->master->lock)> lock(mi->master->lock);
-
-    memcpy(mi->master->replid, mi->master_replid, sizeof(mi->master_replid));
-    mi->master->reploff = mi->master_initial_offset;
-
-    unlinkClient(mi->master);
-    mi->cached_master = mi->master;
-    mi->master = NULL;
-}
-
 /* Free a cached master, called when there are no longer the conditions for
  * a partial resync on reconnection. */
 void replicationDiscardCachedMaster(redisMaster *mi) {

src/sds.h

@@ -424,21 +424,6 @@ public:
         return *this;
     }
 
-    sdsstring &operator=(sdsstring &&other)
-    {
-        sds tmp = m_str;
-        m_str = other.m_str;
-        other.m_str = tmp;
-        return *this;
-    }
-
-    template<typename... Args>
-    sdsstring catfmt(const char *fmt, Args... args)
-    {
-        m_str = sdscatfmt(m_str, fmt, args...);
-        return *this;
-    }
-
     sds release() {
         sds sdsT = m_str;
         m_str = nullptr;

src/server.cpp

@@ -67,7 +67,6 @@
 #include "aelocker.h"
 #include "motd.h"
 #include "t_nhash.h"
-#include "readwritelock.h"
 #ifdef __linux__
 #include <sys/prctl.h>
 #include <sys/mman.h>
@@ -92,10 +91,8 @@ 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
 std::mutex time_thread_mutex;
@@ -2632,8 +2629,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
         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();
@@ -2653,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*/);
-        g_forkLock->acquireRead();
         wakeTimeThread();
+        if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
     }
 }
 
@@ -6223,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;
@@ -6291,10 +6229,7 @@ int redisFork(int purpose) {
 
         openChildInfoPipe();
     }
-    long long startWriteLock = ustime();
-    g_forkLock->releaseRead();
-    executeWithoutGlobalLock([](){ g_forkLock->acquireWrite(); });
-    latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
+    
     if ((childpid = fork()) == 0) {
         /* Child */
         g_pserver->in_fork_child = purpose;
@@ -6303,8 +6238,6 @@ int redisFork(int purpose) {
         closeChildUnusedResourceAfterFork();
     } else {
         /* Parent */
-        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. */
@@ -6367,7 +6300,7 @@ void loadDataFromDisk(void) {
         if (loadAppendOnlyFile(g_pserver->aof_filename) == C_OK)
             serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
     } else if (g_pserver->rdb_filename != NULL || g_pserver->rdb_s3bucketpath != NULL) {
-        rdbSaveInfo rsi;
+        rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
         errno = 0; /* Prevent a stale value from affecting error checking */
         if (rdbLoad(&rsi,RDBFLAGS_NONE) == C_OK) {
             serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds",
@@ -6395,23 +6328,11 @@ void loadDataFromDisk(void) {
                 while ((ln = listNext(&li)))
                 {
                     redisMaster *mi = (redisMaster*)listNodeValue(ln);
-                    if (g_pserver->fActiveReplica) {
-                        for (size_t i = 0; i < rsi.numMasters(); i++) {
-                            if (!strcmp(mi->masterhost, rsi.masters[i].masterhost) && mi->masterport == rsi.masters[i].masterport) {
-                                memcpy(mi->master_replid, rsi.masters[i].master_replid, sizeof(mi->master_replid));
-                                mi->master_initial_offset = rsi.masters[i].master_initial_offset;
-                                replicationCacheMasterUsingMaster(mi);
-                                serverLog(LL_NOTICE, "Cached master recovered from RDB for %s:%d", mi->masterhost, mi->masterport);
-                            }
-                        }
-                    }
-                    else {
-                        /* If we are a replica, create a cached master from this
-                        * information, in order to allow partial resynchronizations
-                        * with masters. */
-                        replicationCacheMasterUsingMyself(mi);
-                        selectDb(mi->cached_master,rsi.repl_stream_db);
-                    }
+                    /* If we are a replica, create a cached master from this
+                    * information, in order to allow partial resynchronizations
+                    * with masters. */
+                    replicationCacheMasterUsingMyself(mi);
+                    selectDb(mi->cached_master,rsi.repl_stream_db);
                 }
             }
         } else if (errno != ENOENT) {
@@ -6633,32 +6554,16 @@ 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)
@@ -6676,7 +6581,6 @@ void *workerThreadMain(void *parg)
     }
 
     moduleAcquireGIL(true); // Normally afterSleep acquires this, but that won't be called on the first run
-    g_forkLock->acquireRead();
     aeEventLoop *el = g_pserver->rgthreadvar[iel].el;
     try
     {
@@ -6685,7 +6589,6 @@ void *workerThreadMain(void *parg)
     catch (ShutdownException)
     {
     }
-    g_forkLock->releaseRead();
     moduleReleaseGIL(true);
     serverAssert(!GlobalLocksAcquired());
     aeDeleteEventLoop(el);
@@ -6995,7 +6898,7 @@ int main(int argc, char **argv) {
             __AFL_INIT();
 #endif
             rio rdb;
-            rdbSaveInfo rsi;
+            rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
             startLoadingFile(stdin, (char*)"stdin", 0);
             rioInitWithFile(&rdb,stdin);
             rdbLoadRio(&rdb,0,&rsi);

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;
@@ -1393,38 +1389,6 @@ struct redisMemOverhead {
     } *db;
 };
 
-struct redisMaster {
-    char *masteruser;               /* AUTH with this user and masterauth with master */
-    char *masterauth;               /* AUTH with this password with master */
-    char *masterhost;               /* Hostname of master */
-    int masterport;                 /* Port of master */
-    client *cached_master;          /* Cached master to be reused for PSYNC. */
-    client *master;
-    /* The following two fields is where we store master PSYNC replid/offset
-     * while the PSYNC is in progress. At the end we'll copy the fields into
-     * the server->master client structure. */
-    char master_replid[CONFIG_RUN_ID_SIZE+1];  /* Master PSYNC runid. */
-    long long master_initial_offset;           /* Master PSYNC offset. */
-
-    bool isActive = false;
-    int repl_state;          /* Replication status if the instance is a replica */
-    off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
-    off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
-    off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */
-    connection *repl_transfer_s;     /* Slave -> Master SYNC socket */
-    int repl_transfer_fd;    /* Slave -> Master SYNC temp file descriptor */
-    char *repl_transfer_tmpfile; /* Slave-> master SYNC temp file name */
-    time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
-    time_t repl_down_since; /* Unix time at which link with master went down */
-
-    unsigned char master_uuid[UUID_BINARY_LEN];  /* Used during sync with master, this is our master's UUID */
-                                                /* After we've connected with our master use the UUID in g_pserver->master */
-    uint64_t mvccLastSync;
-    /* During a handshake the server may have stale keys, we track these here to share once a reciprocal connection is made */
-    std::map<int, std::vector<robj_sharedptr>> *staleKeyMap;
-    int ielReplTransfer = -1;
-};
-
 /* This structure can be optionally passed to RDB save/load functions in
  * order to implement additional functionalities, by storing and loading
  * metadata to the RDB file.
@@ -1433,66 +1397,7 @@ struct redisMaster {
  * replication in order to make sure that chained slaves (slaves of slaves)
  * select the correct DB and are able to accept the stream coming from the
  * top-level master. */
-class rdbSaveInfo {
-public:
-    rdbSaveInfo() {
-        repl_stream_db = -1;
-        repl_id_is_set = 0;
-        memcpy(repl_id, "0000000000000000000000000000000000000000", sizeof(repl_id));
-        repl_offset = -1;
-        fForceSetKey = TRUE;
-        mvccMinThreshold = 0;
-        masters = nullptr;
-        masterCount = 0;
-    }
-    rdbSaveInfo(const rdbSaveInfo &other) {
-        repl_stream_db = other.repl_stream_db;
-        repl_id_is_set = other.repl_id_is_set;
-        memcpy(repl_id, other.repl_id, sizeof(repl_id));
-        repl_offset = other.repl_offset;
-        fForceSetKey = other.fForceSetKey;
-        mvccMinThreshold = other.mvccMinThreshold;
-        masters = (struct redisMaster*)malloc(sizeof(struct redisMaster) * other.masterCount);
-        memcpy(masters, other.masters, sizeof(struct redisMaster) * other.masterCount);
-        masterCount = other.masterCount;
-    }
-    rdbSaveInfo(rdbSaveInfo &&other) : rdbSaveInfo() {
-        swap(*this, other);
-    }
-    rdbSaveInfo &operator=(rdbSaveInfo other) {
-        swap(*this, other);
-        return *this;
-    }
-    ~rdbSaveInfo() {
-        free(masters);
-    }
-    friend void swap(rdbSaveInfo &first, rdbSaveInfo &second) {
-        std::swap(first.repl_stream_db, second.repl_stream_db);
-        std::swap(first.repl_id_is_set, second.repl_id_is_set);
-        std::swap(first.repl_id, second.repl_id);
-        std::swap(first.repl_offset, second.repl_offset);
-        std::swap(first.fForceSetKey, second.fForceSetKey);
-        std::swap(first.mvccMinThreshold, second.mvccMinThreshold);
-        std::swap(first.masters, second.masters);
-        std::swap(first.masterCount, second.masterCount);
-
-    }
-
-    void addMaster(const struct redisMaster &mi) {
-        masterCount++;
-        if (masters == nullptr) {
-            masters = (struct redisMaster*)malloc(sizeof(struct redisMaster));
-        }
-        else {
-            masters = (struct redisMaster*)realloc(masters, sizeof(struct redisMaster) * masterCount);
-        }
-        memcpy(masters + masterCount - 1, &mi, sizeof(struct redisMaster));
-    }
-
-    size_t numMasters() {
-        return masterCount;
-    }
-
+typedef struct rdbSaveInfo {
     /* Used saving and loading. */
     int repl_stream_db;  /* DB to select in g_pserver->master client. */
 
@@ -1502,11 +1407,10 @@ public:
     long long repl_offset;                  /* Replication offset. */
     int fForceSetKey;
     uint64_t mvccMinThreshold;
-    struct redisMaster *masters;
+    struct redisMaster *mi;
+} rdbSaveInfo;
 
-private:
-    size_t masterCount;
-};
+#define RDB_SAVE_INFO_INIT {-1,0,"0000000000000000000000000000000000000000",-1, TRUE, 0, nullptr}
 
 struct malloc_stats {
     size_t zmalloc_used;
@@ -1595,6 +1499,38 @@ struct redisServerThreadVars {
     std::vector<client*> vecclientsProcess;
 };
 
+struct redisMaster {
+    char *masteruser;               /* AUTH with this user and masterauth with master */
+    char *masterauth;               /* AUTH with this password with master */
+    char *masterhost;               /* Hostname of master */
+    int masterport;                 /* Port of master */
+    client *cached_master;          /* Cached master to be reused for PSYNC. */
+    client *master;
+    /* The following two fields is where we store master PSYNC replid/offset
+     * while the PSYNC is in progress. At the end we'll copy the fields into
+     * the server->master client structure. */
+    char master_replid[CONFIG_RUN_ID_SIZE+1];  /* Master PSYNC runid. */
+    long long master_initial_offset;           /* Master PSYNC offset. */
+
+    bool isActive = false;
+    int repl_state;          /* Replication status if the instance is a replica */
+    off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
+    off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
+    off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */
+    connection *repl_transfer_s;     /* Slave -> Master SYNC socket */
+    int repl_transfer_fd;    /* Slave -> Master SYNC temp file descriptor */
+    char *repl_transfer_tmpfile; /* Slave-> master SYNC temp file name */
+    time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
+    time_t repl_down_since; /* Unix time at which link with master went down */
+
+    unsigned char master_uuid[UUID_BINARY_LEN];  /* Used during sync with master, this is our master's UUID */
+                                                /* After we've connected with our master use the UUID in g_pserver->master */
+    uint64_t mvccLastSync;
+    /* During a handshake the server may have stale keys, we track these here to share once a reciprocal connection is made */
+    std::map<int, std::vector<robj_sharedptr>> *staleKeyMap;
+    int ielReplTransfer = -1;
+};
+
 // Const vars are not changed after worker threads are launched
 struct redisServerConst {
     pid_t pid;                  /* Main process pid. */
@@ -2177,7 +2113,6 @@ 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;
@@ -2506,7 +2441,6 @@ void clearReplicationId2(void);
 void mergeReplicationId(const char *);
 void chopReplicationBacklog(void);
 void replicationCacheMasterUsingMyself(struct redisMaster *mi);
-void replicationCacheMasterUsingMaster(struct redisMaster *mi);
 void feedReplicationBacklog(const void *ptr, size_t len);
 void updateMasterAuth();
 void showLatestBacklog();
@@ -2562,7 +2496,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/version.h

@@ -1,4 +1,4 @@
-#define KEYDB_REAL_VERSION "255.255.255"
-#define KEYDB_VERSION_NUM 0x00ffffff
+#define KEYDB_REAL_VERSION "6.2.1"
+#define KEYDB_VERSION_NUM 0x00060201
 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