void aofUpdateCurrentSize(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 rerwite buffer. */
+unsigned long aofRewriteBufferSize(void) {
+ listNode *ln = listLast(server.aof_rewrite_buf_blocks);
+ aofrwblock *block = ln ? ln->value : NULL;
+
+ if (block == NULL) return 0;
+ unsigned long size =
+ (listLength(server.aof_rewrite_buf_blocks)-1) * AOF_RW_BUF_BLOCK_SIZE;
+ size += block->used;
+ return size;
+}
+
+/* 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 ? REDIS_WARNING :
+ REDIS_NOTICE;
+ redisLog(level,"Background AOF buffer size: %lu MB",
+ aofRewriteBufferSize()/(1024*1024));
+ }
+ }
+ }
+}
+
+/* Write the buffer (possibly composed of multiple blocks) into the specified
+ * fd. If no 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 != 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*)(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.appendfd);
- close(server.appendfd);
+ aof_fsync(server.aof_fd);
+ close(server.aof_fd);
- server.appendfd = -1;
- server.appendseldb = -1;
- server.appendonly = 0;
+ 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.bgrewritechildpid != -1) {
+ if (server.aof_child_pid != -1) {
int statloc;
- if (kill(server.bgrewritechildpid,SIGKILL) != -1)
+ redisLog(REDIS_NOTICE,"Killing running AOF rewrite child: %ld",
+ (long) server.aof_child_pid);
+ if (kill(server.aof_child_pid,SIGKILL) != -1)
wait3(&statloc,0,NULL);
/* reset the buffer accumulating changes while the child saves */
- sdsfree(server.bgrewritebuf);
- server.bgrewritebuf = sdsempty();
- server.bgrewritechildpid = -1;
+ aofRewriteBufferReset();
+ aofRemoveTempFile(server.aof_child_pid);
+ server.aof_child_pid = -1;
}
}
/* Called when the user switches from "appendonly no" to "appendonly yes"
* at runtime using the CONFIG command. */
int startAppendOnly(void) {
- server.appendonly = 1;
- server.lastfsync = time(NULL);
- server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
- if (server.appendfd == -1) {
- redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, but I can't open the AOF file: %s",strerror(errno));
+ server.aof_last_fsync = server.unixtime;
+ server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,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) {
- server.appendonly = 0;
- close(server.appendfd);
- redisLog(REDIS_WARNING,"User tried turning on AOF with CONFIG SET but I can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.");
+ close(server.aof_fd);
+ 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 rerwite to be complete
+ * in order to append data on disk. */
+ server.aof_state = REDIS_AOF_WAIT_REWRITE;
return REDIS_OK;
}
ssize_t nwritten;
int sync_in_progress = 0;
- if (sdslen(server.aofbuf) == 0) return;
+ if (sdslen(server.aof_buf) == 0) return;
- if (server.appendfsync == APPENDFSYNC_EVERYSEC)
+ if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0;
- if (server.appendfsync == APPENDFSYNC_EVERYSEC && !force) {
+ 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. */
}
/* 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.");
}
}
* 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 */
- nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf));
- if (nwritten != (signed)sdslen(server.aofbuf)) {
+ nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
+ if (nwritten != (signed)sdslen(server.aof_buf)) {
/* Ooops, we are in troubles. The best thing to do for now is
* aborting instead of giving the illusion that everything is
* working as expected. */
if (nwritten == -1) {
redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno));
} else {
- redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno));
+ redisLog(REDIS_WARNING,"Exiting on short write while writing to "
+ "the append-only file: %s (nwritten=%ld, "
+ "expected=%ld)",
+ strerror(errno),
+ (long)nwritten,
+ (long)sdslen(server.aof_buf));
}
exit(1);
}
- server.appendonly_current_size += nwritten;
+ 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.aofbuf)+sdsavail(server.aofbuf)) < 4000) {
- sdsclear(server.aofbuf);
+ if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
+ sdsclear(server.aof_buf);
} else {
- sdsfree(server.aofbuf);
- server.aofbuf = sdsempty();
+ 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.no_appendfsync_on_rewrite &&
- (server.bgrewritechildpid != -1 || server.bgsavechildpid != -1))
+ 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.appendfsync == APPENDFSYNC_ALWAYS) {
+ if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
/* aof_fsync is defined as fdatasync() for Linux in order to avoid
* flushing metadata. */
- aof_fsync(server.appendfd); /* Let's try to get this data on the disk */
- server.lastfsync = server.unixtime;
- } else if ((server.appendfsync == APPENDFSYNC_EVERYSEC &&
- server.unixtime > server.lastfsync)) {
- if (!sync_in_progress) aof_background_fsync(server.appendfd);
- server.lastfsync = server.unixtime;
+ aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */
+ 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;
}
}
/* The DB this command was targetting is not the same as the last command
* we appendend. To issue a SELECT command is needed. */
- if (dictid != server.appendseldb) {
+ 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.appendseldb = dictid;
+ server.aof_selected_db = dictid;
}
if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
/* 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. */
- server.aofbuf = sdscatlen(server.aofbuf,buf,sdslen(buf));
+ 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.bgrewritechildpid != -1)
- server.bgrewritebuf = sdscatlen(server.bgrewritebuf,buf,sdslen(buf));
+ 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 redisClient *createFakeClient(void) {
selectDb(c,0);
c->fd = -1;
c->querybuf = sdsempty();
+ c->querybuf_peak = 0;
c->argc = 0;
c->argv = NULL;
c->bufpos = 0;
* 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();
listSetFreeMethod(c->reply,decrRefCount);
listSetDupMethod(c->reply,dupClientReplyValue);
struct redisClient *fakeClient;
FILE *fp = fopen(filename,"r");
struct redis_stat sb;
- int appendonly = server.appendonly;
+ int old_aof_state = server.aof_state;
long loops = 0;
if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
- server.appendonly_current_size = 0;
+ server.aof_current_size = 0;
fclose(fp);
return REDIS_ERR;
}
/* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
* to the same file we're about to read. */
- server.appendonly = 0;
+ server.aof_state = REDIS_AOF_OFF;
fakeClient = createFakeClient();
startLoading(fp);
fclose(fp);
freeFakeClient(fakeClient);
- server.appendonly = appendonly;
+ server.aof_state = old_aof_state;
stopLoading();
aofUpdateCurrentSize();
- server.auto_aofrewrite_base_size = server.appendonly_current_size;
+ server.aof_rewrite_base_size = server.aof_current_size;
return REDIS_OK;
readerr:
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 redis.h dependency. */
int rioWriteBulkObject(rio *r, robj *obj) {
while(ziplistGet(p,&vstr,&vlen,&vlong)) {
if (count == 0) {
- int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
- REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+ int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
+ REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
if (rioWriteBulkLongLong(r,vlong) == 0) return 0;
}
p = ziplistNext(zl,p);
- if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+ if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
items--;
}
} else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
robj *eleobj = listNodeValue(ln);
if (count == 0) {
- int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
- REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+ int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
+ REDIS_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 (rioWriteBulkObject(r,eleobj) == 0) return 0;
- if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+ if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
items--;
}
} else {
while(intsetGet(o->ptr,ii++,&llval)) {
if (count == 0) {
- int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
- REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+ int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
+ REDIS_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 == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+ if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
items--;
}
} else if (o->encoding == REDIS_ENCODING_HT) {
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetKey(de);
if (count == 0) {
- int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
- REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
+ int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
+ REDIS_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 (rioWriteBulkObject(r,eleobj) == 0) return 0;
- if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
+ if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
items--;
}
dictReleaseIterator(di);
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 == REDIS_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);
+ 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 : 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 == 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 : 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 (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 an 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;
+ 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 == 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;
+ long long count = 0, items = hashTypeLength(o);
+
+ hi = hashTypeInitIterator(o);
+ while (hashTypeNext(hi) != REDIS_ERR) {
+ if (count == 0) {
+ int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
+ 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;
+}
+
/* 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_AOFREWRITE_ITEMS_PER_CMD items per time
+ * 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;
snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
fp = fopen(tmpfile,"w");
if (!fp) {
- redisLog(REDIS_WARNING, "Failed rewriting the append only file: %s", strerror(errno));
+ redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
return REDIS_ERR;
}
} else if (o->type == REDIS_SET) {
if (rewriteSetObject(&aof,&key,o) == 0) goto werr;
} else if (o->type == REDIS_ZSET) {
- /* Emit the ZADDs needed to rebuild the sorted set */
- char cmd[]="*4\r\n$4\r\nZADD\r\n";
-
- if (o->encoding == REDIS_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);
- redisAssert(eptr != NULL);
- sptr = ziplistNext(zl,eptr);
- redisAssert(sptr != NULL);
-
- while (eptr != NULL) {
- redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
- score = zzlGetScore(sptr);
-
- if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
- if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
- if (rioWriteBulkDouble(&aof,score) == 0) goto werr;
- if (vstr != NULL) {
- if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0)
- goto werr;
- } else {
- if (rioWriteBulkLongLong(&aof,vll) == 0)
- goto werr;
- }
- zzlNext(zl,&eptr,&sptr);
- }
- } 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 (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
- if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
- if (rioWriteBulkDouble(&aof,*score) == 0) goto werr;
- if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr;
- }
- dictReleaseIterator(di);
- } else {
- redisPanic("Unknown sorted set encoding");
- }
+ if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
} else if (o->type == REDIS_HASH) {
- char cmd[]="*4\r\n$4\r\nHSET\r\n";
-
- /* Emit the HSETs needed to rebuild the hash */
- if (o->encoding == REDIS_ENCODING_ZIPMAP) {
- unsigned char *p = zipmapRewind(o->ptr);
- unsigned char *field, *val;
- unsigned int flen, vlen;
-
- while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
- if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
- if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
- if (rioWriteBulkString(&aof,(char*)field,flen) == 0)
- goto werr;
- if (rioWriteBulkString(&aof,(char*)val,vlen) == 0)
- goto werr;
- }
- } else {
- dictIterator *di = dictGetIterator(o->ptr);
- dictEntry *de;
-
- while((de = dictNext(di)) != NULL) {
- robj *field = dictGetKey(de);
- robj *val = dictGetVal(de);
-
- if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
- if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
- if (rioWriteBulkObject(&aof,field) == 0) goto werr;
- if (rioWriteBulkObject(&aof,val) == 0) goto werr;
- }
- dictReleaseIterator(di);
- }
+ if (rewriteHashObject(&aof,&key,o) == 0) goto werr;
} else {
redisPanic("Unknown object type");
}
* 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.bgrewritebuf.
+ * 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.bgrewritebuf into the temp file, and
+ * 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!
*/
pid_t childpid;
long long start;
- if (server.bgrewritechildpid != -1) return REDIS_ERR;
+ if (server.aof_child_pid != -1) return REDIS_ERR;
start = ustime();
if ((childpid = fork()) == 0) {
char tmpfile[256];
if (server.sofd > 0) close(server.sofd);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
- _exit(0);
+ exitFromChild(0);
} else {
- _exit(1);
+ exitFromChild(1);
}
} else {
/* Parent */
}
redisLog(REDIS_NOTICE,
"Background append only file rewriting started by pid %d",childpid);
- server.aofrewrite_scheduled = 0;
- server.bgrewritechildpid = childpid;
+ server.aof_rewrite_scheduled = 0;
+ 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.bgrewritebuf will start
+ * accumulated by the parent into server.aof_rewrite_buf will start
* with a SELECT statement and it will be safe to merge. */
- server.appendseldb = -1;
+ server.aof_selected_db = -1;
return REDIS_OK;
}
return REDIS_OK; /* unreached */
}
void bgrewriteaofCommand(redisClient *c) {
- if (server.bgrewritechildpid != -1) {
+ if (server.aof_child_pid != -1) {
addReplyError(c,"Background append only file rewriting already in progress");
- } else if (server.bgsavechildpid != -1) {
- server.aofrewrite_scheduled = 1;
+ } 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");
unlink(tmpfile);
}
-/* Update the server.appendonly_current_size filed explicitly using stat(2)
+/* Update the server.aof_current_size filed 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 lenght, that is much faster. */
+ * to the current length, that is much faster. */
void aofUpdateCurrentSize(void) {
struct redis_stat sb;
- if (redis_fstat(server.appendfd,&sb) == -1) {
- redisLog(REDIS_WARNING,"Unable to check the AOF length: %s",
+ if (redis_fstat(server.aof_fd,&sb) == -1) {
+ redisLog(REDIS_WARNING,"Unable to obtain the AOF file length. stat: %s",
strerror(errno));
} else {
- server.appendonly_current_size = sb.st_size;
+ server.aof_current_size = sb.st_size;
}
}
void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
if (!bysignal && exitcode == 0) {
int newfd, oldfd;
- int nwritten;
char tmpfile[256];
long long now = ustime();
/* Flush the differences accumulated by the parent to the
* rewritten AOF. */
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
- (int)server.bgrewritechildpid);
+ (int)server.aof_child_pid);
newfd = open(tmpfile,O_WRONLY|O_APPEND);
if (newfd == -1) {
redisLog(REDIS_WARNING,
goto cleanup;
}
- nwritten = write(newfd,server.bgrewritebuf,sdslen(server.bgrewritebuf));
- if (nwritten != (signed)sdslen(server.bgrewritebuf)) {
- if (nwritten == -1) {
- redisLog(REDIS_WARNING,
- "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
- } else {
- redisLog(REDIS_WARNING,
- "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
- }
+ 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;
}
redisLog(REDIS_NOTICE,
- "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten);
+ "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", aofRewriteBufferSize());
/* 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
* 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.appendfd == -1) {
+ 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.appendfilename,O_RDONLY|O_NONBLOCK);
+ 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". */
- if (rename(tmpfile,server.appendfilename) == -1) {
+ if (rename(tmpfile,server.aof_filename) == -1) {
redisLog(REDIS_WARNING,
- "Error trying to rename the temporary AOF: %s", strerror(errno));
+ "Error trying to rename the temporary AOF file: %s", strerror(errno));
close(newfd);
if (oldfd != -1) close(oldfd);
goto cleanup;
}
- if (server.appendfd == -1) {
+ 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.appendfd;
- server.appendfd = newfd;
- if (server.appendfsync == APPENDFSYNC_ALWAYS)
+ oldfd = server.aof_fd;
+ server.aof_fd = newfd;
+ if (server.aof_fsync == AOF_FSYNC_ALWAYS)
aof_fsync(newfd);
- else if (server.appendfsync == APPENDFSYNC_EVERYSEC)
+ else if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
aof_background_fsync(newfd);
- server.appendseldb = -1; /* Make sure SELECT is re-issued */
+ server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
aofUpdateCurrentSize();
- server.auto_aofrewrite_base_size = server.appendonly_current_size;
+ 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.aofbuf);
- server.aofbuf = sdsempty();
+ sdsfree(server.aof_buf);
+ server.aof_buf = sdsempty();
}
- redisLog(REDIS_NOTICE, "Background AOF rewrite successful");
+ 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*)(long)oldfd,NULL,NULL);
}
cleanup:
- sdsfree(server.bgrewritebuf);
- server.bgrewritebuf = sdsempty();
- aofRemoveTempFile(server.bgrewritechildpid);
- server.bgrewritechildpid = -1;
+ aofRewriteBufferReset();
+ aofRemoveTempFile(server.aof_child_pid);
+ server.aof_child_pid = -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;
}
projects / redis.git / blobdiff