diff --git a/src/aof.c b/src/aof.c index 517a19d3..5417e76b 100644 --- a/src/aof.c +++ b/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"); diff --git a/src/redis-benchmark.c b/src/redis-benchmark.c index 7295dc32..e4a40e13 100644 --- a/src/redis-benchmark.c +++ b/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; diff --git a/src/redis-cli.c b/src/redis-cli.c index d0c9d979..328cd3df 100644 --- a/src/redis-cli.c +++ b/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 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. diff --git a/tests/integration/aof-race.tcl b/tests/integration/aof-race.tcl new file mode 100644 index 00000000..207f2073 --- /dev/null +++ b/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] + } +} diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl index 4f3cf01e..476b5870 100644 --- a/tests/test_helper.tcl +++ b/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