Drop min-clients-per-thread to a more reasonable number

* TLS: Session caching configuration support.
* TLS: Remove redundant config initialization.

.github/workflows/ci.yml

@@ -10,6 +10,7 @@ jobs:
     - uses: actions/checkout@v1
     - name: make
       run: |
+          sudo apt-get update
           sudo apt-get -y install uuid-dev libcurl4-openssl-dev
           make BUILD_TLS=yes -j2
     - name: gen-cert
@@ -34,6 +35,7 @@ jobs:
     - uses: actions/checkout@v1
     - name: make -j2
       run: | 
+        sudo apt-get update
         sudo apt-get -y install uuid-dev libcurl4-openssl-dev
         make -j2
 
@@ -50,6 +52,7 @@ jobs:
     - uses: actions/checkout@v1
     - name: make
       run: |
+          sudo apt-get update
           sudo apt-get -y install uuid-dev libcurl4-openssl-dev
           make MALLOC=libc -j2
 

.github/workflows/endurance.yml

@@ -2,7 +2,7 @@ name: Endurance
 
 on:
   schedule:
-    - cron: '30 * * * *'
+    - cron: '0 * * * *'
 
 jobs:
 
@@ -17,6 +17,6 @@ jobs:
           make -j8
     - name: test-multithread (5X)
       run: |
-        sudo apt-get install -y tcl8.5
+        sudo apt-get install -y tcl tcl-tls
         ./runtest --loopn 5 --config server-threads 3 --clients 5 --endurance
 

00-RELEASENOTES

@@ -11,6 +11,76 @@ CRITICAL: There is a critical bug affecting MOST USERS. Upgrade ASAP.
 SECURITY: There are security fixes in the release.
 --------------------------------------------------------------------------------
 
+================================================================================
+Redis 6.0.5     Released Tue Jun 09 11:56:08 CEST 2020
+================================================================================
+
+Upgrade urgency MODERATE: several bugs with moderate impact are fixed here.
+
+The most important issues are listed here:
+
+* Fix handling of speical chars in ACL LOAD.
+* Make Redis Cluster more robust about operation errors that may lead
+  to two clusters to mix together.
+* Revert the sendfile() implementation of RDB transfer. It causes some delay.
+* Fix TLS certificate loading for chained certificates.
+* Fix AOF rewirting of KEEPTTL SET option.
+* Fix MULTI/EXEC behavior during -BUSY script errors.
+
+And this is the full list of commits:
+
+antirez in commit ee8dd01bb:
+ Temporary fix for #7353 issue about EVAL during -BUSY.
+ 1 file changed, 9 insertions(+)
+
+xhe in commit a4a856d53:
+ return the correct proto version HELLO should return the current proto version, while the code hardcoded 3
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+Oran Agra in commit e2046b300:
+ Don't queue commands in an already aborted MULTI state
+ 1 file changed, 7 insertions(+)
+
+Oran Agra in commit b35fdf1de:
+ Avoid rejecting WATCH / UNWATCH, like MULTI/EXEC/DISCARD
+ 1 file changed, 4 insertions(+), 2 deletions(-)
+
+zhaozhao.zz in commit 1d7bf208c:
+ AOF: append origin SET if no expire option
+ 2 files changed, 23 insertions(+), 8 deletions(-)
+
+Oran Agra in commit 676445ad9:
+ fix disconnectSlaves, to try to free each slave.
+ 1 file changed, 1 deletion(-)
+
+zhaozhao.zz in commit 4846c0c8a:
+ donot free protected client in freeClientsInAsyncFreeQueue
+ 1 file changed, 9 insertions(+), 3 deletions(-)
+
+Oran Agra in commit f33de403e:
+ fix pingoff  test race
+ 1 file changed, 1 insertion(+)
+
+Kevin Fwu in commit 49af4d07e:
+ Fix TLS certificate loading for chained certificates.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+antirez in commit 329fddbda:
+ Revert "Implements sendfile for redis."
+ 2 files changed, 2 insertions(+), 55 deletions(-)
+
+antirez in commit 925a2cd5a:
+ Revert "avoid using sendfile if tls-replication is enabled"
+ 1 file changed, 27 insertions(+), 34 deletions(-)
+
+Liu Zhen in commit 84a7a9058:
+ fix clusters mixing accidentally by gossip
+ 1 file changed, 10 insertions(+), 2 deletions(-)
+
+antirez in commit cd63359a1:
+ Fix handling of special chars in ACL LOAD.
+ 1 file changed, 8 insertions(+), 4 deletions(-)
+
 ================================================================================
 Redis 6.0.4     Released Thu May 28 11:36:45 CEST 2020
 ================================================================================

README.md

