Redis 2.2.9

00-RELEASENOTES

@@ -12,6 +12,38 @@ for 2.0.
 CHANGELOG
 ---------
 
+What's new in Redis 2.2.8
+=========================
+
+Redis 2.2.9 is an important bugfix release:
+
+* A Slave used to rewrite the AOF log after a sync with the master, but the
+  rewrite was performed at the wrong time, causing inconsistencies in the
+  AOF file generated.
+* [LR]LPUSH replication in presence of BRPOPLPUSH was broken. Fixed.
+* Memory leak in BRPOPLPUSH fixed.
+* Pub/Sub bug resulting in random crashes fixed.
+
+What's new in Redis 2.2.8
+=========================
+
+* A new form of dict.c (hash table implementation) iterator that performs less
+  copy-on-write of pages, introduced in Redis 2.2.7, caused ZINTERSTORE,
+  ZUNIONSTORE, SINTER, SINTERSTORE commands to behave in the wrong way.
+  This bug is now fixed.
+* Print version info before running the test with 'make test'. This is mainly
+  useful for the Continuous Integration system we run.
+* Fix for DEBUG DIGEST, key may expire on lookup, producing the wrong result.
+* Replication with expire test modified to produce no or less false failures.
+* Fixed Z*STORE when dealing with intsets, regression test added.
+
+What's new in Redis 2.2.7
+=========================
+
+* Fixed bug #543-2 (the issue was reopened with a completely different report)
+  that caused Redis to randomly crash on list push performed against lists
+  with other clients blocked with BLPOP (or variants).
+
 What's new in Redis 2.2.6
 =========================
 

src/debug.c

@@ -80,7 +80,7 @@ void computeDatasetDigest(unsigned char *final) {
         redisDb *db = server.db+j;
 
         if (dictSize(db->dict) == 0) continue;
-        di = dictGetIterator(db->dict);
+        di = dictGetSafeIterator(db->dict);
 
         /* hash the DB id, so the same dataset moved in a different
          * DB will lead to a different digest */
@@ -101,6 +101,11 @@ void computeDatasetDigest(unsigned char *final) {
 
             /* Make sure the key is loaded if VM is active */
             o = lookupKeyRead(db,keyobj);
+            if (o == NULL) {
+                /* Key expired on lookup? Try the next one. */
+                decrRefCount(keyobj);
+                continue;
+            }
 
             aux = htonl(o->type);
             mixDigest(digest,&aux,sizeof(aux));

src/dict.c

@@ -244,9 +244,9 @@ int dictRehashMilliseconds(dict *d, int ms) {
 }
 
 /* This function performs just a step of rehashing, and only if there are
- * not iterators bound to our hash table. When we have iterators in the middle
- * of a rehashing we can't mess with the two hash tables otherwise some element
- * can be missed or duplicated.
+ * no safe iterators bound to our hash table. When we have iterators in the
+ * middle of a rehashing we can't mess with the two hash tables otherwise
+ * some element can be missed or duplicated.
  *
  * This function is called by common lookup or update operations in the
  * dictionary so that the hash table automatically migrates from H1 to H2
@@ -423,17 +423,26 @@ dictIterator *dictGetIterator(dict *d)
     iter->d = d;
     iter->table = 0;
     iter->index = -1;
+    iter->safe = 0;
     iter->entry = NULL;
     iter->nextEntry = NULL;
     return iter;
 }
 
+dictIterator *dictGetSafeIterator(dict *d) {
+    dictIterator *i = dictGetIterator(d);
+
+    i->safe = 1;
+    return i;
+}
+
 dictEntry *dictNext(dictIterator *iter)
 {
     while (1) {
         if (iter->entry == NULL) {
             dictht *ht = &iter->d->ht[iter->table];
-            if (iter->index == -1 && iter->table == 0) iter->d->iterators++;
+            if (iter->safe && iter->index == -1 && iter->table == 0)
+                iter->d->iterators++;
             iter->index++;
             if (iter->index >= (signed) ht->size) {
                 if (dictIsRehashing(iter->d) && iter->table == 0) {
@@ -460,7 +469,8 @@ dictEntry *dictNext(dictIterator *iter)
 
 void dictReleaseIterator(dictIterator *iter)
 {
-    if (!(iter->index == -1 && iter->table == 0)) iter->d->iterators--;
+    if (iter->safe && !(iter->index == -1 && iter->table == 0))
+        iter->d->iterators--;
     zfree(iter);
 }
 

src/dict.h

@@ -74,10 +74,13 @@ typedef struct dict {
     int iterators; /* number of iterators currently running */
 } dict;
 
+/* If safe is set to 1 this is a safe iteartor, that means, you can call
+ * dictAdd, dictFind, and other functions against the dictionary even while
+ * iterating. Otherwise it is a non safe iterator, and only dictNext()
+ * should be called while iterating. */
 typedef struct dictIterator {
     dict *d;
-    int table;
-    int index;
+    int table, index, safe;
     dictEntry *entry, *nextEntry;
 } dictIterator;
 
@@ -132,6 +135,7 @@ dictEntry * dictFind(dict *d, const void *key);
 void *dictFetchValue(dict *d, const void *key);
 int dictResize(dict *d);
 dictIterator *dictGetIterator(dict *d);
+dictIterator *dictGetSafeIterator(dict *d);
 dictEntry *dictNext(dictIterator *iter);
 void dictReleaseIterator(dictIterator *iter);
 dictEntry *dictGetRandomKey(dict *d);

src/networking.c

@@ -416,7 +416,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 
     cfd = anetTcpAccept(server.neterr, fd, cip, &cport);
     if (cfd == AE_ERR) {
-        redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
+        redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
         return;
     }
     redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
@@ -431,7 +431,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 
     cfd = anetUnixAccept(server.neterr, fd);
     if (cfd == AE_ERR) {
-        redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
+        redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
         return;
     }
     redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);
