text: ﻿/*
   * 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.
   */
  
  #ifdef _WIN32
  #include "Win32_Interop/win32_types.h"
  #endif
  
  #include "redis.h"
  #include "bio.h"
  #include "rio.h"
  
  #include <signal.h>
  #include <fcntl.h>
  #include <sys/stat.h>
  #include <sys/types.h>
  #ifndef _WIN32
  #include <sys/time.h>
  #include <sys/resource.h>
  #include <sys/wait.h>
  #else
  #include "Win32_Interop/Win32_QFork.h"
  #endif
  
  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 {
  	PORT_ULONG 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. */
  PORT_ULONG aofRewriteBufferSize(void) {
  	listNode* ln;
  	listIter li;
  	PORT_ULONG 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;
  	REDIS_NOTUSED(el);
  	REDIS_NOTUSED(fd);
  	REDIS_NOTUSED(privdata);
  	REDIS_NOTUSED(mask);
  
  	while (1) {
  		ln = listFirst(server.aof_rewrite_buf_blocks);
  		block = ln ? ln->value : NULL;
  		if (server.aof_stop_sending_diff || !block) {
  #ifndef _WIN32
  			aeDeleteFileEvent(server.el, server.aof_pipe_write_data_to_child,
  				AE_WRITABLE);
  #endif
  			return;
  		}
  		if (block->used > 0) {
  			nwritten = FDAPI_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, PORT_ULONG 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) {
  			PORT_ULONG 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 = (int)listLength(server.aof_rewrite_buf_blocks);         WIN_PORT_FIX /* cast (int) */
  				if (((numblocks + 1) % 10) == 0) {
  					int level = ((numblocks + 1) % 100) == 0 ? REDIS_WARNING :
  						REDIS_NOTICE;
  					redisLog(level, "Background AOF buffer size: %Iu MB", WIN_PORT_FIX /* %lu -> %Iu */
  						aofRewriteBufferSize() / (1024 * 1024));
  				}
  		}
  	}
  #ifndef _WIN32
  	/* 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);
  	}
  #endif
  }
  
  /* 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 = FDAPI_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(REDIS_BIO_AOF_FSYNC, (void*)(PORT_LONG)fd, NULL, NULL);
  }
  
  /* Called when the user switches from "appendonly yes" to "appendonly no"
   * at runtime using the CONFIG command. */
  void stopAppendOnly(void) {
  	redisAssert(server.aof_state != REDIS_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 = REDIS_AOF_OFF;
  	/* rewrite operation in progress? kill it, wait child exit */
  	if (server.aof_child_pid != -1) {
  		POSIX_ONLY(int statloc;)
  
  			redisLog(REDIS_NOTICE, "Killing running AOF rewrite child: %Id", WIN_PORT_FIX /* %ld -> %Id */
  			(PORT_LONG) server.aof_child_pid);
  #ifdef _WIN32
  		AbortForkOperation();
  #else
  		if (kill(server.aof_child_pid, SIGUSR1) != -1)
  			wait3(&statloc, 0, NULL);
  #endif
  		/* 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) {
  	server.aof_last_fsync = server.unixtime;
  	server.aof_fd = FDAPI_open(server.aof_filename, O_WRONLY | O_APPEND | O_CREAT WIN32_ONLY(| _O_BINARY), IF_WIN32(_S_IREAD | _S_IWRITE, 0644));
  	redisAssert(server.aof_state == REDIS_AOF_OFF);
  	if (server.aof_fd == -1) {
  		redisLog(REDIS_WARNING, "Redis needs to enable the AOF but can't open the append only file: %s", strerror(errno));
  		return REDIS_ERR;
  	}
  	if (rewriteAppendOnlyFileBackground() == REDIS_ERR) {
  		close(server.aof_fd);
  		WIN32_ONLY(server.aof_fd = -1;)
  			redisLog(REDIS_WARNING, "Redis needs to enable the AOF but can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.");
  		return REDIS_ERR;
  	}
  	/* We correctly switched on AOF, now wait for the rewrite to be complete
  	 * in order to append data on disk. */
  	server.aof_state = REDIS_AOF_WAIT_REWRITE;
  	return REDIS_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(REDIS_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++;
  			redisLog(REDIS_NOTICE, "Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down Redis.");
  		}
  	}
  	/* 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 = FDAPI_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) {
  				redisLog(REDIS_WARNING, "Error writing to the AOF file: %s",
  					strerror(errno));
  				server.aof_last_write_errno = errno;
  			}
  		}
  		else {
  			if (can_log) {
  				redisLog(REDIS_WARNING, "Short write while writing to "
  					"the AOF file: (nwritten=%lld, "
  					"expected=%lld)",
  					(PORT_LONGLONG)nwritten,
  					(PORT_LONGLONG)sdslen(server.aof_buf));
  			}
  
  			if (FDAPI_ftruncate(server.aof_fd, server.aof_current_size) == -1) {
  				if (can_log) {
  					redisLog(REDIS_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. */
  			redisLog(REDIS_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 = REDIS_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, (int)nwritten, -1);                      WIN_PORT_FIX /* cast (int) */
  			}
  			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 == REDIS_ERR) {
  			redisLog(REDIS_WARNING,
  				"AOF write error looks solved, Redis can write again.");
  			server.aof_last_write_status = REDIS_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) {
  	PORT_LONGLONG 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$%Iu\r\n%s\r\n", WIN_PORT_FIX /* %lu -> %Iu */
  		(PORT_ULONG)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 == REDIS_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, (PORT_ULONG)sdslen(buf));    WIN_PORT_FIX /* cast (PORT_ULONG) */
  
  		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 redisClient* createFakeClient(void) {
  	struct redisClient* 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 = REDIS_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 = REDIS_REPL_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 redisClient* c) {
  	int j;
  
  	for (j = 0; j < c->argc; j++)
  		decrRefCount(c->argv[j]);
  	zfree(c->argv);
  }
  
  void freeFakeClient(struct redisClient* c) {
  	sdsfree(c->querybuf);
  	listRelease(c->reply);
  	listRelease(c->watched_keys);
  	freeClientMultiState(c);
  	zfree(c);
  }
  
  /* Replay the append log file. On success REDIS_OK is returned. On non fatal
   * error (the append only file is zero-length) REDIS_ERR is returned. On
   * fatal error an error message is logged and the program exists. */
  int loadAppendOnlyFile(char* filename) {
  	struct redisClient* fakeClient;
  	FILE* fp = fopen(filename, IF_WIN32("rb", "r"));
  	struct redis_stat sb;
  	int old_aof_state = server.aof_state;
  	PORT_LONG loops = 0;
  	off_t valid_up_to = 0; /* Offset of the latest well-formed command loaded. */
  
  	if (fp && redis_fstat(fileno(fp), &sb) != -1 && sb.st_size == 0) {
  		server.aof_current_size = 0;
  		fclose(fp);
  		return REDIS_ERR;
  	}
  
  	if (fp == NULL) {
  		redisLog(REDIS_WARNING, "Fatal error: can't open the append log file for reading: %s", strerror(errno));
  		exit(1);
  	}
  
  	/* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
  	 * to the same file we're about to read. */
  	server.aof_state = REDIS_AOF_OFF;
  
  	fakeClient = createFakeClient();
  	startLoading(fp);
  
  	while (1) {
  		int argc, j;
  		PORT_ULONG len;
  		robj** argv;
  		char buf[128];
  		sds argsds;
  		struct redisCommand* cmd;
  
  		/* Serve the clients from time to time */
  		if (!(loops++ % 1000)) {
  			loadingProgress((off_t)ftello(fp));                                 WIN_PORT_FIX /* cast (off_t) */
  				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(REDIS_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) {
  			redisLog(REDIS_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 */
  		cmd->proc(fakeClient);
  
  		/* The fake client should not have a reply */
  		redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
  		/* The fake client should never get blocked */
  		redisAssert((fakeClient->flags & REDIS_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);
  		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 & REDIS_MULTI) goto uxeof;
  
  loaded_ok: /* DB loaded, cleanup and return REDIS_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 REDIS_OK;
  
  readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */
  	if (!feof(fp)) {
  		redisLog(REDIS_WARNING, "Unrecoverable error reading the append only file: %s", strerror(errno));
  		exit(1);
  	}
  
  uxeof: /* Unexpected AOF end of file. */
  	if (server.aof_load_truncated) {
  		redisLog(REDIS_WARNING, "!!! Warning: short read while loading the AOF file !!!");
  		redisLog(REDIS_WARNING, "!!! Truncating the AOF at offset %llu !!!",
  			(PORT_ULONGLONG)valid_up_to);
  		if (valid_up_to == -1 || truncate(filename, valid_up_to) == -1) {
  			if (valid_up_to == -1) {
  				redisLog(REDIS_WARNING, "Last valid command offset is invalid");
  			}
  			else {
  				redisLog(REDIS_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) {
  				redisLog(REDIS_WARNING, "Can't seek the end of the AOF file: %s",
  					strerror(errno));
  			}
  			else {
  				redisLog(REDIS_WARNING,
  					"AOF loaded anyway because aof-load-truncated is enabled");
  				goto loaded_ok;
  			}
  		}
  	}
  	redisLog(REDIS_WARNING, "Unexpected end of file reading the append only file. You can: 1) Make a backup of your AOF file, then use ./redis-check-aof --fix <filename>. 2) Alternatively you can set the 'aof-load-truncated' configuration option to yes and restart the server.");
  	exit(1);
  
  fmterr: /* Format error. */
  	redisLog(REDIS_WARNING, "Bad file format reading the append only file: make a backup of your AOF file, then use ./redis-check-aof --fix <filename>");
  	exit(1);
  }
  
  /* ----------------------------------------------------------------------------
   * AOF rewrite
   * ------------------------------------------------------------------------- */
  
   /* Delegate writing an object to writing a bulk string or bulk PORT_LONGLONG.
    * This is not placed in rio.c since that adds the redis.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 == REDIS_ENCODING_INT) {
  		return (int)rioWriteBulkLongLong(r, (PORT_LONG)obj->ptr);               WIN_PORT_FIX /* cast (int) */
  	}
  	else if (sdsEncodedObject(obj)) {
  		return (int)rioWriteBulkString(r, obj->ptr, sdslen(obj->ptr));           WIN_PORT_FIX /* cast (int) */
  	}
  	else {
  		redisPanic("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) {
  	PORT_LONGLONG count = 0, items = listTypeLength(o);
  
  	if (o->encoding == REDIS_ENCODING_ZIPLIST) {
  		unsigned char* zl = o->ptr;
  		unsigned char* p = ziplistIndex(zl, 0);
  		unsigned char* vstr;
  		unsigned int vlen;
  		PORT_LONGLONG vlong;
  
  		while (ziplistGet(p, &vstr, &vlen, &vlong)) {
  			if (count == 0) {
  				int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
  					REDIS_AOF_REWRITE_ITEMS_PER_CMD : (int)items;               WIN_PORT_FIX /* cast (int) */
  
  					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 (vstr) {
  				if (rioWriteBulkString(r, (char*)vstr, vlen) == 0) return 0;
  			}
  			else {
  				if (rioWriteBulkLongLong(r, vlong) == 0) return 0;
  			}
  			p = ziplistNext(zl, p);
  			if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
  			items--;
  		}
  	}
  	else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
  		list* list = o->ptr;
  		listNode* ln;
  		listIter li;
  
  		listRewind(list, &li);
  		while ((ln = listNext(&li))) {
  			robj* eleobj = listNodeValue(ln);
  
  			if (count == 0) {
  				int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
  					REDIS_AOF_REWRITE_ITEMS_PER_CMD : (int)items;               WIN_PORT_FIX /* cast (int) */
  
  					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 (rioWriteBulkObject(r, eleobj) == 0) return 0;
  			if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
  			items--;
  		}
  	}
  	else {
  		redisPanic("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) {
  	PORT_LONGLONG count = 0, items = setTypeSize(o);
  
  	if (o->encoding == REDIS_ENCODING_INTSET) {
  		int ii = 0;
  		int64_t llval;
  
  		while (intsetGet(o->ptr, ii++, &llval)) {
  			if (count == 0) {
  				int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
  					REDIS_AOF_REWRITE_ITEMS_PER_CMD : (int)items;               WIN_PORT_FIX /* cast (int) */
  
  					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 == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
  			items--;
  		}
  	}
  	else if (o->encoding == REDIS_ENCODING_HT) {
  		dictIterator* di = dictGetIterator(o->ptr);
  		dictEntry* de;
  
  		while ((de = dictNext(di)) != NULL) {
  			robj* eleobj = dictGetKey(de);
  			if (count == 0) {
  				int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
  					REDIS_AOF_REWRITE_ITEMS_PER_CMD : (int)items;               WIN_PORT_FIX /* cast (int) */
  
  					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 (rioWriteBulkObject(r, eleobj) == 0) return 0;
  			if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
  			items--;
  		}
  		dictReleaseIterator(di);
  	}
  	else {
  		redisPanic("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) {
  	PORT_LONGLONG count = 0, items = zsetLength(o);
  
  	if (o->encoding == REDIS_ENCODING_ZIPLIST) {
  		unsigned char* zl = o->ptr;
  		unsigned char* eptr, * sptr;
  		unsigned char* vstr;
  		unsigned int vlen;
  		PORT_LONGLONG vll;
  		double score;
  
  		eptr = ziplistIndex(zl, 0);
  		redisAssert(eptr != NULL);
  		sptr = ziplistNext(zl, eptr);
  		redisAssert(sptr != NULL);
  
  		while (eptr != NULL) {
  			redisAssert(ziplistGet(eptr, &vstr, &vlen, &vll));
  			score = zzlGetScore(sptr);
  
  			if (count == 0) {
  				int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
  					REDIS_AOF_REWRITE_ITEMS_PER_CMD : (int)items;               WIN_PORT_FIX /* cast (int) */
  
  					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 == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
  			items--;
  		}
  	}
  	else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
  		zset* zs = o->ptr;
  		dictIterator* di = dictGetIterator(zs->dict);
  		dictEntry* de;
  
  		while ((de = dictNext(di)) != NULL) {
  			robj* eleobj = dictGetKey(de);
  			double* score = dictGetVal(de);
  
  			if (count == 0) {
  				int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
  					REDIS_AOF_REWRITE_ITEMS_PER_CMD : (int)items;               WIN_PORT_FIX /* cast (int) */
  
  					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 (rioWriteBulkObject(r, eleobj) == 0) return 0;
  			if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
  			items--;
  		}
  		dictReleaseIterator(di);
  	}
  	else {
  		redisPanic("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 REDIS_HASH_KEY or REDIS_HASH_VALUE.
   *
   * The function returns 0 on error, non-zero on success. */
  static int rioWriteHashIteratorCursor(rio* r, hashTypeIterator* hi, int what) {
  	if (hi->encoding == REDIS_ENCODING_ZIPLIST) {
  		unsigned char* vstr = NULL;
  		unsigned int vlen = UINT_MAX;
  		PORT_LONGLONG vll = LLONG_MAX;
  
  		hashTypeCurrentFromZiplist(hi, what, &vstr, &vlen, &vll);
  		if (vstr) {
  			return (int)rioWriteBulkString(r, (char*)vstr, vlen);       WIN_PORT_FIX /* cast (int) */
  		}
  		else {
  			return (int)rioWriteBulkLongLong(r, vll);                   WIN_PORT_FIX /* cast (int) */
  		}
  
  	}
  	else if (hi->encoding == REDIS_ENCODING_HT) {
  		robj* value;
  
  		hashTypeCurrentFromHashTable(hi, what, &value);
  		return rioWriteBulkObject(r, value);
  	}
  
  	redisPanic("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;
  	PORT_LONGLONG count = 0, items = hashTypeLength(o);
  
  	hi = hashTypeInitIterator(o);
  	while (hashTypeNext(hi) != REDIS_ERR) {
  		if (count == 0) {
  			int cmd_items = (int)((items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ? WIN_PORT_FIX /* cast (int) */
  				REDIS_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, REDIS_HASH_KEY) == 0) return 0;
  		if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_VALUE) == 0) return 0;
  		if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
  		items--;
  	}
  
  	hashTypeReleaseIterator(hi);
  
  	return 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 =
  		FDAPI_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;
  }
  
  /* 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 REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time
   * are inserted using a single command. */
  int rewriteAppendOnlyFile(char* filename) {
  	dictIterator* di = NULL;
  	dictEntry* de;
  	rio aof;
  	FILE* fp;
  	char tmpfile[256];
  	int j;
  	PORT_LONGLONG now = mstime();
  	char byte;
  	size_t processed = 0;
  
  	/* 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, IF_WIN32("wb", "w"));
  	if (!fp) {
  		redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
  		return REDIS_ERR;
  	}
  
  	server.aof_child_diff = sdsempty();
  	rioInitWithFile(&aof, fp);
  	if (server.aof_rewrite_incremental_fsync)
  		rioSetAutoSync(&aof, REDIS_AOF_AUTOSYNC_BYTES);
  	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);
  		if (!di) {
  			fclose(fp);
  			return REDIS_ERR;
  		}
  
  		/* 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;
  			PORT_LONGLONG 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 == REDIS_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 == REDIS_LIST) {
  				if (rewriteListObject(&aof, &key, o) == 0) goto werr;
  			}
  			else if (o->type == REDIS_SET) {
  				if (rewriteSetObject(&aof, &key, o) == 0) goto werr;
  			}
  			else if (o->type == REDIS_ZSET) {
  				if (rewriteSortedSetObject(&aof, &key, o) == 0) goto werr;
  			}
  			else if (o->type == REDIS_HASH) {
  				if (rewriteHashObject(&aof, &key, o) == 0) goto werr;
  			}
  			else {
  				redisPanic("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 + 1024 * 10) {
  				processed = aof.processed_bytes;
  				aofReadDiffFromParent();
  			}
  		}
  		dictReleaseIterator(di);
  		di = NULL;
  	}
  
  	/* 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 (FDAPI_fsync(FDAPI_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 (FDAPI_write(server.aof_pipe_write_ack_to_parent, "!", 1) != 1) goto werr;
  #ifndef _WIN32
  	if (anetNonBlock(NULL, server.aof_pipe_read_ack_from_parent) != ANET_OK)
  		goto werr;
  #endif
  	/* 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;
  	redisLog(REDIS_NOTICE, "Parent agreed to stop sending diffs. Finalizing AOF...");
  
  	/* Read the final diff if any. */
  	aofReadDiffFromParent();
  
  	/* Write the received diff to the file. */
  	redisLog(REDIS_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 (FDAPI_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) {
  		redisLog(REDIS_WARNING, "Error moving temp append only file on the final destination: %s", strerror(errno));
  		unlink(tmpfile);
  		return REDIS_ERR;
  	}
  	redisLog(REDIS_NOTICE, "SYNC append only file rewrite performed");
  	return REDIS_OK;
  
  werr:
  	redisLog(REDIS_WARNING, "Write error writing append only file on disk: %s", strerror(errno));
  	fclose(fp);
  	unlink(tmpfile);
  	if (di) dictReleaseIterator(di);
  	return REDIS_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;
  	REDIS_NOTUSED(el);
  	REDIS_NOTUSED(privdata);
  	REDIS_NOTUSED(mask);
  
  	if (FDAPI_read(fd, &byte, 1) == 1 && byte == '!') {
  		redisLog(REDIS_NOTICE, "AOF rewrite child asks to stop sending diffs.");
  		server.aof_stop_sending_diff = 1;
  		if (FDAPI_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. */
  			redisLog(REDIS_WARNING, "Can't send ACK to AOF child: %s",
  				strerror(errno));
  		}
  	}
  #ifndef _WIN32
  	/* 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);
  #endif
  }
  
  /* 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 (FDAPI_pipe(fds) == -1) goto error; /* parent -> children data. */
  	if (FDAPI_pipe(fds + 2) == -1) goto error; /* children -> parent ack. */
  	if (FDAPI_pipe(fds + 4) == -1) goto error; /* children -> parent ack. */
  	/* Parent -> children data is non blocking. */
  #ifndef WIN32
  	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;
  #else
  	if (FDAPI_PipeSetNonBlock(fds[0], 1) != 0) goto error;
  	if (FDAPI_PipeSetNonBlock(fds[1], 1) != 0) goto error;
  	if (FDAPI_PipeSetNonBlock(fds[2], 1) != 0) goto error;
  #endif
  
  	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 REDIS_OK;
  
  error:
  	redisLog(REDIS_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 REDIS_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;
  	PORT_LONGLONG start;
  
  	if (server.aof_child_pid != -1) return REDIS_ERR;
  	if (aofCreatePipes() != REDIS_OK) return REDIS_ERR;
  	start = ustime();
  
  #ifndef _WIN32
  	if ((childpid = fork()) == 0) {
  #endif
  		char tmpfile[256];
  
  #ifndef _WIN32
  		/* Child */
  		closeListeningSockets(0);
  		redisSetProcTitle("redis-aof-rewrite");
  #endif
  		snprintf(tmpfile, 256, "temp-rewriteaof-bg-%d.aof", (int)getpid());
  #ifdef _WIN32
  		childpid = BeginForkOperation_Aof(server.aof_pipe_write_ack_to_parent,
  			server.aof_pipe_read_ack_from_parent,
  			server.aof_pipe_read_data_from_parent,
  			tmpfile,
  			&server,
  			sizeof(server),
  			dictGetHashFunctionSeed());
  #else
  		if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
  			size_t private_dirty = zmalloc_get_private_dirty();
  
  			if (private_dirty) {
  				redisLog(REDIS_NOTICE,
  					"AOF rewrite: %lu MB of memory used by copy-on-write",
  					private_dirty / (1024 * 1024));
  			}
  			exitFromChild(0);
  		}
  		else {
  			exitFromChild(1);
  		}
  	}
  	else {
  #endif
  		/* 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) {
  			redisLog(REDIS_WARNING,
  				"Can't rewrite append only file in background: fork: %s",
  				strerror(errno));
  			return REDIS_ERR;
  		}
  		redisLog(REDIS_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 REDIS_OK;
  #ifndef _WIN32
  	}
  #endif
  	return REDIS_OK; /* unreached */
  }
  
  void bgrewriteaofCommand(redisClient* 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() == REDIS_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;
  
  #ifdef _WIN32
  	if (server.aof_fd == -1) {
  		redisLog(REDIS_NOTICE, "Unable to check the AOF length: %s", "appendfd is -1");
  		return;
  	}
  #endif
  
  	latencyStartMonitor(latency);
  	if (redis_fstat(server.aof_fd, &sb) == -1) {
  		redisLog(REDIS_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];
  		PORT_LONGLONG now = ustime();
  		mstime_t latency;
  
  		redisLog(REDIS_NOTICE,
  			"Background AOF rewrite terminated with success");
  
  		/* Flush the differences accumulated by the parent to the
  		 * rewritten AOF. */
  		latencyStartMonitor(latency);
  #ifdef _WIN32
  		snprintf(tmpfile, 256, "temp-rewriteaof-bg-%d.aof",
  			getpid());
  		newfd = FDAPI_open(tmpfile, O_WRONLY | O_APPEND | O_CREAT | _O_BINARY, _S_IREAD | _S_IWRITE);
  #else
  		snprintf(tmpfile, 256, "temp-rewriteaof-bg-%d.aof",
  			(int)server.aof_child_pid);
  		newfd = FDAPI_open(tmpfile, O_WRONLY | O_APPEND);
  #endif
  		if (newfd == -1) {
  			redisLog(REDIS_WARNING,
  				"Unable to open the temporary AOF produced by the child: %s", strerror(errno));
  			goto cleanup;
  		}
  
  		if (aofRewriteBufferWrite(newfd) == -1) {
  			redisLog(REDIS_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);
  
  		redisLog(REDIS_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. */
  #ifdef _WIN32
  		 // On Windows, if AOF is disabled and an AOF file already exists,
  		 // we don't need to open it since it will have to be closed anyway
  		 // before renaming the temp AOF.
  		oldfd = -1; /* We'll set this to the current AOF filedes later. */
  
  		char tmpfile_win_old[256];
  		snprintf(tmpfile_win_old, 256, "temp-rewriteaof-old-%d.aof",
  			(int)server.aof_child_pid);
  
  		if (server.aof_fd != -1) {
  			// AOF enabled, close the existing AOF file
  			close(server.aof_fd);
  			// Now rename the existing AOF file to allow the new file to be renamed
  			if (rename(server.aof_filename, tmpfile_win_old) == -1) {
  				redisLog(REDIS_WARNING,
  					"Error trying to rename the existing AOF to old tempfile: %s", strerror(errno));
  				// Let's clean the Windows-specific temp file here
  				unlink(tmpfile_win_old);
  				goto cleanup;
  			}
  		}
  		latencyStartMonitor(latency);
  		// Close the temp AOF file before renaming it
  		close(newfd);
  		if (rename(tmpfile, server.aof_filename) == -1) {
  			redisLog(REDIS_WARNING,
  				"Error trying to rename the temporary AOF: %s", strerror(errno));
  			if (server.aof_fd != -1) {
  				if (rename(tmpfile_win_old, server.aof_filename) == -1) {
  					redisLog(REDIS_WARNING,
  						"Error trying to rename the existing AOF from old tempfile: %s", strerror(errno));
  					// The Windows-specific temp file couldn't be renamed to
  					// the configured AOF file, that should never happen but
  					// if it happens we leave the file behind in case the user
  					// needs it
  				}
  			}
  			goto cleanup;
  		}
  		/* now open the files again with new names */
  		newfd = FDAPI_open(server.aof_filename, O_WRONLY | O_APPEND | _O_BINARY, 0);
  		if (newfd == -1) {
  			/* Windows fix: More info */
  			redisLog(REDIS_WARNING, "Not able to reopen the temporary AOF file after rename");
  			goto cleanup;
  		}
  		if (server.aof_fd != -1) {
  			server.aof_fd = FDAPI_open(
  				tmpfile_win_old,
  				O_WRONLY | O_APPEND | O_CREAT | _O_BINARY | _O_TEMPORARY,     // _O_TEMPORARY forces delete on close flag in CreateFile call. File will be deleted in REDIS_BIO_CLOSE_FILE job.
  				0644);
  		}
  #else
  		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) {
  			redisLog(REDIS_WARNING,
  				"Error trying to rename the temporary AOF file: %s", strerror(errno));
  			close(newfd);
  			if (oldfd != -1) close(oldfd);
  			goto cleanup;
  		}
  #endif
  		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 = REDIS_OK;
  
  		redisLog(REDIS_NOTICE, "Background AOF rewrite finished successfully");
  		/* Change state from WAIT_REWRITE to ON if needed */
  		if (server.aof_state == REDIS_AOF_WAIT_REWRITE)
  			server.aof_state = REDIS_AOF_ON;
  
  		/* Asynchronously close the overwritten AOF. */
  		if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE, (void*)(PORT_LONG)oldfd, NULL, NULL);
  
  		redisLog(REDIS_VERBOSE,
  			"Background AOF rewrite signal handler took %lldus", ustime() - now);
  	}
  	else if (!bysignal && exitcode != 0) {
  		server.aof_lastbgrewrite_status = REDIS_ERR;
  
  		redisLog(REDIS_WARNING,
  			"Background AOF rewrite terminated with error");
  	}
  	else {
  		server.aof_lastbgrewrite_status = REDIS_ERR;
  
  		redisLog(REDIS_WARNING,
  			"Background AOF rewrite terminated by signal %d", bysignal);
  	}
  
  cleanup:
  	aofClosePipes();
  	aofRewriteBufferReset();
  	aofRemoveTempFile(IF_WIN32(getpid(), 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 == REDIS_AOF_WAIT_REWRITE)
  		server.aof_rewrite_scheduled = 1;
  }
  
  void aofProcessDiffRewriteEvents(aeEventLoop* eventLoop)
  {
  	// only do these checks in the parent process and if an aof rewrite is in progress
  	if (server.aof_child_pid != -1 && server.aof_pipe_read_ack_from_child != -1) {
  		//1) check if more data can be written to the child and write it.
  		// in which case we dont need to send any more diffs to the parent
  		if (server.aof_stop_sending_diff == 0) {
  			aofChildWriteDiffData(eventLoop, server.aof_pipe_write_data_to_child, NULL, 0);
  		}
  
  		//2) check if child has signaled parent to stop sending diffs
  		if (server.aof_stop_sending_diff == 0) {
  			aofChildPipeReadable(eventLoop, server.aof_pipe_read_ack_from_child, NULL, 0);
  		}
  	}
  }
  