1 /* 2 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions are met: 7 * 8 * * Redistributions of source code must retain the above copyright notice, 9 * this list of conditions and the following disclaimer. 10 * * Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * * Neither the name of Redis nor the names of its contributors may be used 14 * to endorse or promote products derived from this software without 15 * specific prior written permission. 16 * 17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27 * POSSIBILITY OF SUCH DAMAGE. 28 */ 29 30 #include "redis.h" 31 #include "bio.h" 32 #include "rio.h" 33 34 #include <signal.h> 35 #include <fcntl.h> 36 #include <sys/stat.h> 37 #include <sys/types.h> 38 #include <sys/time.h> 39 #include <sys/resource.h> 40 #include <sys/wait.h> 41 42 void aofUpdateCurrentSize(void); 43 void aofClosePipes(void); 44 45 /* ---------------------------------------------------------------------------- 46 * AOF rewrite buffer implementation. 47 * 48 * The following code implement a simple buffer used in order to accumulate 49 * changes while the background process is rewriting the AOF file. 50 * 51 * We only need to append, but can't just use realloc with a large block 52 * because 'huge' reallocs are not always handled as one could expect 53 * (via remapping of pages at OS level) but may involve copying data. 54 * 55 * For this reason we use a list of blocks, every block is 56 * AOF_RW_BUF_BLOCK_SIZE bytes. 57 * ------------------------------------------------------------------------- */ 58 59 #define AOF_RW_BUF_BLOCK_SIZE (1024*1024*10) /* 10 MB per block */ 60 61 typedef struct aofrwblock { 62 unsigned long used, free; 63 char buf[AOF_RW_BUF_BLOCK_SIZE]; 64 } aofrwblock; 65 66 /* This function free the old AOF rewrite buffer if needed, and initialize 67 * a fresh new one. It tests for server.aof_rewrite_buf_blocks equal to NULL 68 * so can be used for the first initialization as well. */ 69 void aofRewriteBufferReset(void) { 70 if (server.aof_rewrite_buf_blocks) 71 listRelease(server.aof_rewrite_buf_blocks); 72 73 server.aof_rewrite_buf_blocks = listCreate(); 74 listSetFreeMethod(server.aof_rewrite_buf_blocks,zfree); 75 } 76 77 /* Return the current size of the AOF rerwite buffer. */ 78 unsigned long aofRewriteBufferSize(void) { 79 listNode *ln; 80 listIter li; 81 unsigned long size = 0; 82 83 listRewind(server.aof_rewrite_buf_blocks,&li); 84 while((ln = listNext(&li))) { 85 aofrwblock *block = listNodeValue(ln); 86 size += block->used; 87 } 88 return size; 89 } 90 91 /* Event handler used to send data to the child process doing the AOF 92 * rewrite. We send pieces of our AOF differences buffer so that the final 93 * write when the child finishes the rewrite will be small. */ 94 void aofChildWriteDiffData(aeEventLoop *el, int fd, void *privdata, int mask) { 95 listNode *ln; 96 aofrwblock *block; 97 ssize_t nwritten; 98 99 while(1) { 100 ln = listFirst(server.aof_rewrite_buf_blocks); 101 block = ln ? ln->value : NULL; 102 if (server.aof_stop_sending_diff || !block) { 103 aeDeleteFileEvent(server.el,server.aof_pipe_write_data_to_child, 104 AE_WRITABLE); 105 return; 106 } 107 if (block->used > 0) { 108 nwritten = write(server.aof_pipe_write_data_to_child, 109 block->buf,block->used); 110 if (nwritten <= 0) return; 111 memmove(block->buf,block->buf+nwritten,block->used-nwritten); 112 block->used -= nwritten; 113 } 114 if (block->used == 0) listDelNode(server.aof_rewrite_buf_blocks,ln); 115 } 116 } 117 118 /* Append data to the AOF rewrite buffer, allocating new blocks if needed. */ 119 void aofRewriteBufferAppend(unsigned char *s, unsigned long len) { 120 listNode *ln = listLast(server.aof_rewrite_buf_blocks); 121 aofrwblock *block = ln ? ln->value : NULL; 122 123 while(len) { 124 /* If we already got at least an allocated block, try appending 125 * at least some piece into it. */ 126 if (block) { 127 unsigned long thislen = (block->free < len) ? block->free : len; 128 if (thislen) { /* The current block is not already full. */ 129 memcpy(block->buf+block->used, s, thislen); 130 block->used += thislen; 131 block->free -= thislen; 132 s += thislen; 133 len -= thislen; 134 } 135 } 136 137 if (len) { /* First block to allocate, or need another block. */ 138 int numblocks; 139 140 block = zmalloc(sizeof(*block)); 141 block->free = AOF_RW_BUF_BLOCK_SIZE; 142 block->used = 0; 143 listAddNodeTail(server.aof_rewrite_buf_blocks,block); 144 145 /* Log every time we cross more 10 or 100 blocks, respectively 146 * as a notice or warning. */ 147 numblocks = listLength(server.aof_rewrite_buf_blocks); 148 if (((numblocks+1) % 10) == 0) { 149 int level = ((numblocks+1) % 100) == 0 ? REDIS_WARNING : 150 REDIS_NOTICE; 151 redisLog(level,"Background AOF buffer size: %lu MB", 152 aofRewriteBufferSize()/(1024*1024)); 153 } 154 } 155 } 156 157 /* Install a file event to send data to the rewrite child if there is 158 * not one already. */ 159 if (aeGetFileEvents(server.el,server.aof_pipe_write_data_to_child) == 0) { 160 aeCreateFileEvent(server.el, server.aof_pipe_write_data_to_child, 161 AE_WRITABLE, aofChildWriteDiffData, NULL); 162 } 163 } 164 165 /* Write the buffer (possibly composed of multiple blocks) into the specified 166 * fd. If a short write or any other error happens -1 is returned, 167 * otherwise the number of bytes written is returned. */ 168 ssize_t aofRewriteBufferWrite(int fd) { 169 listNode *ln; 170 listIter li; 171 ssize_t count = 0; 172 173 listRewind(server.aof_rewrite_buf_blocks,&li); 174 while((ln = listNext(&li))) { 175 aofrwblock *block = listNodeValue(ln); 176 ssize_t nwritten; 177 178 if (block->used) { 179 nwritten = write(fd,block->buf,block->used); 180 if (nwritten != block->used) { 181 if (nwritten == 0) errno = EIO; 182 return -1; 183 } 184 count += nwritten; 185 } 186 } 187 return count; 188 } 189 190 /* ---------------------------------------------------------------------------- 191 * AOF file implementation 192 * ------------------------------------------------------------------------- */ 193 194 /* Starts a background task that performs fsync() against the specified 195 * file descriptor (the one of the AOF file) in another thread. */ 196 void aof_background_fsync(int fd) { 197 bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL); 198 } 199 200 /* Called when the user switches from "appendonly yes" to "appendonly no" 201 * at runtime using the CONFIG command. */ 202 void stopAppendOnly(void) { 203 redisAssert(server.aof_state != REDIS_AOF_OFF); 204 flushAppendOnlyFile(1); 205 aof_fsync(server.aof_fd); 206 close(server.aof_fd); 207 208 server.aof_fd = -1; 209 server.aof_selected_db = -1; 210 server.aof_state = REDIS_AOF_OFF; 211 /* rewrite operation in progress? kill it, wait child exit */ 212 if (server.aof_child_pid != -1) { 213 int statloc; 214 215 redisLog(REDIS_NOTICE,"Killing running AOF rewrite child: %ld", 216 (long) server.aof_child_pid); 217 if (kill(server.aof_child_pid,SIGUSR1) != -1) 218 wait3(&statloc,0,NULL); 219 /* reset the buffer accumulating changes while the child saves */ 220 aofRewriteBufferReset(); 221 aofRemoveTempFile(server.aof_child_pid); 222 server.aof_child_pid = -1; 223 server.aof_rewrite_time_start = -1; 224 /* close pipes used for IPC between the two processes. */ 225 aofClosePipes(); 226 } 227 } 228 229 /* Called when the user switches from "appendonly no" to "appendonly yes" 230 * at runtime using the CONFIG command. */ 231 int startAppendOnly(void) { 232 server.aof_last_fsync = server.unixtime; 233 server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644); 234 redisAssert(server.aof_state == REDIS_AOF_OFF); 235 if (server.aof_fd == -1) { 236 redisLog(REDIS_WARNING,"Redis needs to enable the AOF but can't open the append only file: %s",strerror(errno)); 237 return REDIS_ERR; 238 } 239 if (rewriteAppendOnlyFileBackground() == REDIS_ERR) { 240 close(server.aof_fd); 241 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."); 242 return REDIS_ERR; 243 } 244 /* We correctly switched on AOF, now wait for the rerwite to be complete 245 * in order to append data on disk. */ 246 server.aof_state = REDIS_AOF_WAIT_REWRITE; 247 return REDIS_OK; 248 } 249 250 /* Write the append only file buffer on disk. 251 * 252 * Since we are required to write the AOF before replying to the client, 253 * and the only way the client socket can get a write is entering when the 254 * the event loop, we accumulate all the AOF writes in a memory 255 * buffer and write it on disk using this function just before entering 256 * the event loop again. 257 * 258 * About the 'force' argument: 259 * 260 * When the fsync policy is set to 'everysec' we may delay the flush if there 261 * is still an fsync() going on in the background thread, since for instance 262 * on Linux write(2) will be blocked by the background fsync anyway. 263 * When this happens we remember that there is some aof buffer to be 264 * flushed ASAP, and will try to do that in the serverCron() function. 265 * 266 * However if force is set to 1 we'll write regardless of the background 267 * fsync. */ 268 #define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */ 269 void flushAppendOnlyFile(int force) { 270 ssize_t nwritten; 271 int sync_in_progress = 0; 272 mstime_t latency; 273 274 if (sdslen(server.aof_buf) == 0) return; 275 276 if (server.aof_fsync == AOF_FSYNC_EVERYSEC) 277 sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0; 278 279 if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) { 280 /* With this append fsync policy we do background fsyncing. 281 * If the fsync is still in progress we can try to delay 282 * the write for a couple of seconds. */ 283 if (sync_in_progress) { 284 if (server.aof_flush_postponed_start == 0) { 285 /* No previous write postponinig, remember that we are 286 * postponing the flush and return. */ 287 server.aof_flush_postponed_start = server.unixtime; 288 return; 289 } else if (server.unixtime - server.aof_flush_postponed_start < 2) { 290 /* We were already waiting for fsync to finish, but for less 291 * than two seconds this is still ok. Postpone again. */ 292 return; 293 } 294 /* Otherwise fall trough, and go write since we can't wait 295 * over two seconds. */ 296 server.aof_delayed_fsync++; 297 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."); 298 } 299 } 300 /* We want to perform a single write. This should be guaranteed atomic 301 * at least if the filesystem we are writing is a real physical one. 302 * While this will save us against the server being killed I don't think 303 * there is much to do about the whole server stopping for power problems 304 * or alike */ 305 306 latencyStartMonitor(latency); 307 nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf)); 308 latencyEndMonitor(latency); 309 /* We want to capture different events for delayed writes: 310 * when the delay happens with a pending fsync, or with a saving child 311 * active, and when the above two conditions are missing. 312 * We also use an additional event name to save all samples which is 313 * useful for graphing / monitoring purposes. */ 314 if (sync_in_progress) { 315 latencyAddSampleIfNeeded("aof-write-pending-fsync",latency); 316 } else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) { 317 latencyAddSampleIfNeeded("aof-write-active-child",latency); 318 } else { 319 latencyAddSampleIfNeeded("aof-write-alone",latency); 320 } 321 latencyAddSampleIfNeeded("aof-write",latency); 322 323 /* We performed the write so reset the postponed flush sentinel to zero. */ 324 server.aof_flush_postponed_start = 0; 325 326 if (nwritten != (signed)sdslen(server.aof_buf)) { 327 static time_t last_write_error_log = 0; 328 int can_log = 0; 329 330 /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */ 331 if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) { 332 can_log = 1; 333 last_write_error_log = server.unixtime; 334 } 335 336 /* Lof the AOF write error and record the error code. */ 337 if (nwritten == -1) { 338 if (can_log) { 339 redisLog(REDIS_WARNING,"Error writing to the AOF file: %s", 340 strerror(errno)); 341 server.aof_last_write_errno = errno; 342 } 343 } else { 344 if (can_log) { 345 redisLog(REDIS_WARNING,"Short write while writing to " 346 "the AOF file: (nwritten=%lld, " 347 "expected=%lld)", 348 (long long)nwritten, 349 (long long)sdslen(server.aof_buf)); 350 } 351 352 if (ftruncate(server.aof_fd, server.aof_current_size) == -1) { 353 if (can_log) { 354 redisLog(REDIS_WARNING, "Could not remove short write " 355 "from the append-only file. Redis may refuse " 356 "to load the AOF the next time it starts. " 357 "ftruncate: %s", strerror(errno)); 358 } 359 } else { 360 /* If the ftrunacate() succeeded we can set nwritten to 361 * -1 since there is no longer partial data into the AOF. */ 362 nwritten = -1; 363 } 364 server.aof_last_write_errno = ENOSPC; 365 } 366 367 /* Handle the AOF write error. */ 368 if (server.aof_fsync == AOF_FSYNC_ALWAYS) { 369 /* We can't recover when the fsync policy is ALWAYS since the 370 * reply for the client is already in the output buffers, and we 371 * have the contract with the user that on acknowledged write data 372 * is synched on disk. */ 373 redisLog(REDIS_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting..."); 374 exit(1); 375 } else { 376 /* Recover from failed write leaving data into the buffer. However 377 * set an error to stop accepting writes as long as the error 378 * condition is not cleared. */ 379 server.aof_last_write_status = REDIS_ERR; 380 381 /* Trim the sds buffer if there was a partial write, and there 382 * was no way to undo it with ftruncate(2). */ 383 if (nwritten > 0) { 384 server.aof_current_size += nwritten; 385 sdsrange(server.aof_buf,nwritten,-1); 386 } 387 return; /* We'll try again on the next call... */ 388 } 389 } else { 390 /* Successful write(2). If AOF was in error state, restore the 391 * OK state and log the event. */ 392 if (server.aof_last_write_status == REDIS_ERR) { 393 redisLog(REDIS_WARNING, 394 "AOF write error looks solved, Redis can write again."); 395 server.aof_last_write_status = REDIS_OK; 396 } 397 } 398 server.aof_current_size += nwritten; 399 400 /* Re-use AOF buffer when it is small enough. The maximum comes from the 401 * arena size of 4k minus some overhead (but is otherwise arbitrary). */ 402 if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) { 403 sdsclear(server.aof_buf); 404 } else { 405 sdsfree(server.aof_buf); 406 server.aof_buf = sdsempty(); 407 } 408 409 /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are 410 * children doing I/O in the background. */ 411 if (server.aof_no_fsync_on_rewrite && 412 (server.aof_child_pid != -1 || server.rdb_child_pid != -1)) 413 return; 414 415 /* Perform the fsync if needed. */ 416 if (server.aof_fsync == AOF_FSYNC_ALWAYS) { 417 /* aof_fsync is defined as fdatasync() for Linux in order to avoid 418 * flushing metadata. */ 419 latencyStartMonitor(latency); 420 aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */ 421 latencyEndMonitor(latency); 422 latencyAddSampleIfNeeded("aof-fsync-always",latency); 423 server.aof_last_fsync = server.unixtime; 424 } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC && 425 server.unixtime > server.aof_last_fsync)) { 426 if (!sync_in_progress) aof_background_fsync(server.aof_fd); 427 server.aof_last_fsync = server.unixtime; 428 } 429 } 430 431 sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) { 432 char buf[32]; 433 int len, j; 434 robj *o; 435 436 buf[0] = '*'; 437 len = 1+ll2string(buf+1,sizeof(buf)-1,argc); 438 buf[len++] = '\r'; 439 buf[len++] = '\n'; 440 dst = sdscatlen(dst,buf,len); 441 442 for (j = 0; j < argc; j++) { 443 o = getDecodedObject(argv[j]); 444 buf[0] = '$'; 445 len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr)); 446 buf[len++] = '\r'; 447 buf[len++] = '\n'; 448 dst = sdscatlen(dst,buf,len); 449 dst = sdscatlen(dst,o->ptr,sdslen(o->ptr)); 450 dst = sdscatlen(dst,"\r\n",2); 451 decrRefCount(o); 452 } 453 return dst; 454 } 455 456 /* Create the sds representation of an PEXPIREAT command, using 457 * 'seconds' as time to live and 'cmd' to understand what command 458 * we are translating into a PEXPIREAT. 459 * 460 * This command is used in order to translate EXPIRE and PEXPIRE commands 461 * into PEXPIREAT command so that we retain precision in the append only 462 * file, and the time is always absolute and not relative. */ 463 sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) { 464 long long when; 465 robj *argv[3]; 466 467 /* Make sure we can use strtol */ 468 seconds = getDecodedObject(seconds); 469 when = strtoll(seconds->ptr,NULL,10); 470 /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */ 471 if (cmd->proc == expireCommand || cmd->proc == setexCommand || 472 cmd->proc == expireatCommand) 473 { 474 when *= 1000; 475 } 476 /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */ 477 if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || 478 cmd->proc == setexCommand || cmd->proc == psetexCommand) 479 { 480 when += mstime(); 481 } 482 decrRefCount(seconds); 483 484 argv[0] = createStringObject("PEXPIREAT",9); 485 argv[1] = key; 486 argv[2] = createStringObjectFromLongLong(when); 487 buf = catAppendOnlyGenericCommand(buf, 3, argv); 488 decrRefCount(argv[0]); 489 decrRefCount(argv[2]); 490 return buf; 491 } 492 493 void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) { 494 sds buf = sdsempty(); 495 robj *tmpargv[3]; 496 497 /* The DB this command was targeting is not the same as the last command 498 * we appendend. To issue a SELECT command is needed. */ 499 if (dictid != server.aof_selected_db) { 500 char seldb[64]; 501 502 snprintf(seldb,sizeof(seldb),"%d",dictid); 503 buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n", 504 (unsigned long)strlen(seldb),seldb); 505 server.aof_selected_db = dictid; 506 } 507 508 if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || 509 cmd->proc == expireatCommand) { 510 /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */ 511 buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); 512 } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) { 513 /* Translate SETEX/PSETEX to SET and PEXPIREAT */ 514 tmpargv[0] = createStringObject("SET",3); 515 tmpargv[1] = argv[1]; 516 tmpargv[2] = argv[3]; 517 buf = catAppendOnlyGenericCommand(buf,3,tmpargv); 518 decrRefCount(tmpargv[0]); 519 buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); 520 } else { 521 /* All the other commands don't need translation or need the 522 * same translation already operated in the command vector 523 * for the replication itself. */ 524 buf = catAppendOnlyGenericCommand(buf,argc,argv); 525 } 526 527 /* Append to the AOF buffer. This will be flushed on disk just before 528 * of re-entering the event loop, so before the client will get a 529 * positive reply about the operation performed. */ 530 if (server.aof_state == REDIS_AOF_ON) 531 server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf)); 532 533 /* If a background append only file rewriting is in progress we want to 534 * accumulate the differences between the child DB and the current one 535 * in a buffer, so that when the child process will do its work we 536 * can append the differences to the new append only file. */ 537 if (server.aof_child_pid != -1) 538 aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf)); 539 540 sdsfree(buf); 541 } 542 543 /* ---------------------------------------------------------------------------- 544 * AOF loading 545 * ------------------------------------------------------------------------- */ 546 547 /* In Redis commands are always executed in the context of a client, so in 548 * order to load the append only file we need to create a fake client. */ 549 struct redisClient *createFakeClient(void) { 550 struct redisClient *c = zmalloc(sizeof(*c)); 551 552 selectDb(c,0); 553 c->fd = -1; 554 c->name = NULL; 555 c->querybuf = sdsempty(); 556 c->querybuf_peak = 0; 557 c->argc = 0; 558 c->argv = NULL; 559 c->bufpos = 0; 560 c->flags = 0; 561 c->btype = REDIS_BLOCKED_NONE; 562 /* We set the fake client as a slave waiting for the synchronization 563 * so that Redis will not try to send replies to this client. */ 564 c->replstate = REDIS_REPL_WAIT_BGSAVE_START; 565 c->reply = listCreate(); 566 c->reply_bytes = 0; 567 c->obuf_soft_limit_reached_time = 0; 568 c->watched_keys = listCreate(); 569 c->peerid = NULL; 570 listSetFreeMethod(c->reply,decrRefCountVoid); 571 listSetDupMethod(c->reply,dupClientReplyValue); 572 initClientMultiState(c); 573 return c; 574 } 575 576 void freeFakeClient(struct redisClient *c) { 577 sdsfree(c->querybuf); 578 listRelease(c->reply); 579 listRelease(c->watched_keys); 580 freeClientMultiState(c); 581 zfree(c); 582 } 583 584 /* Replay the append log file. On error REDIS_OK is returned. On non fatal 585 * error (the append only file is zero-length) REDIS_ERR is returned. On 586 * fatal error an error message is logged and the program exists. */ 587 int loadAppendOnlyFile(char *filename) { 588 struct redisClient *fakeClient; 589 FILE *fp = fopen(filename,"r"); 590 struct redis_stat sb; 591 int old_aof_state = server.aof_state; 592 long loops = 0; 593 594 if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) { 595 server.aof_current_size = 0; 596 fclose(fp); 597 return REDIS_ERR; 598 } 599 600 if (fp == NULL) { 601 redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno)); 602 exit(1); 603 } 604 605 /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI 606 * to the same file we're about to read. */ 607 server.aof_state = REDIS_AOF_OFF; 608 609 fakeClient = createFakeClient(); 610 startLoading(fp); 611 612 while(1) { 613 int argc, j; 614 unsigned long len; 615 robj **argv; 616 char buf[128]; 617 sds argsds; 618 struct redisCommand *cmd; 619 620 /* Serve the clients from time to time */ 621 if (!(loops++ % 1000)) { 622 loadingProgress(ftello(fp)); 623 processEventsWhileBlocked(); 624 } 625 626 if (fgets(buf,sizeof(buf),fp) == NULL) { 627 if (feof(fp)) 628 break; 629 else 630 goto readerr; 631 } 632 if (buf[0] != '*') goto fmterr; 633 argc = atoi(buf+1); 634 if (argc < 1) goto fmterr; 635 636 argv = zmalloc(sizeof(robj*)*argc); 637 for (j = 0; j < argc; j++) { 638 if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr; 639 if (buf[0] != '$') goto fmterr; 640 len = strtol(buf+1,NULL,10); 641 argsds = sdsnewlen(NULL,len); 642 if (len && fread(argsds,len,1,fp) == 0) goto fmterr; 643 argv[j] = createObject(REDIS_STRING,argsds); 644 if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */ 645 } 646 647 /* Command lookup */ 648 cmd = lookupCommand(argv[0]->ptr); 649 if (!cmd) { 650 redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", (char*)argv[0]->ptr); 651 exit(1); 652 } 653 /* Run the command in the context of a fake client */ 654 fakeClient->argc = argc; 655 fakeClient->argv = argv; 656 cmd->proc(fakeClient); 657 658 /* The fake client should not have a reply */ 659 redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0); 660 /* The fake client should never get blocked */ 661 redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0); 662 663 /* Clean up. Command code may have changed argv/argc so we use the 664 * argv/argc of the client instead of the local variables. */ 665 for (j = 0; j < fakeClient->argc; j++) 666 decrRefCount(fakeClient->argv[j]); 667 zfree(fakeClient->argv); 668 } 669 670 /* This point can only be reached when EOF is reached without errors. 671 * If the client is in the middle of a MULTI/EXEC, log error and quit. */ 672 if (fakeClient->flags & REDIS_MULTI) goto readerr; 673 674 fclose(fp); 675 freeFakeClient(fakeClient); 676 server.aof_state = old_aof_state; 677 stopLoading(); 678 aofUpdateCurrentSize(); 679 server.aof_rewrite_base_size = server.aof_current_size; 680 return REDIS_OK; 681 682 readerr: 683 if (feof(fp)) { 684 redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file"); 685 } else { 686 redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno)); 687 } 688 exit(1); 689 fmterr: 690 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>"); 691 exit(1); 692 } 693 694 /* ---------------------------------------------------------------------------- 695 * AOF rewrite 696 * ------------------------------------------------------------------------- */ 697 698 /* Delegate writing an object to writing a bulk string or bulk long long. 699 * This is not placed in rio.c since that adds the redis.h dependency. */ 700 int rioWriteBulkObject(rio *r, robj *obj) { 701 /* Avoid using getDecodedObject to help copy-on-write (we are often 702 * in a child process when this function is called). */ 703 if (obj->encoding == REDIS_ENCODING_INT) { 704 return rioWriteBulkLongLong(r,(long)obj->ptr); 705 } else if (sdsEncodedObject(obj)) { 706 return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr)); 707 } else { 708 redisPanic("Unknown string encoding"); 709 } 710 } 711 712 /* Emit the commands needed to rebuild a list object. 713 * The function returns 0 on error, 1 on success. */ 714 int rewriteListObject(rio *r, robj *key, robj *o) { 715 long long count = 0, items = listTypeLength(o); 716 717 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 718 unsigned char *zl = o->ptr; 719 unsigned char *p = ziplistIndex(zl,0); 720 unsigned char *vstr; 721 unsigned int vlen; 722 long long vlong; 723 724 while(ziplistGet(p,&vstr,&vlen,&vlong)) { 725 if (count == 0) { 726 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 727 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 728 729 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 730 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0; 731 if (rioWriteBulkObject(r,key) == 0) return 0; 732 } 733 if (vstr) { 734 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0; 735 } else { 736 if (rioWriteBulkLongLong(r,vlong) == 0) return 0; 737 } 738 p = ziplistNext(zl,p); 739 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 740 items--; 741 } 742 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 743 list *list = o->ptr; 744 listNode *ln; 745 listIter li; 746 747 listRewind(list,&li); 748 while((ln = listNext(&li))) { 749 robj *eleobj = listNodeValue(ln); 750 751 if (count == 0) { 752 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 753 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 754 755 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 756 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0; 757 if (rioWriteBulkObject(r,key) == 0) return 0; 758 } 759 if (rioWriteBulkObject(r,eleobj) == 0) return 0; 760 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 761 items--; 762 } 763 } else { 764 redisPanic("Unknown list encoding"); 765 } 766 return 1; 767 } 768 769 /* Emit the commands needed to rebuild a set object. 770 * The function returns 0 on error, 1 on success. */ 771 int rewriteSetObject(rio *r, robj *key, robj *o) { 772 long long count = 0, items = setTypeSize(o); 773 774 if (o->encoding == REDIS_ENCODING_INTSET) { 775 int ii = 0; 776 int64_t llval; 777 778 while(intsetGet(o->ptr,ii++,&llval)) { 779 if (count == 0) { 780 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 781 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 782 783 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 784 if (rioWriteBulkString(r,"SADD",4) == 0) return 0; 785 if (rioWriteBulkObject(r,key) == 0) return 0; 786 } 787 if (rioWriteBulkLongLong(r,llval) == 0) return 0; 788 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 789 items--; 790 } 791 } else if (o->encoding == REDIS_ENCODING_HT) { 792 dictIterator *di = dictGetIterator(o->ptr); 793 dictEntry *de; 794 795 while((de = dictNext(di)) != NULL) { 796 robj *eleobj = dictGetKey(de); 797 if (count == 0) { 798 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 799 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 800 801 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; 802 if (rioWriteBulkString(r,"SADD",4) == 0) return 0; 803 if (rioWriteBulkObject(r,key) == 0) return 0; 804 } 805 if (rioWriteBulkObject(r,eleobj) == 0) return 0; 806 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 807 items--; 808 } 809 dictReleaseIterator(di); 810 } else { 811 redisPanic("Unknown set encoding"); 812 } 813 return 1; 814 } 815 816 /* Emit the commands needed to rebuild a sorted set object. 817 * The function returns 0 on error, 1 on success. */ 818 int rewriteSortedSetObject(rio *r, robj *key, robj *o) { 819 long long count = 0, items = zsetLength(o); 820 821 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 822 unsigned char *zl = o->ptr; 823 unsigned char *eptr, *sptr; 824 unsigned char *vstr; 825 unsigned int vlen; 826 long long vll; 827 double score; 828 829 eptr = ziplistIndex(zl,0); 830 redisAssert(eptr != NULL); 831 sptr = ziplistNext(zl,eptr); 832 redisAssert(sptr != NULL); 833 834 while (eptr != NULL) { 835 redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll)); 836 score = zzlGetScore(sptr); 837 838 if (count == 0) { 839 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 840 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 841 842 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; 843 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0; 844 if (rioWriteBulkObject(r,key) == 0) return 0; 845 } 846 if (rioWriteBulkDouble(r,score) == 0) return 0; 847 if (vstr != NULL) { 848 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0; 849 } else { 850 if (rioWriteBulkLongLong(r,vll) == 0) return 0; 851 } 852 zzlNext(zl,&eptr,&sptr); 853 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 854 items--; 855 } 856 } else if (o->encoding == REDIS_ENCODING_SKIPLIST) { 857 zset *zs = o->ptr; 858 dictIterator *di = dictGetIterator(zs->dict); 859 dictEntry *de; 860 861 while((de = dictNext(di)) != NULL) { 862 robj *eleobj = dictGetKey(de); 863 double *score = dictGetVal(de); 864 865 if (count == 0) { 866 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 867 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 868 869 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; 870 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0; 871 if (rioWriteBulkObject(r,key) == 0) return 0; 872 } 873 if (rioWriteBulkDouble(r,*score) == 0) return 0; 874 if (rioWriteBulkObject(r,eleobj) == 0) return 0; 875 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 876 items--; 877 } 878 dictReleaseIterator(di); 879 } else { 880 redisPanic("Unknown sorted zset encoding"); 881 } 882 return 1; 883 } 884 885 /* Write either the key or the value of the currently selected item of a hash. 886 * The 'hi' argument passes a valid Redis hash iterator. 887 * The 'what' filed specifies if to write a key or a value and can be 888 * either REDIS_HASH_KEY or REDIS_HASH_VALUE. 889 * 890 * The function returns 0 on error, non-zero on success. */ 891 static int rioWriteHashIteratorCursor(rio *r, hashTypeIterator *hi, int what) { 892 if (hi->encoding == REDIS_ENCODING_ZIPLIST) { 893 unsigned char *vstr = NULL; 894 unsigned int vlen = UINT_MAX; 895 long long vll = LLONG_MAX; 896 897 hashTypeCurrentFromZiplist(hi, what, &vstr, &vlen, &vll); 898 if (vstr) { 899 return rioWriteBulkString(r, (char*)vstr, vlen); 900 } else { 901 return rioWriteBulkLongLong(r, vll); 902 } 903 904 } else if (hi->encoding == REDIS_ENCODING_HT) { 905 robj *value; 906 907 hashTypeCurrentFromHashTable(hi, what, &value); 908 return rioWriteBulkObject(r, value); 909 } 910 911 redisPanic("Unknown hash encoding"); 912 return 0; 913 } 914 915 /* Emit the commands needed to rebuild a hash object. 916 * The function returns 0 on error, 1 on success. */ 917 int rewriteHashObject(rio *r, robj *key, robj *o) { 918 hashTypeIterator *hi; 919 long long count = 0, items = hashTypeLength(o); 920 921 hi = hashTypeInitIterator(o); 922 while (hashTypeNext(hi) != REDIS_ERR) { 923 if (count == 0) { 924 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? 925 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items; 926 927 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; 928 if (rioWriteBulkString(r,"HMSET",5) == 0) return 0; 929 if (rioWriteBulkObject(r,key) == 0) return 0; 930 } 931 932 if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_KEY) == 0) return 0; 933 if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_VALUE) == 0) return 0; 934 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0; 935 items--; 936 } 937 938 hashTypeReleaseIterator(hi); 939 940 return 1; 941 } 942 943 /* This function is called by the child rewriting the AOF file to read 944 * the difference accumulated from the parent into a buffer, that is 945 * concatenated at the end of the rewrite. */ 946 ssize_t aofReadDiffFromParent(void) { 947 char buf[65536]; /* Default pipe buffer size on most Linux sytems. */ 948 ssize_t nread, total = 0; 949 950 while ((nread = 951 read(server.aof_pipe_read_data_from_parent,buf,sizeof(buf))) > 0) { 952 server.aof_child_diff = sdscatlen(server.aof_child_diff,buf,nread); 953 total += nread; 954 } 955 return total; 956 } 957 958 /* Write a sequence of commands able to fully rebuild the dataset into 959 * "filename". Used both by REWRITEAOF and BGREWRITEAOF. 960 * 961 * In order to minimize the number of commands needed in the rewritten 962 * log Redis uses variadic commands when possible, such as RPUSH, SADD 963 * and ZADD. However at max REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time 964 * are inserted using a single command. */ 965 int rewriteAppendOnlyFile(char *filename) { 966 dictIterator *di = NULL; 967 dictEntry *de; 968 rio aof; 969 FILE *fp; 970 char tmpfile[256]; 971 int j; 972 long long now = mstime(); 973 char byte; 974 size_t processed = 0; 975 976 /* Note that we have to use a different temp name here compared to the 977 * one used by rewriteAppendOnlyFileBackground() function. */ 978 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid()); 979 fp = fopen(tmpfile,"w"); 980 if (!fp) { 981 redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno)); 982 return REDIS_ERR; 983 } 984 985 server.aof_child_diff = sdsempty(); 986 rioInitWithFile(&aof,fp); 987 if (server.aof_rewrite_incremental_fsync) 988 rioSetAutoSync(&aof,REDIS_AOF_AUTOSYNC_BYTES); 989 for (j = 0; j < server.dbnum; j++) { 990 char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n"; 991 redisDb *db = server.db+j; 992 dict *d = db->dict; 993 if (dictSize(d) == 0) continue; 994 di = dictGetSafeIterator(d); 995 if (!di) { 996 fclose(fp); 997 return REDIS_ERR; 998 } 999 1000 /* SELECT the new DB */ 1001 if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr; 1002 if (rioWriteBulkLongLong(&aof,j) == 0) goto werr; 1003 1004 /* Iterate this DB writing every entry */ 1005 while((de = dictNext(di)) != NULL) { 1006 sds keystr; 1007 robj key, *o; 1008 long long expiretime; 1009 1010 keystr = dictGetKey(de); 1011 o = dictGetVal(de); 1012 initStaticStringObject(key,keystr); 1013 1014 expiretime = getExpire(db,&key); 1015 1016 /* If this key is already expired skip it */ 1017 if (expiretime != -1 && expiretime < now) continue; 1018 1019 /* Save the key and associated value */ 1020 if (o->type == REDIS_STRING) { 1021 /* Emit a SET command */ 1022 char cmd[]="*3\r\n$3\r\nSET\r\n"; 1023 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; 1024 /* Key and value */ 1025 if (rioWriteBulkObject(&aof,&key) == 0) goto werr; 1026 if (rioWriteBulkObject(&aof,o) == 0) goto werr; 1027 } else if (o->type == REDIS_LIST) { 1028 if (rewriteListObject(&aof,&key,o) == 0) goto werr; 1029 } else if (o->type == REDIS_SET) { 1030 if (rewriteSetObject(&aof,&key,o) == 0) goto werr; 1031 } else if (o->type == REDIS_ZSET) { 1032 if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr; 1033 } else if (o->type == REDIS_HASH) { 1034 if (rewriteHashObject(&aof,&key,o) == 0) goto werr; 1035 } else { 1036 redisPanic("Unknown object type"); 1037 } 1038 /* Save the expire time */ 1039 if (expiretime != -1) { 1040 char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n"; 1041 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; 1042 if (rioWriteBulkObject(&aof,&key) == 0) goto werr; 1043 if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr; 1044 } 1045 /* Read some diff from the parent process from time to time. */ 1046 if (aof.processed_bytes > processed+1024*10) { 1047 processed = aof.processed_bytes; 1048 aofReadDiffFromParent(); 1049 } 1050 } 1051 dictReleaseIterator(di); 1052 } 1053 1054 /* Do an initial slow fsync here while the parent is still sending 1055 * data, in order to make the next final fsync faster. */ 1056 if (fflush(fp) == EOF) goto werr; 1057 if (fsync(fileno(fp)) == -1) goto werr; 1058 1059 /* Read again a few times to get more data from the parent. 1060 * We can't read forever (the server may receive data from clients 1061 * fater than it is able to send data to the child), so we try to read 1062 * some more data in a loop as soon as there is a good chance more data 1063 * will come. If it looks like we are wasting time, we abort (this 1064 * happens after 20 ms without new data). */ 1065 int nodata = 0; 1066 mstime_t start = mstime(); 1067 while(mstime()-start < 1000 && nodata < 20) { 1068 if (aeWait(server.aof_pipe_read_data_from_parent, AE_READABLE, 1) <= 0) 1069 { 1070 nodata++; 1071 continue; 1072 } 1073 nodata = 0; /* Start counting from zero, we stop on N *contiguous* 1074 timeouts. */ 1075 aofReadDiffFromParent(); 1076 } 1077 1078 /* Ask the master to stop sending diffs. */ 1079 if (write(server.aof_pipe_write_ack_to_parent,"!",1) != 1) goto werr; 1080 if (anetNonBlock(NULL,server.aof_pipe_read_ack_from_parent) != ANET_OK) 1081 goto werr; 1082 /* We read the ACK from the server using a 10 seconds timeout. Normally 1083 * it should reply ASAP, but just in case we lose its reply, we are sure 1084 * the child will eventually get terminated. */ 1085 if (syncRead(server.aof_pipe_read_ack_from_parent,&byte,1,5000) != 1 || 1086 byte != '!') goto werr; 1087 redisLog(REDIS_NOTICE,"Parent agreed to stop sending diffs. Finalizing AOF..."); 1088 1089 /* Read the final diff if any. */ 1090 aofReadDiffFromParent(); 1091 1092 /* Write the received diff to the file. */ 1093 redisLog(REDIS_NOTICE, 1094 "Concatenating %.2f MB of AOF diff received from parent.", 1095 (double) sdslen(server.aof_child_diff) / (1024*1024)); 1096 if (rioWrite(&aof,server.aof_child_diff,sdslen(server.aof_child_diff)) == 0) 1097 goto werr; 1098 1099 /* Make sure data will not remain on the OS's output buffers */ 1100 if (fflush(fp) == EOF) goto werr; 1101 if (fsync(fileno(fp)) == -1) goto werr; 1102 if (fclose(fp) == EOF) goto werr; 1103 1104 /* Use RENAME to make sure the DB file is changed atomically only 1105 * if the generate DB file is ok. */ 1106 if (rename(tmpfile,filename) == -1) { 1107 redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno)); 1108 unlink(tmpfile); 1109 return REDIS_ERR; 1110 } 1111 redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed"); 1112 return REDIS_OK; 1113 1114 werr: 1115 fclose(fp); 1116 unlink(tmpfile); 1117 redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno)); 1118 if (di) dictReleaseIterator(di); 1119 return REDIS_ERR; 1120 } 1121 1122 /* ---------------------------------------------------------------------------- 1123 * AOF rewrite pipes for IPC 1124 * -------------------------------------------------------------------------- */ 1125 1126 /* This event handler is called when the AOF rewriting child sends us a 1127 * single '!' char to signal we should stop sending buffer diffs. The 1128 * parent sends a '!' as well to acknowledge. */ 1129 void aofChildPipeReadable(aeEventLoop *el, int fd, void *privdata, int mask) { 1130 char byte; 1131 1132 if (read(fd,&byte,1) == 1 && byte == '!') { 1133 redisLog(REDIS_NOTICE,"AOF rewrite child asks to stop sending diffs."); 1134 server.aof_stop_sending_diff = 1; 1135 if (write(server.aof_pipe_write_ack_to_child,"!",1) != 1) { 1136 /* If we can't send the ack, inform the user, but don't try again 1137 * since in the other side the children will use a timeout if the 1138 * kernel can't buffer our write, or, the children was 1139 * terminated. */ 1140 redisLog(REDIS_WARNING,"Can't send ACK to AOF child: %s", 1141 strerror(errno)); 1142 } 1143 } 1144 /* Remove the handler since this can be called only one time during a 1145 * rewrite. */ 1146 aeDeleteFileEvent(server.el,server.aof_pipe_read_ack_from_child,AE_READABLE); 1147 } 1148 1149 /* Create the pipes used for parent - child process IPC during rewrite. 1150 * We have a data pipe used to send AOF incremental diffs to the child, 1151 * and two other pipes used by the children to signal it finished with 1152 * the rewrite so no more data should be written, and another for the 1153 * parent to acknowledge it understood this new condition. */ 1154 int aofCreatePipes(void) { 1155 int fds[6] = {-1, -1, -1, -1, -1, -1}; 1156 int j; 1157 1158 if (pipe(fds) == -1) goto error; /* parent -> children data. */ 1159 if (pipe(fds+2) == -1) goto error; /* children -> parent ack. */ 1160 if (pipe(fds+4) == -1) goto error; /* children -> parent ack. */ 1161 /* Parent -> children data is non blocking. */ 1162 if (anetNonBlock(NULL,fds[0]) != ANET_OK) goto error; 1163 if (anetNonBlock(NULL,fds[1]) != ANET_OK) goto error; 1164 if (aeCreateFileEvent(server.el, fds[2], AE_READABLE, aofChildPipeReadable, NULL) == AE_ERR) goto error; 1165 1166 server.aof_pipe_write_data_to_child = fds[1]; 1167 server.aof_pipe_read_data_from_parent = fds[0]; 1168 server.aof_pipe_write_ack_to_parent = fds[3]; 1169 server.aof_pipe_read_ack_from_child = fds[2]; 1170 server.aof_pipe_write_ack_to_child = fds[5]; 1171 server.aof_pipe_read_ack_from_parent = fds[4]; 1172 server.aof_stop_sending_diff = 0; 1173 return REDIS_OK; 1174 1175 error: 1176 redisLog(REDIS_WARNING,"Error opening /setting AOF rewrite IPC pipes: %s", 1177 strerror(errno)); 1178 for (j = 0; j < 6; j++) if(fds[j] != -1) close(fds[j]); 1179 return REDIS_ERR; 1180 } 1181 1182 void aofClosePipes(void) { 1183 aeDeleteFileEvent(server.el,server.aof_pipe_read_ack_from_child,AE_READABLE); 1184 aeDeleteFileEvent(server.el,server.aof_pipe_write_data_to_child,AE_WRITABLE); 1185 close(server.aof_pipe_write_data_to_child); 1186 close(server.aof_pipe_read_data_from_parent); 1187 close(server.aof_pipe_write_ack_to_parent); 1188 close(server.aof_pipe_read_ack_from_child); 1189 close(server.aof_pipe_write_ack_to_child); 1190 close(server.aof_pipe_read_ack_from_parent); 1191 } 1192 1193 /* ---------------------------------------------------------------------------- 1194 * AOF backgorund rewrite 1195 * ------------------------------------------------------------------------- */ 1196 1197 /* This is how rewriting of the append only file in background works: 1198 * 1199 * 1) The user calls BGREWRITEAOF 1200 * 2) Redis calls this function, that forks(): 1201 * 2a) the child rewrite the append only file in a temp file. 1202 * 2b) the parent accumulates differences in server.aof_rewrite_buf. 1203 * 3) When the child finished '2a' exists. 1204 * 4) The parent will trap the exit code, if it's OK, will append the 1205 * data accumulated into server.aof_rewrite_buf into the temp file, and 1206 * finally will rename(2) the temp file in the actual file name. 1207 * The the new file is reopened as the new append only file. Profit! 1208 */ 1209 int rewriteAppendOnlyFileBackground(void) { 1210 pid_t childpid; 1211 long long start; 1212 1213 if (server.aof_child_pid != -1) return REDIS_ERR; 1214 if (aofCreatePipes() != REDIS_OK) return REDIS_ERR; 1215 start = ustime(); 1216 if ((childpid = fork()) == 0) { 1217 char tmpfile[256]; 1218 1219 /* Child */ 1220 closeListeningSockets(0); 1221 redisSetProcTitle("redis-aof-rewrite"); 1222 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); 1223 if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) { 1224 size_t private_dirty = zmalloc_get_private_dirty(); 1225 1226 if (private_dirty) { 1227 redisLog(REDIS_NOTICE, 1228 "AOF rewrite: %zu MB of memory used by copy-on-write", 1229 private_dirty/(1024*1024)); 1230 } 1231 exitFromChild(0); 1232 } else { 1233 exitFromChild(1); 1234 } 1235 } else { 1236 /* Parent */ 1237 server.stat_fork_time = ustime()-start; 1238 server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */ 1239 latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000); 1240 if (childpid == -1) { 1241 redisLog(REDIS_WARNING, 1242 "Can't rewrite append only file in background: fork: %s", 1243 strerror(errno)); 1244 return REDIS_ERR; 1245 } 1246 redisLog(REDIS_NOTICE, 1247 "Background append only file rewriting started by pid %d",childpid); 1248 server.aof_rewrite_scheduled = 0; 1249 server.aof_rewrite_time_start = time(NULL); 1250 server.aof_child_pid = childpid; 1251 updateDictResizePolicy(); 1252 /* We set appendseldb to -1 in order to force the next call to the 1253 * feedAppendOnlyFile() to issue a SELECT command, so the differences 1254 * accumulated by the parent into server.aof_rewrite_buf will start 1255 * with a SELECT statement and it will be safe to merge. */ 1256 server.aof_selected_db = -1; 1257 replicationScriptCacheFlush(); 1258 return REDIS_OK; 1259 } 1260 return REDIS_OK; /* unreached */ 1261 } 1262 1263 void bgrewriteaofCommand(redisClient *c) { 1264 if (server.aof_child_pid != -1) { 1265 addReplyError(c,"Background append only file rewriting already in progress"); 1266 } else if (server.rdb_child_pid != -1) { 1267 server.aof_rewrite_scheduled = 1; 1268 addReplyStatus(c,"Background append only file rewriting scheduled"); 1269 } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) { 1270 addReplyStatus(c,"Background append only file rewriting started"); 1271 } else { 1272 addReply(c,shared.err); 1273 } 1274 } 1275 1276 void aofRemoveTempFile(pid_t childpid) { 1277 char tmpfile[256]; 1278 1279 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid); 1280 unlink(tmpfile); 1281 } 1282 1283 /* Update the server.aof_current_size filed explicitly using stat(2) 1284 * to check the size of the file. This is useful after a rewrite or after 1285 * a restart, normally the size is updated just adding the write length 1286 * to the current length, that is much faster. */ 1287 void aofUpdateCurrentSize(void) { 1288 struct redis_stat sb; 1289 mstime_t latency; 1290 1291 latencyStartMonitor(latency); 1292 if (redis_fstat(server.aof_fd,&sb) == -1) { 1293 redisLog(REDIS_WARNING,"Unable to obtain the AOF file length. stat: %s", 1294 strerror(errno)); 1295 } else { 1296 server.aof_current_size = sb.st_size; 1297 } 1298 latencyEndMonitor(latency); 1299 latencyAddSampleIfNeeded("aof-fstat",latency); 1300 } 1301 1302 /* A background append only file rewriting (BGREWRITEAOF) terminated its work. 1303 * Handle this. */ 1304 void backgroundRewriteDoneHandler(int exitcode, int bysignal) { 1305 if (!bysignal && exitcode == 0) { 1306 int newfd, oldfd; 1307 char tmpfile[256]; 1308 long long now = ustime(); 1309 mstime_t latency; 1310 1311 redisLog(REDIS_NOTICE, 1312 "Background AOF rewrite terminated with success"); 1313 1314 /* Flush the differences accumulated by the parent to the 1315 * rewritten AOF. */ 1316 latencyStartMonitor(latency); 1317 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", 1318 (int)server.aof_child_pid); 1319 newfd = open(tmpfile,O_WRONLY|O_APPEND); 1320 if (newfd == -1) { 1321 redisLog(REDIS_WARNING, 1322 "Unable to open the temporary AOF produced by the child: %s", strerror(errno)); 1323 goto cleanup; 1324 } 1325 1326 if (aofRewriteBufferWrite(newfd) == -1) { 1327 redisLog(REDIS_WARNING, 1328 "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); 1329 close(newfd); 1330 goto cleanup; 1331 } 1332 latencyEndMonitor(latency); 1333 latencyAddSampleIfNeeded("aof-rewrite-diff-write",latency); 1334 1335 redisLog(REDIS_NOTICE, 1336 "Redidual parent diff successfully flushed to the rewritten AOF (%.2f MB)", (double) aofRewriteBufferSize() / (1024*1024)); 1337 1338 /* The only remaining thing to do is to rename the temporary file to 1339 * the configured file and switch the file descriptor used to do AOF 1340 * writes. We don't want close(2) or rename(2) calls to block the 1341 * server on old file deletion. 1342 * 1343 * There are two possible scenarios: 1344 * 1345 * 1) AOF is DISABLED and this was a one time rewrite. The temporary 1346 * file will be renamed to the configured file. When this file already 1347 * exists, it will be unlinked, which may block the server. 1348 * 1349 * 2) AOF is ENABLED and the rewritten AOF will immediately start 1350 * receiving writes. After the temporary file is renamed to the 1351 * configured file, the original AOF file descriptor will be closed. 1352 * Since this will be the last reference to that file, closing it 1353 * causes the underlying file to be unlinked, which may block the 1354 * server. 1355 * 1356 * To mitigate the blocking effect of the unlink operation (either 1357 * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we 1358 * use a background thread to take care of this. First, we 1359 * make scenario 1 identical to scenario 2 by opening the target file 1360 * when it exists. The unlink operation after the rename(2) will then 1361 * be executed upon calling close(2) for its descriptor. Everything to 1362 * guarantee atomicity for this switch has already happened by then, so 1363 * we don't care what the outcome or duration of that close operation 1364 * is, as long as the file descriptor is released again. */ 1365 if (server.aof_fd == -1) { 1366 /* AOF disabled */ 1367 1368 /* Don't care if this fails: oldfd will be -1 and we handle that. 1369 * One notable case of -1 return is if the old file does 1370 * not exist. */ 1371 oldfd = open(server.aof_filename,O_RDONLY|O_NONBLOCK); 1372 } else { 1373 /* AOF enabled */ 1374 oldfd = -1; /* We'll set this to the current AOF filedes later. */ 1375 } 1376 1377 /* Rename the temporary file. This will not unlink the target file if 1378 * it exists, because we reference it with "oldfd". */ 1379 latencyStartMonitor(latency); 1380 if (rename(tmpfile,server.aof_filename) == -1) { 1381 redisLog(REDIS_WARNING, 1382 "Error trying to rename the temporary AOF file: %s", strerror(errno)); 1383 close(newfd); 1384 if (oldfd != -1) close(oldfd); 1385 goto cleanup; 1386 } 1387 latencyEndMonitor(latency); 1388 latencyAddSampleIfNeeded("aof-rename",latency); 1389 1390 if (server.aof_fd == -1) { 1391 /* AOF disabled, we don't need to set the AOF file descriptor 1392 * to this new file, so we can close it. */ 1393 close(newfd); 1394 } else { 1395 /* AOF enabled, replace the old fd with the new one. */ 1396 oldfd = server.aof_fd; 1397 server.aof_fd = newfd; 1398 if (server.aof_fsync == AOF_FSYNC_ALWAYS) 1399 aof_fsync(newfd); 1400 else if (server.aof_fsync == AOF_FSYNC_EVERYSEC) 1401 aof_background_fsync(newfd); 1402 server.aof_selected_db = -1; /* Make sure SELECT is re-issued */ 1403 aofUpdateCurrentSize(); 1404 server.aof_rewrite_base_size = server.aof_current_size; 1405 1406 /* Clear regular AOF buffer since its contents was just written to 1407 * the new AOF from the background rewrite buffer. */ 1408 sdsfree(server.aof_buf); 1409 server.aof_buf = sdsempty(); 1410 } 1411 1412 server.aof_lastbgrewrite_status = REDIS_OK; 1413 1414 redisLog(REDIS_NOTICE, "Background AOF rewrite finished successfully"); 1415 /* Change state from WAIT_REWRITE to ON if needed */ 1416 if (server.aof_state == REDIS_AOF_WAIT_REWRITE) 1417 server.aof_state = REDIS_AOF_ON; 1418 1419 /* Asynchronously close the overwritten AOF. */ 1420 if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL); 1421 1422 redisLog(REDIS_VERBOSE, 1423 "Background AOF rewrite signal handler took %lldus", ustime()-now); 1424 } else if (!bysignal && exitcode != 0) { 1425 server.aof_lastbgrewrite_status = REDIS_ERR; 1426 1427 redisLog(REDIS_WARNING, 1428 "Background AOF rewrite terminated with error"); 1429 } else { 1430 server.aof_lastbgrewrite_status = REDIS_ERR; 1431 1432 redisLog(REDIS_WARNING, 1433 "Background AOF rewrite terminated by signal %d", bysignal); 1434 } 1435 1436 cleanup: 1437 aofClosePipes(); 1438 aofRewriteBufferReset(); 1439 aofRemoveTempFile(server.aof_child_pid); 1440 server.aof_child_pid = -1; 1441 server.aof_rewrite_time_last = time(NULL)-server.aof_rewrite_time_start; 1442 server.aof_rewrite_time_start = -1; 1443 /* Schedule a new rewrite if we are waiting for it to switch the AOF ON. */ 1444 if (server.aof_state == REDIS_AOF_WAIT_REWRITE) 1445 server.aof_rewrite_scheduled = 1; 1446 } 1447