Modules shared API: export new core APIs.

Based on ideas and code in PR #5560 by @MeirShpilraien.

src/Makefile

@@ -21,6 +21,11 @@ NODEPS:=clean distclean
 
 # Default settings
 STD=-std=c99 -pedantic -DREDIS_STATIC=''
+ifneq (,$(findstring clang,$(CC)))
+ifneq (,$(findstring FreeBSD,$(uname_S)))
+  STD+=-Wno-c11-extensions
+endif
+endif
 WARN=-Wall -W -Wno-missing-field-initializers
 OPT=$(OPTIMIZATION)
 
@@ -41,6 +46,10 @@ endif
 # To get ARM stack traces if Redis crashes we need a special C flag.
 ifneq (,$(filter aarch64 armv,$(uname_M)))
         CFLAGS+=-funwind-tables
+else
+ifneq (,$(findstring armv,$(uname_M)))
+        CFLAGS+=-funwind-tables
+endif
 endif
 
 # Backwards compatibility for selecting an allocator
@@ -93,10 +102,20 @@ else
 ifeq ($(uname_S),OpenBSD)
 	# OpenBSD
 	FINAL_LIBS+= -lpthread
+	ifeq ($(USE_BACKTRACE),yes)
+	    FINAL_CFLAGS+= -DUSE_BACKTRACE -I/usr/local/include
+	    FINAL_LDFLAGS+= -L/usr/local/lib
+	    FINAL_LIBS+= -lexecinfo
+    	endif
+
 else
 ifeq ($(uname_S),FreeBSD)
 	# FreeBSD
-	FINAL_LIBS+= -lpthread
+	FINAL_LIBS+= -lpthread -lexecinfo
+else
+ifeq ($(uname_S),DragonFly)
+	# FreeBSD
+	FINAL_LIBS+= -lpthread -lexecinfo
 else
 	# All the other OSes (notably Linux)
 	FINAL_LDFLAGS+= -rdynamic
@@ -106,6 +125,7 @@ endif
 endif
 endif
 endif
+endif
 # Include paths to dependencies
 FINAL_CFLAGS+= -I../deps/hiredis -I../deps/linenoise -I../deps/lua/src
 

src/atomicvar.h

@@ -1,7 +1,7 @@
 /* This file implements atomic counters using __atomic or __sync macros if
  * available, otherwise synchronizing different threads using a mutex.
  *
- * The exported interaface is composed of three macros:
+ * The exported interface is composed of three macros:
  *
  * atomicIncr(var,count) -- Increment the atomic counter
  * atomicGetIncr(var,oldvalue_var,count) -- Get and increment the atomic counter

src/cluster.c

@@ -5164,10 +5164,10 @@ try_again:
         serverAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));
     }
 
-    int expired = 0; /* Number of keys that we'll find already expired.
-                        Note that serializing large keys may take some time
-                        so certain keys that were found non expired by the
-                        lookupKey() function, may be expired later. */
+    int non_expired = 0; /* Number of keys that we'll find non expired.
+                            Note that serializing large keys may take some time
+                            so certain keys that were found non expired by the
+                            lookupKey() function, may be expired later. */
 
     /* Create RESTORE payload and generate the protocol to call the command. */
     for (j = 0; j < num_keys; j++) {
@@ -5177,11 +5177,16 @@ try_again:
         if (expireat != -1) {
             ttl = expireat-mstime();
             if (ttl < 0) {
-                expired++;
                 continue;
             }
             if (ttl < 1) ttl = 1;
         }
+
+        /* Relocate valid (non expired) keys into the array in successive
+         * positions to remove holes created by the keys that were present
+         * in the first lookup but are now expired after the second lookup. */
+        kv[non_expired++] = kv[j];
+
         serverAssertWithInfo(c,NULL,
             rioWriteBulkCount(&cmd,'*',replace ? 5 : 4));
 
@@ -5209,6 +5214,9 @@ try_again:
             serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"REPLACE",7));
     }
 
