diff options
Diffstat (limited to 'src/aof.c')
| -rw-r--r-- | src/aof.c | 1568 |
1 files changed, 1568 insertions, 0 deletions
diff --git a/src/aof.c b/src/aof.c new file mode 100644 index 0000000..f73da1e --- /dev/null +++ b/src/aof.c @@ -0,0 +1,1568 @@ +/* + * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of Redis nor the names of its contributors may be used + * to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#include "server.h" +#include "bio.h" +#include "rio.h" + +#include <signal.h> +#include <fcntl.h> +#include <sys/stat.h> +#include <sys/types.h> +#include <sys/time.h> +#include <sys/resource.h> +#include <sys/wait.h> +#include <sys/param.h> + +void aofUpdateCurrentSize(void); +void aofClosePipes(void); + +/* ---------------------------------------------------------------------------- + * AOF rewrite buffer implementation. + * + * The following code implement a simple buffer used in order to accumulate + * changes while the background process is rewriting the AOF file. + * + * We only need to append, but can't just use realloc with a large block + * because 'huge' reallocs are not always handled as one could expect + * (via remapping of pages at OS level) but may involve copying data. + * + * For this reason we use a list of blocks, every block is + * AOF_RW_BUF_BLOCK_SIZE bytes. + * ------------------------------------------------------------------------- */ + +#define AOF_RW_BUF_BLOCK_SIZE (1024*1024*10) /* 10 MB per block */ + +typedef struct aofrwblock { + unsigned long used, free; + char buf[AOF_RW_BUF_BLOCK_SIZE]; +} aofrwblock; + +/* This function free the old AOF rewrite buffer if needed, and initialize + * a fresh new one. It tests for server.aof_rewrite_buf_blocks equal to NULL + * so can be used for the first initialization as well. */ +void aofRewriteBufferReset(void) { + if (server.aof_rewrite_buf_blocks) + listRelease(server.aof_rewrite_buf_blocks); + + server.aof_rewrite_buf_blocks = listCreate(); + listSetFreeMethod(server.aof_rewrite_buf_blocks,zfree); +} + +/* Return the current size of the AOF rewrite buffer. */ +unsigned long aofRewriteBufferSize(void) { + listNode *ln; + listIter li; + unsigned long size = 0; + + listRewind(server.aof_rewrite_buf_blocks,&li); + while((ln = listNext(&li))) { + aofrwblock *block = listNodeValue(ln); + size += block->used; + } + return size; +} + +/* Event handler used to send data to the child process doing the AOF + * rewrite. We send pieces of our AOF differences buffer so that the final + * write when the child finishes the rewrite will be small. */ +void aofChildWriteDiffData(aeEventLoop *el, int fd, void *privdata, int mask) { + listNode *ln; + aofrwblock *block; + ssize_t nwritten; + UNUSED(el); + UNUSED(fd); + UNUSED(privdata); + UNUSED(mask); + + while(1) { + ln = listFirst(server.aof_rewrite_buf_blocks); + block = ln ? ln->value : NULL; + if (server.aof_stop_sending_diff || !block) { + aeDeleteFileEvent(server.el,server.aof_pipe_write_data_to_child, + AE_WRITABLE); + return; + } + if (block->used > 0) { + nwritten = write(server.aof_pipe_write_data_to_child, + block->buf,block->used); + if (nwritten <= 0) return; + memmove(block->buf,block->buf+nwritten,block->used-nwritten); + block->used -= nwritten; + } + if (block->used == 0) listDelNode(server.aof_rewrite_buf_blocks,ln); + } +} + +/* Append data to the AOF rewrite buffer, allocating new blocks if needed. */ +void aofRewriteBufferAppend(unsigned char *s, unsigned long len) { + listNode *ln = listLast(server.aof_rewrite_buf_blocks); + aofrwblock *block = ln ? ln->value : NULL; + + while(len) { + /* If we already got at least an allocated block, try appending + * at least some piece into it. */ + if (block) { + unsigned long thislen = (block->free < len) ? block->free : len; + if (thislen) { /* The current block is not already full. */ + memcpy(block->buf+block->used, s, thislen); + block->used += thislen; + block->free -= thislen; + s += thislen; + len -= thislen; + } + } + + if (len) { /* First block to allocate, or need another block. */ + int numblocks; + + block = zmalloc(sizeof(*block)); + block->free = AOF_RW_BUF_BLOCK_SIZE; + block->used = 0; + listAddNodeTail(server.aof_rewrite_buf_blocks,block); + + /* Log every time we cross more 10 or 100 blocks, respectively + * as a notice or warning. */ + numblocks = listLength(server.aof_rewrite_buf_blocks); + if (((numblocks+1) % 10) == 0) { + int level = ((numblocks+1) % 100) == 0 ? LL_WARNING : + LL_NOTICE; + serverLog(level,"Background AOF buffer size: %lu MB", + aofRewriteBufferSize()/(1024*1024)); + } + } + } + + /* Install a file event to send data to the rewrite child if there is + * not one already. */ + if (aeGetFileEvents(server.el,server.aof_pipe_write_data_to_child) == 0) { + aeCreateFileEvent(server.el, server.aof_pipe_write_data_to_child, + AE_WRITABLE, aofChildWriteDiffData, NULL); + } +} + +/* Write the buffer (possibly composed of multiple blocks) into the specified + * fd. If a short write or any other error happens -1 is returned, + * otherwise the number of bytes written is returned. */ +ssize_t aofRewriteBufferWrite(int fd) { + listNode *ln; + listIter li; + ssize_t count = 0; + + listRewind(server.aof_rewrite_buf_blocks,&li); + while((ln = listNext(&li))) { + aofrwblock *block = listNodeValue(ln); + ssize_t nwritten; + + if (block->used) { + nwritten = write(fd,block->buf,block->used); + if (nwritten != (ssize_t)block->used) { + if (nwritten == 0) errno = EIO; + return -1; + } + count += nwritten; + } + } + return count; +} + +/* ---------------------------------------------------------------------------- + * AOF file implementation + * ------------------------------------------------------------------------- */ + +/* Starts a background task that performs fsync() against the specified + * file descriptor (the one of the AOF file) in another thread. */ +void aof_background_fsync(int fd) { + bioCreateBackgroundJob(BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL); +} + +/* Called when the user switches from "appendonly yes" to "appendonly no" + * at runtime using the CONFIG command. */ +void stopAppendOnly(void) { + serverAssert(server.aof_state != AOF_OFF); + flushAppendOnlyFile(1); + aof_fsync(server.aof_fd); + close(server.aof_fd); + + server.aof_fd = -1; + server.aof_selected_db = -1; + server.aof_state = AOF_OFF; + /* rewrite operation in progress? kill it, wait child exit */ + if (server.aof_child_pid != -1) { + int statloc; + + serverLog(LL_NOTICE,"Killing running AOF rewrite child: %ld", + (long) server.aof_child_pid); + if (kill(server.aof_child_pid,SIGUSR1) != -1) { + while(wait3(&statloc,0,NULL) != server.aof_child_pid); + } + /* reset the buffer accumulating changes while the child saves */ + aofRewriteBufferReset(); + aofRemoveTempFile(server.aof_child_pid); + server.aof_child_pid = -1; + server.aof_rewrite_time_start = -1; + /* close pipes used for IPC between the two processes. */ + aofClosePipes(); + } +} + +/* Called when the user switches from "appendonly no" to "appendonly yes" + * at runtime using the CONFIG command. */ +int startAppendOnly(void) { + char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */ + + server.aof_last_fsync = server.unixtime; + server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644); + serverAssert(server.aof_state == AOF_OFF); + if (server.aof_fd == -1) { + char *cwdp = getcwd(cwd,MAXPATHLEN); + + serverLog(LL_WARNING, + "Redis needs to enable the AOF but can't open the " + "append only file %s (in server root dir %s): %s", + server.aof_filename, + cwdp ? cwdp : "unknown", + strerror(errno)); + return C_ERR; + } + if (server.rdb_child_pid != -1) { + server.aof_rewrite_scheduled = 1; + serverLog(LL_WARNING,"AOF was enabled but there is already a child process saving an RDB file on disk. An AOF background was scheduled to start when possible."); + } else if (rewriteAppendOnlyFileBackground() == C_ERR) { + close(server.aof_fd); + 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."); + return C_ERR; + } + /* We correctly switched on AOF, now wait for the rewrite to be complete + * in order to append data on disk. */ + server.aof_state = AOF_WAIT_REWRITE; + return C_OK; +} + +/* Write the append only file buffer on disk. + * + * Since we are required to write the AOF before replying to the client, + * and the only way the client socket can get a write is entering when the + * the event loop, we accumulate all the AOF writes in a memory + * buffer and write it on disk using this function just before entering + * the event loop again. + * + * About the 'force' argument: + * + * When the fsync policy is set to 'everysec' we may delay the flush if there + * is still an fsync() going on in the background thread, since for instance + * on Linux write(2) will be blocked by the background fsync anyway. + * When this happens we remember that there is some aof buffer to be + * flushed ASAP, and will try to do that in the serverCron() function. + * + * However if force is set to 1 we'll write regardless of the background + * fsync. */ +#define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */ +void flushAppendOnlyFile(int force) { + ssize_t nwritten; + int sync_in_progress = 0; + mstime_t latency; + + if (sdslen(server.aof_buf) == 0) return; + + if (server.aof_fsync == AOF_FSYNC_EVERYSEC) + sync_in_progress = bioPendingJobsOfType(BIO_AOF_FSYNC) != 0; + + if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) { + /* With this append fsync policy we do background fsyncing. + * If the fsync is still in progress we can try to delay + * the write for a couple of seconds. */ + if (sync_in_progress) { + if (server.aof_flush_postponed_start == 0) { + /* No previous write postponing, remember that we are + * postponing the flush and return. */ + server.aof_flush_postponed_start = server.unixtime; + return; + } else if (server.unixtime - server.aof_flush_postponed_start < 2) { + /* We were already waiting for fsync to finish, but for less + * than two seconds this is still ok. Postpone again. */ + return; + } + /* Otherwise fall trough, and go write since we can't wait + * over two seconds. */ + server.aof_delayed_fsync++; + 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."); + } + } + /* We want to perform a single write. This should be guaranteed atomic + * at least if the filesystem we are writing is a real physical one. + * While this will save us against the server being killed I don't think + * there is much to do about the whole server stopping for power problems + * or alike */ + + latencyStartMonitor(latency); + nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf)); + latencyEndMonitor(latency); + /* We want to capture different events for delayed writes: + * when the delay happens with a pending fsync, or with a saving child + * active, and when the above two conditions are missing. + * We also use an additional event name to save all samples which is + * useful for graphing / monitoring purposes. */ + if (sync_in_progress) { + latencyAddSampleIfNeeded("aof-write-pending-fsync",latency); + } else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) { + latencyAddSampleIfNeeded("aof-write-active-child",latency); + } else { + latencyAddSampleIfNeeded("aof-write-alone",latency); + } + latencyAddSampleIfNeeded("aof-write",latency); + + /* We performed the write so reset the postponed flush sentinel to zero. */ + server.aof_flush_postponed_start = 0; + + if (nwritten != (signed)sdslen(server.aof_buf)) { + static time_t last_write_error_log = 0; + int can_log = 0; + + /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */ + if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) { + can_log = 1; + last_write_error_log = server.unixtime; + } + + /* Log the AOF write error and record the error code. */ + if (nwritten == -1) { + if (can_log) { + serverLog(LL_WARNING,"Error writing to the AOF file: %s", + strerror(errno)); + server.aof_last_write_errno = errno; + } + } else { + if (can_log) { + serverLog(LL_WARNING,"Short write while writing to " + "the AOF file: (nwritten=%lld, " + "expected=%lld)", + (long long)nwritten, + (long long)sdslen(server.aof_buf)); + } + + if (ftruncate(server.aof_fd, server.aof_current_size) == -1) { + if (can_log) { + serverLog(LL_WARNING, "Could not remove short write " + "from the append-only file. Redis may refuse " + "to load the AOF the next time it starts. " + "ftruncate: %s", strerror(errno)); + } + } else { + /* If the ftruncate() succeeded we can set nwritten to + * -1 since there is no longer partial data into the AOF. */ + nwritten = -1; + } + server.aof_last_write_errno = ENOSPC; + } + + /* Handle the AOF write error. */ + if (server.aof_fsync == AOF_FSYNC_ALWAYS) { + /* We can't recover when the fsync policy is ALWAYS since the + * reply for the client is already in the output buffers, and we + * have the contract with the user that on acknowledged write data + * is synced on disk. */ + serverLog(LL_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting..."); + exit(1); + } else { + /* Recover from failed write leaving data into the buffer. However + * set an error to stop accepting writes as long as the error + * condition is not cleared. */ + server.aof_last_write_status = C_ERR; + + /* Trim the sds buffer if there was a partial write, and there + * was no way to undo it with ftruncate(2). */ + if (nwritten > 0) { + server.aof_current_size += nwritten; + sdsrange(server.aof_buf,nwritten,-1); + } + return; /* We'll try again on the next call... */ + } + } else { + /* Successful write(2). If AOF was in error state, restore the + * OK state and log the event. */ + if (server.aof_last_write_status == C_ERR) { + serverLog(LL_WARNING, + "AOF write error looks solved, Redis can write again."); + server.aof_last_write_status = C_OK; + } + } + server.aof_current_size += nwritten; + + /* Re-use AOF buffer when it is small enough. The maximum comes from the + * arena size of 4k minus some overhead (but is otherwise arbitrary). */ + if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) { + sdsclear(server.aof_buf); + } else { + sdsfree(server.aof_buf); + server.aof_buf = sdsempty(); + } + + /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are + * children doing I/O in the background. */ + if (server.aof_no_fsync_on_rewrite && + (server.aof_child_pid != -1 || server.rdb_child_pid != -1)) + return; + + /* Perform the fsync if needed. */ + if (server.aof_fsync == AOF_FSYNC_ALWAYS) { + /* aof_fsync is defined as fdatasync() for Linux in order to avoid + * flushing metadata. */ + latencyStartMonitor(latency); + aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */ + latencyEndMonitor(latency); + latencyAddSampleIfNeeded("aof-fsync-always",latency); + server.aof_last_fsync = server.unixtime; + } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC && + server.unixtime > server.aof_last_fsync)) { + if (!sync_in_progress) aof_background_fsync(server.aof_fd); + server.aof_last_fsync = server.unixtime; + } +} + +sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) { + char buf[32]; + int len, j; + robj *o; + + buf[0] = '*'; + len = 1+ll2string(buf+1,sizeof(buf)-1,argc); + buf[len++] = '\r'; + buf[len++] = '\n'; + dst = sdscatlen(dst,buf,len); + + for (j = 0; j < argc; j++) { + o = getDecodedObject(argv[j]); + buf[0] = '$'; + len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr)); + buf[len++] = '\r'; + buf[len++] = '\n'; + dst = sdscatlen(dst,buf,len); + dst = sdscatlen(dst,o->ptr,sdslen(o->ptr)); + dst = sdscatlen(dst,"\r\n",2); + decrRefCount(o); + } + return dst; +} + +/* Create the sds representation of an PEXPIREAT command, using + * 'seconds' as time to live and 'cmd' to understand what command + * we are translating into a PEXPIREAT. + * + * This command is used in order to translate EXPIRE and PEXPIRE commands + * into PEXPIREAT command so that we retain precision in the append only + * file, and the time is always absolute and not relative. */ +sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) { + long long when; + robj *argv[3]; + + /* Make sure we can use strtoll */ + seconds = getDecodedObject(seconds); + when = strtoll(seconds->ptr,NULL,10); + /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */ + if (cmd->proc == expireCommand || cmd->proc == setexCommand || + cmd->proc == expireatCommand) + { + when *= 1000; + } + /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */ + if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || + cmd->proc == setexCommand || cmd->proc == psetexCommand) + { + when += mstime(); + } + decrRefCount(seconds); + + argv[0] = createStringObject("PEXPIREAT",9); + argv[1] = key; + argv[2] = createStringObjectFromLongLong(when); + buf = catAppendOnlyGenericCommand(buf, 3, argv); + decrRefCount(argv[0]); + decrRefCount(argv[2]); + return buf; +} + +void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) { + sds buf = sdsempty(); + robj *tmpargv[3]; + + /* The DB this command was targeting is not the same as the last command + * we appended. To issue a SELECT command is needed. */ + if (dictid != server.aof_selected_db) { + char seldb[64]; + + snprintf(seldb,sizeof(seldb),"%d",dictid); + buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n", + (unsigned long)strlen(seldb),seldb); + server.aof_selected_db = dictid; + } + + if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || + cmd->proc == expireatCommand) { + /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */ + buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); + } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) { + /* Translate SETEX/PSETEX to SET and PEXPIREAT */ + tmpargv[0] = createStringObject("SET",3); + tmpargv[1] = argv[1]; + tmpargv[2] = argv[3]; + buf = catAppendOnlyGenericCommand(buf,3,tmpargv); + decrRefCount(tmpargv[0]); + buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); + } else { + /* All the other commands don't need translation or need the + * same translation already operated in the command vector + * for the replication itself. */ + buf = catAppendOnlyGenericCommand(buf,argc,argv); + } + + /* Append to the AOF buffer. This will be flushed on disk just before + * of re-entering the event loop, so before the client will get a + * positive reply about the operation performed. */ + if (server.aof_state == AOF_ON) + server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf)); + + /* If a background append only file rewriting is in progress we want to + * accumulate the differences between the child DB and the current one + * in a buffer, so that when the child process will do its work we + * can append the differences to the new append only file. */ + if (server.aof_child_pid != -1) + aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf)); + + sdsfree(buf); +} + +/* ---------------------------------------------------------------------------- + * AOF loading + * ------------------------------------------------------------------------- */ + +/* In Redis commands are always executed in the context of a client, so in + * order to load the append only file we need to create a fake client. */ +struct client *createFakeClient(void) { + struct client *c = zmalloc(sizeof(*c)); + + selectDb(c,0); + c->fd = -1; + c->name = NULL; + c->querybuf = sdsempty(); + c->querybuf_peak = 0; + c->argc = 0; + c->argv = NULL; + c->bufpos = 0; + c->flags = 0; + c->btype = BLOCKED_NONE; + /* We set the fake client as a slave waiting for the synchronization + * so that Redis will not try to send replies to this client. */ + c->replstate = SLAVE_STATE_WAIT_BGSAVE_START; + c->reply = listCreate(); + c->reply_bytes = 0; + c->obuf_soft_limit_reached_time = 0; + c->watched_keys = listCreate(); + c->peerid = NULL; + listSetFreeMethod(c->reply,decrRefCountVoid); + listSetDupMethod(c->reply,dupClientReplyValue); + initClientMultiState(c); + return c; +} + +void freeFakeClientArgv(struct client *c) { + int j; + + for (j = 0; j < c->argc; j++) + decrRefCount(c->argv[j]); + zfree(c->argv); +} + +void freeFakeClient(struct client *c) { + sdsfree(c->querybuf); + listRelease(c->reply); + listRelease(c->watched_keys); + freeClientMultiState(c); + zfree(c); +} + +/* Replay the append log file. On success C_OK is returned. On non fatal + * error (the append only file is zero-length) C_ERR is returned. On + * fatal error an error message is logged and the program exists. */ +int loadAppendOnlyFile(char *filename) { + struct client *fakeClient; + FILE *fp = fopen(filename,"r"); + struct redis_stat sb; + int old_aof_state = server.aof_state; + long loops = 0; + off_t valid_up_to = 0; /* Offset of latest well-formed command loaded. */ + + if (fp == NULL) { + serverLog(LL_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno)); + exit(1); + } + + /* Handle a zero-length AOF file as a special case. An emtpy AOF file + * is a valid AOF because an empty server with AOF enabled will create + * a zero length file at startup, that will remain like that if no write + * operation is received. */ + if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) { + server.aof_current_size = 0; + fclose(fp); + return C_ERR; + } + + /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI + * to the same file we're about to read. */ + server.aof_state = AOF_OFF; + + fakeClient = createFakeClient(); + startLoading(fp); + + /* Check if this AOF file has an RDB preamble. In that case we need to + * load the RDB file and later continue loading the AOF tail. */ + char sig[5]; /* "REDIS" */ + if (fread(sig,1,5,fp) != 5 || memcmp(sig,"REDIS",5) != 0) { + /* No RDB preamble, seek back at 0 offset. */ + if (fseek(fp,0,SEEK_SET) == -1) goto readerr; + } else { + /* RDB preamble. Pass loading the RDB functions. */ + rio rdb; + + serverLog(LL_NOTICE,"Reading RDB preamble from AOF file..."); + if (fseek(fp,0,SEEK_SET) == -1) goto readerr; + rioInitWithFile(&rdb,fp); + if (rdbLoadRio(&rdb,NULL) != C_OK) { + serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted"); + goto readerr; + } else { + serverLog(LL_NOTICE,"Reading the remaining AOF tail..."); + } + } + + /* Read the actual AOF file, in REPL format, command by command. */ + while(1) { + int argc, j; + unsigned long len; + robj **argv; + char buf[128]; + sds argsds; + struct redisCommand *cmd; + + /* Serve the clients from time to time */ + if (!(loops++ % 1000)) { + loadingProgress(ftello(fp)); + processEventsWhileBlocked(); + } + + if (fgets(buf,sizeof(buf),fp) == NULL) { + if (feof(fp)) + break; + else + goto readerr; + } + if (buf[0] != '*') goto fmterr; + if (buf[1] == '\0') goto readerr; + argc = atoi(buf+1); + if (argc < 1) goto fmterr; + + argv = zmalloc(sizeof(robj*)*argc); + fakeClient->argc = argc; + fakeClient->argv = argv; + + for (j = 0; j < argc; j++) { + if (fgets(buf,sizeof(buf),fp) == NULL) { + fakeClient->argc = j; /* Free up to j-1. */ + freeFakeClientArgv(fakeClient); + goto readerr; + } + if (buf[0] != '$') goto fmterr; + len = strtol(buf+1,NULL,10); + argsds = sdsnewlen(NULL,len); + if (len && fread(argsds,len,1,fp) == 0) { + sdsfree(argsds); + fakeClient->argc = j; /* Free up to j-1. */ + freeFakeClientArgv(fakeClient); + goto readerr; + } + argv[j] = createObject(OBJ_STRING,argsds); + if (fread(buf,2,1,fp) == 0) { + fakeClient->argc = j+1; /* Free up to j. */ + freeFakeClientArgv(fakeClient); + goto readerr; /* discard CRLF */ + } + } + + /* Command lookup */ + cmd = lookupCommand(argv[0]->ptr); + if (!cmd) { + serverLog(LL_WARNING,"Unknown command '%s' reading the append only file", (char*)argv[0]->ptr); + exit(1); + } + + /* Run the command in the context of a fake client */ + fakeClient->cmd = cmd; + cmd->proc(fakeClient); + + /* The fake client should not have a reply */ + serverAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0); + /* The fake client should never get blocked */ + serverAssert((fakeClient->flags & CLIENT_BLOCKED) == 0); + + /* Clean up. Command code may have changed argv/argc so we use the + * argv/argc of the client instead of the local variables. */ + freeFakeClientArgv(fakeClient); + fakeClient->cmd = NULL; + if (server.aof_load_truncated) valid_up_to = ftello(fp); + } + + /* This point can only be reached when EOF is reached without errors. + * If the client is in the middle of a MULTI/EXEC, log error and quit. */ + if (fakeClient->flags & CLIENT_MULTI) goto uxeof; + +loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */ + fclose(fp); + freeFakeClient(fakeClient); + server.aof_state = old_aof_state; + stopLoading(); + aofUpdateCurrentSize(); + server.aof_rewrite_base_size = server.aof_current_size; + return C_OK; + +readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */ + if (!feof(fp)) { + if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */ + serverLog(LL_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno)); + exit(1); + } + +uxeof: /* Unexpected AOF end of file. */ + if (server.aof_load_truncated) { + serverLog(LL_WARNING,"!!! Warning: short read while loading the AOF file !!!"); + serverLog(LL_WARNING,"!!! Truncating the AOF at offset %llu !!!", + (unsigned long long) valid_up_to); + if (valid_up_to == -1 || truncate(filename,valid_up_to) == -1) { + if (valid_up_to == -1) { + serverLog(LL_WARNING,"Last valid command offset is invalid"); + } else { + serverLog(LL_WARNING,"Error truncating the AOF file: %s", + strerror(errno)); + } + } else { + /* Make sure the AOF file descriptor points to the end of the + * file after the truncate call. */ + if (server.aof_fd != -1 && lseek(server.aof_fd,0,SEEK_END) == -1) { + serverLog(LL_WARNING,"Can't seek the end of the AOF file: %s", + strerror(errno)); + } else { + serverLog(LL_WARNING, + "AOF loaded anyway because aof-load-truncated is enabled"); + goto loaded_ok; + } + } + } + if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */ + 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."); + exit(1); + +fmterr: /* Format error. */ + if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */ + 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>"); + exit(1); +} + +/* ---------------------------------------------------------------------------- + * AOF rewrite + * ------------------------------------------------------------------------- */ + +/* Delegate writing an object to writing a bulk string or bulk long long. + * This is not placed in rio.c since that adds the server.h dependency. */ +int rioWriteBulkObject(rio *r, robj *obj) { + /* Avoid using getDecodedObject to help copy-on-write (we are often + * in a child process when this function is called). */ + if (obj->encoding == OBJ_ENCODING_INT) { + return rioWriteBulkLongLong(r,(long)obj->ptr); + } else if (sdsEncodedObject(obj)) { + return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr)); + } else { + serverPanic("Unknown string encoding"); + } +} + +/* Emit the commands needed to rebuild a list object. + * The function returns 0 on error, 1 on success. */ +int rewriteListObject(rio *r, robj *key, robj *o) { + long long count = 0, items = listTypeLength(o); + + if (o->encoding == OBJ_ENCODING_QUICKLIST) { + quicklist *list = o->ptr; + quicklistIter *li = quicklistGetIterator(list, AL_START_HEAD); + quicklistEntry entry; + + while (quicklistNext(li,&entry)) { + if (count == 0) { + int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? + AOF_REWRITE_ITEMS_PER_CMD : items; + if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; + if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0; + if (rioWriteBulkObject(r,key) == 0) return 0; + } + + if (entry.value) { + if (rioWriteBulkString(r,(char*)entry.value,entry.sz) == 0) return 0; + } else { + if (rioWriteBulkLongLong(r,entry.longval) == 0) return 0; + } + if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0; + items--; + } + quicklistReleaseIterator(li); + } else { + serverPanic("Unknown list encoding"); + } + return 1; +} + +/* Emit the commands needed to rebuild a set object. + * The function returns 0 on error, 1 on success. */ +int rewriteSetObject(rio *r, robj *key, robj *o) { + long long count = 0, items = setTypeSize(o); + + if (o->encoding == OBJ_ENCODING_INTSET) { + int ii = 0; + int64_t llval; + + while(intsetGet(o->ptr,ii++,&llval)) { + if (count == 0) { + int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? + AOF_REWRITE_ITEMS_PER_CMD : items; + + if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; + if (rioWriteBulkString(r,"SADD",4) == 0) return 0; + if (rioWriteBulkObject(r,key) == 0) return 0; + } + if (rioWriteBulkLongLong(r,llval) == 0) return 0; + if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0; + items--; + } + } else if (o->encoding == OBJ_ENCODING_HT) { + dictIterator *di = dictGetIterator(o->ptr); + dictEntry *de; + + while((de = dictNext(di)) != NULL) { + sds ele = dictGetKey(de); + if (count == 0) { + int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? + AOF_REWRITE_ITEMS_PER_CMD : items; + + if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0; + if (rioWriteBulkString(r,"SADD",4) == 0) return 0; + if (rioWriteBulkObject(r,key) == 0) return 0; + } + if (rioWriteBulkString(r,ele,sdslen(ele)) == 0) return 0; + if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0; + items--; + } + dictReleaseIterator(di); + } else { + serverPanic("Unknown set encoding"); + } + return 1; +} + +/* Emit the commands needed to rebuild a sorted set object. + * The function returns 0 on error, 1 on success. */ +int rewriteSortedSetObject(rio *r, robj *key, robj *o) { + long long count = 0, items = zsetLength(o); + + if (o->encoding == OBJ_ENCODING_ZIPLIST) { + unsigned char *zl = o->ptr; + unsigned char *eptr, *sptr; + unsigned char *vstr; + unsigned int vlen; + long long vll; + double score; + + eptr = ziplistIndex(zl,0); + serverAssert(eptr != NULL); + sptr = ziplistNext(zl,eptr); + serverAssert(sptr != NULL); + + while (eptr != NULL) { + serverAssert(ziplistGet(eptr,&vstr,&vlen,&vll)); + score = zzlGetScore(sptr); + + if (count == 0) { + int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? + AOF_REWRITE_ITEMS_PER_CMD : items; + + if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; + if (rioWriteBulkString(r,"ZADD",4) == 0) return 0; + if (rioWriteBulkObject(r,key) == 0) return 0; + } + if (rioWriteBulkDouble(r,score) == 0) return 0; + if (vstr != NULL) { + if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0; + } else { + if (rioWriteBulkLongLong(r,vll) == 0) return 0; + } + zzlNext(zl,&eptr,&sptr); + if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0; + items--; + } + } else if (o->encoding == OBJ_ENCODING_SKIPLIST) { + zset *zs = o->ptr; + dictIterator *di = dictGetIterator(zs->dict); + dictEntry *de; + + while((de = dictNext(di)) != NULL) { + sds ele = dictGetKey(de); + double *score = dictGetVal(de); + + if (count == 0) { + int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? + AOF_REWRITE_ITEMS_PER_CMD : items; + + if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; + if (rioWriteBulkString(r,"ZADD",4) == 0) return 0; + if (rioWriteBulkObject(r,key) == 0) return 0; + } + if (rioWriteBulkDouble(r,*score) == 0) return 0; + if (rioWriteBulkString(r,ele,sdslen(ele)) == 0) return 0; + if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0; + items--; + } + dictReleaseIterator(di); + } else { + serverPanic("Unknown sorted zset encoding"); + } + return 1; +} + +/* Write either the key or the value of the currently selected item of a hash. + * The 'hi' argument passes a valid Redis hash iterator. + * The 'what' filed specifies if to write a key or a value and can be + * either OBJ_HASH_KEY or OBJ_HASH_VALUE. + * + * The function returns 0 on error, non-zero on success. */ +static int rioWriteHashIteratorCursor(rio *r, hashTypeIterator *hi, int what) { + if (hi->encoding == OBJ_ENCODING_ZIPLIST) { + unsigned char *vstr = NULL; + unsigned int vlen = UINT_MAX; + long long vll = LLONG_MAX; + + hashTypeCurrentFromZiplist(hi, what, &vstr, &vlen, &vll); + if (vstr) + return rioWriteBulkString(r, (char*)vstr, vlen); + else + return rioWriteBulkLongLong(r, vll); + } else if (hi->encoding == OBJ_ENCODING_HT) { + sds value = hashTypeCurrentFromHashTable(hi, what); + return rioWriteBulkString(r, value, sdslen(value)); + } + + serverPanic("Unknown hash encoding"); + return 0; +} + +/* Emit the commands needed to rebuild a hash object. + * The function returns 0 on error, 1 on success. */ +int rewriteHashObject(rio *r, robj *key, robj *o) { + hashTypeIterator *hi; + long long count = 0, items = hashTypeLength(o); + + hi = hashTypeInitIterator(o); + while (hashTypeNext(hi) != C_ERR) { + if (count == 0) { + int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ? + AOF_REWRITE_ITEMS_PER_CMD : items; + + if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0; + if (rioWriteBulkString(r,"HMSET",5) == 0) return 0; + if (rioWriteBulkObject(r,key) == 0) return 0; + } + + if (rioWriteHashIteratorCursor(r, hi, OBJ_HASH_KEY) == 0) return 0; + if (rioWriteHashIteratorCursor(r, hi, OBJ_HASH_VALUE) == 0) return 0; + if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0; + items--; + } + + hashTypeReleaseIterator(hi); + + return 1; +} + +/* Call the module type callback in order to rewrite a data type + * that is exported by a module and is not handled by Redis itself. + * The function returns 0 on error, 1 on success. */ +int rewriteModuleObject(rio *r, robj *key, robj *o) { + RedisModuleIO io; + moduleValue *mv = o->ptr; + moduleType *mt = mv->type; + moduleInitIOContext(io,mt,r); + mt->aof_rewrite(&io,key,mv->value); + if (io.ctx) { + moduleFreeContext(io.ctx); + zfree(io.ctx); + } + return io.error ? 0 : 1; +} + +/* This function is called by the child rewriting the AOF file to read + * the difference accumulated from the parent into a buffer, that is + * concatenated at the end of the rewrite. */ +ssize_t aofReadDiffFromParent(void) { + char buf[65536]; /* Default pipe buffer size on most Linux systems. */ + ssize_t nread, total = 0; + + while ((nread = + read(server.aof_pipe_read_data_from_parent,buf,sizeof(buf))) > 0) { + server.aof_child_diff = sdscatlen(server.aof_child_diff,buf,nread); + total += nread; + } + return total; +} + +int rewriteAppendOnlyFileRio(rio *aof) { + dictIterator *di = NULL; + dictEntry *de; + size_t processed = 0; + long long now = mstime(); + int j; + + for (j = 0; j < server.dbnum; j++) { + char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n"; + redisDb *db = server.db+j; + dict *d = db->dict; + if (dictSize(d) == 0) continue; + di = dictGetSafeIterator(d); + + /* SELECT the new DB */ + if (rioWrite(aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr; + if (rioWriteBulkLongLong(aof,j) == 0) goto werr; + + /* Iterate this DB writing every entry */ + while((de = dictNext(di)) != NULL) { + sds keystr; + robj key, *o; + long long expiretime; + + keystr = dictGetKey(de); + o = dictGetVal(de); + initStaticStringObject(key,keystr); + + expiretime = getExpire(db,&key); + + /* If this key is already expired skip it */ + if (expiretime != -1 && expiretime < now) continue; + + /* Save the key and associated value */ + if (o->type == OBJ_STRING) { + /* Emit a SET command */ + char cmd[]="*3\r\n$3\r\nSET\r\n"; + if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr; + /* Key and value */ + if (rioWriteBulkObject(aof,&key) == 0) goto werr; + if (rioWriteBulkObject(aof,o) == 0) goto werr; + } else if (o->type == OBJ_LIST) { + if (rewriteListObject(aof,&key,o) == 0) goto werr; + } else if (o->type == OBJ_SET) { + if (rewriteSetObject(aof,&key,o) == 0) goto werr; + } else if (o->type == OBJ_ZSET) { + if (rewriteSortedSetObject(aof,&key,o) == 0) goto werr; + } else if (o->type == OBJ_HASH) { + if (rewriteHashObject(aof,&key,o) == 0) goto werr; + } else if (o->type == OBJ_MODULE) { + if (rewriteModuleObject(aof,&key,o) == 0) goto werr; + } else { + serverPanic("Unknown object type"); + } + /* Save the expire time */ + if (expiretime != -1) { + char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n"; + if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr; + if (rioWriteBulkObject(aof,&key) == 0) goto werr; + if (rioWriteBulkLongLong(aof,expiretime) == 0) goto werr; + } + /* Read some diff from the parent process from time to time. */ + if (aof->processed_bytes > processed+AOF_READ_DIFF_INTERVAL_BYTES) { + processed = aof->processed_bytes; + aofReadDiffFromParent(); + } + } + dictReleaseIterator(di); + di = NULL; + } + return C_OK; + +werr: + if (di) dictReleaseIterator(di); + return C_ERR; +} + +/* Write a sequence of commands able to fully rebuild the dataset into + * "filename". Used both by REWRITEAOF and BGREWRITEAOF. + * + * In order to minimize the number of commands needed in the rewritten + * log Redis uses variadic commands when possible, such as RPUSH, SADD + * and ZADD. However at max AOF_REWRITE_ITEMS_PER_CMD items per time + * are inserted using a single command. */ +int rewriteAppendOnlyFile(char *filename) { + rio aof; + FILE *fp; + char tmpfile[256]; + char byte; + + /* Note that we have to use a different temp name here compared to the + * one used by rewriteAppendOnlyFileBackground() function. */ + snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid()); + fp = fopen(tmpfile,"w"); + if (!fp) { + serverLog(LL_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno)); + return C_ERR; + } + + server.aof_child_diff = sdsempty(); + rioInitWithFile(&aof,fp); + + if (server.aof_rewrite_incremental_fsync) + rioSetAutoSync(&aof,AOF_AUTOSYNC_BYTES); + + if (server.aof_use_rdb_preamble) { + int error; + if (rdbSaveRio(&aof,&error,RDB_SAVE_AOF_PREAMBLE,NULL) == C_ERR) { + errno = error; + goto werr; + } + } else { + if (rewriteAppendOnlyFileRio(&aof) == C_ERR) goto werr; + } + + /* Do an initial slow fsync here while the parent is still sending + * data, in order to make the next final fsync faster. */ + if (fflush(fp) == EOF) goto werr; + if (fsync(fileno(fp)) == -1) goto werr; + + /* Read again a few times to get more data from the parent. + * We can't read forever (the server may receive data from clients + * faster than it is able to send data to the child), so we try to read + * some more data in a loop as soon as there is a good chance more data + * will come. If it looks like we are wasting time, we abort (this + * happens after 20 ms without new data). */ + int nodata = 0; + mstime_t start = mstime(); + while(mstime()-start < 1000 && nodata < 20) { + if (aeWait(server.aof_pipe_read_data_from_parent, AE_READABLE, 1) <= 0) + { + nodata++; + continue; + } + nodata = 0; /* Start counting from zero, we stop on N *contiguous* + timeouts. */ + aofReadDiffFromParent(); + } + + /* Ask the master to stop sending diffs. */ + if (write(server.aof_pipe_write_ack_to_parent,"!",1) != 1) goto werr; + if (anetNonBlock(NULL,server.aof_pipe_read_ack_from_parent) != ANET_OK) + goto werr; + /* We read the ACK from the server using a 10 seconds timeout. Normally + * it should reply ASAP, but just in case we lose its reply, we are sure + * the child will eventually get terminated. */ + if (syncRead(server.aof_pipe_read_ack_from_parent,&byte,1,5000) != 1 || + byte != '!') goto werr; + serverLog(LL_NOTICE,"Parent agreed to stop sending diffs. Finalizing AOF..."); + + /* Read the final diff if any. */ + aofReadDiffFromParent(); + + /* Write the received diff to the file. */ + serverLog(LL_NOTICE, + "Concatenating %.2f MB of AOF diff received from parent.", + (double) sdslen(server.aof_child_diff) / (1024*1024)); + if (rioWrite(&aof,server.aof_child_diff,sdslen(server.aof_child_diff)) == 0) + goto werr; + + /* Make sure data will not remain on the OS's output buffers */ + if (fflush(fp) == EOF) goto werr; + if (fsync(fileno(fp)) == -1) goto werr; + if (fclose(fp) == EOF) goto werr; + + /* Use RENAME to make sure the DB file is changed atomically only + * if the generate DB file is ok. */ + if (rename(tmpfile,filename) == -1) { + serverLog(LL_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno)); + unlink(tmpfile); + return C_ERR; + } + serverLog(LL_NOTICE,"SYNC append only file rewrite performed"); + return C_OK; + +werr: + serverLog(LL_WARNING,"Write error writing append only file on disk: %s", strerror(errno)); + fclose(fp); + unlink(tmpfile); + return C_ERR; +} + +/* ---------------------------------------------------------------------------- + * AOF rewrite pipes for IPC + * -------------------------------------------------------------------------- */ + +/* This event handler is called when the AOF rewriting child sends us a + * single '!' char to signal we should stop sending buffer diffs. The + * parent sends a '!' as well to acknowledge. */ +void aofChildPipeReadable(aeEventLoop *el, int fd, void *privdata, int mask) { + char byte; + UNUSED(el); + UNUSED(privdata); + UNUSED(mask); + + if (read(fd,&byte,1) == 1 && byte == '!') { + serverLog(LL_NOTICE,"AOF rewrite child asks to stop sending diffs."); + server.aof_stop_sending_diff = 1; + if (write(server.aof_pipe_write_ack_to_child,"!",1) != 1) { + /* If we can't send the ack, inform the user, but don't try again + * since in the other side the children will use a timeout if the + * kernel can't buffer our write, or, the children was + * terminated. */ + serverLog(LL_WARNING,"Can't send ACK to AOF child: %s", + strerror(errno)); + } + } + /* Remove the handler since this can be called only one time during a + * rewrite. */ + aeDeleteFileEvent(server.el,server.aof_pipe_read_ack_from_child,AE_READABLE); +} + +/* Create the pipes used for parent - child process IPC during rewrite. + * We have a data pipe used to send AOF incremental diffs to the child, + * and two other pipes used by the children to signal it finished with + * the rewrite so no more data should be written, and another for the + * parent to acknowledge it understood this new condition. */ +int aofCreatePipes(void) { + int fds[6] = {-1, -1, -1, -1, -1, -1}; + int j; + + if (pipe(fds) == -1) goto error; /* parent -> children data. */ + if (pipe(fds+2) == -1) goto error; /* children -> parent ack. */ + if (pipe(fds+4) == -1) goto error; /* children -> parent ack. */ + /* Parent -> children data is non blocking. */ + if (anetNonBlock(NULL,fds[0]) != ANET_OK) goto error; + if (anetNonBlock(NULL,fds[1]) != ANET_OK) goto error; + if (aeCreateFileEvent(server.el, fds[2], AE_READABLE, aofChildPipeReadable, NULL) == AE_ERR) goto error; + + server.aof_pipe_write_data_to_child = fds[1]; + server.aof_pipe_read_data_from_parent = fds[0]; + server.aof_pipe_write_ack_to_parent = fds[3]; + server.aof_pipe_read_ack_from_child = fds[2]; + server.aof_pipe_write_ack_to_child = fds[5]; + server.aof_pipe_read_ack_from_parent = fds[4]; + server.aof_stop_sending_diff = 0; + return C_OK; + +error: + serverLog(LL_WARNING,"Error opening /setting AOF rewrite IPC pipes: %s", + strerror(errno)); + for (j = 0; j < 6; j++) if(fds[j] != -1) close(fds[j]); + return C_ERR; +} + +void aofClosePipes(void) { + aeDeleteFileEvent(server.el,server.aof_pipe_read_ack_from_child,AE_READABLE); + aeDeleteFileEvent(server.el,server.aof_pipe_write_data_to_child,AE_WRITABLE); + close(server.aof_pipe_write_data_to_child); + close(server.aof_pipe_read_data_from_parent); + close(server.aof_pipe_write_ack_to_parent); + close(server.aof_pipe_read_ack_from_child); + close(server.aof_pipe_write_ack_to_child); + close(server.aof_pipe_read_ack_from_parent); +} + +/* ---------------------------------------------------------------------------- + * AOF background rewrite + * ------------------------------------------------------------------------- */ + +/* This is how rewriting of the append only file in background works: + * + * 1) The user calls BGREWRITEAOF + * 2) Redis calls this function, that forks(): + * 2a) the child rewrite the append only file in a temp file. + * 2b) the parent accumulates differences in server.aof_rewrite_buf. + * 3) When the child finished '2a' exists. + * 4) The parent will trap the exit code, if it's OK, will append the + * data accumulated into server.aof_rewrite_buf into the temp file, and + * finally will rename(2) the temp file in the actual file name. + * The the new file is reopened as the new append only file. Profit! + */ +int rewriteAppendOnlyFileBackground(void) { + pid_t childpid; + long long start; + + if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR; + if (aofCreatePipes() != C_OK) return C_ERR; + openChildInfoPipe(); + start = ustime(); + if ((childpid = fork()) == 0) { + char tmpfile[256]; + + /* Child */ + closeListeningSockets(0); + redisSetProcTitle("redis-aof-rewrite"); + snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); + if (rewriteAppendOnlyFile(tmpfile) == C_OK) { + size_t private_dirty = zmalloc_get_private_dirty(-1); + + if (private_dirty) { + serverLog(LL_NOTICE, + "AOF rewrite: %zu MB of memory used by copy-on-write", + private_dirty/(1024*1024)); + } + + server.child_info_data.cow_size = private_dirty; + sendChildInfo(CHILD_INFO_TYPE_AOF); + exitFromChild(0); + } else { + exitFromChild(1); + } + } else { + /* Parent */ + server.stat_fork_time = ustime()-start; + server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */ + latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000); + if (childpid == -1) { + closeChildInfoPipe(); + serverLog(LL_WARNING, + "Can't rewrite append only file in background: fork: %s", + strerror(errno)); + aofClosePipes(); + return C_ERR; + } + serverLog(LL_NOTICE, + "Background append only file rewriting started by pid %d",childpid); + server.aof_rewrite_scheduled = 0; + server.aof_rewrite_time_start = time(NULL); + server.aof_child_pid = childpid; + updateDictResizePolicy(); + /* We set appendseldb to -1 in order to force the next call to the + * feedAppendOnlyFile() to issue a SELECT command, so the differences + * accumulated by the parent into server.aof_rewrite_buf will start + * with a SELECT statement and it will be safe to merge. */ + server.aof_selected_db = -1; + replicationScriptCacheFlush(); + return C_OK; + } + return C_OK; /* unreached */ +} + +void bgrewriteaofCommand(client *c) { + if (server.aof_child_pid != -1) { + addReplyError(c,"Background append only file rewriting already in progress"); + } else if (server.rdb_child_pid != -1) { + server.aof_rewrite_scheduled = 1; + addReplyStatus(c,"Background append only file rewriting scheduled"); + } else if (rewriteAppendOnlyFileBackground() == C_OK) { + addReplyStatus(c,"Background append only file rewriting started"); + } else { + addReply(c,shared.err); + } +} + +void aofRemoveTempFile(pid_t childpid) { + char tmpfile[256]; + + snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid); + unlink(tmpfile); +} + +/* Update the server.aof_current_size field explicitly using stat(2) + * to check the size of the file. This is useful after a rewrite or after + * a restart, normally the size is updated just adding the write length + * to the current length, that is much faster. */ +void aofUpdateCurrentSize(void) { + struct redis_stat sb; + mstime_t latency; + + latencyStartMonitor(latency); + if (redis_fstat(server.aof_fd,&sb) == -1) { + serverLog(LL_WARNING,"Unable to obtain the AOF file length. stat: %s", + strerror(errno)); + } else { + server.aof_current_size = sb.st_size; + } + latencyEndMonitor(latency); + latencyAddSampleIfNeeded("aof-fstat",latency); +} + +/* A background append only file rewriting (BGREWRITEAOF) terminated its work. + * Handle this. */ +void backgroundRewriteDoneHandler(int exitcode, int bysignal) { + if (!bysignal && exitcode == 0) { + int newfd, oldfd; + char tmpfile[256]; + long long now = ustime(); + mstime_t latency; + + serverLog(LL_NOTICE, + "Background AOF rewrite terminated with success"); + + /* Flush the differences accumulated by the parent to the + * rewritten AOF. */ + latencyStartMonitor(latency); + snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", + (int)server.aof_child_pid); + newfd = open(tmpfile,O_WRONLY|O_APPEND); + if (newfd == -1) { + serverLog(LL_WARNING, + "Unable to open the temporary AOF produced by the child: %s", strerror(errno)); + goto cleanup; + } + + if (aofRewriteBufferWrite(newfd) == -1) { + serverLog(LL_WARNING, + "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); + close(newfd); + goto cleanup; + } + latencyEndMonitor(latency); + latencyAddSampleIfNeeded("aof-rewrite-diff-write",latency); + + serverLog(LL_NOTICE, + "Residual parent diff successfully flushed to the rewritten AOF (%.2f MB)", (double) aofRewriteBufferSize() / (1024*1024)); + + /* The only remaining thing to do is to rename the temporary file to + * the configured file and switch the file descriptor used to do AOF + * writes. We don't want close(2) or rename(2) calls to block the + * server on old file deletion. + * + * There are two possible scenarios: + * + * 1) AOF is DISABLED and this was a one time rewrite. The temporary + * file will be renamed to the configured file. When this file already + * exists, it will be unlinked, which may block the server. + * + * 2) AOF is ENABLED and the rewritten AOF will immediately start + * receiving writes. After the temporary file is renamed to the + * configured file, the original AOF file descriptor will be closed. + * Since this will be the last reference to that file, closing it + * causes the underlying file to be unlinked, which may block the + * server. + * + * To mitigate the blocking effect of the unlink operation (either + * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we + * use a background thread to take care of this. First, we + * make scenario 1 identical to scenario 2 by opening the target file + * when it exists. The unlink operation after the rename(2) will then + * be executed upon calling close(2) for its descriptor. Everything to + * guarantee atomicity for this switch has already happened by then, so + * we don't care what the outcome or duration of that close operation + * is, as long as the file descriptor is released again. */ + if (server.aof_fd == -1) { + /* AOF disabled */ + + /* Don't care if this fails: oldfd will be -1 and we handle that. + * One notable case of -1 return is if the old file does + * not exist. */ + oldfd = open(server.aof_filename,O_RDONLY|O_NONBLOCK); + } else { + /* AOF enabled */ + oldfd = -1; /* We'll set this to the current AOF filedes later. */ + } + + /* Rename the temporary file. This will not unlink the target file if + * it exists, because we reference it with "oldfd". */ + latencyStartMonitor(latency); + if (rename(tmpfile,server.aof_filename) == -1) { + serverLog(LL_WARNING, + "Error trying to rename the temporary AOF file %s into %s: %s", + tmpfile, + server.aof_filename, + strerror(errno)); + close(newfd); + if (oldfd != -1) close(oldfd); + goto cleanup; + } + latencyEndMonitor(latency); + latencyAddSampleIfNeeded("aof-rename",latency); + + if (server.aof_fd == -1) { + /* AOF disabled, we don't need to set the AOF file descriptor + * to this new file, so we can close it. */ + close(newfd); + } else { + /* AOF enabled, replace the old fd with the new one. */ + oldfd = server.aof_fd; + server.aof_fd = newfd; + if (server.aof_fsync == AOF_FSYNC_ALWAYS) + aof_fsync(newfd); + else if (server.aof_fsync == AOF_FSYNC_EVERYSEC) + aof_background_fsync(newfd); + server.aof_selected_db = -1; /* Make sure SELECT is re-issued */ + aofUpdateCurrentSize(); + server.aof_rewrite_base_size = server.aof_current_size; + + /* Clear regular AOF buffer since its contents was just written to + * the new AOF from the background rewrite buffer. */ + sdsfree(server.aof_buf); + server.aof_buf = sdsempty(); + } + + server.aof_lastbgrewrite_status = C_OK; + + serverLog(LL_NOTICE, "Background AOF rewrite finished successfully"); + /* Change state from WAIT_REWRITE to ON if needed */ + if (server.aof_state == AOF_WAIT_REWRITE) + server.aof_state = AOF_ON; + + /* Asynchronously close the overwritten AOF. */ + if (oldfd != -1) bioCreateBackgroundJob(BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL); + + serverLog(LL_VERBOSE, + "Background AOF rewrite signal handler took %lldus", ustime()-now); + } else if (!bysignal && exitcode != 0) { + /* SIGUSR1 is whitelisted, so we have a way to kill a child without + * tirggering an error conditon. */ + if (bysignal != SIGUSR1) + server.aof_lastbgrewrite_status = C_ERR; + serverLog(LL_WARNING, + "Background AOF rewrite terminated with error"); + } else { + server.aof_lastbgrewrite_status = C_ERR; + + serverLog(LL_WARNING, + "Background AOF rewrite terminated by signal %d", bysignal); + } + +cleanup: + aofClosePipes(); + aofRewriteBufferReset(); + aofRemoveTempFile(server.aof_child_pid); + server.aof_child_pid = -1; + server.aof_rewrite_time_last = time(NULL)-server.aof_rewrite_time_start; + server.aof_rewrite_time_start = -1; + /* Schedule a new rewrite if we are waiting for it to switch the AOF ON. */ + if (server.aof_state == AOF_WAIT_REWRITE) + server.aof_rewrite_scheduled = 1; +} |