X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/57b0738011007e47ebe25d5c81acfe333c561e02..4be855e757b800a9f848823fcee8b7faf3e1eba0:/src/aof.c diff --git a/src/aof.c b/src/aof.c index 1ed2363a..4a463bde 100644 --- a/src/aof.c +++ b/src/aof.c @@ -1,4 +1,6 @@ #include "redis.h" +#include "bio.h" +#include "rio.h" #include #include @@ -8,10 +10,16 @@ #include #include +void aofUpdateCurrentSize(void); + +void aof_background_fsync(int fd) { + bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL); +} + /* Called when the user switches from "appendonly yes" to "appendonly no" * at runtime using the CONFIG command. */ void stopAppendOnly(void) { - flushAppendOnlyFile(); + flushAppendOnlyFile(1); aof_fsync(server.appendfd); close(server.appendfd); @@ -19,15 +27,15 @@ void stopAppendOnly(void) { server.appendseldb = -1; server.appendonly = 0; /* rewrite operation in progress? kill it, wait child exit */ - if (server.bgsavechildpid != -1) { + if (server.bgrewritechildpid != -1) { int statloc; - if (kill(server.bgsavechildpid,SIGKILL) != -1) + if (kill(server.bgrewritechildpid,SIGKILL) != -1) wait3(&statloc,0,NULL); /* reset the buffer accumulating changes while the child saves */ sdsfree(server.bgrewritebuf); server.bgrewritebuf = sdsempty(); - server.bgsavechildpid = -1; + server.bgrewritechildpid = -1; } } @@ -44,7 +52,7 @@ int startAppendOnly(void) { if (rewriteAppendOnlyFileBackground() == REDIS_ERR) { server.appendonly = 0; close(server.appendfd); - redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, I can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.",strerror(errno)); + redisLog(REDIS_WARNING,"User tried turning on AOF with CONFIG SET but I can't trigger a background AOF rewrite operation. Check the above logs for more info about the error."); return REDIS_ERR; } return REDIS_OK; @@ -56,79 +64,155 @@ int startAppendOnly(void) { * and the only way the client socket can get a write is entering when the * the event loop, we accumulate all the AOF writes in a memory * buffer and write it on disk using this function just before entering - * the event loop again. */ -void flushAppendOnlyFile(void) { - time_t now; + * the event loop again. + * + * About the 'force' argument: + * + * When the fsync policy is set to 'everysec' we may delay the flush if there + * is still an fsync() going on in the background thread, since for instance + * on Linux write(2) will be blocked by the background fsync anyway. + * When this happens we remember that there is some aof buffer to be + * flushed ASAP, and will try to do that in the serverCron() function. + * + * However if force is set to 1 we'll write regardless of the background + * fsync. */ +void flushAppendOnlyFile(int force) { ssize_t nwritten; + int sync_in_progress = 0; if (sdslen(server.aofbuf) == 0) return; + if (server.appendfsync == APPENDFSYNC_EVERYSEC) + sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0; + + if (server.appendfsync == APPENDFSYNC_EVERYSEC && !force) { + /* With this append fsync policy we do background fsyncing. + * If the fsync is still in progress we can try to delay + * the write for a couple of seconds. */ + if (sync_in_progress) { + if (server.aof_flush_postponed_start == 0) { + /* No previous write postponinig, remember that we are + * postponing the flush and return. */ + server.aof_flush_postponed_start = server.unixtime; + return; + } else if (server.unixtime - server.aof_flush_postponed_start < 2) { + /* We were already waiting for fsync to finish, but for less + * than two seconds this is still ok. Postpone again. */ + return; + } + /* Otherwise fall trough, and go write since we can't wait + * over two seconds. */ + redisLog(REDIS_NOTICE,"Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down Redis."); + } + } + /* If you are following this code path, then we are going to write so + * set reset the postponed flush sentinel to zero. */ + server.aof_flush_postponed_start = 0; + /* We want to perform a single write. This should be guaranteed atomic * at least if the filesystem we are writing is a real physical one. * While this will save us against the server being killed I don't think * there is much to do about the whole server stopping for power problems * or alike */ - nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf)); - if (nwritten != (signed)sdslen(server.aofbuf)) { + nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf)); + if (nwritten != (signed)sdslen(server.aofbuf)) { /* Ooops, we are in troubles. The best thing to do for now is * aborting instead of giving the illusion that everything is * working as expected. */ - if (nwritten == -1) { + if (nwritten == -1) { redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno)); - } else { + } else { redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno)); - } - exit(1); + } + exit(1); + } + server.appendonly_current_size += nwritten; + + /* Re-use AOF buffer when it is small enough. The maximum comes from the + * arena size of 4k minus some overhead (but is otherwise arbitrary). */ + if ((sdslen(server.aofbuf)+sdsavail(server.aofbuf)) < 4000) { + sdsclear(server.aofbuf); + } else { + sdsfree(server.aofbuf); + server.aofbuf = sdsempty(); } - sdsfree(server.aofbuf); - server.aofbuf = sdsempty(); - /* Don't Fsync if no-appendfsync-on-rewrite is set to yes and we have - * childs performing heavy I/O on disk. */ + /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are + * children doing I/O in the background. */ if (server.no_appendfsync_on_rewrite && (server.bgrewritechildpid != -1 || server.bgsavechildpid != -1)) return; - /* Fsync if needed */ - now = time(NULL); - if (server.appendfsync == APPENDFSYNC_ALWAYS || - (server.appendfsync == APPENDFSYNC_EVERYSEC && - now-server.lastfsync > 1)) - { + + /* Perform the fsync if needed. */ + if (server.appendfsync == APPENDFSYNC_ALWAYS) { /* aof_fsync is defined as fdatasync() for Linux in order to avoid * flushing metadata. */ aof_fsync(server.appendfd); /* Let's try to get this data on the disk */ - server.lastfsync = now; + server.lastfsync = server.unixtime; + } else if ((server.appendfsync == APPENDFSYNC_EVERYSEC && + server.unixtime > server.lastfsync)) { + if (!sync_in_progress) aof_background_fsync(server.appendfd); + server.lastfsync = server.unixtime; } } -sds catAppendOnlyGenericCommand(sds buf, int argc, robj **argv) { - int j; - buf = sdscatprintf(buf,"*%d\r\n",argc); +sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) { + char buf[32]; + int len, j; + robj *o; + + buf[0] = '*'; + len = 1+ll2string(buf+1,sizeof(buf)-1,argc); + buf[len++] = '\r'; + buf[len++] = '\n'; + dst = sdscatlen(dst,buf,len); + for (j = 0; j < argc; j++) { - robj *o = getDecodedObject(argv[j]); - buf = sdscatprintf(buf,"$%lu\r\n",(unsigned long)sdslen(o->ptr)); - buf = sdscatlen(buf,o->ptr,sdslen(o->ptr)); - buf = sdscatlen(buf,"\r\n",2); + o = getDecodedObject(argv[j]); + buf[0] = '$'; + len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr)); + buf[len++] = '\r'; + buf[len++] = '\n'; + dst = sdscatlen(dst,buf,len); + dst = sdscatlen(dst,o->ptr,sdslen(o->ptr)); + dst = sdscatlen(dst,"\r\n",2); decrRefCount(o); } - return buf; + return dst; } -sds catAppendOnlyExpireAtCommand(sds buf, robj *key, robj *seconds) { - int argc = 3; - long when; +/* Create the sds representation of an PEXPIREAT command, using + * 'seconds' as time to live and 'cmd' to understand what command + * we are translating into a PEXPIREAT. + * + * This command is used in order to translate EXPIRE and PEXPIRE commands + * into PEXPIREAT command so that we retain precision in the append only + * file, and the time is always absolute and not relative. */ +sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) { + long long when; robj *argv[3]; /* Make sure we can use strtol */ seconds = getDecodedObject(seconds); - when = time(NULL)+strtol(seconds->ptr,NULL,10); + when = strtoll(seconds->ptr,NULL,10); + /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */ + if (cmd->proc == expireCommand || cmd->proc == setexCommand || + cmd->proc == expireatCommand) + { + when *= 1000; + } + /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */ + if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || + cmd->proc == setexCommand || cmd->proc == psetexCommand) + { + when += mstime(); + } decrRefCount(seconds); - argv[0] = createStringObject("EXPIREAT",8); + argv[0] = createStringObject("PEXPIREAT",9); argv[1] = key; - argv[2] = createObject(REDIS_STRING, - sdscatprintf(sdsempty(),"%ld",when)); - buf = catAppendOnlyGenericCommand(buf, argc, argv); + argv[2] = createStringObjectFromLongLong(when); + buf = catAppendOnlyGenericCommand(buf, 3, argv); decrRefCount(argv[0]); decrRefCount(argv[2]); return buf; @@ -149,18 +233,22 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a server.appendseldb = dictid; } - if (cmd->proc == expireCommand) { - /* Translate EXPIRE into EXPIREAT */ - buf = catAppendOnlyExpireAtCommand(buf,argv[1],argv[2]); - } else if (cmd->proc == setexCommand) { - /* Translate SETEX to SET and EXPIREAT */ + if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || + cmd->proc == expireatCommand) { + /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */ + buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); + } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) { + /* Translate SETEX/PSETEX to SET and PEXPIREAT */ tmpargv[0] = createStringObject("SET",3); tmpargv[1] = argv[1]; tmpargv[2] = argv[3]; buf = catAppendOnlyGenericCommand(buf,3,tmpargv); decrRefCount(tmpargv[0]); - buf = catAppendOnlyExpireAtCommand(buf,argv[1],argv[2]); + buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); } else { + /* All the other commands don't need translation or need the + * same translation already operated in the command vector + * for the replication itself. */ buf = catAppendOnlyGenericCommand(buf,argc,argv); } @@ -189,6 +277,7 @@ struct redisClient *createFakeClient(void) { c->querybuf = sdsempty(); c->argc = 0; c->argv = NULL; + c->bufpos = 0; c->flags = 0; /* We set the fake client as a slave waiting for the synchronization * so that Redis will not try to send replies to this client. */ @@ -217,9 +306,13 @@ int loadAppendOnlyFile(char *filename) { FILE *fp = fopen(filename,"r"); struct redis_stat sb; int appendonly = server.appendonly; + long loops = 0; - if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) + if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) { + server.appendonly_current_size = 0; + fclose(fp); return REDIS_ERR; + } if (fp == NULL) { redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno)); @@ -231,6 +324,8 @@ int loadAppendOnlyFile(char *filename) { server.appendonly = 0; fakeClient = createFakeClient(); + startLoading(fp); + while(1) { int argc, j; unsigned long len; @@ -238,7 +333,12 @@ int loadAppendOnlyFile(char *filename) { char buf[128]; sds argsds; struct redisCommand *cmd; - int force_swapout; + + /* Serve the clients from time to time */ + if (!(loops++ % 1000)) { + loadingProgress(ftello(fp)); + aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); + } if (fgets(buf,sizeof(buf),fp) == NULL) { if (feof(fp)) @@ -248,6 +348,8 @@ int loadAppendOnlyFile(char *filename) { } if (buf[0] != '*') goto fmterr; argc = atoi(buf+1); + if (argc < 1) goto fmterr; + argv = zmalloc(sizeof(robj*)*argc); for (j = 0; j < argc; j++) { if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr; @@ -265,9 +367,6 @@ int loadAppendOnlyFile(char *filename) { redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", argv[0]->ptr); exit(1); } - /* Try object encoding */ - if (cmd->flags & REDIS_CMD_BULK) - argv[argc-1] = tryObjectEncoding(argv[argc-1]); /* Run the command in the context of a fake client */ fakeClient->argc = argc; fakeClient->argv = argv; @@ -275,21 +374,14 @@ int loadAppendOnlyFile(char *filename) { /* The fake client should not have a reply */ redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0); - - /* Clean up, ready for the next command */ - for (j = 0; j < argc; j++) decrRefCount(argv[j]); - zfree(argv); - - /* Handle swapping while loading big datasets when VM is on */ - force_swapout = 0; - if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32) - force_swapout = 1; - - if (server.vm_enabled && force_swapout) { - while (zmalloc_used_memory() > server.vm_max_memory) { - if (vmSwapOneObjectBlocking() == REDIS_ERR) break; - } - } + /* The fake client should never get blocked */ + redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0); + + /* Clean up. Command code may have changed argv/argc so we use the + * argv/argc of the client instead of the local variables. */ + for (j = 0; j < fakeClient->argc; j++) + decrRefCount(fakeClient->argv[j]); + zfree(fakeClient->argv); } /* This point can only be reached when EOF is reached without errors. @@ -299,6 +391,9 @@ int loadAppendOnlyFile(char *filename) { fclose(fp); freeFakeClient(fakeClient); server.appendonly = appendonly; + stopLoading(); + aofUpdateCurrentSize(); + server.auto_aofrewrite_base_size = server.appendonly_current_size; return REDIS_OK; readerr: @@ -309,54 +404,19 @@ readerr: } exit(1); fmterr: - redisLog(REDIS_WARNING,"Bad file format reading the append only file"); + redisLog(REDIS_WARNING,"Bad file format reading the append only file: make a backup of your AOF file, then use ./redis-check-aof --fix "); exit(1); } -/* Write binary-safe string into a file in the bulkformat - * $\r\n\r\n */ -int fwriteBulkString(FILE *fp, char *s, unsigned long len) { - char cbuf[128]; - int clen; - cbuf[0] = '$'; - clen = 1+ll2string(cbuf+1,sizeof(cbuf)-1,len); - cbuf[clen++] = '\r'; - cbuf[clen++] = '\n'; - if (fwrite(cbuf,clen,1,fp) == 0) return 0; - if (len > 0 && fwrite(s,len,1,fp) == 0) return 0; - if (fwrite("\r\n",2,1,fp) == 0) return 0; - return 1; -} - -/* Write a double value in bulk format $\r\n\r\n */ -int fwriteBulkDouble(FILE *fp, double d) { - char buf[128], dbuf[128]; - - snprintf(dbuf,sizeof(dbuf),"%.17g\r\n",d); - snprintf(buf,sizeof(buf),"$%lu\r\n",(unsigned long)strlen(dbuf)-2); - if (fwrite(buf,strlen(buf),1,fp) == 0) return 0; - if (fwrite(dbuf,strlen(dbuf),1,fp) == 0) return 0; - return 1; -} - -/* Write a long value in bulk format $\r\n\r\n */ -int fwriteBulkLongLong(FILE *fp, long long l) { - char bbuf[128], lbuf[128]; - unsigned int blen, llen; - llen = ll2string(lbuf,32,l); - blen = snprintf(bbuf,sizeof(bbuf),"$%u\r\n%s\r\n",llen,lbuf); - if (fwrite(bbuf,blen,1,fp) == 0) return 0; - return 1; -} - -/* Delegate writing an object to writing a bulk string or bulk long long. */ -int fwriteBulkObject(FILE *fp, robj *obj) { +/* Delegate writing an object to writing a bulk string or bulk long long. + * This is not placed in rio.c since that adds the redis.h dependency. */ +int rioWriteBulkObject(rio *r, robj *obj) { /* Avoid using getDecodedObject to help copy-on-write (we are often * in a child process when this function is called). */ if (obj->encoding == REDIS_ENCODING_INT) { - return fwriteBulkLongLong(fp,(long)obj->ptr); + return rioWriteBulkLongLong(r,(long)obj->ptr); } else if (obj->encoding == REDIS_ENCODING_RAW) { - return fwriteBulkString(fp,obj->ptr,sdslen(obj->ptr)); + return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr)); } else { redisPanic("Unknown string encoding"); } @@ -367,10 +427,11 @@ int fwriteBulkObject(FILE *fp, robj *obj) { int rewriteAppendOnlyFile(char *filename) { dictIterator *di = NULL; dictEntry *de; + rio aof; FILE *fp; char tmpfile[256]; int j; - time_t now = time(NULL); + long long now = mstime(); /* Note that we have to use a different temp name here compared to the * one used by rewriteAppendOnlyFileBackground() function. */ @@ -380,52 +441,43 @@ int rewriteAppendOnlyFile(char *filename) { redisLog(REDIS_WARNING, "Failed rewriting the append only file: %s", strerror(errno)); return REDIS_ERR; } + + rioInitWithFile(&aof,fp); for (j = 0; j < server.dbnum; j++) { char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n"; redisDb *db = server.db+j; dict *d = db->dict; if (dictSize(d) == 0) continue; - di = dictGetIterator(d); + di = dictGetSafeIterator(d); if (!di) { fclose(fp); return REDIS_ERR; } /* SELECT the new DB */ - if (fwrite(selectcmd,sizeof(selectcmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkLongLong(fp,j) == 0) goto werr; + if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr; + if (rioWriteBulkLongLong(&aof,j) == 0) goto werr; /* Iterate this DB writing every entry */ while((de = dictNext(di)) != NULL) { - sds keystr = dictGetEntryKey(de); + sds keystr; robj key, *o; - time_t expiretime; - int swapped; + long long expiretime; - keystr = dictGetEntryKey(de); - o = dictGetEntryVal(de); + keystr = dictGetKey(de); + o = dictGetVal(de); initStaticStringObject(key,keystr); - /* If the value for this key is swapped, load a preview in memory. - * We use a "swapped" flag to remember if we need to free the - * value object instead to just increment the ref count anyway - * in order to avoid copy-on-write of pages if we are forked() */ - if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY || - o->storage == REDIS_VM_SWAPPING) { - swapped = 0; - } else { - o = vmPreviewObject(o); - swapped = 1; - } + expiretime = getExpire(db,&key); /* Save the key and associated value */ if (o->type == REDIS_STRING) { /* Emit a SET command */ char cmd[]="*3\r\n$3\r\nSET\r\n"; - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; + if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; /* Key and value */ - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkObject(fp,o) == 0) goto werr; + 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"; @@ -437,13 +489,13 @@ int rewriteAppendOnlyFile(char *filename) { long long vlong; while(ziplistGet(p,&vstr,&vlen,&vlong)) { - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; + if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; + if (rioWriteBulkObject(&aof,&key) == 0) goto werr; if (vstr) { - if (fwriteBulkString(fp,(char*)vstr,vlen) == 0) + if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0) goto werr; } else { - if (fwriteBulkLongLong(fp,vlong) == 0) + if (rioWriteBulkLongLong(&aof,vlong) == 0) goto werr; } p = ziplistNext(zl,p); @@ -457,9 +509,9 @@ int rewriteAppendOnlyFile(char *filename) { while((ln = listNext(&li))) { robj *eleobj = listNodeValue(ln); - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkObject(fp,eleobj) == 0) goto werr; + 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"); @@ -472,18 +524,18 @@ int rewriteAppendOnlyFile(char *filename) { int ii = 0; int64_t llval; while(intsetGet(o->ptr,ii++,&llval)) { - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkLongLong(fp,llval) == 0) goto werr; + 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 = dictGetEntryKey(de); - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkObject(fp,eleobj) == 0) goto werr; + 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 { @@ -491,21 +543,55 @@ int rewriteAppendOnlyFile(char *filename) { } } else if (o->type == REDIS_ZSET) { /* Emit the ZADDs needed to rebuild the sorted set */ - zset *zs = o->ptr; - dictIterator *di = dictGetIterator(zs->dict); - dictEntry *de; - - while((de = dictNext(di)) != NULL) { - char cmd[]="*4\r\n$4\r\nZADD\r\n"; - robj *eleobj = dictGetEntryKey(de); - double *score = dictGetEntryVal(de); - - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkDouble(fp,*score) == 0) goto werr; - if (fwriteBulkObject(fp,eleobj) == 0) goto werr; + 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"); } - dictReleaseIterator(di); } else if (o->type == REDIS_HASH) { char cmd[]="*4\r\n$4\r\nHSET\r\n"; @@ -516,11 +602,11 @@ int rewriteAppendOnlyFile(char *filename) { unsigned int flen, vlen; while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) { - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkString(fp,(char*)field,flen) == 0) + 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 (fwriteBulkString(fp,(char*)val,vlen) == 0) + if (rioWriteBulkString(&aof,(char*)val,vlen) == 0) goto werr; } } else { @@ -528,13 +614,13 @@ int rewriteAppendOnlyFile(char *filename) { dictEntry *de; while((de = dictNext(di)) != NULL) { - robj *field = dictGetEntryKey(de); - robj *val = dictGetEntryVal(de); + robj *field = dictGetKey(de); + robj *val = dictGetVal(de); - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkObject(fp,field) == 0) goto werr; - if (fwriteBulkObject(fp,val) == 0) goto werr; + 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); } @@ -543,14 +629,13 @@ int rewriteAppendOnlyFile(char *filename) { } /* Save the expire time */ if (expiretime != -1) { - char cmd[]="*3\r\n$8\r\nEXPIREAT\r\n"; + char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n"; /* If this key is already expired skip it */ if (expiretime < now) continue; - if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; - if (fwriteBulkObject(fp,&key) == 0) goto werr; - if (fwriteBulkLongLong(fp,expiretime) == 0) goto werr; + if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; + if (rioWriteBulkObject(&aof,&key) == 0) goto werr; + if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr; } - if (swapped) decrRefCount(o); } dictReleaseIterator(di); } @@ -592,15 +677,16 @@ werr: */ int rewriteAppendOnlyFileBackground(void) { pid_t childpid; + long long start; if (server.bgrewritechildpid != -1) return REDIS_ERR; - if (server.vm_enabled) waitEmptyIOJobsQueue(); + start = ustime(); if ((childpid = fork()) == 0) { - /* Child */ char tmpfile[256]; - if (server.vm_enabled) vmReopenSwapFile(); - close(server.fd); + /* Child */ + if (server.ipfd > 0) close(server.ipfd); + if (server.sofd > 0) close(server.sofd); snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) { _exit(0); @@ -609,6 +695,7 @@ int rewriteAppendOnlyFileBackground(void) { } } else { /* Parent */ + server.stat_fork_time = ustime()-start; if (childpid == -1) { redisLog(REDIS_WARNING, "Can't rewrite append only file in background: fork: %s", @@ -617,6 +704,7 @@ int rewriteAppendOnlyFileBackground(void) { } redisLog(REDIS_NOTICE, "Background append only file rewriting started by pid %d",childpid); + server.aofrewrite_scheduled = 0; server.bgrewritechildpid = childpid; updateDictResizePolicy(); /* We set appendseldb to -1 in order to force the next call to the @@ -631,12 +719,12 @@ int rewriteAppendOnlyFileBackground(void) { void bgrewriteaofCommand(redisClient *c) { if (server.bgrewritechildpid != -1) { - addReplySds(c,sdsnew("-ERR background append only file rewriting already in progress\r\n")); - return; - } - if (rewriteAppendOnlyFileBackground() == REDIS_OK) { - char *status = "+Background append only file rewriting started\r\n"; - addReplySds(c,sdsnew(status)); + addReplyError(c,"Background append only file rewriting already in progress"); + } else if (server.bgsavechildpid != -1) { + server.aofrewrite_scheduled = 1; + addReplyStatus(c,"Background append only file rewriting scheduled"); + } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) { + addReplyStatus(c,"Background append only file rewriting started"); } else { addReply(c,shared.err); } @@ -649,61 +737,146 @@ void aofRemoveTempFile(pid_t childpid) { unlink(tmpfile); } +/* Update the server.appendonly_current_size filed explicitly using stat(2) + * to check the size of the file. This is useful after a rewrite or after + * a restart, normally the size is updated just adding the write length + * to the current lenght, that is much faster. */ +void aofUpdateCurrentSize(void) { + struct redis_stat sb; + + if (redis_fstat(server.appendfd,&sb) == -1) { + redisLog(REDIS_WARNING,"Unable to check the AOF length: %s", + strerror(errno)); + } else { + server.appendonly_current_size = sb.st_size; + } +} + /* A background append only file rewriting (BGREWRITEAOF) terminated its work. * Handle this. */ -void backgroundRewriteDoneHandler(int statloc) { - int exitcode = WEXITSTATUS(statloc); - int bysignal = WIFSIGNALED(statloc); - +void backgroundRewriteDoneHandler(int exitcode, int bysignal) { if (!bysignal && exitcode == 0) { - int fd; + int newfd, oldfd; + int nwritten; char tmpfile[256]; + long long now = ustime(); redisLog(REDIS_NOTICE, - "Background append only file rewriting terminated with success"); - /* Now it's time to flush the differences accumulated by the parent */ - snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) server.bgrewritechildpid); - fd = open(tmpfile,O_WRONLY|O_APPEND); - if (fd == -1) { - redisLog(REDIS_WARNING, "Not able to open the temp append only file produced by the child: %s", strerror(errno)); + "Background AOF rewrite terminated with success"); + + /* Flush the differences accumulated by the parent to the + * rewritten AOF. */ + snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", + (int)server.bgrewritechildpid); + newfd = open(tmpfile,O_WRONLY|O_APPEND); + if (newfd == -1) { + redisLog(REDIS_WARNING, + "Unable to open the temporary AOF produced by the child: %s", strerror(errno)); goto cleanup; } - /* Flush our data... */ - if (write(fd,server.bgrewritebuf,sdslen(server.bgrewritebuf)) != - (signed) sdslen(server.bgrewritebuf)) { - redisLog(REDIS_WARNING, "Error or short write trying to flush the parent diff of the append log file in the child temp file: %s", strerror(errno)); - close(fd); + + nwritten = write(newfd,server.bgrewritebuf,sdslen(server.bgrewritebuf)); + if (nwritten != (signed)sdslen(server.bgrewritebuf)) { + if (nwritten == -1) { + redisLog(REDIS_WARNING, + "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); + } else { + redisLog(REDIS_WARNING, + "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); + } + close(newfd); goto cleanup; } - redisLog(REDIS_NOTICE,"Parent diff flushed into the new append log file with success (%lu bytes)",sdslen(server.bgrewritebuf)); - /* Now our work is to rename the temp file into the stable file. And - * switch the file descriptor used by the server for append only. */ + + redisLog(REDIS_NOTICE, + "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten); + + /* The only remaining thing to do is to rename the temporary file to + * the configured file and switch the file descriptor used to do AOF + * writes. We don't want close(2) or rename(2) calls to block the + * server on old file deletion. + * + * There are two possible scenarios: + * + * 1) AOF is DISABLED and this was a one time rewrite. The temporary + * file will be renamed to the configured file. When this file already + * exists, it will be unlinked, which may block the server. + * + * 2) AOF is ENABLED and the rewritten AOF will immediately start + * receiving writes. After the temporary file is renamed to the + * configured file, the original AOF file descriptor will be closed. + * Since this will be the last reference to that file, closing it + * causes the underlying file to be unlinked, which may block the + * server. + * + * To mitigate the blocking effect of the unlink operation (either + * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we + * use a background thread to take care of this. First, we + * make scenario 1 identical to scenario 2 by opening the target file + * when it exists. The unlink operation after the rename(2) will then + * be executed upon calling close(2) for its descriptor. Everything to + * guarantee atomicity for this switch has already happened by then, so + * we don't care what the outcome or duration of that close operation + * is, as long as the file descriptor is released again. */ + if (server.appendfd == -1) { + /* AOF disabled */ + + /* Don't care if this fails: oldfd will be -1 and we handle that. + * One notable case of -1 return is if the old file does + * not exist. */ + oldfd = open(server.appendfilename,O_RDONLY|O_NONBLOCK); + } else { + /* AOF enabled */ + oldfd = -1; /* We'll set this to the current AOF filedes later. */ + } + + /* Rename the temporary file. This will not unlink the target file if + * it exists, because we reference it with "oldfd". */ if (rename(tmpfile,server.appendfilename) == -1) { - redisLog(REDIS_WARNING,"Can't rename the temp append only file into the stable one: %s", strerror(errno)); - close(fd); + redisLog(REDIS_WARNING, + "Error trying to rename the temporary AOF: %s", strerror(errno)); + close(newfd); + if (oldfd != -1) close(oldfd); goto cleanup; } - /* Mission completed... almost */ - redisLog(REDIS_NOTICE,"Append only file successfully rewritten."); - if (server.appendfd != -1) { - /* If append only is actually enabled... */ - close(server.appendfd); - server.appendfd = fd; - if (server.appendfsync != APPENDFSYNC_NO) aof_fsync(fd); - server.appendseldb = -1; /* Make sure it will issue SELECT */ - redisLog(REDIS_NOTICE,"The new append only file was selected for future appends."); + + if (server.appendfd == -1) { + /* AOF disabled, we don't need to set the AOF file descriptor + * to this new file, so we can close it. */ + close(newfd); } else { - /* If append only is disabled we just generate a dump in this - * format. Why not? */ - close(fd); + /* AOF enabled, replace the old fd with the new one. */ + oldfd = server.appendfd; + server.appendfd = newfd; + if (server.appendfsync == APPENDFSYNC_ALWAYS) + aof_fsync(newfd); + else if (server.appendfsync == APPENDFSYNC_EVERYSEC) + aof_background_fsync(newfd); + 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"); + + /* Asynchronously close the overwritten AOF. */ + if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL); + + redisLog(REDIS_VERBOSE, + "Background AOF rewrite signal handler took %lldus", ustime()-now); } else if (!bysignal && exitcode != 0) { - redisLog(REDIS_WARNING, "Background append only file rewriting error"); + redisLog(REDIS_WARNING, + "Background AOF rewrite terminated with error"); } else { redisLog(REDIS_WARNING, - "Background append only file rewriting terminated by signal %d", - WTERMSIG(statloc)); + "Background AOF rewrite terminated by signal %d", bysignal); } + cleanup: sdsfree(server.bgrewritebuf); server.bgrewritebuf = sdsempty();