src/aof.c

   1 #include "redis.h"
   2
   3 #include <signal.h>
   4 #include <fcntl.h>
   5 #include <sys/stat.h>
   6 #include <sys/types.h>
   7 #include <sys/time.h>
   8 #include <sys/resource.h>
   9 #include <sys/wait.h>
  10
  11 /* Called when the user switches from "appendonly yes" to "appendonly no"
  12  * at runtime using the CONFIG command. */
  13 void stopAppendOnly(void) {
  14     flushAppendOnlyFile();
  15     aof_fsync(server.appendfd);
  16     close(server.appendfd);
  17
  18     server.appendfd = -1;
  19     server.appendseldb = -1;
  20     server.appendonly = 0;
  21     /* rewrite operation in progress? kill it, wait child exit */
  22     if (server.bgsavechildpid != -1) {
  23         int statloc;
  24
  25         if (kill(server.bgsavechildpid,SIGKILL) != -1)
  26             wait3(&statloc,0,NULL);
  27         /* reset the buffer accumulating changes while the child saves */
  28         sdsfree(server.bgrewritebuf);
  29         server.bgrewritebuf = sdsempty();
  30         server.bgsavechildpid = -1;
  31     }
  32 }
  33
  34 /* Called when the user switches from "appendonly no" to "appendonly yes"
  35  * at runtime using the CONFIG command. */
  36 int startAppendOnly(void) {
  37     server.appendonly = 1;
  38     server.lastfsync = time(NULL);
  39     server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
  40     if (server.appendfd == -1) {
  41         redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, but I can't open the AOF file: %s",strerror(errno));
  42         return REDIS_ERR;
  43     }
  44     if (rewriteAppendOnlyFileBackground() == REDIS_ERR) {
  45         server.appendonly = 0;
  46         close(server.appendfd);
  47         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));
  48         return REDIS_ERR;
  49     }
  50     return REDIS_OK;
  51 }
  52
  53 /* Write the append only file buffer on disk.
  54  *
  55  * Since we are required to write the AOF before replying to the client,
  56  * and the only way the client socket can get a write is entering when the
  57  * the event loop, we accumulate all the AOF writes in a memory
  58  * buffer and write it on disk using this function just before entering
  59  * the event loop again. */
  60 void flushAppendOnlyFile(void) {
  61     time_t now;
  62     ssize_t nwritten;
  63
  64     if (sdslen(server.aofbuf) == 0) return;
  65
  66     /* We want to perform a single write. This should be guaranteed atomic
  67      * at least if the filesystem we are writing is a real physical one.
  68      * While this will save us against the server being killed I don't think
  69      * there is much to do about the whole server stopping for power problems
  70      * or alike */
  71      nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf));
  72      if (nwritten != (signed)sdslen(server.aofbuf)) {
  73         /* Ooops, we are in troubles. The best thing to do for now is
  74          * aborting instead of giving the illusion that everything is
  75          * working as expected. */
  76          if (nwritten == -1) {
  77             redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno));
  78          } else {
  79             redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno));
  80          }
  81          exit(1);
  82     }
  83     sdsfree(server.aofbuf);
  84     server.aofbuf = sdsempty();
  85
  86     /* Don't Fsync if no-appendfsync-on-rewrite is set to yes and we have
  87      * childs performing heavy I/O on disk. */
  88     if (server.no_appendfsync_on_rewrite &&
  89         (server.bgrewritechildpid != -1 || server.bgsavechildpid != -1))
  90             return;
  91     /* Fsync if needed */
  92     now = time(NULL);
  93     if (server.appendfsync == APPENDFSYNC_ALWAYS ||
  94         (server.appendfsync == APPENDFSYNC_EVERYSEC &&
  95          now-server.lastfsync > 1))
  96     {
  97         /* aof_fsync is defined as fdatasync() for Linux in order to avoid
  98          * flushing metadata. */
  99         aof_fsync(server.appendfd); /* Let's try to get this data on the disk */
 100         server.lastfsync = now;
 101     }
 102 }
 103
 104 sds catAppendOnlyGenericCommand(sds buf, int argc, robj **argv) {
 105     int j;
 106     buf = sdscatprintf(buf,"*%d\r\n",argc);
 107     for (j = 0; j < argc; j++) {
 108         robj *o = getDecodedObject(argv[j]);
 109         buf = sdscatprintf(buf,"$%lu\r\n",(unsigned long)sdslen(o->ptr));
 110         buf = sdscatlen(buf,o->ptr,sdslen(o->ptr));
 111         buf = sdscatlen(buf,"\r\n",2);
 112         decrRefCount(o);
 113     }
 114     return buf;
 115 }
 116
 117 sds catAppendOnlyExpireAtCommand(sds buf, robj *key, robj *seconds) {
 118     int argc = 3;
 119     long when;
 120     robj *argv[3];
 121
 122     /* Make sure we can use strtol */
 123     seconds = getDecodedObject(seconds);
 124     when = time(NULL)+strtol(seconds->ptr,NULL,10);
 125     decrRefCount(seconds);
 126
 127     argv[0] = createStringObject("EXPIREAT",8);
 128     argv[1] = key;
 129     argv[2] = createObject(REDIS_STRING,
 130         sdscatprintf(sdsempty(),"%ld",when));
 131     buf = catAppendOnlyGenericCommand(buf, argc, argv);
 132     decrRefCount(argv[0]);
 133     decrRefCount(argv[2]);
 134     return buf;
 135 }
 136
 137 void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
 138     sds buf = sdsempty();
 139     robj *tmpargv[3];
 140
 141     /* The DB this command was targetting is not the same as the last command
 142      * we appendend. To issue a SELECT command is needed. */
 143     if (dictid != server.appendseldb) {
 144         char seldb[64];
 145
 146         snprintf(seldb,sizeof(seldb),"%d",dictid);
 147         buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
 148             (unsigned long)strlen(seldb),seldb);
 149         server.appendseldb = dictid;
 150     }
 151
 152     if (cmd->proc == expireCommand) {
 153         /* Translate EXPIRE into EXPIREAT */
 154         buf = catAppendOnlyExpireAtCommand(buf,argv[1],argv[2]);
 155     } else if (cmd->proc == setexCommand) {
 156         /* Translate SETEX to SET and EXPIREAT */
 157         tmpargv[0] = createStringObject("SET",3);
 158         tmpargv[1] = argv[1];
 159         tmpargv[2] = argv[3];
 160         buf = catAppendOnlyGenericCommand(buf,3,tmpargv);
 161         decrRefCount(tmpargv[0]);
 162         buf = catAppendOnlyExpireAtCommand(buf,argv[1],argv[2]);
 163     } else {
 164         buf = catAppendOnlyGenericCommand(buf,argc,argv);
 165     }
 166
 167     /* Append to the AOF buffer. This will be flushed on disk just before
 168      * of re-entering the event loop, so before the client will get a
 169      * positive reply about the operation performed. */
 170     server.aofbuf = sdscatlen(server.aofbuf,buf,sdslen(buf));
 171
 172     /* If a background append only file rewriting is in progress we want to
 173      * accumulate the differences between the child DB and the current one
 174      * in a buffer, so that when the child process will do its work we
 175      * can append the differences to the new append only file. */
 176     if (server.bgrewritechildpid != -1)
 177         server.bgrewritebuf = sdscatlen(server.bgrewritebuf,buf,sdslen(buf));
 178
 179     sdsfree(buf);
 180 }
 181
 182 /* In Redis commands are always executed in the context of a client, so in
 183  * order to load the append only file we need to create a fake client. */
 184 struct redisClient *createFakeClient(void) {
 185     struct redisClient *c = zmalloc(sizeof(*c));
 186
 187     selectDb(c,0);
 188     c->fd = -1;
 189     c->querybuf = sdsempty();
 190     c->argc = 0;
 191     c->argv = NULL;
 192     c->bufpos = 0;
 193     c->flags = 0;
 194     /* We set the fake client as a slave waiting for the synchronization
 195      * so that Redis will not try to send replies to this client. */
 196     c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
 197     c->reply = listCreate();
 198     c->watched_keys = listCreate();
 199     listSetFreeMethod(c->reply,decrRefCount);
 200     listSetDupMethod(c->reply,dupClientReplyValue);
 201     initClientMultiState(c);
 202     return c;
 203 }
 204
 205 void freeFakeClient(struct redisClient *c) {
 206     sdsfree(c->querybuf);
 207     listRelease(c->reply);
 208     listRelease(c->watched_keys);
 209     freeClientMultiState(c);
 210     zfree(c);
 211 }
 212
 213 /* Replay the append log file. On error REDIS_OK is returned. On non fatal
 214  * error (the append only file is zero-length) REDIS_ERR is returned. On
 215  * fatal error an error message is logged and the program exists. */
 216 int loadAppendOnlyFile(char *filename) {
 217     struct redisClient *fakeClient;
 218     FILE *fp = fopen(filename,"r");
 219     struct redis_stat sb;
 220     int appendonly = server.appendonly;
 221
 222     if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
 223         return REDIS_ERR;
 224
 225     if (fp == NULL) {
 226         redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno));
 227         exit(1);
 228     }
 229
 230     /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
 231      * to the same file we're about to read. */
 232     server.appendonly = 0;
 233
 234     fakeClient = createFakeClient();
 235     while(1) {
 236         int argc, j;
 237         unsigned long len;
 238         robj **argv;
 239         char buf[128];
 240         sds argsds;
 241         struct redisCommand *cmd;
 242         int force_swapout;
 243
 244         if (fgets(buf,sizeof(buf),fp) == NULL) {
 245             if (feof(fp))
 246                 break;
 247             else
 248                 goto readerr;
 249         }
 250         if (buf[0] != '*') goto fmterr;
 251         argc = atoi(buf+1);
 252         argv = zmalloc(sizeof(robj*)*argc);
 253         for (j = 0; j < argc; j++) {
 254             if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr;
 255             if (buf[0] != '$') goto fmterr;
 256             len = strtol(buf+1,NULL,10);
 257             argsds = sdsnewlen(NULL,len);
 258             if (len && fread(argsds,len,1,fp) == 0) goto fmterr;
 259             argv[j] = createObject(REDIS_STRING,argsds);
 260             if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */
 261         }
 262
 263         /* Command lookup */
 264         cmd = lookupCommand(argv[0]->ptr);
 265         if (!cmd) {
 266             redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", argv[0]->ptr);
 267             exit(1);
 268         }
 269         /* Try object encoding */
 270         if (cmd->flags & REDIS_CMD_BULK)
 271             argv[argc-1] = tryObjectEncoding(argv[argc-1]);
 272         /* Run the command in the context of a fake client */
 273         fakeClient->argc = argc;
 274         fakeClient->argv = argv;
 275         cmd->proc(fakeClient);
 276
 277         /* The fake client should not have a reply */
 278         redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
 279
 280         /* Clean up, ready for the next command */
 281         for (j = 0; j < argc; j++) decrRefCount(argv[j]);
 282         zfree(argv);
 283
 284         /* Handle swapping while loading big datasets when VM is on */
 285         force_swapout = 0;
 286         if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32)
 287             force_swapout = 1;
 288
 289         if (server.vm_enabled && force_swapout) {
 290             while (zmalloc_used_memory() > server.vm_max_memory) {
 291                 if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
 292             }
 293         }
 294     }
 295
 296     /* This point can only be reached when EOF is reached without errors.
 297      * If the client is in the middle of a MULTI/EXEC, log error and quit. */
 298     if (fakeClient->flags & REDIS_MULTI) goto readerr;
 299
 300     fclose(fp);
 301     freeFakeClient(fakeClient);
 302     server.appendonly = appendonly;
 303     return REDIS_OK;
 304
 305 readerr:
 306     if (feof(fp)) {
 307         redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file");
 308     } else {
 309         redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno));
 310     }
 311     exit(1);
 312 fmterr:
 313     redisLog(REDIS_WARNING,"Bad file format reading the append only file");
 314     exit(1);
 315 }
 316
 317 /* Write binary-safe string into a file in the bulkformat
 318  * $<count>\r\n<payload>\r\n */
 319 int fwriteBulkString(FILE *fp, char *s, unsigned long len) {
 320     char cbuf[128];
 321     int clen;
 322     cbuf[0] = '$';
 323     clen = 1+ll2string(cbuf+1,sizeof(cbuf)-1,len);
 324     cbuf[clen++] = '\r';
 325     cbuf[clen++] = '\n';
 326     if (fwrite(cbuf,clen,1,fp) == 0) return 0;
 327     if (len > 0 && fwrite(s,len,1,fp) == 0) return 0;
 328     if (fwrite("\r\n",2,1,fp) == 0) return 0;
 329     return 1;
 330 }
 331
 332 /* Write a double value in bulk format $<count>\r\n<payload>\r\n */
 333 int fwriteBulkDouble(FILE *fp, double d) {
 334     char buf[128], dbuf[128];
 335
 336     snprintf(dbuf,sizeof(dbuf),"%.17g\r\n",d);
 337     snprintf(buf,sizeof(buf),"$%lu\r\n",(unsigned long)strlen(dbuf)-2);
 338     if (fwrite(buf,strlen(buf),1,fp) == 0) return 0;
 339     if (fwrite(dbuf,strlen(dbuf),1,fp) == 0) return 0;
 340     return 1;
 341 }
 342
 343 /* Write a long value in bulk format $<count>\r\n<payload>\r\n */
 344 int fwriteBulkLongLong(FILE *fp, long long l) {
 345     char bbuf[128], lbuf[128];
 346     unsigned int blen, llen;
 347     llen = ll2string(lbuf,32,l);
 348     blen = snprintf(bbuf,sizeof(bbuf),"$%u\r\n%s\r\n",llen,lbuf);
 349     if (fwrite(bbuf,blen,1,fp) == 0) return 0;
 350     return 1;
 351 }
 352
 353 /* Delegate writing an object to writing a bulk string or bulk long long. */
 354 int fwriteBulkObject(FILE *fp, robj *obj) {
 355     /* Avoid using getDecodedObject to help copy-on-write (we are often
 356      * in a child process when this function is called). */
 357     if (obj->encoding == REDIS_ENCODING_INT) {
 358         return fwriteBulkLongLong(fp,(long)obj->ptr);
 359     } else if (obj->encoding == REDIS_ENCODING_RAW) {
 360         return fwriteBulkString(fp,obj->ptr,sdslen(obj->ptr));
 361     } else {
 362         redisPanic("Unknown string encoding");
 363     }
 364 }
 365
 366 /* Write a sequence of commands able to fully rebuild the dataset into
 367  * "filename". Used both by REWRITEAOF and BGREWRITEAOF. */
 368 int rewriteAppendOnlyFile(char *filename) {
 369     dictIterator *di = NULL;
 370     dictEntry *de;
 371     FILE *fp;
 372     char tmpfile[256];
 373     int j;
 374     time_t now = time(NULL);
 375
 376     /* Note that we have to use a different temp name here compared to the
 377      * one used by rewriteAppendOnlyFileBackground() function. */
 378     snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
 379     fp = fopen(tmpfile,"w");
 380     if (!fp) {
 381         redisLog(REDIS_WARNING, "Failed rewriting the append only file: %s", strerror(errno));
 382         return REDIS_ERR;
 383     }
 384     for (j = 0; j < server.dbnum; j++) {
 385         char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
 386         redisDb *db = server.db+j;
 387         dict *d = db->dict;
 388         if (dictSize(d) == 0) continue;
 389         di = dictGetIterator(d);
 390         if (!di) {
 391             fclose(fp);
 392             return REDIS_ERR;
 393         }
 394
 395         /* SELECT the new DB */
 396         if (fwrite(selectcmd,sizeof(selectcmd)-1,1,fp) == 0) goto werr;
 397         if (fwriteBulkLongLong(fp,j) == 0) goto werr;
 398
 399         /* Iterate this DB writing every entry */
 400         while((de = dictNext(di)) != NULL) {
 401             sds keystr = dictGetEntryKey(de);
 402             robj key, *o;
 403             time_t expiretime;
 404             int swapped;
 405
 406             keystr = dictGetEntryKey(de);
 407             o = dictGetEntryVal(de);
 408             initStaticStringObject(key,keystr);
 409             /* If the value for this key is swapped, load a preview in memory.
 410              * We use a "swapped" flag to remember if we need to free the
 411              * value object instead to just increment the ref count anyway
 412              * in order to avoid copy-on-write of pages if we are forked() */
 413             if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY ||
 414                 o->storage == REDIS_VM_SWAPPING) {
 415                 swapped = 0;
 416             } else {
 417                 o = vmPreviewObject(o);
 418                 swapped = 1;
 419             }
 420             expiretime = getExpire(db,&key);
 421
 422             /* Save the key and associated value */
 423             if (o->type == REDIS_STRING) {
 424                 /* Emit a SET command */
 425                 char cmd[]="*3\r\n$3\r\nSET\r\n";
 426                 if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 427                 /* Key and value */
 428                 if (fwriteBulkObject(fp,&key) == 0) goto werr;
 429                 if (fwriteBulkObject(fp,o) == 0) goto werr;
 430             } else if (o->type == REDIS_LIST) {
 431                 /* Emit the RPUSHes needed to rebuild the list */
 432                 char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
 433                 if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 434                     unsigned char *zl = o->ptr;
 435                     unsigned char *p = ziplistIndex(zl,0);
 436                     unsigned char *vstr;
 437                     unsigned int vlen;
 438                     long long vlong;
 439
 440                     while(ziplistGet(p,&vstr,&vlen,&vlong)) {
 441                         if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 442                         if (fwriteBulkObject(fp,&key) == 0) goto werr;
 443                         if (vstr) {
 444                             if (fwriteBulkString(fp,(char*)vstr,vlen) == 0)
 445                                 goto werr;
 446                         } else {
 447                             if (fwriteBulkLongLong(fp,vlong) == 0)
 448                                 goto werr;
 449                         }
 450                         p = ziplistNext(zl,p);
 451                     }
 452                 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
 453                     list *list = o->ptr;
 454                     listNode *ln;
 455                     listIter li;
 456
 457                     listRewind(list,&li);
 458                     while((ln = listNext(&li))) {
 459                         robj *eleobj = listNodeValue(ln);
 460
 461                         if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 462                         if (fwriteBulkObject(fp,&key) == 0) goto werr;
 463                         if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
 464                     }
 465                 } else {
 466                     redisPanic("Unknown list encoding");
 467                 }
 468             } else if (o->type == REDIS_SET) {
 469                 char cmd[]="*3\r\n$4\r\nSADD\r\n";
 470
 471                 /* Emit the SADDs needed to rebuild the set */
 472                 if (o->encoding == REDIS_ENCODING_INTSET) {
 473                     int ii = 0;
 474                     int64_t llval;
 475                     while(intsetGet(o->ptr,ii++,&llval)) {
 476                         if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 477                         if (fwriteBulkObject(fp,&key) == 0) goto werr;
 478                         if (fwriteBulkLongLong(fp,llval) == 0) goto werr;
 479                     }
 480                 } else if (o->encoding == REDIS_ENCODING_HT) {
 481                     dictIterator *di = dictGetIterator(o->ptr);
 482                     dictEntry *de;
 483                     while((de = dictNext(di)) != NULL) {
 484                         robj *eleobj = dictGetEntryKey(de);
 485                         if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 486                         if (fwriteBulkObject(fp,&key) == 0) goto werr;
 487                         if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
 488                     }
 489                     dictReleaseIterator(di);
 490                 } else {
 491                     redisPanic("Unknown set encoding");
 492                 }
 493             } else if (o->type == REDIS_ZSET) {
 494                 /* Emit the ZADDs needed to rebuild the sorted set */
 495                 zset *zs = o->ptr;
 496                 dictIterator *di = dictGetIterator(zs->dict);
 497                 dictEntry *de;
 498
 499                 while((de = dictNext(di)) != NULL) {
 500                     char cmd[]="*4\r\n$4\r\nZADD\r\n";
 501                     robj *eleobj = dictGetEntryKey(de);
 502                     double *score = dictGetEntryVal(de);
 503
 504                     if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 505                     if (fwriteBulkObject(fp,&key) == 0) goto werr;
 506                     if (fwriteBulkDouble(fp,*score) == 0) goto werr;
 507                     if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
 508                 }
 509                 dictReleaseIterator(di);
 510             } else if (o->type == REDIS_HASH) {
 511                 char cmd[]="*4\r\n$4\r\nHSET\r\n";
 512
 513                 /* Emit the HSETs needed to rebuild the hash */
 514                 if (o->encoding == REDIS_ENCODING_ZIPMAP) {
 515                     unsigned char *p = zipmapRewind(o->ptr);
 516                     unsigned char *field, *val;
 517                     unsigned int flen, vlen;
 518
 519                     while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
 520                         if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 521                         if (fwriteBulkObject(fp,&key) == 0) goto werr;
 522                         if (fwriteBulkString(fp,(char*)field,flen) == 0)
 523                             goto werr;
 524                         if (fwriteBulkString(fp,(char*)val,vlen) == 0)
 525                             goto werr;
 526                     }
 527                 } else {
 528                     dictIterator *di = dictGetIterator(o->ptr);
 529                     dictEntry *de;
 530
 531                     while((de = dictNext(di)) != NULL) {
 532                         robj *field = dictGetEntryKey(de);
 533                         robj *val = dictGetEntryVal(de);
 534
 535                         if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 536                         if (fwriteBulkObject(fp,&key) == 0) goto werr;
 537                         if (fwriteBulkObject(fp,field) == 0) goto werr;
 538                         if (fwriteBulkObject(fp,val) == 0) goto werr;
 539                     }
 540                     dictReleaseIterator(di);
 541                 }
 542             } else {
 543                 redisPanic("Unknown object type");
 544             }
 545             /* Save the expire time */
 546             if (expiretime != -1) {
 547                 char cmd[]="*3\r\n$8\r\nEXPIREAT\r\n";
 548                 /* If this key is already expired skip it */
 549                 if (expiretime < now) continue;
 550                 if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
 551                 if (fwriteBulkObject(fp,&key) == 0) goto werr;
 552                 if (fwriteBulkLongLong(fp,expiretime) == 0) goto werr;
 553             }
 554             if (swapped) decrRefCount(o);
 555         }
 556         dictReleaseIterator(di);
 557     }
 558
 559     /* Make sure data will not remain on the OS's output buffers */
 560     fflush(fp);
 561     aof_fsync(fileno(fp));
 562     fclose(fp);
 563
 564     /* Use RENAME to make sure the DB file is changed atomically only
 565      * if the generate DB file is ok. */
 566     if (rename(tmpfile,filename) == -1) {
 567         redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
 568         unlink(tmpfile);
 569         return REDIS_ERR;
 570     }
 571     redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");
 572     return REDIS_OK;
 573
 574 werr:
 575     fclose(fp);
 576     unlink(tmpfile);
 577     redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
 578     if (di) dictReleaseIterator(di);
 579     return REDIS_ERR;
 580 }
 581
 582 /* This is how rewriting of the append only file in background works:
 583  *
 584  * 1) The user calls BGREWRITEAOF
 585  * 2) Redis calls this function, that forks():
 586  *    2a) the child rewrite the append only file in a temp file.
 587  *    2b) the parent accumulates differences in server.bgrewritebuf.
 588  * 3) When the child finished '2a' exists.
 589  * 4) The parent will trap the exit code, if it's OK, will append the
 590  *    data accumulated into server.bgrewritebuf into the temp file, and
 591  *    finally will rename(2) the temp file in the actual file name.
 592  *    The the new file is reopened as the new append only file. Profit!
 593  */
 594 int rewriteAppendOnlyFileBackground(void) {
 595     pid_t childpid;
 596
 597     if (server.bgrewritechildpid != -1) return REDIS_ERR;
 598     if (server.vm_enabled) waitEmptyIOJobsQueue();
 599     if ((childpid = fork()) == 0) {
 600         /* Child */
 601         char tmpfile[256];
 602
 603         if (server.vm_enabled) vmReopenSwapFile();
 604         close(server.fd);
 605         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
 606         if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
 607             _exit(0);
 608         } else {
 609             _exit(1);
 610         }
 611     } else {
 612         /* Parent */
 613         if (childpid == -1) {
 614             redisLog(REDIS_WARNING,
 615                 "Can't rewrite append only file in background: fork: %s",
 616                 strerror(errno));
 617             return REDIS_ERR;
 618         }
 619         redisLog(REDIS_NOTICE,
 620             "Background append only file rewriting started by pid %d",childpid);
 621         server.bgrewritechildpid = childpid;
 622         updateDictResizePolicy();
 623         /* We set appendseldb to -1 in order to force the next call to the
 624          * feedAppendOnlyFile() to issue a SELECT command, so the differences
 625          * accumulated by the parent into server.bgrewritebuf will start
 626          * with a SELECT statement and it will be safe to merge. */
 627         server.appendseldb = -1;
 628         return REDIS_OK;
 629     }
 630     return REDIS_OK; /* unreached */
 631 }
 632
 633 void bgrewriteaofCommand(redisClient *c) {
 634     if (server.bgrewritechildpid != -1) {
 635         addReplySds(c,sdsnew("-ERR background append only file rewriting already in progress\r\n"));
 636         return;
 637     }
 638     if (rewriteAppendOnlyFileBackground() == REDIS_OK) {
 639         char *status = "+Background append only file rewriting started\r\n";
 640         addReplySds(c,sdsnew(status));
 641     } else {
 642         addReply(c,shared.err);
 643     }
 644 }
 645
 646 void aofRemoveTempFile(pid_t childpid) {
 647     char tmpfile[256];
 648
 649     snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
 650     unlink(tmpfile);
 651 }
 652
 653 /* A background append only file rewriting (BGREWRITEAOF) terminated its work.
 654  * Handle this. */
 655 void backgroundRewriteDoneHandler(int statloc) {
 656     int exitcode = WEXITSTATUS(statloc);
 657     int bysignal = WIFSIGNALED(statloc);
 658
 659     if (!bysignal && exitcode == 0) {
 660         int fd;
 661         char tmpfile[256];
 662
 663         redisLog(REDIS_NOTICE,
 664             "Background append only file rewriting terminated with success");
 665         /* Now it's time to flush the differences accumulated by the parent */
 666         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) server.bgrewritechildpid);
 667         fd = open(tmpfile,O_WRONLY|O_APPEND);
 668         if (fd == -1) {
 669             redisLog(REDIS_WARNING, "Not able to open the temp append only file produced by the child: %s", strerror(errno));
 670             goto cleanup;
 671         }
 672         /* Flush our data... */
 673         if (write(fd,server.bgrewritebuf,sdslen(server.bgrewritebuf)) !=
 674                 (signed) sdslen(server.bgrewritebuf)) {
 675             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));
 676             close(fd);
 677             goto cleanup;
 678         }
 679         redisLog(REDIS_NOTICE,"Parent diff flushed into the new append log file with success (%lu bytes)",sdslen(server.bgrewritebuf));
 680         /* Now our work is to rename the temp file into the stable file. And
 681          * switch the file descriptor used by the server for append only. */
 682         if (rename(tmpfile,server.appendfilename) == -1) {
 683             redisLog(REDIS_WARNING,"Can't rename the temp append only file into the stable one: %s", strerror(errno));
 684             close(fd);
 685             goto cleanup;
 686         }
 687         /* Mission completed... almost */
 688         redisLog(REDIS_NOTICE,"Append only file successfully rewritten.");
 689         if (server.appendfd != -1) {
 690             /* If append only is actually enabled... */
 691             close(server.appendfd);
 692             server.appendfd = fd;
 693             if (server.appendfsync != APPENDFSYNC_NO) aof_fsync(fd);
 694             server.appendseldb = -1; /* Make sure it will issue SELECT */
 695             redisLog(REDIS_NOTICE,"The new append only file was selected for future appends.");
 696         } else {
 697             /* If append only is disabled we just generate a dump in this
 698              * format. Why not? */
 699             close(fd);
 700         }
 701     } else if (!bysignal && exitcode != 0) {
 702         redisLog(REDIS_WARNING, "Background append only file rewriting error");
 703     } else {
 704         redisLog(REDIS_WARNING,
 705             "Background append only file rewriting terminated by signal %d",
 706             WTERMSIG(statloc));
 707     }
 708 cleanup:
 709     sdsfree(server.bgrewritebuf);
 710     server.bgrewritebuf = sdsempty();
 711     aofRemoveTempFile(server.bgrewritechildpid);
 712     server.bgrewritechildpid = -1;
 713 }