@@ -872,3 +872,28 @@ void getClientsMaxBuffers(unsigned long *longest_output_list,
     *biggest_input_buffer = bib;
 }
 
+void rewriteClientCommandVector(redisClient *c, int argc, ...) {
+    va_list ap;
+    int j;
+    robj **argv; /* The new argument vector */
+
+    argv = zmalloc(sizeof(robj*)*argc);
+    va_start(ap,argc);
+    for (j = 0; j < argc; j++) {
+        robj *a;
+        
+        a = va_arg(ap, robj*);
+        argv[j] = a;
+        incrRefCount(a);
+    }
+    /* We free the objects in the original vector at the end, so we are
+     * sure that if the same objects are reused in the new vector the
+     * refcount gets incremented before it gets decremented. */
+    for (j = 0; j < c->argc; j++) decrRefCount(c->argv[j]);
+    zfree(c->argv);
+    /* Replace argv and argc with our new versions. */
+    c->argv = argv;
+    c->argc = argc;
+    va_end(ap);
+}
+

src/pubsub.c

@@ -141,7 +141,7 @@ int pubsubUnsubscribePattern(redisClient *c, robj *pattern, int notify) {
 /* Unsubscribe from all the channels. Return the number of channels the
  * client was subscribed from. */
 int pubsubUnsubscribeAllChannels(redisClient *c, int notify) {
-    dictIterator *di = dictGetIterator(c->pubsub_channels);
+    dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
     dictEntry *de;
     int count = 0;
 

src/redis.h

@@ -661,6 +661,7 @@ void addReplyMultiBulkLen(redisClient *c, long length);
 void *dupClientReplyValue(void *o);
 void getClientsMaxBuffers(unsigned long *longest_output_list,
                           unsigned long *biggest_input_buffer);
+void rewriteClientCommandVector(redisClient *c, int argc, ...);
 
 #ifdef __GNUC__
 void addReplyErrorFormat(redisClient *c, const char *fmt, ...)

src/replication.c

@@ -390,6 +390,8 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
         server.master->authenticated = 1;
         server.replstate = REDIS_REPL_CONNECTED;
         redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
+        /* Rewrite the AOF file now that the dataset changed. */
+        if (server.appendonly) rewriteAppendOnlyFileBackground();
     }
 }
 
