Merge remote-tracking branch 'origin/unstable' into bg-aof-2

src/aof.c

@@ -812,6 +812,11 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
             server.appendseldb = -1; /* Make sure SELECT is re-issued */
             aofUpdateCurrentSize();
             server.auto_aofrewrite_base_size = server.appendonly_current_size;
+
+            /* Clear regular AOF buffer since its contents was just written to
+             * the new AOF from the background rewrite buffer. */
+            sdsfree(server.aofbuf);
+            server.aofbuf = sdsempty();
         }
 
         redisLog(REDIS_NOTICE, "Background AOF rewrite successful");

src/redis-benchmark.c

@@ -53,9 +53,10 @@ static struct config {
     int hostport;
     const char *hostsocket;
     int numclients;
-    int requests;
     int liveclients;
-    int donerequests;
+    int requests;
+    int requests_issued;
+    int requests_finished;
     int keysize;
     int datasize;
     int randomkeys;
@@ -148,7 +149,7 @@ static void randomizeClientKey(client c) {
 }
 
 static void clientDone(client c) {
-    if (config.donerequests == config.requests) {
+    if (config.requests_finished == config.requests) {
         freeClient(c);
         aeStop(config.el);
         return;
@@ -189,8 +190,8 @@ static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
                 exit(1);
             }
 
-            if (config.donerequests < config.requests)
-                config.latency[config.donerequests++] = c->latency;
+            if (config.requests_finished < config.requests)
+                config.latency[config.requests_finished++] = c->latency;
             clientDone(c);
         }
     }
@@ -202,8 +203,15 @@ static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
     REDIS_NOTUSED(fd);
     REDIS_NOTUSED(mask);
 
-    /* When nothing was written yet, randomize keys and set start time. */
+    /* Initialize request when nothing was written. */
     if (c->written == 0) {
+        /* Enforce upper bound to number of requests. */
+        if (config.requests_issued++ >= config.requests) {
+            freeClient(c);
+            return;
+        }
+
+        /* Really initialize: randomize keys and set start time. */
         if (config.randomkeys) randomizeClientKey(c);
         c->start = ustime();
         c->latency = -1;
@@ -286,10 +294,10 @@ static void showLatencyReport(void) {
     int i, curlat = 0;
     float perc, reqpersec;
 
-    reqpersec = (float)config.donerequests/((float)config.totlatency/1000);
+    reqpersec = (float)config.requests_finished/((float)config.totlatency/1000);
     if (!config.quiet) {
         printf("====== %s ======\n", config.title);
-        printf("  %d requests completed in %.2f seconds\n", config.donerequests,
+        printf("  %d requests completed in %.2f seconds\n", config.requests_finished,
             (float)config.totlatency/1000);
         printf("  %d parallel clients\n", config.numclients);
         printf("  %d bytes payload\n", config.datasize);
@@ -314,7 +322,8 @@ static void benchmark(const char *title, const char *cmd, int len) {
     client c;
 
     config.title = title;
-    config.donerequests = 0;
+    config.requests_issued = 0;
+    config.requests_finished = 0;
 
     c = createClient(cmd,len);
     createMissingClients(c);
@@ -416,7 +425,7 @@ int showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData
     REDIS_NOTUSED(clientData);
 
     float dt = (float)(mstime()-config.start)/1000.0;
-    float rps = (float)config.donerequests/dt;
+    float rps = (float)config.requests_finished/dt;
     printf("%s: %.2f\r", config.title, rps);
     fflush(stdout);
     return 250; /* every 250ms */
@@ -438,7 +447,6 @@ int main(int argc, const char **argv) {
     config.el = aeCreateEventLoop();
     aeCreateTimeEvent(config.el,1,showThroughput,NULL,NULL);
     config.keepalive = 1;
-    config.donerequests = 0;
     config.datasize = 3;
     config.randomkeys = 0;
     config.randomkeys_keyspacelen = 0;

src/redis-cli.c

@@ -61,6 +61,7 @@ static struct config {
     int shutdown;
     int monitor_mode;
     int pubsub_mode;
+    int latency_mode;
     int stdinarg; /* get last arg from stdin. (-x option) */
     char *auth;
     int raw_output; /* output mode per command */
@@ -567,6 +568,8 @@ static int parseOptions(int argc, char **argv) {
             i++;
         } else if (!strcmp(argv[i],"--raw")) {
             config.raw_output = 1;
+        } else if (!strcmp(argv[i],"--latency")) {
+            config.latency_mode = 1;
         } else if (!strcmp(argv[i],"-d") && !lastarg) {
             sdsfree(config.mb_delim);
             config.mb_delim = sdsnew(argv[i+1]);
@@ -617,6 +620,7 @@ static void usage() {
 "  -x               Read last argument from STDIN\n"
 "  -d <delimiter>   Multi-bulk delimiter in for raw formatting (default: \\n)\n"
 "  --raw            Use raw formatting for replies (default when STDOUT is not a tty)\n"
+"  --latency        Enter a special mode continuously sampling latency.\n"
 "  --help           Output this help and exit\n"
 "  --version        Output version and exit\n"
 "\n"
@@ -739,6 +743,38 @@ static int noninteractive(int argc, char **argv) {
     return retval;
 }
 
+static void latencyMode(void) {
+    redisReply *reply;
+    long long start, latency, min, max, tot, count = 0;
+    double avg;
+
+    if (!context) exit(1);
+    while(1) {
+        start = mstime();
+        reply = redisCommand(context,"PING");
+        if (reply == NULL) {
+            fprintf(stderr,"\nI/O error\n");
+            exit(1);
+        }
+        latency = mstime()-start;
+        freeReplyObject(reply);
+        count++;
+        if (count == 1) {
+            min = max = tot = latency;
+            avg = (double) latency;
+        } else {
+            if (latency < min) min = latency;
+            if (latency > max) max = latency;
+            tot += latency;
+            avg = (double) tot/count;
+        }
+        printf("\x1b[0G\x1b[2Kmin: %lld, max: %lld, avg: %.2f (%lld samples)",
+            min, max, avg, count);
+        fflush(stdout);
+        usleep(10000);
+    }
+}
+
 int main(int argc, char **argv) {
     int firstarg;
 
@@ -752,6 +788,7 @@ int main(int argc, char **argv) {
     config.shutdown = 0;
     config.monitor_mode = 0;
     config.pubsub_mode = 0;
+    config.latency_mode = 0;
     config.stdinarg = 0;
     config.auth = NULL;
     config.raw_output = !isatty(fileno(stdout)) && (getenv("FAKETTY") == NULL);
@@ -762,6 +799,12 @@ int main(int argc, char **argv) {
     argc -= firstarg;
     argv += firstarg;
 
+    /* Start in latency mode if appropriate */
+    if (config.latency_mode) {
+        cliConnect(0);
+        latencyMode();
+    }
+
     /* Start interactive mode when no command is provided */
     if (argc == 0) {
         /* Note that in repl mode we don't abort on connection error.

tests/integration/aof-race.tcl

@@ -0,0 +1,35 @@
+set defaults { appendonly {yes} appendfilename {appendonly.aof} }
+set server_path [tmpdir server.aof]
+set aof_path "$server_path/appendonly.aof"
+
+proc start_server_aof {overrides code} {
+    upvar defaults defaults srv srv server_path server_path
+    set config [concat $defaults $overrides]
+    start_server [list overrides $config] $code
+}
+
+tags {"aof"} {
+    # Specific test for a regression where internal buffers were not properly
+    # cleaned after a child responsible for an AOF rewrite exited. This buffer
+    # was subsequently appended to the new AOF, resulting in duplicate commands.
+    start_server_aof [list dir $server_path] {
+        set client [redis [srv host] [srv port]]
+        set bench [open "|src/redis-benchmark -q -p [srv port] -c 20 -n 20000 incr foo" "r+"]
+        after 100
+
+        # Benchmark should be running by now: start background rewrite
+        $client bgrewriteaof
+
+        # Read until benchmark pipe reaches EOF
+        while {[string length [read $bench]] > 0} {}
+
+        # Check contents of foo
+        assert_equal 20000 [$client get foo]
+    }
+
+    # Restart server to replay AOF
+    start_server_aof [list dir $server_path] {
+        set client [redis [srv host] [srv port]]
+        assert_equal 20000 [$client get foo]
+    }
+}

tests/test_helper.tcl

@@ -29,6 +29,7 @@ set ::all_tests {
     integration/replication-2
     integration/replication-3
     integration/aof
+    integration/aof-race
     unit/pubsub
     unit/slowlog
     unit/scripting