1 #include "redis.h" 2 #include "bio.h" 3 #include "rio.h" 4 5 #include <signal.h> 6 #include <fcntl.h> 7 #include <sys/stat.h> 8 #include <sys/types.h> 9 #include <sys/time.h> 10 #include <sys/resource.h> 11 #include <sys/wait.h> 12 13 void aofUpdateCurrentSize(void); 14 15 void aof_background_fsync(int fd) { 16 bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL); 17 } 18 19 /* Called when the user switches from "appendonly yes" to "appendonly no" 20 * at runtime using the CONFIG command. */ 21 void stopAppendOnly(void) { 22 redisAssert(server.aof_state != REDIS_AOF_OFF); 23 flushAppendOnlyFile(1); 24 aof_fsync(server.aof_fd); 25 close(server.aof_fd); 26 27 server.aof_fd = -1; 28 server.aof_selected_db = -1; 29 server.aof_state = REDIS_AOF_OFF; 30 /* rewrite operation in progress? kill it, wait child exit */ 31 if (server.aof_child_pid != -1) { 32 int statloc; 33 34 redisLog(REDIS_NOTICE,"Killing running AOF rewrite child: %ld", 35 (long) server.aof_child_pid); 36 if (kill(server.aof_child_pid,SIGKILL) != -1) 37 wait3(&statloc,0,NULL); 38 /* reset the buffer accumulating changes while the child saves */ 39 sdsfree(server.aof_rewrite_buf); 40 server.aof_rewrite_buf = sdsempty(); 41 aofRemoveTempFile(server.aof_child_pid); 42 server.aof_child_pid = -1; 43 } 44 } 45 46 /* Called when the user switches from "appendonly no" to "appendonly yes" 47 * at runtime using the CONFIG command. */ 48 int startAppendOnly(void) { 49 server.aof_last_fsync = time(NULL); 50 server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644); 51 redisAssert(server.aof_state == REDIS_AOF_OFF); 52 if (server.aof_fd == -1) { 53 redisLog(REDIS_WARNING,"Redis needs to enable the AOF but can't open the append only file: %s",strerror(errno)); 54 return REDIS_ERR; 55 } 56 if (rewriteAppendOnlyFileBackground() == REDIS_ERR) { 57 close(server.aof_fd); 58 redisLog(REDIS_WARNING,"Redis needs to enable the AOF but can't trigger a background AOF rewrite operation. Check the above logs for more info about the error."); 59 return REDIS_ERR; 60 } 61 /* We correctly switched on AOF, now wait for the rerwite to be complete 62 * in order to append data on disk. */ 63 server.aof_state = REDIS_AOF_WAIT_REWRITE; 64 return REDIS_OK; 65 } 66 67 /* Write the append only file buffer on disk. 68 * 69 * Since we are required to write the AOF before replying to the client, 70 * and the only way the client socket can get a write is entering when the 71 * the event loop, we accumulate all the AOF writes in a memory 72 * buffer and write it on disk using this function just before entering 73 * the event loop again. 74 * 75 * About the 'force' argument: 76 * 77 * When the fsync policy is set to 'everysec' we may delay the flush if there 78 * is still an fsync() going on in the background thread, since for instance 79 * on Linux write(2) will be blocked by the background fsync anyway. 80 * When this happens we remember that there is some aof buffer to be 81 * flushed ASAP, and will try to do that in the serverCron() function. 82 * 83 * However if force is set to 1 we'll write regardless of the background 84 * fsync. */ 85 void flushAppendOnlyFile(int force) { 86 ssize_t nwritten; 87 int sync_in_progress = 0; 88 89 if (sdslen(server.aof_buf) == 0) return; 90 91 if (server.aof_fsync == AOF_FSYNC_EVERYSEC) 92 sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0; 93 94 if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) { 95 /* With this append fsync policy we do background fsyncing. 96 * If the fsync is still in progress we can try to delay 97 * the write for a couple of seconds. */ 98 if (sync_in_progress) { 99 if (server.aof_flush_postponed_start == 0) { 100 /* No previous write postponinig, remember that we are 101 * postponing the flush and return. */ 102 server.aof_flush_postponed_start = server.unixtime; 103 return; 104 } else if (server.unixtime - server.aof_flush_postponed_start < 2) { 105 /* We were already waiting for fsync to finish, but for less 106 * than two seconds this is still ok. Postpone again. */ 107 return; 108 } 109 /* Otherwise fall trough, and go write since we can't wait 110 * over two seconds. */ 111 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."); 112 } 113 } 114 /* If you are following this code path, then we are going to write so 115 * set reset the postponed flush sentinel to zero. */ 116 server.aof_flush_postponed_start = 0; 117 118 /* We want to perform a single write. This should be guaranteed atomic 119 * at least if the filesystem we are writing is a real physical one. 120 * While this will save us against the server being killed I don't think 121 * there is much to do about the whole server stopping for power problems 122 * or alike */ 123 nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf)); 124 if (nwritten != (signed)sdslen(server.aof_buf)) { 125 /* Ooops, we are in troubles. The best thing to do for now is 126 * aborting instead of giving the illusion that everything is 127 * working as expected. */ 128 if (nwritten == -1) { 129 redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno)); 130 } else { 131 redisLog(REDIS_WARNING,"Exiting on short write while writing to " 132 "the append-only file: %s (nwritten=%ld, " 133 "expected=%ld)", 134 strerror(errno), 135 (long)nwritten, 136 (long)sdslen(server.aof_buf)); 137 } 138 exit(1); 139 } 140 server.aof_current_size += nwritten; 141 142 /* Re-use AOF buffer when it is small enough. The maximum comes from the 143 * arena size of 4k minus some overhead (but is otherwise arbitrary). */ 144 if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) { 145 sdsclear(server.aof_buf); 146 } else { 147 sdsfree(server.aof_buf); 148 server.aof_buf = sdsempty(); 149 } 150 151 /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are 152 * children doing I/O in the background. */ 153 if (server.aof_no_fsync_on_rewrite && 154 (server.aof_child_pid != -1 || server.rdb_child_pid != -1)) 155 return; 156 157 /* Perform the fsync if needed. */ 158 if (server.aof_fsync == AOF_FSYNC_ALWAYS) { 159 /* aof_fsync is defined as fdatasync() for Linux in order to avoid 160 * flushing metadata. */ 161 aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */ 162 server.aof_last_fsync = server.unixtime; 163 } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC && 164 server.unixtime > server.aof_last_fsync)) { 165 if (!sync_in_progress) aof_background_fsync(server.aof_fd); 166 server.aof_last_fsync = server.unixtime; 167 } 168 } 169 170 sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) { 171 char buf[32]; 172 int len, j; 173 robj *o; 174 175 buf[0] = '*'; 176 len = 1+ll2string(buf+1,sizeof(buf)-1,argc); 177 buf[len++] = '\r'; 178 buf[len++] = '\n'; 179 dst = sdscatlen(dst,buf,len); 180 181 for (j = 0; j < argc; j++) { 182 o = getDecodedObject(argv[j]); 183 buf[0] = '$'; 184 len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr)); 185 buf[len++] = '\r'; 186 buf[len++] = '\n'; 187 dst = sdscatlen(dst,buf,len); 188 dst = sdscatlen(dst,o->ptr,sdslen(o->ptr)); 189 dst = sdscatlen(dst,"\r\n",2); 190 decrRefCount(o); 191 } 192 return dst; 193 } 194 195 /* Create the sds representation of an PEXPIREAT command, using 196 * 'seconds' as time to live and 'cmd' to understand what command 197 * we are translating into a PEXPIREAT. 198 * 199 * This command is used in order to translate EXPIRE and PEXPIRE commands 200 * into PEXPIREAT command so that we retain precision in the append only 201 * file, and the time is always absolute and not relative. */ 202 sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) { 203 long long when; 204 robj *argv[3]; 205 206 /* Make sure we can use strtol */ 207 seconds = getDecodedObject(seconds); 208 when = strtoll(seconds->ptr,NULL,10); 209 /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */ 210 if (cmd->proc == expireCommand || cmd->proc == setexCommand || 211 cmd->proc == expireatCommand) 212 { 213 when *= 1000; 214 } 215 /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */ 216 if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || 217 cmd->proc == setexCommand || cmd->proc == psetexCommand) 218 { 219 when += mstime(); 220 } 221 decrRefCount(seconds); 222 223 argv[0] = createStringObject("PEXPIREAT",9); 224 argv[1] = key; 225 argv[2] = createStringObjectFromLongLong(when); 226 buf = catAppendOnlyGenericCommand(buf, 3, argv); 227 decrRefCount(argv[0]); 228 decrRefCount(argv[2]); 229 return buf; 230 } 231 232 void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) { 233 sds buf = sdsempty(); 234 robj *tmpargv[3]; 235 236 /* The DB this command was targetting is not the same as the last command 237 * we appendend. To issue a SELECT command is needed. */ 238 if (dictid != server.aof_selected_db) { 239 char seldb[64]; 240 241 snprintf(seldb,sizeof(seldb),"%d",dictid); 242 buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n", 243 (unsigned long)strlen(seldb),seldb); 244 server.aof_selected_db = dictid; 245 } 246 247 if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || 248 cmd->proc == expireatCommand) { 249 /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */ 250 buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); 251 } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) { 252 /* Translate SETEX/PSETEX to SET and PEXPIREAT */ 253 tmpargv[0] = createStringObject("SET",3); 254 tmpargv[1] = argv[1]; 255 tmpargv[2] = argv[3]; 256 buf = catAppendOnlyGenericCommand(buf,3,tmpargv); 257 decrRefCount(tmpargv[0]); 258 buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); 259 } else { 260 /* All the other commands don't need translation or need the 261 * same translation already operated in the command vector 262 * for the replication itself. */ 263 buf = catAppendOnlyGenericCommand(buf,argc,argv); 264 } 265 266 /* Append to the AOF buffer. This will be flushed on disk just before 267 * of re-entering the event loop, so before the client will get a 268 * positive reply about the operation performed. */ 269 if (server.aof_state == REDIS_AOF_ON) 270 server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf)); 271 272 /* If a background append only file rewriting is in progress we want to 273 * accumulate the differences between the child DB and the current one 274 * in a buffer, so that when the child process will do its work we 275 * can append the differences to the new append only file. */ 276 if (server.aof_child_pid != -1) 277 server.aof_rewrite_buf = sdscatlen(server.aof_rewrite_buf,buf,sdslen(buf)); 278 279 sdsfree(buf); 280 } 281 282 /* In Redis commands are always executed in the context of a client, so in 283 * order to load the append only file we need to create a fake client. */ 284 struct redisClient *createFakeClient(void) { 285 struct redisClient *c = zmalloc(sizeof(*c)); 286 287 selectDb(c,0); 288 c->fd = -1; 289 c->querybuf = sdsempty(); 290 c->argc = 0; 291 c->argv = NULL; 292 c->bufpos = 0; 293 c->flags = 0; 294 /* We set the fake client as a slave waiting for the synchronization 295 * so that Redis will not try to send replies to this client. */ 296 c->replstate = REDIS_REPL_WAIT_BGSAVE_START; 297 c->reply = listCreate(); 298 c->reply_bytes = 0; 299 c->obuf_soft_limit_reached_time = 0; 300 c->watched_keys = listCreate(); 301 listSetFreeMethod(c->reply,decrRefCount); 302 listSetDupMethod(c->reply,dupClientReplyValue); 303 initClientMultiState(c); 304 return c; 305 } 306 307 void freeFakeClient(struct redisClient *c) { 308 sdsfree(c->querybuf); 309 listRelease(c->reply); 310 listRelease(c->watched_keys); 311 freeClientMultiState(c); 312 zfree(c); 313 } 314 315 /* Replay the append log file. On error REDIS_OK is returned. On non fatal 316 * error (the append only file is zero-length) REDIS_ERR is returned. On 317 * fatal error an error message is logged and the program exists. */ 318 int loadAppendOnlyFile(char *filename) { 319 struct redisClient *fakeClient; 320 FILE *fp = fopen(filename,"r"); 321 struct redis_stat sb; 322 int old_aof_state = server.aof_state; 323 long loops = 0; 324 325 if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) { 326 server.aof_current_size = 0; 327 fclose(fp); 328 return REDIS_ERR; 329 } 330 331 if (fp == NULL) { 332 redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno)); 333 exit(1); 334 } 335 336 /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI 337 * to the same file we're about to read. */ 338 server.aof_state = REDIS_AOF_OFF; 339 340 fakeClient = createFakeClient(); 341 startLoading(fp); 342 343 while(1) { 344 int argc, j; 345 unsigned long len; 346 robj **argv; 347 char buf[128]; 348 sds argsds; 349 struct redisCommand *cmd; 350 351 /* Serve the clients from time to time */ 352 if (!(loops++ % 1000)) { 353 loadingProgress(ftello(fp)); 354 aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); 355 } 356 357 if (fgets(buf,sizeof(buf),fp) == NULL) { 358 if (feof(fp)) 359 break; 360 else 361 goto readerr; 362 } 363 if (buf[0] != '*') goto fmterr; 364 argc = atoi(buf+1); 365 if (argc < 1) goto fmterr; 366 367 argv = zmalloc(sizeof(robj*)*argc); 368 for (j = 0; j < argc; j++) { 369 if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr; 370 if (buf[0] != '$') goto fmterr; 371 len = strtol(buf+1,NULL,10); 372 argsds = sdsnewlen(NULL,len); 373 if (len && fread(argsds,len,1,fp) == 0) goto fmterr; 374 argv[j] = createObject(REDIS_STRING,argsds); 375 if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */ 376 } 377 378 /* Command lookup */ 379 cmd = lookupCommand(argv[0]->ptr); 380 if (!cmd) { 381 redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", argv[0]->ptr); 382 exit(1); 383 } 384 /* Run the command in the context of a fake client */ 385 fakeClient->argc = argc; 386 fakeClient->argv = argv; 387 cmd->proc(fakeClient); 388 389 /* The fake client should not have a reply */ 390 redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0); 391 /* The fake client should never get blocked */ 392 redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0); 393 394 /* Clean up. Command code may have changed argv/argc so we use the 395 * argv/argc of the client instead of the local variables. */ 396 for (j = 0; j < fakeClient->argc; j++) 397 decrRefCount(fakeClient->argv[j]); 398 zfree(fakeClient->argv); 399 } 400 401 /* This point can only be reached when EOF is reached without errors. 402 * If the client is in the middle of a MULTI/EXEC, log error and quit. */ 403 if (fakeClient->flags & REDIS_MULTI) goto readerr; 404 405 fclose(fp); 406 freeFakeClient(fakeClient); 407 server.aof_state = old_aof_state; 408 stopLoading(); 409 aofUpdateCurrentSize(); 410 server.aof_rewrite_base_size = server.aof_current_size; 411 return REDIS_OK; 412 413 readerr: 414 if (feof(fp)) { 415 redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file"); 416 } else { 417 redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno)); 418 } 419 exit(1); 420 fmterr: 421 redisLog(REDIS_WARNING,"Bad file format reading the append only file: make a backup of your AOF file, then use ./redis-check-aof --fix <filename>"); 422 exit(1); 423 } 424 425 /* Delegate writing an object to writing a bulk string or bulk long long. 426 * This is not placed in rio.c since that adds the redis.h dependency. */ 427 int rioWriteBulkObject(rio *r, robj *obj) { 428 /* Avoid using getDecodedObject to help copy-on-write (we are often 429 * in a child process when this function is called). */ 430 if (obj->encoding == REDIS_ENCODING_INT) { 431 return rioWriteBulkLongLong(r,(long)obj->ptr); 432 } else if (obj->encoding == REDIS_ENCODING_RAW) { 433 return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr)); 434 } else { 435 redisPanic("Unknown string encoding"); 436 } 437 } 438 439 /* Emit the commands needed to rebuild a list object. 440 * The function returns 0 on error, 1 on success. */ 441 int rewriteListObject(rio *r, robj *key, robj *o) { 442 long long count = 0, items = listTypeLength(o); 443 444 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 445 unsigned char *zl = o->ptr; 446 unsigned char *p = ziplistIndex(zl,0); 447 unsigned char *vstr; 448 unsigned int vlen; 449 long long vlong; 450 451 while(ziplistGet(p,&vstr,&vlen,&vlong)) { 452 if (count == 0) { 453 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 454 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 455 456 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 457 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0; 458 if (rioWriteBulkObject(r,key) == 0) return 0; 459 } 460 if (vstr) { 461 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0; 462 } else { 463 if (rioWriteBulkLongLong(r,vlong) == 0) return 0; 464 } 465 p = ziplistNext(zl,p); 466 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 467 items--; 468 } 469 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 470 list *list = o->ptr; 471 listNode *ln; 472 listIter li; 473 474 listRewind(list,&li); 475 while((ln = listNext(&li))) { 476 robj *eleobj = listNodeValue(ln); 477 478 if (count == 0) { 479 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 480 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 481 482 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 483 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0; 484 if (rioWriteBulkObject(r,key) == 0) return 0; 485 } 486 if (rioWriteBulkObject(r,eleobj) == 0) return 0; 487 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 488 items--; 489 } 490 } else { 491 redisPanic("Unknown list encoding"); 492 } 493 return 1; 494 } 495 496 /* Emit the commands needed to rebuild a set object. 497 * The function returns 0 on error, 1 on success. */ 498 int rewriteSetObject(rio *r, robj *key, robj *o) { 499 long long count = 0, items = setTypeSize(o); 500 501 if (o->encoding == REDIS_ENCODING_INTSET) { 502 int ii = 0; 503 int64_t llval; 504 505 while(intsetGet(o->ptr,ii++,&llval)) { 506 if (count == 0) { 507 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 508 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 509 510 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 511 if (rioWriteBulkString(r,"SADD",4) == 0) return 0; 512 if (rioWriteBulkObject(r,key) == 0) return 0; 513 } 514 if (rioWriteBulkLongLong(r,llval) == 0) return 0; 515 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 516 items--; 517 } 518 } else if (o->encoding == REDIS_ENCODING_HT) { 519 dictIterator *di = dictGetIterator(o->ptr); 520 dictEntry *de; 521 522 while((de = dictNext(di)) != NULL) { 523 robj *eleobj = dictGetKey(de); 524 if (count == 0) { 525 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 526 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 527 528 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 529 if (rioWriteBulkString(r,"SADD",4) == 0) return 0; 530 if (rioWriteBulkObject(r,key) == 0) return 0; 531 } 532 if (rioWriteBulkObject(r,eleobj) == 0) return 0; 533 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 534 items--; 535 } 536 dictReleaseIterator(di); 537 } else { 538 redisPanic("Unknown set encoding"); 539 } 540 return 1; 541 } 542 543 /* Emit the commands needed to rebuild a sorted set object. 544 * The function returns 0 on error, 1 on success. */ 545 int rewriteSortedSetObject(rio *r, robj *key, robj *o) { 546 long long count = 0, items = zsetLength(o); 547 548 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 549 unsigned char *zl = o->ptr; 550 unsigned char *eptr, *sptr; 551 unsigned char *vstr; 552 unsigned int vlen; 553 long long vll; 554 double score; 555 556 eptr = ziplistIndex(zl,0); 557 redisAssert(eptr != NULL); 558 sptr = ziplistNext(zl,eptr); 559 redisAssert(sptr != NULL); 560 561 while (eptr != NULL) { 562 redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll)); 563 score = zzlGetScore(sptr); 564 565 if (count == 0) { 566 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 567 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 568 569 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; 570 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0; 571 if (rioWriteBulkObject(r,key) == 0) return 0; 572 } 573 if (rioWriteBulkDouble(r,score) == 0) return 0; 574 if (vstr != NULL) { 575 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0; 576 } else { 577 if (rioWriteBulkLongLong(r,vll) == 0) return 0; 578 } 579 zzlNext(zl,&eptr,&sptr); 580 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 581 items--; 582 } 583 } else if (o->encoding == REDIS_ENCODING_SKIPLIST) { 584 zset *zs = o->ptr; 585 dictIterator *di = dictGetIterator(zs->dict); 586 dictEntry *de; 587 588 while((de = dictNext(di)) != NULL) { 589 robj *eleobj = dictGetKey(de); 590 double *score = dictGetVal(de); 591 592 if (count == 0) { 593 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 594 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 595 596 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; 597 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0; 598 if (rioWriteBulkObject(r,key) == 0) return 0; 599 } 600 if (rioWriteBulkDouble(r,*score) == 0) return 0; 601 if (rioWriteBulkObject(r,eleobj) == 0) return 0; 602 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 603 items--; 604 } 605 dictReleaseIterator(di); 606 } else { 607 redisPanic("Unknown sorted zset encoding"); 608 } 609 return 1; 610 } 611 612 /* Write either the key or the value of the currently selected item of an hash. 613 * The 'hi' argument passes a valid Redis hash iterator. 614 * The 'what' filed specifies if to write a key or a value and can be 615 * either REDIS_HASH_KEY or REDIS_HASH_VALUE. 616 * 617 * The function returns 0 on error, non-zero on success. */ 618 static int rioWriteHashIteratorCursor(rio *r, hashTypeIterator *hi, int what) { 619 if (hi->encoding == REDIS_ENCODING_ZIPLIST) { 620 unsigned char *vstr = NULL; 621 unsigned int vlen = UINT_MAX; 622 long long vll = LLONG_MAX; 623 624 hashTypeCurrentFromZiplist(hi, what, &vstr, &vlen, &vll); 625 if (vstr) { 626 return rioWriteBulkString(r, (char*)vstr, vlen); 627 } else { 628 return rioWriteBulkLongLong(r, vll); 629 } 630 631 } else if (hi->encoding == REDIS_ENCODING_HT) { 632 robj *value; 633 634 hashTypeCurrentFromHashTable(hi, what, &value); 635 return rioWriteBulkObject(r, value); 636 } 637 638 redisPanic("Unknown hash encoding"); 639 return 0; 640 } 641 642 /* Emit the commands needed to rebuild a hash object. 643 * The function returns 0 on error, 1 on success. */ 644 int rewriteHashObject(rio *r, robj *key, robj *o) { 645 hashTypeIterator *hi; 646 long long count = 0, items = hashTypeLength(o); 647 648 hi = hashTypeInitIterator(o); 649 while (hashTypeNext(hi) != REDIS_ERR) { 650 if (count == 0) { 651 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 652 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 653 654 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; 655 if (rioWriteBulkString(r,"HMSET",5) == 0) return 0; 656 if (rioWriteBulkObject(r,key) == 0) return 0; 657 } 658 659 if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_KEY) == 0) return 0; 660 if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_VALUE) == 0) return 0; 661 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 662 items--; 663 } 664 665 hashTypeReleaseIterator(hi); 666 667 return 1; 668 } 669 670 /* Write a sequence of commands able to fully rebuild the dataset into 671 * "filename". Used both by REWRITEAOF and BGREWRITEAOF. 672 * 673 * In order to minimize the number of commands needed in the rewritten 674 * log Redis uses variadic commands when possible, such as RPUSH, SADD 675 * and ZADD. However at max REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time 676 * are inserted using a single command. */ 677 int rewriteAppendOnlyFile(char *filename) { 678 dictIterator *di = NULL; 679 dictEntry *de; 680 rio aof; 681 FILE *fp; 682 char tmpfile[256]; 683 int j; 684 long long now = mstime(); 685 686 /* Note that we have to use a different temp name here compared to the 687 * one used by rewriteAppendOnlyFileBackground() function. */ 688 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid()); 689 fp = fopen(tmpfile,"w"); 690 if (!fp) { 691 redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno)); 692 return REDIS_ERR; 693 } 694 695 rioInitWithFile(&aof,fp); 696 for (j = 0; j < server.dbnum; j++) { 697 char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n"; 698 redisDb *db = server.db+j; 699 dict *d = db->dict; 700 if (dictSize(d) == 0) continue; 701 di = dictGetSafeIterator(d); 702 if (!di) { 703 fclose(fp); 704 return REDIS_ERR; 705 } 706 707 /* SELECT the new DB */ 708 if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr; 709 if (rioWriteBulkLongLong(&aof,j) == 0) goto werr; 710 711 /* Iterate this DB writing every entry */ 712 while((de = dictNext(di)) != NULL) { 713 sds keystr; 714 robj key, *o; 715 long long expiretime; 716 717 keystr = dictGetKey(de); 718 o = dictGetVal(de); 719 initStaticStringObject(key,keystr); 720 721 expiretime = getExpire(db,&key); 722 723 /* Save the key and associated value */ 724 if (o->type == REDIS_STRING) { 725 /* Emit a SET command */ 726 char cmd[]="*3\r\n$3\r\nSET\r\n"; 727 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; 728 /* Key and value */ 729 if (rioWriteBulkObject(&aof,&key) == 0) goto werr; 730 if (rioWriteBulkObject(&aof,o) == 0) goto werr; 731 } else if (o->type == REDIS_LIST) { 732 if (rewriteListObject(&aof,&key,o) == 0) goto werr; 733 } else if (o->type == REDIS_SET) { 734 if (rewriteSetObject(&aof,&key,o) == 0) goto werr; 735 } else if (o->type == REDIS_ZSET) { 736 if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr; 737 } else if (o->type == REDIS_HASH) { 738 if (rewriteHashObject(&aof,&key,o) == 0) goto werr; 739 } else { 740 redisPanic("Unknown object type"); 741 } 742 /* Save the expire time */ 743 if (expiretime != -1) { 744 char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n"; 745 /* If this key is already expired skip it */ 746 if (expiretime < now) continue; 747 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; 748 if (rioWriteBulkObject(&aof,&key) == 0) goto werr; 749 if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr; 750 } 751 } 752 dictReleaseIterator(di); 753 } 754 755 /* Make sure data will not remain on the OS's output buffers */ 756 fflush(fp); 757 aof_fsync(fileno(fp)); 758 fclose(fp); 759 760 /* Use RENAME to make sure the DB file is changed atomically only 761 * if the generate DB file is ok. */ 762 if (rename(tmpfile,filename) == -1) { 763 redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno)); 764 unlink(tmpfile); 765 return REDIS_ERR; 766 } 767 redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed"); 768 return REDIS_OK; 769 770 werr: 771 fclose(fp); 772 unlink(tmpfile); 773 redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno)); 774 if (di) dictReleaseIterator(di); 775 return REDIS_ERR; 776 } 777 778 /* This is how rewriting of the append only file in background works: 779 * 780 * 1) The user calls BGREWRITEAOF 781 * 2) Redis calls this function, that forks(): 782 * 2a) the child rewrite the append only file in a temp file. 783 * 2b) the parent accumulates differences in server.aof_rewrite_buf. 784 * 3) When the child finished '2a' exists. 785 * 4) The parent will trap the exit code, if it's OK, will append the 786 * data accumulated into server.aof_rewrite_buf into the temp file, and 787 * finally will rename(2) the temp file in the actual file name. 788 * The the new file is reopened as the new append only file. Profit! 789 */ 790 int rewriteAppendOnlyFileBackground(void) { 791 pid_t childpid; 792 long long start; 793 794 if (server.aof_child_pid != -1) return REDIS_ERR; 795 start = ustime(); 796 if ((childpid = fork()) == 0) { 797 char tmpfile[256]; 798 799 /* Child */ 800 if (server.ipfd > 0) close(server.ipfd); 801 if (server.sofd > 0) close(server.sofd); 802 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); 803 if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) { 804 _exit(0); 805 } else { 806 _exit(1); 807 } 808 } else { 809 /* Parent */ 810 server.stat_fork_time = ustime()-start; 811 if (childpid == -1) { 812 redisLog(REDIS_WARNING, 813 "Can't rewrite append only file in background: fork: %s", 814 strerror(errno)); 815 return REDIS_ERR; 816 } 817 redisLog(REDIS_NOTICE, 818 "Background append only file rewriting started by pid %d",childpid); 819 server.aof_rewrite_scheduled = 0; 820 server.aof_child_pid = childpid; 821 updateDictResizePolicy(); 822 /* We set appendseldb to -1 in order to force the next call to the 823 * feedAppendOnlyFile() to issue a SELECT command, so the differences 824 * accumulated by the parent into server.aof_rewrite_buf will start 825 * with a SELECT statement and it will be safe to merge. */ 826 server.aof_selected_db = -1; 827 return REDIS_OK; 828 } 829 return REDIS_OK; /* unreached */ 830 } 831 832 void bgrewriteaofCommand(redisClient *c) { 833 if (server.aof_child_pid != -1) { 834 addReplyError(c,"Background append only file rewriting already in progress"); 835 } else if (server.rdb_child_pid != -1) { 836 server.aof_rewrite_scheduled = 1; 837 addReplyStatus(c,"Background append only file rewriting scheduled"); 838 } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) { 839 addReplyStatus(c,"Background append only file rewriting started"); 840 } else { 841 addReply(c,shared.err); 842 } 843 } 844 845 void aofRemoveTempFile(pid_t childpid) { 846 char tmpfile[256]; 847 848 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid); 849 unlink(tmpfile); 850 } 851 852 /* Update the server.aof_current_size filed explicitly using stat(2) 853 * to check the size of the file. This is useful after a rewrite or after 854 * a restart, normally the size is updated just adding the write length 855 * to the current length, that is much faster. */ 856 void aofUpdateCurrentSize(void) { 857 struct redis_stat sb; 858 859 if (redis_fstat(server.aof_fd,&sb) == -1) { 860 redisLog(REDIS_WARNING,"Unable to obtain the AOF file length. stat: %s", 861 strerror(errno)); 862 } else { 863 server.aof_current_size = sb.st_size; 864 } 865 } 866 867 /* A background append only file rewriting (BGREWRITEAOF) terminated its work. 868 * Handle this. */ 869 void backgroundRewriteDoneHandler(int exitcode, int bysignal) { 870 if (!bysignal && exitcode == 0) { 871 int newfd, oldfd; 872 int nwritten; 873 char tmpfile[256]; 874 long long now = ustime(); 875 876 redisLog(REDIS_NOTICE, 877 "Background AOF rewrite terminated with success"); 878 879 /* Flush the differences accumulated by the parent to the 880 * rewritten AOF. */ 881 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", 882 (int)server.aof_child_pid); 883 newfd = open(tmpfile,O_WRONLY|O_APPEND); 884 if (newfd == -1) { 885 redisLog(REDIS_WARNING, 886 "Unable to open the temporary AOF produced by the child: %s", strerror(errno)); 887 goto cleanup; 888 } 889 890 nwritten = write(newfd,server.aof_rewrite_buf,sdslen(server.aof_rewrite_buf)); 891 if (nwritten != (signed)sdslen(server.aof_rewrite_buf)) { 892 if (nwritten == -1) { 893 redisLog(REDIS_WARNING, 894 "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); 895 } else { 896 redisLog(REDIS_WARNING, 897 "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); 898 } 899 close(newfd); 900 goto cleanup; 901 } 902 903 redisLog(REDIS_NOTICE, 904 "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten); 905 906 /* The only remaining thing to do is to rename the temporary file to 907 * the configured file and switch the file descriptor used to do AOF 908 * writes. We don't want close(2) or rename(2) calls to block the 909 * server on old file deletion. 910 * 911 * There are two possible scenarios: 912 * 913 * 1) AOF is DISABLED and this was a one time rewrite. The temporary 914 * file will be renamed to the configured file. When this file already 915 * exists, it will be unlinked, which may block the server. 916 * 917 * 2) AOF is ENABLED and the rewritten AOF will immediately start 918 * receiving writes. After the temporary file is renamed to the 919 * configured file, the original AOF file descriptor will be closed. 920 * Since this will be the last reference to that file, closing it 921 * causes the underlying file to be unlinked, which may block the 922 * server. 923 * 924 * To mitigate the blocking effect of the unlink operation (either 925 * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we 926 * use a background thread to take care of this. First, we 927 * make scenario 1 identical to scenario 2 by opening the target file 928 * when it exists. The unlink operation after the rename(2) will then 929 * be executed upon calling close(2) for its descriptor. Everything to 930 * guarantee atomicity for this switch has already happened by then, so 931 * we don't care what the outcome or duration of that close operation 932 * is, as long as the file descriptor is released again. */ 933 if (server.aof_fd == -1) { 934 /* AOF disabled */ 935 936 /* Don't care if this fails: oldfd will be -1 and we handle that. 937 * One notable case of -1 return is if the old file does 938 * not exist. */ 939 oldfd = open(server.aof_filename,O_RDONLY|O_NONBLOCK); 940 } else { 941 /* AOF enabled */ 942 oldfd = -1; /* We'll set this to the current AOF filedes later. */ 943 } 944 945 /* Rename the temporary file. This will not unlink the target file if 946 * it exists, because we reference it with "oldfd". */ 947 if (rename(tmpfile,server.aof_filename) == -1) { 948 redisLog(REDIS_WARNING, 949 "Error trying to rename the temporary AOF file: %s", strerror(errno)); 950 close(newfd); 951 if (oldfd != -1) close(oldfd); 952 goto cleanup; 953 } 954 955 if (server.aof_fd == -1) { 956 /* AOF disabled, we don't need to set the AOF file descriptor 957 * to this new file, so we can close it. */ 958 close(newfd); 959 } else { 960 /* AOF enabled, replace the old fd with the new one. */ 961 oldfd = server.aof_fd; 962 server.aof_fd = newfd; 963 if (server.aof_fsync == AOF_FSYNC_ALWAYS) 964 aof_fsync(newfd); 965 else if (server.aof_fsync == AOF_FSYNC_EVERYSEC) 966 aof_background_fsync(newfd); 967 server.aof_selected_db = -1; /* Make sure SELECT is re-issued */ 968 aofUpdateCurrentSize(); 969 server.aof_rewrite_base_size = server.aof_current_size; 970 971 /* Clear regular AOF buffer since its contents was just written to 972 * the new AOF from the background rewrite buffer. */ 973 sdsfree(server.aof_buf); 974 server.aof_buf = sdsempty(); 975 } 976 977 redisLog(REDIS_NOTICE, "Background AOF rewrite finished successfully"); 978 /* Change state from WAIT_REWRITE to ON if needed */ 979 if (server.aof_state == REDIS_AOF_WAIT_REWRITE) 980 server.aof_state = REDIS_AOF_ON; 981 982 /* Asynchronously close the overwritten AOF. */ 983 if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL); 984 985 redisLog(REDIS_VERBOSE, 986 "Background AOF rewrite signal handler took %lldus", ustime()-now); 987 } else if (!bysignal && exitcode != 0) { 988 redisLog(REDIS_WARNING, 989 "Background AOF rewrite terminated with error"); 990 } else { 991 redisLog(REDIS_WARNING, 992 "Background AOF rewrite terminated by signal %d", bysignal); 993 } 994 995 cleanup: 996 sdsfree(server.aof_rewrite_buf); 997 server.aof_rewrite_buf = sdsempty(); 998 aofRemoveTempFile(server.aof_child_pid); 999 server.aof_child_pid = -1; 1000 /* Schedule a new rewrite if we are waiting for it to switch the AOF ON. */ 1001 if (server.aof_state == REDIS_AOF_WAIT_REWRITE) 1002 server.aof_rewrite_scheduled = 1; 1003 } 1004