@@ -519,7 +521,6 @@ void replicationCron(void) {
         redisLog(REDIS_NOTICE,"Connecting to MASTER...");
         if (syncWithMaster() == REDIS_OK) {
             redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started: SYNC sent");
-            if (server.appendonly) rewriteAppendOnlyFileBackground();
         }
     }
     

src/t_list.c

@@ -635,7 +635,9 @@ void lremCommand(redisClient *c) {
  * as well. This command was originally proposed by Ezra Zygmuntowicz.
  */
 
-void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value) {
+void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, robj *dstobj, robj *value) {
+    robj *aux;
+
     if (!handleClientsWaitingListPush(c,dstkey,value)) {
         /* Create the list if the key does not exist */
         if (!dstobj) {
@@ -643,9 +645,25 @@ void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value
             dbAdd(c->db,dstkey,dstobj);
         } else {
             touchWatchedKey(c->db,dstkey);
-            server.dirty++;
         }
         listTypePush(dstobj,value,REDIS_HEAD);
+        /* If we are pushing as a result of LPUSH against a key
+         * watched by BLPOPLPUSH, we need to rewrite the command vector.
+         * But if this is called directly by RPOPLPUSH (either directly
+         * or via a BRPOPLPUSH where the popped list exists)
+         * we should replicate the BRPOPLPUSH command itself. */
+        if (c != origclient) {
+            aux = createStringObject("LPUSH",5);
+            rewriteClientCommandVector(origclient,3,aux,dstkey,value);
+            decrRefCount(aux);
+        } else {
+            /* Make sure to always use RPOPLPUSH in the replication / AOF,
+             * even if the original command was BRPOPLPUSH. */
+            aux = createStringObject("RPOPLPUSH",9);
+            rewriteClientCommandVector(origclient,3,aux,c->argv[1],c->argv[2]);
+            decrRefCount(aux);
+        }
+        server.dirty++;
     }
 
     /* Always send the pushed value to the client. */
@@ -661,16 +679,22 @@ void rpoplpushCommand(redisClient *c) {
         addReply(c,shared.nullbulk);
     } else {
         robj *dobj = lookupKeyWrite(c->db,c->argv[2]);
+        robj *touchedkey = c->argv[1];
+
         if (dobj && checkType(c,dobj,REDIS_LIST)) return;
         value = listTypePop(sobj,REDIS_TAIL);
-        rpoplpushHandlePush(c,c->argv[2],dobj,value);
+        /* We saved touched key, and protect it, since rpoplpushHandlePush
+         * may change the client command argument vector. */
+        incrRefCount(touchedkey);
+        rpoplpushHandlePush(c,c,c->argv[2],dobj,value);
 
         /* listTypePop returns an object with its refcount incremented */
         decrRefCount(value);
 
         /* Delete the source list when it is empty */
-        if (listTypeLength(sobj) == 0) dbDelete(c->db,c->argv[1]);
-        touchWatchedKey(c->db,c->argv[1]);
+        if (listTypeLength(sobj) == 0) dbDelete(c->db,touchedkey);
+        touchWatchedKey(c->db,touchedkey);
+        decrRefCount(touchedkey);
         server.dirty++;
     }
 }
