Merge branch 'master' into intset-split

Conflicts:
	src/Makefile
	src/t_set.c

BETATESTING.txt

@@ -1,9 +0,0 @@
-This is a stable release, for beta testing make sure to download the latest source code from Git:
-
-    git clone git://github.com/antirez/redis.git
-
-It's also possibe to download the latest source code as a tarball:
-
-    http://github.com/antirez/redis/tree/master
-
-(use the download button)

INSTALL

@@ -0,0 +1,19 @@
+To compile Redis, do the following:
+
+    cd src; make
+
+The compilation will produce a redis-server binary.
+Copy this file where you want.
+
+Run the server using the following command line:
+
+    /path/to/redis-server
+
+This will start a Redis server with the default configuration.
+
+Otherwise if you want to provide your configuration use:
+
+    /path/to/redis-server /path/to/redis.conf
+
+You can find an example redis.conf file in the root directory
+of this source distribution.

Makefile

@@ -0,0 +1,14 @@
+# Top level makefile, the real shit is at src/Makefile
+
+TARGETS=32bit noopt test
+
+all:
+	cd src && $(MAKE) $@
+
+install: dummy
+	cd src && $(MAKE) $@
+
+$(TARGETS) clean:
+	cd src && $(MAKE) $@
+
+dummy:

TODO

@@ -6,6 +6,8 @@ VERSION 2.2 TODO (Optimizations and latency)
 
 * Support for syslog(3).
 * Change the implementation of ZCOUNT to use the augmented skiplist in order to be much faster.
+* Add an explicit test for MULTI/EXEC reloaded in the AOF.
+* Command table -> hash table, with support for command renaming
 
 VM TODO
 =======
@@ -56,3 +58,9 @@ KNOWN BUGS
 ==========
 
 * LRANGE and other commands are using 32 bit integers for ranges, and overflows are not detected. So LRANGE mylist 0 23498204823094823904823904 will have random effects.
+
+REDIS CLI TODO
+==============
+
+* Computer parsable output generation
+* Memoize return values so that they can be used later as arguments, like $1

src/Makefile

@@ -15,6 +15,10 @@ endif
 CCOPT= $(CFLAGS) $(CCLINK) $(ARCH) $(PROF)
 DEBUG?= -g -rdynamic -ggdb 
 
+INSTALL_TOP= /usr/local
+INSTALL_BIN= $(INSTALL_TOP)/bin
+INSTALL= cp -p
+
 OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o vm.o pubsub.o multi.o debug.o sort.o intset.o
 BENCHOBJ = ae.o anet.o redis-benchmark.o sds.o adlist.o zmalloc.o
 CLIOBJ = anet.o sds.o adlist.o redis-cli.o zmalloc.o linenoise.o
@@ -110,3 +114,10 @@ noopt:
 
 32bitgprof:
 	make PROF="-pg" ARCH="-arch i386"
+
+install: all
+	$(INSTALL) $(PRGNAME) $(INSTALL_BIN)
+	$(INSTALL) $(BENCHPRGNAME) $(INSTALL_BIN)
+	$(INSTALL) $(CLIPRGNAME) $(INSTALL_BIN)
+	$(INSTALL) $(CHECKDUMPPRGNAME) $(INSTALL_BIN)
+	$(INSTALL) $(CHECKAOFPRGNAME) $(INSTALL_BIN)

src/aof.c

@@ -194,6 +194,7 @@ struct redisClient *createFakeClient(void) {
      * so that Redis will not try to send replies to this client. */
     c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
     c->reply = listCreate();
+    c->watched_keys = listCreate();
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);
     initClientMultiState(c);
