Merge remote-tracking branch 'origin/unstable' into unstable

diff --git a/src/aof.c b/src/aof.c
index 4a463bde0664b34e39ceb5a9da9fe96bc485bdc5..bfa8163d3fd57e32d4aa2bf6b90e1ed5f840f3cc 100644 (file)
--- a/src/aof.c
+++ b/src/aof.c
@@ -422,8 +422,236 @@ int rioWriteBulkObject(rio *r, robj *obj) {
     }
 }
 
+/* Emit the commands needed to rebuild a list object.
+ * The function returns 0 on error, 1 on success. */
+int rewriteListObject(rio *r, robj *key, robj *o) {
+    long long count = 0, items = listTypeLength(o);
+
+    if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+        unsigned char *zl = o->ptr;
+        unsigned char *p = ziplistIndex(zl,0);
+        unsigned char *vstr;
+        unsigned int vlen;
+        long long vlong;
+
+        while(ziplistGet(p,&vstr,&vlen,&vlong)) {
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
+                if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (vstr) {
+                if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
+            } else {
+                if (rioWriteBulkLongLong(r,vlong) == 0) return 0;
+            }
+            p = ziplistNext(zl,p);
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+    } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
+        list *list = o->ptr;
+        listNode *ln;
+        listIter li;
+
+        listRewind(list,&li);
+        while((ln = listNext(&li))) {
+            robj *eleobj = listNodeValue(ln);
+
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
+                if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (rioWriteBulkObject(r,eleobj) == 0) return 0;
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+    } else {
+        redisPanic("Unknown list encoding");
+    }
+    return 1;
+}
+
+/* Emit the commands needed to rebuild a set object.
+ * The function returns 0 on error, 1 on success. */
+int rewriteSetObject(rio *r, robj *key, robj *o) {
+    long long count = 0, items = setTypeSize(o);
+
+    if (o->encoding == REDIS_ENCODING_INTSET) {
+        int ii = 0;
+        int64_t llval;
+
+        while(intsetGet(o->ptr,ii++,&llval)) {
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
+                if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (rioWriteBulkLongLong(r,llval) == 0) return 0;
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+    } else if (o->encoding == REDIS_ENCODING_HT) {
+        dictIterator *di = dictGetIterator(o->ptr);
+        dictEntry *de;
+
+        while((de = dictNext(di)) != NULL) {
+            robj *eleobj = dictGetKey(de);
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
+                if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (rioWriteBulkObject(r,eleobj) == 0) return 0;
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+        dictReleaseIterator(di);
+    } else {
+        redisPanic("Unknown set encoding");
+    }
+    return 1;
+}
+
+/* Emit the commands needed to rebuild a sorted set object.
+ * The function returns 0 on error, 1 on success. */
+int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
+    long long count = 0, items = zsetLength(o);
+
+    if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+        unsigned char *zl = o->ptr;
+        unsigned char *eptr, *sptr;
+        unsigned char *vstr;
+        unsigned int vlen;
+        long long vll;
+        double score;
+
+        eptr = ziplistIndex(zl,0);
+        redisAssert(eptr != NULL);
+        sptr = ziplistNext(zl,eptr);
+        redisAssert(sptr != NULL);
+
+        while (eptr != NULL) {
+            redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
+            score = zzlGetScore(sptr);
+
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
+                if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (rioWriteBulkDouble(r,score) == 0) return 0;
+            if (vstr != NULL) {
+                if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
+            } else {
+                if (rioWriteBulkLongLong(r,vll) == 0) return 0;
+            }
+            zzlNext(zl,&eptr,&sptr);
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+    } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
+        zset *zs = o->ptr;
+        dictIterator *di = dictGetIterator(zs->dict);
+        dictEntry *de;
+
+        while((de = dictNext(di)) != NULL) {
+            robj *eleobj = dictGetKey(de);
+            double *score = dictGetVal(de);
+
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
+                if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (rioWriteBulkDouble(r,*score) == 0) return 0;
+            if (rioWriteBulkObject(r,eleobj) == 0) return 0;
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+        dictReleaseIterator(di);
+    } else {
+        redisPanic("Unknown sorted zset encoding");
+    }
+    return 1;
+}
+
+/* Emit the commands needed to rebuild a hash object.
+ * The function returns 0 on error, 1 on success. */
+int rewriteHashObject(rio *r, robj *key, robj *o) {
+    long long count = 0, items = hashTypeLength(o);
+
+    if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+        unsigned char *p = zipmapRewind(o->ptr);
+        unsigned char *field, *val;
+        unsigned int flen, vlen;
+
+        while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
+                if (rioWriteBulkString(r,"HMSET",5) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (rioWriteBulkString(r,(char*)field,flen) == 0) return 0;
+            if (rioWriteBulkString(r,(char*)val,vlen) == 0) return 0;
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+    } else {
+        dictIterator *di = dictGetIterator(o->ptr);
+        dictEntry *de;
+
+        while((de = dictNext(di)) != NULL) {
+            robj *field = dictGetKey(de);
+            robj *val = dictGetVal(de);
+
+            if (count == 0) {
+                int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
+                    REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+
+                if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
+                if (rioWriteBulkString(r,"HMSET",5) == 0) return 0;
+                if (rioWriteBulkObject(r,key) == 0) return 0;
+            }
+            if (rioWriteBulkObject(r,field) == 0) return 0;
+            if (rioWriteBulkObject(r,val) == 0) return 0;
+            if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+            items--;
+        }
+        dictReleaseIterator(di);
+    }
+    return 1;
+}
+
 /* Write a sequence of commands able to fully rebuild the dataset into
- * "filename". Used both by REWRITEAOF and BGREWRITEAOF. */
+ * "filename". Used both by REWRITEAOF and BGREWRITEAOF.
+ *
+ * In order to minimize the number of commands needed in the rewritten
+ * log Redis uses variadic commands when possible, such as RPUSH, SADD
+ * and ZADD. However at max REDIS_AOFREWRITE_ITEMS_PER_CMD items per time
+ * are inserted using a single command. */
 int rewriteAppendOnlyFile(char *filename) {
     dictIterator *di = NULL;
     dictEntry *de;
@@ -479,151 +707,13 @@ int rewriteAppendOnlyFile(char *filename) {
                 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
                 if (rioWriteBulkObject(&aof,o) == 0) goto werr;
             } else if (o->type == REDIS_LIST) {
-                /* Emit the RPUSHes needed to rebuild the list */
-                char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
-                if (o->encoding == REDIS_ENCODING_ZIPLIST) {
-                    unsigned char *zl = o->ptr;
-                    unsigned char *p = ziplistIndex(zl,0);
-                    unsigned char *vstr;
-                    unsigned int vlen;
-                    long long vlong;
-
-                    while(ziplistGet(p,&vstr,&vlen,&vlong)) {
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (vstr) {
-                            if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0)
-                                goto werr;
-                        } else {
-                            if (rioWriteBulkLongLong(&aof,vlong) == 0)
-                                goto werr;
-                        }
-                        p = ziplistNext(zl,p);
-                    }
-                } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
-                    list *list = o->ptr;
-                    listNode *ln;
-                    listIter li;
-
-                    listRewind(list,&li);
-                    while((ln = listNext(&li))) {
-                        robj *eleobj = listNodeValue(ln);
-
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr;
-                    }
-                } else {
-                    redisPanic("Unknown list encoding");
-                }
+                if (rewriteListObject(&aof,&key,o) == 0) goto werr;
             } else if (o->type == REDIS_SET) {
-                char cmd[]="*3\r\n$4\r\nSADD\r\n";
-
-                /* Emit the SADDs needed to rebuild the set */
-                if (o->encoding == REDIS_ENCODING_INTSET) {
-                    int ii = 0;
-                    int64_t llval;
-                    while(intsetGet(o->ptr,ii++,&llval)) {
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (rioWriteBulkLongLong(&aof,llval) == 0) goto werr;
-                    }
-                } else if (o->encoding == REDIS_ENCODING_HT) {
-                    dictIterator *di = dictGetIterator(o->ptr);
-                    dictEntry *de;
-                    while((de = dictNext(di)) != NULL) {
-                        robj *eleobj = dictGetKey(de);
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr;
-                    }
-                    dictReleaseIterator(di);
-                } else {
-                    redisPanic("Unknown set encoding");
-                }
+                if (rewriteSetObject(&aof,&key,o) == 0) goto werr;
             } else if (o->type == REDIS_ZSET) {
-                /* Emit the ZADDs needed to rebuild the sorted set */
-                char cmd[]="*4\r\n$4\r\nZADD\r\n";
-
-                if (o->encoding == REDIS_ENCODING_ZIPLIST) {
-                    unsigned char *zl = o->ptr;
-                    unsigned char *eptr, *sptr;
-                    unsigned char *vstr;
-                    unsigned int vlen;
-                    long long vll;
-                    double score;
-
-                    eptr = ziplistIndex(zl,0);
-                    redisAssert(eptr != NULL);
-                    sptr = ziplistNext(zl,eptr);
-                    redisAssert(sptr != NULL);
-
-                    while (eptr != NULL) {
-                        redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
-                        score = zzlGetScore(sptr);
-
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (rioWriteBulkDouble(&aof,score) == 0) goto werr;
-                        if (vstr != NULL) {
-                            if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0)
-                                goto werr;
-                        } else {
-                            if (rioWriteBulkLongLong(&aof,vll) == 0)
-                                goto werr;
-                        }
-                        zzlNext(zl,&eptr,&sptr);
-                    }
-                } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
-                    zset *zs = o->ptr;
-                    dictIterator *di = dictGetIterator(zs->dict);
-                    dictEntry *de;
-
-                    while((de = dictNext(di)) != NULL) {
-                        robj *eleobj = dictGetKey(de);
-                        double *score = dictGetVal(de);
-
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (rioWriteBulkDouble(&aof,*score) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr;
-                    }
-                    dictReleaseIterator(di);
-                } else {
-                    redisPanic("Unknown sorted set encoding");
-                }
+                if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
             } else if (o->type == REDIS_HASH) {
-                char cmd[]="*4\r\n$4\r\nHSET\r\n";
-
-                /* Emit the HSETs needed to rebuild the hash */
-                if (o->encoding == REDIS_ENCODING_ZIPMAP) {
-                    unsigned char *p = zipmapRewind(o->ptr);
-                    unsigned char *field, *val;
-                    unsigned int flen, vlen;
-
-                    while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (rioWriteBulkString(&aof,(char*)field,flen) == 0)
-                            goto werr;
-                        if (rioWriteBulkString(&aof,(char*)val,vlen) == 0)
-                            goto werr;
-                    }
-                } else {
-                    dictIterator *di = dictGetIterator(o->ptr);
-                    dictEntry *de;
-
-                    while((de = dictNext(di)) != NULL) {
-                        robj *field = dictGetKey(de);
-                        robj *val = dictGetVal(de);
-
-                        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,field) == 0) goto werr;
-                        if (rioWriteBulkObject(&aof,val) == 0) goto werr;
-                    }
-                    dictReleaseIterator(di);
-                }
+                if (rewriteHashObject(&aof,&key,o) == 0) goto werr;
             } else {
                 redisPanic("Unknown object type");
             }