@@ -772,6 +796,7 @@ void unblockClientWaitingData(redisClient *c) {
     /* Cleanup the client structure */
     zfree(c->bpop.keys);
     c->bpop.keys = NULL;
+    if (c->bpop.target) decrRefCount(c->bpop.target);
     c->bpop.target = NULL;
     c->flags &= ~REDIS_BLOCKED;
     c->flags |= REDIS_UNBLOCKED;
@@ -815,26 +840,27 @@ int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
         receiver = ln->value;
         dstkey = receiver->bpop.target;
 
+        /* Protect receiver->bpop.target, that will be freed by
+         * the next unblockClientWaitingData() call. */
+        if (dstkey) incrRefCount(dstkey);
         /* This should remove the first element of the "clients" list. */
         unblockClientWaitingData(receiver);
-        redisAssert(ln != listFirst(clients));
 
         if (dstkey == NULL) {
             /* BRPOP/BLPOP */
             addReplyMultiBulkLen(receiver,2);
             addReplyBulk(receiver,key);
             addReplyBulk(receiver,ele);
-            return 1;
+            return 1; /* Serve just the first client as in B[RL]POP semantics */
         } else {
             /* BRPOPLPUSH, note that receiver->db is always equal to c->db. */
             dstobj = lookupKeyWrite(receiver->db,dstkey);
-            if (dstobj && checkType(receiver,dstobj,REDIS_LIST)) {
-                decrRefCount(dstkey);
-            } else {
-                rpoplpushHandlePush(receiver,dstkey,dstobj,ele);
+            if (!(dstobj && checkType(receiver,dstobj,REDIS_LIST))) {
+                rpoplpushHandlePush(c,receiver,dstkey,dstobj,ele);
                 decrRefCount(dstkey);
                 return 1;
             }
+            decrRefCount(dstkey);
         }
     }
 

src/t_set.c

@@ -325,7 +325,7 @@ void scardCommand(redisClient *c) {
 }
 
 void spopCommand(redisClient *c) {
-    robj *set, *ele;
+    robj *set, *ele, *aux;
     int64_t llele;
     int encoding;
 
@@ -341,16 +341,11 @@ void spopCommand(redisClient *c) {
         setTypeRemove(set,ele);
     }
 
-    /* Change argv to replicate as SREM */
-    c->argc = 3;
-    c->argv = zrealloc(c->argv,sizeof(robj*)*(c->argc));
-
-    /* Overwrite SREM with SPOP (same length) */
-    redisAssert(sdslen(c->argv[0]->ptr) == 4);
-    memcpy(c->argv[0]->ptr, "SREM", 4);
-
-    /* Popped element already has incremented refcount */
-    c->argv[2] = ele;
+    /* Replicate/AOF this command as an SREM operation */
+    aux = createStringObject("SREM",4);
+    rewriteClientCommandVector(c,3,aux,c->argv[1],ele);
+    decrRefCount(ele);
+    decrRefCount(aux);
 
     addReplyBulk(c,ele);
     if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
@@ -433,6 +428,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum,
     si = setTypeInitIterator(sets[0]);
     while((encoding = setTypeNext(si,&eleobj,&intobj)) != -1) {
         for (j = 1; j < setnum; j++) {
+            if (sets[j] == sets[0]) continue;
             if (encoding == REDIS_ENCODING_INTSET) {
                 /* intset with intset is simple... and fast */
                 if (sets[j]->encoding == REDIS_ENCODING_INTSET &&

src/t_zset.c

@@ -615,6 +615,10 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
             if (obj->type == REDIS_ZSET) {
                 src[i].dict = ((zset*)obj->ptr)->dict;
             } else if (obj->type == REDIS_SET) {
+                if (obj->encoding == REDIS_ENCODING_INTSET)
+                    setTypeConvert(obj, REDIS_ENCODING_HT);
+
+                redisAssert(obj->encoding == REDIS_ENCODING_HT);
                 src[i].dict = (obj->ptr);
             } else {
                 zfree(src);
@@ -682,7 +686,18 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
 
                 score = src[0].weight * zunionInterDictValue(de);
                 for (j = 1; j < setnum; j++) {
-                    dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
+                    dictEntry *other;
+
+                    /* If it's the same dictionary don't lookup as we are not
+                     * in the context of a safe iterator. It's the same
+                     * dictionary so we are sure the element is inside.
+                     * This happens on SINTERSTORE dest 2 mykey mykey. */
+                    if (src[j].dict == src[0].dict) {
+                        other = de;
+                    } else {
+                        other = dictFind(src[j].dict,dictGetEntryKey(de));
+                    }
+
                     if (other) {
                         value = src[j].weight * zunionInterDictValue(other);
                         zunionInterAggregate(&score,value,aggregate);
@@ -720,10 +735,19 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                 /* because the zsets are sorted by size, its only possible
                  * for sets at larger indices to hold this entry */
                 for (j = (i+1); j < setnum; j++) {
-                    dictEntry *other = dictFind(src[j].dict,dictGetEntryKey(de));
-                    if (other) {
-                        value = src[j].weight * zunionInterDictValue(other);
+                    /* It is not safe to access the zset we are
+                     * iterating, so explicitly check for equal object. */
+                    if (src[j].dict == src[i].dict) {
+                        value = src[i].weight * zunionInterDictValue(de);
                         zunionInterAggregate(&score,value,aggregate);
+                    } else {
+                        dictEntry *other;
+
+                        other = dictFind(src[j].dict,dictGetEntryKey(de));
+                        if (other) {
+                            value = src[j].weight * zunionInterDictValue(other);
+                            zunionInterAggregate(&score,value,aggregate);
+                        }
                     }
                 }
 

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "2.2.6"
+#define REDIS_VERSION "2.2.9"

tests/integration/replication.tcl

@@ -6,6 +6,24 @@ start_server {tags {"repl"}} {
             s -1 role
         } {slave}
 
+        test {BRPOPLPUSH replication, when blocking against empty list} {
+            set rd [redis_deferring_client]
+            $rd brpoplpush a b 5
+            r lpush a foo
+            after 1000
+            assert_equal [r debug digest] [r -1 debug digest]
+        }
+
+        test {BRPOPLPUSH replication, list exists} {
+            set rd [redis_deferring_client]
+            r lpush c 1
+            r lpush c 2
+            r lpush c 3
+            $rd brpoplpush c d 5
+            after 1000
+            assert_equal [r debug digest] [r -1 debug digest]
+        }
+
         test {MASTER and SLAVE dataset should be identical after complex ops} {
             createComplexDataset r 10000
             after 500
@@ -27,6 +45,8 @@ start_server {tags {"repl"}} {
         test {MASTER and SLAVE consistency with expire} {
             createComplexDataset r 50000 useexpire
             after 4000 ;# Make sure everything expired before taking the digest
+            r keys *   ;# Force DEL syntesizing to slave
+            after 1000 ;# Wait another second. Now everything should be fine.
             if {[r debug digest] ne [r -1 debug digest]} {
                 set csv1 [csvdump r]
                 set csv2 [csvdump {r -1}]

tests/test_helper.tcl

@@ -108,6 +108,7 @@ proc cleanup {} {
 }
 
 proc execute_everything {} {
+    execute_tests "unit/printver"
     execute_tests "unit/auth"
     execute_tests "unit/protocol"
     execute_tests "unit/basic"

tests/unit/printver.tcl

@@ -0,0 +1,6 @@
+start_server {} {
+    set i [r info]
+    regexp {redis_version:(.*?)\r\n} $i - version
+    regexp {redis_git_sha1:(.*?)\r\n} $i - sha1
+    puts "Testing Redis version $version ($sha1)"
+}

tests/unit/type/list.tcl

@@ -262,6 +262,14 @@ start_server {
         r exec
     } {foo bar {} {} {bar foo}}
 
+    test {BRPOPLPUSH timeout} {
+      set rd [redis_deferring_client]
+
+      $rd brpoplpush foo_list bar_list 1
+      after 2000
+      $rd read
+    } {}
+
     foreach {pop} {BLPOP BRPOP} {
         test "$pop: with single empty list argument" {
             set rd [redis_deferring_client]

tests/unit/type/zset.tcl

@@ -484,6 +484,13 @@ start_server {tags {"zset"}} {
     test {ZINTERSTORE with AGGREGATE MAX} {
         list [r zinterstore zsetc 2 zseta zsetb aggregate max] [r zrange zsetc 0 -1 withscores]
     } {2 {b 2 c 3}}
+
+    test {ZINTERSTORE regression with two sets, intset+hashtable} {
+        r del seta setb setc
+        r sadd set1 a
+        r sadd set2 10
+        r zinterstore set3 2 set1 set2
+    } {0}
     
     foreach cmd {ZUNIONSTORE ZINTERSTORE} {
         test "$cmd with +inf/-inf scores" {