@@ -203,6 +204,7 @@ struct redisClient *createFakeClient(void) {
 void freeFakeClient(struct redisClient *c) {
     sdsfree(c->querybuf);
     listRelease(c->reply);
+    listRelease(c->watched_keys);
     freeClientMultiState(c);
     zfree(c);
 }

src/db.c

@@ -45,8 +45,7 @@ robj *lookupKeyRead(redisDb *db, robj *key) {
 }
 
 robj *lookupKeyWrite(redisDb *db, robj *key) {
-    deleteIfVolatile(db,key);
+    expireIfNeeded(db,key);
-    touchWatchedKey(db,key);
     return lookupKey(db,key);
 }
 
@@ -322,7 +321,6 @@ void renameGenericCommand(redisClient *c, int nx) {
         return;
 
     incrRefCount(o);
-    deleteIfVolatile(c->db,c->argv[2]);
     if (dbAdd(c->db,c->argv[2],o) == REDIS_ERR) {
         if (nx) {
             decrRefCount(o);
@@ -332,6 +330,7 @@ void renameGenericCommand(redisClient *c, int nx) {
         dbReplace(c->db,c->argv[2],o);
     }
     dbDelete(c->db,c->argv[1]);
+    touchWatchedKey(c->db,c->argv[1]);
     touchWatchedKey(c->db,c->argv[2]);
     server.dirty++;
     addReply(c,nx ? shared.cone : shared.ok);
@@ -375,7 +374,6 @@ void moveCommand(redisClient *c) {
     }
 
     /* Try to add the element to the target DB */
-    deleteIfVolatile(dst,c->argv[1]);
     if (dbAdd(dst,c->argv[1],o) == REDIS_ERR) {
         addReply(c,shared.czero);
         return;
@@ -396,23 +394,16 @@ int removeExpire(redisDb *db, robj *key) {
     /* An expire may only be removed if there is a corresponding entry in the
      * main dict. Otherwise, the key will never be freed. */
     redisAssert(dictFind(db->dict,key->ptr) != NULL);
-    if (dictDelete(db->expires,key->ptr) == DICT_OK) {
+    return dictDelete(db->expires,key->ptr) == DICT_OK;
-        return 1;
-    } else {
-        return 0;
-    }
 }
 
-int setExpire(redisDb *db, robj *key, time_t when) {
+void setExpire(redisDb *db, robj *key, time_t when) {
     dictEntry *de;
 
     /* Reuse the sds from the main dict in the expire dict */
-    redisAssert((de = dictFind(db->dict,key->ptr)) != NULL);
+    de = dictFind(db->dict,key->ptr);
-    if (dictAdd(db->expires,dictGetEntryKey(de),(void*)when) == DICT_ERR) {
+    redisAssert(de != NULL);
-        return 0;
+    dictReplace(db->expires,dictGetEntryKey(de),(void*)when);
-    } else {
-        return 1;
-    }
 }
 
 /* Return the expire time of the specified key, or -1 if no expire
@@ -430,8 +421,46 @@ time_t getExpire(redisDb *db, robj *key) {
     return (time_t) dictGetEntryVal(de);
 }
 
+/* Propagate expires into slaves and the AOF file.
+ * When a key expires in the master, a DEL operation for this key is sent
+ * to all the slaves and the AOF file if enabled.
+ *
+ * This way the key expiry is centralized in one place, and since both
+ * AOF and the master->slave link guarantee operation ordering, everything
+ * will be consistent even if we allow write operations against expiring
+ * keys. */
+void propagateExpire(redisDb *db, robj *key) {
+    struct redisCommand *cmd;
+    robj *argv[2];
+
+    cmd = lookupCommand("del");
+    argv[0] = createStringObject("DEL",3);
+    argv[1] = key;
+    incrRefCount(key);
+
+    if (server.appendonly)
+        feedAppendOnlyFile(cmd,db->id,argv,2);
+    if (listLength(server.slaves))
+        replicationFeedSlaves(server.slaves,db->id,argv,2);
+
+    decrRefCount(argv[0]);
+    decrRefCount(argv[1]);
+}
+
 int expireIfNeeded(redisDb *db, robj *key) {
     time_t when = getExpire(db,key);
+
+    /* If we are running in the context of a slave, 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 time(NULL) > when;
+    }
+
     if (when < 0) return 0;
 
     /* Return when this key has not expired */
@@ -440,15 +469,7 @@ int expireIfNeeded(redisDb *db, robj *key) {
     /* Delete the key */
     server.stat_expiredkeys++;
     server.dirty++;
-    return dbDelete(db,key);
+    propagateExpire(db,key);
-}
-
-int deleteIfVolatile(redisDb *db, robj *key) {
-    if (getExpire(db,key) < 0) return 0;
-
-    /* Delete the key */
-    server.stat_expiredkeys++;
-    server.dirty++;
     return dbDelete(db,key);
 }
 
@@ -472,15 +493,14 @@ void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
     if (seconds <= 0) {
         if (dbDelete(c->db,key)) server.dirty++;
         addReply(c, shared.cone);
+        touchWatchedKey(c->db,key);
         return;
     } else {
         time_t when = time(NULL)+seconds;
-        if (setExpire(c->db,key,when)) {
+        setExpire(c->db,key,when);
-            addReply(c,shared.cone);
+        addReply(c,shared.cone);
-            server.dirty++;
+        touchWatchedKey(c->db,key);
-        } else {
+        server.dirty++;
-            addReply(c,shared.czero);
-        }
         return;
     }
 }
@@ -505,4 +525,18 @@ void ttlCommand(redisClient *c) {
     addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",ttl));
 }
 
+void persistCommand(redisClient *c) {
+    dictEntry *de;
 
+    de = dictFind(c->db->dict,c->argv[1]->ptr);
+    if (de == NULL) {
+        addReply(c,shared.czero);
+    } else {
+        if (removeExpire(c->db,c->argv[1])) {
+            addReply(c,shared.cone);
+            server.dirty++;
+        } else {
+            addReply(c,shared.czero);
+        }
+    }
+}

src/dict.c

@@ -52,33 +52,6 @@
  * around when there is a child performing saving operations. */
 static int dict_can_resize = 1;
 
-/* ---------------------------- Utility funcitons --------------------------- */
-
-static void _dictPanic(const char *fmt, ...)
-{
-    va_list ap;
-
-    va_start(ap, fmt);
-    fprintf(stderr, "\nDICT LIBRARY PANIC: ");
-    vfprintf(stderr, fmt, ap);
-    fprintf(stderr, "\n\n");
-    va_end(ap);
-}
-
-/* ------------------------- Heap Management Wrappers------------------------ */
-
-static void *_dictAlloc(size_t size)
-{
-    void *p = zmalloc(size);
-    if (p == NULL)
-        _dictPanic("Out of memory");
-    return p;
-}
-
-static void _dictFree(void *ptr) {
-    zfree(ptr);
-}
-
 /* -------------------------- private prototypes ---------------------------- */
 
 static int _dictExpandIfNeeded(dict *ht);
@@ -132,7 +105,7 @@ static void _dictReset(dictht *ht)
 dict *dictCreate(dictType *type,
         void *privDataPtr)
 {
-    dict *d = _dictAlloc(sizeof(*d));
+    dict *d = zmalloc(sizeof(*d));
 
     _dictInit(d,type,privDataPtr);
     return d;
@@ -175,14 +148,12 @@ int dictExpand(dict *d, unsigned long size)
     if (dictIsRehashing(d) || d->ht[0].used > size)
         return DICT_ERR;
 
+    /* Allocate the new hashtable and initialize all pointers to NULL */
     n.size = realsize;
     n.sizemask = realsize-1;
-    n.table = _dictAlloc(realsize*sizeof(dictEntry*));
+    n.table = zcalloc(realsize*sizeof(dictEntry*));
     n.used = 0;
 
-    /* Initialize all the pointers to NULL */
-    memset(n.table, 0, realsize*sizeof(dictEntry*));
-
     /* Is this the first initialization? If so it's not really a rehashing
      * we just set the first hash table so that it can accept keys. */
     if (d->ht[0].table == NULL) {
@@ -208,7 +179,7 @@ int dictRehash(dict *d, int n) {
 
         /* Check if we already rehashed the whole table... */
         if (d->ht[0].used == 0) {
-            _dictFree(d->ht[0].table);
+            zfree(d->ht[0].table);
             d->ht[0] = d->ht[1];
             _dictReset(&d->ht[1]);
             d->rehashidx = -1;
@@ -285,7 +256,7 @@ int dictAdd(dict *d, void *key, void *val)
 
     /* Allocates the memory and stores key */
     ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0];
-    entry = _dictAlloc(sizeof(*entry));
+    entry = zmalloc(sizeof(*entry));
     entry->next = ht->table[index];
     ht->table[index] = entry;
     ht->used++;
@@ -348,7 +319,7 @@ static int dictGenericDelete(dict *d, const void *key, int nofree)
                     dictFreeEntryKey(d, he);
                     dictFreeEntryVal(d, he);
                 }
-                _dictFree(he);
+                zfree(he);
                 d->ht[table].used--;
                 return DICT_OK;
             }
@@ -382,13 +353,13 @@ int _dictClear(dict *d, dictht *ht)
             nextHe = he->next;
             dictFreeEntryKey(d, he);
             dictFreeEntryVal(d, he);
-            _dictFree(he);
+            zfree(he);
             ht->used--;
             he = nextHe;
         }
     }
     /* Free the table and the allocated cache structure */
-    _dictFree(ht->table);
+    zfree(ht->table);
     /* Re-initialize the table */
     _dictReset(ht);
     return DICT_OK; /* never fails */
@@ -399,7 +370,7 @@ void dictRelease(dict *d)
 {
     _dictClear(d,&d->ht[0]);
     _dictClear(d,&d->ht[1]);
-    _dictFree(d);
+    zfree(d);
 }
 
 dictEntry *dictFind(dict *d, const void *key)
@@ -432,7 +403,7 @@ void *dictFetchValue(dict *d, const void *key) {
 
 dictIterator *dictGetIterator(dict *d)
 {
-    dictIterator *iter = _dictAlloc(sizeof(*iter));
+    dictIterator *iter = zmalloc(sizeof(*iter));
 
     iter->d = d;
     iter->table = 0;
@@ -475,7 +446,7 @@ dictEntry *dictNext(dictIterator *iter)
 void dictReleaseIterator(dictIterator *iter)
 {
     if (!(iter->index == -1 && iter->table == 0)) iter->d->iterators--;
-    _dictFree(iter);
+    zfree(iter);
 }
 
 /* Return a random entry from the hash table. Useful to
@@ -644,6 +615,12 @@ void dictDisableResize(void) {
     dict_can_resize = 0;
 }
 
+#if 0
+
+/* The following are just example hash table types implementations.
+ * Not useful for Redis so they are commented out.
+ */
+
 /* ----------------------- StringCopy Hash Table Type ------------------------*/
 
 static unsigned int _dictStringCopyHTHashFunction(const void *key)
@@ -651,10 +628,10 @@ static unsigned int _dictStringCopyHTHashFunction(const void *key)
     return dictGenHashFunction(key, strlen(key));
 }
 
-static void *_dictStringCopyHTKeyDup(void *privdata, const void *key)
+static void *_dictStringDup(void *privdata, const void *key)
 {
     int len = strlen(key);
-    char *copy = _dictAlloc(len+1);
+    char *copy = zmalloc(len+1);
     DICT_NOTUSED(privdata);
 
     memcpy(copy, key, len);
@@ -662,17 +639,6 @@ static void *_dictStringCopyHTKeyDup(void *privdata, const void *key)
     return copy;
 }
 
-static void *_dictStringKeyValCopyHTValDup(void *privdata, const void *val)
-{
-    int len = strlen(val);
-    char *copy = _dictAlloc(len+1);
-    DICT_NOTUSED(privdata);
-
-    memcpy(copy, val, len);
-    copy[len] = '\0';
-    return copy;
-}
-
 static int _dictStringCopyHTKeyCompare(void *privdata, const void *key1,
         const void *key2)
 {
@@ -681,47 +647,41 @@ static int _dictStringCopyHTKeyCompare(void *privdata, const void *key1,
     return strcmp(key1, key2) == 0;
 }
 
-static void _dictStringCopyHTKeyDestructor(void *privdata, void *key)
+static void _dictStringDestructor(void *privdata, void *key)
 {
     DICT_NOTUSED(privdata);
 
-    _dictFree((void*)key); /* ATTENTION: const cast */
+    zfree(key);
-}
-
-static void _dictStringKeyValCopyHTValDestructor(void *privdata, void *val)
-{
-    DICT_NOTUSED(privdata);
-
-    _dictFree((void*)val); /* ATTENTION: const cast */
 }
 
 dictType dictTypeHeapStringCopyKey = {
-    _dictStringCopyHTHashFunction,        /* hash function */
+    _dictStringCopyHTHashFunction, /* hash function */
-    _dictStringCopyHTKeyDup,              /* key dup */
+    _dictStringDup,                /* key dup */
-    NULL,                               /* val dup */
+    NULL,                          /* val dup */
-    _dictStringCopyHTKeyCompare,          /* key compare */
+    _dictStringCopyHTKeyCompare,   /* key compare */
-    _dictStringCopyHTKeyDestructor,       /* key destructor */
+    _dictStringDestructor,         /* key destructor */
-    NULL                                /* val destructor */
+    NULL                           /* val destructor */
 };
 
 /* This is like StringCopy but does not auto-duplicate the key.
  * It's used for intepreter's shared strings. */
 dictType dictTypeHeapStrings = {
-    _dictStringCopyHTHashFunction,        /* hash function */
+    _dictStringCopyHTHashFunction, /* hash function */
-    NULL,                               /* key dup */
+    NULL,                          /* key dup */
-    NULL,                               /* val dup */
+    NULL,                          /* val dup */
-    _dictStringCopyHTKeyCompare,          /* key compare */
+    _dictStringCopyHTKeyCompare,   /* key compare */
-    _dictStringCopyHTKeyDestructor,       /* key destructor */
+    _dictStringDestructor,         /* key destructor */
-    NULL                                /* val destructor */
+    NULL                           /* val destructor */
 };
 
 /* This is like StringCopy but also automatically handle dynamic
  * allocated C strings as values. */
 dictType dictTypeHeapStringCopyKeyValue = {
-    _dictStringCopyHTHashFunction,        /* hash function */
+    _dictStringCopyHTHashFunction, /* hash function */
-    _dictStringCopyHTKeyDup,              /* key dup */
+    _dictStringDup,                /* key dup */
-    _dictStringKeyValCopyHTValDup,        /* val dup */
+    _dictStringDup,                /* val dup */
-    _dictStringCopyHTKeyCompare,          /* key compare */
+    _dictStringCopyHTKeyCompare,   /* key compare */
-    _dictStringCopyHTKeyDestructor,       /* key destructor */
+    _dictStringDestructor,         /* key destructor */
-    _dictStringKeyValCopyHTValDestructor, /* val destructor */
+    _dictStringDestructor,         /* val destructor */
 };
+#endif

src/linenoise.c

@@ -70,6 +70,7 @@
  */
 
 #include "fmacros.h"
+
 #include <termios.h>
 #include <unistd.h>
 #include <stdlib.h>
@@ -81,13 +82,14 @@
 #include <sys/ioctl.h>
 #include <unistd.h>
 
+#define LINENOISE_DEFAULT_HISTORY_MAX_LEN 100
 #define LINENOISE_MAX_LINE 4096
 static char *unsupported_term[] = {"dumb","cons25",NULL};
 
 static struct termios orig_termios; /* in order to restore at exit */
 static int rawmode = 0; /* for atexit() function to check if restore is needed*/
 static int atexit_registered = 0; /* register atexit just 1 time */
-static int history_max_len = 100;
+static int history_max_len = LINENOISE_DEFAULT_HISTORY_MAX_LEN;
 static int history_len = 0;
 char **history = NULL;
 
@@ -219,11 +221,10 @@ static int linenoisePrompt(int fd, char *buf, size_t buflen, const char *prompt)
         if (nread <= 0) return len;
         switch(c) {
         case 13:    /* enter */
-            history_len--;
-            return len;
         case 4:     /* ctrl-d */
             history_len--;
-            return (len == 0) ? -1 : (int)len;
+            free(history[history_len]);
+            return (len == 0 && c == 4) ? -1 : (int)len;
         case 3:     /* ctrl-c */
             errno = EAGAIN;
             return -1;
@@ -396,7 +397,7 @@ int linenoiseHistoryAdd(const char *line) {
     char *linecopy;
 
     if (history_max_len == 0) return 0;
-    if (history == 0) {
+    if (history == NULL) {
         history = malloc(sizeof(char*)*history_max_len);
         if (history == NULL) return 0;
         memset(history,0,(sizeof(char*)*history_max_len));
@@ -404,6 +405,7 @@ int linenoiseHistoryAdd(const char *line) {
     linecopy = strdup(line);
     if (!linecopy) return 0;
     if (history_len == history_max_len) {
+        free(history[0]);
         memmove(history,history+1,sizeof(char*)*(history_max_len-1));
         history_len--;
     }
@@ -431,3 +433,39 @@ int linenoiseHistorySetMaxLen(int len) {
         history_len = history_max_len;
     return 1;
 }
+
+/* Save the history in the specified file. On success 0 is returned
+ * otherwise -1 is returned. */
+int linenoiseHistorySave(char *filename) {
+    FILE *fp = fopen(filename,"w");
+    int j;
+    
+    if (fp == NULL) return -1;
+    for (j = 0; j < history_len; j++)
+        fprintf(fp,"%s\n",history[j]);
+    fclose(fp);
+    return 0;
+}
+
+/* Load the history from the specified file. If the file does not exist
+ * zero is returned and no operation is performed.
+ *
+ * If the file exists and the operation succeeded 0 is returned, otherwise
+ * on error -1 is returned. */
+int linenoiseHistoryLoad(char *filename) {
+    FILE *fp = fopen(filename,"r");
+    char buf[LINENOISE_MAX_LINE];
+    
+    if (fp == NULL) return -1;
+
+    while (fgets(buf,LINENOISE_MAX_LINE,fp) != NULL) {
+        char *p;
+        
+        p = strchr(buf,'\r');
+        if (!p) p = strchr(buf,'\n');
+        if (p) *p = '\0';
+        linenoiseHistoryAdd(buf);
+    }
+    fclose(fp);
+    return 0;
+}

src/linenoise.h

@@ -35,7 +35,9 @@
 #define __LINENOISE_H
 
 char *linenoise(const char *prompt);
-int linenoiseHistoryAdd(char *line);
+int linenoiseHistoryAdd(const char *line);
 int linenoiseHistorySetMaxLen(int len);
+int linenoiseHistorySave(char *filename);
+int linenoiseHistoryLoad(char *filename);
 
 #endif /* __LINENOISE_H */

src/networking.c

@@ -235,19 +235,24 @@ void freeClient(redisClient *c) {
     ln = listSearchKey(server.clients,c);
     redisAssert(ln != NULL);
     listDelNode(server.clients,ln);
-    /* Remove from the list of clients that are now ready to be restarted
+    /* Remove from the list of clients waiting for swapped keys, or ready
-     * after waiting for swapped keys */
+     * to be restarted, but not yet woken up again. */
-    if (c->flags & REDIS_IO_WAIT && listLength(c->io_keys) == 0) {
+    if (c->flags & REDIS_IO_WAIT) {
-        ln = listSearchKey(server.io_ready_clients,c);
+        redisAssert(server.vm_enabled);
-        if (ln) {
+        if (listLength(c->io_keys) == 0) {
+            ln = listSearchKey(server.io_ready_clients,c);
+
+            /* When this client is waiting to be woken up (REDIS_IO_WAIT),
+             * it should be present in the list io_ready_clients */
+            redisAssert(ln != NULL);
             listDelNode(server.io_ready_clients,ln);
-            server.vm_blocked_clients--;
+        } else {
+            while (listLength(c->io_keys)) {
+                ln = listFirst(c->io_keys);
+                dontWaitForSwappedKey(c,ln->value);
+            }
         }
-    }
+        server.vm_blocked_clients--;
-    /* Remove from the list of clients waiting for swapped keys */
-    while (server.vm_enabled && listLength(c->io_keys)) {
-        ln = listFirst(c->io_keys);
-        dontWaitForSwappedKey(c,ln->value);
     }
     listRelease(c->io_keys);
     /* Master/slave cleanup */

src/object.c

@@ -1,5 +1,6 @@
 #include "redis.h"
 #include <pthread.h>
+#include <math.h>
 
 robj *createObject(int type, void *ptr) {
     robj *o;
@@ -11,8 +12,7 @@ robj *createObject(int type, void *ptr) {
         listDelNode(server.objfreelist,head);
         if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
     } else {
-        if (server.vm_enabled)
+        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
-            pthread_mutex_unlock(&server.obj_freelist_mutex);
         o = zmalloc(sizeof(*o));
     }
     o->type = type;
@@ -36,7 +36,8 @@ robj *createStringObject(char *ptr, size_t len) {
 
 robj *createStringObjectFromLongLong(long long value) {
     robj *o;
-    if (value >= 0 && value < REDIS_SHARED_INTEGERS) {
+    if (value >= 0 && value < REDIS_SHARED_INTEGERS &&
+        pthread_equal(pthread_self(),server.mainthread)) {
         incrRefCount(shared.integers[value]);
         o = shared.integers[value];
     } else {
@@ -197,6 +198,7 @@ void decrRefCount(void *obj) {
         case REDIS_HASH: freeHashObject(o); break;
         default: redisPanic("Unknown object type"); break;
         }
+        o->ptr = NULL; /* defensive programming. We'll see NULL in traces. */
         if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
         if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
             !listAddNodeHead(server.objfreelist,o))
@@ -232,8 +234,15 @@ robj *tryObjectEncoding(robj *o) {
     /* Check if we can represent this string as a long integer */
     if (isStringRepresentableAsLong(s,&value) == REDIS_ERR) return o;
 
-    /* Ok, this object can be encoded */
+    /* Ok, this object can be encoded...
-    if (value >= 0 && value < REDIS_SHARED_INTEGERS) {
+     *
+     * Can I use a shared object? Only if the object is inside a given
+     * range and if this is the main thread, since when VM is enabled we
+     * have the constraint that I/O thread should only handle non-shared
+     * objects, in order to avoid race conditions (we don't have per-object
+     * locking). */
+    if (value >= 0 && value < REDIS_SHARED_INTEGERS &&
+        pthread_equal(pthread_self(),server.mainthread)) {
         decrRefCount(o);
         incrRefCount(shared.integers[value]);
         return shared.integers[value];
@@ -329,7 +338,7 @@ int getDoubleFromObject(robj *o, double *target) {
         redisAssert(o->type == REDIS_STRING);
         if (o->encoding == REDIS_ENCODING_RAW) {
             value = strtod(o->ptr, &eptr);
-            if (eptr[0] != '\0') return REDIS_ERR;
+            if (eptr[0] != '\0' || isnan(value)) return REDIS_ERR;
         } else if (o->encoding == REDIS_ENCODING_INT) {
             value = (long)o->ptr;
         } else {

src/redis-cli.c

@@ -29,6 +29,7 @@
  */
 
 #include "fmacros.h"
+#include "version.h"
 
 #include <stdio.h>
 #include <string.h>
@@ -60,6 +61,7 @@ static struct config {
     int pubsub_mode;
     int raw_output;
     char *auth;
+    char *historyfile;
 } config;
 
 static int cliReadReply(int fd);
@@ -315,6 +317,9 @@ static int parseOptions(int argc, char **argv) {
             config.interactive = 1;
         } else if (!strcmp(argv[i],"-c")) {
             config.argn_from_stdin = 1;
+        } else if (!strcmp(argv[i],"-v")) {
+            printf("redis-cli shipped with Redis verison %s\n", REDIS_VERSION);
+            exit(0);
         } else {
             break;
         }
@@ -340,7 +345,7 @@ static sds readArgFromStdin(void) {
 }
 
 static void usage() {
-    fprintf(stderr, "usage: redis-cli [-h host] [-p port] [-a authpw] [-r repeat_times] [-n db_num] [-i] cmd arg1 arg2 arg3 ... argN\n");
+    fprintf(stderr, "usage: redis-cli [-iv] [-h host] [-p port] [-a authpw] [-r repeat_times] [-n db_num] cmd arg1 arg2 arg3 ... argN\n");
     fprintf(stderr, "usage: echo \"argN\" | redis-cli -c [-h host] [-p port] [-a authpw] [-r repeat_times] [-n db_num] cmd arg1 arg2 ... arg(N-1)\n");
     fprintf(stderr, "\nIf a pipe from standard input is detected this data is used as last argument.\n\n");
     fprintf(stderr, "example: cat /etc/passwd | redis-cli set my_passwd\n");
@@ -361,80 +366,17 @@ static char **convertToSds(int count, char** args) {
   return sds;
 }
 
-static char **splitArguments(char *line, int *argc) {
-    char *p = line;
-    char *current = NULL;
-    char **vector = NULL;
-
-    *argc = 0;
-    while(1) {
-        /* skip blanks */
-        while(*p && isspace(*p)) p++;
-        if (*p) {
-            /* get a token */
-            int inq=0; /* set to 1 if we are in "quotes" */
-            int done = 0;
-
-            if (current == NULL) current = sdsempty();
-            while(!done) {
-                if (inq) {
-                    if (*p == '\\' && *(p+1)) {
-                        char c;
-
-                        p++;
-                        switch(*p) {
-                        case 'n': c = '\n'; break;
-                        case 'r': c = '\r'; break;
-                        case 't': c = '\t'; break;
-                        case 'b': c = '\b'; break;
-                        case 'a': c = '\a'; break;
-                        default: c = *p; break;
-                        }
-                        current = sdscatlen(current,&c,1);
-                    } else if (*p == '"') {
-                        done = 1;
-                    } else {
-                        current = sdscatlen(current,p,1);
-                    }
-                } else {
-                    switch(*p) {
-                    case ' ':
-                    case '\n':
-                    case '\r':
-                    case '\t':
-                    case '\0':
-                        done=1;
-                        break;
-                    case '"':
-                        inq=1;
-                        break;
-                    default:
-                        current = sdscatlen(current,p,1);
-                        break;
-                    }
-                }
-                if (*p) p++;
-            }
-            /* add the token to the vector */
-            vector = zrealloc(vector,((*argc)+1)*sizeof(char*));
-            vector[*argc] = current;
-            (*argc)++;
-            current = NULL;
-        } else {
-            return vector;
-        }
-    }
-}
-
 #define LINE_BUFLEN 4096
 static void repl() {
     int argc, j;
-    char *line, **argv;
+    char *line;
+    sds *argv;
 
     while((line = linenoise("redis> ")) != NULL) {
         if (line[0] != '\0') {
-            argv = splitArguments(line,&argc);
+            argv = sdssplitargs(line,&argc);
             linenoiseHistoryAdd(line);
+            if (config.historyfile) linenoiseHistorySave(config.historyfile);
             if (argc > 0) {
                 if (strcasecmp(argv[0],"quit") == 0 ||
                     strcasecmp(argv[0],"exit") == 0)
@@ -468,6 +410,13 @@ int main(int argc, char **argv) {
     config.pubsub_mode = 0;
     config.raw_output = 0;
     config.auth = NULL;
+    config.historyfile = NULL;
+
+    if (getenv("HOME") != NULL) {
+        config.historyfile = malloc(256);
+        snprintf(config.historyfile,256,"%s/.rediscli_history",getenv("HOME"));
+        linenoiseHistoryLoad(config.historyfile);
+    }
 
     firstarg = parseOptions(argc,argv);
     argc -= firstarg;

src/redis.c

@@ -74,6 +74,7 @@ struct redisCommand readonlyCommandTable[] = {
     {"setex",setexCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,0,0,0},
     {"append",appendCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
     {"substr",substrCommand,4,REDIS_CMD_INLINE,NULL,1,1,1},
+    {"strlen",strlenCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
     {"del",delCommand,-2,REDIS_CMD_INLINE,NULL,0,0,0},
     {"exists",existsCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
     {"incr",incrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
@@ -169,6 +170,7 @@ struct redisCommand readonlyCommandTable[] = {
     {"info",infoCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
     {"monitor",monitorCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
     {"ttl",ttlCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
+    {"persist",persistCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
     {"slaveof",slaveofCommand,3,REDIS_CMD_INLINE,NULL,0,0,0},
     {"debug",debugCommand,-2,REDIS_CMD_INLINE,NULL,0,0,0},
     {"config",configCommand,-2,REDIS_CMD_BULK,NULL,0,0,0},
@@ -186,23 +188,22 @@ struct redisCommand readonlyCommandTable[] = {
 void redisLog(int level, const char *fmt, ...) {
     va_list ap;
     FILE *fp;
+    char *c = ".-*#";
+    char buf[64];
+    time_t now;
+
+    if (level < server.verbosity) return;
 
     fp = (server.logfile == NULL) ? stdout : fopen(server.logfile,"a");
     if (!fp) return;
 
     va_start(ap, fmt);
-    if (level >= server.verbosity) {
+    now = time(NULL);
-        char *c = ".-*#";
+    strftime(buf,64,"%d %b %H:%M:%S",localtime(&now));
-        char buf[64];
+    fprintf(fp,"[%d] %s %c ",(int)getpid(),buf,c[level]);
-        time_t now;
+    vfprintf(fp, fmt, ap);
-
+    fprintf(fp,"\n");
-        now = time(NULL);
+    fflush(fp);
-        strftime(buf,64,"%d %b %H:%M:%S",localtime(&now));
-        fprintf(fp,"[%d] %s %c ",(int)getpid(),buf,c[level]);
-        vfprintf(fp, fmt, ap);
-        fprintf(fp,"\n");
-        fflush(fp);
-    }
     va_end(ap);
 
     if (server.logfile) fclose(fp);
@@ -435,6 +436,48 @@ void updateDictResizePolicy(void) {
 
 /* ======================= Cron: called every 100 ms ======================== */
 
+/* Try to expire a few timed out keys. The algorithm used is adaptive and
+ * will use few CPU cycles if there are few expiring keys, otherwise
+ * it will get more aggressive to avoid that too much memory is used by
+ * keys that can be removed from the keyspace. */
+void activeExpireCycle(void) {
+    int j;
+
+    for (j = 0; j < server.dbnum; j++) {
+        int expired;
+        redisDb *db = server.db+j;
+
+        /* Continue to expire if at the end of the cycle more than 25%
+         * of the keys were expired. */
+        do {
+            long num = dictSize(db->expires);
+            time_t now = time(NULL);
+
+            expired = 0;
+            if (num > REDIS_EXPIRELOOKUPS_PER_CRON)
+                num = REDIS_EXPIRELOOKUPS_PER_CRON;
+            while (num--) {
+                dictEntry *de;
+                time_t t;
+
+                if ((de = dictGetRandomKey(db->expires)) == NULL) break;
+                t = (time_t) dictGetEntryVal(de);
+                if (now > t) {
+                    sds key = dictGetEntryKey(de);
+                    robj *keyobj = createStringObject(key,sdslen(key));
+
+                    propagateExpire(db,keyobj);
+                    dbDelete(db,keyobj);
+                    decrRefCount(keyobj);
+                    expired++;
+                    server.stat_expiredkeys++;
+                }
+            }
+        } while (expired > REDIS_EXPIRELOOKUPS_PER_CRON/4);
+    }
+}
+
+
 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     int j, loops = server.cronloops++;
     REDIS_NOTUSED(eventLoop);
@@ -533,41 +576,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
          }
     }
 
-    /* Try to expire a few timed out keys. The algorithm used is adaptive and
+    /* Expire a few keys per cycle, only if this is a master.
-     * will use few CPU cycles if there are few expiring keys, otherwise
+     * On slaves we wait for DEL operations synthesized by the master
-     * it will get more aggressive to avoid that too much memory is used by
+     * in order to guarantee a strict consistency. */
-     * keys that can be removed from the keyspace. */
+    if (server.masterhost == NULL) activeExpireCycle();
-    for (j = 0; j < server.dbnum; j++) {
-        int expired;
-        redisDb *db = server.db+j;
-
-        /* Continue to expire if at the end of the cycle more than 25%
-         * of the keys were expired. */
-        do {
-            long num = dictSize(db->expires);
-            time_t now = time(NULL);
-
-            expired = 0;
-            if (num > REDIS_EXPIRELOOKUPS_PER_CRON)
-                num = REDIS_EXPIRELOOKUPS_PER_CRON;
-            while (num--) {
-                dictEntry *de;
-                time_t t;
-
-                if ((de = dictGetRandomKey(db->expires)) == NULL) break;
-                t = (time_t) dictGetEntryVal(de);
-                if (now > t) {
-                    sds key = dictGetEntryKey(de);
-                    robj *keyobj = createStringObject(key,sdslen(key));
-
-                    dbDelete(db,keyobj);
-                    decrRefCount(keyobj);
-                    expired++;
-                    server.stat_expiredkeys++;
-                }
-            }
-        } while (expired > REDIS_EXPIRELOOKUPS_PER_CRON/4);
-    }
 
     /* Swap a few keys on disk if we are over the memory limit and VM
      * is enbled. Try to free objects from the free list first. */
@@ -761,6 +773,7 @@ void initServer() {
     signal(SIGPIPE, SIG_IGN);
     setupSigSegvAction();
 
+    server.mainthread = pthread_self();
     server.devnull = fopen("/dev/null","w");
     if (server.devnull == NULL) {
         redisLog(REDIS_WARNING, "Can't open /dev/null: %s", server.neterr);
@@ -827,7 +840,7 @@ int qsortRedisCommands(const void *r1, const void *r2) {
 
 void sortCommandTable() {
     /* Copy and sort the read-only version of the command table */
-    commandTable = (struct redisCommand*)malloc(sizeof(readonlyCommandTable));
+    commandTable = (struct redisCommand*)zmalloc(sizeof(readonlyCommandTable));
     memcpy(commandTable,readonlyCommandTable,sizeof(readonlyCommandTable));
     qsort(commandTable,
         sizeof(readonlyCommandTable)/sizeof(struct redisCommand),

src/redis.h

@@ -16,6 +16,7 @@
 #include <unistd.h>
 #include <errno.h>
 #include <inttypes.h>
+#include <pthread.h>
 
 #include "ae.h"     /* Event driven programming library */
 #include "sds.h"    /* Dynamic safe strings */
@@ -329,6 +330,7 @@ struct sharedObjectsStruct {
 
 /* Global server state structure */
 struct redisServer {
+    pthread_t mainthread;
     int port;
     int fd;
     redisDb *db;
@@ -775,10 +777,10 @@ void resetServerSaveParams();
 
 /* db.c -- Keyspace access API */
 int removeExpire(redisDb *db, robj *key);
+void propagateExpire(redisDb *db, robj *key);
 int expireIfNeeded(redisDb *db, robj *key);
-int deleteIfVolatile(redisDb *db, robj *key);
 time_t getExpire(redisDb *db, robj *key);
-int setExpire(redisDb *db, robj *key, time_t when);
+void setExpire(redisDb *db, robj *key, time_t when);
 robj *lookupKey(redisDb *db, robj *key);
 robj *lookupKeyRead(redisDb *db, robj *key);
 robj *lookupKeyWrite(redisDb *db, robj *key);
@@ -861,6 +863,7 @@ void expireCommand(redisClient *c);
 void expireatCommand(redisClient *c);
 void getsetCommand(redisClient *c);
 void ttlCommand(redisClient *c);
+void persistCommand(redisClient *c);
 void slaveofCommand(redisClient *c);
 void debugCommand(redisClient *c);
 void msetCommand(redisClient *c);
@@ -882,6 +885,7 @@ void blpopCommand(redisClient *c);
 void brpopCommand(redisClient *c);
 void appendCommand(redisClient *c);
 void substrCommand(redisClient *c);
+void strlenCommand(redisClient *c);
 void zrankCommand(redisClient *c);
 void zrevrankCommand(redisClient *c);
 void hsetCommand(redisClient *c);
@@ -908,4 +912,11 @@ void publishCommand(redisClient *c);
 void watchCommand(redisClient *c);
 void unwatchCommand(redisClient *c);
 
+#if defined(__GNUC__)
+void *calloc(size_t count, size_t size) __attribute__ ((deprecated));
+void free(void *ptr) __attribute__ ((deprecated));
+void *malloc(size_t size) __attribute__ ((deprecated));
+void *realloc(void *ptr, size_t size) __attribute__ ((deprecated));
+#endif
+
 #endif

src/sds.c

@@ -382,3 +382,80 @@ sds sdscatrepr(sds s, char *p, size_t len) {
     }
     return sdscatlen(s,"\"",1);
 }
+
+/* Split a line into arguments, where every argument can be in the
+ * following programming-language REPL-alike form:
+ *
+ * foo bar "newline are supported\n" and "\xff\x00otherstuff"
+ *
+ * The number of arguments is stored into *argc, and an array
+ * of sds is returned. The caller should sdsfree() all the returned
+ * strings and finally zfree() the array itself.
+ *
+ * Note that sdscatrepr() is able to convert back a string into
+ * a quoted string in the same format sdssplitargs() is able to parse.
+ */
+sds *sdssplitargs(char *line, int *argc) {
+    char *p = line;
+    char *current = NULL;
+    char **vector = NULL;
+
+    *argc = 0;
+    while(1) {
+        /* skip blanks */
+        while(*p && isspace(*p)) p++;
+        if (*p) {
+            /* get a token */
+            int inq=0; /* set to 1 if we are in "quotes" */
+            int done = 0;
+
+            if (current == NULL) current = sdsempty();
+            while(!done) {
+                if (inq) {
+                    if (*p == '\\' && *(p+1)) {
+                        char c;
+
+                        p++;
+                        switch(*p) {
+                        case 'n': c = '\n'; break;
+                        case 'r': c = '\r'; break;
+                        case 't': c = '\t'; break;
+                        case 'b': c = '\b'; break;
+                        case 'a': c = '\a'; break;
+                        default: c = *p; break;
+                        }
+                        current = sdscatlen(current,&c,1);
+                    } else if (*p == '"') {
+                        done = 1;
+                    } else {
+                        current = sdscatlen(current,p,1);
+                    }
+                } else {
+                    switch(*p) {
+                    case ' ':
+                    case '\n':
+                    case '\r':
+                    case '\t':
+                    case '\0':
+                        done=1;
+                        break;
+                    case '"':
+                        inq=1;
+                        break;
+                    default:
+                        current = sdscatlen(current,p,1);
+                        break;
+                    }
+                }
+                if (*p) p++;
+            }
+            /* add the token to the vector */
+            vector = zrealloc(vector,((*argc)+1)*sizeof(char*));
+            vector[*argc] = current;
+            (*argc)++;
+            current = NULL;
+        } else {
+            return vector;
+        }
+    }
+}

src/sds.h

@@ -70,5 +70,6 @@ void sdstolower(sds s);
 void sdstoupper(sds s);
 sds sdsfromlonglong(long long value);
 sds sdscatrepr(sds s, char *p, size_t len);
+sds *sdssplitargs(char *line, int *argc);
 
 #endif

src/sort.c

@@ -364,6 +364,7 @@ void sortCommand(redisClient *c) {
          * SORT result is empty a new key is set and maybe the old content
          * replaced. */
         server.dirty += 1+outputlen;
+        touchWatchedKey(c->db,storekey);
         addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",outputlen));
     }
 

src/t_hash.c

@@ -224,6 +224,7 @@ void hsetCommand(redisClient *c) {
     hashTypeTryObjectEncoding(o,&c->argv[2], &c->argv[3]);
     update = hashTypeSet(o,c->argv[2],c->argv[3]);
     addReply(c, update ? shared.czero : shared.cone);
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
 }
 
@@ -238,6 +239,7 @@ void hsetnxCommand(redisClient *c) {
         hashTypeTryObjectEncoding(o,&c->argv[2], &c->argv[3]);
         hashTypeSet(o,c->argv[2],c->argv[3]);
         addReply(c, shared.cone);
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
     }
 }
@@ -258,6 +260,7 @@ void hmsetCommand(redisClient *c) {
         hashTypeSet(o,c->argv[i],c->argv[i+1]);
     }
     addReply(c, shared.ok);
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
 }
 
@@ -284,6 +287,7 @@ void hincrbyCommand(redisClient *c) {
     hashTypeSet(o,c->argv[2],new);
     decrRefCount(new);
     addReplyLongLong(c,value);
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
 }
 
@@ -330,6 +334,7 @@ void hdelCommand(redisClient *c) {
     if (hashTypeDelete(o,c->argv[2])) {
         if (hashTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
         addReply(c,shared.cone);
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
     } else {
         addReply(c,shared.czero);

src/t_list.c

@@ -273,12 +273,14 @@ void pushGenericCommand(redisClient *c, int where) {
             return;
         }
         if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
+            touchWatchedKey(c->db,c->argv[1]);
             addReply(c,shared.cone);
             return;
         }
     }
     listTypePush(lobj,c->argv[2],where);
     addReplyLongLong(c,listTypeLength(lobj));
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
 }
 
@@ -327,6 +329,7 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
             if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
                 ziplistLen(subject->ptr) > server.list_max_ziplist_entries)
                     listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
+            touchWatchedKey(c->db,c->argv[1]);
             server.dirty++;
         } else {
             /* Notify client of a failed insert */
@@ -335,6 +338,7 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
         }
     } else {
         listTypePush(subject,val,where);
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
     }
 
@@ -419,6 +423,7 @@ void lsetCommand(redisClient *c) {
             o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr));
             decrRefCount(value);
             addReply(c,shared.ok);
+            touchWatchedKey(c->db,c->argv[1]);
             server.dirty++;
         }
     } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
@@ -430,6 +435,7 @@ void lsetCommand(redisClient *c) {
             listNodeValue(ln) = value;
             incrRefCount(value);
             addReply(c,shared.ok);
+            touchWatchedKey(c->db,c->argv[1]);
             server.dirty++;
         }
     } else {
@@ -448,6 +454,7 @@ void popGenericCommand(redisClient *c, int where) {
         addReplyBulk(c,value);
         decrRefCount(value);
         if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
     }
 }
@@ -476,11 +483,10 @@ void lrangeCommand(redisClient *c) {
     if (start < 0) start = llen+start;
     if (end < 0) end = llen+end;
     if (start < 0) start = 0;
-    if (end < 0) end = 0;
 
-    /* indexes sanity checks */
+    /* Invariant: start >= 0, so this test will be true when end < 0.
+     * The range is empty when start > end or start >= length. */
     if (start > end || start >= llen) {
-        /* Out of range start or start > end result in empty list */
         addReply(c,shared.emptymultibulk);
         return;
     }
@@ -516,9 +522,9 @@ void ltrimCommand(redisClient *c) {
     if (start < 0) start = llen+start;
     if (end < 0) end = llen+end;
     if (start < 0) start = 0;
-    if (end < 0) end = 0;
 
-    /* indexes sanity checks */
+    /* Invariant: start >= 0, so this test will be true when end < 0.
+     * The range is empty when start > end or start >= length. */
     if (start > end || start >= llen) {
         /* Out of range start or start > end result in empty list */
         ltrim = llen;
@@ -547,6 +553,7 @@ void ltrimCommand(redisClient *c) {
         redisPanic("Unknown list encoding");
     }
     if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
     addReply(c,shared.ok);
 }
@@ -588,6 +595,7 @@ void lremCommand(redisClient *c) {
 
     if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]);
     addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
+    if (removed) touchWatchedKey(c->db,c->argv[1]);
 }
 
 /* This is the semantic of this command:
@@ -636,6 +644,7 @@ void rpoplpushcommand(redisClient *c) {
 
         /* Delete the source list when it is empty */
         if (listTypeLength(sobj) == 0) dbDelete(c->db,c->argv[1]);
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
     }
 }

src/t_set.c

@@ -188,6 +188,7 @@ void saddCommand(redisClient *c) {
         }
     }
     if (setTypeAdd(set,c->argv[2])) {
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
         addReply(c,shared.cone);
     } else {
@@ -203,6 +204,7 @@ void sremCommand(redisClient *c) {
 
     if (setTypeRemove(set,c->argv[2])) {
         if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
         addReply(c,shared.cone);
     } else {
@@ -241,6 +243,8 @@ void smoveCommand(redisClient *c) {
 
     /* Remove the src set from the database when empty */
     if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]);
+    touchWatchedKey(c->db,c->argv[1]);
+    touchWatchedKey(c->db,c->argv[2]);
     server.dirty++;
 
     /* Create the destination set when it doesn't exist */
@@ -289,6 +293,7 @@ void spopCommand(redisClient *c) {
         addReplyBulk(c,ele);
         decrRefCount(ele);
         if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
     }
 }
@@ -325,8 +330,10 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
         if (!setobj) {
             zfree(sets);
             if (dstkey) {
-                if (dbDelete(c->db,dstkey))
+                if (dbDelete(c->db,dstkey)) {
+                    touchWatchedKey(c->db,dstkey);
                     server.dirty++;
+                }
                 addReply(c,shared.czero);
             } else {
                 addReply(c,shared.emptymultibulk);
@@ -390,6 +397,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }
+        touchWatchedKey(c->db,dstkey);
         server.dirty++;
     } else {
         lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",cardinality);
@@ -481,6 +489,7 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *
             decrRefCount(dstset);
             addReply(c,shared.czero);
         }
+        touchWatchedKey(c->db,dstkey);
         server.dirty++;
     }
     zfree(sets);

src/t_string.c

@@ -17,8 +17,6 @@ void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj *expir
         }
     }
 
-    touchWatchedKey(c->db,key);
-    if (nx) deleteIfVolatile(c->db,key);
     retval = dbAdd(c->db,key,val);
     if (retval == REDIS_ERR) {
         if (!nx) {
@@ -31,6 +29,7 @@ void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj *expir
     } else {
         incrRefCount(val);
     }
+    touchWatchedKey(c->db,key);
     server.dirty++;
     removeExpire(c->db,key);
     if (expire) setExpire(c->db,key,time(NULL)+seconds);
@@ -72,6 +71,7 @@ void getsetCommand(redisClient *c) {
     if (getGenericCommand(c) == REDIS_ERR) return;
     dbReplace(c->db,c->argv[1],c->argv[2]);
     incrRefCount(c->argv[2]);
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
     removeExpire(c->db,c->argv[1]);
 }
@@ -120,6 +120,7 @@ void msetGenericCommand(redisClient *c, int nx) {
         dbReplace(c->db,c->argv[j],c->argv[j+1]);
         incrRefCount(c->argv[j+1]);
         removeExpire(c->db,c->argv[j]);
+        touchWatchedKey(c->db,c->argv[j]);
     }
     server.dirty += (c->argc-1)/2;
     addReply(c, nx ? shared.cone : shared.ok);
@@ -144,6 +145,7 @@ void incrDecrCommand(redisClient *c, long long incr) {
     value += incr;
     o = createStringObjectFromLongLong(value);
     dbReplace(c->db,c->argv[1],o);
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
     addReply(c,shared.colon);
     addReply(c,o);
@@ -207,6 +209,7 @@ void appendCommand(redisClient *c) {
         }
         totlen = sdslen(o->ptr);
     }
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
     addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",(unsigned long)totlen));
 }
@@ -248,4 +251,13 @@ void substrCommand(redisClient *c) {
     decrRefCount(o);
 }
 
+void strlenCommand(redisClient *c) {
+    robj *o;
 
+    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
+        checkType(c,o,REDIS_STRING)) return;
+
+    o = getDecodedObject(o);
+    addReplyLongLong(c,sdslen(o->ptr));
+    decrRefCount(o);
+}

src/t_zset.c

@@ -327,11 +327,6 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, i
     zset *zs;
     double *score;
 
-    if (isnan(scoreval)) {
-        addReplySds(c,sdsnew("-ERR provide score is Not A Number (nan)\r\n"));
-        return;
-    }
-
     zsetobj = lookupKeyWrite(c->db,key);
     if (zsetobj == NULL) {
         zsetobj = createZsetObject();
@@ -361,7 +356,7 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, i
         }
         if (isnan(*score)) {
             addReplySds(c,
-                sdsnew("-ERR resulting score is Not A Number (nan)\r\n"));
+                sdsnew("-ERR resulting score is not a number (NaN)\r\n"));
             zfree(score);
             /* Note that we don't need to check if the zset may be empty and
              * should be removed here, as we can only obtain Nan as score if
@@ -379,6 +374,7 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, i
         incrRefCount(ele); /* added to hash */
         zslInsert(zs->zsl,*score,ele);
         incrRefCount(ele); /* added to skiplist */
+        touchWatchedKey(c->db,c->argv[1]);
         server.dirty++;
         if (doincrement)
             addReplyDouble(c,*score);
@@ -402,6 +398,7 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, i
             incrRefCount(ele);
             /* Update the score in the hash table */
             dictReplace(zs->dict,ele,score);
+            touchWatchedKey(c->db,c->argv[1]);
             server.dirty++;
         } else {
             zfree(score);
@@ -415,15 +412,13 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scoreval, i
 
 void zaddCommand(redisClient *c) {
     double scoreval;
-
+    if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
-    if (getDoubleFromObjectOrReply(c, c->argv[2], &scoreval, NULL) != REDIS_OK) return;
     zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,0);
 }
 
 void zincrbyCommand(redisClient *c) {
     double scoreval;
-
+    if (getDoubleFromObjectOrReply(c,c->argv[2],&scoreval,NULL) != REDIS_OK) return;
-    if (getDoubleFromObjectOrReply(c, c->argv[2], &scoreval, NULL) != REDIS_OK) return;
     zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,1);
 }
 
@@ -452,6 +447,7 @@ void zremCommand(redisClient *c) {
     dictDelete(zs->dict,c->argv[2]);
     if (htNeedsResize(zs->dict)) dictResize(zs->dict);
     if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
+    touchWatchedKey(c->db,c->argv[1]);
     server.dirty++;
     addReply(c,shared.cone);
 }
@@ -473,6 +469,7 @@ void zremrangebyscoreCommand(redisClient *c) {
     deleted = zslDeleteRangeByScore(zs->zsl,min,max,zs->dict);
     if (htNeedsResize(zs->dict)) dictResize(zs->dict);
     if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
+    if (deleted) touchWatchedKey(c->db,c->argv[1]);
     server.dirty += deleted;
     addReplyLongLong(c,deleted);
 }
@@ -497,9 +494,9 @@ void zremrangebyrankCommand(redisClient *c) {
     if (start < 0) start = llen+start;
     if (end < 0) end = llen+end;
     if (start < 0) start = 0;
-    if (end < 0) end = 0;
 
-    /* indexes sanity checks */
+    /* Invariant: start >= 0, so this test will be true when end < 0.
+     * The range is empty when start > end or start >= length. */
     if (start > end || start >= llen) {
         addReply(c,shared.czero);
         return;
@@ -511,6 +508,7 @@ void zremrangebyrankCommand(redisClient *c) {
     deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
     if (htNeedsResize(zs->dict)) dictResize(zs->dict);
     if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
+    if (deleted) touchWatchedKey(c->db,c->argv[1]);
     server.dirty += deleted;
     addReplyLongLong(c, deleted);
 }
@@ -536,6 +534,10 @@ int qsortCompareZsetopsrcByCardinality(const void *s1, const void *s2) {
 inline static void zunionInterAggregate(double *target, double val, int aggregate) {
     if (aggregate == REDIS_AGGR_SUM) {
         *target = *target + val;
+        /* The result of adding two doubles is NaN when one variable
+         * is +inf and the other is -inf. When these numbers are added,
+         * we maintain the convention of the result being 0.0. */
+        if (isnan(*target)) *target = 0.0;
     } else if (aggregate == REDIS_AGGR_MIN) {
         *target = val < *target ? val : *target;
     } else if (aggregate == REDIS_AGGR_MAX) {
@@ -554,6 +556,7 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
     zset *dstzset;
     dictIterator *di;
     dictEntry *de;
+    int touched = 0;
 
     /* expect setnum input keys to be given */
     setnum = atoi(c->argv[2]->ptr);
@@ -598,8 +601,12 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
             if (remaining >= (setnum + 1) && !strcasecmp(c->argv[j]->ptr,"weights")) {
                 j++; remaining--;
                 for (i = 0; i < setnum; i++, j++, remaining--) {
-                    if (getDoubleFromObjectOrReply(c, c->argv[j], &src[i].weight, NULL) != REDIS_OK)
+                    if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight,
+                            "weight value is not a double") != REDIS_OK)
+                    {
+                        zfree(src);
                         return;
+                    }
                 }
             } else if (remaining >= 2 && !strcasecmp(c->argv[j]->ptr,"aggregate")) {
                 j++; remaining--;
@@ -698,10 +705,15 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
         redisAssert(op == REDIS_OP_INTER || op == REDIS_OP_UNION);
     }
 
-    dbDelete(c->db,dstkey);
+    if (dbDelete(c->db,dstkey)) {
+        touchWatchedKey(c->db,dstkey);
+        touched = 1;
+        server.dirty++;
+    }
     if (dstzset->zsl->length) {
         dbAdd(c->db,dstkey,dstobj);
         addReplyLongLong(c, dstzset->zsl->length);
+        if (!touched) touchWatchedKey(c->db,dstkey);
         server.dirty++;
     } else {
         decrRefCount(dstobj);
@@ -750,11 +762,10 @@ void zrangeGenericCommand(redisClient *c, int reverse) {
     if (start < 0) start = llen+start;
     if (end < 0) end = llen+end;
     if (start < 0) start = 0;
-    if (end < 0) end = 0;
 
-    /* indexes sanity checks */
+    /* Invariant: start >= 0, so this test will be true when end < 0.
+     * The range is empty when start > end or start >= length. */
     if (start > end || start >= llen) {
-        /* Out of range start or start > end result in empty list */
         addReply(c,shared.emptymultibulk);
         return;
     }

src/vm.c

@@ -86,10 +86,9 @@ void vmInit(void) {
     } else {
         redisLog(REDIS_NOTICE,"Swap file allocated with success");
     }
-    server.vm_bitmap = zmalloc((server.vm_pages+7)/8);
+    server.vm_bitmap = zcalloc((server.vm_pages+7)/8);
     redisLog(REDIS_VERBOSE,"Allocated %lld bytes page table for %lld pages",
         (long long) (server.vm_pages+7)/8, server.vm_pages);
-    memset(server.vm_bitmap,0,(server.vm_pages+7)/8);
 
     /* Initialize threaded I/O (used by Virtual Memory) */
     server.io_newjobs = listCreate();
@@ -1075,6 +1074,11 @@ int dontWaitForSwappedKey(redisClient *c, robj *key) {
     listIter li;
     struct dictEntry *de;
 
+    /* The key object might be destroyed when deleted from the c->io_keys
+     * list (and the "key" argument is physically the same object as the
+     * object inside the list), so we need to protect it. */
+    incrRefCount(key);
+
     /* Remove the key from the list of keys this client is waiting for. */
     listRewind(c->io_keys,&li);
     while ((ln = listNext(&li)) != NULL) {
@@ -1095,6 +1099,7 @@ int dontWaitForSwappedKey(redisClient *c, robj *key) {
     if (listLength(l) == 0)
         dictDelete(c->db->io_keys,key);
 
+    decrRefCount(key);
     return listLength(c->io_keys) == 0;
 }
 

src/ziplist.c

@@ -23,6 +23,8 @@
 #include "zmalloc.h"
 #include "ziplist.h"
 
+int ll2string(char *s, size_t len, long long value);
+
 /* Important note: the ZIP_END value is used to depict the end of the
  * ziplist structure. When a pointer contains an entry, the first couple
  * of bytes contain the encoded length of the previous entry. This length
@@ -174,15 +176,27 @@ static int zipPrevLenByteDiff(unsigned char *p, unsigned int len) {
 }
 
 /* Check if string pointed to by 'entry' can be encoded as an integer.
- * Stores the integer value in 'v' and its encoding in 'encoding'.
+ * Stores the integer value in 'v' and its encoding in 'encoding'. */
- * Warning: this function requires a NULL-terminated string! */
+static int zipTryEncoding(unsigned char *entry, unsigned int entrylen, long long *v, unsigned char *encoding) {
-static int zipTryEncoding(unsigned char *entry, long long *v, unsigned char *encoding) {
     long long value;
     char *eptr;
+    char buf[32];
 
+    if (entrylen >= 32 || entrylen == 0) return 0;
     if (entry[0] == '-' || (entry[0] >= '0' && entry[0] <= '9')) {
-        value = strtoll((char*)entry,&eptr,10);
+        int slen;
+
+        /* Perform a back-and-forth conversion to make sure that
+         * the string turned into an integer is not losing any info. */
+        memcpy(buf,entry,entrylen);
+        buf[entrylen] = '\0';
+        value = strtoll(buf,&eptr,10);
         if (eptr[0] != '\0') return 0;
+        slen = ll2string(buf,32,value);
+        if (entrylen != (unsigned)slen || memcmp(buf,entry,slen)) return 0;
+
+        /* Great, the string can be encoded. Check what's the smallest
+         * of our encoding types that can hold this value. */
         if (value >= INT16_MIN && value <= INT16_MAX) {
             *encoding = ZIP_ENC_INT16;
         } else if (value >= INT32_MIN && value <= INT32_MAX) {
@@ -329,7 +343,7 @@ static unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsig
     }
 
     /* See if the entry can be encoded */
-    if (zipTryEncoding(s,&value,&encoding)) {
+    if (zipTryEncoding(s,slen,&value,&encoding)) {
         reqlen = zipEncodingSize(encoding);
     } else {
         reqlen = slen;
@@ -505,7 +519,7 @@ unsigned int ziplistCompare(unsigned char *p, unsigned char *sstr, unsigned int
         }
     } else {
         /* Try to compare encoded values */
-        if (zipTryEncoding(sstr,&sval,&sencoding)) {
+        if (zipTryEncoding(sstr,slen,&sval,&sencoding)) {
             if (entry.encoding == sencoding) {
                 zval = zipLoadInteger(p+entry.headersize,entry.encoding);
                 return zval == sval;

src/zmalloc.c

@@ -89,6 +89,20 @@ void *zmalloc(size_t size) {
 #endif
 }
 
+void *zcalloc(size_t size) {
+    void *ptr = calloc(1, size+PREFIX_SIZE);
+
+    if (!ptr) zmalloc_oom(size);
+#ifdef HAVE_MALLOC_SIZE
+    increment_used_memory(redis_malloc_size(ptr));
+    return ptr;
+#else
+    *((size_t*)ptr) = size;
+    increment_used_memory(size+PREFIX_SIZE);
+    return (char*)ptr+PREFIX_SIZE;
+#endif
+}
+
 void *zrealloc(void *ptr, size_t size) {
 #ifndef HAVE_MALLOC_SIZE
     void *realptr;

src/zmalloc.h

@@ -32,6 +32,7 @@
 #define _ZMALLOC_H
 
 void *zmalloc(size_t size);
+void *zcalloc(size_t size);
 void *zrealloc(void *ptr, size_t size);
 void zfree(void *ptr);
 char *zstrdup(const char *s);

tests/integration/replication.tcl

@@ -1,3 +1,49 @@
+start_server {tags {"repl"}} {
+    start_server {} {
+        test {First server should have role slave after SLAVEOF} {
+            r -1 slaveof [srv 0 host] [srv 0 port]
+            after 1000
+            s -1 role
+        } {slave}
+
+        test {MASTER and SLAVE dataset should be identical after complex ops} {
+            createComplexDataset r 10000
+            after 500
+            if {[r debug digest] ne [r -1 debug digest]} {
+                set csv1 [csvdump r]
+                set csv2 [csvdump {r -1}]
+                set fd [open /tmp/repldump1.txt w]
+                puts -nonewline $fd $csv1
+                close $fd
+                set fd [open /tmp/repldump2.txt w]
+                puts -nonewline $fd $csv2
+                close $fd
+                puts "Master - Slave inconsistency"
+                puts "Run diff -u against /tmp/repldump*.txt for more info"
+            }
+            assert_equal [r debug digest] [r -1 debug digest]
+        }
+
+        test {MASTER and SLAVE consistency with expire} {
+            createComplexDataset r 50000 useexpire
+            after 4000 ;# Make sure everything expired before taking the digest
+            if {[r debug digest] ne [r -1 debug digest]} {
+                set csv1 [csvdump r]
+                set csv2 [csvdump {r -1}]
+                set fd [open /tmp/repldump1.txt w]
+                puts -nonewline $fd $csv1
+                close $fd
+                set fd [open /tmp/repldump2.txt w]
+                puts -nonewline $fd $csv2
+                close $fd
+                puts "Master - Slave inconsistency"
+                puts "Run diff -u against /tmp/repldump*.txt for more info"
+            }
+            assert_equal [r debug digest] [r -1 debug digest]
+        }
+    }
+}
+
 start_server {tags {"repl"}} {
     r set mykey foo
     

tests/support/test.tcl

@@ -33,9 +33,14 @@ proc assert_error {pattern code} {
 }
 
 proc assert_encoding {enc key} {
-    # swapped out value doesn't have encoding, so swap in first
+    # Swapped out values don't have an encoding, so make sure that
-    r debug swapin $key
+    # the value is swapped in before checking the encoding.
-    assert_match "* encoding:$enc *" [r debug object $key]
+    set dbg [r debug object $key]
+    while {[string match "* swapped at:*" $dbg]} {
+        r debug swapin $key
+        set dbg [r debug object $key]
+    }
+    assert_match "* encoding:$enc *" $dbg
 }
 
 proc assert_type {type key} {

tests/support/util.tcl

@@ -44,7 +44,7 @@ proc warnings_from_file {filename} {
 
 # Return value for INFO property
 proc status {r property} {
-    if {[regexp "\r\n$property:(.*?)\r\n" [$r info] _ value]} {
+    if {[regexp "\r\n$property:(.*?)\r\n" [{*}$r info] _ value]} {
         set _ $value
     }
 }
@@ -127,11 +127,32 @@ proc randomKey {} {
     }
 }
 
-proc createComplexDataset {r ops} {
+proc findKeyWithType {r type} {
+    for {set j 0} {$j < 20} {incr j} {
+        set k [{*}$r randomkey]
+        if {$k eq {}} {
+            return {}
+        }
+        if {[{*}$r type $k] eq $type} {
+            return $k
+        }
+    }
+    return {}
+}
+
+proc createComplexDataset {r ops {opt {}}} {
     for {set j 0} {$j < $ops} {incr j} {
         set k [randomKey]
+        set k2 [randomKey]
         set f [randomValue]
         set v [randomValue]
+
+        if {[lsearch -exact $opt useexpire] != -1} {
+            if {rand() < 0.1} {
+                {*}$r expire [randomKey] [randomInt 2]
+            }
+        }
+
         randpath {
             set d [expr {rand()}]
         } {
@@ -145,21 +166,23 @@ proc createComplexDataset {r ops} {
         } {
             randpath {set d +inf} {set d -inf}
         }
-        set t [$r type $k]
+        set t [{*}$r type $k]
 
         if {$t eq {none}} {
             randpath {
-                $r set $k $v
+                {*}$r set $k $v
             } {
-                $r lpush $k $v
+                {*}$r lpush $k $v
             } {
-                $r sadd $k $v
+                {*}$r sadd $k $v
             } {
-                $r zadd $k $d $v
+                {*}$r zadd $k $d $v
             } {
-                $r hset $k $f $v
+                {*}$r hset $k $f $v
+            } {
+                {*}$r del $k
             }
-            set t [$r type $k]
+            set t [{*}$r type $k]
         }
 
         switch $t {
@@ -167,23 +190,45 @@ proc createComplexDataset {r ops} {
                 # Nothing to do
             }
             {list} {
-                randpath {$r lpush $k $v} \
+                randpath {{*}$r lpush $k $v} \
-                        {$r rpush $k $v} \
+                        {{*}$r rpush $k $v} \
-                        {$r lrem $k 0 $v} \
+                        {{*}$r lrem $k 0 $v} \
-                        {$r rpop $k} \
+                        {{*}$r rpop $k} \
-                        {$r lpop $k}
+                        {{*}$r lpop $k}
             }
             {set} {
-                randpath {$r sadd $k $v} \
+                randpath {{*}$r sadd $k $v} \
-                        {$r srem $k $v}
+                        {{*}$r srem $k $v} \
+                        {
+                            set otherset [findKeyWithType r set]
+                            if {$otherset ne {}} {
+                                randpath {
+                                    {*}$r sunionstore $k2 $k $otherset
+                                } {
+                                    {*}$r sinterstore $k2 $k $otherset
+                                } {
+                                    {*}$r sdiffstore $k2 $k $otherset
+                                }
+                            }
+                        }
             }
             {zset} {
-                randpath {$r zadd $k $d $v} \
+                randpath {{*}$r zadd $k $d $v} \
-                        {$r zrem $k $v}
+                        {{*}$r zrem $k $v} \
+                        {
+                            set otherzset [findKeyWithType r zset]
+                            if {$otherzset ne {}} {
+                                randpath {
+                                    {*}$r zunionstore $k2 2 $k $otherzset
+                                } {
+                                    {*}$r zinterstore $k2 2 $k $otherzset
+                                }
+                            }
+                        }
             }
             {hash} {
-                randpath {$r hset $k $f $v} \
+                randpath {{*}$r hset $k $f $v} \
-                        {$r hdel $k $f}
+                        {{*}$r hdel $k $f}
             }
         }
     }
@@ -196,3 +241,52 @@ proc formatCommand {args} {
     }
     set _ $cmd
 }
+
+proc csvdump r {
+    set o {}
+    foreach k [lsort [{*}$r keys *]] {
+        set type [{*}$r type $k]
+        append o [csvstring $k] , [csvstring $type] ,
+        switch $type {
+            string {
+                append o [csvstring [{*}$r get $k]] "\n"
+            }
+            list {
+                foreach e [{*}$r lrange $k 0 -1] {
+                    append o [csvstring $e] ,
+                }
+                append o "\n"
+            }
+            set {
+                foreach e [lsort [{*}$r smembers $k]] {
+                    append o [csvstring $e] ,
+                }
+                append o "\n"
+            }
+            zset {
+                foreach e [{*}$r zrange $k 0 -1 withscores] {
+                    append o [csvstring $e] ,
+                }
+                append o "\n"
+            }
+            hash {
+                set fields [{*}$r hgetall $k]
+                set newfields {}
+                foreach {k v} $fields {
+                    lappend newfields [list $k $v]
+                }
+                set fields [lsort -index 0 $newfields]
+                foreach kv $fields {
+                    append o [csvstring [lindex $kv 0]] ,
+                    append o [csvstring [lindex $kv 1]] ,
+                }
+                append o "\n"
+            }
+        }
+    }
+    return $o
+}
+
+proc csvstring s {
+    return "\"$s\""
+}

tests/test_helper.tcl

@@ -102,13 +102,13 @@ proc main {} {
     execute_tests "unit/expire"
     execute_tests "unit/other"
     execute_tests "unit/cas"
-    
+
+    cleanup
     puts "\n[expr $::passed+$::failed] tests, $::passed passed, $::failed failed"
     if {$::failed > 0} {
         puts "\n*** WARNING!!! $::failed FAILED TESTS ***\n"
+        exit 1
     }
-
-    cleanup
 }
 
 # parse arguments

tests/unit/basic.tcl

@@ -148,12 +148,11 @@ start_server {tags {"basic"}} {
         r get novar2
     } {foobared}
 
-    test {SETNX will overwrite EXPIREing key} {
+    test {SETNX against volatile key} {
         r set x 10
         r expire x 10000
-        r setnx x 20
+        list [r setnx x 20] [r get x]
-        r get x
+    } {0 10}
-    } {20}
 
     test {EXISTS} {
         set res {}
@@ -362,10 +361,17 @@ start_server {tags {"basic"}} {
         list [r msetnx x1 xxx y2 yyy] [r get x1] [r get y2]
     } {1 xxx yyy}
 
-    test {MSETNX should remove all the volatile keys even on failure} {
+    test {STRLEN against non existing key} {
-        r mset x 1 y 2 z 3
+        r strlen notakey
-        r expire y 10000
+    } {0}
-        r expire z 10000
+
-        list [r msetnx x A y B z C] [r mget x y z]
+    test {STRLEN against integer} {
-    } {0 {1 {} {}}}
+        r set myinteger -555
+        r strlen myinteger
+    } {4}
+
+    test {STRLEN against plain string} {
+        r set mystring "foozzz0123456789 baz"
+        r strlen mystring
+    }
 }

tests/unit/cas.tcl

@@ -111,4 +111,25 @@ start_server {tags {"cas"}} {
         r ping
         r exec
     } {PONG}
+
+    test {WATCH will consider touched keys target of EXPIRE} {
+        r del x
+        r set x foo
+        r watch x
+        r expire x 10
+        r multi
+        r ping
+        r exec
+    } {}
+
+    test {WATCH will not consider touched expired keys} {
+        r del x
+        r set x foo
+        r expire x 2
+        r watch x
+        after 3000
+        r multi
+        r ping
+        r exec
+    } {PONG}
 }

tests/unit/expire.tcl

@@ -1,12 +1,13 @@
 start_server {tags {"expire"}} {
-    test {EXPIRE - don't set timeouts multiple times} {
+    test {EXPIRE - set timeouts multiple times} {
         r set x foobar
         set v1 [r expire x 5]
         set v2 [r ttl x]
         set v3 [r expire x 10]
         set v4 [r ttl x]
+        r expire x 4
         list $v1 $v2 $v3 $v4
-    } {1 5 0 5}
+    } {1 5 1 10}
 
     test {EXPIRE - It should be still possible to read 'x'} {
         r get x
@@ -19,13 +20,13 @@ start_server {tags {"expire"}} {
         } {{} 0}
     }
 
-    test {EXPIRE - Delete on write policy} {
+    test {EXPIRE - write on expire should work} {
         r del x
         r lpush x foo
         r expire x 1000
         r lpush x bar
         r lrange x 0 -1
-    } {bar}
+    } {bar foo}
 
     test {EXPIREAT - Check for EXPIRE alike behavior} {
         r del x
@@ -59,4 +60,15 @@ start_server {tags {"expire"}} {
         catch {r setex z -10 foo} e
         set _ $e
     } {*invalid expire*}
+
+    test {PERSIST can undo an EXPIRE} {
+        r set x foo
+        r expire x 50
+        list [r ttl x] [r persist x] [r ttl x] [r get x]
+    } {50 1 -1 foo}
+
+    test {PERSIST returns 0 against non existing or non volatile keys} {
+        r set x foo
+        list [r persist foo] [r persist nokeyatall]
+    } {0 0}
 }

tests/unit/other.tcl

@@ -46,23 +46,56 @@ start_server {} {
         set _ $err
     } {*invalid*}
 
-    if {![catch {package require sha1}]} {
+    tags {consistency} {
-        test {Check consistency of different data types after a reload} {
+        if {![catch {package require sha1}]} {
-            r flushdb
+            test {Check consistency of different data types after a reload} {
-            createComplexDataset r 10000
+                r flushdb
-            set sha1 [r debug digest]
+                createComplexDataset r 10000
-            r debug reload
+                set dump [csvdump r]
-            set sha1_after [r debug digest]
+                set sha1 [r debug digest]
-            expr {$sha1 eq $sha1_after}
+                r debug reload
-        } {1}
+                set sha1_after [r debug digest]
+                if {$sha1 eq $sha1_after} {
+                    set _ 1
+                } else {
+                    set newdump [csvdump r]
+                    puts "Consistency test failed!"
+                    puts "You can inspect the two dumps in /tmp/repldump*.txt"
 
-        test {Same dataset digest if saving/reloading as AOF?} {
+                    set fd [open /tmp/repldump1.txt w]
-            r bgrewriteaof
+                    puts $fd $dump
-            waitForBgrewriteaof r
+                    close $fd
-            r debug loadaof
+                    set fd [open /tmp/repldump2.txt w]
-            set sha1_after [r debug digest]
+                    puts $fd $newdump
-            expr {$sha1 eq $sha1_after}
+                    close $fd
-        } {1}
+
+                    set _ 0
+                }
+            } {1}
+
+            test {Same dataset digest if saving/reloading as AOF?} {
+                r bgrewriteaof
+                waitForBgrewriteaof r
+                r debug loadaof
+                set sha1_after [r debug digest]
+                if {$sha1 eq $sha1_after} {
+                    set _ 1
+                } else {
+                    set newdump [csvdump r]
+                    puts "Consistency test failed!"
+                    puts "You can inspect the two dumps in /tmp/aofdump*.txt"
+
+                    set fd [open /tmp/aofdump1.txt w]
+                    puts $fd $dump
+                    close $fd
+                    set fd [open /tmp/aofdump2.txt w]
+                    puts $fd $newdump
+                    close $fd
+
+                    set _ 0
+                }
+            } {1}
+        }
     }
 
     test {EXPIRES after a reload (snapshot + append only file)} {

tests/unit/type/list.tcl

@@ -5,6 +5,12 @@ start_server {
         "list-max-ziplist-entries" 256
     }
 } {
+    # We need a value larger than list-max-ziplist-value to make sure
+    # the list has the right encoding when it is swapped in again.
+    array set largevalue {}
+    set largevalue(ziplist) "hello"
+    set largevalue(linkedlist) [string repeat "hello" 4]
+
     test {LPUSH, RPUSH, LLENGTH, LINDEX - ziplist} {
         # first lpush then rpush
         assert_equal 1 [r lpush myziplist1 a]
@@ -28,28 +34,25 @@ start_server {
     }
 
     test {LPUSH, RPUSH, LLENGTH, LINDEX - regular list} {
-        # use a string of length 17 to ensure a regular list is used
-        set large_value "aaaaaaaaaaaaaaaaa"
-
         # first lpush then rpush
-        assert_equal 1 [r lpush mylist1 $large_value]
+        assert_equal 1 [r lpush mylist1 $largevalue(linkedlist)]
         assert_encoding linkedlist mylist1
         assert_equal 2 [r rpush mylist1 b]
         assert_equal 3 [r rpush mylist1 c]
         assert_equal 3 [r llen mylist1]
-        assert_equal $large_value [r lindex mylist1 0]
+        assert_equal $largevalue(linkedlist) [r lindex mylist1 0]
         assert_equal b [r lindex mylist1 1]
         assert_equal c [r lindex mylist1 2]
 
         # first rpush then lpush
-        assert_equal 1 [r rpush mylist2 $large_value]
+        assert_equal 1 [r rpush mylist2 $largevalue(linkedlist)]
         assert_encoding linkedlist mylist2
         assert_equal 2 [r lpush mylist2 b]
         assert_equal 3 [r lpush mylist2 c]
         assert_equal 3 [r llen mylist2]
         assert_equal c [r lindex mylist2 0]
         assert_equal b [r lindex mylist2 1]
-        assert_equal $large_value [r lindex mylist2 2]
+        assert_equal $largevalue(linkedlist) [r lindex mylist2 2]
     }
 
     test {DEL a list - ziplist} {
@@ -72,16 +75,14 @@ start_server {
 
     proc create_linkedlist {key entries} {
         r del $key
-        r rpush $key "aaaaaaaaaaaaaaaaa"
         foreach entry $entries { r rpush $key $entry }
-        assert_equal "aaaaaaaaaaaaaaaaa" [r lpop $key]
         assert_encoding linkedlist $key
     }
 
-    foreach type {ziplist linkedlist} {
+    foreach {type large} [array get largevalue] {
         test "BLPOP, BRPOP: single existing list - $type" {
             set rd [redis_deferring_client]
-            create_$type blist {a b c d}
+            create_$type blist "a b $large c d"
 
             $rd blpop blist 1
             assert_equal {blist a} [$rd read]
@@ -96,8 +97,8 @@ start_server {
 
         test "BLPOP, BRPOP: multiple existing lists - $type" {
             set rd [redis_deferring_client]
-            create_$type blist1 {a b c}
+            create_$type blist1 "a $large c"
-            create_$type blist2 {d e f}
+            create_$type blist2 "d $large f"
 
             $rd blpop blist1 blist2 1
             assert_equal {blist1 a} [$rd read]
@@ -117,7 +118,7 @@ start_server {
         test "BLPOP, BRPOP: second list has an entry - $type" {
             set rd [redis_deferring_client]
             r del blist1
-            create_$type blist2 {d e f}
+            create_$type blist2 "d $large f"
 
             $rd blpop blist1 blist2 1
             assert_equal {blist2 d} [$rd read]
@@ -179,26 +180,26 @@ start_server {
         assert_equal 0 [r llen xlist]
     }
 
-    foreach type {ziplist linkedlist} {
+    foreach {type large} [array get largevalue] {
         test "LPUSHX, RPUSHX - $type" {
-            create_$type xlist {b c}
+            create_$type xlist "$large c"
             assert_equal 3 [r rpushx xlist d]
             assert_equal 4 [r lpushx xlist a]
-            assert_equal {a b c d} [r lrange xlist 0 -1]
+            assert_equal "a $large c d" [r lrange xlist 0 -1]
         }
 
         test "LINSERT - $type" {
-            create_$type xlist {a b c d}
+            create_$type xlist "a $large c d"
             assert_equal 5 [r linsert xlist before c zz]
-            assert_equal {a b zz c d} [r lrange xlist 0 10]
+            assert_equal "a $large zz c d" [r lrange xlist 0 10]
             assert_equal 6 [r linsert xlist after c yy]
-            assert_equal {a b zz c yy d} [r lrange xlist 0 10]
+            assert_equal "a $large zz c yy d" [r lrange xlist 0 10]
             assert_equal 7 [r linsert xlist after d dd]
             assert_equal -1 [r linsert xlist after bad ddd]
-            assert_equal {a b zz c yy d dd} [r lrange xlist 0 10]
+            assert_equal "a $large zz c yy d dd" [r lrange xlist 0 10]
             assert_equal 8 [r linsert xlist before a aa]
             assert_equal -1 [r linsert xlist before bad aaa]
-            assert_equal {aa a b zz c yy d dd} [r lrange xlist 0 10]
+            assert_equal "aa a $large zz c yy d dd" [r lrange xlist 0 10]
 
             # check inserting integer encoded value
             assert_equal 9 [r linsert xlist before aa 42]
@@ -207,14 +208,14 @@ start_server {
     }
 
     test {LPUSHX, RPUSHX convert from ziplist to list} {
-        set large_value "aaaaaaaaaaaaaaaaa"
+        set large $largevalue(linkedlist)
 
         # convert when a large value is pushed
         create_ziplist xlist a
-        assert_equal 2 [r rpushx xlist $large_value]
+        assert_equal 2 [r rpushx xlist $large]
         assert_encoding linkedlist xlist
         create_ziplist xlist a
-        assert_equal 2 [r lpushx xlist $large_value]
+        assert_equal 2 [r lpushx xlist $large]
         assert_encoding linkedlist xlist
 
         # convert when the length threshold is exceeded
@@ -227,14 +228,14 @@ start_server {
     }
 
     test {LINSERT convert from ziplist to list} {
-        set large_value "aaaaaaaaaaaaaaaaa"
+        set large $largevalue(linkedlist)
 
         # convert when a large value is inserted
         create_ziplist xlist a
-        assert_equal 2 [r linsert xlist before a $large_value]
+        assert_equal 2 [r linsert xlist before a $large]
         assert_encoding linkedlist xlist
         create_ziplist xlist a
-        assert_equal 2 [r linsert xlist after a $large_value]
+        assert_equal 2 [r linsert xlist after a $large]
         assert_encoding linkedlist xlist
 
         # convert when the length threshold is exceeded
@@ -320,32 +321,38 @@ start_server {
         assert_error ERR* {r rpush mylist 0}
     }
 
-    foreach type {ziplist linkedlist} {
+    foreach {type large} [array get largevalue] {
         test "RPOPLPUSH base case - $type" {
             r del mylist1 mylist2
-            create_$type mylist1 {a b c d}
+            create_$type mylist1 "a $large c d"
             assert_equal d [r rpoplpush mylist1 mylist2]
             assert_equal c [r rpoplpush mylist1 mylist2]
-            assert_equal {a b} [r lrange mylist1 0 -1]
+            assert_equal "a $large" [r lrange mylist1 0 -1]
-            assert_equal {c d} [r lrange mylist2 0 -1]
+            assert_equal "c d" [r lrange mylist2 0 -1]
             assert_encoding ziplist mylist2
         }
 
         test "RPOPLPUSH with the same list as src and dst - $type" {
-            create_$type mylist {a b c}
+            create_$type mylist "a $large c"
-            assert_equal {a b c} [r lrange mylist 0 -1]
+            assert_equal "a $large c" [r lrange mylist 0 -1]
             assert_equal c [r rpoplpush mylist mylist]
-            assert_equal {c a b} [r lrange mylist 0 -1]
+            assert_equal "c a $large" [r lrange mylist 0 -1]
         }
 
-        foreach othertype {ziplist linkedlist} {
+        foreach {othertype otherlarge} [array get largevalue] {
             test "RPOPLPUSH with $type source and existing target $othertype" {
-                create_$type srclist {a b c d}
+                create_$type srclist "a b c $large"
-                create_$othertype dstlist {x}
+                create_$othertype dstlist "$otherlarge"
-                assert_equal d [r rpoplpush srclist dstlist]
+                assert_equal $large [r rpoplpush srclist dstlist]
                 assert_equal c [r rpoplpush srclist dstlist]
-                assert_equal {a b} [r lrange srclist 0 -1]
+                assert_equal "a b" [r lrange srclist 0 -1]
-                assert_equal {c d x} [r lrange dstlist 0 -1]
+                assert_equal "c $large $otherlarge" [r lrange dstlist 0 -1]
+
+                # When we rpoplpush'ed a large value, dstlist should be
+                # converted to the same encoding as srclist.
+                if {$type eq "linkedlist"} {
+                    assert_encoding linkedlist dstlist
+                }
             }
         }
     }
@@ -378,10 +385,10 @@ start_server {
         assert_equal {} [r rpoplpush srclist dstlist]
     } {}
 
-    foreach type {ziplist linkedlist} {
+    foreach {type large} [array get largevalue] {
         test "Basic LPOP/RPOP - $type" {
-            create_$type mylist {0 1 2}
+            create_$type mylist "$large 1 2"
-            assert_equal 0 [r lpop mylist]
+            assert_equal $large [r lpop mylist]
             assert_equal 2 [r rpop mylist]
             assert_equal 1 [r lpop mylist]
             assert_equal 0 [r llen mylist]
@@ -416,28 +423,28 @@ start_server {
         }
     }
 
-    foreach type {ziplist linkedlist} {
+    foreach {type large} [array get largevalue] {
         test "LRANGE basics - $type" {
-            create_$type mylist {0 1 2 3 4 5 6 7 8 9}
+            create_$type mylist "$large 1 2 3 4 5 6 7 8 9"
             assert_equal {1 2 3 4 5 6 7 8} [r lrange mylist 1 -2]
             assert_equal {7 8 9} [r lrange mylist -3 -1]
             assert_equal {4} [r lrange mylist 4 4]
         }
 
         test "LRANGE inverted indexes - $type" {
-            create_$type mylist {0 1 2 3 4 5 6 7 8 9}
+            create_$type mylist "$large 1 2 3 4 5 6 7 8 9"
             assert_equal {} [r lrange mylist 6 2]
         }
 
         test "LRANGE out of range indexes including the full list - $type" {
-            create_$type mylist {1 2 3}
+            create_$type mylist "$large 1 2 3"
-            assert_equal {1 2 3} [r lrange mylist -1000 1000]
+            assert_equal "$large 1 2 3" [r lrange mylist -1000 1000]
         }
 
         test "LRANGE out of range negative end index - $type" {
-            create_$type mylist {1 2 3}
+            create_$type mylist "$large 1 2 3"
-            assert_equal {1} [r lrange mylist 0 -3]
+            assert_equal $large [r lrange mylist 0 -4]
-            assert_equal {} [r lrange mylist 0 -4]
+            assert_equal {} [r lrange mylist 0 -5]
         }
     }
 
@@ -445,27 +452,28 @@ start_server {
         assert_equal {} [r lrange nosuchkey 0 1]
     }
 
-    foreach type {ziplist linkedlist} {
+    foreach {type large} [array get largevalue] {
         proc trim_list {type min max} {
+            upvar 1 large large
             r del mylist
-            create_$type mylist {1 2 3 4 5}
+            create_$type mylist "1 2 3 4 $large"
             r ltrim mylist $min $max
             r lrange mylist 0 -1
         }
 
         test "LTRIM basics - $type" {
-            assert_equal {1} [trim_list $type 0 0]
+            assert_equal "1" [trim_list $type 0 0]
-            assert_equal {1 2} [trim_list $type 0 1]
+            assert_equal "1 2" [trim_list $type 0 1]
-            assert_equal {1 2 3} [trim_list $type 0 2]
+            assert_equal "1 2 3" [trim_list $type 0 2]
-            assert_equal {2 3} [trim_list $type 1 2]
+            assert_equal "2 3" [trim_list $type 1 2]
-            assert_equal {2 3 4 5} [trim_list $type 1 -1]
+            assert_equal "2 3 4 $large" [trim_list $type 1 -1]
-            assert_equal {2 3 4} [trim_list $type 1 -2]
+            assert_equal "2 3 4" [trim_list $type 1 -2]
-            assert_equal {4 5} [trim_list $type -2 -1]
+            assert_equal "4 $large" [trim_list $type -2 -1]
-            assert_equal {5} [trim_list $type -1 -1]
+            assert_equal "$large" [trim_list $type -1 -1]
-            assert_equal {1 2 3 4 5} [trim_list $type -5 -1]
+            assert_equal "1 2 3 4 $large" [trim_list $type -5 -1]
-            assert_equal {1 2 3 4 5} [trim_list $type -10 10]
+            assert_equal "1 2 3 4 $large" [trim_list $type -10 10]
-            assert_equal {1 2 3 4 5} [trim_list $type 0 5]
+            assert_equal "1 2 3 4 $large" [trim_list $type 0 5]
-            assert_equal {1 2 3 4 5} [trim_list $type 0 10]
+            assert_equal "1 2 3 4 $large" [trim_list $type 0 10]
         }
 
         test "LTRIM out of range negative end index - $type" {
@@ -478,20 +486,19 @@ start_server {
                 set mylist {}
                 set startlen 32
                 r del mylist
+
+                # Start with the large value to ensure the
+                # right encoding is used.
+                r rpush mylist $large
+                lappend mylist $large
+
                 for {set i 0} {$i < $startlen} {incr i} {
                     set str [randomInt 9223372036854775807]
                     r rpush mylist $str
                     lappend mylist $str
                 }
 
-                # do a push/pop of a large value to convert to a real list
+                for {set i 0} {$i < 1000} {incr i} {
-                if {$type eq "list"} {
-                    r rpush mylist "aaaaaaaaaaaaaaaaa"
-                    r rpop mylist
-                    assert_encoding linkedlist mylist
-                }
-
-                for {set i 0} {$i < 10000} {incr i} {
                     set min [expr {int(rand()*$startlen)}]
                     set max [expr {$min+int(rand()*$startlen)}]
                     set mylist [lrange $mylist $min $max]
@@ -508,12 +515,12 @@ start_server {
         }
     }
 
-    foreach type {ziplist linkedlist} {
+    foreach {type large} [array get largevalue] {
         test "LSET - $type" {
-            create_$type mylist {99 98 97 96 95}
+            create_$type mylist "99 98 $large 96 95"
             r lset mylist 1 foo
             r lset mylist -1 bar
-            assert_equal {99 foo 97 96 bar} [r lrange mylist 0 -1]
+            assert_equal "99 foo $large 96 bar" [r lrange mylist 0 -1]
         }
 
         test "LSET out of range index - $type" {
@@ -530,38 +537,38 @@ start_server {
         assert_error ERR*value* {r lset nolist 0 foo}
     }
 
-    foreach type {ziplist linkedlist} {
+    foreach {type e} [array get largevalue] {
         test "LREM remove all the occurrences - $type" {
-            create_$type mylist {foo bar foobar foobared zap bar test foo}
+            create_$type mylist "$e foo bar foobar foobared zap bar test foo"
             assert_equal 2 [r lrem mylist 0 bar]
-            assert_equal {foo foobar foobared zap test foo} [r lrange mylist 0 -1]
+            assert_equal "$e foo foobar foobared zap test foo" [r lrange mylist 0 -1]
         }
 
         test "LREM remove the first occurrence - $type" {
             assert_equal 1 [r lrem mylist 1 foo]
-            assert_equal {foobar foobared zap test foo} [r lrange mylist 0 -1]
+            assert_equal "$e foobar foobared zap test foo" [r lrange mylist 0 -1]
         }
 
         test "LREM remove non existing element - $type" {
             assert_equal 0 [r lrem mylist 1 nosuchelement]
-            assert_equal {foobar foobared zap test foo} [r lrange mylist 0 -1]
+            assert_equal "$e foobar foobared zap test foo" [r lrange mylist 0 -1]
         }
 
         test "LREM starting from tail with negative count - $type" {
-            create_$type mylist {foo bar foobar foobared zap bar test foo foo}
+            create_$type mylist "$e foo bar foobar foobared zap bar test foo foo"
             assert_equal 1 [r lrem mylist -1 bar]
-            assert_equal {foo bar foobar foobared zap test foo foo} [r lrange mylist 0 -1]
+            assert_equal "$e foo bar foobar foobared zap test foo foo" [r lrange mylist 0 -1]
         }
 
         test "LREM starting from tail with negative count (2) - $type" {
             assert_equal 2 [r lrem mylist -2 foo]
-            assert_equal {foo bar foobar foobared zap test} [r lrange mylist 0 -1]
+            assert_equal "$e foo bar foobar foobared zap test" [r lrange mylist 0 -1]
         }
 
         test "LREM deleting objects that may be int encoded - $type" {
-            create_$type myotherlist {1 2 3}
+            create_$type myotherlist "$e 1 2 3"
             assert_equal 1 [r lrem myotherlist 1 2]
-            assert_equal 2 [r llen myotherlist]
+            assert_equal 3 [r llen myotherlist]
         }
     }
 }

tests/unit/type/zset.tcl

@@ -433,6 +433,42 @@ start_server {tags {"zset"}} {
         list [r zinterstore zsetc 2 zseta zsetb aggregate max] [r zrange zsetc 0 -1 withscores]
     } {2 {b 2 c 3}}
     
+    foreach cmd {ZUNIONSTORE ZINTERSTORE} {
+        test "$cmd with +inf/-inf scores" {
+            r del zsetinf1 zsetinf2
+
+            r zadd zsetinf1 +inf key
+            r zadd zsetinf2 +inf key
+            r $cmd zsetinf3 2 zsetinf1 zsetinf2
+            assert_equal inf [r zscore zsetinf3 key]
+
+            r zadd zsetinf1 -inf key
+            r zadd zsetinf2 +inf key
+            r $cmd zsetinf3 2 zsetinf1 zsetinf2
+            assert_equal 0 [r zscore zsetinf3 key]
+
+            r zadd zsetinf1 +inf key
+            r zadd zsetinf2 -inf key
+            r $cmd zsetinf3 2 zsetinf1 zsetinf2
+            assert_equal 0 [r zscore zsetinf3 key]
+
+            r zadd zsetinf1 -inf key
+            r zadd zsetinf2 -inf key
+            r $cmd zsetinf3 2 zsetinf1 zsetinf2
+            assert_equal -inf [r zscore zsetinf3 key]
+        }
+
+        test "$cmd with NaN weights" {
+            r del zsetinf1 zsetinf2
+
+            r zadd zsetinf1 1.0 key
+            r zadd zsetinf2 1.0 key
+            assert_error "*weight value is not a double*" {
+                r $cmd zsetinf3 2 zsetinf1 zsetinf2 weights nan nan
+            }
+        }
+    }
+
     tags {"slow"} {
         test {ZSETs skiplist implementation backlink consistency test} {
             set diff 0
@@ -477,22 +513,16 @@ start_server {tags {"zset"}} {
         } {}
     }
 
-    test {ZSET element can't be set to nan with ZADD} {
+    test {ZSET element can't be set to NaN with ZADD} {
-        set e {}
+        assert_error "*not a double*" {r zadd myzset nan abc}
-        catch {r zadd myzset nan abc} e
+    }
-        set _ $e
-    } {*Not A Number*}
 
-    test {ZSET element can't be set to nan with ZINCRBY} {
+    test {ZSET element can't be set to NaN with ZINCRBY} {
-        set e {}
+        assert_error "*not a double*" {r zadd myzset nan abc}
-        catch {r zincrby myzset nan abc} e
+    }
-        set _ $e
-    } {*Not A Number*}
 
-    test {ZINCRBY calls leading to Nan are refused} {
+    test {ZINCRBY calls leading to NaN result in error} {
-        set e {}
         r zincrby myzset +inf abc
-        catch {r zincrby myzset -inf abc} e
+        assert_error "*NaN*" {r zincrby myzset -inf abc}
-        set _ $e
+    }
-    } {*Not A Number*}
 }