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