]> git.saurik.com Git - redis.git/blame - src/aof.c
Added a top-function comment to rioWriteHashIteratorCursor() to better specify what...
[redis.git] / src / aof.c
CommitLineData
e2641e09 1#include "redis.h"
986630af 2#include "bio.h"
f9c6f39b 3#include "rio.h"
e2641e09 4
5#include <signal.h>
6#include <fcntl.h>
7#include <sys/stat.h>
3688d7f3 8#include <sys/types.h>
9#include <sys/time.h>
10#include <sys/resource.h>
11#include <sys/wait.h>
e2641e09 12
b333e239 13void aofUpdateCurrentSize(void);
14
4b77700a 15void aof_background_fsync(int fd) {
9a35eb22 16 bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL);
4b77700a 17}
18
e2641e09 19/* Called when the user switches from "appendonly yes" to "appendonly no"
20 * at runtime using the CONFIG command. */
21void stopAppendOnly(void) {
e394114d 22 redisAssert(server.aof_state != REDIS_AOF_OFF);
db3c2a4f 23 flushAppendOnlyFile(1);
ff2145ad 24 aof_fsync(server.aof_fd);
25 close(server.aof_fd);
e2641e09 26
ff2145ad 27 server.aof_fd = -1;
28 server.aof_selected_db = -1;
e394114d 29 server.aof_state = REDIS_AOF_OFF;
e2641e09 30 /* rewrite operation in progress? kill it, wait child exit */
ff2145ad 31 if (server.aof_child_pid != -1) {
e2641e09 32 int statloc;
33
b941417c 34 redisLog(REDIS_NOTICE,"Killing running AOF rewrite child: %ld",
35 (long) server.aof_child_pid);
ff2145ad 36 if (kill(server.aof_child_pid,SIGKILL) != -1)
e2641e09 37 wait3(&statloc,0,NULL);
38 /* reset the buffer accumulating changes while the child saves */
ff2145ad 39 sdsfree(server.aof_rewrite_buf);
40 server.aof_rewrite_buf = sdsempty();
41 aofRemoveTempFile(server.aof_child_pid);
42 server.aof_child_pid = -1;
e2641e09 43 }
44}
45
46/* Called when the user switches from "appendonly no" to "appendonly yes"
47 * at runtime using the CONFIG command. */
48int startAppendOnly(void) {
ff2145ad 49 server.aof_last_fsync = time(NULL);
50 server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644);
e394114d 51 redisAssert(server.aof_state == REDIS_AOF_OFF);
ff2145ad 52 if (server.aof_fd == -1) {
e7a2e7c1 53 redisLog(REDIS_WARNING,"Redis needs to enable the AOF but can't open the append only file: %s",strerror(errno));
e2641e09 54 return REDIS_ERR;
55 }
56 if (rewriteAppendOnlyFileBackground() == REDIS_ERR) {
ff2145ad 57 close(server.aof_fd);
e7a2e7c1 58 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.");
e2641e09 59 return REDIS_ERR;
60 }
e7a2e7c1 61 /* We correctly switched on AOF, now wait for the rerwite to be complete
62 * in order to append data on disk. */
e394114d 63 server.aof_state = REDIS_AOF_WAIT_REWRITE;
e2641e09 64 return REDIS_OK;
65}
66
67/* Write the append only file buffer on disk.
68 *
69 * Since we are required to write the AOF before replying to the client,
70 * and the only way the client socket can get a write is entering when the
71 * the event loop, we accumulate all the AOF writes in a memory
72 * buffer and write it on disk using this function just before entering
db3c2a4f 73 * the event loop again.
74 *
75 * About the 'force' argument:
76 *
77 * When the fsync policy is set to 'everysec' we may delay the flush if there
78 * is still an fsync() going on in the background thread, since for instance
79 * on Linux write(2) will be blocked by the background fsync anyway.
80 * When this happens we remember that there is some aof buffer to be
81 * flushed ASAP, and will try to do that in the serverCron() function.
82 *
83 * However if force is set to 1 we'll write regardless of the background
84 * fsync. */
85void flushAppendOnlyFile(int force) {
e2641e09 86 ssize_t nwritten;
db3c2a4f 87 int sync_in_progress = 0;
e2641e09 88
ff2145ad 89 if (sdslen(server.aof_buf) == 0) return;
e2641e09 90
2c915bcf 91 if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
db3c2a4f 92 sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0;
93
2c915bcf 94 if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
db3c2a4f 95 /* With this append fsync policy we do background fsyncing.
96 * If the fsync is still in progress we can try to delay
97 * the write for a couple of seconds. */
98 if (sync_in_progress) {
99 if (server.aof_flush_postponed_start == 0) {
100 /* No previous write postponinig, remember that we are
101 * postponing the flush and return. */
102 server.aof_flush_postponed_start = server.unixtime;
103 return;
104 } else if (server.unixtime - server.aof_flush_postponed_start < 2) {
e7aec180 105 /* We were already waiting for fsync to finish, but for less
db3c2a4f 106 * than two seconds this is still ok. Postpone again. */
107 return;
108 }
109 /* Otherwise fall trough, and go write since we can't wait
110 * over two seconds. */
77ca5fcb 111 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.");
db3c2a4f 112 }
113 }
114 /* If you are following this code path, then we are going to write so
115 * set reset the postponed flush sentinel to zero. */
116 server.aof_flush_postponed_start = 0;
117
e2641e09 118 /* We want to perform a single write. This should be guaranteed atomic
119 * at least if the filesystem we are writing is a real physical one.
120 * While this will save us against the server being killed I don't think
121 * there is much to do about the whole server stopping for power problems
122 * or alike */
ff2145ad 123 nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
124 if (nwritten != (signed)sdslen(server.aof_buf)) {
e2641e09 125 /* Ooops, we are in troubles. The best thing to do for now is
126 * aborting instead of giving the illusion that everything is
127 * working as expected. */
a57225c2 128 if (nwritten == -1) {
e2641e09 129 redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno));
a57225c2 130 } else {
e51b79f3 131 redisLog(REDIS_WARNING,"Exiting on short write while writing to "
132 "the append-only file: %s (nwritten=%ld, "
133 "expected=%ld)",
134 strerror(errno),
135 (long)nwritten,
136 (long)sdslen(server.aof_buf));
a57225c2
PN
137 }
138 exit(1);
e2641e09 139 }
2c915bcf 140 server.aof_current_size += nwritten;
e2641e09 141
f990782f
PN
142 /* Re-use AOF buffer when it is small enough. The maximum comes from the
143 * arena size of 4k minus some overhead (but is otherwise arbitrary). */
ff2145ad 144 if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
145 sdsclear(server.aof_buf);
f990782f 146 } else {
ff2145ad 147 sdsfree(server.aof_buf);
148 server.aof_buf = sdsempty();
f990782f
PN
149 }
150
29732248
PN
151 /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
152 * children doing I/O in the background. */
2c915bcf 153 if (server.aof_no_fsync_on_rewrite &&
f48cd4b9 154 (server.aof_child_pid != -1 || server.rdb_child_pid != -1))
e2641e09 155 return;
29732248
PN
156
157 /* Perform the fsync if needed. */
2c915bcf 158 if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
e2641e09 159 /* aof_fsync is defined as fdatasync() for Linux in order to avoid
160 * flushing metadata. */
ff2145ad 161 aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */
162 server.aof_last_fsync = server.unixtime;
2c915bcf 163 } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
ff2145ad 164 server.unixtime > server.aof_last_fsync)) {
165 if (!sync_in_progress) aof_background_fsync(server.aof_fd);
166 server.aof_last_fsync = server.unixtime;
e2641e09 167 }
168}
169
d1ec6c8b
PN
170sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) {
171 char buf[32];
172 int len, j;
173 robj *o;
174
175 buf[0] = '*';
176 len = 1+ll2string(buf+1,sizeof(buf)-1,argc);
177 buf[len++] = '\r';
178 buf[len++] = '\n';
179 dst = sdscatlen(dst,buf,len);
180
e2641e09 181 for (j = 0; j < argc; j++) {
d1ec6c8b
PN
182 o = getDecodedObject(argv[j]);
183 buf[0] = '$';
184 len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr));
185 buf[len++] = '\r';
186 buf[len++] = '\n';
187 dst = sdscatlen(dst,buf,len);
188 dst = sdscatlen(dst,o->ptr,sdslen(o->ptr));
189 dst = sdscatlen(dst,"\r\n",2);
e2641e09 190 decrRefCount(o);
191 }
d1ec6c8b 192 return dst;
e2641e09 193}
194
12d293ca 195/* Create the sds representation of an PEXPIREAT command, using
196 * 'seconds' as time to live and 'cmd' to understand what command
197 * we are translating into a PEXPIREAT.
198 *
199 * This command is used in order to translate EXPIRE and PEXPIRE commands
200 * into PEXPIREAT command so that we retain precision in the append only
201 * file, and the time is always absolute and not relative. */
202sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) {
203 long long when;
e2641e09 204 robj *argv[3];
205
206 /* Make sure we can use strtol */
207 seconds = getDecodedObject(seconds);
12d293ca 208 when = strtoll(seconds->ptr,NULL,10);
209 /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */
210 if (cmd->proc == expireCommand || cmd->proc == setexCommand ||
211 cmd->proc == expireatCommand)
212 {
213 when *= 1000;
214 }
215 /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */
216 if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
217 cmd->proc == setexCommand || cmd->proc == psetexCommand)
218 {
219 when += mstime();
220 }
e2641e09 221 decrRefCount(seconds);
222
12d293ca 223 argv[0] = createStringObject("PEXPIREAT",9);
e2641e09 224 argv[1] = key;
12d293ca 225 argv[2] = createStringObjectFromLongLong(when);
226 buf = catAppendOnlyGenericCommand(buf, 3, argv);
e2641e09 227 decrRefCount(argv[0]);
228 decrRefCount(argv[2]);
229 return buf;
230}
231
232void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
087f4140 233 sds buf = sdsempty();
e2641e09 234 robj *tmpargv[3];
235
236 /* The DB this command was targetting is not the same as the last command
237 * we appendend. To issue a SELECT command is needed. */
ff2145ad 238 if (dictid != server.aof_selected_db) {
e2641e09 239 char seldb[64];
240
241 snprintf(seldb,sizeof(seldb),"%d",dictid);
242 buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
243 (unsigned long)strlen(seldb),seldb);
ff2145ad 244 server.aof_selected_db = dictid;
e2641e09 245 }
246
12d293ca 247 if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
248 cmd->proc == expireatCommand) {
249 /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */
250 buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
251 } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) {
252 /* Translate SETEX/PSETEX to SET and PEXPIREAT */
e2641e09 253 tmpargv[0] = createStringObject("SET",3);
254 tmpargv[1] = argv[1];
255 tmpargv[2] = argv[3];
256 buf = catAppendOnlyGenericCommand(buf,3,tmpargv);
257 decrRefCount(tmpargv[0]);
12d293ca 258 buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
e2641e09 259 } else {
12d293ca 260 /* All the other commands don't need translation or need the
261 * same translation already operated in the command vector
262 * for the replication itself. */
e2641e09 263 buf = catAppendOnlyGenericCommand(buf,argc,argv);
264 }
265
266 /* Append to the AOF buffer. This will be flushed on disk just before
267 * of re-entering the event loop, so before the client will get a
e394114d 268 * positive reply about the operation performed. */
269 if (server.aof_state == REDIS_AOF_ON)
ff2145ad 270 server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf));
e2641e09 271
272 /* If a background append only file rewriting is in progress we want to
273 * accumulate the differences between the child DB and the current one
274 * in a buffer, so that when the child process will do its work we
275 * can append the differences to the new append only file. */
ff2145ad 276 if (server.aof_child_pid != -1)
277 server.aof_rewrite_buf = sdscatlen(server.aof_rewrite_buf,buf,sdslen(buf));
e2641e09 278
279 sdsfree(buf);
280}
281
282/* In Redis commands are always executed in the context of a client, so in
283 * order to load the append only file we need to create a fake client. */
284struct redisClient *createFakeClient(void) {
285 struct redisClient *c = zmalloc(sizeof(*c));
286
287 selectDb(c,0);
288 c->fd = -1;
289 c->querybuf = sdsempty();
290 c->argc = 0;
291 c->argv = NULL;
2403fc9f 292 c->bufpos = 0;
e2641e09 293 c->flags = 0;
294 /* We set the fake client as a slave waiting for the synchronization
295 * so that Redis will not try to send replies to this client. */
296 c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
297 c->reply = listCreate();
3853c168 298 c->reply_bytes = 0;
7eac2a75 299 c->obuf_soft_limit_reached_time = 0;
b67d2345 300 c->watched_keys = listCreate();
e2641e09 301 listSetFreeMethod(c->reply,decrRefCount);
302 listSetDupMethod(c->reply,dupClientReplyValue);
303 initClientMultiState(c);
304 return c;
305}
306
307void freeFakeClient(struct redisClient *c) {
308 sdsfree(c->querybuf);
309 listRelease(c->reply);
b67d2345 310 listRelease(c->watched_keys);
e2641e09 311 freeClientMultiState(c);
312 zfree(c);
313}
314
315/* Replay the append log file. On error REDIS_OK is returned. On non fatal
316 * error (the append only file is zero-length) REDIS_ERR is returned. On
317 * fatal error an error message is logged and the program exists. */
318int loadAppendOnlyFile(char *filename) {
319 struct redisClient *fakeClient;
320 FILE *fp = fopen(filename,"r");
321 struct redis_stat sb;
e394114d 322 int old_aof_state = server.aof_state;
97e7f8ae 323 long loops = 0;
e2641e09 324
4aec2ec8 325 if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
2c915bcf 326 server.aof_current_size = 0;
4aec2ec8 327 fclose(fp);
e2641e09 328 return REDIS_ERR;
4aec2ec8 329 }
e2641e09 330
331 if (fp == NULL) {
332 redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno));
333 exit(1);
334 }
335
336 /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
337 * to the same file we're about to read. */
e394114d 338 server.aof_state = REDIS_AOF_OFF;
e2641e09 339
340 fakeClient = createFakeClient();
97e7f8ae 341 startLoading(fp);
342
e2641e09 343 while(1) {
344 int argc, j;
345 unsigned long len;
346 robj **argv;
347 char buf[128];
348 sds argsds;
349 struct redisCommand *cmd;
e2641e09 350
97e7f8ae 351 /* Serve the clients from time to time */
352 if (!(loops++ % 1000)) {
353 loadingProgress(ftello(fp));
354 aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
355 }
356
e2641e09 357 if (fgets(buf,sizeof(buf),fp) == NULL) {
358 if (feof(fp))
359 break;
360 else
361 goto readerr;
362 }
363 if (buf[0] != '*') goto fmterr;
364 argc = atoi(buf+1);
be6f6395
KM
365 if (argc < 1) goto fmterr;
366
e2641e09 367 argv = zmalloc(sizeof(robj*)*argc);
368 for (j = 0; j < argc; j++) {
369 if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr;
370 if (buf[0] != '$') goto fmterr;
371 len = strtol(buf+1,NULL,10);
372 argsds = sdsnewlen(NULL,len);
373 if (len && fread(argsds,len,1,fp) == 0) goto fmterr;
374 argv[j] = createObject(REDIS_STRING,argsds);
375 if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */
376 }
377
378 /* Command lookup */
379 cmd = lookupCommand(argv[0]->ptr);
380 if (!cmd) {
381 redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", argv[0]->ptr);
382 exit(1);
383 }
e2641e09 384 /* Run the command in the context of a fake client */
385 fakeClient->argc = argc;
386 fakeClient->argv = argv;
387 cmd->proc(fakeClient);
57b07380
PN
388
389 /* The fake client should not have a reply */
390 redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
ef67a2fc 391 /* The fake client should never get blocked */
392 redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0);
57b07380 393
45b0f6fb
PN
394 /* Clean up. Command code may have changed argv/argc so we use the
395 * argv/argc of the client instead of the local variables. */
396 for (j = 0; j < fakeClient->argc; j++)
397 decrRefCount(fakeClient->argv[j]);
398 zfree(fakeClient->argv);
e2641e09 399 }
400
401 /* This point can only be reached when EOF is reached without errors.
402 * If the client is in the middle of a MULTI/EXEC, log error and quit. */
403 if (fakeClient->flags & REDIS_MULTI) goto readerr;
404
405 fclose(fp);
406 freeFakeClient(fakeClient);
e394114d 407 server.aof_state = old_aof_state;
97e7f8ae 408 stopLoading();
b333e239 409 aofUpdateCurrentSize();
2c915bcf 410 server.aof_rewrite_base_size = server.aof_current_size;
e2641e09 411 return REDIS_OK;
412
413readerr:
414 if (feof(fp)) {
415 redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file");
416 } else {
417 redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno));
418 }
419 exit(1);
420fmterr:
412e457c 421 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>");
e2641e09 422 exit(1);
423}
424
7271198c
PN
425/* Delegate writing an object to writing a bulk string or bulk long long.
426 * This is not placed in rio.c since that adds the redis.h dependency. */
427int rioWriteBulkObject(rio *r, robj *obj) {
428 /* Avoid using getDecodedObject to help copy-on-write (we are often
429 * in a child process when this function is called). */
430 if (obj->encoding == REDIS_ENCODING_INT) {
431 return rioWriteBulkLongLong(r,(long)obj->ptr);
432 } else if (obj->encoding == REDIS_ENCODING_RAW) {
433 return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr));
434 } else {
435 redisPanic("Unknown string encoding");
436 }
437}
438
5b250096 439/* Emit the commands needed to rebuild a list object.
440 * The function returns 0 on error, 1 on success. */
441int rewriteListObject(rio *r, robj *key, robj *o) {
442 long long count = 0, items = listTypeLength(o);
443
444 if (o->encoding == REDIS_ENCODING_ZIPLIST) {
445 unsigned char *zl = o->ptr;
446 unsigned char *p = ziplistIndex(zl,0);
447 unsigned char *vstr;
448 unsigned int vlen;
449 long long vlong;
450
451 while(ziplistGet(p,&vstr,&vlen,&vlong)) {
452 if (count == 0) {
2c915bcf 453 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
454 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
8d875ccb 455
5b250096 456 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
457 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
458 if (rioWriteBulkObject(r,key) == 0) return 0;
459 }
460 if (vstr) {
461 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
462 } else {
463 if (rioWriteBulkLongLong(r,vlong) == 0) return 0;
464 }
465 p = ziplistNext(zl,p);
2c915bcf 466 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
5b250096 467 items--;
468 }
469 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
470 list *list = o->ptr;
471 listNode *ln;
472 listIter li;
473
474 listRewind(list,&li);
475 while((ln = listNext(&li))) {
476 robj *eleobj = listNodeValue(ln);
477
38c06fa0 478 if (count == 0) {
2c915bcf 479 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
480 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
8d875ccb 481
38c06fa0 482 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
483 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
484 if (rioWriteBulkObject(r,key) == 0) return 0;
485 }
5b250096 486 if (rioWriteBulkObject(r,eleobj) == 0) return 0;
2c915bcf 487 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
38c06fa0 488 items--;
5b250096 489 }
490 } else {
491 redisPanic("Unknown list encoding");
492 }
493 return 1;
494}
495
8d875ccb 496/* Emit the commands needed to rebuild a set object.
497 * The function returns 0 on error, 1 on success. */
498int rewriteSetObject(rio *r, robj *key, robj *o) {
499 long long count = 0, items = setTypeSize(o);
500
501 if (o->encoding == REDIS_ENCODING_INTSET) {
502 int ii = 0;
503 int64_t llval;
504
505 while(intsetGet(o->ptr,ii++,&llval)) {
506 if (count == 0) {
2c915bcf 507 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
508 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
8d875ccb 509
510 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
511 if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
512 if (rioWriteBulkObject(r,key) == 0) return 0;
513 }
514 if (rioWriteBulkLongLong(r,llval) == 0) return 0;
2c915bcf 515 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
8d875ccb 516 items--;
517 }
518 } else if (o->encoding == REDIS_ENCODING_HT) {
519 dictIterator *di = dictGetIterator(o->ptr);
520 dictEntry *de;
521
522 while((de = dictNext(di)) != NULL) {
523 robj *eleobj = dictGetKey(de);
524 if (count == 0) {
2c915bcf 525 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
526 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
8d875ccb 527
528 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
529 if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
530 if (rioWriteBulkObject(r,key) == 0) return 0;
531 }
532 if (rioWriteBulkObject(r,eleobj) == 0) return 0;
2c915bcf 533 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
8d875ccb 534 items--;
535 }
536 dictReleaseIterator(di);
537 } else {
538 redisPanic("Unknown set encoding");
539 }
540 return 1;
541}
542
7df9b141 543/* Emit the commands needed to rebuild a sorted set object.
544 * The function returns 0 on error, 1 on success. */
545int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
546 long long count = 0, items = zsetLength(o);
547
548 if (o->encoding == REDIS_ENCODING_ZIPLIST) {
549 unsigned char *zl = o->ptr;
550 unsigned char *eptr, *sptr;
551 unsigned char *vstr;
552 unsigned int vlen;
553 long long vll;
554 double score;
555
556 eptr = ziplistIndex(zl,0);
557 redisAssert(eptr != NULL);
558 sptr = ziplistNext(zl,eptr);
559 redisAssert(sptr != NULL);
560
561 while (eptr != NULL) {
562 redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
563 score = zzlGetScore(sptr);
564
565 if (count == 0) {
2c915bcf 566 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
567 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
7df9b141 568
569 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
570 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
571 if (rioWriteBulkObject(r,key) == 0) return 0;
572 }
573 if (rioWriteBulkDouble(r,score) == 0) return 0;
574 if (vstr != NULL) {
575 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
576 } else {
577 if (rioWriteBulkLongLong(r,vll) == 0) return 0;
578 }
579 zzlNext(zl,&eptr,&sptr);
2c915bcf 580 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
7df9b141 581 items--;
582 }
583 } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
584 zset *zs = o->ptr;
585 dictIterator *di = dictGetIterator(zs->dict);
586 dictEntry *de;
587
588 while((de = dictNext(di)) != NULL) {
589 robj *eleobj = dictGetKey(de);
590 double *score = dictGetVal(de);
591
592 if (count == 0) {
2c915bcf 593 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
594 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
7df9b141 595
596 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
597 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
598 if (rioWriteBulkObject(r,key) == 0) return 0;
599 }
600 if (rioWriteBulkDouble(r,*score) == 0) return 0;
601 if (rioWriteBulkObject(r,eleobj) == 0) return 0;
2c915bcf 602 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
7df9b141 603 items--;
604 }
605 dictReleaseIterator(di);
606 } else {
607 redisPanic("Unknown sorted zset encoding");
608 }
609 return 1;
610}
611
addc0327 612/* Write either the key or the value of the currently selected item of an hash.
613 * The 'hi' argument passes a valid Redis hash iterator.
614 * The 'what' filed specifies if to write a key or a value and can be
615 * either REDIS_HASH_KEY or REDIS_HASH_VALUE.
616 *
617 * The function returns 0 on error, non-zero on success. */
ebd85e9a
PN
618static int rioWriteHashIteratorCursor(rio *r, hashTypeIterator *hi, int what) {
619 if (hi->encoding == REDIS_ENCODING_ZIPLIST) {
620 unsigned char *vstr = NULL;
621 unsigned int vlen = UINT_MAX;
622 long long vll = LLONG_MAX;
623
624 hashTypeCurrentFromZiplist(hi, what, &vstr, &vlen, &vll);
625 if (vstr) {
626 return rioWriteBulkString(r, (char*)vstr, vlen);
627 } else {
628 return rioWriteBulkLongLong(r, vll);
629 }
630
631 } else if (hi->encoding == REDIS_ENCODING_HT) {
632 robj *value;
633
634 hashTypeCurrentFromHashTable(hi, what, &value);
635 return rioWriteBulkObject(r, value);
636 }
637
638 redisPanic("Unknown hash encoding");
639 return 0;
640}
641
54ecc0e7 642/* Emit the commands needed to rebuild a hash object.
643 * The function returns 0 on error, 1 on success. */
644int rewriteHashObject(rio *r, robj *key, robj *o) {
ebd85e9a 645 hashTypeIterator *hi;
54ecc0e7 646 long long count = 0, items = hashTypeLength(o);
647
ebd85e9a
PN
648 hi = hashTypeInitIterator(o);
649 while (hashTypeNext(hi) != REDIS_ERR) {
650 if (count == 0) {
651 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
652 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
54ecc0e7 653
ebd85e9a
PN
654 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
655 if (rioWriteBulkString(r,"HMSET",5) == 0) return 0;
656 if (rioWriteBulkObject(r,key) == 0) return 0;
54ecc0e7 657 }
54ecc0e7 658
ebd85e9a
PN
659 if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_KEY) == 0) return 0;
660 if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_VALUE) == 0) return 0;
661 if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
662 items--;
663 }
54ecc0e7 664
ebd85e9a 665 hashTypeReleaseIterator(hi);
54ecc0e7 666
54ecc0e7 667 return 1;
668}
669
e2641e09 670/* Write a sequence of commands able to fully rebuild the dataset into
5b250096 671 * "filename". Used both by REWRITEAOF and BGREWRITEAOF.
672 *
673 * In order to minimize the number of commands needed in the rewritten
674 * log Redis uses variadic commands when possible, such as RPUSH, SADD
2c915bcf 675 * and ZADD. However at max REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time
5b250096 676 * are inserted using a single command. */
e2641e09 677int rewriteAppendOnlyFile(char *filename) {
678 dictIterator *di = NULL;
679 dictEntry *de;
7271198c 680 rio aof;
e2641e09 681 FILE *fp;
682 char tmpfile[256];
683 int j;
4be855e7 684 long long now = mstime();
e2641e09 685
686 /* Note that we have to use a different temp name here compared to the
687 * one used by rewriteAppendOnlyFileBackground() function. */
688 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
689 fp = fopen(tmpfile,"w");
690 if (!fp) {
e51b79f3 691 redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
e2641e09 692 return REDIS_ERR;
693 }
7271198c 694
f96a8a80 695 rioInitWithFile(&aof,fp);
e2641e09 696 for (j = 0; j < server.dbnum; j++) {
697 char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
698 redisDb *db = server.db+j;
699 dict *d = db->dict;
700 if (dictSize(d) == 0) continue;
591f29e0 701 di = dictGetSafeIterator(d);
e2641e09 702 if (!di) {
703 fclose(fp);
704 return REDIS_ERR;
705 }
706
707 /* SELECT the new DB */
7271198c
PN
708 if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
709 if (rioWriteBulkLongLong(&aof,j) == 0) goto werr;
e2641e09 710
711 /* Iterate this DB writing every entry */
712 while((de = dictNext(di)) != NULL) {
6901fe77 713 sds keystr;
e2641e09 714 robj key, *o;
4be855e7 715 long long expiretime;
e2641e09 716
c0ba9ebe 717 keystr = dictGetKey(de);
718 o = dictGetVal(de);
e2641e09 719 initStaticStringObject(key,keystr);
16d77878 720
e2641e09 721 expiretime = getExpire(db,&key);
722
723 /* Save the key and associated value */
724 if (o->type == REDIS_STRING) {
725 /* Emit a SET command */
726 char cmd[]="*3\r\n$3\r\nSET\r\n";
7271198c 727 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
e2641e09 728 /* Key and value */
7271198c
PN
729 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
730 if (rioWriteBulkObject(&aof,o) == 0) goto werr;
e2641e09 731 } else if (o->type == REDIS_LIST) {
5b250096 732 if (rewriteListObject(&aof,&key,o) == 0) goto werr;
e2641e09 733 } else if (o->type == REDIS_SET) {
8d875ccb 734 if (rewriteSetObject(&aof,&key,o) == 0) goto werr;
e2641e09 735 } else if (o->type == REDIS_ZSET) {
7df9b141 736 if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
e2641e09 737 } else if (o->type == REDIS_HASH) {
54ecc0e7 738 if (rewriteHashObject(&aof,&key,o) == 0) goto werr;
e2641e09 739 } else {
740 redisPanic("Unknown object type");
741 }
742 /* Save the expire time */
743 if (expiretime != -1) {
12d293ca 744 char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
e2641e09 745 /* If this key is already expired skip it */
746 if (expiretime < now) continue;
7271198c
PN
747 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
748 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
b0b74486 749 if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;
e2641e09 750 }
e2641e09 751 }
752 dictReleaseIterator(di);
753 }
754
755 /* Make sure data will not remain on the OS's output buffers */
756 fflush(fp);
757 aof_fsync(fileno(fp));
758 fclose(fp);
759
760 /* Use RENAME to make sure the DB file is changed atomically only
761 * if the generate DB file is ok. */
762 if (rename(tmpfile,filename) == -1) {
763 redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
764 unlink(tmpfile);
765 return REDIS_ERR;
766 }
767 redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");
768 return REDIS_OK;
769
770werr:
771 fclose(fp);
772 unlink(tmpfile);
773 redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
774 if (di) dictReleaseIterator(di);
775 return REDIS_ERR;
776}
777
778/* This is how rewriting of the append only file in background works:
779 *
780 * 1) The user calls BGREWRITEAOF
781 * 2) Redis calls this function, that forks():
782 * 2a) the child rewrite the append only file in a temp file.
ff2145ad 783 * 2b) the parent accumulates differences in server.aof_rewrite_buf.
e2641e09 784 * 3) When the child finished '2a' exists.
785 * 4) The parent will trap the exit code, if it's OK, will append the
ff2145ad 786 * data accumulated into server.aof_rewrite_buf into the temp file, and
e2641e09 787 * finally will rename(2) the temp file in the actual file name.
788 * The the new file is reopened as the new append only file. Profit!
789 */
790int rewriteAppendOnlyFileBackground(void) {
791 pid_t childpid;
615e414c 792 long long start;
e2641e09 793
ff2145ad 794 if (server.aof_child_pid != -1) return REDIS_ERR;
615e414c 795 start = ustime();
e2641e09 796 if ((childpid = fork()) == 0) {
e2641e09 797 char tmpfile[256];
798
615e414c 799 /* Child */
a5639e7d
PN
800 if (server.ipfd > 0) close(server.ipfd);
801 if (server.sofd > 0) close(server.sofd);
e2641e09 802 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
803 if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
804 _exit(0);
805 } else {
806 _exit(1);
807 }
808 } else {
809 /* Parent */
615e414c 810 server.stat_fork_time = ustime()-start;
e2641e09 811 if (childpid == -1) {
812 redisLog(REDIS_WARNING,
813 "Can't rewrite append only file in background: fork: %s",
814 strerror(errno));
815 return REDIS_ERR;
816 }
817 redisLog(REDIS_NOTICE,
818 "Background append only file rewriting started by pid %d",childpid);
2c915bcf 819 server.aof_rewrite_scheduled = 0;
ff2145ad 820 server.aof_child_pid = childpid;
e2641e09 821 updateDictResizePolicy();
822 /* We set appendseldb to -1 in order to force the next call to the
823 * feedAppendOnlyFile() to issue a SELECT command, so the differences
ff2145ad 824 * accumulated by the parent into server.aof_rewrite_buf will start
e2641e09 825 * with a SELECT statement and it will be safe to merge. */
ff2145ad 826 server.aof_selected_db = -1;
e2641e09 827 return REDIS_OK;
828 }
829 return REDIS_OK; /* unreached */
830}
831
832void bgrewriteaofCommand(redisClient *c) {
ff2145ad 833 if (server.aof_child_pid != -1) {
3ab20376 834 addReplyError(c,"Background append only file rewriting already in progress");
f48cd4b9 835 } else if (server.rdb_child_pid != -1) {
2c915bcf 836 server.aof_rewrite_scheduled = 1;
9e40bce3 837 addReplyStatus(c,"Background append only file rewriting scheduled");
b333e239 838 } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) {
3ab20376 839 addReplyStatus(c,"Background append only file rewriting started");
e2641e09 840 } else {
841 addReply(c,shared.err);
842 }
843}
844
845void aofRemoveTempFile(pid_t childpid) {
846 char tmpfile[256];
847
848 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
849 unlink(tmpfile);
850}
851
2c915bcf 852/* Update the server.aof_current_size filed explicitly using stat(2)
b333e239 853 * to check the size of the file. This is useful after a rewrite or after
854 * a restart, normally the size is updated just adding the write length
2f0f0d95 855 * to the current length, that is much faster. */
b333e239 856void aofUpdateCurrentSize(void) {
857 struct redis_stat sb;
858
ff2145ad 859 if (redis_fstat(server.aof_fd,&sb) == -1) {
e51b79f3 860 redisLog(REDIS_WARNING,"Unable to obtain the AOF file length. stat: %s",
b333e239 861 strerror(errno));
862 } else {
2c915bcf 863 server.aof_current_size = sb.st_size;
b333e239 864 }
865}
866
e2641e09 867/* A background append only file rewriting (BGREWRITEAOF) terminated its work.
868 * Handle this. */
36c17a53 869void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
e2641e09 870 if (!bysignal && exitcode == 0) {
b454056d
PN
871 int newfd, oldfd;
872 int nwritten;
e2641e09 873 char tmpfile[256];
b454056d 874 long long now = ustime();
e2641e09 875
876 redisLog(REDIS_NOTICE,
b454056d
PN
877 "Background AOF rewrite terminated with success");
878
986630af 879 /* Flush the differences accumulated by the parent to the
880 * rewritten AOF. */
881 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
ff2145ad 882 (int)server.aof_child_pid);
b454056d
PN
883 newfd = open(tmpfile,O_WRONLY|O_APPEND);
884 if (newfd == -1) {
885 redisLog(REDIS_WARNING,
886 "Unable to open the temporary AOF produced by the child: %s", strerror(errno));
e2641e09 887 goto cleanup;
888 }
b454056d 889
ff2145ad 890 nwritten = write(newfd,server.aof_rewrite_buf,sdslen(server.aof_rewrite_buf));
891 if (nwritten != (signed)sdslen(server.aof_rewrite_buf)) {
b454056d
PN
892 if (nwritten == -1) {
893 redisLog(REDIS_WARNING,
894 "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
895 } else {
896 redisLog(REDIS_WARNING,
897 "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
898 }
899 close(newfd);
e2641e09 900 goto cleanup;
901 }
b454056d
PN
902
903 redisLog(REDIS_NOTICE,
904 "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten);
905
906 /* The only remaining thing to do is to rename the temporary file to
907 * the configured file and switch the file descriptor used to do AOF
986630af 908 * writes. We don't want close(2) or rename(2) calls to block the
909 * server on old file deletion.
910 *
911 * There are two possible scenarios:
b454056d
PN
912 *
913 * 1) AOF is DISABLED and this was a one time rewrite. The temporary
914 * file will be renamed to the configured file. When this file already
915 * exists, it will be unlinked, which may block the server.
916 *
917 * 2) AOF is ENABLED and the rewritten AOF will immediately start
918 * receiving writes. After the temporary file is renamed to the
919 * configured file, the original AOF file descriptor will be closed.
920 * Since this will be the last reference to that file, closing it
921 * causes the underlying file to be unlinked, which may block the
922 * server.
923 *
924 * To mitigate the blocking effect of the unlink operation (either
925 * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we
986630af 926 * use a background thread to take care of this. First, we
b454056d
PN
927 * make scenario 1 identical to scenario 2 by opening the target file
928 * when it exists. The unlink operation after the rename(2) will then
929 * be executed upon calling close(2) for its descriptor. Everything to
930 * guarantee atomicity for this switch has already happened by then, so
931 * we don't care what the outcome or duration of that close operation
932 * is, as long as the file descriptor is released again. */
ff2145ad 933 if (server.aof_fd == -1) {
b454056d 934 /* AOF disabled */
b454056d 935
986630af 936 /* Don't care if this fails: oldfd will be -1 and we handle that.
937 * One notable case of -1 return is if the old file does
938 * not exist. */
2c915bcf 939 oldfd = open(server.aof_filename,O_RDONLY|O_NONBLOCK);
b454056d
PN
940 } else {
941 /* AOF enabled */
986630af 942 oldfd = -1; /* We'll set this to the current AOF filedes later. */
b454056d
PN
943 }
944
945 /* Rename the temporary file. This will not unlink the target file if
946 * it exists, because we reference it with "oldfd". */
2c915bcf 947 if (rename(tmpfile,server.aof_filename) == -1) {
b454056d 948 redisLog(REDIS_WARNING,
e51b79f3 949 "Error trying to rename the temporary AOF file: %s", strerror(errno));
b454056d 950 close(newfd);
986630af 951 if (oldfd != -1) close(oldfd);
e2641e09 952 goto cleanup;
953 }
b454056d 954
ff2145ad 955 if (server.aof_fd == -1) {
986630af 956 /* AOF disabled, we don't need to set the AOF file descriptor
957 * to this new file, so we can close it. */
b454056d
PN
958 close(newfd);
959 } else {
986630af 960 /* AOF enabled, replace the old fd with the new one. */
ff2145ad 961 oldfd = server.aof_fd;
962 server.aof_fd = newfd;
2c915bcf 963 if (server.aof_fsync == AOF_FSYNC_ALWAYS)
4b77700a 964 aof_fsync(newfd);
2c915bcf 965 else if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
4b77700a 966 aof_background_fsync(newfd);
ff2145ad 967 server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
b333e239 968 aofUpdateCurrentSize();
2c915bcf 969 server.aof_rewrite_base_size = server.aof_current_size;
5f54a5e6
PN
970
971 /* Clear regular AOF buffer since its contents was just written to
972 * the new AOF from the background rewrite buffer. */
ff2145ad 973 sdsfree(server.aof_buf);
974 server.aof_buf = sdsempty();
e2641e09 975 }
b454056d 976
e51b79f3 977 redisLog(REDIS_NOTICE, "Background AOF rewrite finished successfully");
e394114d 978 /* Change state from WAIT_REWRITE to ON if needed */
979 if (server.aof_state == REDIS_AOF_WAIT_REWRITE)
980 server.aof_state = REDIS_AOF_ON;
b454056d
PN
981
982 /* Asynchronously close the overwritten AOF. */
50be9b97 983 if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL);
b454056d
PN
984
985 redisLog(REDIS_VERBOSE,
986 "Background AOF rewrite signal handler took %lldus", ustime()-now);
e2641e09 987 } else if (!bysignal && exitcode != 0) {
b454056d
PN
988 redisLog(REDIS_WARNING,
989 "Background AOF rewrite terminated with error");
e2641e09 990 } else {
991 redisLog(REDIS_WARNING,
b454056d 992 "Background AOF rewrite terminated by signal %d", bysignal);
e2641e09 993 }
b454056d 994
e2641e09 995cleanup:
ff2145ad 996 sdsfree(server.aof_rewrite_buf);
997 server.aof_rewrite_buf = sdsempty();
998 aofRemoveTempFile(server.aof_child_pid);
999 server.aof_child_pid = -1;
e394114d 1000 /* Schedule a new rewrite if we are waiting for it to switch the AOF ON. */
1001 if (server.aof_state == REDIS_AOF_WAIT_REWRITE)
2c915bcf 1002 server.aof_rewrite_scheduled = 1;
e2641e09 1003}