+    /* Fix the actual number of keys we are migrating. */
+    num_keys = non_expired;
+
     /* Transfer the query to the other node in 64K chunks. */
     errno = 0;
     {
@@ -5250,7 +5258,7 @@ try_again:
      * command name itself. */
     if (!copy) newargv = zmalloc(sizeof(robj*)*(num_keys+1));
 
-    for (j = 0; j < num_keys-expired; j++) {
+    for (j = 0; j < num_keys; j++) {
         if (syncReadLine(cs->fd, buf2, sizeof(buf2), timeout) <= 0) {
             socket_error = 1;
             break;

src/config.c

@@ -120,7 +120,7 @@ const char *configEnumGetName(configEnum *ce, int val) {
     return NULL;
 }
 
-/* Wrapper for configEnumGetName() returning "unknown" insetad of NULL if
+/* Wrapper for configEnumGetName() returning "unknown" instead of NULL if
  * there is no match. */
 const char *configEnumGetNameOrUnknown(configEnum *ce, int val) {
     const char *name = configEnumGetName(ce,val);
@@ -684,7 +684,7 @@ void loadServerConfigFromString(char *config) {
                 goto loaderr;
             }
         } else if ((!strcasecmp(argv[0],"cluster-slave-no-failover") ||
-                    !strcasecmp(argv[0],"cluster-replica-no-failiver")) &&
+                    !strcasecmp(argv[0],"cluster-replica-no-failover")) &&
                    argc == 2)
         {
             server.cluster_slave_no_failover = yesnotoi(argv[1]);
@@ -1248,7 +1248,7 @@ void configSetCommand(client *c) {
             if (server.maxmemory < zmalloc_used_memory()) {
                 serverLog(LL_WARNING,"WARNING: the new maxmemory value set via CONFIG SET is smaller than the current memory usage. This will result in key eviction and/or the inability to accept new write commands depending on the maxmemory-policy.");
             }
-            freeMemoryIfNeeded();
+            freeMemoryIfNeededAndSafe();
         }
     } config_set_memory_field(
       "proto-max-bulk-len",server.proto_max_bulk_len) {

src/config.h

@@ -62,7 +62,9 @@
 #endif
 
 /* Test for backtrace() */
-#if defined(__APPLE__) || (defined(__linux__) && defined(__GLIBC__))
+#if defined(__APPLE__) || (defined(__linux__) && defined(__GLIBC__)) || \
+    defined(__FreeBSD__) || (defined(__OpenBSD__) && defined(USE_BACKTRACE))\
+ || defined(__DragonFly__)
 #define HAVE_BACKTRACE 1
 #endif
 

src/db.c

@@ -38,6 +38,8 @@
  * C-level DB API
  *----------------------------------------------------------------------------*/
 
+int keyIsExpired(redisDb *db, robj *key);
+
 /* Update LFU when an object is accessed.
  * Firstly, decrement the counter if the decrement time is reached.
  * Then logarithmically increment the counter, and update the access time. */
@@ -102,7 +104,10 @@ robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
         /* Key expired. If we are in the context of a master, expireIfNeeded()
          * returns 0 only when the key does not exist at all, so it's safe
          * to return NULL ASAP. */
-        if (server.masterhost == NULL) return NULL;
+        if (server.masterhost == NULL) {
+            server.stat_keyspace_misses++;
+            return NULL;
+        }
 
         /* However if we are in the context of a slave, expireIfNeeded() will
          * not really try to expire the key, it only returns information
@@ -121,6 +126,7 @@ robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
             server.current_client->cmd &&
             server.current_client->cmd->flags & CMD_READONLY)
         {
+            server.stat_keyspace_misses++;
             return NULL;
         }
     }
@@ -206,7 +212,7 @@ void dbOverwrite(redisDb *db, robj *key, robj *val) {
  * 2) clients WATCHing for the destination key notified.
  * 3) The expire time of the key is reset (the key is made persistent).
  *
- * All the new keys in the database should be craeted via this interface. */
+ * All the new keys in the database should be created via this interface. */
 void setKey(redisDb *db, robj *key, robj *val) {
     if (lookupKeyWrite(db,key) == NULL) {
         dbAdd(db,key,val);
@@ -543,7 +549,7 @@ void keysCommand(client *c) {
 
         if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
             keyobj = createStringObject(key,sdslen(key));
-            if (expireIfNeeded(c->db,keyobj) == 0) {
+            if (!keyIsExpired(c->db,keyobj)) {
                 addReplyBulk(c,keyobj);
                 numkeys++;
             }
@@ -1120,6 +1126,25 @@ void propagateExpire(redisDb *db, robj *key, int lazy) {
     decrRefCount(argv[1]);
 }
 
+/* Check if the key is expired. */
+int keyIsExpired(redisDb *db, robj *key) {
+    mstime_t when = getExpire(db,key);
+
+    if (when < 0) return 0; /* No expire for this key */
+
+    /* Don't expire anything while loading. It will be done later. */
+    if (server.loading) return 0;
+
+    /* If we are in the context of a Lua script, we pretend that time is
+     * blocked to when the Lua script started. This way a key can expire
+     * only the first time it is accessed and not in the middle of the
+     * script execution, making propagation to slaves / AOF consistent.
+     * See issue #1525 on Github for more information. */
+    mstime_t now = server.lua_caller ? server.lua_time_start : mstime();
+
+    return now > when;
+}
+
 /* This function is called when we are going to perform some operation
  * in a given key, but such key may be already logically expired even if
  * it still exists in the database. The main way this function is called
@@ -1140,32 +1165,17 @@ void propagateExpire(redisDb *db, robj *key, int lazy) {
  * The return value of the function is 0 if the key is still valid,
  * otherwise the function returns 1 if the key is expired. */
 int expireIfNeeded(redisDb *db, robj *key) {
-    mstime_t when = getExpire(db,key);
-    mstime_t now;
+    if (!keyIsExpired(db,key)) return 0;
 
-    if (when < 0) return 0; /* No expire for this key */
-
-    /* Don't expire anything while loading. It will be done later. */
-    if (server.loading) return 0;
-
-    /* If we are in the context of a Lua script, we pretend that time is
-     * blocked to when the Lua script started. This way a key can expire
-     * only the first time it is accessed and not in the middle of the
-     * script execution, making propagation to slaves / AOF consistent.
-     * See issue #1525 on Github for more information. */
-    now = server.lua_caller ? server.lua_time_start : mstime();
-
-    /* If we are running in the context of a slave, return ASAP:
+    /* If we are running in the context of a slave, instead of
+     * evicting the expired key from the database, we return ASAP:
      * the slave key expiration is controlled by the master that will
      * send us synthesized DEL operations for expired keys.
      *
      * Still we try to return the right information to the caller,
      * that is, 0 if we think the key should be still valid, 1 if
      * we think the key is expired at this time. */
-    if (server.masterhost != NULL) return now > when;
-
-    /* Return when this key has not expired */
-    if (now <= when) return 0;
+    if (server.masterhost != NULL) return 1;
 
     /* Delete the key */
     server.stat_expiredkeys++;

src/debug.c

@@ -37,7 +37,11 @@
 
 #ifdef HAVE_BACKTRACE
 #include <execinfo.h>
+#ifndef __OpenBSD__
 #include <ucontext.h>
+#else
+typedef ucontext_t sigcontext_t;
+#endif
 #include <fcntl.h>
 #include "bio.h"
 #include <unistd.h>
@@ -70,7 +74,7 @@ void xorDigest(unsigned char *digest, void *ptr, size_t len) {
         digest[j] ^= hash[j];
 }
 
-void xorObjectDigest(unsigned char *digest, robj *o) {
+void xorStringObjectDigest(unsigned char *digest, robj *o) {
     o = getDecodedObject(o);
     xorDigest(digest,o->ptr,sdslen(o->ptr));
     decrRefCount(o);
@@ -100,12 +104,151 @@ void mixDigest(unsigned char *digest, void *ptr, size_t len) {
     SHA1Final(digest,&ctx);
 }
 
-void mixObjectDigest(unsigned char *digest, robj *o) {
+void mixStringObjectDigest(unsigned char *digest, robj *o) {
     o = getDecodedObject(o);
     mixDigest(digest,o->ptr,sdslen(o->ptr));
     decrRefCount(o);
 }
 
+/* This function computes the digest of a data structure stored in the
+ * object 'o'. It is the core of the DEBUG DIGEST command: when taking the
+ * digest of a whole dataset, we take the digest of the key and the value
+ * pair, and xor all those together.
+ *
+ * Note that this function does not reset the initial 'digest' passed, it
+ * will continue mixing this object digest to anything that was already
+ * present. */
+void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o) {
+    uint32_t aux = htonl(o->type);
+    mixDigest(digest,&aux,sizeof(aux));
+    long long expiretime = getExpire(db,keyobj);
+    char buf[128];
+
+    /* Save the key and associated value */
+    if (o->type == OBJ_STRING) {
+        mixStringObjectDigest(digest,o);
+    } else if (o->type == OBJ_LIST) {
+        listTypeIterator *li = listTypeInitIterator(o,0,LIST_TAIL);
+        listTypeEntry entry;
+        while(listTypeNext(li,&entry)) {
+            robj *eleobj = listTypeGet(&entry);
+            mixStringObjectDigest(digest,eleobj);
+            decrRefCount(eleobj);
+        }
+        listTypeReleaseIterator(li);
+    } else if (o->type == OBJ_SET) {
+        setTypeIterator *si = setTypeInitIterator(o);
+        sds sdsele;
+        while((sdsele = setTypeNextObject(si)) != NULL) {
+            xorDigest(digest,sdsele,sdslen(sdsele));
+            sdsfree(sdsele);
+        }
+        setTypeReleaseIterator(si);
+    } else if (o->type == OBJ_ZSET) {
+        unsigned char eledigest[20];
+
+        if (o->encoding == OBJ_ENCODING_ZIPLIST) {
+            unsigned char *zl = o->ptr;
+            unsigned char *eptr, *sptr;
+            unsigned char *vstr;
+            unsigned int vlen;
+            long long vll;
+            double score;
+
+            eptr = ziplistIndex(zl,0);
+            serverAssert(eptr != NULL);
+            sptr = ziplistNext(zl,eptr);
+            serverAssert(sptr != NULL);
+
+            while (eptr != NULL) {
+                serverAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
+                score = zzlGetScore(sptr);
+
+                memset(eledigest,0,20);
+                if (vstr != NULL) {
+                    mixDigest(eledigest,vstr,vlen);
+                } else {
+                    ll2string(buf,sizeof(buf),vll);
+                    mixDigest(eledigest,buf,strlen(buf));
+                }
+
+                snprintf(buf,sizeof(buf),"%.17g",score);
+                mixDigest(eledigest,buf,strlen(buf));
+                xorDigest(digest,eledigest,20);
+                zzlNext(zl,&eptr,&sptr);
+            }
+        } else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
+            zset *zs = o->ptr;
+            dictIterator *di = dictGetIterator(zs->dict);
+            dictEntry *de;
+
+            while((de = dictNext(di)) != NULL) {
+                sds sdsele = dictGetKey(de);
+                double *score = dictGetVal(de);
+
+                snprintf(buf,sizeof(buf),"%.17g",*score);
+                memset(eledigest,0,20);
+                mixDigest(eledigest,sdsele,sdslen(sdsele));
+                mixDigest(eledigest,buf,strlen(buf));
+                xorDigest(digest,eledigest,20);
+            }
+            dictReleaseIterator(di);
+        } else {
+            serverPanic("Unknown sorted set encoding");
+        }
+    } else if (o->type == OBJ_HASH) {
+        hashTypeIterator *hi = hashTypeInitIterator(o);
+        while (hashTypeNext(hi) != C_ERR) {
+            unsigned char eledigest[20];
+            sds sdsele;
+
+            memset(eledigest,0,20);
+            sdsele = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_KEY);
+            mixDigest(eledigest,sdsele,sdslen(sdsele));
+            sdsfree(sdsele);
+            sdsele = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_VALUE);
+            mixDigest(eledigest,sdsele,sdslen(sdsele));
+            sdsfree(sdsele);
+            xorDigest(digest,eledigest,20);
+        }
+        hashTypeReleaseIterator(hi);
+    } else if (o->type == OBJ_STREAM) {
+        streamIterator si;
+        streamIteratorStart(&si,o->ptr,NULL,NULL,0);
+        streamID id;
+        int64_t numfields;
+
+        while(streamIteratorGetID(&si,&id,&numfields)) {
+            sds itemid = sdscatfmt(sdsempty(),"%U.%U",id.ms,id.seq);
+            mixDigest(digest,itemid,sdslen(itemid));
+            sdsfree(itemid);
+
+            while(numfields--) {
+                unsigned char *field, *value;
+                int64_t field_len, value_len;
+                streamIteratorGetField(&si,&field,&value,
+                                           &field_len,&value_len);
+                mixDigest(digest,field,field_len);
+                mixDigest(digest,value,value_len);
+            }
+        }
+        streamIteratorStop(&si);
+    } else if (o->type == OBJ_MODULE) {
+        RedisModuleDigest md;
+        moduleValue *mv = o->ptr;
+        moduleType *mt = mv->type;
+        moduleInitDigestContext(md);
+        if (mt->digest) {
+            mt->digest(&md,mv->value);
+            xorDigest(digest,md.x,sizeof(md.x));
+        }
+    } else {
+        serverPanic("Unknown object type");
+    }
+    /* If the key has an expire, add it to the mix */
+    if (expiretime != -1) xorDigest(digest,"!!expire!!",10);
+}
+
 /* Compute the dataset digest. Since keys, sets elements, hashes elements
  * are not ordered, we use a trick: every aggregate digest is the xor
  * of the digests of their elements. This way the order will not change
@@ -114,7 +257,6 @@ void mixObjectDigest(unsigned char *digest, robj *o) {
  * a different digest. */
 void computeDatasetDigest(unsigned char *final) {
     unsigned char digest[20];
-    char buf[128];
     dictIterator *di = NULL;
     dictEntry *de;
     int j;
@@ -137,7 +279,6 @@ void computeDatasetDigest(unsigned char *final) {
         while((de = dictNext(di)) != NULL) {
             sds key;
             robj *keyobj, *o;
-            long long expiretime;
 
             memset(digest,0,20); /* This key-val digest */
             key = dictGetKey(de);
@@ -146,134 +287,8 @@ void computeDatasetDigest(unsigned char *final) {
             mixDigest(digest,key,sdslen(key));
 
             o = dictGetVal(de);
+            xorObjectDigest(db,keyobj,digest,o);
 
-            aux = htonl(o->type);
-            mixDigest(digest,&aux,sizeof(aux));
-            expiretime = getExpire(db,keyobj);
-
-            /* Save the key and associated value */
-            if (o->type == OBJ_STRING) {
-                mixObjectDigest(digest,o);
-            } else if (o->type == OBJ_LIST) {
-                listTypeIterator *li = listTypeInitIterator(o,0,LIST_TAIL);
-                listTypeEntry entry;
-                while(listTypeNext(li,&entry)) {
-                    robj *eleobj = listTypeGet(&entry);
-                    mixObjectDigest(digest,eleobj);
-                    decrRefCount(eleobj);
-                }
-                listTypeReleaseIterator(li);
-            } else if (o->type == OBJ_SET) {
-                setTypeIterator *si = setTypeInitIterator(o);
-                sds sdsele;
-                while((sdsele = setTypeNextObject(si)) != NULL) {
-                    xorDigest(digest,sdsele,sdslen(sdsele));
-                    sdsfree(sdsele);
-                }
-                setTypeReleaseIterator(si);
-            } else if (o->type == OBJ_ZSET) {
-                unsigned char eledigest[20];
-
-                if (o->encoding == OBJ_ENCODING_ZIPLIST) {
-                    unsigned char *zl = o->ptr;
-                    unsigned char *eptr, *sptr;
-                    unsigned char *vstr;
-                    unsigned int vlen;
-                    long long vll;
-                    double score;
-
-                    eptr = ziplistIndex(zl,0);
-                    serverAssert(eptr != NULL);
-                    sptr = ziplistNext(zl,eptr);
-                    serverAssert(sptr != NULL);
-
-                    while (eptr != NULL) {
-                        serverAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
-                        score = zzlGetScore(sptr);
-
-                        memset(eledigest,0,20);
-                        if (vstr != NULL) {
-                            mixDigest(eledigest,vstr,vlen);
-                        } else {
-                            ll2string(buf,sizeof(buf),vll);
-                            mixDigest(eledigest,buf,strlen(buf));
-                        }
-
-                        snprintf(buf,sizeof(buf),"%.17g",score);
-                        mixDigest(eledigest,buf,strlen(buf));
-                        xorDigest(digest,eledigest,20);
-                        zzlNext(zl,&eptr,&sptr);
-                    }
-                } else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
-                    zset *zs = o->ptr;
-                    dictIterator *di = dictGetIterator(zs->dict);
-                    dictEntry *de;
-
-                    while((de = dictNext(di)) != NULL) {
-                        sds sdsele = dictGetKey(de);
-                        double *score = dictGetVal(de);
-
-                        snprintf(buf,sizeof(buf),"%.17g",*score);
-                        memset(eledigest,0,20);
-                        mixDigest(eledigest,sdsele,sdslen(sdsele));
-                        mixDigest(eledigest,buf,strlen(buf));
-                        xorDigest(digest,eledigest,20);
-                    }
-                    dictReleaseIterator(di);
-                } else {
-                    serverPanic("Unknown sorted set encoding");
-                }
-            } else if (o->type == OBJ_HASH) {
-                hashTypeIterator *hi = hashTypeInitIterator(o);
-                while (hashTypeNext(hi) != C_ERR) {
-                    unsigned char eledigest[20];
-                    sds sdsele;
-
-                    memset(eledigest,0,20);
-                    sdsele = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_KEY);
-                    mixDigest(eledigest,sdsele,sdslen(sdsele));
-                    sdsfree(sdsele);
-                    sdsele = hashTypeCurrentObjectNewSds(hi,OBJ_HASH_VALUE);
-                    mixDigest(eledigest,sdsele,sdslen(sdsele));
-                    sdsfree(sdsele);
-                    xorDigest(digest,eledigest,20);
-                }
-                hashTypeReleaseIterator(hi);
-            } else if (o->type == OBJ_STREAM) {
-                streamIterator si;
-                streamIteratorStart(&si,o->ptr,NULL,NULL,0);
-                streamID id;
-                int64_t numfields;
-
-                while(streamIteratorGetID(&si,&id,&numfields)) {
-                    sds itemid = sdscatfmt(sdsempty(),"%U.%U",id.ms,id.seq);
-                    mixDigest(digest,itemid,sdslen(itemid));
-                    sdsfree(itemid);
-
-                    while(numfields--) {
-                        unsigned char *field, *value;
-                        int64_t field_len, value_len;
-                        streamIteratorGetField(&si,&field,&value,
-                                                   &field_len,&value_len);
-                        mixDigest(digest,field,field_len);
-                        mixDigest(digest,value,value_len);
-                    }
-                }
-                streamIteratorStop(&si);
-            } else if (o->type == OBJ_MODULE) {
-                RedisModuleDigest md;
-                moduleValue *mv = o->ptr;
-                moduleType *mt = mv->type;
-                moduleInitDigestContext(md);
-                if (mt->digest) {
-                    mt->digest(&md,mv->value);
-                    xorDigest(digest,md.x,sizeof(md.x));
-                }
-            } else {
-                serverPanic("Unknown object type");
-            }
-            /* If the key has an expire, add it to the mix */
-            if (expiretime != -1) xorDigest(digest,"!!expire!!",10);
             /* We can finally xor the key-val digest to the final digest */
             xorDigest(final,digest,20);
             decrRefCount(keyobj);
@@ -289,6 +304,7 @@ void debugCommand(client *c) {
 "CHANGE-REPL-ID -- Change the replication IDs of the instance. Dangerous, should be used only for testing the replication subsystem.",
 "CRASH-AND-RECOVER <milliseconds> -- Hard crash and restart after <milliseconds> delay.",
 "DIGEST -- Output a hex signature representing the current DB content.",
+"DIGEST-VALUE <key-1> ... <key-N>-- Output a hex signature of the values of all the specified keys.",
 "ERROR <string> -- Return a Redis protocol error with <string> as message. Useful for clients unit tests to simulate Redis errors.",
 "LOG <message> -- write message to the server log.",
 "HTSTATS <dbid> -- Return hash table statistics of the specified Redis database.",
@@ -306,6 +322,7 @@ void debugCommand(client *c) {
 "SLEEP <seconds> -- Stop the server for <seconds>. Decimals allowed.",
 "STRUCTSIZE -- Return the size of different Redis core C structures.",
 "ZIPLIST <key> -- Show low level info about the ziplist encoding.",
+"STRINGMATCH-TEST -- Run a fuzz tester against the stringmatchlen() function.",
 NULL
         };
         addReplyHelp(c, help);
@@ -332,7 +349,6 @@ NULL
         zfree(ptr);
         addReply(c,shared.ok);
     } else if (!strcasecmp(c->argv[1]->ptr,"assert")) {
-        if (c->argc >= 3) c->argv[2] = tryObjectEncoding(c->argv[2]);
         serverAssertWithInfo(c,c->argv[0],1 == 2);
     } else if (!strcasecmp(c->argv[1]->ptr,"log") && c->argc == 3) {
         serverLog(LL_WARNING, "DEBUG LOG: %s", (char*)c->argv[2]->ptr);
@@ -491,15 +507,28 @@ NULL
         }
         addReply(c,shared.ok);
     } else if (!strcasecmp(c->argv[1]->ptr,"digest") && c->argc == 2) {
+        /* DEBUG DIGEST (form without keys specified) */
         unsigned char digest[20];
         sds d = sdsempty();
-        int j;
 
         computeDatasetDigest(digest);
-        for (j = 0; j < 20; j++)
-            d = sdscatprintf(d, "%02x",digest[j]);
+        for (int i = 0; i < 20; i++) d = sdscatprintf(d, "%02x",digest[i]);
         addReplyStatus(c,d);
         sdsfree(d);
+    } else if (!strcasecmp(c->argv[1]->ptr,"digest-value") && c->argc >= 2) {
+        /* DEBUG DIGEST-VALUE key key key ... key. */
+        addReplyMultiBulkLen(c,c->argc-2);
+        for (int j = 2; j < c->argc; j++) {
+            unsigned char digest[20];
+            memset(digest,0,20); /* Start with a clean result */
+            robj *o = lookupKeyReadWithFlags(c->db,c->argv[j],LOOKUP_NOTOUCH);
+            if (o) xorObjectDigest(c->db,c->argv[j],digest,o);
+
+            sds d = sdsempty();
+            for (int i = 0; i < 20; i++) d = sdscatprintf(d, "%02x",digest[i]);
+            addReplyStatus(c,d);
+            sdsfree(d);
+        }
     } else if (!strcasecmp(c->argv[1]->ptr,"sleep") && c->argc == 3) {
         double dtime = strtod(c->argv[2]->ptr,NULL);
         long long utime = dtime*1000000;
@@ -591,6 +620,10 @@ NULL
         changeReplicationId();
         clearReplicationId2();
         addReply(c,shared.ok);
+    } else if (!strcasecmp(c->argv[1]->ptr,"stringmatch-test") && c->argc == 2)
+    {
+        stringmatchlen_fuzz_test();
+        addReplyStatus(c,"Apparently Redis did not crash: test passed");
     } else {
         addReplySubcommandSyntaxError(c);
         return;
@@ -729,6 +762,22 @@ static void *getMcontextEip(ucontext_t *uc) {
     #elif defined(__aarch64__) /* Linux AArch64 */
     return (void*) uc->uc_mcontext.pc;
     #endif
+#elif defined(__FreeBSD__)
+    /* FreeBSD */
+    #if defined(__i386__)
+    return (void*) uc->uc_mcontext.mc_eip;
+    #elif defined(__x86_64__)
+    return (void*) uc->uc_mcontext.mc_rip;
+    #endif
+#elif defined(__OpenBSD__)
+    /* OpenBSD */
+    #if defined(__i386__)
+    return (void*) uc->sc_eip;
+    #elif defined(__x86_64__)
+    return (void*) uc->sc_rip;
+    #endif
+#elif defined(__DragonFly__)
+    return (void*) uc->uc_mcontext.mc_rip;
 #else
     return NULL;
 #endif
@@ -870,6 +919,145 @@ void logRegisters(ucontext_t *uc) {
     );
     logStackContent((void**)uc->uc_mcontext.gregs[15]);
     #endif
+#elif defined(__FreeBSD__)
+    #if defined(__x86_64__)
+    serverLog(LL_WARNING,
+    "\n"
+    "RAX:%016lx RBX:%016lx\nRCX:%016lx RDX:%016lx\n"
+    "RDI:%016lx RSI:%016lx\nRBP:%016lx RSP:%016lx\n"
+    "R8 :%016lx R9 :%016lx\nR10:%016lx R11:%016lx\n"
+    "R12:%016lx R13:%016lx\nR14:%016lx R15:%016lx\n"
+    "RIP:%016lx EFL:%016lx\nCSGSFS:%016lx",
+        (unsigned long) uc->uc_mcontext.mc_rax,
+        (unsigned long) uc->uc_mcontext.mc_rbx,
+        (unsigned long) uc->uc_mcontext.mc_rcx,
+        (unsigned long) uc->uc_mcontext.mc_rdx,
+        (unsigned long) uc->uc_mcontext.mc_rdi,
+        (unsigned long) uc->uc_mcontext.mc_rsi,
+        (unsigned long) uc->uc_mcontext.mc_rbp,
+        (unsigned long) uc->uc_mcontext.mc_rsp,
+        (unsigned long) uc->uc_mcontext.mc_r8,
+        (unsigned long) uc->uc_mcontext.mc_r9,
+        (unsigned long) uc->uc_mcontext.mc_r10,
+        (unsigned long) uc->uc_mcontext.mc_r11,
+        (unsigned long) uc->uc_mcontext.mc_r12,
+        (unsigned long) uc->uc_mcontext.mc_r13,
+        (unsigned long) uc->uc_mcontext.mc_r14,
+        (unsigned long) uc->uc_mcontext.mc_r15,
+        (unsigned long) uc->uc_mcontext.mc_rip,
+        (unsigned long) uc->uc_mcontext.mc_rflags,
+        (unsigned long) uc->uc_mcontext.mc_cs
+    );
+    logStackContent((void**)uc->uc_mcontext.mc_rsp);
+    #elif defined(__i386__)
+    serverLog(LL_WARNING,
+    "\n"
+    "EAX:%08lx EBX:%08lx ECX:%08lx EDX:%08lx\n"
+    "EDI:%08lx ESI:%08lx EBP:%08lx ESP:%08lx\n"
+    "SS :%08lx EFL:%08lx EIP:%08lx CS:%08lx\n"
+    "DS :%08lx ES :%08lx FS :%08lx GS:%08lx",
+        (unsigned long) uc->uc_mcontext.mc_eax,
+        (unsigned long) uc->uc_mcontext.mc_ebx,
+        (unsigned long) uc->uc_mcontext.mc_ebx,
+        (unsigned long) uc->uc_mcontext.mc_edx,
+        (unsigned long) uc->uc_mcontext.mc_edi,
+        (unsigned long) uc->uc_mcontext.mc_esi,
+        (unsigned long) uc->uc_mcontext.mc_ebp,
+        (unsigned long) uc->uc_mcontext.mc_esp,
+        (unsigned long) uc->uc_mcontext.mc_ss,
+        (unsigned long) uc->uc_mcontext.mc_eflags,
+        (unsigned long) uc->uc_mcontext.mc_eip,
+        (unsigned long) uc->uc_mcontext.mc_cs,
+        (unsigned long) uc->uc_mcontext.mc_es,
+        (unsigned long) uc->uc_mcontext.mc_fs,
+        (unsigned long) uc->uc_mcontext.mc_gs
+    );
+    logStackContent((void**)uc->uc_mcontext.mc_esp);
+    #endif
+#elif defined(__OpenBSD__)
+    #if defined(__x86_64__)
+    serverLog(LL_WARNING,
+    "\n"
+    "RAX:%016lx RBX:%016lx\nRCX:%016lx RDX:%016lx\n"
+    "RDI:%016lx RSI:%016lx\nRBP:%016lx RSP:%016lx\n"
+    "R8 :%016lx R9 :%016lx\nR10:%016lx R11:%016lx\n"
+    "R12:%016lx R13:%016lx\nR14:%016lx R15:%016lx\n"
+    "RIP:%016lx EFL:%016lx\nCSGSFS:%016lx",
+        (unsigned long) uc->sc_rax,
+        (unsigned long) uc->sc_rbx,
+        (unsigned long) uc->sc_rcx,
+        (unsigned long) uc->sc_rdx,
+        (unsigned long) uc->sc_rdi,
+        (unsigned long) uc->sc_rsi,
+        (unsigned long) uc->sc_rbp,
+        (unsigned long) uc->sc_rsp,
+        (unsigned long) uc->sc_r8,
+        (unsigned long) uc->sc_r9,
+        (unsigned long) uc->sc_r10,
+        (unsigned long) uc->sc_r11,
+        (unsigned long) uc->sc_r12,
+        (unsigned long) uc->sc_r13,
+        (unsigned long) uc->sc_r14,
+        (unsigned long) uc->sc_r15,
+        (unsigned long) uc->sc_rip,
+        (unsigned long) uc->sc_rflags,
+        (unsigned long) uc->sc_cs
+    );
+    logStackContent((void**)uc->sc_rsp);
+    #elif defined(__i386__)
+    serverLog(LL_WARNING,
+    "\n"
+    "EAX:%08lx EBX:%08lx ECX:%08lx EDX:%08lx\n"
+    "EDI:%08lx ESI:%08lx EBP:%08lx ESP:%08lx\n"
+    "SS :%08lx EFL:%08lx EIP:%08lx CS:%08lx\n"
+    "DS :%08lx ES :%08lx FS :%08lx GS:%08lx",
+        (unsigned long) uc->sc_eax,
+        (unsigned long) uc->sc_ebx,
+        (unsigned long) uc->sc_ebx,
+        (unsigned long) uc->sc_edx,
+        (unsigned long) uc->sc_edi,
+        (unsigned long) uc->sc_esi,
+        (unsigned long) uc->sc_ebp,
+        (unsigned long) uc->sc_esp,
+        (unsigned long) uc->sc_ss,
+        (unsigned long) uc->sc_eflags,
+        (unsigned long) uc->sc_eip,
+        (unsigned long) uc->sc_cs,
+        (unsigned long) uc->sc_es,
+        (unsigned long) uc->sc_fs,
+        (unsigned long) uc->sc_gs
+    );
+    logStackContent((void**)uc->sc_esp);
+    #endif
+#elif defined(__DragonFly__)
+    serverLog(LL_WARNING,
+    "\n"
+    "RAX:%016lx RBX:%016lx\nRCX:%016lx RDX:%016lx\n"
+    "RDI:%016lx RSI:%016lx\nRBP:%016lx RSP:%016lx\n"
+    "R8 :%016lx R9 :%016lx\nR10:%016lx R11:%016lx\n"
+    "R12:%016lx R13:%016lx\nR14:%016lx R15:%016lx\n"
+    "RIP:%016lx EFL:%016lx\nCSGSFS:%016lx",
+        (unsigned long) uc->uc_mcontext.mc_rax,
+        (unsigned long) uc->uc_mcontext.mc_rbx,
+        (unsigned long) uc->uc_mcontext.mc_rcx,
+        (unsigned long) uc->uc_mcontext.mc_rdx,
+        (unsigned long) uc->uc_mcontext.mc_rdi,
+        (unsigned long) uc->uc_mcontext.mc_rsi,
+        (unsigned long) uc->uc_mcontext.mc_rbp,
+        (unsigned long) uc->uc_mcontext.mc_rsp,
+        (unsigned long) uc->uc_mcontext.mc_r8,
+        (unsigned long) uc->uc_mcontext.mc_r9,
+        (unsigned long) uc->uc_mcontext.mc_r10,
+        (unsigned long) uc->uc_mcontext.mc_r11,
+        (unsigned long) uc->uc_mcontext.mc_r12,
+        (unsigned long) uc->uc_mcontext.mc_r13,
+        (unsigned long) uc->uc_mcontext.mc_r14,
+        (unsigned long) uc->uc_mcontext.mc_r15,
+        (unsigned long) uc->uc_mcontext.mc_rip,
+        (unsigned long) uc->uc_mcontext.mc_rflags,
+        (unsigned long) uc->uc_mcontext.mc_cs
+    );
+    logStackContent((void**)uc->uc_mcontext.mc_rsp);
 #else
     serverLog(LL_WARNING,
         "  Dumping of registers not supported for this OS/arch");
@@ -1189,6 +1377,8 @@ void serverLogHexDump(int level, char *descr, void *value, size_t len) {
 void watchdogSignalHandler(int sig, siginfo_t *info, void *secret) {
 #ifdef HAVE_BACKTRACE
     ucontext_t *uc = (ucontext_t*) secret;
+#else
+    (void)secret;
 #endif
     UNUSED(info);
     UNUSED(sig);

src/evict.c

@@ -364,7 +364,7 @@ size_t freeMemoryGetNotCountedMemory(void) {
         }
     }
     if (server.aof_state != AOF_OFF) {
-        overhead += sdslen(server.aof_buf)+aofRewriteBufferSize();
+        overhead += sdsalloc(server.aof_buf)+aofRewriteBufferSize();
     }
     return overhead;
 }
@@ -444,8 +444,8 @@ int getMaxmemoryState(size_t *total, size_t *logical, size_t *tofree, float *lev
  * Otehrwise if we are over the memory limit, but not enough memory
  * was freed to return back under the limit, the function returns C_ERR. */
 int freeMemoryIfNeeded(void) {
-    /* By default slaves should ignore maxmemory and just be masters excat
-     * copies. */
+    /* By default replicas should ignore maxmemory
+     * and just be masters exact copies. */
     if (server.masterhost && server.repl_slave_ignore_maxmemory) return C_OK;
 
     size_t mem_reported, mem_tofree, mem_freed;
@@ -622,3 +622,14 @@ cant_free:
     return C_ERR;
 }
 
+/* This is a wrapper for freeMemoryIfNeeded() that only really calls the
+ * function if right now there are the conditions to do so safely:
+ *
+ * - There must be no script in timeout condition.
+ * - Nor we are loading data right now.
+ *
+ */
+int freeMemoryIfNeededAndSafe(void) {
+    if (server.lua_timedout || server.loading) return C_OK;
+    return freeMemoryIfNeeded();
+}

src/intset.c

@@ -123,7 +123,7 @@ static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
     } else {
         /* Check for the case where we know we cannot find the value,
          * but do know the insert position. */
-        if (value > _intsetGet(is,intrev32ifbe(is->length)-1)) {
+        if (value > _intsetGet(is,max)) {
             if (pos) *pos = intrev32ifbe(is->length);
             return 0;
         } else if (value < _intsetGet(is,0)) {

src/latency.c

@@ -562,11 +562,21 @@ sds latencyCommandGenSparkeline(char *event, struct latencyTimeSeries *ts) {
  *
  * LATENCY HISTORY: return time-latency samples for the specified event.
  * LATENCY LATEST: return the latest latency for all the events classes.
- * LATENCY DOCTOR: returns an human readable analysis of instance latency.
+ * LATENCY DOCTOR: returns a human readable analysis of instance latency.
  * LATENCY GRAPH: provide an ASCII graph of the latency of the specified event.
  * LATENCY RESET: reset data of a specified event or all the data if no event provided.
  */
 void latencyCommand(client *c) {
+    const char *help[] = {
+"DOCTOR              -- Returns a human readable latency analysis report.",
+"GRAPH   <event>     -- Returns an ASCII latency graph for the event class.",
+"HISTORY <event>     -- Returns time-latency samples for the event class.",
+"LATEST              -- Returns the latest latency samples for all events.",
+"RESET   [event ...] -- Resets latency data of one or more event classes.",
+"                       (default: reset all data for all event classes)",
+"HELP                -- Prints this help.",
+NULL
+    };
     struct latencyTimeSeries *ts;
 
     if (!strcasecmp(c->argv[1]->ptr,"history") && c->argc == 3) {
@@ -611,8 +621,10 @@ void latencyCommand(client *c) {
                 resets += latencyResetEvent(c->argv[j]->ptr);
             addReplyLongLong(c,resets);
         }
+    } else if (!strcasecmp(c->argv[1]->ptr,"help") && c->argc >= 2) {
+        addReplyHelp(c, help);
     } else {
-        addReply(c,shared.syntaxerr);
+        addReplySubcommandSyntaxError(c);
     }
     return;
 

src/module.c

@@ -47,9 +47,21 @@ struct RedisModule {
     int ver;        /* Module version. We use just progressive integers. */
     int apiver;     /* Module API version as requested during initialization.*/
     list *types;    /* Module data types. */
+    list *usedby;   /* List of modules using APIs from this one. */
+    list *using;    /* List of modules we use some APIs of. */
 };
 typedef struct RedisModule RedisModule;
 
+/* This represents a shared API. Shared APIs will be used to populate
+ * the server.sharedapi dictionary, mapping names of APIs exported by
+ * modules for other modules to use, to their structure specifying the
+ * function pointer that can be called. */
+struct RedisModuleSharedAPI {
+    void *func;
+    RedisModule *module;
+};
+typedef struct RedisModuleSharedAPI RedisModuleSharedAPI;
+
 static dict *modules; /* Hash table of modules. SDS -> RedisModule ptr.*/
 
 /* Entries in the context->amqueue array, representing objects to free
@@ -700,6 +712,8 @@ void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int api
     module->ver = ver;
     module->apiver = apiver;
     module->types = listCreate();
+    module->usedby = listCreate();
+    module->using = listCreate();
     ctx->module = module;
 }
 
@@ -4615,6 +4629,121 @@ void RM_GetRandomHexChars(char *dst, size_t len) {
     getRandomHexChars(dst,len);
 }
 
+/* --------------------------------------------------------------------------
+ * Modules API exporting / importing
+ * -------------------------------------------------------------------------- */
+
+/* This function is called by a module in order to export some API with a
+ * given name. Other modules will be able to use this API by calling the
+ * symmetrical function RM_GetSharedAPI() and casting the return value to
+ * the right function pointer.
+ *
+ * The function will return REDISMODULE_OK if the name is not already taken,
+ * otherwise REDISMODULE_ERR will be returned and no operation will be
+ * performed.
+ *
+ * IMPORTANT: the apiname argument should be a string literal with static
+ * lifetime. The API relies on the fact that it will always be valid in
+ * the future. */
+int RM_ExportSharedAPI(RedisModuleCtx *ctx, const char *apiname, void *func) {
+    RedisModuleSharedAPI *sapi = zmalloc(sizeof(*sapi));
+    sapi->module = ctx->module;
+    sapi->func = func;
+    if (dictAdd(server.sharedapi, (char*)apiname, sapi) != DICT_OK) {
+        zfree(sapi);
+        return REDISMODULE_ERR;
+    }
+    return REDISMODULE_OK;
+}
+
+/* Request an exported API pointer. The return value is just a void pointer
+ * that the caller of this function will be required to cast to the right
+ * function pointer, so this is a private contract between modules.
+ *
+ * If the requested API is not available then NULL is returned. Because
+ * modules can be loaded at different times with different order, this
+ * function calls should be put inside some module generic API registering
+ * step, that is called every time a module attempts to execute a
+ * command that requires external APIs: if some API cannot be resolved, the
+ * command should return an error.
+ *
+ * Here is an exmaple:
+ *
+ *     int ... myCommandImplementation() {
+ *        if (getExternalAPIs() == 0) {
+ *             reply with an error here if we cannot have the APIs
+ *        }
+ *        // Use the API:
+ *        myFunctionPointer(foo);
+ *     }
+ *
+ * And the function registerAPI() is:
+ *
+ *     int getExternalAPIs(void) {
+ *         static int api_loaded = 0;
+ *         if (api_loaded != 0) return 1; // APIs already resolved.
+ *
+ *         myFunctionPointer = RedisModule_GetOtherModuleAPI("...");
+ *         if (myFunctionPointer == NULL) return 0;
+ *
+ *         return 1;
+ *     }
+ */
+void *RM_GetSharedAPI(RedisModuleCtx *ctx, const char *apiname) {
+    dictEntry *de = dictFind(server.sharedapi, apiname);
+    if (de == NULL) return NULL;
+    RedisModuleSharedAPI *sapi = dictGetVal(de);
+    if (listSearchKey(sapi->module->usedby,ctx->module) == NULL) {
+        listAddNodeTail(sapi->module->usedby,ctx->module);
+        listAddNodeTail(ctx->module->using,sapi->module);
+    }
+    return sapi->func;
+}
+
+/* Remove all the APIs registered by the specified module. Usually you
+ * want this when the module is going to be unloaded. This function
+ * assumes that's caller responsibility to make sure the APIs are not
+ * used by other modules.
+ *
+ * The number of unregistered APIs is returned. */
+int moduleUnregisterSharedAPI(RedisModule *module) {
+    int count = 0;
+    dictIterator *di = dictGetSafeIterator(server.sharedapi);
+    dictEntry *de;
+    while ((de = dictNext(di)) != NULL) {
+        const char *apiname = dictGetKey(de);
+        RedisModuleSharedAPI *sapi = dictGetVal(de);
+        if (sapi->module == module) {
+            dictDelete(server.sharedapi,apiname);
+            zfree(sapi);
+            count++;
+        }
+    }
+    dictReleaseIterator(di);
+    return count;
+}
+
+/* Remove the specified module as an user of APIs of ever other module.
+ * This is usually called when a module is unloaded.
+ *
+ * Returns the number of modules this module was using APIs from. */
+int moduleUnregisterUsedAPI(RedisModule *module) {
+    listIter li;
+    listNode *ln;
+    int count = 0;
+
+    listRewind(module->using,&li);
+    while((ln = listNext(&li))) {
+        RedisModule *used = ln->value;
+        listNode *ln = listSearchKey(used->usedby,module);
+        if (ln) {
+            listDelNode(module->using,ln);
+            count++;
+        }
+    }
+    return count;
+}
+
 /* --------------------------------------------------------------------------
  * Modules API internals
  * -------------------------------------------------------------------------- */
@@ -4758,6 +4887,8 @@ int moduleLoad(const char *path, void **module_argv, int module_argc) {
     if (onload((void*)&ctx,module_argv,module_argc) == REDISMODULE_ERR) {
         if (ctx.module) {
             moduleUnregisterCommands(ctx.module);
+            moduleUnregisterSharedAPI(ctx.module);
+            moduleUnregisterUsedAPI(ctx.module);
             moduleFreeModuleStructure(ctx.module);
         }
         dlclose(handle);
@@ -4787,14 +4918,17 @@ int moduleUnload(sds name) {
     if (module == NULL) {
         errno = ENOENT;
         return REDISMODULE_ERR;
-    }
-
-    if (listLength(module->types)) {
+    } else if (listLength(module->types)) {
         errno = EBUSY;
         return REDISMODULE_ERR;
+    } else if (listLength(module->usedby)) {
+        errno = EPERM;
+        return REDISMODULE_ERR;
     }
 
     moduleUnregisterCommands(module);
+    moduleUnregisterSharedAPI(module);
+    moduleUnregisterUsedAPI(module);
 
     /* Remove any notification subscribers this module might have */
     moduleUnsubscribeNotifications(module);
@@ -4856,7 +4990,12 @@ NULL
                 errmsg = "no such module with that name";
                 break;
             case EBUSY:
-                errmsg = "the module exports one or more module-side data types, can't unload";
+                errmsg = "the module exports one or more module-side data "
+                         "types, can't unload";
+                break;
+            case EPERM:
+                errmsg = "the module exports APIs used by other modules. "
+                         "Please unload them first and try again";
                 break;
             default:
                 errmsg = "operation not possible.";
@@ -4894,6 +5033,7 @@ size_t moduleCount(void) {
  * file so that's easy to seek it to add new entries. */
 void moduleRegisterCoreAPI(void) {
     server.moduleapi = dictCreate(&moduleAPIDictType,NULL);
+    server.sharedapi = dictCreate(&moduleAPIDictType,NULL);
     REGISTER_API(Alloc);
     REGISTER_API(Calloc);
     REGISTER_API(Realloc);
@@ -5044,4 +5184,6 @@ void moduleRegisterCoreAPI(void) {
     REGISTER_API(DictPrev);
     REGISTER_API(DictCompareC);
     REGISTER_API(DictCompare);
+    REGISTER_API(ExportSharedAPI);
+    REGISTER_API(GetSharedAPI);
 }

src/multi.c

@@ -35,6 +35,7 @@
 void initClientMultiState(client *c) {
     c->mstate.commands = NULL;
     c->mstate.count = 0;
+    c->mstate.cmd_flags = 0;
 }
 
 /* Release all the resources associated with MULTI/EXEC state */
@@ -67,6 +68,7 @@ void queueMultiCommand(client *c) {
     for (j = 0; j < c->argc; j++)
         incrRefCount(mc->argv[j]);
     c->mstate.count++;
+    c->mstate.cmd_flags |= c->cmd->flags;
 }
 
 void discardTransaction(client *c) {
@@ -137,6 +139,21 @@ void execCommand(client *c) {
         goto handle_monitor;
     }
 
+    /* If there are write commands inside the transaction, and this is a read
+     * only slave, we want to send an error. This happens when the transaction
+     * was initiated when the instance was a master or a writable replica and
+     * then the configuration changed (for example instance was turned into
+     * a replica). */
+    if (!server.loading && server.masterhost && server.repl_slave_ro &&
+        !(c->flags & CLIENT_MASTER) && c->mstate.cmd_flags & CMD_WRITE)
+    {
+        addReplyError(c,
+            "Transaction contains write commands but instance "
+            "is now a read-only replica. EXEC aborted.");
+        discardTransaction(c);
+        goto handle_monitor;
+    }
+
     /* Exec all the queued commands */
     unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
     orig_argv = c->argv;

src/networking.c

@@ -365,19 +365,13 @@ void addReplyErrorLength(client *c, const char *s, size_t len) {
      * Where the master must propagate the first change even if the second
      * will produce an error. However it is useful to log such events since
      * they are rare and may hint at errors in a script or a bug in Redis. */
-    if (c->flags & (CLIENT_MASTER|CLIENT_SLAVE)) {
+    if (c->flags & (CLIENT_MASTER|CLIENT_SLAVE) && !(c->flags & CLIENT_MONITOR)) {
         char* to = c->flags & CLIENT_MASTER? "master": "replica";
         char* from = c->flags & CLIENT_MASTER? "replica": "master";
         char *cmdname = c->lastcmd ? c->lastcmd->name : "<unknown>";
         serverLog(LL_WARNING,"== CRITICAL == This %s is sending an error "
                              "to its %s: '%s' after processing the command "
                              "'%s'", from, to, s, cmdname);
-        /* Here we want to panic because when a master is sending an
-         * error to some slave in the context of replication, this can
-         * only create some kind of offset or data desynchronization. Better
-         * to catch it ASAP and crash instead of continuing. */
-        if (c->flags & CLIENT_SLAVE)
-            serverPanic("Continuing is unsafe: replication protocol violation.");
     }
 }
 
@@ -1470,7 +1464,7 @@ void processInputBuffer(client *c) {
     }
 
     /* Trim to pos */
-    if (c->qb_pos) {
+    if (server.current_client != NULL && c->qb_pos) {
         sdsrange(c->querybuf,c->qb_pos,-1);
         c->qb_pos = 0;
     }
@@ -2109,6 +2103,7 @@ int checkClientOutputBufferLimits(client *c) {
  * called from contexts where the client can't be freed safely, i.e. from the
  * lower level functions pushing data inside the client output buffers. */
 void asyncCloseClientOnOutputBufferLimitReached(client *c) {
+    if (c->fd == -1) return; /* It is unsafe to free fake clients. */
     serverAssert(c->reply_bytes < SIZE_MAX-(1024*64));
     if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return;
     if (checkClientOutputBufferLimits(c)) {

src/object.c

@@ -185,7 +185,7 @@ robj *createStringObjectFromLongDouble(long double value, int humanfriendly) {
 /* Duplicate a string object, with the guarantee that the returned object
  * has the same encoding as the original one.
  *
- * This function also guarantees that duplicating a small integere object
+ * This function also guarantees that duplicating a small integer object
  * (or a string object that contains a representation of a small integer)
  * will always result in a fresh object that is unshared (refcount == 1).
  *
@@ -1011,7 +1011,7 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
 
     mem = 0;
     if (server.aof_state != AOF_OFF) {
-        mem += sdslen(server.aof_buf);
+        mem += sdsalloc(server.aof_buf);
         mem += aofRewriteBufferSize();
     }
     mh->aof_buffer = mem;
@@ -1285,9 +1285,18 @@ NULL
  *
  * Usage: MEMORY usage <key> */
 void memoryCommand(client *c) {
-    robj *o;
-
-    if (!strcasecmp(c->argv[1]->ptr,"usage") && c->argc >= 3) {
+    if (!strcasecmp(c->argv[1]->ptr,"help") && c->argc == 2) {
+        const char *help[] = {
+"DOCTOR - Return memory problems reports.",
+"MALLOC-STATS -- Return internal statistics report from the memory allocator.",
+"PURGE -- Attempt to purge dirty pages for reclamation by the allocator.",
+"STATS -- Return information about the memory usage of the server.",
+"USAGE <key> [SAMPLES <count>] -- Return memory in bytes used by <key> and its value. Nested values are sampled up to <count> times (default: 5).",
+NULL
+        };
+        addReplyHelp(c, help);
+    } else if (!strcasecmp(c->argv[1]->ptr,"usage") && c->argc >= 3) {
+        dictEntry *de;
         long long samples = OBJ_COMPUTE_SIZE_DEF_SAMPLES;
         for (int j = 3; j < c->argc; j++) {
             if (!strcasecmp(c->argv[j]->ptr,"samples") &&
@@ -1306,10 +1315,12 @@ void memoryCommand(client *c) {
                 return;
             }
         }
-        if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk))
-                == NULL) return;
-        size_t usage = objectComputeSize(o,samples);
-        usage += sdsAllocSize(c->argv[2]->ptr);
+        if ((de = dictFind(c->db->dict,c->argv[2]->ptr)) == NULL) {
+            addReply(c, shared.nullbulk);
+            return;
+        }
+        size_t usage = objectComputeSize(dictGetVal(de),samples);
+        usage += sdsAllocSize(dictGetKey(de));
         usage += sizeof(dictEntry);
         addReplyLongLong(c,usage);
     } else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) {
@@ -1434,19 +1445,7 @@ void memoryCommand(client *c) {
         addReply(c, shared.ok);
         /* Nothing to do for other allocators. */
 #endif
-    } else if (!strcasecmp(c->argv[1]->ptr,"help") && c->argc == 2) {
-        addReplyMultiBulkLen(c,5);
-        addReplyBulkCString(c,
-"MEMORY DOCTOR                        - Outputs memory problems report");
-        addReplyBulkCString(c,
-"MEMORY USAGE <key> [SAMPLES <count>] - Estimate memory usage of key");
-        addReplyBulkCString(c,
-"MEMORY STATS                         - Show memory usage details");
-        addReplyBulkCString(c,
-"MEMORY PURGE                         - Ask the allocator to release memory");
-        addReplyBulkCString(c,
-"MEMORY MALLOC-STATS                  - Show allocator internal stats");
     } else {
-        addReplyError(c,"Syntax error. Try MEMORY HELP");
+        addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try MEMORY HELP", (char*)c->argv[1]->ptr);
     }
 }

src/quicklist.h

@@ -40,7 +40,7 @@
  * container: 2 bits, NONE=1, ZIPLIST=2.
  * recompress: 1 bit, bool, true if node is temporarry decompressed for usage.
  * attempted_compress: 1 bit, boolean, used for verifying during testing.
- * extra: 12 bits, free for future use; pads out the remainder of 32 bits */
+ * extra: 10 bits, free for future use; pads out the remainder of 32 bits */
 typedef struct quicklistNode {
     struct quicklistNode *prev;
     struct quicklistNode *next;

src/rax.c

@@ -1406,7 +1406,7 @@ int raxIteratorNextStep(raxIterator *it, int noup) {
     }
 }
 
-/* Seek the grestest key in the subtree at the current node. Return 0 on
+/* Seek the greatest key in the subtree at the current node. Return 0 on
  * out of memory, otherwise 1. This is an helper function for different
  * iteration functions below. */
 int raxSeekGreatest(raxIterator *it) {

src/rdb.c

@@ -1645,6 +1645,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
              * node: the entries inside the listpack itself are delta-encoded
              * relatively to this ID. */
             sds nodekey = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL);
+            if (nodekey == NULL) {
+                rdbExitReportCorruptRDB("Stream master ID loading failed: invalid encoding or I/O error.");
+            }
             if (sdslen(nodekey) != sizeof(streamID)) {
                 rdbExitReportCorruptRDB("Stream node key entry is not the "
                                         "size of a stream ID");

src/redis-cli.c

@@ -67,6 +67,7 @@
 #define REDIS_CLI_HISTFILE_DEFAULT ".rediscli_history"
 #define REDIS_CLI_RCFILE_ENV "REDISCLI_RCFILE"
 #define REDIS_CLI_RCFILE_DEFAULT ".redisclirc"
+#define REDIS_CLI_AUTH_ENV "REDISCLI_AUTH"
 
 #define CLUSTER_MANAGER_SLOTS               16384
 #define CLUSTER_MANAGER_MIGRATE_TIMEOUT     60000
@@ -116,6 +117,7 @@
 #define CLUSTER_MANAGER_CMD_FLAG_REPLACE        1 << 6
 #define CLUSTER_MANAGER_CMD_FLAG_COPY           1 << 7
 #define CLUSTER_MANAGER_CMD_FLAG_COLOR          1 << 8
+#define CLUSTER_MANAGER_CMD_FLAG_CHECK_OWNERS   1 << 9
 
 #define CLUSTER_MANAGER_OPT_GETFRIENDS  1 << 0
 #define CLUSTER_MANAGER_OPT_COLD        1 << 1
@@ -1377,6 +1379,9 @@ static int parseOptions(int argc, char **argv) {
         } else if (!strcmp(argv[i],"--cluster-use-empty-masters")) {
             config.cluster_manager_command.flags |=
                 CLUSTER_MANAGER_CMD_FLAG_EMPTYMASTER;
+        } else if (!strcmp(argv[i],"--cluster-search-multiple-owners")) {
+            config.cluster_manager_command.flags |=
+                CLUSTER_MANAGER_CMD_FLAG_CHECK_OWNERS;
         } else if (!strcmp(argv[i],"-v") || !strcmp(argv[i], "--version")) {
             sds version = cliVersion();
             printf("redis-cli %s\n", version);
@@ -1419,6 +1424,14 @@ static int parseOptions(int argc, char **argv) {
     return i;
 }
 
+static void parseEnv() {
+    /* Set auth from env, but do not overwrite CLI arguments if passed */
+    char *auth = getenv(REDIS_CLI_AUTH_ENV);
+    if (auth != NULL && config.auth == NULL) {
+        config.auth = auth;
+    }
+}
+
 static sds readArgFromStdin(void) {
     char buf[1024];
     sds arg = sdsempty();
@@ -1446,6 +1459,9 @@ static void usage(void) {
 "  -p <port>          Server port (default: 6379).\n"
 "  -s <socket>        Server socket (overrides hostname and port).\n"
 "  -a <password>      Password to use when connecting to the server.\n"
+"                     You can also use the " REDIS_CLI_AUTH_ENV " environment\n"
+"                     variable to pass this password more safely\n"
+"                     (if both are used, this argument takes predecence).\n"
 "  -u <uri>           Server URI.\n"
 "  -r <repeat>        Execute specified command N times.\n"
 "  -i <interval>      When -r is used, waits <interval> seconds per command.\n"
@@ -1834,7 +1850,7 @@ static int evalMode(int argc, char **argv) {
         if (eval_ldb) {
             if (!config.eval_ldb) {
                 /* If the debugging session ended immediately, there was an
-                 * error compiling the script. Show it and don't enter
+                 * error compiling the script. Show it and they don't enter
                  * the REPL at all. */
                 printf("Eval debugging session can't start:\n");
                 cliReadReply(0);
@@ -1917,6 +1933,7 @@ static dictType clusterManagerDictType = {
 };
 
 typedef int clusterManagerCommandProc(int argc, char **argv);
+typedef int (*clusterManagerOnReplyError)(redisReply *reply, int bulk_idx);
 
 /* Cluster Manager helper functions */
 
@@ -1978,14 +1995,17 @@ typedef struct clusterManagerCommandDef {
 clusterManagerCommandDef clusterManagerCommands[] = {
     {"create", clusterManagerCommandCreate, -2, "host1:port1 ... hostN:portN",
      "replicas <arg>"},
-    {"check", clusterManagerCommandCheck, -1, "host:port", NULL},
+    {"check", clusterManagerCommandCheck, -1, "host:port",
+     "search-multiple-owners"},
     {"info", clusterManagerCommandInfo, -1, "host:port", NULL},
-    {"fix", clusterManagerCommandFix, -1, "host:port", NULL},
+    {"fix", clusterManagerCommandFix, -1, "host:port",
+     "search-multiple-owners"},
     {"reshard", clusterManagerCommandReshard, -1, "host:port",
-     "from <arg>,to <arg>,slots <arg>,yes,timeout <arg>,pipeline <arg>"},
+     "from <arg>,to <arg>,slots <arg>,yes,timeout <arg>,pipeline <arg>,"
+     "replace"},
     {"rebalance", clusterManagerCommandRebalance, -1, "host:port",
      "weight <node1=w1...nodeN=wN>,use-empty-masters,"
-     "timeout <arg>,simulate,pipeline <arg>,threshold <arg>"},
+     "timeout <arg>,simulate,pipeline <arg>,threshold <arg>,replace"},
     {"add-node", clusterManagerCommandAddNode, 2,
      "new_host:new_port existing_host:existing_port", "slave,master-id <arg>"},
     {"del-node", clusterManagerCommandDeleteNode, 2, "host:port node_id",NULL},
@@ -2176,6 +2196,44 @@ static int clusterManagerCheckRedisReply(clusterManagerNode *n,
     return 1;
 }
 
+/* Execute MULTI command on a cluster node. */
+static int clusterManagerStartTransaction(clusterManagerNode *node) {
+    redisReply *reply = CLUSTER_MANAGER_COMMAND(node, "MULTI");
+    int success = clusterManagerCheckRedisReply(node, reply, NULL);
+    if (reply) freeReplyObject(reply);
+    return success;
+}
+
+/* Execute EXEC command on a cluster node. */
+static int clusterManagerExecTransaction(clusterManagerNode *node,
+                                         clusterManagerOnReplyError onerror)
+{
+    redisReply *reply = CLUSTER_MANAGER_COMMAND(node, "EXEC");
+    int success = clusterManagerCheckRedisReply(node, reply, NULL);
+    if (success) {
+        if (reply->type != REDIS_REPLY_ARRAY) {
+            success = 0;
+            goto cleanup;
+        }
+        size_t i;
+        for (i = 0; i < reply->elements; i++) {
+            redisReply *r = reply->element[i];
+            char *err = NULL;
+            success = clusterManagerCheckRedisReply(node, r, &err);
+            if (!success && onerror) success = onerror(r, i);
+            if (err) {
+                if (!success)
+                    CLUSTER_MANAGER_PRINT_REPLY_ERROR(node, err);
+                zfree(err);
+            }
+            if (!success) break;
+        }
+    }
+cleanup:
+    if (reply) freeReplyObject(reply);
+    return success;
+}
+
 static int clusterManagerNodeConnect(clusterManagerNode *node) {
     if (node->context) redisFree(node->context);
     node->context = redisConnect(node->ip, node->port);
@@ -2726,8 +2784,7 @@ static int clusterManagerSetSlot(clusterManagerNode *node1,
         if (err != NULL) {
             *err = zmalloc((reply->len + 1) * sizeof(char));
             strcpy(*err, reply->str);
-            CLUSTER_MANAGER_PRINT_REPLY_ERROR(node1, err);
-        }
+        } else CLUSTER_MANAGER_PRINT_REPLY_ERROR(node1, reply->str);
         goto cleanup;
     }
 cleanup:
@@ -2735,6 +2792,84 @@ cleanup:
     return success;
 }
 
+static int clusterManagerClearSlotStatus(clusterManagerNode *node, int slot) {
+    redisReply *reply = CLUSTER_MANAGER_COMMAND(node,
+        "CLUSTER SETSLOT %d %s", slot, "STABLE");
+    int success = clusterManagerCheckRedisReply(node, reply, NULL);
+    if (reply) freeReplyObject(reply);
+    return success;
+}
+
+static int clusterManagerDelSlot(clusterManagerNode *node, int slot,
+                                 int ignore_unassigned_err)
+{
+    redisReply *reply = CLUSTER_MANAGER_COMMAND(node,
+        "CLUSTER DELSLOTS %d", slot);
+    char *err = NULL;
+    int success = clusterManagerCheckRedisReply(node, reply, &err);
+    if (!success && reply && reply->type == REDIS_REPLY_ERROR &&
+        ignore_unassigned_err &&
+        strstr(reply->str, "already unassigned") != NULL) success = 1;
+    if (!success && err != NULL) {
+        CLUSTER_MANAGER_PRINT_REPLY_ERROR(node, err);
+        zfree(err);
+    }
+    if (reply) freeReplyObject(reply);
+    return success;
+}
+
+static int clusterManagerAddSlot(clusterManagerNode *node, int slot) {
+    redisReply *reply = CLUSTER_MANAGER_COMMAND(node,
+        "CLUSTER ADDSLOTS %d", slot);
+    int success = clusterManagerCheckRedisReply(node, reply, NULL);
+    if (reply) freeReplyObject(reply);
+    return success;
+}
+
+static signed int clusterManagerCountKeysInSlot(clusterManagerNode *node,
+                                                int slot)
+{
+    redisReply *reply = CLUSTER_MANAGER_COMMAND(node,
+        "CLUSTER COUNTKEYSINSLOT %d", slot);
+    int count = -1;
+    int success = clusterManagerCheckRedisReply(node, reply, NULL);
+    if (success && reply->type == REDIS_REPLY_INTEGER) count = reply->integer;
+    if (reply) freeReplyObject(reply);
+    return count;
+}
+
+static int clusterManagerBumpEpoch(clusterManagerNode *node) {
+    redisReply *reply = CLUSTER_MANAGER_COMMAND(node, "CLUSTER BUMPEPOCH");
+    int success = clusterManagerCheckRedisReply(node, reply, NULL);
+    if (reply) freeReplyObject(reply);
+    return success;
+}
+
+static int clusterManagerIgnoreUnassignedErr(redisReply *reply, int bulk_idx) {
+    if (bulk_idx == 0 && reply) {
+        if (reply->type == REDIS_REPLY_ERROR)
+            return strstr(reply->str, "already unassigned") != NULL;
+    }
+    return 0;
+}
+
+static int clusterManagerSetSlotOwner(clusterManagerNode *owner,
+                                      int slot,
+                                      int do_clear)
+{
+    int success = clusterManagerStartTransaction(owner);
+    if (!success) return 0;
+    /* Ensure the slot is not already assigned. */
+    clusterManagerDelSlot(owner, slot, 1);
+    /* Add the slot and bump epoch. */
+    clusterManagerAddSlot(owner, slot);
+    if (do_clear) clusterManagerClearSlotStatus(owner, slot);
+    clusterManagerBumpEpoch(owner);
+    success = clusterManagerExecTransaction(owner,
+        clusterManagerIgnoreUnassignedErr);
+    return success;
+}
+
 /* Migrate keys taken from reply->elements. It returns the reply from the
  * MIGRATE command, or NULL if something goes wrong. If the argument 'dots'
  * is not NULL, a dot will be printed for every migrated key. */
@@ -2815,8 +2950,8 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
                                            char **err)
 {
     int success = 1;
-    int do_fix = (config.cluster_manager_command.flags &
-                  CLUSTER_MANAGER_CMD_FLAG_FIX);
+    int replace_existing_keys = (config.cluster_manager_command.flags &
+            (CLUSTER_MANAGER_CMD_FLAG_FIX | CLUSTER_MANAGER_CMD_FLAG_REPLACE));
     while (1) {
         char *dots = NULL;
         redisReply *reply = NULL, *migrate_reply = NULL;
@@ -2830,7 +2965,7 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
             if (err != NULL) {
                 *err = zmalloc((reply->len + 1) * sizeof(char));
                 strcpy(*err, reply->str);
-                CLUSTER_MANAGER_PRINT_REPLY_ERROR(source, err);
+                CLUSTER_MANAGER_PRINT_REPLY_ERROR(source, *err);
             }
             goto next;
         }
@@ -2847,15 +2982,23 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
                                                          dots);
         if (migrate_reply == NULL) goto next;
         if (migrate_reply->type == REDIS_REPLY_ERROR) {
-            if (do_fix && strstr(migrate_reply->str, "BUSYKEY")) {
+            int is_busy = strstr(migrate_reply->str, "BUSYKEY") != NULL;
+            int not_served = strstr(migrate_reply->str, "slot not served") != NULL;
+            if (replace_existing_keys && (is_busy || not_served)) {
+                /* If the key already exists, try to migrate keys
+                 * adding REPLACE option.
+                 * If the key's slot is not served, try to assign slot
+                 * to the target node. */
+                if (not_served)
+                    clusterManagerSetSlot(source, target, slot, "node", NULL);
                 clusterManagerLogWarn("*** Target key exists. "
                                       "Replacing it for FIX.\n");
                 freeReplyObject(migrate_reply);
-                /* Try to migrate keys adding REPLACE option. */
                 migrate_reply = clusterManagerMigrateKeysInReply(source,
                                                                  target,
                                                                  reply,
-                                                                 1, timeout,
+                                                                 is_busy,
+                                                                 timeout,
                                                                  NULL);
                 success = (migrate_reply != NULL &&
                            migrate_reply->type != REDIS_REPLY_ERROR);
@@ -2941,7 +3084,7 @@ static int clusterManagerMoveSlot(clusterManagerNode *source,
                 if (err != NULL) {
                     *err = zmalloc((r->len + 1) * sizeof(char));
                     strcpy(*err, r->str);
-                    CLUSTER_MANAGER_PRINT_REPLY_ERROR(n, err);
+                    CLUSTER_MANAGER_PRINT_REPLY_ERROR(n, *err);
                 }
             }
             freeReplyObject(r);
@@ -3301,8 +3444,8 @@ static sds clusterManagerGetConfigSignature(clusterManagerNode *node) {
                 nodename = token;
                 tot_size = (p - token);
                 name_len = tot_size++; // Make room for ':' in tot_size
-            } else if (i == 8) break;
-            i++;
+            }
+            if (++i == 8) break;
         }
         if (i != 8) continue;
         if (nodename == NULL) continue;
@@ -3341,7 +3484,7 @@ static sds clusterManagerGetConfigSignature(clusterManagerNode *node) {
             char *sp = cfg + name_len;
             *(sp++) = ':';
             for (i = 0; i < c; i++) {
-                if (i > 0) *(sp++) = '|';
+                if (i > 0) *(sp++) = ',';
                 int slen = strlen(slots[i]);
                 memcpy(sp, slots[i], slen);
                 sp += slen;
@@ -3499,6 +3642,34 @@ static clusterManagerNode *clusterManagerNodeWithLeastReplicas() {
     return node;
 }
 
+/* This function returns a random master node, return NULL if none */
+
+static clusterManagerNode *clusterManagerNodeMasterRandom() {
+    int master_count = 0;
+    int idx;
+    listIter li;
+    listNode *ln;
+    listRewind(cluster_manager.nodes, &li);
+    while ((ln = listNext(&li)) != NULL) {
+        clusterManagerNode *n = ln->value;
+        if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
+        master_count++;
+    }
+
+    srand(time(NULL));
+    idx = rand() % master_count;
+    listRewind(cluster_manager.nodes, &li);
+    while ((ln = listNext(&li)) != NULL) {
+        clusterManagerNode *n = ln->value;
+        if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
+        if (!idx--) {
+            return n;
+        }
+    }
+    /* Can not be reached */
+    return NULL;
+}
+
 static int clusterManagerFixSlotsCoverage(char *all_slots) {
     int i, fixed = 0;
     list *none = NULL, *single = NULL, *multi = NULL;
@@ -3571,26 +3742,22 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
                "across the cluster:\n");
         clusterManagerPrintSlotsList(none);
         if (confirmWithYes("Fix these slots by covering with a random node?")){
-            srand(time(NULL));
             listIter li;
             listNode *ln;
             listRewind(none, &li);
             while ((ln = listNext(&li)) != NULL) {
                 sds slot = ln->value;
-                long idx = (long) (rand() % listLength(cluster_manager.nodes));
-                listNode *node_n = listIndex(cluster_manager.nodes, idx);
-                assert(node_n != NULL);
-                clusterManagerNode *n = node_n->value;
+                int s = atoi(slot);
+                clusterManagerNode *n = clusterManagerNodeMasterRandom();
                 clusterManagerLogInfo(">>> Covering slot %s with %s:%d\n",
                                       slot, n->ip, n->port);
-                redisReply *r = CLUSTER_MANAGER_COMMAND(n,
-                    "CLUSTER ADDSLOTS %s", slot);
-                if (!clusterManagerCheckRedisReply(n, r, NULL)) fixed = -1;
-                if (r) freeReplyObject(r);
-                if (fixed < 0) goto cleanup;
+                if (!clusterManagerSetSlotOwner(n, s, 0)) {
+                    fixed = -1;
+                    goto cleanup;
+                }
                 /* Since CLUSTER ADDSLOTS succeeded, we also update the slot
                  * info into the node struct, in order to keep it synced */
-                n->slots[atoi(slot)] = 1;
+                n->slots[s] = 1;
                 fixed++;
             }
         }
@@ -3598,7 +3765,7 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
 
     /*  Handle case "2": keys only in one node. */
     if (listLength(single) > 0) {
-        printf("The following uncovered slots  have keys in just one node:\n");
+        printf("The following uncovered slots have keys in just one node:\n");
         clusterManagerPrintSlotsList(single);
         if (confirmWithYes("Fix these slots by covering with those nodes?")){
             listIter li;
@@ -3606,6 +3773,7 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
             listRewind(single, &li);
             while ((ln = listNext(&li)) != NULL) {
                 sds slot = ln->value;
+                int s = atoi(slot);
                 dictEntry *entry = dictFind(clusterManagerUncoveredSlots, slot);
                 assert(entry != NULL);
                 list *nodes = (list *) dictGetVal(entry);
@@ -3614,11 +3782,10 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
                 clusterManagerNode *n = fn->value;
                 clusterManagerLogInfo(">>> Covering slot %s with %s:%d\n",
                                       slot, n->ip, n->port);
-                redisReply *r = CLUSTER_MANAGER_COMMAND(n,
-                    "CLUSTER ADDSLOTS %s", slot);
-                if (!clusterManagerCheckRedisReply(n, r, NULL)) fixed = -1;
-                if (r) freeReplyObject(r);
-                if (fixed < 0) goto cleanup;
+                if (!clusterManagerSetSlotOwner(n, s, 0)) {
+                    fixed = -1;
+                    goto cleanup;
+                }
                 /* Since CLUSTER ADDSLOTS succeeded, we also update the slot
                  * info into the node struct, in order to keep it synced */
                 n->slots[atoi(slot)] = 1;
@@ -3651,16 +3818,10 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
                 clusterManagerLogInfo(">>> Covering slot %s moving keys "
                                       "to %s:%d\n", slot,
                                       target->ip, target->port);
-                redisReply *r = CLUSTER_MANAGER_COMMAND(target,
-                    "CLUSTER ADDSLOTS %s", slot);
-                if (!clusterManagerCheckRedisReply(target, r, NULL)) fixed = -1;
-                if (r) freeReplyObject(r);
-                if (fixed < 0) goto cleanup;
-                r = CLUSTER_MANAGER_COMMAND(target,
-                    "CLUSTER SETSLOT %s %s", slot, "STABLE");
-                if (!clusterManagerCheckRedisReply(target, r, NULL)) fixed = -1;
-                if (r) freeReplyObject(r);
-                if (fixed < 0) goto cleanup;
+                if (!clusterManagerSetSlotOwner(target, s, 1)) {
+                    fixed = -1;
+                    goto cleanup;
+                }
                 /* Since CLUSTER ADDSLOTS succeeded, we also update the slot
                  * info into the node struct, in order to keep it synced */
                 target->slots[atoi(slot)] = 1;
@@ -3670,16 +3831,16 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
                 while ((nln = listNext(&nli)) != NULL) {
                     clusterManagerNode *src = nln->value;
                     if (src == target) continue;
+                    /* Assign the slot to target node in the source node. */
+                    if (!clusterManagerSetSlot(src, target, s, "NODE", NULL))
+                        fixed = -1;
+                    if (fixed < 0) goto cleanup;
                     /* Set the source node in 'importing' state
                      * (even if we will actually migrate keys away)
                      * in order to avoid receiving redirections
                      * for MIGRATE. */
-                    redisReply *r = CLUSTER_MANAGER_COMMAND(src,
-                        "CLUSTER SETSLOT %s %s %s", slot,
-                        "IMPORTING", target->name);
-                    if (!clusterManagerCheckRedisReply(target, r, NULL))
-                        fixed = -1;
-                    if (r) freeReplyObject(r);
+                    if (!clusterManagerSetSlot(src, target, s,
+                                               "IMPORTING", NULL)) fixed = -1;
                     if (fixed < 0) goto cleanup;
                     int opts = CLUSTER_MANAGER_OPT_VERBOSE |
                                CLUSTER_MANAGER_OPT_COLD;
@@ -3687,6 +3848,9 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
                         fixed = -1;
                         goto cleanup;
                     }
+                    if (!clusterManagerClearSlotStatus(src, s))
+                        fixed = -1;
+                    if (fixed < 0) goto cleanup;
                 }
                 fixed++;
             }
@@ -3720,11 +3884,22 @@ static int clusterManagerFixOpenSlot(int slot) {
     while ((ln = listNext(&li)) != NULL) {
         clusterManagerNode *n = ln->value;
         if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
-        if (n->slots[slot]) {
-            if (owner == NULL) owner = n;
-            listAddNodeTail(owners, n);
+        if (n->slots[slot]) listAddNodeTail(owners, n);
+        else {
+            redisReply *r = CLUSTER_MANAGER_COMMAND(n,
+                "CLUSTER COUNTKEYSINSLOT %d", slot);
+            success = clusterManagerCheckRedisReply(n, r, NULL);
+            if (success && r->integer > 0) {
+                clusterManagerLogWarn("*** Found keys about slot %d "
+                                      "in non-owner node %s:%d!\n", slot,
+                                      n->ip, n->port);
+                listAddNodeTail(owners, n);
+            }
+            if (r) freeReplyObject(r);
+            if (!success) goto cleanup;
         }
     }
+    if (listLength(owners) == 1) owner = listFirst(owners)->value;
     listRewind(cluster_manager.nodes, &li);
     while ((ln = listNext(&li)) != NULL) {
         clusterManagerNode *n = ln->value;
@@ -3735,7 +3910,7 @@ static int clusterManagerFixOpenSlot(int slot) {
                 sds migrating_slot = n->migrating[i];
                 if (atoi(migrating_slot) == slot) {
                     char *sep = (listLength(migrating) == 0 ? "" : ",");
-                    migrating_str = sdscatfmt(migrating_str, "%s%S:%u",
+                    migrating_str = sdscatfmt(migrating_str, "%s%s:%u",
                                               sep, n->ip, n->port);
                     listAddNodeTail(migrating, n);
                     is_migrating = 1;
@@ -3748,7 +3923,7 @@ static int clusterManagerFixOpenSlot(int slot) {
                 sds importing_slot = n->importing[i];
                 if (atoi(importing_slot) == slot) {
                     char *sep = (listLength(importing) == 0 ? "" : ",");
-                    importing_str = sdscatfmt(importing_str, "%s%S:%u",
+                    importing_str = sdscatfmt(importing_str, "%s%s:%u",
                                               sep, n->ip, n->port);
                     listAddNodeTail(importing, n);
                     is_importing = 1;
@@ -3767,15 +3942,20 @@ static int clusterManagerFixOpenSlot(int slot) {
                 clusterManagerLogWarn("*** Found keys about slot %d "
                                       "in node %s:%d!\n", slot, n->ip,
                                       n->port);
+                char *sep = (listLength(importing) == 0 ? "" : ",");
+                importing_str = sdscatfmt(importing_str, "%s%S:%u",
+                                          sep, n->ip, n->port);
                 listAddNodeTail(importing, n);
             }
             if (r) freeReplyObject(r);
             if (!success) goto cleanup;
         }
     }
-    printf("Set as migrating in: %s\n", migrating_str);
-    printf("Set as importing in: %s\n", importing_str);
-    /* If there is no slot owner, set as owner the slot with the biggest
+    if (sdslen(migrating_str) > 0)
+        printf("Set as migrating in: %s\n", migrating_str);
+    if (sdslen(importing_str) > 0)
+        printf("Set as importing in: %s\n", importing_str);
+    /* If there is no slot owner, set as owner the node with the biggest
      * number of keys, among the set of migrating / importing nodes. */
     if (owner == NULL) {
         clusterManagerLogInfo(">>> Nobody claims ownership, "
@@ -3793,15 +3973,9 @@ static int clusterManagerFixOpenSlot(int slot) {
         // Use ADDSLOTS to assign the slot.
         clusterManagerLogWarn("*** Configuring %s:%d as the slot owner\n",
                               owner->ip, owner->port);
-        redisReply *reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER "
-                                                    "SETSLOT %d %s",
-                                                    slot, "STABLE");
-        success = clusterManagerCheckRedisReply(owner, reply, NULL);
-        if (reply) freeReplyObject(reply);
+        success = clusterManagerClearSlotStatus(owner, slot);
         if (!success) goto cleanup;
-        reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER ADDSLOTS %d", slot);
-        success = clusterManagerCheckRedisReply(owner, reply, NULL);
-        if (reply) freeReplyObject(reply);
+        success = clusterManagerSetSlotOwner(owner, slot, 0);
         if (!success) goto cleanup;
         /* Since CLUSTER ADDSLOTS succeeded, we also update the slot
          * info into the node struct, in order to keep it synced */
@@ -3809,9 +3983,7 @@ static int clusterManagerFixOpenSlot(int slot) {
         /* Make sure this information will propagate. Not strictly needed
          * since there is no past owner, so all the other nodes will accept
          * whatever epoch this node will claim the slot with. */
-        reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER BUMPEPOCH");
-        success = clusterManagerCheckRedisReply(owner, reply, NULL);
-        if (reply) freeReplyObject(reply);
+        success = clusterManagerBumpEpoch(owner);
         if (!success) goto cleanup;
         /* Remove the owner from the list of migrating/importing
          * nodes. */
@@ -3827,32 +3999,38 @@ static int clusterManagerFixOpenSlot(int slot) {
      * in migrating state, since migrating is a valid state only for
      * slot owners. */
     if (listLength(owners) > 1) {
-        owner = clusterManagerGetNodeWithMostKeysInSlot(owners, slot, NULL);
+        /* Owner cannot be NULL at this point, since if there are more owners,
+         * the owner has been set in the previous condition (owner == NULL). */
+        assert(owner != NULL);
         listRewind(owners, &li);
-        redisReply *reply = NULL;
         while ((ln = listNext(&li)) != NULL) {
             clusterManagerNode *n = ln->value;
             if (n == owner) continue;
-            reply = CLUSTER_MANAGER_COMMAND(n, "CLUSTER DELSLOT %d", slot);
-            success = clusterManagerCheckRedisReply(n, reply, NULL);
-            if (reply) freeReplyObject(reply);
+            success = clusterManagerDelSlot(n, slot, 1);
+            if (!success) goto cleanup;
+            n->slots[slot] = 0;
+            /* Assign the slot to the owner in the node 'n' configuration.' */
+            success = clusterManagerSetSlot(n, owner, slot, "node", NULL);
             if (!success) goto cleanup;
             success = clusterManagerSetSlot(n, owner, slot, "importing", NULL);
             if (!success) goto cleanup;
-            clusterManagerRemoveNodeFromList(importing, n); //Avoid duplicates
+            /* Avoid duplicates. */
+            clusterManagerRemoveNodeFromList(importing, n);
             listAddNodeTail(importing, n);
+            /* Ensure that the node is not in the migrating list. */
+            clusterManagerRemoveNodeFromList(migrating, n);
         }
-        reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER BUMPEPOCH");
-        success = clusterManagerCheckRedisReply(owner, reply, NULL);
-        if (reply) freeReplyObject(reply);
-        if (!success) goto cleanup;
     }
     int move_opts = CLUSTER_MANAGER_OPT_VERBOSE;
-    /* Case 1: The slot is in migrating state in one slot, and in
-     *         importing state in 1 slot. That's trivial to address. */
+    /* Case 1: The slot is in migrating state in one node, and in
+     *         importing state in 1 node. That's trivial to address. */
     if (listLength(migrating) == 1 && listLength(importing) == 1) {
         clusterManagerNode *src = listFirst(migrating)->value;
         clusterManagerNode *dst = listFirst(importing)->value;
+        clusterManagerLogInfo(">>> Case 1: Moving slot %d from "
+                              "%s:%d to %s:%d\n", slot,
+                              src->ip, src->port, dst->ip, dst->port);
+        move_opts |= CLUSTER_MANAGER_OPT_UPDATE;
         success = clusterManagerMoveSlot(src, dst, slot, move_opts, NULL);
     }
     /* Case 2: There are multiple nodes that claim the slot as importing,
@@ -3860,7 +4038,7 @@ static int clusterManagerFixOpenSlot(int slot) {
      * the slot. In this case we just move all the keys to the owner
      * according to the configuration. */
     else if (listLength(migrating) == 0 && listLength(importing) > 0) {
-        clusterManagerLogInfo(">>> Moving all the %d slot keys to its "
+        clusterManagerLogInfo(">>> Case 2: Moving all the %d slot keys to its "
                               "owner %s:%d\n", slot, owner->ip, owner->port);
         move_opts |= CLUSTER_MANAGER_OPT_COLD;
         listRewind(importing, &li);
@@ -3871,38 +4049,117 @@ static int clusterManagerFixOpenSlot(int slot) {
             if (!success) goto cleanup;
             clusterManagerLogInfo(">>> Setting %d as STABLE in "
                                   "%s:%d\n", slot, n->ip, n->port);
-
-            redisReply *r = CLUSTER_MANAGER_COMMAND(n, "CLUSTER SETSLOT %d %s",
-                                                    slot, "STABLE");
-            success = clusterManagerCheckRedisReply(n, r, NULL);
-            if (r) freeReplyObject(r);
+            success = clusterManagerClearSlotStatus(n, slot);
             if (!success) goto cleanup;
         }
+        /* Since the slot has been moved in "cold" mode, ensure that all the
+         * other nodes update their own configuration about the slot itself. */
+        listRewind(cluster_manager.nodes, &li);
+        while ((ln = listNext(&li)) != NULL) {
+            clusterManagerNode *n = ln->value;
+            if (n == owner) continue;
+            if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
+            success = clusterManagerSetSlot(n, owner, slot, "NODE", NULL);
+            if (!success) goto cleanup;
+        }
+    }
+    /* Case 3: The slot is in migrating state in one node but multiple
+     * other nodes claim to be in importing state and don't have any key in
+     * the slot. We search for the importing node having the same ID as
+     * the destination node of the migrating node.
+     * In that case we move the slot from the migrating node to this node and
+     * we close the importing states on all the other importing nodes.
+     * If no importing node has the same ID as the destination node of the
+     * migrating node, the slot's state is closed on both the migrating node
+     * and the importing nodes. */
+    else if (listLength(migrating) == 1 && listLength(importing) > 1) {
+        int try_to_fix = 1;
+        clusterManagerNode *src = listFirst(migrating)->value;
+        clusterManagerNode *dst = NULL;
+        sds target_id = NULL;
+        for (int i = 0; i < src->migrating_count; i += 2) {
+            sds migrating_slot = src->migrating[i];
+            if (atoi(migrating_slot) == slot) {
+                target_id = src->migrating[i + 1];
+                break;
+            }
+        }
+        assert(target_id != NULL);
+        listIter li;
+        listNode *ln;
+        listRewind(importing, &li);
+        while ((ln = listNext(&li)) != NULL) {
+            clusterManagerNode *n = ln->value;
+            int count = clusterManagerCountKeysInSlot(n, slot);
+            if (count > 0) {
+                try_to_fix = 0;
+                break;
+            }
+            if (strcmp(n->name, target_id) == 0) dst = n;
+        }
+        if (!try_to_fix) goto unhandled_case;
+        if (dst != NULL) {
+            clusterManagerLogInfo(">>> Case 3: Moving slot %d from %s:%d to "
+                                  "%s:%d and closing it on all the other "
+                                  "importing nodes.\n",
+                                  slot, src->ip, src->port,
+                                  dst->ip, dst->port);
+            /* Move the slot to the destination node. */
+            success = clusterManagerMoveSlot(src, dst, slot, move_opts, NULL);
+            if (!success) goto cleanup;
+            /* Close slot on all the other importing nodes. */
+            listRewind(importing, &li);
+            while ((ln = listNext(&li)) != NULL) {
+                clusterManagerNode *n = ln->value;
+                if (dst == n) continue;
+                success = clusterManagerClearSlotStatus(n, slot);
+                if (!success) goto cleanup;
+            }
+        } else {
+            clusterManagerLogInfo(">>> Case 3: Closing slot %d on both "
+                                  "migrating and importing nodes.\n", slot);
+            /* Close the slot on both the migrating node and the importing
+             * nodes. */
+            success = clusterManagerClearSlotStatus(src, slot);
+            if (!success) goto cleanup;
+            listRewind(importing, &li);
+            while ((ln = listNext(&li)) != NULL) {
+                clusterManagerNode *n = ln->value;
+                success = clusterManagerClearSlotStatus(n, slot);
+                if (!success) goto cleanup;
+            }
+        }
     } else {
         int try_to_close_slot = (listLength(importing) == 0 &&
                                  listLength(migrating) == 1);
         if (try_to_close_slot) {
             clusterManagerNode *n = listFirst(migrating)->value;
-            redisReply *r = CLUSTER_MANAGER_COMMAND(n,
-                "CLUSTER GETKEYSINSLOT %d %d", slot, 10);
-            success = clusterManagerCheckRedisReply(n, r, NULL);
-            if (r) {
-                if (success) try_to_close_slot = (r->elements == 0);
-                freeReplyObject(r);
+            if (!owner || owner != n) {
+                redisReply *r = CLUSTER_MANAGER_COMMAND(n,
+                    "CLUSTER GETKEYSINSLOT %d %d", slot, 10);
+                success = clusterManagerCheckRedisReply(n, r, NULL);
+                if (r) {
+                    if (success) try_to_close_slot = (r->elements == 0);
+                    freeReplyObject(r);
+                }
+                if (!success) goto cleanup;
             }
-            if (!success) goto cleanup;
         }
-    /* Case 3: There are no slots claiming to be in importing state, but
-     * there is a migrating node that actually don't have any key. We
-     * can just close the slot, probably a reshard interrupted in the middle. */
+        /* Case 4: There are no slots claiming to be in importing state, but
+         * there is a migrating node that actually don't have any key or is the
+         * slot owner. We can just close the slot, probably a reshard
+         * interrupted in the middle. */
         if (try_to_close_slot) {
             clusterManagerNode *n = listFirst(migrating)->value;
+            clusterManagerLogInfo(">>> Case 4: Closing slot %d on %s:%d\n",
+                                  slot, n->ip, n->port);
             redisReply *r = CLUSTER_MANAGER_COMMAND(n, "CLUSTER SETSLOT %d %s",
                                                     slot, "STABLE");
             success = clusterManagerCheckRedisReply(n, r, NULL);
             if (r) freeReplyObject(r);
             if (!success) goto cleanup;
         } else {
+unhandled_case:
             success = 0;
             clusterManagerLogErr("[ERR] Sorry, redis-cli can't fix this slot "
                                  "yet (work in progress). Slot is set as "
@@ -3920,17 +4177,55 @@ cleanup:
     return success;
 }
 
+static int clusterManagerFixMultipleSlotOwners(int slot, list *owners) {
+    clusterManagerLogInfo(">>> Fixing multiple owners for slot %d...\n", slot);
+    int success = 0;
+    assert(listLength(owners) > 1);
+    clusterManagerNode *owner = clusterManagerGetNodeWithMostKeysInSlot(owners,
+                                                                        slot,
+                                                                        NULL);
+    if (!owner) owner = listFirst(owners)->value;
+    clusterManagerLogInfo(">>> Setting slot %d owner: %s:%d\n",
+                          slot, owner->ip, owner->port);
+    /* Set the slot owner. */
+    if (!clusterManagerSetSlotOwner(owner, slot, 0)) return 0;
+    listIter li;
+    listNode *ln;
+    listRewind(cluster_manager.nodes, &li);
+    /* Update configuration in all the other master nodes by assigning the slot
+     * itself to the new owner, and by eventually migrating keys if the node
+     * has keys for the slot. */
+    while ((ln = listNext(&li)) != NULL) {
+        clusterManagerNode *n = ln->value;
+        if (n == owner) continue;
+        if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
+        int count = clusterManagerCountKeysInSlot(n, slot);
+        success = (count >= 0);
+        if (!success) break;
+        clusterManagerDelSlot(n, slot, 1);
+        if (!clusterManagerSetSlot(n, owner, slot, "node", NULL)) return 0;
+        if (count > 0) {
+            int opts = CLUSTER_MANAGER_OPT_VERBOSE |
+                       CLUSTER_MANAGER_OPT_COLD;
+            success = clusterManagerMoveSlot(n, owner, slot, opts, NULL);
+            if (!success) break;
+        }
+    }
+    return success;
+}
+
 static int clusterManagerCheckCluster(int quiet) {
     listNode *ln = listFirst(cluster_manager.nodes);
     if (!ln) return 0;
-    int result = 1;
-    int do_fix = config.cluster_manager_command.flags &
-                 CLUSTER_MANAGER_CMD_FLAG_FIX;
     clusterManagerNode *node = ln->value;
     clusterManagerLogInfo(">>> Performing Cluster Check (using node %s:%d)\n",
                           node->ip, node->port);
+    int result = 1, consistent = 0;
+    int do_fix = config.cluster_manager_command.flags &
+                 CLUSTER_MANAGER_CMD_FLAG_FIX;
     if (!quiet) clusterManagerShowNodes();
-    if (!clusterManagerIsConfigConsistent()) {
+    consistent = clusterManagerIsConfigConsistent();
+    if (!consistent) {
         sds err = sdsnew("[ERR] Nodes don't agree about configuration!");
         clusterManagerOnError(err);
         result = 0;
@@ -3938,7 +4233,7 @@ static int clusterManagerCheckCluster(int quiet) {
         clusterManagerLogOk("[OK] All nodes agree about slots "
                             "configuration.\n");
     }
-    // Check open slots
+    /* Check open slots */
     clusterManagerLogInfo(">>> Check for open slots...\n");
     listIter li;
     listRewind(cluster_manager.nodes, &li);
@@ -3997,7 +4292,7 @@ static int clusterManagerCheckCluster(int quiet) {
         clusterManagerLogErr("%s.\n", (char *) errstr);
         sdsfree(errstr);
         if (do_fix) {
-            // Fix open slots.
+            /* Fix open slots. */
             dictReleaseIterator(iter);
             iter = dictGetIterator(open_slots);
             while ((entry = dictNext(iter)) != NULL) {
@@ -4025,12 +4320,58 @@ static int clusterManagerCheckCluster(int quiet) {
         result = 0;
         if (do_fix/* && result*/) {
             dictType dtype = clusterManagerDictType;
+            dtype.keyDestructor = dictSdsDestructor;
             dtype.valDestructor = dictListDestructor;
             clusterManagerUncoveredSlots = dictCreate(&dtype, NULL);
             int fixed = clusterManagerFixSlotsCoverage(slots);
             if (fixed > 0) result = 1;
         }
     }
+    int search_multiple_owners = config.cluster_manager_command.flags &
+                                 CLUSTER_MANAGER_CMD_FLAG_CHECK_OWNERS;
+    if (search_multiple_owners) {
+        /* Check whether there are multiple owners, even when slots are
+         * fully covered and there are no open slots. */
+        clusterManagerLogInfo(">>> Check for multiple slot owners...\n");
+        int slot = 0;
+        for (; slot < CLUSTER_MANAGER_SLOTS; slot++) {
+            listIter li;
+            listNode *ln;
+            listRewind(cluster_manager.nodes, &li);
+            list *owners = listCreate();
+            while ((ln = listNext(&li)) != NULL) {
+                clusterManagerNode *n = ln->value;
+                if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
+                if (n->slots[slot]) listAddNodeTail(owners, n);
+                else {
+                    /* Nodes having keys for the slot will be considered
+                     * owners too. */
+                    int count = clusterManagerCountKeysInSlot(n, slot);
+                    if (count > 0) listAddNodeTail(owners, n);
+                }
+            }
+            if (listLength(owners) > 1) {
+                result = 0;
+                clusterManagerLogErr("[WARNING] Slot %d has %d owners:\n",
+                                     slot, listLength(owners));
+                listRewind(owners, &li);
+                while ((ln = listNext(&li)) != NULL) {
+                    clusterManagerNode *n = ln->value;
+                    clusterManagerLogErr("    %s:%d\n", n->ip, n->port);
+                }
+                if (do_fix) {
+                    result = clusterManagerFixMultipleSlotOwners(slot, owners);
+                    if (!result) {
+                        clusterManagerLogErr("Failed to fix multiple owners "
+                                             "for slot %d\n", slot);
+                        listRelease(owners);
+                        break;
+                    }
+                }
+            }
+            listRelease(owners);
+        }
+    }
     return result;
 }
 
@@ -4061,7 +4402,7 @@ static clusterManagerNode *clusterNodeForResharding(char *id,
 static list *clusterManagerComputeReshardTable(list *sources, int numslots) {
     list *moved = listCreate();
     int src_count = listLength(sources), i = 0, tot_slots = 0, j;
-    clusterManagerNode **sorted = zmalloc(src_count * sizeof(**sorted));
+    clusterManagerNode **sorted = zmalloc(src_count * sizeof(*sorted));
     listIter li;
     listNode *ln;
     listRewind(sources, &li);
@@ -5100,10 +5441,13 @@ static int clusterManagerCommandSetTimeout(int argc, char **argv) {
                               n->port);
         ok_count++;
         continue;
-reply_err:
+reply_err:;
+        int need_free = 0;
         if (err == NULL) err = "";
+        else need_free = 1;
         clusterManagerLogErr("ERR setting node-timeot for %s:%d: %s\n", n->ip,
                              n->port, err);
+        if (need_free) zfree(err);
         err_count++;
     }
     clusterManagerLogInfo(">>> New node timeout set. %d OK, %d ERR.\n",
@@ -5280,7 +5624,7 @@ static int clusterManagerCommandCall(int argc, char **argv) {
         if (status != REDIS_OK || reply == NULL )
             printf("%s:%d: Failed!\n", n->ip, n->port);
         else {
-            sds formatted_reply = cliFormatReplyTTY(reply, "");
+            sds formatted_reply = cliFormatReplyRaw(reply);
             printf("%s:%d: %s\n", n->ip, n->port, (char *) formatted_reply);
             sdsfree(formatted_reply);
         }
@@ -6657,6 +7001,8 @@ int main(int argc, char **argv) {
     argc -= firstarg;
     argv += firstarg;
 
+    parseEnv();
+
     /* Cluster Manager mode */
     if (CLUSTER_MANAGER_MODE()) {
         clusterManagerCommandProc *proc = validateClusterManagerCommand();

src/redismodule.h

@@ -332,6 +332,8 @@ void REDISMODULE_API_FUNC(RedisModule_GetRandomBytes)(unsigned char *dst, size_t
 void REDISMODULE_API_FUNC(RedisModule_GetRandomHexChars)(char *dst, size_t len);
 void REDISMODULE_API_FUNC(RedisModule_SetDisconnectCallback)(RedisModuleBlockedClient *bc, RedisModuleDisconnectFunc callback);
 void REDISMODULE_API_FUNC(RedisModule_SetClusterFlags)(RedisModuleCtx *ctx, uint64_t flags);
+int REDISMODULE_API_FUNC(RedisModule_ExportSharedAPI)(RedisModuleCtx *ctx, const char *apiname, void *func);
+void *REDISMODULE_API_FUNC(RedisModule_GetSharedAPI)(RedisModuleCtx *ctx, const char *apiname);
 #endif
 
 /* This is included inline inside each Redis module. */
@@ -492,6 +494,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(GetRandomBytes);
     REDISMODULE_GET_API(GetRandomHexChars);
     REDISMODULE_GET_API(SetClusterFlags);
+    REDISMODULE_GET_API(ExportSharedAPI);
+    REDISMODULE_GET_API(GetSharedAPI);
 #endif
 
     if (RedisModule_IsModuleNameBusy && RedisModule_IsModuleNameBusy(name)) return REDISMODULE_ERR;

src/replication.c

@@ -1245,6 +1245,18 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
     if (eof_reached) {
         int aof_is_enabled = server.aof_state != AOF_OFF;
 
+        /* Ensure background save doesn't overwrite synced data */
+        if (server.rdb_child_pid != -1) {
+            serverLog(LL_NOTICE,
+                "Replica is about to load the RDB file received from the "
+                "master, but there is a pending RDB child running. "
+                "Killing process %ld and removing its temp file to avoid "
+                "any race",
+                    (long) server.rdb_child_pid);
+            kill(server.rdb_child_pid,SIGUSR1);
+            rdbRemoveTempFile(server.rdb_child_pid);
+        }
+
         if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
             serverLog(LL_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> REPLICA synchronization: %s", strerror(errno));
             cancelReplicationHandshake();

src/sds.c

@@ -695,7 +695,7 @@ sds sdscatfmt(sds s, char const *fmt, ...) {
  * s = sdstrim(s,"Aa. :");
  * printf("%s\n", s);
  *
- * Output will be just "Hello World".
+ * Output will be just "HelloWorld".
  */
 sds sdstrim(sds s, const char *cset) {
     char *start, *end, *sp, *ep;

src/sentinel.c

@@ -452,13 +452,15 @@ struct redisCommand sentinelcmds[] = {
     {"info",sentinelInfoCommand,-1,"",0,NULL,0,0,0,0,0},
     {"role",sentinelRoleCommand,1,"l",0,NULL,0,0,0,0,0},
     {"client",clientCommand,-2,"rs",0,NULL,0,0,0,0,0},
-    {"shutdown",shutdownCommand,-1,"",0,NULL,0,0,0,0,0}
+    {"shutdown",shutdownCommand,-1,"",0,NULL,0,0,0,0,0},
+    {"auth",authCommand,2,"sltF",0,NULL,0,0,0,0,0}
 };
 
 /* This function overwrites a few normal Redis config default with Sentinel
  * specific defaults. */
 void initSentinelConfig(void) {
     server.port = REDIS_SENTINEL_PORT;
+    server.protected_mode = 0; /* Sentinel must be exposed. */
 }
 
 /* Perform the Sentinel mode initialization. */
@@ -1941,12 +1943,25 @@ werr:
 /* Send the AUTH command with the specified master password if needed.
  * Note that for slaves the password set for the master is used.
  *
+ * In case this Sentinel requires a password as well, via the "requirepass"
+ * configuration directive, we assume we should use the local password in
+ * order to authenticate when connecting with the other Sentinels as well.
+ * So basically all the Sentinels share the same password and use it to
+ * authenticate reciprocally.
+ *
  * We don't check at all if the command was successfully transmitted
  * to the instance as if it fails Sentinel will detect the instance down,
  * will disconnect and reconnect the link and so forth. */
 void sentinelSendAuthIfNeeded(sentinelRedisInstance *ri, redisAsyncContext *c) {
-    char *auth_pass = (ri->flags & SRI_MASTER) ? ri->auth_pass :
-                                                 ri->master->auth_pass;
+    char *auth_pass = NULL;
+
+    if (ri->flags & SRI_MASTER) {
+        auth_pass = ri->auth_pass;
+    } else if (ri->flags & SRI_SLAVE) {
+        auth_pass = ri->master->auth_pass;
+    } else if (ri->flags & SRI_SENTINEL) {
+        if (server.requirepass) auth_pass = server.requirepass;
+    }
 
     if (auth_pass) {
         if (redisAsyncCommand(c, sentinelDiscardReplyCallback, ri, "%s %s",

src/server.c

@@ -1526,10 +1526,10 @@ void initServerConfig(void) {
     server.runid[CONFIG_RUN_ID_SIZE] = '\0';
     changeReplicationId();
     clearReplicationId2();
-    server.timezone = timezone; /* Initialized by tzset(). */
+    server.timezone = getTimeZone(); /* Initialized by tzset(). */
     server.configfile = NULL;
     server.executable = NULL;
-    server.config_hz = CONFIG_DEFAULT_HZ;
+    server.hz = server.config_hz = CONFIG_DEFAULT_HZ;
     server.dynamic_hz = CONFIG_DEFAULT_DYNAMIC_HZ;
     server.arch_bits = (sizeof(long) == 8) ? 64 : 32;
     server.port = CONFIG_DEFAULT_SERVER_PORT;
@@ -1958,9 +1958,13 @@ int listenToPort(int port, int *fds, int *count) {
         }
         if (fds[*count] == ANET_ERR) {
             serverLog(LL_WARNING,
-                "Creating Server TCP listening socket %s:%d: %s",
+                "Could not create server TCP listening socket %s:%d: %s",
                 server.bindaddr[j] ? server.bindaddr[j] : "*",
                 port, server.neterr);
+                if (errno == ENOPROTOOPT     || errno == EPROTONOSUPPORT ||
+                    errno == ESOCKTNOSUPPORT || errno == EPFNOSUPPORT ||
+                    errno == EAFNOSUPPORT    || errno == EADDRNOTAVAIL)
+                    continue;
             return C_ERR;
         }
         anetNonBlock(NULL,fds[*count]);
@@ -2603,24 +2607,23 @@ int processCommand(client *c) {
     }
 
     /* Handle the maxmemory directive.
-     *
-     * First we try to free some memory if possible (if there are volatile
-     * keys in the dataset). If there are not the only thing we can do
-     * is returning an error.
      *
      * Note that we do not want to reclaim memory if we are here re-entering
      * the event loop since there is a busy Lua script running in timeout
-     * condition, to avoid mixing the propagation of scripts with the propagation
-     * of DELs due to eviction. */
+     * condition, to avoid mixing the propagation of scripts with the
+     * propagation of DELs due to eviction. */
     if (server.maxmemory && !server.lua_timedout) {
-        int out_of_memory = freeMemoryIfNeeded() == C_ERR;
+        int out_of_memory = freeMemoryIfNeededAndSafe() == C_ERR;
         /* freeMemoryIfNeeded may flush slave output buffers. This may result
          * into a slave, that may be the active client, to be freed. */
         if (server.current_client == NULL) return C_ERR;
 
         /* It was impossible to free enough memory, and the command the client
-         * is trying to execute is denied during OOM conditions? Error. */
-        if ((c->cmd->flags & CMD_DENYOOM) && out_of_memory) {
+         * is trying to execute is denied during OOM conditions or the client
+         * is in MULTI/EXEC context? Error. */
+        if (out_of_memory &&
+            (c->cmd->flags & CMD_DENYOOM ||
+             (c->flags & CLIENT_MULTI && c->cmd->proc != execCommand))) {
             flagTransaction(c);
             addReply(c, shared.oomerr);
             return C_OK;
@@ -3240,11 +3243,11 @@ sds genRedisInfoString(char *section) {
             "allocator_frag_ratio:%.2f\r\n"
             "allocator_frag_bytes:%zu\r\n"
             "allocator_rss_ratio:%.2f\r\n"
-            "allocator_rss_bytes:%zu\r\n"
+            "allocator_rss_bytes:%zd\r\n"
             "rss_overhead_ratio:%.2f\r\n"
-            "rss_overhead_bytes:%zu\r\n"
+            "rss_overhead_bytes:%zd\r\n"
             "mem_fragmentation_ratio:%.2f\r\n"
-            "mem_fragmentation_bytes:%zu\r\n"
+            "mem_fragmentation_bytes:%zd\r\n"
             "mem_not_counted_for_evict:%zu\r\n"
             "mem_replication_backlog:%zu\r\n"
             "mem_clients_slaves:%zu\r\n"

src/server.h

@@ -654,6 +654,9 @@ typedef struct multiCmd {
 typedef struct multiState {
     multiCmd *commands;     /* Array of MULTI commands */
     int count;              /* Total number of MULTI commands */
+    int cmd_flags;          /* The accumulated command flags OR-ed together.
+                               So if at least a command has a given flag, it
+                               will be set in this field. */
     int minreplicas;        /* MINREPLICAS for synchronous replication */
     time_t minreplicas_timeout; /* MINREPLICAS timeout as unixtime. */
 } multiState;
@@ -864,11 +867,11 @@ struct redisMemOverhead {
     float dataset_perc;
     float peak_perc;
     float total_frag;
-    size_t total_frag_bytes;
+    ssize_t total_frag_bytes;
     float allocator_frag;
-    size_t allocator_frag_bytes;
+    ssize_t allocator_frag_bytes;
     float allocator_rss;
-    size_t allocator_rss_bytes;
+    ssize_t allocator_rss_bytes;
     float rss_extra;
     size_t rss_extra_bytes;
     size_t num_dbs;
@@ -951,7 +954,9 @@ struct redisServer {
     size_t initial_memory_usage; /* Bytes used after initialization. */
     int always_show_logo;       /* Show logo even for non-stdout logging. */
     /* Modules */
-    dict *moduleapi;            /* Exported APIs dictionary for modules. */
+    dict *moduleapi;            /* Exported core APIs dictionary for modules. */
+    dict *sharedapi;            /* Like moduleapi but containing the APIs that
+                                   modules share with each other. */
     list *loadmodule_queue;     /* List of modules to load at startup. */
     int module_blocked_pipe[2]; /* Pipe used to awake the event loop if a
                                    client blocked on a module command needs
@@ -1699,6 +1704,7 @@ int zslLexValueLteMax(sds value, zlexrangespec *spec);
 int getMaxmemoryState(size_t *total, size_t *logical, size_t *tofree, float *level);
 size_t freeMemoryGetNotCountedMemory();
 int freeMemoryIfNeeded(void);
+int freeMemoryIfNeededAndSafe(void);
 int processCommand(client *c);
 void setupSignalHandlers(void);
 struct redisCommand *lookupCommand(sds name);

src/setproctitle.c

@@ -39,7 +39,7 @@
 #include <errno.h>	/* errno program_invocation_name program_invocation_short_name */
 
 #if !defined(HAVE_SETPROCTITLE)
-#if (defined __NetBSD__ || defined __FreeBSD__ || defined __OpenBSD__)
+#if (defined __NetBSD__ || defined __FreeBSD__ || defined __OpenBSD__ || defined __DragonFly__)
 #define HAVE_SETPROCTITLE 1
 #else
 #define HAVE_SETPROCTITLE 0

src/t_stream.c

@@ -37,7 +37,7 @@
  * mark the entry as deleted, or having the same field as the "master"
  * entry at the start of the listpack> */
 #define STREAM_ITEM_FLAG_NONE 0             /* No special flags. */
-#define STREAM_ITEM_FLAG_DELETED (1<<0)     /* Entry is delted. Skip it. */
+#define STREAM_ITEM_FLAG_DELETED (1<<0)     /* Entry is deleted. Skip it. */
 #define STREAM_ITEM_FLAG_SAMEFIELDS (1<<1)  /* Same fields as master entry. */
 
 void streamFreeCG(streamCG *cg);
@@ -165,7 +165,7 @@ int streamCompareID(streamID *a, streamID *b) {
  * Returns the new entry ID populating the 'added_id' structure.
  *
  * If 'use_id' is not NULL, the ID is not auto-generated by the function,
- * but instead the passed ID is uesd to add the new entry. In this case
+ * but instead the passed ID is used to add the new entry. In this case
  * adding the entry may fail as specified later in this comment.
  *
  * The function returns C_OK if the item was added, this is always true
@@ -223,13 +223,13 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
      *
      * count and deleted just represent respectively the total number of
      * entries inside the listpack that are valid, and marked as deleted
-     * (delted flag in the entry flags set). So the total number of items
+     * (deleted flag in the entry flags set). So the total number of items
      * actually inside the listpack (both deleted and not) is count+deleted.
      *
      * The real entries will be encoded with an ID that is just the
      * millisecond and sequence difference compared to the key stored at
      * the radix tree node containing the listpack (delta encoding), and
-     * if the fields of the entry are the same as the master enty fields, the
+     * if the fields of the entry are the same as the master entry fields, the
      * entry flags will specify this fact and the entry fields and number
      * of fields will be omitted (see later in the code of this function).
      *
@@ -486,7 +486,7 @@ int64_t streamTrimByLength(stream *s, size_t maxlen, int approx) {
  *  }
  *  streamIteratorStop(&myiterator); */
 void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev) {
-    /* Intialize the iterator and translates the iteration start/stop
+    /* Initialize the iterator and translates the iteration start/stop
      * elements into a 128 big big-endian number. */
     if (start) {
         streamEncodeID(si->start_key,start);
@@ -564,7 +564,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
                 si->lp_ele = lpLast(si->lp);
             }
         } else if (si->rev) {
-            /* If we are itereating in the reverse order, and this is not
+            /* If we are iterating in the reverse order, and this is not
              * the first entry emitted for this listpack, then we already
              * emitted the current entry, and have to go back to the previous
              * one. */
@@ -751,7 +751,7 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
 }
 
 /* Stop the stream iterator. The only cleanup we need is to free the rax
- * itereator, since the stream iterator itself is supposed to be stack
+ * iterator, since the stream iterator itself is supposed to be stack
  * allocated. */
 void streamIteratorStop(streamIterator *si) {
     raxStop(&si->ri);
@@ -847,11 +847,15 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
     decrRefCount(argv[4]);
 }
 
-/* Send the specified range to the client 'c'. The range the client will
- * receive is between start and end inclusive, if 'count' is non zero, no more
- * than 'count' elemnets are sent. The 'end' pointer can be NULL to mean that
- * we want all the elements from 'start' till the end of the stream. If 'rev'
- * is non zero, elements are produced in reversed order from end to start.
+/* Send the stream items in the specified range to the client 'c'. The range
+ * the client will receive is between start and end inclusive, if 'count' is
+ * non zero, no more than 'count' elements are sent.
+ *
+ * The 'end' pointer can be NULL to mean that we want all the elements from
+ * 'start' till the end of the stream. If 'rev' is non zero, elements are
+ * produced in reversed order from end to start.
+ *
+ * The function returns the number of entries emitted.
  *
  * If group and consumer are not NULL, the function performs additional work:
  * 1. It updates the last delivered ID in the group in case we are
@@ -863,15 +867,15 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
  *
  * The behavior may be modified passing non-zero flags:
  *
- * STREAM_RWR_NOACK: Do not craete PEL entries, that is, the point "3" above
+ * STREAM_RWR_NOACK: Do not create PEL entries, that is, the point "3" above
  *                   is not performed.
  * STREAM_RWR_RAWENTRIES: Do not emit array boundaries, but just the entries,
  *                        and return the number of entries emitted as usually.
  *                        This is used when the function is just used in order
  *                        to emit data and there is some higher level logic.
  *
- * The final argument 'spi' (stream propagatino info pointer) is a structure
- * filled with information needed to propagte the command execution to AOF
+ * The final argument 'spi' (stream propagation info pointer) is a structure
+ * filled with information needed to propagate the command execution to AOF
  * and slaves, in the case a consumer group was passed: we need to generate
  * XCLAIM commands to create the pending list into AOF/slaves in that case.
  *
@@ -890,6 +894,7 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
 #define STREAM_RWR_NOACK (1<<0)         /* Do not create entries in the PEL. */
 #define STREAM_RWR_RAWENTRIES (1<<1)    /* Do not emit protocol for array
                                            boundaries, just the entries. */
+#define STREAM_RWR_HISTORY (1<<2)       /* Only serve consumer local PEL. */
 size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count, int rev, streamCG *group, streamConsumer *consumer, int flags, streamPropInfo *spi) {
     void *arraylen_ptr = NULL;
     size_t arraylen = 0;
@@ -898,15 +903,12 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
     streamID id;
     int propagate_last_id = 0;
 
-    /* If a group was passed, we check if the request is about messages
-     * never delivered so far (normally this happens when ">" ID is passed).
-     *
-     * If instead the client is asking for some history, we serve it
-     * using a different function, so that we return entries *solely*
-     * from its own PEL. This ensures each consumer will always and only
-     * see the history of messages delivered to it and not yet confirmed
+    /* If the client is asking for some history, we serve it using a
+     * different function, so that we return entries *solely* from its
+     * own PEL. This ensures each consumer will always and only see
+     * the history of messages delivered to it and not yet confirmed
      * as delivered. */
-    if (group && streamCompareID(start,&group->last_id) <= 0) {
+    if (group && (flags & STREAM_RWR_HISTORY)) {
         return streamReplyWithRangeFromConsumerPEL(c,s,start,end,count,
                                                    consumer);
     }
@@ -1023,7 +1025,7 @@ size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start
         if (end && memcmp(ri.key,end,ri.key_len) > 0) break;
         streamID thisid;
         streamDecodeID(ri.key,&thisid);
-        if (streamReplyWithRange(c,s,&thisid,NULL,1,0,NULL,NULL,
+        if (streamReplyWithRange(c,s,&thisid,&thisid,1,0,NULL,NULL,
                                  STREAM_RWR_RAWENTRIES,NULL) == 0)
         {
             /* Note that we may have a not acknowledged entry in the PEL
@@ -1136,7 +1138,7 @@ invalid:
 }
 
 /* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
- * 0, to be used when - and + are accetable IDs. */
+ * 0, to be used when - and + are acceptable IDs. */
 int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
     return streamGenericParseIDOrReply(c,o,id,missing_seq,0);
 }
@@ -1260,7 +1262,7 @@ void xrangeGenericCommand(client *c, int rev) {
     robj *o;
     stream *s;
     streamID startid, endid;
-    long long count = 0;
+    long long count = -1;
     robj *startarg = rev ? c->argv[3] : c->argv[2];
     robj *endarg = rev ? c->argv[2] : c->argv[3];
 
@@ -1287,7 +1289,13 @@ void xrangeGenericCommand(client *c, int rev) {
     if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
         || checkType(c,o,OBJ_STREAM)) return;
     s = o->ptr;
-    streamReplyWithRange(c,s,&startid,&endid,count,rev,NULL,NULL,0,NULL);
+
+    if (count == 0) {
+        addReply(c,shared.nullmultibulk);
+    } else {
+        if (count == -1) count = 0;
+        streamReplyWithRange(c,s,&startid,&endid,count,rev,NULL,NULL,0,NULL);
+    }
 }
 
 /* XRANGE key start end [COUNT <n>] */
@@ -1464,8 +1472,10 @@ void xreadCommand(client *c) {
         stream *s = o->ptr;
         streamID *gt = ids+i; /* ID must be greater than this. */
         int serve_synchronously = 0;
+        int serve_history = 0; /* True for XREADGROUP with ID != ">". */
 
-        /* Check if there are the conditions to serve the client synchronously. */
+        /* Check if there are the conditions to serve the client
+         * synchronously. */
         if (groups) {
             /* If the consumer is blocked on a group, we always serve it
              * synchronously (serving its local history) if the ID specified
@@ -1474,6 +1484,7 @@ void xreadCommand(client *c) {
                 gt->seq != UINT64_MAX)
             {
                 serve_synchronously = 1;
+                serve_history = 1;
             } else {
                 /* We also want to serve a consumer in a consumer group
                  * synchronously in case the group top item delivered is smaller
@@ -1509,9 +1520,12 @@ void xreadCommand(client *c) {
             if (groups) consumer = streamLookupConsumer(groups[i],
                                                         consumername->ptr,1);
             streamPropInfo spi = {c->argv[i+streams_arg],groupname};
+            int flags = 0;
+            if (noack) flags |= STREAM_RWR_NOACK;
+            if (serve_history) flags |= STREAM_RWR_HISTORY;
             streamReplyWithRange(c,s,&start,NULL,count,0,
                                  groups ? groups[i] : NULL,
-                                 consumer, noack, &spi);
+                                 consumer, flags, &spi);
             if (groups) server.dirty++;
         }
     }
@@ -1815,7 +1829,9 @@ NULL
     }
 }
 
-/* Set the internal "last ID" of a stream. */
+/* XSETID <stream> <groupname> <id>
+ *
+ * Set the internal "last ID" of a stream. */
 void xsetidCommand(client *c) {
     robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr);
     if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
@@ -1892,7 +1908,7 @@ void xackCommand(client *c) {
     addReplyLongLong(c,acknowledged);
 }
 
-/* XPENDING <key> <group> [<start> <stop> <count>] [<consumer>]
+/* XPENDING <key> <group> [<start> <stop> <count> [<consumer>]]
  *
  * If start and stop are omitted, the command just outputs information about
  * the amount of pending messages for the key/group pair, together with
@@ -1921,6 +1937,7 @@ void xpendingCommand(client *c) {
     if (c->argc >= 6) {
         if (getLongLongFromObjectOrReply(c,c->argv[5],&count,NULL) == C_ERR)
             return;
+        if (count < 0) count = 0;
         if (streamParseIDOrReply(c,c->argv[3],&startid,0) == C_ERR)
             return;
         if (streamParseIDOrReply(c,c->argv[4],&endid,UINT64_MAX) == C_ERR)
@@ -2146,7 +2163,7 @@ void xclaimCommand(client *c) {
 
     /* If we stopped because some IDs cannot be parsed, perhaps they
      * are trailing options. */
-    time_t now = mstime();
+    mstime_t now = mstime();
     streamID last_id = {0,0};
     int propagate_last_id = 0;
     for (; j < c->argc; j++) {
@@ -2265,8 +2282,9 @@ void xclaimCommand(client *c) {
             if (justid) {
                 addReplyStreamID(c,&id);
             } else {
-                streamReplyWithRange(c,o->ptr,&id,NULL,1,0,NULL,NULL,
-                                     STREAM_RWR_RAWENTRIES,NULL);
+                size_t emitted = streamReplyWithRange(c,o->ptr,&id,&id,1,0,
+                                    NULL,NULL,STREAM_RWR_RAWENTRIES,NULL);
+                if (!emitted) addReply(c,shared.nullbulk);
             }
             arraylen++;
 

src/t_string.c

@@ -301,24 +301,22 @@ void mgetCommand(client *c) {
 }
 
 void msetGenericCommand(client *c, int nx) {
-    int j, busykeys = 0;
+    int j;
 
     if ((c->argc % 2) == 0) {
         addReplyError(c,"wrong number of arguments for MSET");
         return;
     }
+
     /* Handle the NX flag. The MSETNX semantic is to return zero and don't
-     * set nothing at all if at least one already key exists. */
+     * set anything if at least one key alerady exists. */
     if (nx) {
         for (j = 1; j < c->argc; j += 2) {
             if (lookupKeyWrite(c->db,c->argv[j]) != NULL) {
-                busykeys++;
+                addReply(c, shared.czero);
+                return;
             }
         }
-        if (busykeys) {
-            addReply(c, shared.czero);
-            return;
-        }
     }
 
     for (j = 1; j < c->argc; j += 2) {

src/t_zset.c

@@ -574,12 +574,12 @@ int zslParseLexRangeItem(robj *item, sds *dest, int *ex) {
     switch(c[0]) {
     case '+':
         if (c[1] != '\0') return C_ERR;
-        *ex = 0;
+        *ex = 1;
         *dest = shared.maxstring;
         return C_OK;
     case '-':
         if (c[1] != '\0') return C_ERR;
-        *ex = 0;
+        *ex = 1;
         *dest = shared.minstring;
         return C_OK;
     case '(':
@@ -652,9 +652,8 @@ int zslIsInLexRange(zskiplist *zsl, zlexrangespec *range) {
     zskiplistNode *x;
 
     /* Test for ranges that will always be empty. */
-    if (sdscmplex(range->min,range->max) > 1 ||
-            (sdscmp(range->min,range->max) == 0 &&
-            (range->minex || range->maxex)))
+    int cmp = sdscmplex(range->min,range->max);
+    if (cmp > 0 || (cmp == 0 && (range->minex || range->maxex)))
         return 0;
     x = zsl->tail;
     if (x == NULL || !zslLexValueGteMin(x->ele,range))
@@ -927,9 +926,8 @@ int zzlIsInLexRange(unsigned char *zl, zlexrangespec *range) {
     unsigned char *p;
 
     /* Test for ranges that will always be empty. */
-    if (sdscmplex(range->min,range->max) > 1 ||
-            (sdscmp(range->min,range->max) == 0 &&
-            (range->minex || range->maxex)))
+    int cmp = sdscmplex(range->min,range->max);
+    if (cmp > 0 || (cmp == 0 && (range->minex || range->maxex)))
         return 0;
 
     p = ziplistIndex(zl,-2); /* Last element. */

src/util.c

@@ -39,6 +39,7 @@
 #include <float.h>
 #include <stdint.h>
 #include <errno.h>
+#include <time.h>
 
 #include "util.h"
 #include "sha1.h"
@@ -47,7 +48,7 @@
 int stringmatchlen(const char *pattern, int patternLen,
         const char *string, int stringLen, int nocase)
 {
-    while(patternLen) {
+    while(patternLen && stringLen) {
         switch(pattern[0]) {
         case '*':
             while (pattern[1] == '*') {
@@ -170,6 +171,22 @@ int stringmatch(const char *pattern, const char *string, int nocase) {
     return stringmatchlen(pattern,strlen(pattern),string,strlen(string),nocase);
 }
 
+/* Fuzz stringmatchlen() trying to crash it with bad input. */
+int stringmatchlen_fuzz_test(void) {
+    char str[32];
+    char pat[32];
+    int cycles = 10000000;
+    int total_matches = 0;
+    while(cycles--) {
+        int strlen = rand() % sizeof(str);
+        int patlen = rand() % sizeof(pat);
+        for (int j = 0; j < strlen; j++) str[j] = rand() % 128;
+        for (int j = 0; j < patlen; j++) pat[j] = rand() % 128;
+        total_matches += stringmatchlen(pat, patlen, str, strlen, 0);
+    }
+    return total_matches;
+}
+
 /* Convert a string representing an amount of memory into the number of
  * bytes, so for instance memtoll("1Gb") will return 1073741824 that is
  * (1024*1024*1024).
@@ -605,7 +622,7 @@ void getRandomHexChars(char *p, size_t len) {
  * already, this will be detected and handled correctly.
  *
  * The function does not try to normalize everything, but only the obvious
- * case of one or more "../" appearning at the start of "filename"
+ * case of one or more "../" appearing at the start of "filename"
  * relative path. */
 sds getAbsolutePath(char *filename) {
     char cwd[1024];
@@ -652,6 +669,24 @@ sds getAbsolutePath(char *filename) {
     return abspath;
 }
 
+/*
+ * Gets the proper timezone in a more portable fashion
+ * i.e timezone variables are linux specific.
+ */
+
+unsigned long getTimeZone(void) {
+#ifdef __linux__
+    return timezone;
+#else
+    struct timeval tv;
+    struct timezone tz;
+
+    gettimeofday(&tv, &tz);
+
+    return tz.tz_minuteswest * 60UL;
+#endif
+}
+
 /* Return true if the specified path is just a file basename without any
  * relative or absolute path. This function just checks that no / or \
  * character exists inside the specified path, that's enough in the

src/util.h

@@ -40,6 +40,7 @@
 
 int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase);
 int stringmatch(const char *p, const char *s, int nocase);
+int stringmatchlen_fuzz_test(void);
 long long memtoll(const char *p, int *err);
 uint32_t digits10(uint64_t v);
 uint32_t sdigits10(int64_t v);
@@ -50,6 +51,7 @@ int string2ld(const char *s, size_t slen, long double *dp);
 int d2string(char *buf, size_t len, double value);
 int ld2string(char *buf, size_t len, long double value, int humanfriendly);
 sds getAbsolutePath(char *filename);
+unsigned long getTimeZone(void);
 int pathIsBaseName(char *path);
 
 #ifdef REDIS_TEST

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "5.0.0"
+#define REDIS_VERSION "999.999.999"

tests/integration/replication.tcl

@@ -275,7 +275,6 @@ start_server {tags {"repl"}} {
     start_server {} {
         test "Master stream is correctly processed while the replica has a script in -BUSY state" {
             set slave [srv 0 client]
-            puts [srv 0 port]
             $slave config set lua-time-limit 500
             $slave slaveof $master_host $master_port
 

tests/support/server.tcl

@@ -4,7 +4,7 @@ set ::valgrind_errors {}
 
 proc start_server_error {config_file error} {
     set err {}
-    append err "Cant' start the Redis server\n"
+    append err "Can't start the Redis server\n"
     append err "CONFIGURATION:"
     append err [exec cat $config_file]
     append err "\nERROR:"

tests/support/util.tcl

@@ -91,6 +91,14 @@ proc wait_for_sync r {
     }
 }
 
+proc wait_for_ofs_sync {r1 r2} {
+    wait_for_condition 50 100 {
+        [status $r1 master_repl_offset] eq [status $r2 master_repl_offset]
+    } else {
+        fail "replica didn't sync in time"
+    }
+}
+
 # Random integer between 0 and max (excluded).
 proc randomInt {max} {
     expr {int(rand()*$max)}

tests/unit/memefficiency.tcl

@@ -90,6 +90,7 @@ start_server {tags {"defrag"}} {
         test "Active defrag big keys" {
             r flushdb
             r config resetstat
+            r config set save "" ;# prevent bgsave from interfereing with save below
             r config set activedefrag no
             r config set active-defrag-max-scan-fields 1000
             r config set active-defrag-threshold-lower 5

tests/unit/type/stream-cgroups.tcl

@@ -108,6 +108,93 @@ start_server {
         assert {$c == 5}
     }
 
+    test {XREADGROUP will not report data on empty history. Bug #5577} {
+        r del events
+        r xadd events * a 1
+        r xadd events * b 2
+        r xadd events * c 3
+        r xgroup create events mygroup 0
+
+        # Current local PEL should be empty
+        set res [r xpending events mygroup - + 10]
+        assert {[llength $res] == 0}
+
+        # So XREADGROUP should read an empty history as well
+        set res [r xreadgroup group mygroup myconsumer count 3 streams events 0]
+        assert {[llength [lindex $res 0 1]] == 0}
+
+        # We should fetch all the elements in the stream asking for >
+        set res [r xreadgroup group mygroup myconsumer count 3 streams events >]
+        assert {[llength [lindex $res 0 1]] == 3}
+
+        # Now the history is populated with three not acked entries
+        set res [r xreadgroup group mygroup myconsumer count 3 streams events 0]
+        assert {[llength [lindex $res 0 1]] == 3}
+    }
+
+    test {XREADGROUP history reporting of deleted entries. Bug #5570} {
+        r del mystream
+        r XGROUP CREATE mystream mygroup $ MKSTREAM
+        r XADD mystream 1 field1 A
+        r XREADGROUP GROUP mygroup myconsumer STREAMS mystream >
+        r XADD mystream MAXLEN 1 2 field1 B
+        r XREADGROUP GROUP mygroup myconsumer STREAMS mystream >
+
+        # Now we have two pending entries, however one should be deleted
+        # and one should be ok (we should only see "B")
+        set res [r XREADGROUP GROUP mygroup myconsumer STREAMS mystream 0-1]
+        assert {[lindex $res 0 1 0] == {1-0 {}}}
+        assert {[lindex $res 0 1 1] == {2-0 {field1 B}}}
+    }
+
+    test {XCLAIM can claim PEL items from another consumer} {
+        # Add 3 items into the stream, and create a consumer group
+        r del mystream
+        set id1 [r XADD mystream * a 1]
+        set id2 [r XADD mystream * b 2]
+        set id3 [r XADD mystream * c 3]
+        r XGROUP CREATE mystream mygroup 0
+
+        # Client 1 reads item 1 from the stream without acknowledgements.
+        # Client 2 then claims pending item 1 from the PEL of client 1
+        set reply [
+            r XREADGROUP GROUP mygroup client1 count 1 STREAMS mystream >
+        ]
+        assert {[llength [lindex $reply 0 1 0 1]] == 2}
+        assert {[lindex $reply 0 1 0 1] eq {a 1}}
+        r debug sleep 0.2
+        set reply [
+            r XCLAIM mystream mygroup client2 10 $id1
+        ]
+        assert {[llength [lindex $reply 0 1]] == 2}
+        assert {[lindex $reply 0 1] eq {a 1}}
+
+        # Client 1 reads another 2 items from stream
+        r XREADGROUP GROUP mygroup client1 count 2 STREAMS mystream >
+        r debug sleep 0.2
+
+        # Delete item 2 from the stream. Now client 1 has PEL that contains
+        # only item 3. Try to use client 2 to claim the deleted item 2
+        # from the PEL of client 1, this should return nil
+        r XDEL mystream $id2
+        set reply [
+            r XCLAIM mystream mygroup client2 10 $id2
+        ]
+        assert {[llength $reply] == 1}
+        assert_equal "" [lindex $reply 0]
+
+        # Delete item 3 from the stream. Now client 1 has PEL that is empty.
+        # Try to use client 2 to claim the deleted item 3 from the PEL
+        # of client 1, this should return nil
+        r debug sleep 0.2
+        r XDEL mystream $id3
+        set reply [
+            r XCLAIM mystream mygroup client2 10 $id3
+        ]
+        assert {[llength $reply] == 1}
+        assert_equal "" [lindex $reply 0]
+    }
+
     start_server {} {
         set master [srv -1 client]
         set master_host [srv -1 host]
@@ -144,6 +231,8 @@ start_server {
                     }
                 }
 
+                wait_for_ofs_sync $master $slave
+
                 # Turn slave into master
                 $slave slaveof no one
 

tests/unit/type/zset.tcl

@@ -388,7 +388,7 @@ start_server {tags {"zset"}} {
                               0 omega}
         }
 
-        test "ZRANGEBYLEX/ZREVRANGEBYLEX/ZCOUNT basics" {
+        test "ZRANGEBYLEX/ZREVRANGEBYLEX/ZLEXCOUNT basics" {
             create_default_lex_zset
 
             # inclusive range
@@ -416,6 +416,22 @@ start_server {tags {"zset"}} {
             assert_equal {} [r zrevrangebylex zset \[elez \[elex]
             assert_equal {} [r zrevrangebylex zset (hill (omega]
         }
+        
+        test "ZLEXCOUNT advanced" {
+            create_default_lex_zset
+    
+            assert_equal 9 [r zlexcount zset - +]
+            assert_equal 0 [r zlexcount zset + -]
+            assert_equal 0 [r zlexcount zset + \[c]
+            assert_equal 0 [r zlexcount zset \[c -]
+            assert_equal 8 [r zlexcount zset \[bar +]
+            assert_equal 5 [r zlexcount zset \[bar \[foo]
+            assert_equal 4 [r zlexcount zset \[bar (foo]
+            assert_equal 4 [r zlexcount zset (bar \[foo]
+            assert_equal 3 [r zlexcount zset (bar (foo]
+            assert_equal 5 [r zlexcount zset - (foo]
+            assert_equal 1 [r zlexcount zset (maxstring +]
+        }
 
         test "ZRANGEBYSLEX with LIMIT" {
             create_default_lex_zset