]> git.saurik.com Git - redis.git/blob - src/aof.c
e67b4c50e2a49f1f9e890421292e58f336a27a5e
[redis.git] / src / aof.c
1 #include "redis.h"
2 #include "bio.h"
3 #include "rio.h"
4
5 #include <signal.h>
6 #include <fcntl.h>
7 #include <sys/stat.h>
8 #include <sys/types.h>
9 #include <sys/time.h>
10 #include <sys/resource.h>
11 #include <sys/wait.h>
12
13 void aofUpdateCurrentSize(void);
14
15 void aof_background_fsync(int fd) {
16 bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL);
17 }
18
19 /* Called when the user switches from "appendonly yes" to "appendonly no"
20 * at runtime using the CONFIG command. */
21 void stopAppendOnly(void) {
22 flushAppendOnlyFile(1);
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 */
30 if (server.bgrewritechildpid != -1) {
31 int statloc;
32
33 if (kill(server.bgrewritechildpid,SIGKILL) != -1)
34 wait3(&statloc,0,NULL);
35 /* reset the buffer accumulating changes while the child saves */
36 sdsfree(server.bgrewritebuf);
37 server.bgrewritebuf = sdsempty();
38 server.bgrewritechildpid = -1;
39 }
40 }
41
42 /* Called when the user switches from "appendonly no" to "appendonly yes"
43 * at runtime using the CONFIG command. */
44 int 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);
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.");
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
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. */
79 void flushAppendOnlyFile(int force) {
80 ssize_t nwritten;
81 int sync_in_progress = 0;
82
83 if (sdslen(server.aofbuf) == 0) return;
84
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) {
99 /* We were already waiting for fsync to finish, but for less
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. */
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.");
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
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 */
117 nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf));
118 if (nwritten != (signed)sdslen(server.aofbuf)) {
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. */
122 if (nwritten == -1) {
123 redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno));
124 } else {
125 redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno));
126 }
127 exit(1);
128 }
129 server.appendonly_current_size += nwritten;
130
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
140 /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
141 * children doing I/O in the background. */
142 if (server.no_appendfsync_on_rewrite &&
143 (server.bgrewritechildpid != -1 || server.bgsavechildpid != -1))
144 return;
145
146 /* Perform the fsync if needed. */
147 if (server.appendfsync == APPENDFSYNC_ALWAYS) {
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 */
151 server.lastfsync = server.unixtime;
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;
156 }
157 }
158
159 sds 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
170 for (j = 0; j < argc; j++) {
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);
179 decrRefCount(o);
180 }
181 return dst;
182 }
183
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. */
191 sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) {
192 long long when;
193 robj *argv[3];
194
195 /* Make sure we can use strtol */
196 seconds = getDecodedObject(seconds);
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 }
210 decrRefCount(seconds);
211
212 argv[0] = createStringObject("PEXPIREAT",9);
213 argv[1] = key;
214 argv[2] = createStringObjectFromLongLong(when);
215 buf = catAppendOnlyGenericCommand(buf, 3, argv);
216 decrRefCount(argv[0]);
217 decrRefCount(argv[2]);
218 return buf;
219 }
220
221 void 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
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 */
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]);
247 buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
248 } else {
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. */
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. */
272 struct 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;
280 c->bufpos = 0;
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();
286 c->watched_keys = listCreate();
287 listSetFreeMethod(c->reply,decrRefCount);
288 listSetDupMethod(c->reply,dupClientReplyValue);
289 initClientMultiState(c);
290 return c;
291 }
292
293 void freeFakeClient(struct redisClient *c) {
294 sdsfree(c->querybuf);
295 listRelease(c->reply);
296 listRelease(c->watched_keys);
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. */
304 int loadAppendOnlyFile(char *filename) {
305 struct redisClient *fakeClient;
306 FILE *fp = fopen(filename,"r");
307 struct redis_stat sb;
308 int appendonly = server.appendonly;
309 long loops = 0;
310
311 if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
312 server.appendonly_current_size = 0;
313 fclose(fp);
314 return REDIS_ERR;
315 }
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();
327 startLoading(fp);
328
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;
336
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
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);
351 if (argc < 1) goto fmterr;
352
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 }
370 /* Run the command in the context of a fake client */
371 fakeClient->argc = argc;
372 fakeClient->argv = argv;
373 cmd->proc(fakeClient);
374
375 /* The fake client should not have a reply */
376 redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
377 /* The fake client should never get blocked */
378 redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0);
379
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);
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;
394 stopLoading();
395 aofUpdateCurrentSize();
396 server.auto_aofrewrite_base_size = server.appendonly_current_size;
397 return REDIS_OK;
398
399 readerr:
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);
406 fmterr:
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>");
408 exit(1);
409 }
410
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. */
413 int 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
425 /* Emit the commands needed to rebuild a list object.
426 * The function returns 0 on error, 1 on success. */
427 int 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
442 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
443 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
444 if (rioWriteBulkObject(r,key) == 0) return 0;
445 }
446 if (vstr) {
447 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
448 } else {
449 if (rioWriteBulkLongLong(r,vlong) == 0) return 0;
450 }
451 p = ziplistNext(zl,p);
452 if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
453 items--;
454 }
455 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
456 list *list = o->ptr;
457 listNode *ln;
458 listIter li;
459
460 listRewind(list,&li);
461 while((ln = listNext(&li))) {
462 robj *eleobj = listNodeValue(ln);
463
464 if (count == 0) {
465 int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
466 REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
467
468 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
469 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
470 if (rioWriteBulkObject(r,key) == 0) return 0;
471 }
472 if (rioWriteBulkObject(r,eleobj) == 0) return 0;
473 if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
474 items--;
475 }
476 } else {
477 redisPanic("Unknown list encoding");
478 }
479 return 1;
480 }
481
482 /* Emit the commands needed to rebuild a set object.
483 * The function returns 0 on error, 1 on success. */
484 int rewriteSetObject(rio *r, robj *key, robj *o) {
485 long long count = 0, items = setTypeSize(o);
486
487 if (o->encoding == REDIS_ENCODING_INTSET) {
488 int ii = 0;
489 int64_t llval;
490
491 while(intsetGet(o->ptr,ii++,&llval)) {
492 if (count == 0) {
493 int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
494 REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
495
496 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
497 if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
498 if (rioWriteBulkObject(r,key) == 0) return 0;
499 }
500 if (rioWriteBulkLongLong(r,llval) == 0) return 0;
501 if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
502 items--;
503 }
504 } else if (o->encoding == REDIS_ENCODING_HT) {
505 dictIterator *di = dictGetIterator(o->ptr);
506 dictEntry *de;
507
508 while((de = dictNext(di)) != NULL) {
509 robj *eleobj = dictGetKey(de);
510 if (count == 0) {
511 int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
512 REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
513
514 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
515 if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
516 if (rioWriteBulkObject(r,key) == 0) return 0;
517 }
518 if (rioWriteBulkObject(r,eleobj) == 0) return 0;
519 if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
520 items--;
521 }
522 dictReleaseIterator(di);
523 } else {
524 redisPanic("Unknown set encoding");
525 }
526 return 1;
527 }
528
529 /* Emit the commands needed to rebuild a sorted set object.
530 * The function returns 0 on error, 1 on success. */
531 int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
532 long long count = 0, items = zsetLength(o);
533
534 if (o->encoding == REDIS_ENCODING_ZIPLIST) {
535 unsigned char *zl = o->ptr;
536 unsigned char *eptr, *sptr;
537 unsigned char *vstr;
538 unsigned int vlen;
539 long long vll;
540 double score;
541
542 eptr = ziplistIndex(zl,0);
543 redisAssert(eptr != NULL);
544 sptr = ziplistNext(zl,eptr);
545 redisAssert(sptr != NULL);
546
547 while (eptr != NULL) {
548 redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
549 score = zzlGetScore(sptr);
550
551 if (count == 0) {
552 int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
553 REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
554
555 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
556 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
557 if (rioWriteBulkObject(r,key) == 0) return 0;
558 }
559 if (rioWriteBulkDouble(r,score) == 0) return 0;
560 if (vstr != NULL) {
561 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
562 } else {
563 if (rioWriteBulkLongLong(r,vll) == 0) return 0;
564 }
565 zzlNext(zl,&eptr,&sptr);
566 if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
567 items--;
568 }
569 } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
570 zset *zs = o->ptr;
571 dictIterator *di = dictGetIterator(zs->dict);
572 dictEntry *de;
573
574 while((de = dictNext(di)) != NULL) {
575 robj *eleobj = dictGetKey(de);
576 double *score = dictGetVal(de);
577
578 if (count == 0) {
579 int cmd_items = (items > REDIS_AOFREWRITE_ITEMS_PER_CMD) ?
580 REDIS_AOFREWRITE_ITEMS_PER_CMD : items;
581
582 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
583 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
584 if (rioWriteBulkObject(r,key) == 0) return 0;
585 }
586 if (rioWriteBulkDouble(r,*score) == 0) return 0;
587 if (rioWriteBulkObject(r,eleobj) == 0) return 0;
588 if (++count == REDIS_AOFREWRITE_ITEMS_PER_CMD) count = 0;
589 items--;
590 }
591 dictReleaseIterator(di);
592 } else {
593 redisPanic("Unknown sorted zset encoding");
594 }
595 return 1;
596 }
597
598 /* Write a sequence of commands able to fully rebuild the dataset into
599 * "filename". Used both by REWRITEAOF and BGREWRITEAOF.
600 *
601 * In order to minimize the number of commands needed in the rewritten
602 * log Redis uses variadic commands when possible, such as RPUSH, SADD
603 * and ZADD. However at max REDIS_AOFREWRITE_ITEMS_PER_CMD items per time
604 * are inserted using a single command. */
605 int rewriteAppendOnlyFile(char *filename) {
606 dictIterator *di = NULL;
607 dictEntry *de;
608 rio aof;
609 FILE *fp;
610 char tmpfile[256];
611 int j;
612 long long now = mstime();
613
614 /* Note that we have to use a different temp name here compared to the
615 * one used by rewriteAppendOnlyFileBackground() function. */
616 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
617 fp = fopen(tmpfile,"w");
618 if (!fp) {
619 redisLog(REDIS_WARNING, "Failed rewriting the append only file: %s", strerror(errno));
620 return REDIS_ERR;
621 }
622
623 rioInitWithFile(&aof,fp);
624 for (j = 0; j < server.dbnum; j++) {
625 char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
626 redisDb *db = server.db+j;
627 dict *d = db->dict;
628 if (dictSize(d) == 0) continue;
629 di = dictGetSafeIterator(d);
630 if (!di) {
631 fclose(fp);
632 return REDIS_ERR;
633 }
634
635 /* SELECT the new DB */
636 if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
637 if (rioWriteBulkLongLong(&aof,j) == 0) goto werr;
638
639 /* Iterate this DB writing every entry */
640 while((de = dictNext(di)) != NULL) {
641 sds keystr;
642 robj key, *o;
643 long long expiretime;
644
645 keystr = dictGetKey(de);
646 o = dictGetVal(de);
647 initStaticStringObject(key,keystr);
648
649 expiretime = getExpire(db,&key);
650
651 /* Save the key and associated value */
652 if (o->type == REDIS_STRING) {
653 /* Emit a SET command */
654 char cmd[]="*3\r\n$3\r\nSET\r\n";
655 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
656 /* Key and value */
657 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
658 if (rioWriteBulkObject(&aof,o) == 0) goto werr;
659 } else if (o->type == REDIS_LIST) {
660 if (rewriteListObject(&aof,&key,o) == 0) goto werr;
661 } else if (o->type == REDIS_SET) {
662 if (rewriteSetObject(&aof,&key,o) == 0) goto werr;
663 } else if (o->type == REDIS_ZSET) {
664 if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
665 } else if (o->type == REDIS_HASH) {
666 char cmd[]="*4\r\n$4\r\nHSET\r\n";
667
668 /* Emit the HSETs needed to rebuild the hash */
669 if (o->encoding == REDIS_ENCODING_ZIPMAP) {
670 unsigned char *p = zipmapRewind(o->ptr);
671 unsigned char *field, *val;
672 unsigned int flen, vlen;
673
674 while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
675 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
676 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
677 if (rioWriteBulkString(&aof,(char*)field,flen) == 0)
678 goto werr;
679 if (rioWriteBulkString(&aof,(char*)val,vlen) == 0)
680 goto werr;
681 }
682 } else {
683 dictIterator *di = dictGetIterator(o->ptr);
684 dictEntry *de;
685
686 while((de = dictNext(di)) != NULL) {
687 robj *field = dictGetKey(de);
688 robj *val = dictGetVal(de);
689
690 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
691 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
692 if (rioWriteBulkObject(&aof,field) == 0) goto werr;
693 if (rioWriteBulkObject(&aof,val) == 0) goto werr;
694 }
695 dictReleaseIterator(di);
696 }
697 } else {
698 redisPanic("Unknown object type");
699 }
700 /* Save the expire time */
701 if (expiretime != -1) {
702 char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
703 /* If this key is already expired skip it */
704 if (expiretime < now) continue;
705 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
706 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
707 if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;
708 }
709 }
710 dictReleaseIterator(di);
711 }
712
713 /* Make sure data will not remain on the OS's output buffers */
714 fflush(fp);
715 aof_fsync(fileno(fp));
716 fclose(fp);
717
718 /* Use RENAME to make sure the DB file is changed atomically only
719 * if the generate DB file is ok. */
720 if (rename(tmpfile,filename) == -1) {
721 redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
722 unlink(tmpfile);
723 return REDIS_ERR;
724 }
725 redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");
726 return REDIS_OK;
727
728 werr:
729 fclose(fp);
730 unlink(tmpfile);
731 redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
732 if (di) dictReleaseIterator(di);
733 return REDIS_ERR;
734 }
735
736 /* This is how rewriting of the append only file in background works:
737 *
738 * 1) The user calls BGREWRITEAOF
739 * 2) Redis calls this function, that forks():
740 * 2a) the child rewrite the append only file in a temp file.
741 * 2b) the parent accumulates differences in server.bgrewritebuf.
742 * 3) When the child finished '2a' exists.
743 * 4) The parent will trap the exit code, if it's OK, will append the
744 * data accumulated into server.bgrewritebuf into the temp file, and
745 * finally will rename(2) the temp file in the actual file name.
746 * The the new file is reopened as the new append only file. Profit!
747 */
748 int rewriteAppendOnlyFileBackground(void) {
749 pid_t childpid;
750 long long start;
751
752 if (server.bgrewritechildpid != -1) return REDIS_ERR;
753 start = ustime();
754 if ((childpid = fork()) == 0) {
755 char tmpfile[256];
756
757 /* Child */
758 if (server.ipfd > 0) close(server.ipfd);
759 if (server.sofd > 0) close(server.sofd);
760 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
761 if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
762 _exit(0);
763 } else {
764 _exit(1);
765 }
766 } else {
767 /* Parent */
768 server.stat_fork_time = ustime()-start;
769 if (childpid == -1) {
770 redisLog(REDIS_WARNING,
771 "Can't rewrite append only file in background: fork: %s",
772 strerror(errno));
773 return REDIS_ERR;
774 }
775 redisLog(REDIS_NOTICE,
776 "Background append only file rewriting started by pid %d",childpid);
777 server.aofrewrite_scheduled = 0;
778 server.bgrewritechildpid = childpid;
779 updateDictResizePolicy();
780 /* We set appendseldb to -1 in order to force the next call to the
781 * feedAppendOnlyFile() to issue a SELECT command, so the differences
782 * accumulated by the parent into server.bgrewritebuf will start
783 * with a SELECT statement and it will be safe to merge. */
784 server.appendseldb = -1;
785 return REDIS_OK;
786 }
787 return REDIS_OK; /* unreached */
788 }
789
790 void bgrewriteaofCommand(redisClient *c) {
791 if (server.bgrewritechildpid != -1) {
792 addReplyError(c,"Background append only file rewriting already in progress");
793 } else if (server.bgsavechildpid != -1) {
794 server.aofrewrite_scheduled = 1;
795 addReplyStatus(c,"Background append only file rewriting scheduled");
796 } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) {
797 addReplyStatus(c,"Background append only file rewriting started");
798 } else {
799 addReply(c,shared.err);
800 }
801 }
802
803 void aofRemoveTempFile(pid_t childpid) {
804 char tmpfile[256];
805
806 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
807 unlink(tmpfile);
808 }
809
810 /* Update the server.appendonly_current_size filed explicitly using stat(2)
811 * to check the size of the file. This is useful after a rewrite or after
812 * a restart, normally the size is updated just adding the write length
813 * to the current lenght, that is much faster. */
814 void aofUpdateCurrentSize(void) {
815 struct redis_stat sb;
816
817 if (redis_fstat(server.appendfd,&sb) == -1) {
818 redisLog(REDIS_WARNING,"Unable to check the AOF length: %s",
819 strerror(errno));
820 } else {
821 server.appendonly_current_size = sb.st_size;
822 }
823 }
824
825 /* A background append only file rewriting (BGREWRITEAOF) terminated its work.
826 * Handle this. */
827 void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
828 if (!bysignal && exitcode == 0) {
829 int newfd, oldfd;
830 int nwritten;
831 char tmpfile[256];
832 long long now = ustime();
833
834 redisLog(REDIS_NOTICE,
835 "Background AOF rewrite terminated with success");
836
837 /* Flush the differences accumulated by the parent to the
838 * rewritten AOF. */
839 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
840 (int)server.bgrewritechildpid);
841 newfd = open(tmpfile,O_WRONLY|O_APPEND);
842 if (newfd == -1) {
843 redisLog(REDIS_WARNING,
844 "Unable to open the temporary AOF produced by the child: %s", strerror(errno));
845 goto cleanup;
846 }
847
848 nwritten = write(newfd,server.bgrewritebuf,sdslen(server.bgrewritebuf));
849 if (nwritten != (signed)sdslen(server.bgrewritebuf)) {
850 if (nwritten == -1) {
851 redisLog(REDIS_WARNING,
852 "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
853 } else {
854 redisLog(REDIS_WARNING,
855 "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
856 }
857 close(newfd);
858 goto cleanup;
859 }
860
861 redisLog(REDIS_NOTICE,
862 "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten);
863
864 /* The only remaining thing to do is to rename the temporary file to
865 * the configured file and switch the file descriptor used to do AOF
866 * writes. We don't want close(2) or rename(2) calls to block the
867 * server on old file deletion.
868 *
869 * There are two possible scenarios:
870 *
871 * 1) AOF is DISABLED and this was a one time rewrite. The temporary
872 * file will be renamed to the configured file. When this file already
873 * exists, it will be unlinked, which may block the server.
874 *
875 * 2) AOF is ENABLED and the rewritten AOF will immediately start
876 * receiving writes. After the temporary file is renamed to the
877 * configured file, the original AOF file descriptor will be closed.
878 * Since this will be the last reference to that file, closing it
879 * causes the underlying file to be unlinked, which may block the
880 * server.
881 *
882 * To mitigate the blocking effect of the unlink operation (either
883 * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we
884 * use a background thread to take care of this. First, we
885 * make scenario 1 identical to scenario 2 by opening the target file
886 * when it exists. The unlink operation after the rename(2) will then
887 * be executed upon calling close(2) for its descriptor. Everything to
888 * guarantee atomicity for this switch has already happened by then, so
889 * we don't care what the outcome or duration of that close operation
890 * is, as long as the file descriptor is released again. */
891 if (server.appendfd == -1) {
892 /* AOF disabled */
893
894 /* Don't care if this fails: oldfd will be -1 and we handle that.
895 * One notable case of -1 return is if the old file does
896 * not exist. */
897 oldfd = open(server.appendfilename,O_RDONLY|O_NONBLOCK);
898 } else {
899 /* AOF enabled */
900 oldfd = -1; /* We'll set this to the current AOF filedes later. */
901 }
902
903 /* Rename the temporary file. This will not unlink the target file if
904 * it exists, because we reference it with "oldfd". */
905 if (rename(tmpfile,server.appendfilename) == -1) {
906 redisLog(REDIS_WARNING,
907 "Error trying to rename the temporary AOF: %s", strerror(errno));
908 close(newfd);
909 if (oldfd != -1) close(oldfd);
910 goto cleanup;
911 }
912
913 if (server.appendfd == -1) {
914 /* AOF disabled, we don't need to set the AOF file descriptor
915 * to this new file, so we can close it. */
916 close(newfd);
917 } else {
918 /* AOF enabled, replace the old fd with the new one. */
919 oldfd = server.appendfd;
920 server.appendfd = newfd;
921 if (server.appendfsync == APPENDFSYNC_ALWAYS)
922 aof_fsync(newfd);
923 else if (server.appendfsync == APPENDFSYNC_EVERYSEC)
924 aof_background_fsync(newfd);
925 server.appendseldb = -1; /* Make sure SELECT is re-issued */
926 aofUpdateCurrentSize();
927 server.auto_aofrewrite_base_size = server.appendonly_current_size;
928
929 /* Clear regular AOF buffer since its contents was just written to
930 * the new AOF from the background rewrite buffer. */
931 sdsfree(server.aofbuf);
932 server.aofbuf = sdsempty();
933 }
934
935 redisLog(REDIS_NOTICE, "Background AOF rewrite successful");
936
937 /* Asynchronously close the overwritten AOF. */
938 if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL);
939
940 redisLog(REDIS_VERBOSE,
941 "Background AOF rewrite signal handler took %lldus", ustime()-now);
942 } else if (!bysignal && exitcode != 0) {
943 redisLog(REDIS_WARNING,
944 "Background AOF rewrite terminated with error");
945 } else {
946 redisLog(REDIS_WARNING,
947 "Background AOF rewrite terminated by signal %d", bysignal);
948 }
949
950 cleanup:
951 sdsfree(server.bgrewritebuf);
952 server.bgrewritebuf = sdsempty();
953 aofRemoveTempFile(server.bgrewritechildpid);
954 server.bgrewritechildpid = -1;
955 }