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