@@ -1,6 +1,5 @@
 ![Current Release](https://img.shields.io/github/release/JohnSully/KeyDB.svg)
 ![CI](https://github.com/JohnSully/KeyDB/workflows/CI/badge.svg?branch=unstable)
-[![Join the chat at https://gitter.im/KeyDB/community](https://badges.gitter.im/KeyDB/community.svg)](https://gitter.im/KeyDB/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 [![StackShare](http://img.shields.io/badge/tech-stack-0690fa.svg?style=flat)](https://stackshare.io/eq-alpha-technology-inc/eq-alpha-technology-inc)
 
 ##### New! Want to extend KeyDB with Javascript?  Try [ModJS](https://github.com/JohnSully/ModJS)

TLS.md

@@ -56,8 +56,6 @@ Note that unlike Redis, KeyDB fully supports multithreading of TLS connections.
 To-Do List
 ----------
 
-- [ ] Add session caching support. Check if/how it's handled by clients to
-  assess how useful/important it is.
 - [ ] redis-benchmark support. The current implementation is a mix of using
   hiredis for parsing and basic networking (establishing connections), but
   directly manipulating sockets for most actions. This will need to be cleaned

keydb.conf

@@ -199,6 +199,22 @@ tcp-keepalive 300
 #
 # tls-prefer-server-ciphers yes
 
+# By default, TLS session caching is enabled to allow faster and less expensive
+# reconnections by clients that support it. Use the following directive to disable
+# caching.
+#
+# tls-session-caching no
+
+# Change the default number of TLS sessions cached. A zero value sets the cache
+# to unlimited size. The default size is 20480.
+#
+# tls-session-cache-size 5000
+
+# Change the default timeout of cached TLS sessions. The default timeout is 300
+# seconds.
+#
+# tls-session-cache-timeout 60
+
 ################################# GENERAL #####################################
 
 # By default KeyDB does not run as a daemon. Use 'yes' if you need it.
@@ -401,6 +417,20 @@ dir ./
 #
 replica-serve-stale-data yes
 
+# Active Replicas will allow read only data access while loading remote RDBs
+# provided they are permitted to serve stale data.  As an option you may also
+# permit them to accept write commands.  This is an EXPERIMENTAL feature and
+# may result in commands not being fully synchronized
+#
+# allow-write-during-load no
+
+# You can modify the number of masters necessary to form a replica quorum when
+# multi-master is enabled and replica-serve-stale-data is "no".  By default 
+# this is set to -1 which implies the number of known masters (e.g. those
+# you added with replicaof)
+#
+# replica-quorum -1
+
 # You can configure a replica instance to accept writes or not. Writing against
 # a replica instance may be useful to store some ephemeral data (because data
 # written on a replica will be easily deleted after resync with the master) but
@@ -1811,6 +1841,15 @@ jemalloc-bg-thread yes
 # Set bgsave child process to cpu affinity 1,10,11
 # bgsave_cpulist 1,10-11
 
+# The minimum number of clients on a thread before KeyDB assigns new connections to a different thread
+#  Tuning this parameter is a tradeoff between locking overhead and distributing the workload over multiple cores
+# min-clients-per-thread 50
+
+# Avoid forwarding RREPLAY messages to other masters?
+#   WARNING: This setting is dangerous! You must be certain all masters are connected to each
+#   other in a true mesh topology or data loss will occur!
+#   This command can be used to reduce multimaster bus traffic
+# multi-master-no-forward no
 
 # Path to directory for file backed scratchpad.  The file backed scratchpad
 # reduces memory requirements by storing rarely accessed data on disk 

pkg/deb/master_changelog

@@ -1,3 +1,14 @@
+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/Makefile

@@ -402,7 +402,7 @@ bench: $(REDIS_BENCHMARK_NAME)
 	$(MAKE) CXXFLAGS="-m32" CFLAGS="-m32" LDFLAGS="-m32"
 
 gcov:
-	$(MAKE) REDIS_CFLAGS="-fprofile-arcs -ftest-coverage -DCOVERAGE_TEST" REDIS_LDFLAGS="-fprofile-arcs -ftest-coverage"
+	$(MAKE) REDIS_CXXFLAGS="-fprofile-arcs -ftest-coverage -DCOVERAGE_TEST" REDIS_CFLAGS="-fprofile-arcs -ftest-coverage -DCOVERAGE_TEST" REDIS_LDFLAGS="-fprofile-arcs -ftest-coverage"
 
 noopt:
 	$(MAKE) OPTIMIZATION="-O0"

src/acl.cpp

@@ -735,10 +735,11 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  * EEXIST: You are adding a key pattern after "*" was already added. This is
  *         almost surely an error on the user side.
  * ENODEV: The password you are trying to remove from the user does not exist.
- * EBADMSG: The hash you are trying to add is not a valid hash. 
+ * EBADMSG: The hash you are trying to add is not a valid hash.
  */
 int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     if (oplen == -1) oplen = strlen(op);
+    if (oplen == 0) return C_OK; /* Empty string is a no-operation. */
     if (!strcasecmp(op,"on")) {
         u->flags |= USER_FLAG_ENABLED;
         u->flags &= ~USER_FLAG_DISABLED;
@@ -1300,7 +1301,7 @@ sds ACLLoadFromFile(const char *filename) {
         if (lines[i][0] == '\0') continue;
 
         /* Split into arguments */
-        argv = sdssplitargs(lines[i],&argc);
+        argv = sdssplitlen(lines[i],sdslen(lines[i])," ",1,&argc);
         if (argv == NULL) {
             errors = sdscatprintf(errors,
                      "%s:%d: unbalanced quotes in acl line. ",
@@ -1332,11 +1333,14 @@ sds ACLLoadFromFile(const char *filename) {
             continue;
         }
 
-        /* Try to process the line using the fake user to validate iif
-         * the rules are able to apply cleanly. */
+        /* Try to process the line using the fake user to validate if
+         * the rules are able to apply cleanly. At this stage we also
+         * trim trailing spaces, so that we don't have to handle that
+         * in ACLSetUser(). */
         ACLSetUser(fakeuser,"reset",-1);
         int j;
         for (j = 2; j < argc; j++) {
+            argv[j] = sdstrim(argv[j],"\t\r\n");
             if (ACLSetUser(fakeuser,argv[j],sdslen(argv[j])) != C_OK) {
                 const char *errmsg = ACLSetUserStringError();
                 errors = sdscatprintf(errors,

src/ae.cpp

@@ -234,7 +234,7 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
 
     int ret = AE_OK;
     
-    aeCommand cmd = {};
+    aeCommand cmd;
     cmd.op = AE_ASYNC_OP::CreateFileEvent;
     cmd.fd = fd;
     cmd.mask = mask;
@@ -258,9 +258,11 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
     
     if (fSynchronous)
     {
-        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::defer_lock);
+        {
+        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::adopt_lock);
         cmd.pctl->cv.wait(ulock);
         ret = cmd.pctl->rval;
+        }
         delete cmd.pctl;
     }
 
@@ -294,14 +296,13 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     }
 
     aeCommand cmd = {};
-    memset(&cmd, 0, sizeof(aeCommand));
     cmd.op = AE_ASYNC_OP::PostCppFunction;
     cmd.pfn = new (MALLOC_LOCAL) std::function<void()>(fn);
     cmd.pctl = nullptr;
     cmd.fLock = fLock;
     if (fSynchronous)
     {
-        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl();
+        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl;
         cmd.pctl->mutexcv.lock();
     }
 
@@ -312,9 +313,11 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     int ret = AE_OK;
     if (fSynchronous)
     {
-        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::defer_lock);
+        {
+        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::adopt_lock);
         cmd.pctl->cv.wait(ulock);
         ret = cmd.pctl->rval;
+        }
         delete cmd.pctl;
     }
     return ret;
@@ -454,7 +457,7 @@ void aeDeleteFileEventAsync(aeEventLoop *eventLoop, int fd, int mask)
 {
     if (eventLoop == g_eventLoopThisThread)
         return aeDeleteFileEvent(eventLoop, fd, mask);
-    aeCommand cmd;
+    aeCommand cmd = {};
     cmd.op = AE_ASYNC_OP::DeleteFileEvent;
     cmd.fd = fd;
     cmd.mask = mask;

src/aelocker.h

@@ -9,11 +9,14 @@ public:
     {
     }
 
-    void arm(client *c) // if a client is passed, then the client is already locked
+    void arm(client *c, bool fIfNeeded = false) // if a client is passed, then the client is already locked
     {
         if (m_fArmed)
             return;
         
+        if (fIfNeeded && aeThreadOwnsLock())
+            return;
+
         serverAssertDebug(!GlobalLocksAcquired());
 
         if (c != nullptr)

src/aof.cpp

@@ -671,19 +671,23 @@ sds catCommandForAofAndActiveReplication(sds buf, struct redisCommand *cmd, robj
     } else if (cmd->proc == setCommand && argc > 3) {
         int i;
         robj *exarg = NULL, *pxarg = NULL;
-        /* Translate SET [EX seconds][PX milliseconds] to SET and PEXPIREAT */
-        buf = catAppendOnlyGenericCommand(buf,3,argv);
         for (i = 3; i < argc; i ++) {
             if (!strcasecmp(szFromObj(argv[i]), "ex")) exarg = argv[i+1];
             if (!strcasecmp(szFromObj(argv[i]), "px")) pxarg = argv[i+1];
         }
         serverAssert(!(exarg && pxarg));
-        if (exarg)
-            buf = catAppendOnlyExpireAtCommand(buf,cserver.expireCommand,argv[1],
-                                               exarg);
-        if (pxarg)
-            buf = catAppendOnlyExpireAtCommand(buf,cserver.pexpireCommand,argv[1],
-                                               pxarg);
+        if (exarg || pxarg) {
+            /* Translate SET [EX seconds][PX milliseconds] to SET and PEXPIREAT */
+            buf = catAppendOnlyGenericCommand(buf,3,argv);
+            if (exarg)
+                buf = catAppendOnlyExpireAtCommand(buf,cserver.expireCommand,argv[1],
+                                                   exarg);
+            if (pxarg)
+                buf = catAppendOnlyExpireAtCommand(buf,cserver.pexpireCommand,argv[1],
+                                                   pxarg);
+        } else {
+            buf = catAppendOnlyGenericCommand(buf,argc,argv);
+        }
     } else if (cmd->proc == expireMemberCommand || cmd->proc == expireMemberAtCommand ||
         cmd->proc == pexpireMemberAtCommand) {
         /* Translate subkey expire commands to PEXPIREMEMBERAT */
@@ -1422,7 +1426,8 @@ int rewriteAppendOnlyFileRio(rio *aof) {
         /* Iterate this DB writing every entry */
         while((de = dictNext(di)) != NULL) {
             sds keystr;
-            robj key, *o;
+            redisObjectStack key;
+            robj *o = nullptr;
 
             keystr = (sds)dictGetKey(de);
             o = (robj*)dictGetVal(de);

src/cluster.cpp

@@ -1501,7 +1501,10 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
             }
         } else {
             /* If it's not in NOADDR state and we don't have it, we
-             * start a handshake process against this IP/PORT pairs.
+             * add it to our trusted dict with exact nodeid and flag.
+             * Note that we cannot simply start a handshake against
+             * this IP/PORT pairs, since IP/PORT can be reused already,
+             * otherwise we risk joining another cluster.
              *
              * Note that we require that the sender of this gossip message
              * is a well known node in our cluster, otherwise we risk
@@ -1510,7 +1513,12 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
                 !(flags & CLUSTER_NODE_NOADDR) &&
                 !clusterBlacklistExists(g->nodename))
             {
-                clusterStartHandshake(g->ip,ntohs(g->port),ntohs(g->cport));
+                clusterNode *node;
+                node = createClusterNode(g->nodename, flags);
+                memcpy(node->ip,g->ip,NET_IP_STR_LEN);
+                node->port = ntohs(g->port);
+                node->cport = ntohs(g->cport);
+                clusterAddNode(node);
             }
         }
 

src/config.cpp

@@ -518,7 +518,7 @@ void loadServerConfigFromString(char *config) {
             err = "KeyDB not compliled with scratch-file support.";
             goto loaderr;
 #endif
-        } else if (!strcasecmp(argv[0],"server-threads") && argc == 2) {
+        } else if ((!strcasecmp(argv[0],"server-threads") || !strcasecmp(argv[0],"io-threads")) && argc == 2) {
             cserver.cthreads = atoi(argv[1]);
             if (cserver.cthreads <= 0 || cserver.cthreads > MAX_EVENT_LOOPS) {
                 err = "Invalid number of threads specified";
@@ -2167,13 +2167,20 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
     return 1;
 }
 
+static int validateMultiMasterNoForward(int val, const char **) {
+    if (val) {
+        serverLog(LL_WARNING, "WARNING: multi-master-no-forward is set, you *must* use a mesh topology or dataloss will occur");
+    }
+    return 1;
+}
+
 #ifdef USE_OPENSSL
 static int updateTlsCfg(char *val, char *prev, const char **err) {
     UNUSED(val);
     UNUSED(prev);
     UNUSED(err);
     if (tlsConfigure(&g_pserver->tls_ctx_config) == C_ERR) {
-        *err = "Unable to configure tls-cert-file. Check server logs.";
+        *err = "Unable to update TLS configuration. Check server logs.";
         return 0;
     }
     return 1;
@@ -2183,8 +2190,15 @@ static int updateTlsCfgBool(int val, int prev, const char **err) {
     UNUSED(prev);
     return updateTlsCfg(NULL, NULL, err);
 }
+
+static int updateTlsCfgInt(long long val, long long prev, const char **err) {
+    UNUSED(val);
+    UNUSED(prev);
+    return updateTlsCfg(NULL, NULL, err);
+}
 #endif  /* USE_OPENSSL */
 
+int fDummy = false;
 standardConfig configs[] = {
     /* Bool configs */
     createBoolConfig("rdbchecksum", NULL, IMMUTABLE_CONFIG, g_pserver->rdb_checksum, 1, NULL, NULL),
@@ -2200,6 +2214,7 @@ standardConfig configs[] = {
     createBoolConfig("lazyfree-lazy-eviction", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_eviction, 0, NULL, NULL),
     createBoolConfig("lazyfree-lazy-expire", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_expire, 0, NULL, NULL),
     createBoolConfig("lazyfree-lazy-server-del", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_server_del, 0, NULL, NULL),
+    createBoolConfig("lazyfree-lazy-user-del", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_user_del , 0, NULL, NULL),
     createBoolConfig("repl-disable-tcp-nodelay", NULL, MODIFIABLE_CONFIG, g_pserver->repl_disable_tcp_nodelay, 0, NULL, NULL),
     createBoolConfig("repl-diskless-sync", NULL, MODIFIABLE_CONFIG, g_pserver->repl_diskless_sync, 0, NULL, NULL),
     createBoolConfig("aof-rewrite-incremental-fsync", NULL, MODIFIABLE_CONFIG, g_pserver->aof_rewrite_incremental_fsync, 1, NULL, NULL),
@@ -2220,6 +2235,9 @@ standardConfig configs[] = {
     createBoolConfig("cluster-enabled", NULL, IMMUTABLE_CONFIG, g_pserver->cluster_enabled, 0, NULL, NULL),
     createBoolConfig("appendonly", NULL, MODIFIABLE_CONFIG, g_pserver->aof_enabled, 0, NULL, updateAppendonly),
     createBoolConfig("cluster-allow-reads-when-down", NULL, MODIFIABLE_CONFIG, g_pserver->cluster_allow_reads_when_down, 0, NULL, NULL),
+    createBoolConfig("multi-master-no-forward", NULL, MODIFIABLE_CONFIG, cserver.multimaster_no_forward, 0, validateMultiMasterNoForward, NULL),
+    createBoolConfig("allow-write-during-load", NULL, MODIFIABLE_CONFIG, g_pserver->fWriteDuringActiveLoad, 0, NULL, NULL),
+    createBoolConfig("io-threads-do-reads", NULL, IMMUTABLE_CONFIG, fDummy, 0, NULL, NULL),
 
     /* String Configs */
     createStringConfig("aclfile", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->acl_filename, "", NULL, NULL),
@@ -2276,7 +2294,8 @@ standardConfig configs[] = {
     createIntConfig("hz", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->config_hz, CONFIG_DEFAULT_HZ, INTEGER_CONFIG, NULL, updateHZ),
     createIntConfig("min-replicas-to-write", "min-slaves-to-write", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->repl_min_slaves_to_write, 0, INTEGER_CONFIG, NULL, updateGoodSlaves),
     createIntConfig("min-replicas-max-lag", "min-slaves-max-lag", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->repl_min_slaves_max_lag, 10, INTEGER_CONFIG, NULL, updateGoodSlaves),
-    createIntConfig("min-clients-per-thread", NULL, MODIFIABLE_CONFIG, 0, 400, cserver.thread_min_client_threshold, 50, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("min-clients-per-thread", NULL, MODIFIABLE_CONFIG, 0, 400, cserver.thread_min_client_threshold, 20, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("replica-quorum", NULL, MODIFIABLE_CONFIG, -1, INT_MAX, g_pserver->repl_quorum, -1, INTEGER_CONFIG, NULL, NULL),
     /* Unsigned int configs */
     createUIntConfig("maxclients", NULL, MODIFIABLE_CONFIG, 1, UINT_MAX, g_pserver->maxclients, 10000, INTEGER_CONFIG, NULL, updateMaxclients),
 
@@ -2314,10 +2333,13 @@ standardConfig configs[] = {
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, IMMUTABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, NULL), /* TCP port. */
+    createIntConfig("tls-session-cache-size", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->tls_ctx_config.session_cache_size, 20*1024, INTEGER_CONFIG, NULL, updateTlsCfgInt),
+    createIntConfig("tls-session-cache-timeout", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->tls_ctx_config.session_cache_timeout, 300, INTEGER_CONFIG, NULL, updateTlsCfgInt),
     createBoolConfig("tls-cluster", NULL, MODIFIABLE_CONFIG, g_pserver->tls_cluster, 0, NULL, NULL),
     createBoolConfig("tls-replication", NULL, MODIFIABLE_CONFIG, g_pserver->tls_replication, 0, NULL, NULL),
     createBoolConfig("tls-auth-clients", NULL, MODIFIABLE_CONFIG, g_pserver->tls_auth_clients, 1, NULL, NULL),
     createBoolConfig("tls-prefer-server-ciphers", NULL, MODIFIABLE_CONFIG, g_pserver->tls_ctx_config.prefer_server_ciphers, 0, NULL, updateTlsCfgBool),
+    createBoolConfig("tls-session-caching", NULL, MODIFIABLE_CONFIG, g_pserver->tls_ctx_config.session_caching, 1, NULL, updateTlsCfgBool),
     createStringConfig("tls-cert-file", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->tls_ctx_config.cert_file, NULL, NULL, updateTlsCfg),
     createStringConfig("tls-key-file", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->tls_ctx_config.key_file, NULL, NULL, updateTlsCfg),
     createStringConfig("tls-dh-params-file", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->tls_ctx_config.dh_params_file, NULL, NULL, updateTlsCfg),

src/connection.cpp

@@ -164,7 +164,7 @@ static int connSocketWrite(connection *conn, const void *data, size_t data_len)
     int ret = write(conn->fd, data, data_len);
     if (ret < 0 && errno != EAGAIN) {
         conn->last_errno = errno;
-        conn->state = CONN_STATE_ERROR;
+        conn->state.store(CONN_STATE_ERROR, std::memory_order_relaxed);
     }
 
     return ret;
@@ -173,10 +173,10 @@ static int connSocketWrite(connection *conn, const void *data, size_t data_len)
 static int connSocketRead(connection *conn, void *buf, size_t buf_len) {
     int ret = read(conn->fd, buf, buf_len);
     if (!ret) {
-        conn->state = CONN_STATE_CLOSED;
+        conn->state.store(CONN_STATE_CLOSED, std::memory_order_release);
     } else if (ret < 0 && errno != EAGAIN) {
         conn->last_errno = errno;
-        conn->state = CONN_STATE_ERROR;
+        conn->state.store(CONN_STATE_ERROR, std::memory_order_release);
     }
 
     return ret;
@@ -253,14 +253,14 @@ static void connSocketEventHandler(struct aeEventLoop *el, int fd, void *clientD
     UNUSED(fd);
     connection *conn = (connection*)clientData;
 
-    if (conn->state == CONN_STATE_CONNECTING &&
+    if (conn->state.load(std::memory_order_relaxed) == CONN_STATE_CONNECTING &&
             (mask & AE_WRITABLE) && conn->conn_handler) {
 
         if (connGetSocketError(conn)) {
             conn->last_errno = errno;
-            conn->state = CONN_STATE_ERROR;
+            conn->state.store(CONN_STATE_ERROR, std::memory_order_release);
         } else {
-            conn->state = CONN_STATE_CONNECTED;
+            conn->state.store(CONN_STATE_CONNECTED, std::memory_order_release);
         }
 
         if (!conn->write_handler) aeDeleteFileEvent(serverTL->el,conn->fd,AE_WRITABLE);
@@ -407,7 +407,7 @@ int connRecvTimeout(connection *conn, long long ms) {
 }
 
 int connGetState(connection *conn) {
-    return conn->state;
+    return conn->state.load(std::memory_order_relaxed);
 }
 
 void connSetThreadAffinity(connection *conn, int cpu) {

src/db.cpp

@@ -93,7 +93,7 @@ static robj *lookupKey(redisDb *db, robj *key, int flags) {
         updateDbValAccess(de, flags);
 
         if (flags & LOOKUP_UPDATEMVCC) {
-            val->mvcc_tstamp = getMvccTstamp();
+            setMvccTstamp(val, getMvccTstamp());
         }
         return val;
     } else {
@@ -206,7 +206,9 @@ int dbAddCore(redisDb *db, robj *key, robj *val) {
     serverAssert(!val->FExpires());
     sds copy = sdsdup(szFromObj(key));
     int retval = dictAdd(db->pdict, copy, val);
-    val->mvcc_tstamp = key->mvcc_tstamp = getMvccTstamp();
+    uint64_t mvcc = getMvccTstamp();
+    setMvccTstamp(key, mvcc);
+    setMvccTstamp(val, mvcc);
 
     if (retval == DICT_OK)
     {
@@ -256,7 +258,7 @@ void dbOverwriteCore(redisDb *db, dictEntry *de, robj *key, robj *val, bool fUpd
     if (fUpdateMvcc) {
         if (val->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
             val = dupStringObject(val);
-        val->mvcc_tstamp = getMvccTstamp();
+        setMvccTstamp(val, getMvccTstamp());
     }
 
     dictSetVal(db->pdict, de, val);
@@ -291,12 +293,12 @@ int dbMerge(redisDb *db, robj *key, robj *val, int fReplace)
             return (dbAddCore(db, key, val) == DICT_OK);
 
         robj *old = (robj*)dictGetVal(de);
-        if (old->mvcc_tstamp <= val->mvcc_tstamp)
+        if (mvccFromObj(old) <= mvccFromObj(val))
         {
             dbOverwriteCore(db, de, key, val, false, true);
             return true;
         }
-        
+
         return false;
     }
     else
@@ -683,6 +685,13 @@ void existsCommand(client *c) {
     addReplyLongLong(c,count);
 }
 
+void mexistsCommand(client *c) {
+    addReplyArrayLen(c, c->argc - 1);
+    for (int j = 1; j < c->argc; ++j) {
+        addReplyBool(c, lookupKeyRead(c->db, c->argv[j]));
+    }
+}
+
 void selectCommand(client *c) {
     long id;
 
@@ -1219,8 +1228,7 @@ int dbSwapDatabases(long id1, long id2) {
     if (id1 < 0 || id1 >= cserver.dbnum ||
         id2 < 0 || id2 >= cserver.dbnum) return C_ERR;
     if (id1 == id2) return C_OK;
-    redisDb aux; 
-    memcpy(&aux, &g_pserver->db[id1], sizeof(redisDb));
+    redisDb aux(g_pserver->db[id1]);
     redisDb *db1 = &g_pserver->db[id1], *db2 = &g_pserver->db[id2];
 
     /* Swap hash tables. Note that we don't swap blocking_keys,
@@ -1387,7 +1395,7 @@ void setExpire(client *c, redisDb *db, robj *key, robj *subkey, long long when)
         db->setexpire->insert(e);
     }
 
-    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0;
+    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0 && !g_pserver->fActiveReplica;
     if (c && writable_slave && !(c->flags & CLIENT_MASTER))
         rememberSlaveKeyWithExpire(db,key);
 }
@@ -1424,7 +1432,7 @@ void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
     ((robj*)dictGetVal(kde))->SetFExpires(true);
 
 
-    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0;
+    int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0 && !g_pserver->fActiveReplica;
     if (c && writable_slave && !(c->flags & CLIENT_MASTER))
         rememberSlaveKeyWithExpire(db,key);
 }
@@ -1480,7 +1488,6 @@ void propagateExpire(redisDb *db, robj *key, int lazy) {
 void propagateSubkeyExpire(redisDb *db, int type, robj *key, robj *subkey)
 {
     robj *argv[3];
-    robj objT;
     redisCommand *cmd = nullptr;
     switch (type)
     {

src/defrag.cpp

@@ -55,7 +55,8 @@ bool replaceSateliteOSetKeyPtr(expireset &set, sds oldkey, sds newkey);
  * returns NULL in case the allocatoin wasn't moved.
  * when it returns a non-null value, the old pointer was already released
  * and should NOT be accessed. */
-void* activeDefragAlloc(void *ptr) {
+template<typename TPTR>
+TPTR* activeDefragAlloc(TPTR *ptr) {
     size_t size;
     void *newptr;
     if(!je_get_defrag_hint(ptr)) {
@@ -70,7 +71,14 @@ void* activeDefragAlloc(void *ptr) {
     newptr = zmalloc_no_tcache(size);
     memcpy(newptr, ptr, size);
     zfree_no_tcache(ptr);
-    return newptr;
+    return (TPTR*)newptr;
+}
+
+template<>
+robj* activeDefragAlloc(robj *o) {
+    void *pvSrc = allocPtrFromObj(o);
+    void *pvDst = activeDefragAlloc(pvSrc);
+    return objFromAllocPtr(pvDst);
 }
 
 /*Defrag helper for sds strings

src/expire.cpp

@@ -80,7 +80,7 @@ void activeExpireCycleExpire(redisDb *db, expireEntry &e, long long now) {
     robj *val = (robj*)dictGetVal(de);
     int deleted = 0;
 
-    robj objKey;
+    redisObjectStack objKey;
     initStaticStringObject(objKey, (char*)e.key());
     bool fTtlChanged = false;
 
@@ -145,7 +145,7 @@ void activeExpireCycleExpire(redisDb *db, expireEntry &e, long long now) {
             serverAssert(false);
         }
         
-        robj objSubkey;
+        redisObjectStack objSubkey;
         initStaticStringObject(objSubkey, (char*)pfat->nextExpireEntry().spsubkey.get());
         propagateSubkeyExpire(db, val->type, &objKey, &objSubkey);
         
@@ -745,3 +745,90 @@ void touchCommand(client *c) {
     addReplyLongLong(c,touched);
 }
 
+expireEntryFat::~expireEntryFat()
+{
+    if (m_dictIndex != nullptr)
+        dictRelease(m_dictIndex);
+}
+
+void expireEntryFat::createIndex()
+{
+    serverAssert(m_dictIndex == nullptr);
+    m_dictIndex = dictCreate(&keyptrDictType, nullptr);
+
+    for (auto &entry : m_vecexpireEntries)
+    {
+        if (entry.spsubkey != nullptr)
+        {
+            dictEntry *de = dictAddRaw(m_dictIndex, (void*)entry.spsubkey.get(), nullptr);
+            de->v.s64 = entry.when;
+        }
+    }
+}
+
+void expireEntryFat::expireSubKey(const char *szSubkey, long long when)
+{
+    if (m_vecexpireEntries.size() >= INDEX_THRESHOLD && m_dictIndex == nullptr)
+        createIndex();
+
+    // First check if the subkey already has an expiration
+    if (m_dictIndex != nullptr && szSubkey != nullptr)
+    {
+        dictEntry *de = dictFind(m_dictIndex, szSubkey);
+        if (de != nullptr)
+        {
+            auto itr = std::lower_bound(m_vecexpireEntries.begin(), m_vecexpireEntries.end(), de->v.u64);
+            while (itr != m_vecexpireEntries.end() && itr->when == de->v.s64)
+            {
+                bool fFound = false;
+                if (szSubkey == nullptr && itr->spsubkey == nullptr) {
+                    fFound = true;
+                } else if (szSubkey != nullptr && itr->spsubkey != nullptr && sdscmp((sds)itr->spsubkey.get(), (sds)szSubkey) == 0) {
+                    fFound = true;
+                }
+                if (fFound) {
+                    m_vecexpireEntries.erase(itr);
+                    dictDelete(m_dictIndex, szSubkey);
+                    break;
+                }
+                ++itr;
+            }
+        }
+    }
+    else
+    {
+        for (auto &entry : m_vecexpireEntries)
+        {
+            if (szSubkey != nullptr)
+            {
+                // if this is a subkey expiry then its not a match if the expireEntry is either for the
+                //  primary key or a different subkey
+                if (entry.spsubkey == nullptr || sdscmp((sds)entry.spsubkey.get(), (sds)szSubkey) != 0)
+                    continue;
+            }
+            else
+            {
+                if (entry.spsubkey != nullptr)
+                    continue;
+            }
+            m_vecexpireEntries.erase(m_vecexpireEntries.begin() + (&entry - m_vecexpireEntries.data()));
+            break;
+        }
+    }
+    auto itrInsert = std::lower_bound(m_vecexpireEntries.begin(), m_vecexpireEntries.end(), when);
+    const char *subkey = (szSubkey) ? sdsdup(szSubkey) : nullptr;
+    auto itr = m_vecexpireEntries.emplace(itrInsert, when, subkey);
+    if (m_dictIndex && subkey) {
+        dictEntry *de = dictAddRaw(m_dictIndex, (void*)itr->spsubkey.get(), nullptr);
+        de->v.s64 = when;
+    }
+}
+
+void expireEntryFat::popfrontExpireEntry()
+{ 
+    if (m_dictIndex != nullptr && m_vecexpireEntries.begin()->spsubkey) {
+        int res = dictDelete(m_dictIndex, (void*)m_vecexpireEntries.begin()->spsubkey.get());
+        serverAssert(res == DICT_OK);
+    }
+    m_vecexpireEntries.erase(m_vecexpireEntries.begin());
+}

src/expire.h

@@ -0,0 +1,220 @@
+#pragma once
+
+class expireEntryFat
+{
+    friend class expireEntry;
+    static const int INDEX_THRESHOLD = 16;
+public:
+    struct subexpireEntry
+    {
+        long long when;
+        std::unique_ptr<const char, void(*)(const char*)> spsubkey;
+
+        subexpireEntry(long long when, const char *subkey)
+            : when(when), spsubkey(subkey, sdsfree)
+        {}
+
+        bool operator<(long long when) const noexcept { return this->when < when; }
+        bool operator<(const subexpireEntry &se) { return this->when < se.when; }
+    };
+
+private:
+    sds m_keyPrimary;
+    std::vector<subexpireEntry> m_vecexpireEntries;  // Note a NULL for the sds portion means the expire is for the primary key
+    dict *m_dictIndex = nullptr;
+
+    void createIndex();
+public:
+    expireEntryFat(sds keyPrimary)
+        : m_keyPrimary(keyPrimary)
+        {}
+    ~expireEntryFat();
+
+    long long when() const noexcept { return m_vecexpireEntries.front().when; }
+    const char *key() const noexcept { return m_keyPrimary; }
+
+    bool operator<(long long when) const noexcept { return this->when() <  when; }
+
+    void expireSubKey(const char *szSubkey, long long when);
+
+    bool FEmpty() const noexcept { return m_vecexpireEntries.empty(); }
+    const subexpireEntry &nextExpireEntry() const noexcept { return m_vecexpireEntries.front(); }
+    void popfrontExpireEntry();
+    const subexpireEntry &operator[](size_t idx) { return m_vecexpireEntries[idx]; }
+    size_t size() const noexcept { return m_vecexpireEntries.size(); }
+};
+
+class expireEntry {
+    union
+    {
+        sds m_key;
+        expireEntryFat *m_pfatentry;
+    } u;
+    long long m_when;   // LLONG_MIN means this is a fat entry and we should use the pointer
+
+public:
+    class iter
+    {
+        friend class expireEntry;
+        expireEntry *m_pentry = nullptr;
+        size_t m_idx = 0;
+
+    public:
+        iter(expireEntry *pentry, size_t idx)
+            : m_pentry(pentry), m_idx(idx)
+        {}
+
+        iter &operator++() { ++m_idx; return *this; }
+        
+        const char *subkey() const
+        {
+            if (m_pentry->FFat())
+                return (*m_pentry->pfatentry())[m_idx].spsubkey.get();
+            return nullptr;
+        }
+        long long when() const
+        {
+            if (m_pentry->FFat())
+                return (*m_pentry->pfatentry())[m_idx].when;
+            return m_pentry->when();
+        }
+
+        bool operator!=(const iter &other)
+        {
+            return m_idx != other.m_idx;
+        }
+
+        const iter &operator*() const { return *this; }
+    };
+
+    expireEntry(sds key, const char *subkey, long long when)
+    {
+        if (subkey != nullptr)
+        {
+            m_when = LLONG_MIN;
+            u.m_pfatentry = new (MALLOC_LOCAL) expireEntryFat(key);
+            u.m_pfatentry->expireSubKey(subkey, when);
+        }
+        else
+        {
+            u.m_key = key;
+            m_when = when;
+        }
+    }
+
+    expireEntry(expireEntryFat *pfatentry)
+    {
+        u.m_pfatentry = pfatentry;
+        m_when = LLONG_MIN;
+    }
+
+    expireEntry(expireEntry &&e)
+    {
+        u.m_key = e.u.m_key;
+        m_when = e.m_when;
+        e.u.m_key = (char*)key();  // we do this so it can still be found in the set
+        e.m_when = 0;
+    }
+
+    ~expireEntry()
+    {
+        if (FFat())
+            delete u.m_pfatentry;
+    }
+
+    void setKeyUnsafe(sds key)
+    {
+        if (FFat())
+            u.m_pfatentry->m_keyPrimary = key;
+        else
+            u.m_key = key;
+    }
+
+    inline bool FFat() const noexcept { return m_when == LLONG_MIN; }
+    expireEntryFat *pfatentry() { assert(FFat()); return u.m_pfatentry; }
+
+
+    bool operator==(const char *key) const noexcept
+    { 
+        return this->key() == key; 
+    }
+
+    bool operator<(const expireEntry &e) const noexcept
+    { 
+        return when() < e.when(); 
+    }
+    bool operator<(long long when) const noexcept
+    { 
+        return this->when() < when;
+    }
+
+    const char *key() const noexcept
+    { 
+        if (FFat())
+            return u.m_pfatentry->key();
+        return u.m_key;
+    }
+    long long when() const noexcept
+    { 
+        if (FFat())
+            return u.m_pfatentry->when();
+        return m_when; 
+    }
+
+    void update(const char *subkey, long long when)
+    {
+        if (!FFat())
+        {
+            if (subkey == nullptr)
+            {
+                m_when = when;
+                return;
+            }
+            else
+            {
+                // we have to upgrade to a fat entry
+                long long whenT = m_when;
+                sds keyPrimary = u.m_key;
+                m_when = LLONG_MIN;
+                u.m_pfatentry = new (MALLOC_LOCAL) expireEntryFat(keyPrimary);
+                u.m_pfatentry->expireSubKey(nullptr, whenT);
+                // at this point we're fat so fall through
+            }
+        }
+        u.m_pfatentry->expireSubKey(subkey, when);
+    }
+    
+    iter begin() { return iter(this, 0); }
+    iter end()
+    {
+        if (FFat())
+            return iter(this, u.m_pfatentry->size());
+        return iter(this, 1);
+    }
+    
+    void erase(iter &itr)
+    {
+        if (!FFat())
+            throw -1;   // assert
+        pfatentry()->m_vecexpireEntries.erase(
+            pfatentry()->m_vecexpireEntries.begin() + itr.m_idx);
+    }
+
+    bool FGetPrimaryExpire(long long *pwhen)
+    {
+        *pwhen = -1;
+        for (auto itr : *this)
+        {
+            if (itr.subkey() == nullptr)
+            {
+                *pwhen = itr.when();
+                return true;
+            }
+        }
+        return false;
+    }
+
+    explicit operator const char*() const noexcept { return key(); }
+    explicit operator long long() const noexcept { return when(); }
+};
+typedef semiorderedset<expireEntry, const char *, true /*expireEntry can be memmoved*/> expireset;

src/fastlock.cpp

@@ -35,7 +35,11 @@
 #include <sched.h>
 #include <atomic>
 #include <assert.h>
+#ifdef __FreeBSD__
+#include <pthread_np.h>
+#else
 #include <pthread.h>
+#endif
 #include <limits.h>
 #include <map>
 #ifdef __linux__
@@ -167,7 +171,12 @@ extern "C" pid_t gettid()
 #else
 	if (pidCache == -1) {
 		uint64_t tidT;
+#ifdef __FreeBSD__
+// Check https://github.com/ClickHouse/ClickHouse/commit/8d51824ddcb604b6f179a0216f0d32ba5612bd2e
+                tidT = pthread_getthreadid_np();
+#else
 		pthread_threadid_np(nullptr, &tidT);
+#endif
 		serverAssert(tidT < UINT_MAX);
 		pidCache = (int)tidT;
 	}
@@ -343,7 +352,9 @@ extern "C" void fastlock_lock(struct fastlock *lock)
     unsigned myticket = __atomic_fetch_add(&lock->m_ticket.m_avail, 1, __ATOMIC_RELEASE);
     unsigned cloops = 0;
     ticket ticketT;
-    unsigned loopLimit = g_fHighCpuPressure ? 0x10000 : 0x100000;
+    int fHighPressure;
+    __atomic_load(&g_fHighCpuPressure, &fHighPressure, __ATOMIC_RELAXED);
+    unsigned loopLimit = fHighPressure ? 0x10000 : 0x100000;
 
     for (;;)
     {
@@ -478,19 +489,22 @@ void fastlock_auto_adjust_waits()
 {
 #ifdef __linux__
     struct sysinfo sysinf;
-    auto fHighPressurePrev = g_fHighCpuPressure;
+    int fHighPressurePrev, fHighPressureNew;
+    __atomic_load(&g_fHighCpuPressure, &fHighPressurePrev, __ATOMIC_RELAXED);
+    fHighPressureNew = fHighPressurePrev;
     memset(&sysinf, 0, sizeof sysinf);
     if (!sysinfo(&sysinf)) {
         auto avgCoreLoad = sysinf.loads[0] / get_nprocs();
-        g_fHighCpuPressure = (avgCoreLoad > ((1 << SI_LOAD_SHIFT) * 0.9));
-        if (g_fHighCpuPressure)
+        int fHighPressureNew = (avgCoreLoad > ((1 << SI_LOAD_SHIFT) * 0.9));
+        __atomic_store(&g_fHighCpuPressure, &fHighPressureNew, __ATOMIC_RELEASE);
+        if (fHighPressureNew)
             serverLog(!fHighPressurePrev ?  3 /*LL_WARNING*/ : 1 /* LL_VERBOSE */, "NOTICE: Detuning locks due to high load per core: %.2f%%", avgCoreLoad / (double)(1 << SI_LOAD_SHIFT)*100.0);
     }
 
-    if (!g_fHighCpuPressure && fHighPressurePrev) {
+    if (!fHighPressureNew && fHighPressurePrev) {
         serverLog(3 /*LL_WARNING*/, "NOTICE: CPU pressure reduced");
     }
 #else
     g_fHighCpuPressure = g_fTestMode;
 #endif
-}
+}

src/help.h

@@ -1325,7 +1325,12 @@ struct commandHelp {
     "Rename a hash key, copying the value.",
     4,
     "6.5.3"
-    }
+    },
+    { "KEYDB.MEXISTS",
+    "key [key ...]",
+    "Determine if a key exists",
+    0,
+    "6.5.12" },
 };
 
 #endif

src/module.cpp

@@ -4895,14 +4895,17 @@ void RM_FreeThreadSafeContext(RedisModuleCtx *ctx) {
     zfree(ctx);
 }
 
+static bool g_fModuleThread = false;
 /* Acquire the server lock before executing a thread safe API call.
  * This is not needed for `RedisModule_Reply*` calls when there is
  * a blocked client connected to the thread safe context. */
 void RM_ThreadSafeContextLock(RedisModuleCtx *ctx) {
     UNUSED(ctx);
-    moduleAcquireGIL(FALSE /*fServerThread*/);
-    if (serverTL == nullptr)
+    if (serverTL == nullptr) {
         serverTL = &g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN];    // arbitrary module threads get the main thread context
+        g_fModuleThread = true;
+    }
+    moduleAcquireGIL(FALSE /*fServerThread*/);
 }
 
 /* Release the server lock after a thread safe API call was executed. */
@@ -4911,10 +4914,20 @@ void RM_ThreadSafeContextUnlock(RedisModuleCtx *ctx) {
     moduleReleaseGIL(FALSE /*fServerThread*/);
 }
 
+// A module may be triggered synchronously in a non-module context.  In this scenario we don't lock again
+//  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;
+}
 void moduleAcquireGIL(int fServerThread) {
     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);
 
@@ -4932,6 +4945,10 @@ void moduleAcquireGIL(int fServerThread) {
 void moduleReleaseGIL(int fServerThread) {
     std::unique_lock<std::mutex> lock(s_mutex);
 
+    if (FModuleCallBackLock(fServerThread)) {
+        return;
+    }
+
     if (fServerThread)
     {
         --s_cAcquisitionsServer;

src/multi.cpp

@@ -59,6 +59,13 @@ void queueMultiCommand(client *c) {
     multiCmd *mc;
     int j;
 
+    /* No sense to waste memory if the transaction is already aborted.
+     * this is useful in case client sends these in a pipeline, or doesn't
+     * bother to read previous responses and didn't notice the multi was already
+     * aborted. */
+    if (c->flags & CLIENT_DIRTY_EXEC)
+        return;
+
     c->mstate.commands = (multiCmd*)zrealloc(c->mstate.commands,
             sizeof(multiCmd)*(c->mstate.count+1), MALLOC_LOCAL);
     mc = c->mstate.commands+c->mstate.count;
@@ -131,6 +138,15 @@ void execCommand(client *c) {
         return;
     }
 
+    /* If we are in -BUSY state, flag the transaction and return the
+     * -BUSY error, like Redis <= 5. This is a temporary fix, may be changed
+     *  ASAP, see issue #7353 on Github. */
+    if (g_pserver->lua_timedout) {
+        flagTransaction(c);
+        addReply(c, shared.slowscripterr);
+        return;
+    }
+
     /* Check if we need to abort the EXEC because:
      * 1) Some WATCHed key was touched.
      * 2) There was a previous error while queueing commands.

src/networking.cpp

@@ -56,7 +56,7 @@ size_t getStringObjectSdsUsedMemory(robj *o) {
     serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);
     switch(o->encoding) {
     case OBJ_ENCODING_RAW: return sdsZmallocSize((sds)ptrFromObj(o));
-    case OBJ_ENCODING_EMBSTR: return zmalloc_size(o)-sizeof(robj);
+    case OBJ_ENCODING_EMBSTR: return zmalloc_size(allocPtrFromObj(o))-sizeof(robj);
     default: return 0; /* Just integer encoding for now. */
     }
 }
@@ -1264,9 +1264,10 @@ static void acceptCommonHandler(connection *conn, int flags, char *ip, int iel)
 void acceptOnThread(connection *conn, int flags, char *cip)
 {
     int ielCur = ielFromEventLoop(serverTL->el);
+    bool fBootLoad = (g_pserver->loading == LOADING_BOOT);
 
     int ielTarget = 0;
-    if (g_pserver->loading)
+    if (fBootLoad)
     {
         ielTarget = IDX_EVENT_LOOP_MAIN;    // During load only the main thread is active
     }
@@ -1290,10 +1291,10 @@ 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);
-            if (!g_fTestMode && !g_pserver->loading)
+            if (!g_fTestMode && !fBootLoad)
                 rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
             zfree(szT);
         });
@@ -1302,7 +1303,7 @@ void acceptOnThread(connection *conn, int flags, char *cip)
             return;
         // If res != AE_OK we can still try to accept on the local thread
     }
-    if (!g_fTestMode && !g_pserver->loading)
+    if (!g_fTestMode && !fBootLoad)
         rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
 
     aeAcquireLock();
@@ -1645,7 +1646,7 @@ int freeClientsInAsyncFreeQueue(int iel) {
     while((ln = listNext(&li))) 
     {
         client *c = (client*)listNodeValue(ln);
-        if (c->iel == iel)
+        if (c->iel == iel && !(c->flags & CLIENT_PROTECTED))
         {
             vecclientsFree.push_back(c);
             listDelNode(g_pserver->clients_to_close, ln);
@@ -1928,9 +1929,12 @@ int handleClientsWithPendingWrites(int iel, int aof_state) {
         }
     }
 
-    AeLocker locker;
-    locker.arm(nullptr);
-    ProcessPendingAsyncWrites();
+    if (listLength(serverTL->clients_pending_asyncwrite))
+    {
+        AeLocker locker;
+        locker.arm(nullptr);
+        ProcessPendingAsyncWrites();
+    }
 
     return processed;
 }
@@ -2256,7 +2260,7 @@ int processMultibulkBuffer(client *c) {
  * 1. The client is reset unless there are reasons to avoid doing it.
  * 2. In the case of master clients, the replication offset is updated.
  * 3. Propagate commands we got from our master to replicas down the line. */
-void commandProcessed(client *c) {
+void commandProcessed(client *c, int flags) {
     long long prev_offset = c->reploff;
     if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
         /* Update the applied replication offset of our master. */
@@ -2284,7 +2288,7 @@ void commandProcessed(client *c) {
             ae.arm(c);
         long long applied = c->reploff - prev_offset;
         if (applied) {
-            if (!g_pserver->fActiveReplica)
+            if (!g_pserver->fActiveReplica && (flags & CMD_CALL_PROPAGATE))
             {
                 replicationFeedSlavesFromMasterStream(g_pserver->slaves,
                     c->pending_querybuf, applied);
@@ -2305,9 +2309,10 @@ void commandProcessed(client *c) {
 int processCommandAndResetClient(client *c, int flags) {
     int deadclient = 0;
     serverTL->current_client = c;
-    AeLocker locker;
-    if (processCommand(c, flags, locker) == C_OK) {
-        commandProcessed(c);
+    serverAssert(GlobalLocksAcquired());
+    
+    if (processCommand(c, flags) == C_OK) {
+        commandProcessed(c, flags);
     }
     if (serverTL->current_client == NULL) deadclient = 1;
     serverTL->current_client = NULL;
@@ -2457,14 +2462,26 @@ void readQueryFromClient(connection *conn) {
         return;
     }
 
-    /* There is more data in the client input buffer, continue parsing it
-     * in case to check if there is a full command to execute. */
-    processInputBuffer(c, CMD_CALL_FULL);
+    serverTL->vecclientsProcess.push_back(c);
+}
+
+void processClients()
+{
+    serverAssert(GlobalLocksAcquired());
+
+    for (client *c : serverTL->vecclientsProcess) {
+        /* There is more data in the client input buffer, continue parsing it
+        * in case to check if there is a full command to execute. */
+        std::unique_lock<fastlock> ul(c->lock);
+        processInputBuffer(c, CMD_CALL_FULL);
+    }
+
     if (listLength(serverTL->clients_pending_asyncwrite))
     {
-        aelock.arm(c);
         ProcessPendingAsyncWrites();
     }
+    
+    serverTL->vecclientsProcess.clear();
 }
 
 void getClientsMaxBuffers(unsigned long *longest_output_list,
@@ -2840,7 +2857,7 @@ NULL
         if (c->name)
             addReplyBulk(c,c->name);
         else
-            addReplyNull(c, shared.nullbulk);
+            addReplyNull(c);
     } else if (!strcasecmp(szFromObj(c->argv[1]),"pause") && c->argc == 3) {
         /* CLIENT PAUSE */
         long long duration;
@@ -3057,7 +3074,7 @@ void helloCommand(client *c) {
     addReplyBulkCString(c,KEYDB_SET_VERSION);
 
     addReplyBulkCString(c,"proto");
-    addReplyLongLong(c,3);
+    addReplyLongLong(c,ver);
 
     addReplyBulkCString(c,"id");
     addReplyLongLong(c,c->id);
@@ -3422,7 +3439,7 @@ void processEventsWhileBlocked(int iel) {
     
 
     aeReleaseLock();
-    serverAssertDebug(!GlobalLocksAcquired());
+    serverAssert(!GlobalLocksAcquired());
     try
     {
         while (iterations--) {

src/object.cpp

@@ -41,12 +41,15 @@
 /* ===================== Creation and parsing of objects ==================== */
 
 robj *createObject(int type, void *ptr) {
-    robj *o = (robj*)zcalloc(sizeof(*o), MALLOC_SHARED);
+    size_t mvccExtraBytes = g_pserver->fActiveReplica ? sizeof(redisObjectExtended) : 0;
+    char *oB = (char*)zcalloc(sizeof(robj)+mvccExtraBytes, MALLOC_SHARED);
+    robj *o = reinterpret_cast<robj*>(oB + mvccExtraBytes);
+    
     o->type = type;
     o->encoding = OBJ_ENCODING_RAW;
     o->m_ptr = ptr;
     o->setrefcount(1);
-    o->mvcc_tstamp = OBJ_MVCC_INVALID;
+    setMvccTstamp(o, OBJ_MVCC_INVALID);
 
     /* Set the LRU to the current lruclock (minutes resolution), or
      * alternatively the LFU counter. */
@@ -95,13 +98,16 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
     size_t allocsize = sizeof(struct sdshdr8)+len+1;
     if (allocsize < sizeof(void*))
         allocsize = sizeof(void*);
-    robj *o = (robj*)zcalloc(sizeof(robj)+allocsize-sizeof(o->m_ptr), MALLOC_SHARED);
+
+    size_t mvccExtraBytes = g_pserver->fActiveReplica ? sizeof(redisObjectExtended) : 0;
+    char *oB = (char*)zcalloc(sizeof(robj)+allocsize-sizeof(redisObject::m_ptr)+mvccExtraBytes, MALLOC_SHARED);
+    robj *o = reinterpret_cast<robj*>(oB + mvccExtraBytes);
     struct sdshdr8 *sh = (sdshdr8*)(&o->m_ptr);
 
     o->type = OBJ_STRING;
     o->encoding = OBJ_ENCODING_EMBSTR;
     o->setrefcount(1);
-    o->mvcc_tstamp = OBJ_MVCC_INVALID;
+    setMvccTstamp(o, OBJ_MVCC_INVALID);
 
     if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
         o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
@@ -129,8 +135,9 @@ 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. */
-#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 48
-static_assert((sizeof(redisObject)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT-8) == 64, "Max EMBSTR obj should be 64 bytes total");
+#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) {
     if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT)
         return createEmbeddedStringObject(ptr,len);
@@ -390,7 +397,11 @@ void decrRefCount(robj_roptr o) {
         case OBJ_CRON: freeCronObject(o); break;
         default: serverPanic("Unknown object type"); break;
         }
-        zfree(o.unsafe_robjcast());
+        if (g_pserver->fActiveReplica) {
+            zfree(reinterpret_cast<redisObjectExtended*>(o.unsafe_robjcast())-1);
+        } else {
+            zfree(o.unsafe_robjcast());
+        }
     } else {
         if (prev <= 0) serverPanic("decrRefCount against refcount <= 0");
     }
@@ -1320,7 +1331,8 @@ NULL
     } else if (!strcasecmp(szFromObj(c->argv[1]), "lastmodified") && c->argc == 3) {
         if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
                 == NULL) return;
-        addReplyLongLong(c, (g_pserver->mstime - (o->mvcc_tstamp >> MVCC_MS_SHIFT)) / 1000);
+        uint64_t mvcc = mvccFromObj(o);
+        addReplyLongLong(c, (g_pserver->mstime - (mvcc >> MVCC_MS_SHIFT)) / 1000);
     } else {
         addReplySubcommandSyntaxError(c);
     }
@@ -1362,7 +1374,7 @@ NULL
             }
         }
         if ((de = dictFind(c->db->pdict,ptrFromObj(c->argv[2]))) == NULL) {
-            addReplyNull(c, shared.nullbulk);
+            addReplyNull(c);
             return;
         }
         size_t usage = objectComputeSize((robj*)dictGetVal(de),samples);
@@ -1500,3 +1512,39 @@ void redisObject::setrefcount(unsigned ref)
     serverAssert(!FExpires());
     refcount.store(ref, std::memory_order_relaxed); 
 }
+
+redisObjectStack::redisObjectStack()
+{
+    // We need to ensure the Extended Object is first in the class layout
+    serverAssert(reinterpret_cast<ptrdiff_t>(static_cast<redisObject*>(this)) != reinterpret_cast<ptrdiff_t>(this));
+}
+
+void *allocPtrFromObj(robj_roptr o) {
+    if (g_pserver->fActiveReplica)
+        return reinterpret_cast<redisObjectExtended*>(o.unsafe_robjcast()) - 1;
+    return o.unsafe_robjcast();
+}
+
+robj *objFromAllocPtr(void *pv) {
+    if (g_pserver->fActiveReplica) {
+        return reinterpret_cast<robj*>(reinterpret_cast<redisObjectExtended*>(pv)+1);
+    } 
+    return reinterpret_cast<robj*>(pv);
+}
+
+uint64_t mvccFromObj(robj_roptr o)
+{
+    if (g_pserver->fActiveReplica) {
+        redisObjectExtended *oe = reinterpret_cast<redisObjectExtended*>(o.unsafe_robjcast()) - 1;
+        return oe->mvcc_tstamp;
+    }
+    return OBJ_MVCC_INVALID;
+}
+
+void setMvccTstamp(robj *o, uint64_t mvcc)
+{
+    if (!g_pserver->fActiveReplica)
+        return;
+    redisObjectExtended *oe = reinterpret_cast<redisObjectExtended*>(o) - 1;
+    oe->mvcc_tstamp = mvcc;
+}

src/pubsub.cpp

@@ -89,7 +89,7 @@ void addReplyPubsubUnsubscribed(client *c, robj *channel) {
     if (channel)
         addReplyBulk(c,channel);
     else
-        addReplyNull(c, shared.nullbulk);
+        addReplyNull(c);
     addReplyLongLong(c,clientSubscriptionsCount(c));
 }
 
@@ -117,7 +117,7 @@ void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) {
     if (pattern)
         addReplyBulk(c,pattern);
     else
-        addReplyNull(c, shared.nullbulk);
+        addReplyNull(c);
     addReplyLongLong(c,clientSubscriptionsCount(c));
 }
 

src/rdb.cpp

@@ -1089,8 +1089,10 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, expireEntry *pexpire) {
     }
 
     char szT[32];
-    snprintf(szT, 32, "%" PRIu64, val->mvcc_tstamp);
-    if (rdbSaveAuxFieldStrStr(rdb,"mvcc-tstamp", szT) == -1) return -1;
+    if (g_pserver->fActiveReplica) {
+        snprintf(szT, 32, "%" PRIu64, mvccFromObj(val));
+        if (rdbSaveAuxFieldStrStr(rdb,"mvcc-tstamp", szT) == -1) return -1;
+    }
 
     /* Save type, key, value */
     if (rdbSaveObjectType(rdb,val) == -1) return -1;
@@ -1144,7 +1146,7 @@ int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
 
 int saveKey(rio *rdb, redisDb *db, int flags, size_t *processed, const char *keystr, robj *o)
 {    
-    robj key;
+    redisObjectStack key;
 
     initStaticStringObject(key,(char*)keystr);
     expireEntry *pexpire = getExpire(db, &key);
@@ -1997,7 +1999,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, uint64_t mvcc_tstamp) {
             exit(1);
         }
         RedisModuleIO io;
-        robj keyobj;
+        redisObjectStack keyobj;
         initStaticStringObject(keyobj,key);
         moduleInitIOContext(io,mt,rdb,&keyobj);
         io.ver = (rdbtype == RDB_TYPE_MODULE) ? 1 : 2;
@@ -2046,7 +2048,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, uint64_t mvcc_tstamp) {
         return NULL;
     }
 
-    o->mvcc_tstamp = mvcc_tstamp;
+    setMvccTstamp(o, mvcc_tstamp);
     serverAssert(!o->FExpires());
     return o;
 }
@@ -2055,7 +2057,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, uint64_t mvcc_tstamp) {
  * needed to provide loading stats. */
 void startLoading(size_t size, int rdbflags) {
     /* Load the DB */
-    g_pserver->loading = 1;
+    g_pserver->loading = (rdbflags & RDBFLAGS_REPLICATION) ? LOADING_REPLICATION : LOADING_BOOT;
     g_pserver->loading_start_time = time(NULL);
     g_pserver->loading_loaded_bytes = 0;
     g_pserver->loading_total_bytes = size;
@@ -2314,7 +2316,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
                     }
                 }
                 else {
-                    redisObject keyobj;
+                    redisObjectStack keyobj;
                     initStaticStringObject(keyobj,key);
                     setExpire(NULL, db, &keyobj, subexpireKey, strtoll(szFromObj(auxval), nullptr, 10));
                     decrRefCount(subexpireKey);
@@ -2398,23 +2400,22 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
             key = nullptr;
             goto eoferr;
         }
-        bool fStaleMvccKey = (rsi) ? val->mvcc_tstamp < rsi->mvccMinThreshold : false;
+        bool fStaleMvccKey = (rsi) ? mvccFromObj(val) < rsi->mvccMinThreshold : false;
 
         /* Check if the key already expired. This function is used when loading
          * an RDB file from disk, either at startup, or when an RDB was
          * received from the master. In the latter case, the master is
          * responsible for key expiry. If we would expire keys here, the
          * snapshot taken by the master may not be reflected on the replica. */
-        robj keyobj;
+        redisObjectStack keyobj;
         initStaticStringObject(keyobj,key);
         bool fExpiredKey = iAmMaster() && !(rdbflags&RDBFLAGS_AOF_PREAMBLE) && expiretime != -1 && expiretime < now;
         if (fStaleMvccKey || fExpiredKey) {
             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 *objKeyDup = createStringObject(key, sdslen(key));
-                rsi->mi->staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);
-                decrRefCount(objKeyDup);
+                robj_sharedptr objKeyDup(createStringObject(key, sdslen(key)));
+                rsi->mi->staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);                
             }
             fLastKeyExpired = true;
             sdsfree(key);

src/replication.cpp

@@ -1166,6 +1166,7 @@ LError:
  * full resync. */
 void replconfCommand(client *c) {
     int j;
+    bool fCapaCommand = false;
 
     if ((c->argc % 2) == 0) {
         /* Number of arguments must be odd to make sure that every
@@ -1176,6 +1177,7 @@ void replconfCommand(client *c) {
 
     /* Process every option-value pair. */
     for (j = 1; j < c->argc; j+=2) {
+        fCapaCommand = false;
         if (!strcasecmp((const char*)ptrFromObj(c->argv[j]),"listening-port")) {
             long port;
 
@@ -1200,6 +1202,8 @@ void replconfCommand(client *c) {
                 c->slave_capa |= SLAVE_CAPA_PSYNC2;
             else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "activeExpire"))
                 c->slave_capa |= SLAVE_CAPA_ACTIVE_EXPIRE;
+
+            fCapaCommand = true;
         } else if (!strcasecmp((const char*)ptrFromObj(c->argv[j]),"ack")) {
             /* REPLCONF ACK is used by replica to inform the master the amount
              * of replication stream that it processed so far. It is an
@@ -1242,7 +1246,16 @@ void replconfCommand(client *c) {
             return;
         }
     }
-    addReply(c,shared.ok);
+
+    if (fCapaCommand) {
+        sds reply = sdsnew("+OK");
+        if (g_pserver->fActiveReplica)
+            reply = sdscat(reply, " active-replica");
+        reply = sdscat(reply, "\r\n");
+        addReplySds(c, reply);
+    } else {
+        addReply(c,shared.ok);
+    }
 }
 
 /* This function puts a replica in the online state, and should be called just
@@ -1370,30 +1383,28 @@ void sendBulkToSlave(connection *conn) {
      * try to use sendfile system call if supported, unless tls is enabled.
      * fallback to normal read+write otherwise. */
     nwritten = 0;
-    if (!nwritten) {
-        ssize_t buflen;
-        char buf[PROTO_IOBUF_LEN];
+    ssize_t buflen;
+    char buf[PROTO_IOBUF_LEN];
 
-        lseek(replica->repldbfd,replica->repldboff,SEEK_SET);
-        buflen = read(replica->repldbfd,buf,PROTO_IOBUF_LEN);
-        if (buflen <= 0) {
-            serverLog(LL_WARNING,"Read error sending DB to replica: %s",
-                (buflen == 0) ? "premature EOF" : strerror(errno));
+    lseek(replica->repldbfd,replica->repldboff,SEEK_SET);
+    buflen = read(replica->repldbfd,buf,PROTO_IOBUF_LEN);
+    if (buflen <= 0) {
+        serverLog(LL_WARNING,"Read error sending DB to replica: %s",
+            (buflen == 0) ? "premature EOF" : strerror(errno));
+        ul.unlock();
+        aeLock.arm(nullptr);
+        freeClient(replica);
+        return;
+    }
+    if ((nwritten = connWrite(conn,buf,buflen)) == -1) {
+        if (connGetState(conn) != CONN_STATE_CONNECTED) {
+            serverLog(LL_WARNING,"Write error sending DB to replica: %s",
+                connGetLastError(conn));
             ul.unlock();
             aeLock.arm(nullptr);
             freeClient(replica);
-            return;
-        }
-        if ((nwritten = connWrite(conn,buf,buflen)) == -1) {
-            if (connGetState(conn) != CONN_STATE_CONNECTED) {
-                serverLog(LL_WARNING,"Write error sending DB to replica: %s",
-                    connGetLastError(conn));
-                ul.unlock();
-                aeLock.arm(nullptr);
-                freeClient(replica);
-            }
-            return;
         }
+        return;
     }
 
     replica->repldboff += nwritten;
@@ -2069,7 +2080,6 @@ void readSyncBulkPayload(connection *conn) {
      *
      * 2. Or when we are done reading from the socket to the RDB file, in
      *    such case we want just to read the RDB file in memory. */
-    serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
 
     /* We need to stop any AOF rewriting child before flusing and parsing
      * the RDB, otherwise we'll create a copy-on-write disaster. */
@@ -2089,7 +2099,10 @@ void readSyncBulkPayload(connection *conn) {
      * (Where disklessLoadMakeBackups left server.db empty) because we
      * want to execute all the auxiliary logic of emptyDb (Namely,
      * fire module events) */
-    emptyDb(-1,empty_db_flags,replicationEmptyDbCallback);
+    if (!fUpdate) {
+        serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
+        emptyDb(-1,empty_db_flags,replicationEmptyDbCallback);
+    }
 
     /* Before loading the DB into memory we need to delete the readable
      * handler, otherwise it will get called recursively since
@@ -2557,6 +2570,30 @@ int slaveTryPartialResynchronization(redisMaster *mi, connection *conn, int read
     return PSYNC_NOT_SUPPORTED;
 }
 
+void parseMasterCapa(redisMaster *mi, sds strcapa)
+{
+    if (sdslen(strcapa) < 1 || strcapa[0] != '+')
+        return;
+
+    char *szStart = strcapa + 1;    // skip the +
+    char *pchEnd = szStart;
+
+    mi->isActive = false;
+    for (;;)
+    {
+        if (*pchEnd == ' ' || *pchEnd == '\0') {
+            // Parse the word
+            if (strncmp(szStart, "active-replica", pchEnd - szStart) == 0) {
+                mi->isActive = true;
+            }
+            szStart = pchEnd + 1;
+        }
+        if (*pchEnd == '\0')
+            break;
+        ++pchEnd;
+    }
+}
+
 /* This handler fires when the non blocking connect was able to
  * establish a connection with the master. */
 void syncWithMaster(connection *conn) {
@@ -2750,16 +2787,8 @@ void syncWithMaster(connection *conn) {
      *
      * The master will ignore capabilities it does not understand. */
     if (mi->repl_state == REPL_STATE_SEND_CAPA) {
-        if (g_pserver->fActiveReplica)
-        {
-            err = sendSynchronousCommand(mi, SYNC_CMD_WRITE,conn,"REPLCONF",
-                    "capa","eof","capa","psync2","capa","activeExpire",NULL);
-        }
-        else
-        {
-            err = sendSynchronousCommand(mi, SYNC_CMD_WRITE,conn,"REPLCONF",
-                    "capa","eof","capa","psync2",NULL);
-        }
+        err = sendSynchronousCommand(mi, SYNC_CMD_WRITE,conn,"REPLCONF",
+                "capa","eof","capa","psync2","capa","activeExpire",NULL);
         if (err) goto write_error;
         sdsfree(err);
         mi->repl_state = REPL_STATE_RECEIVE_CAPA;
@@ -2774,6 +2803,8 @@ void syncWithMaster(connection *conn) {
         if (err[0] == '-') {
             serverLog(LL_NOTICE,"(Non critical) Master does not understand "
                                   "REPLCONF capa: %s", err);
+        } else {
+            parseMasterCapa(mi, err);
         }
         sdsfree(err);
         mi->repl_state = REPL_STATE_SEND_PSYNC;
@@ -3151,15 +3182,46 @@ void replicaofCommand(client *c) {
         return;
     }
 
-    /* The special host/port combination "NO" "ONE" turns the instance
-     * into a master. Otherwise the new master address is set. */
-    if (!strcasecmp((const char*)ptrFromObj(c->argv[1]),"no") &&
+    if (c->argc > 3) {
+        if (c->argc != 4) {
+            addReplyError(c, "Invalid arguments");
+            return;
+        }
+        if (!strcasecmp((const char*)ptrFromObj(c->argv[1]),"remove")) {
+            listIter li;
+            listNode *ln;
+            bool fRemoved = false;
+            long port;
+            string2l(szFromObj(c->argv[3]), sdslen(szFromObj(c->argv[3])), &port);
+        LRestart:
+            listRewind(g_pserver->masters, &li);
+            while ((ln = listNext(&li))) {
+                redisMaster *mi = (redisMaster*)listNodeValue(ln);
+                if (mi->masterport != port)
+                    continue;
+                if (sdscmp(szFromObj(c->argv[2]), mi->masterhost) == 0) {
+                    replicationUnsetMaster(mi);
+                    fRemoved = true;
+                    goto LRestart;
+                }
+            }
+            if (!fRemoved) {
+                addReplyError(c, "Master not found");
+                return;
+            } else if (listLength(g_pserver->masters) == 0) {
+                goto LLogNoMaster;
+            }
+        }
+    } else if (!strcasecmp((const char*)ptrFromObj(c->argv[1]),"no") &&
         !strcasecmp((const char*)ptrFromObj(c->argv[2]),"one")) {
+        /* The special host/port combination "NO" "ONE" turns the instance
+         * into a master. Otherwise the new master address is set. */
         if (listLength(g_pserver->masters)) {
             while (listLength(g_pserver->masters))
             {
                 replicationUnsetMaster((redisMaster*)listNodeValue(listFirst(g_pserver->masters)));
             }
+        LLogNoMaster:
             sds client = catClientInfoString(sdsempty(),c);
             serverLog(LL_NOTICE,"MASTER MODE enabled (user request from '%s')",
                 client);
@@ -3976,12 +4038,20 @@ int FBrokenLinkToMaster()
     listNode *ln;
     listRewind(g_pserver->masters, &li);
 
+    int connected = 0;
     while ((ln = listNext(&li)))
     {
         redisMaster *mi = (redisMaster*)listNodeValue(ln);
-        if (mi->repl_state != REPL_STATE_CONNECTED)
-            return true;
+        if (mi->repl_state == REPL_STATE_CONNECTED)
+            ++connected;
     }
+
+    if (g_pserver->repl_quorum < 0) {
+        return connected < (int)listLength(g_pserver->masters);
+    } else {
+        return connected < g_pserver->repl_quorum;
+    }
+
     return false;
 }
 
@@ -4218,7 +4288,7 @@ void replicaReplayCommand(client *c)
     serverTL->current_client = current_clientSave;
 
     // call() will not propogate this for us, so we do so here
-    if (!s_pstate->FCancelled() && s_pstate->FFirst())
+    if (!s_pstate->FCancelled() && s_pstate->FFirst() && !cserver.multimaster_no_forward)
         alsoPropagate(cserver.rreplayCommand,c->db->id,c->argv,c->argc,PROPAGATE_AOF|PROPAGATE_REPL);
     
     s_pstate->Pop();

src/server.cpp

@@ -247,6 +247,10 @@ struct redisCommand redisCommandTable[] = {
      "read-only fast @keyspace",
      0,NULL,1,-1,1,0,0,0},
 
+    {"keydb.mexists",mexistsCommand,-2,
+     "read-only fast @keyspace",
+     0,NULL,1,-1,1,0,0,0},
+
     {"setbit",setbitCommand,4,
      "write use-memory @bitmap",
      0,NULL,1,1,1,0,0,0},
@@ -768,7 +772,7 @@ struct redisCommand redisCommandTable[] = {
      "admin no-script ok-stale",
      0,NULL,0,0,0,0,0,0},
 
-    {"replicaof",replicaofCommand,3,
+    {"replicaof",replicaofCommand,-3,
      "admin no-script ok-stale",
      0,NULL,0,0,0,0,0,0},
 
@@ -809,11 +813,11 @@ struct redisCommand redisCommandTable[] = {
      0,NULL,0,0,0,0,0,0},
 
     {"watch",watchCommand,-2,
-     "no-script fast @transaction",
+     "no-script fast ok-loading ok-stale @transaction",
      0,NULL,1,-1,1,0,0,0},
 
     {"unwatch",unwatchCommand,1,
-     "no-script fast @transaction",
+     "no-script fast ok-loading ok-stale @transaction",
      0,NULL,0,0,0,0,0,0},
 
     {"cluster",clusterCommand,-2,
@@ -1731,6 +1735,18 @@ void clientsCron(int iel) {
     freeClientsInAsyncFreeQueue(iel);
 }
 
+bool expireOwnKeys()
+{
+    if (iAmMaster()) {
+        return true;
+    } else if (!g_pserver->fActiveReplica && (listLength(g_pserver->masters) == 1)) {
+        redisMaster *mi = (redisMaster*)listNodeValue(listFirst(g_pserver->masters));
+        if (mi->isActive)
+            return true;
+    }
+    return false;
+}
+
 /* This function handles 'background' operations we are required to do
  * incrementally in Redis databases, such as active key expiring, resizing,
  * rehashing. */
@@ -1738,7 +1754,7 @@ void databasesCron(void) {
     /* Expire keys by random sampling. Not required for slaves
      * as master will synthesize DELs for us. */
     if (g_pserver->active_expire_enabled) {
-        if (iAmMaster()) {
+        if (expireOwnKeys()) {
             activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
         } else {
             expireSlaveKeys();
@@ -2181,6 +2197,7 @@ int serverCronLite(struct aeEventLoop *eventLoop, long long id, void *clientData
 }
 
 extern int ProcessingEventsWhileBlocked;
+void processClients();
 
 /* This function gets called every time Redis is entering the
  * main loop of the event driven library, that is, before to sleep
@@ -2199,6 +2216,7 @@ extern int ProcessingEventsWhileBlocked;
 void beforeSleep(struct aeEventLoop *eventLoop) {
     UNUSED(eventLoop);
     int iel = ielFromEventLoop(eventLoop);
+    processClients();
 
     /* Handle precise timeouts of blocked clients. */
     handleBlockedClientsTimeout();
@@ -2283,6 +2301,7 @@ void beforeSleepLite(struct aeEventLoop *eventLoop)
     
     /* Try to process pending commands for clients that were just unblocked. */
     aeAcquireLock();
+    processClients();
     if (listLength(g_pserver->rgthreadvar[iel].unblocked_clients)) {
         processUnblockedClients(iel);
     }
@@ -2454,6 +2473,7 @@ void initMasterInfo(redisMaster *master)
     master->cached_master = NULL;
     master->master_initial_offset = -1;
     
+    master->isActive = false;
 
     master->repl_state = REPL_STATE_NONE;
     master->repl_down_since = 0; /* Never connected, repl is down since EVER. */
@@ -3544,7 +3564,7 @@ void call(client *c, int flags) {
             !(flags & CMD_CALL_PROPAGATE_AOF))
                 propagate_flags &= ~PROPAGATE_AOF;
 
-        if (c->cmd->flags & CMD_SKIP_PROPOGATE)
+        if ((c->cmd->flags & CMD_SKIP_PROPOGATE) && g_pserver->fActiveReplica)
             propagate_flags &= ~PROPAGATE_REPL;
 
         /* Call propagate() only if at least one of AOF / replication
@@ -3629,12 +3649,12 @@ void call(client *c, int flags) {
  * If C_OK is returned the client is still alive and valid and
  * other operations can be performed by the caller. Otherwise
  * if C_ERR is returned the client was destroyed (i.e. after QUIT). */
-int processCommand(client *c, int callFlags, AeLocker &locker) {
+int processCommand(client *c, int callFlags) {
     AssertCorrectThread(c);
+    serverAssert(GlobalLocksAcquired());
 
     if (moduleHasCommandFilters())
     {
-        locker.arm(c);
         moduleCallCommandFilters(c);
     }
 
@@ -3686,9 +3706,6 @@ int processCommand(client *c, int callFlags, AeLocker &locker) {
 
     /* Check if the user can run this command according to the current
      * ACLs. */
-    if (c->puser && !(c->puser->flags & USER_FLAG_ALLCOMMANDS))
-        locker.arm(c);  // ACLs require the lock
-
     int acl_keypos;
     int acl_retval = ACLCheckCommandPerm(c,&acl_keypos);
     if (acl_retval != ACL_OK) {
@@ -3716,7 +3733,6 @@ int processCommand(client *c, int callFlags, AeLocker &locker) {
         !(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0 &&
           c->cmd->proc != execCommand))
     {
-        locker.arm(c);
         int hashslot;
         int error_code;
         clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,
@@ -3731,9 +3747,6 @@ int processCommand(client *c, int callFlags, AeLocker &locker) {
             return C_OK;
         }
     }
-    
-    if (!locker.isArmed())
-        locker.arm(c);
 
     incrementMvccTstamp();
 
@@ -3849,8 +3862,12 @@ int processCommand(client *c, int callFlags, AeLocker &locker) {
     /* Loading DB? Return an error if the command has not the
      * CMD_LOADING flag. */
     if (g_pserver->loading && !(c->cmd->flags & CMD_LOADING)) {
-        addReply(c, shared.loadingerr);
-        return C_OK;
+        /* Active Replicas can execute read only commands, and optionally write commands */
+        if (!(g_pserver->loading == LOADING_REPLICATION && g_pserver->fActiveReplica && ((c->cmd->flags & CMD_READONLY) || g_pserver->fWriteDuringActiveLoad)))
+        {
+            addReply(c, shared.loadingerr);
+            return C_OK;
+        }
     }
 
     /* Lua script too slow? Only allow a limited number of commands.
@@ -3866,6 +3883,8 @@ int processCommand(client *c, int callFlags, AeLocker &locker) {
           c->cmd->proc != multiCommand &&
           c->cmd->proc != execCommand &&
           c->cmd->proc != discardCommand &&
+          c->cmd->proc != watchCommand &&
+          c->cmd->proc != unwatchCommand &&
         !(c->cmd->proc == shutdownCommand &&
           c->argc == 2 &&
           tolower(((char*)ptrFromObj(c->argv[1]))[0]) == 'n') &&
@@ -4488,7 +4507,7 @@ sds genRedisInfoString(const char *section) {
             "aof_last_cow_size:%zu\r\n"
             "module_fork_in_progress:%d\r\n"
             "module_fork_last_cow_size:%zu\r\n",
-            g_pserver->loading.load(std::memory_order_relaxed),
+            !!g_pserver->loading.load(std::memory_order_relaxed),   /* Note: libraries expect 1 or 0 here so coerce our enum */
             g_pserver->dirty,
             g_pserver->rdb_child_pid != -1,
             (intmax_t)g_pserver->lastsave,
@@ -4637,64 +4656,61 @@ sds genRedisInfoString(const char *section) {
             listLength(g_pserver->masters) == 0 ? "master" 
                 : g_pserver->fActiveReplica ? "active-replica" : "slave");
         if (listLength(g_pserver->masters)) {
-            listIter li;
-            listNode *ln;
-            listRewind(g_pserver->masters, &li);
-            bool fAllUp = true;
-            while ((ln = listNext(&li))) {
-                redisMaster *mi = (redisMaster*)listNodeValue(ln);
-                fAllUp = fAllUp && mi->repl_state == REPL_STATE_CONNECTED;
-            }
-
             info = sdscatprintf(info, "master_global_link_status:%s\r\n",
-                fAllUp ? "up" : "down");
+                FBrokenLinkToMaster() ? "down" : "up");
 
             int cmasters = 0;
+            listIter li;
+            listNode *ln;
             listRewind(g_pserver->masters, &li);
             while ((ln = listNext(&li)))
             {
                 long long slave_repl_offset = 1;
                 redisMaster *mi = (redisMaster*)listNodeValue(ln);
-                info = sdscatprintf(info, "Master %d: \r\n", cmasters);
-                ++cmasters;
 
                 if (mi->master)
                     slave_repl_offset = mi->master->reploff;
                 else if (mi->cached_master)
                     slave_repl_offset = mi->cached_master->reploff;
 
+                char master_prefix[128] = "";
+                if (cmasters != 0) {
+                    snprintf(master_prefix, sizeof(master_prefix), "_%d", cmasters);
+                }
+
                 info = sdscatprintf(info,
-                    "master_host:%s\r\n"
-                    "master_port:%d\r\n"
-                    "master_link_status:%s\r\n"
-                    "master_last_io_seconds_ago:%d\r\n"
-                    "master_sync_in_progress:%d\r\n"
+                    "master%s_host:%s\r\n"
+                    "master%s_port:%d\r\n"
+                    "master%s_link_status:%s\r\n"
+                    "master%s_last_io_seconds_ago:%d\r\n"
+                    "master%s_sync_in_progress:%d\r\n"
                     "slave_repl_offset:%lld\r\n"
-                    ,mi->masterhost,
-                    mi->masterport,
-                    (mi->repl_state == REPL_STATE_CONNECTED) ?
+                    ,master_prefix, mi->masterhost,
+                    master_prefix, mi->masterport,
+                    master_prefix, (mi->repl_state == REPL_STATE_CONNECTED) ?
                         "up" : "down",
-                    mi->master ?
+                    master_prefix, mi->master ?
                     ((int)(g_pserver->unixtime-mi->master->lastinteraction)) : -1,
-                    mi->repl_state == REPL_STATE_TRANSFER,
+                    master_prefix, mi->repl_state == REPL_STATE_TRANSFER,
                     slave_repl_offset
                 );
 
                 if (mi->repl_state == REPL_STATE_TRANSFER) {
                     info = sdscatprintf(info,
-                        "master_sync_left_bytes:%lld\r\n"
-                        "master_sync_last_io_seconds_ago:%d\r\n"
-                        , (long long)
+                        "master%s_sync_left_bytes:%lld\r\n"
+                        "master%s_sync_last_io_seconds_ago:%d\r\n"
+                        , master_prefix, (long long)
                             (mi->repl_transfer_size - mi->repl_transfer_read),
-                        (int)(g_pserver->unixtime-mi->repl_transfer_lastio)
+                        master_prefix, (int)(g_pserver->unixtime-mi->repl_transfer_lastio)
                     );
                 }
 
                 if (mi->repl_state != REPL_STATE_CONNECTED) {
                     info = sdscatprintf(info,
-                        "master_link_down_since_seconds:%jd\r\n",
-                        (intmax_t)g_pserver->unixtime-mi->repl_down_since);
+                        "master%s_link_down_since_seconds:%jd\r\n",
+                        master_prefix, (intmax_t)g_pserver->unixtime-mi->repl_down_since);
                 }
+                ++cmasters;
             }
             info = sdscatprintf(info,
                 "slave_priority:%d\r\n"

src/server.h

@@ -94,6 +94,7 @@ typedef long long ustime_t; /* microsecond time type. */
 #include "semiorderedset.h"
 #include "connection.h" /* Connection abstraction */
 #include "serverassert.h"
+#include "expire.h"
 
 #define REDISMODULE_CORE 1
 #include "redismodule.h"    /* Redis modules API defines. */
@@ -104,6 +105,9 @@ typedef long long ustime_t; /* microsecond time type. */
 #include "endianconv.h"
 #include "crc64.h"
 
+#define LOADING_BOOT 1
+#define LOADING_REPLICATION 2
+
 extern int g_fTestMode;
 
 struct redisObject;
@@ -166,22 +170,24 @@ class robj_sharedptr
 
 public:
     robj_sharedptr()
-        : m_ptr(nullptr)
-        {}
-    robj_sharedptr(redisObject *ptr)
-        : m_ptr(ptr)
-        {
+    : m_ptr(nullptr)
+    {}
+    explicit robj_sharedptr(redisObject *ptr)
+    : m_ptr(ptr)
+    {
+        if(m_ptr)
             incrRefCount(ptr);
-        }
+    }
     ~robj_sharedptr()
     {
         if (m_ptr)
             decrRefCount(m_ptr);
     }
     robj_sharedptr(const robj_sharedptr& other)
-    {
-        m_ptr = other.m_ptr;
-        incrRefCount(m_ptr);
+    : m_ptr(other.m_ptr)
+    {        
+        if(m_ptr)
+            incrRefCount(m_ptr);
     }
 
     robj_sharedptr(robj_sharedptr&& other)
@@ -192,41 +198,19 @@ public:
 
     robj_sharedptr &operator=(const robj_sharedptr& other)
     {
-        if (m_ptr)
-            decrRefCount(m_ptr);
-        m_ptr = other.m_ptr;
-        incrRefCount(m_ptr);
+        robj_sharedptr tmp(other);
+        using std::swap;
+        swap(m_ptr, tmp.m_ptr);
         return *this;
     }
     robj_sharedptr &operator=(redisObject *ptr)
     {
-        if (m_ptr)
-            decrRefCount(m_ptr);
-        m_ptr = ptr;
-        incrRefCount(m_ptr);
+        robj_sharedptr tmp(ptr);
+        using std::swap;
+        swap(m_ptr, tmp.m_ptr);
         return *this;
     }
-
-    bool operator==(const robj_sharedptr &other) const
-    {
-        return m_ptr == other.m_ptr;
-    }
-
-    bool operator==(const void *p) const
-    {
-        return m_ptr == p;
-    }
-
-    bool operator!=(const robj_sharedptr &other) const
-    {
-        return m_ptr != other.m_ptr;
-    }
-
-    bool operator!=(const void *p) const
-    {
-        return m_ptr != p;
-    }
-
+    
     redisObject* operator->() const
     {
         return m_ptr;
@@ -237,7 +221,7 @@ public:
         return !m_ptr;
     }
 
-    operator bool() const{
+    explicit operator bool() const{
         return !!m_ptr;
     }
 
@@ -247,8 +231,39 @@ public:
     }
 
     redisObject *get() { return m_ptr; }
+    const redisObject *get() const { return m_ptr; }
 };
 
+inline bool operator==(const robj_sharedptr &lhs, const robj_sharedptr &rhs)
+{
+    return lhs.get() == rhs.get();
+}
+
+inline bool operator!=(const robj_sharedptr &lhs, const robj_sharedptr &rhs)
+{
+    return !(lhs == rhs);
+}
+
+inline bool operator==(const robj_sharedptr &lhs, const void *p)
+{
+    return lhs.get() == p;
+}
+
+inline bool operator==(const void *p, const robj_sharedptr &rhs)
+{
+    return rhs == p;
+}
+
+inline bool operator!=(const robj_sharedptr &lhs, const void *p)
+{
+    return !(lhs == p);
+}
+
+inline bool operator!=(const void *p, const robj_sharedptr &rhs)
+{
+    return !(rhs == p);
+}
+
 /* Error codes */
 #define C_OK                    0
 #define C_ERR                   -1
@@ -785,7 +800,16 @@ typedef struct RedisModuleDigest {
 
 #define MVCC_MS_SHIFT 20
 
-typedef struct redisObject {
+// This struct will be allocated ahead of the ROBJ when needed
+struct redisObjectExtended {
+    uint64_t mvcc_tstamp;
+};
+
+typedef class redisObject {
+protected:
+    redisObject() {}
+
+public:
     unsigned type:4;
     unsigned encoding:4;
     unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
@@ -794,7 +818,6 @@ typedef struct redisObject {
 private:
     mutable std::atomic<unsigned> refcount {0};
 public:
-    uint64_t mvcc_tstamp;
     void *m_ptr;
 
     inline bool FExpires() const { return refcount.load(std::memory_order_relaxed) >> 31; }
@@ -805,7 +828,18 @@ public:
     void addref() const { refcount.fetch_add(1, std::memory_order_relaxed); }
     unsigned release() const { return refcount.fetch_sub(1, std::memory_order_seq_cst) & ~(1U << 31); }
 } robj;
-static_assert(sizeof(redisObject) == 24, "object size is critical, don't increase");
+static_assert(sizeof(redisObject) == 16, "object size is critical, don't increase");
+
+class redisObjectStack : public redisObjectExtended, public redisObject
+{
+public:
+    redisObjectStack();
+};
+
+uint64_t mvccFromObj(robj_roptr o);
+void setMvccTstamp(redisObject *o, uint64_t mvcc);
+void *allocPtrFromObj(robj_roptr o);
+robj *objFromAllocPtr(void *pv);
 
 __attribute__((always_inline)) inline const void *ptrFromObj(robj_roptr &o)
 {
@@ -831,243 +865,6 @@ __attribute__((always_inline)) inline char *szFromObj(const robj *o)
     return (char*)ptrFromObj(o);
 }
 
-class expireEntryFat
-{
-    friend class expireEntry;
-public:
-    struct subexpireEntry
-    {
-        long long when;
-        std::unique_ptr<const char, void(*)(const char*)> spsubkey;
-
-        subexpireEntry(long long when, const char *subkey)
-            : when(when), spsubkey(subkey, sdsfree)
-        {}
-
-        bool operator<(long long when) const noexcept { return this->when < when; }
-        bool operator<(const subexpireEntry &se) { return this->when < se.when; }
-    };
-
-private:
-    sds m_keyPrimary;
-    std::vector<subexpireEntry> m_vecexpireEntries;  // Note a NULL for the sds portion means the expire is for the primary key
-
-public:
-    expireEntryFat(sds keyPrimary)
-        : m_keyPrimary(keyPrimary)
-        {}
-    long long when() const noexcept { return m_vecexpireEntries.front().when; }
-    const char *key() const noexcept { return m_keyPrimary; }
-
-    bool operator<(long long when) const noexcept { return this->when() <  when; }
-
-    void expireSubKey(const char *szSubkey, long long when)
-    {
-        // First check if the subkey already has an expiration
-        for (auto &entry : m_vecexpireEntries)
-        {
-            if (szSubkey != nullptr)
-            {
-                // if this is a subkey expiry then its not a match if the expireEntry is either for the
-                //  primary key or a different subkey
-                if (entry.spsubkey == nullptr || sdscmp((sds)entry.spsubkey.get(), (sds)szSubkey) != 0)
-                    continue;
-            }
-            else
-            {
-                if (entry.spsubkey != nullptr)
-                    continue;
-            }
-            m_vecexpireEntries.erase(m_vecexpireEntries.begin() + (&entry - m_vecexpireEntries.data()));
-            break;
-        }
-        auto itrInsert = std::lower_bound(m_vecexpireEntries.begin(), m_vecexpireEntries.end(), when);
-        const char *subkey = (szSubkey) ? sdsdup(szSubkey) : nullptr;
-        m_vecexpireEntries.emplace(itrInsert, when, subkey);
-    }
-
-    bool FEmpty() const noexcept { return m_vecexpireEntries.empty(); }
-    const subexpireEntry &nextExpireEntry() const noexcept { return m_vecexpireEntries.front(); }
-    void popfrontExpireEntry() { m_vecexpireEntries.erase(m_vecexpireEntries.begin()); }
-    const subexpireEntry &operator[](size_t idx) { return m_vecexpireEntries[idx]; }
-    size_t size() const noexcept { return m_vecexpireEntries.size(); }
-};
-
-class expireEntry {
-    union
-    {
-        sds m_key;
-        expireEntryFat *m_pfatentry;
-    } u;
-    long long m_when;   // LLONG_MIN means this is a fat entry and we should use the pointer
-
-public:
-    class iter
-    {
-        friend class expireEntry;
-        expireEntry *m_pentry = nullptr;
-        size_t m_idx = 0;
-
-    public:
-        iter(expireEntry *pentry, size_t idx)
-            : m_pentry(pentry), m_idx(idx)
-        {}
-
-        iter &operator++() { ++m_idx; return *this; }
-        
-        const char *subkey() const
-        {
-            if (m_pentry->FFat())
-                return (*m_pentry->pfatentry())[m_idx].spsubkey.get();
-            return nullptr;
-        }
-        long long when() const
-        {
-            if (m_pentry->FFat())
-                return (*m_pentry->pfatentry())[m_idx].when;
-            return m_pentry->when();
-        }
-
-        bool operator!=(const iter &other)
-        {
-            return m_idx != other.m_idx;
-        }
-
-        const iter &operator*() const { return *this; }
-    };
-
-    expireEntry(sds key, const char *subkey, long long when)
-    {
-        if (subkey != nullptr)
-        {
-            m_when = LLONG_MIN;
-            u.m_pfatentry = new (MALLOC_LOCAL) expireEntryFat(key);
-            u.m_pfatentry->expireSubKey(subkey, when);
-        }
-        else
-        {
-            u.m_key = key;
-            m_when = when;
-        }
-    }
-
-    expireEntry(expireEntryFat *pfatentry)
-    {
-        u.m_pfatentry = pfatentry;
-        m_when = LLONG_MIN;
-    }
-
-    expireEntry(expireEntry &&e)
-    {
-        u.m_key = e.u.m_key;
-        m_when = e.m_when;
-        e.u.m_key = (char*)key();  // we do this so it can still be found in the set
-        e.m_when = 0;
-    }
-
-    ~expireEntry()
-    {
-        if (FFat())
-            delete u.m_pfatentry;
-    }
-
-    void setKeyUnsafe(sds key)
-    {
-        if (FFat())
-            u.m_pfatentry->m_keyPrimary = key;
-        else
-            u.m_key = key;
-    }
-
-    inline bool FFat() const noexcept { return m_when == LLONG_MIN; }
-    expireEntryFat *pfatentry() { assert(FFat()); return u.m_pfatentry; }
-
-
-    bool operator==(const char *key) const noexcept
-    { 
-        return this->key() == key; 
-    }
-
-    bool operator<(const expireEntry &e) const noexcept
-    { 
-        return when() < e.when(); 
-    }
-    bool operator<(long long when) const noexcept
-    { 
-        return this->when() < when;
-    }
-
-    const char *key() const noexcept
-    { 
-        if (FFat())
-            return u.m_pfatentry->key();
-        return u.m_key;
-    }
-    long long when() const noexcept
-    { 
-        if (FFat())
-            return u.m_pfatentry->when();
-        return m_when; 
-    }
-
-    void update(const char *subkey, long long when)
-    {
-        if (!FFat())
-        {
-            if (subkey == nullptr)
-            {
-                m_when = when;
-                return;
-            }
-            else
-            {
-                // we have to upgrade to a fat entry
-                long long whenT = m_when;
-                sds keyPrimary = u.m_key;
-                m_when = LLONG_MIN;
-                u.m_pfatentry = new (MALLOC_LOCAL) expireEntryFat(keyPrimary);
-                u.m_pfatentry->expireSubKey(nullptr, whenT);
-                // at this point we're fat so fall through
-            }
-        }
-        u.m_pfatentry->expireSubKey(subkey, when);
-    }
-    
-    iter begin() { return iter(this, 0); }
-    iter end()
-    {
-        if (FFat())
-            return iter(this, u.m_pfatentry->size());
-        return iter(this, 1);
-    }
-    
-    void erase(iter &itr)
-    {
-        if (!FFat())
-            throw -1;   // assert
-        pfatentry()->m_vecexpireEntries.erase(
-            pfatentry()->m_vecexpireEntries.begin() + itr.m_idx);
-    }
-
-    bool FGetPrimaryExpire(long long *pwhen)
-    {
-        *pwhen = -1;
-        for (auto itr : *this)
-        {
-            if (itr.subkey() == nullptr)
-            {
-                *pwhen = itr.when();
-                return true;
-            }
-        }
-        return false;
-    }
-
-    explicit operator const char*() const noexcept { return key(); }
-    explicit operator long long() const noexcept { return when(); }
-};
-typedef semiorderedset<expireEntry, const char *, true /*expireEntry can be memmoved*/> expireset;
-
 /* The a string name for an object's type as listed above
  * Native types are checked against the OBJ_STRING, OBJ_LIST, OBJ_* defines,
  * and Module types have their registered name returned. */
@@ -1511,6 +1308,9 @@ typedef struct redisTLSContextConfig {
     char *ciphers;
     char *ciphersuites;
     int prefer_server_ciphers;
+    int session_caching;
+    int session_cache_size;
+    int session_cache_timeout;
 } redisTLSContextConfig;
 
 /*-----------------------------------------------------------------------------
@@ -1555,6 +1355,7 @@ struct redisServerThreadVars {
     char neterr[ANET_ERR_LEN];   /* Error buffer for anet.c */
     long unsigned commandsExecuted = 0;
     bool fRetrySetAofEvent = false;
+    std::vector<client*> vecclientsProcess;
 };
 
 struct redisMaster {
@@ -1570,6 +1371,7 @@ struct redisMaster {
     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. */
@@ -1637,6 +1439,7 @@ struct redisServerConst {
     unsigned char uuid[UUID_BINARY_LEN];         /* This server's UUID - populated on boot */
     bool fUsePro = false;
     int thread_min_client_threshold = 50;
+    int multimaster_no_forward;
 };
 
 struct redisServer {
@@ -1859,6 +1662,7 @@ struct redisServer {
     int repl_syncio_timeout; /* Timeout for synchronous I/O calls */
     int repl_disable_tcp_nodelay;   /* Disable TCP_NODELAY after SYNC? */
     int repl_serve_stale_data; /* Serve stale data when link is down? */
+    int repl_quorum;           /* For multimaster what do we consider a quorum? -1 means all master must be online */
     int repl_slave_ro;          /* Slave is read only? */
     int repl_slave_ignore_maxmemory;    /* If true slaves do not evict. */
     int slave_priority;             /* Reported in INFO and used by Sentinel. */
@@ -1979,6 +1783,7 @@ struct redisServer {
     int watchdog_period;  /* Software watchdog period in ms. 0 = off */
 
     int fActiveReplica;                          /* Can this replica also be a master? */
+    int fWriteDuringActiveLoad;                  /* Can this active-replica write during an RDB load? */
 
     // Format:
     //  Lower 20 bits: a counter incrementing for each command executed in the same millisecond
@@ -2545,7 +2350,7 @@ int getMaxmemoryState(size_t *total, size_t *logical, size_t *tofree, float *lev
 size_t freeMemoryGetNotCountedMemory();
 int freeMemoryIfNeeded(void);
 int freeMemoryIfNeededAndSafe(void);
-int processCommand(client *c, int callFlags, class AeLocker &locker);
+int processCommand(client *c, int callFlags);
 void setupSignalHandlers(void);
 struct redisCommand *lookupCommand(sds name);
 struct redisCommand *lookupCommandByCString(const char *s);
@@ -2810,6 +2615,7 @@ void getCommand(client *c);
 void delCommand(client *c);
 void unlinkCommand(client *c);
 void existsCommand(client *c);
+void mexistsCommand(client *c);
 void setbitCommand(client *c);
 void getbitCommand(client *c);
 void bitfieldCommand(client *c);

src/storage.h

@@ -1,8 +1,6 @@
 #ifndef __STORAGE_H__
 #define __STORAGE_H__
 
-#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 48   // Note: also defined in object.c - should always match
-
 #ifdef __cplusplus
 extern "C" {
 #endif

src/t_list.cpp

@@ -331,7 +331,7 @@ void lindexCommand(client *c) {
             addReplyBulk(c,value);
             decrRefCount(value);
         } else {
-            addReplyNull(c,shared.nullbulk);
+            addReplyNull(c);
         }
     } else {
         serverPanic("Unknown list encoding");

src/tls.cpp

@@ -150,8 +150,6 @@ void tlsInit(void) {
         serverLog(LL_WARNING, "OpenSSL: Failed to seed random number generator.");
     }
 
-    /* Server configuration */
-    g_pserver->tls_auth_clients = 1;    /* Secure by default */
     tlsInitThread();
 }
 
@@ -193,6 +191,15 @@ int tlsConfigure(redisTLSContextConfig *ctx_config) {
     SSL_CTX_set_options(ctx, SSL_OP_DONT_INSERT_EMPTY_FRAGMENTS);
 #endif
 
+    if (ctx_config->session_caching) {
+        SSL_CTX_set_session_cache_mode(ctx, SSL_SESS_CACHE_SERVER);
+        SSL_CTX_sess_set_cache_size(ctx, ctx_config->session_cache_size);
+        SSL_CTX_set_timeout(ctx, ctx_config->session_cache_timeout);
+        SSL_CTX_set_session_id_context(ctx, (const unsigned char*) "KeyDB", 5);
+    } else {
+        SSL_CTX_set_session_cache_mode(ctx, SSL_SESS_CACHE_OFF);
+    }
+
     protocols = parseProtocolsConfig(ctx_config->protocols);
     if (protocols == -1) goto error;
 
@@ -226,7 +233,7 @@ int tlsConfigure(redisTLSContextConfig *ctx_config) {
     SSL_CTX_set_ecdh_auto(ctx, 1);
 #endif
 
-    if (SSL_CTX_use_certificate_file(ctx, ctx_config->cert_file, SSL_FILETYPE_PEM) <= 0) {
+    if (SSL_CTX_use_certificate_chain_file(ctx, ctx_config->cert_file) <= 0) {
         ERR_error_string_n(ERR_get_error(), errbuf, sizeof(errbuf));
         serverLog(LL_WARNING, "Failed to load certificate: %s: %s", ctx_config->cert_file, errbuf);
         goto error;

src/tracking.cpp

@@ -101,8 +101,8 @@ void disableTracking(client *c) {
 
 /* Set the client 'c' to track the prefix 'prefix'. If the client 'c' is
  * already registered for the specified prefix, no operation is performed. */
-void enableBcastTrackingForPrefix(client *c, const char *prefix, size_t plen) {
-    bcastState *bs = (bcastState*)raxFind(PrefixTable,(unsigned char*)prefix,sdslen(prefix));
+static void enableBcastTrackingForPrefix(client *c, const char *prefix, size_t plen) {
+    bcastState *bs = (bcastState*)raxFind(PrefixTable,(unsigned char*)prefix,plen);
     /* If this is the first client subscribing to such prefix, create
      * the prefix in the table. */
     if (bs == raxNotFound) {

src/version.h

@@ -1,3 +1,3 @@
-#define KEYDB_REAL_VERSION "0.0.0"
+#define KEYDB_REAL_VERSION "6.0.16"
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config
 

tests/integration/psync2-pingoff.tcl

@@ -64,6 +64,7 @@ start_server {} {
         # make sure replication is still alive and kicking
         $R(1) incr x
         wait_for_condition 50 1000 {
+            [status $R(0) loading] == 0 &&
             [$R(0) get x] == 1
         } else {
             fail "replica didn't get incr"

tests/integration/replication-active.tcl

@@ -237,3 +237,18 @@ start_server {tags {"active-repl"} overrides {active-replica yes}} {
     }
 }
 }
+
+start_server {tags {"active-repl"} overrides {active-replica yes}} {
+    set slave [srv 0 client]
+    set slave_host [srv 0 host]
+    set slave_port [srv 0 port]
+    start_server {tags {"active-repl"} overrides { active-replica yes}} {
+        r set testkeyB bar
+        test {Active Replica Merges Database On Sync} {
+            $slave set testkeyA foo
+            r replicaof $slave_host $slave_port
+	    after 1000
+	    assert_equal 2 [r dbsize]
+	}
+    }
+}

tests/integration/replication-multimaster.tcl

@@ -1,4 +1,6 @@
 foreach topology {mesh ring} {
+
+foreach noforward [expr {[string equal $topology "mesh"] ? {no yes} : {no}}] {
 start_server {tags {"multi-master"} overrides {hz 500 active-replica yes multi-master yes}} {
 start_server {overrides {hz 500 active-replica yes multi-master yes}} {
 start_server {overrides {hz 500 active-replica yes multi-master yes}} {
@@ -8,8 +10,12 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         set R($j) [srv [expr 0-$j] client]
         set R_host($j) [srv [expr 0-$j] host]
         set R_port($j) [srv [expr 0-$j] port]
+
+	$R($j) config set multi-master-no-forward $noforward
     }
 
+    set topology_name "$topology[expr {[string equal $noforward "yes"] ? " no-forward" : ""}]"
+
     # Initialize as mesh
     if [string equal $topology "mesh"] {
     for {set j 0} {$j < 4} {incr j} {
@@ -31,7 +37,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         $R(3) replicaof $R_host(2) $R_port(2)
     }
 
-    test "$topology all nodes up" {
+    test "$topology_name all nodes up" {
         for {set j 0} {$j < 4} {incr j} {
             wait_for_condition 50 100 {
                 [string match {*master_global_link_status:up*} [$R($j) info replication]]
@@ -41,7 +47,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         }
     }
 
-    test "$topology replicates to all nodes" {
+    test "$topology_name replicates to all nodes" {
         $R(0) set testkey foo
         after 500
         for {set n 0} {$n < 4} {incr n} {
@@ -53,7 +59,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         }
     }
 
-    test "$topology replicates only once" {
+    test "$topology_name replicates only once" {
         $R(0) set testkey 1
         after 500
         #wait_for_condition 50 100 {
@@ -74,7 +80,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         }
     }
 
-    test "$topology transaction replicates only once" {
+    test "$topology_name transaction replicates only once" {
         for {set j 0} {$j < 1000} {incr j} {
             $R(0) set testkey 1
             $R(0) multi
@@ -90,6 +96,27 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
             }
         }
     }
+
+    # Keep this test last since it muchs with the config
+    if [string equal $topology "mesh"] {
+    test "$topology_name quorum respected" {
+        $R(0) config set replica-serve-stale-data no
+
+        # No issues when all nodes are connected with default settings
+        $R(0) get testkey
+
+        # No issues when quorum is equal to the number of nodes
+        $R(0) config set replica-quorum 3
+        $R(0) get testkey
+
+        $R(0) config set replica-quorum 4
+        catch {
+            $R(0) get testkey
+        } e
+        assert_match {*MASTER is down*} $e
+    }
+    }
+}
 }
 }
 }

tests/modules/hooks.c

@@ -30,6 +30,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
+#define REDISMODULE_EXPERIMENTAL_API 1
 #include "redismodule.h"
 #include <stdio.h>
 #include <string.h>
@@ -143,10 +144,12 @@ void flushdbCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void
 {
     REDISMODULE_NOT_USED(e);
 
+    RedisModule_ThreadSafeContextLock(ctx);
     RedisModuleFlushInfo *fi = data;
     char *keyname = (sub == REDISMODULE_SUBEVENT_FLUSHDB_START) ?
         "flush-start" : "flush-end";
     LogNumericEvent(ctx, keyname, fi->dbnum);
+    RedisModule_ThreadSafeContextUnlock(ctx);
 }
 
 void roleChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)

tests/unit/expire.tcl

@@ -313,4 +313,14 @@ start_server {tags {"expire"}} {
         after 3000
         assert_equal [r dbsize] 0
     }
+
+    test {SET - use KEEPTTL option, TTL should not be removed after loadaof} {
+        r config set appendonly yes
+        r set foo bar EX 100
+        r set foo bar2 KEEPTTL
+        after 2000
+        r debug loadaof
+        set ttl [r ttl foo]
+        assert {$ttl <= 98 && $ttl > 90}
+    }
 }

tests/unit/introspection.tcl

@@ -78,17 +78,8 @@ start_server {tags {"introspection"}} {
             syslog-facility
             databases
             port
-            io-threads
             tls-port
-            tls-prefer-server-ciphers
-            tls-cert-file
-            tls-key-file
-            tls-dh-params-file
-            tls-ca-cert-file
-            tls-ca-cert-dir
-            tls-protocols
-            tls-ciphers
-            tls-ciphersuites
+            io-threads
             logfile
             unixsocketperm
             slaveof
@@ -101,6 +92,23 @@ start_server {tags {"introspection"}} {
             bgsave_cpulist
         }
 
+        if {!$::tls} {
+            append skip_configs {
+                tls-prefer-server-ciphers
+                tls-session-cache-timeout
+                tls-session-cache-size
+                tls-session-caching
+                tls-cert-file
+                tls-key-file
+                tls-dh-params-file
+                tls-ca-cert-file
+                tls-ca-cert-dir
+                tls-protocols
+                tls-ciphers
+                tls-ciphersuites
+            }
+        }
+
         set configs {}
         foreach {k v} [r config get *] {
             if {[lsearch $skip_configs $k] != -1} {

tests/unit/multi.tcl

@@ -363,6 +363,9 @@ start_server {tags {"multi"}} {
         set xx [r get xx]
         # make sure that either the whole transcation passed or none of it (we actually expect none)
         assert { $xx == 1 || $xx == 3}
+        # Discard the transaction since EXEC likely got -BUSY error
+        # so the client is still in MULTI state.
+        catch { $rd2 discard ;$rd2 read } e
         # check that the connection is no longer in multi state
         $rd2 ping asdf
         set pong [$rd2 read]

tests/unit/pubsub.tcl

@@ -109,6 +109,7 @@ start_server {tags {"pubsub"}} {
         unsubscribe $rd1
         # Wait for a response to the unsub
         __consume_subscribe_messages $rd1 unsubscribe {chan1 chan2 chan3}
+	after 1
         assert_equal 0 [r publish chan1 hello]
         assert_equal 0 [r publish chan2 hello]
         assert_equal 0 [r publish chan3 hello]