diff --git a/src/redis.h b/src/redis.h
index c425691edc4b5e04dcd31fe3fd301ff9f99ae865..6b9bad08d7a6b11cc82b38bd4c9514a37f248eae 100644 (file)
--- a/src/redis.h
+++ b/src/redis.h
@@ -54,6 +54,7 @@
 #define REDIS_MAX_LOGMSG_LEN    1024 /* Default maximum length of syslog messages */
 #define REDIS_AUTO_AOFREWRITE_PERC  100
 #define REDIS_AUTO_AOFREWRITE_MIN_SIZE (1024*1024)
+#define REDIS_AOFREWRITE_ITEMS_PER_CMD 64
 #define REDIS_SLOWLOG_LOG_SLOWER_THAN 10000
 #define REDIS_SLOWLOG_MAX_LEN 64
 #define REDIS_MAX_CLIENTS 10000

diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl
index 2918fe7af60b4e45320e8420bd8867addb97230a..459784cd6386997199fe710c5146728aa4bcb7b2 100644 (file)
--- a/tests/test_helper.tcl
+++ b/tests/test_helper.tcl
@@ -25,6 +25,7 @@ set ::all_tests {
     unit/other
     unit/cas
     unit/quit
+    unit/aofrw
     integration/replication
     integration/replication-2
     integration/replication-3

diff --git a/tests/unit/aofrw.tcl b/tests/unit/aofrw.tcl
new file mode 100644 (file)
index 0000000..a558ed4
--- /dev/null
+++ b/tests/unit/aofrw.tcl
@@ -0,0 +1,107 @@
+start_server {tags {"aofrw"}} {
+    foreach d {string int} {
+        foreach e {ziplist linkedlist} {
+            test "AOF rewrite of list with $e encoding, $d data" {
+                r flushall
+                if {$e eq {ziplist}} {set len 10} else {set len 1000}
+                for {set j 0} {$j < $len} {incr j} {
+                    if {$d eq {string}} {
+                        set data [randstring 0 16 alpha]
+                    } else {
+                        set data [randomInt 4000000000]
+                    }
+                    r lpush key $data
+                }
+                assert_equal [r object encoding key] $e
+                set d1 [r debug digest]
+                r bgrewriteaof
+                waitForBgrewriteaof r
+                r debug loadaof
+                set d2 [r debug digest]
+                if {$d1 ne $d2} {
+                    error "assertion:$d1 is not equal to $d2"
+                }
+            }
+        }
+    }
+
+    foreach d {string int} {
+        foreach e {intset hashtable} {
+            test "AOF rewrite of set with $e encoding, $d data" {
+                r flushall
+                if {$e eq {intset}} {set len 10} else {set len 1000}
+                for {set j 0} {$j < $len} {incr j} {
+                    if {$d eq {string}} {
+                        set data [randstring 0 16 alpha]
+                    } else {
+                        set data [randomInt 4000000000]
+                    }
+                    r sadd key $data
+                }
+                if {$d ne {string}} {
+                    assert_equal [r object encoding key] $e
+                }
+                set d1 [r debug digest]
+                r bgrewriteaof
+                waitForBgrewriteaof r
+                r debug loadaof
+                set d2 [r debug digest]
+                if {$d1 ne $d2} {
+                    error "assertion:$d1 is not equal to $d2"
+                }
+            }
+        }
+    }
+
+    foreach d {string int} {
+        foreach e {zipmap hashtable} {
+            test "AOF rewrite of hash with $e encoding, $d data" {
+                r flushall
+                if {$e eq {zipmap}} {set len 10} else {set len 1000}
+                for {set j 0} {$j < $len} {incr j} {
+                    if {$d eq {string}} {
+                        set data [randstring 0 16 alpha]
+                    } else {
+                        set data [randomInt 4000000000]
+                    }
+                    r hset key $data $data
+                }
+                assert_equal [r object encoding key] $e
+                set d1 [r debug digest]
+                r bgrewriteaof
+                waitForBgrewriteaof r
+                r debug loadaof
+                set d2 [r debug digest]
+                if {$d1 ne $d2} {
+                    error "assertion:$d1 is not equal to $d2"
+                }
+            }
+        }
+    }
+
+    foreach d {string int} {
+        foreach e {ziplist skiplist} {
+            test "AOF rewrite of zset with $e encoding, $d data" {
+                r flushall
+                if {$e eq {ziplist}} {set len 10} else {set len 1000}
+                for {set j 0} {$j < $len} {incr j} {
+                    if {$d eq {string}} {
+                        set data [randstring 0 16 alpha]
+                    } else {
+                        set data [randomInt 4000000000]
+                    }
+                    r zadd key [expr rand()] $data
+                }
+                assert_equal [r object encoding key] $e
+                set d1 [r debug digest]
+                r bgrewriteaof
+                waitForBgrewriteaof r
+                r debug loadaof
+                set d2 [r debug digest]
+                if {$d1 ne $d2} {
+                    error "assertion:$d1 is not equal to $d2"
+                }
+            }
+        }
+    }
+}