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