]> git.saurik.com Git - redis.git/blame_incremental - src/aof.c
rewrite INCRBYFLOAT as SETs for AOF/replication
[redis.git] / src / aof.c
... / ...
CommitLineData
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
13void aofUpdateCurrentSize(void);
14
15void 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. */
21void 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. */
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);
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. */
79void 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
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
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. */
191sds 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
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
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. */
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;
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
293void 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. */
304int 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
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:
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. */
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
425/* Write a sequence of commands able to fully rebuild the dataset into
426 * "filename". Used both by REWRITEAOF and BGREWRITEAOF. */
427int rewriteAppendOnlyFile(char *filename) {
428 dictIterator *di = NULL;
429 dictEntry *de;
430 rio aof;
431 FILE *fp;
432 char tmpfile[256];
433 int j;
434 time_t now = time(NULL);
435
436 /* Note that we have to use a different temp name here compared to the
437 * one used by rewriteAppendOnlyFileBackground() function. */
438 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
439 fp = fopen(tmpfile,"w");
440 if (!fp) {
441 redisLog(REDIS_WARNING, "Failed rewriting the append only file: %s", strerror(errno));
442 return REDIS_ERR;
443 }
444
445 rioInitWithFile(&aof,fp);
446 for (j = 0; j < server.dbnum; j++) {
447 char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
448 redisDb *db = server.db+j;
449 dict *d = db->dict;
450 if (dictSize(d) == 0) continue;
451 di = dictGetSafeIterator(d);
452 if (!di) {
453 fclose(fp);
454 return REDIS_ERR;
455 }
456
457 /* SELECT the new DB */
458 if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
459 if (rioWriteBulkLongLong(&aof,j) == 0) goto werr;
460
461 /* Iterate this DB writing every entry */
462 while((de = dictNext(di)) != NULL) {
463 sds keystr;
464 robj key, *o;
465 time_t expiretime;
466
467 keystr = dictGetKey(de);
468 o = dictGetVal(de);
469 initStaticStringObject(key,keystr);
470
471 expiretime = getExpire(db,&key);
472
473 /* Save the key and associated value */
474 if (o->type == REDIS_STRING) {
475 /* Emit a SET command */
476 char cmd[]="*3\r\n$3\r\nSET\r\n";
477 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
478 /* Key and value */
479 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
480 if (rioWriteBulkObject(&aof,o) == 0) goto werr;
481 } else if (o->type == REDIS_LIST) {
482 /* Emit the RPUSHes needed to rebuild the list */
483 char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
484 if (o->encoding == REDIS_ENCODING_ZIPLIST) {
485 unsigned char *zl = o->ptr;
486 unsigned char *p = ziplistIndex(zl,0);
487 unsigned char *vstr;
488 unsigned int vlen;
489 long long vlong;
490
491 while(ziplistGet(p,&vstr,&vlen,&vlong)) {
492 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
493 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
494 if (vstr) {
495 if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0)
496 goto werr;
497 } else {
498 if (rioWriteBulkLongLong(&aof,vlong) == 0)
499 goto werr;
500 }
501 p = ziplistNext(zl,p);
502 }
503 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
504 list *list = o->ptr;
505 listNode *ln;
506 listIter li;
507
508 listRewind(list,&li);
509 while((ln = listNext(&li))) {
510 robj *eleobj = listNodeValue(ln);
511
512 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
513 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
514 if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr;
515 }
516 } else {
517 redisPanic("Unknown list encoding");
518 }
519 } else if (o->type == REDIS_SET) {
520 char cmd[]="*3\r\n$4\r\nSADD\r\n";
521
522 /* Emit the SADDs needed to rebuild the set */
523 if (o->encoding == REDIS_ENCODING_INTSET) {
524 int ii = 0;
525 int64_t llval;
526 while(intsetGet(o->ptr,ii++,&llval)) {
527 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
528 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
529 if (rioWriteBulkLongLong(&aof,llval) == 0) goto werr;
530 }
531 } else if (o->encoding == REDIS_ENCODING_HT) {
532 dictIterator *di = dictGetIterator(o->ptr);
533 dictEntry *de;
534 while((de = dictNext(di)) != NULL) {
535 robj *eleobj = dictGetKey(de);
536 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
537 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
538 if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr;
539 }
540 dictReleaseIterator(di);
541 } else {
542 redisPanic("Unknown set encoding");
543 }
544 } else if (o->type == REDIS_ZSET) {
545 /* Emit the ZADDs needed to rebuild the sorted set */
546 char cmd[]="*4\r\n$4\r\nZADD\r\n";
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 (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
566 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
567 if (rioWriteBulkDouble(&aof,score) == 0) goto werr;
568 if (vstr != NULL) {
569 if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0)
570 goto werr;
571 } else {
572 if (rioWriteBulkLongLong(&aof,vll) == 0)
573 goto werr;
574 }
575 zzlNext(zl,&eptr,&sptr);
576 }
577 } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
578 zset *zs = o->ptr;
579 dictIterator *di = dictGetIterator(zs->dict);
580 dictEntry *de;
581
582 while((de = dictNext(di)) != NULL) {
583 robj *eleobj = dictGetKey(de);
584 double *score = dictGetVal(de);
585
586 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
587 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
588 if (rioWriteBulkDouble(&aof,*score) == 0) goto werr;
589 if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr;
590 }
591 dictReleaseIterator(di);
592 } else {
593 redisPanic("Unknown sorted set encoding");
594 }
595 } else if (o->type == REDIS_HASH) {
596 char cmd[]="*4\r\n$4\r\nHSET\r\n";
597
598 /* Emit the HSETs needed to rebuild the hash */
599 if (o->encoding == REDIS_ENCODING_ZIPMAP) {
600 unsigned char *p = zipmapRewind(o->ptr);
601 unsigned char *field, *val;
602 unsigned int flen, vlen;
603
604 while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
605 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
606 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
607 if (rioWriteBulkString(&aof,(char*)field,flen) == 0)
608 goto werr;
609 if (rioWriteBulkString(&aof,(char*)val,vlen) == 0)
610 goto werr;
611 }
612 } else {
613 dictIterator *di = dictGetIterator(o->ptr);
614 dictEntry *de;
615
616 while((de = dictNext(di)) != NULL) {
617 robj *field = dictGetKey(de);
618 robj *val = dictGetVal(de);
619
620 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
621 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
622 if (rioWriteBulkObject(&aof,field) == 0) goto werr;
623 if (rioWriteBulkObject(&aof,val) == 0) goto werr;
624 }
625 dictReleaseIterator(di);
626 }
627 } else {
628 redisPanic("Unknown object type");
629 }
630 /* Save the expire time */
631 if (expiretime != -1) {
632 char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
633 /* If this key is already expired skip it */
634 if (expiretime < now) continue;
635 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
636 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
637 if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;
638 }
639 }
640 dictReleaseIterator(di);
641 }
642
643 /* Make sure data will not remain on the OS's output buffers */
644 fflush(fp);
645 aof_fsync(fileno(fp));
646 fclose(fp);
647
648 /* Use RENAME to make sure the DB file is changed atomically only
649 * if the generate DB file is ok. */
650 if (rename(tmpfile,filename) == -1) {
651 redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
652 unlink(tmpfile);
653 return REDIS_ERR;
654 }
655 redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");
656 return REDIS_OK;
657
658werr:
659 fclose(fp);
660 unlink(tmpfile);
661 redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
662 if (di) dictReleaseIterator(di);
663 return REDIS_ERR;
664}
665
666/* This is how rewriting of the append only file in background works:
667 *
668 * 1) The user calls BGREWRITEAOF
669 * 2) Redis calls this function, that forks():
670 * 2a) the child rewrite the append only file in a temp file.
671 * 2b) the parent accumulates differences in server.bgrewritebuf.
672 * 3) When the child finished '2a' exists.
673 * 4) The parent will trap the exit code, if it's OK, will append the
674 * data accumulated into server.bgrewritebuf into the temp file, and
675 * finally will rename(2) the temp file in the actual file name.
676 * The the new file is reopened as the new append only file. Profit!
677 */
678int rewriteAppendOnlyFileBackground(void) {
679 pid_t childpid;
680 long long start;
681
682 if (server.bgrewritechildpid != -1) return REDIS_ERR;
683 start = ustime();
684 if ((childpid = fork()) == 0) {
685 char tmpfile[256];
686
687 /* Child */
688 if (server.ipfd > 0) close(server.ipfd);
689 if (server.sofd > 0) close(server.sofd);
690 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
691 if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
692 _exit(0);
693 } else {
694 _exit(1);
695 }
696 } else {
697 /* Parent */
698 server.stat_fork_time = ustime()-start;
699 if (childpid == -1) {
700 redisLog(REDIS_WARNING,
701 "Can't rewrite append only file in background: fork: %s",
702 strerror(errno));
703 return REDIS_ERR;
704 }
705 redisLog(REDIS_NOTICE,
706 "Background append only file rewriting started by pid %d",childpid);
707 server.aofrewrite_scheduled = 0;
708 server.bgrewritechildpid = childpid;
709 updateDictResizePolicy();
710 /* We set appendseldb to -1 in order to force the next call to the
711 * feedAppendOnlyFile() to issue a SELECT command, so the differences
712 * accumulated by the parent into server.bgrewritebuf will start
713 * with a SELECT statement and it will be safe to merge. */
714 server.appendseldb = -1;
715 return REDIS_OK;
716 }
717 return REDIS_OK; /* unreached */
718}
719
720void bgrewriteaofCommand(redisClient *c) {
721 if (server.bgrewritechildpid != -1) {
722 addReplyError(c,"Background append only file rewriting already in progress");
723 } else if (server.bgsavechildpid != -1) {
724 server.aofrewrite_scheduled = 1;
725 addReplyStatus(c,"Background append only file rewriting scheduled");
726 } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) {
727 addReplyStatus(c,"Background append only file rewriting started");
728 } else {
729 addReply(c,shared.err);
730 }
731}
732
733void aofRemoveTempFile(pid_t childpid) {
734 char tmpfile[256];
735
736 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
737 unlink(tmpfile);
738}
739
740/* Update the server.appendonly_current_size filed explicitly using stat(2)
741 * to check the size of the file. This is useful after a rewrite or after
742 * a restart, normally the size is updated just adding the write length
743 * to the current lenght, that is much faster. */
744void aofUpdateCurrentSize(void) {
745 struct redis_stat sb;
746
747 if (redis_fstat(server.appendfd,&sb) == -1) {
748 redisLog(REDIS_WARNING,"Unable to check the AOF length: %s",
749 strerror(errno));
750 } else {
751 server.appendonly_current_size = sb.st_size;
752 }
753}
754
755/* A background append only file rewriting (BGREWRITEAOF) terminated its work.
756 * Handle this. */
757void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
758 if (!bysignal && exitcode == 0) {
759 int newfd, oldfd;
760 int nwritten;
761 char tmpfile[256];
762 long long now = ustime();
763
764 redisLog(REDIS_NOTICE,
765 "Background AOF rewrite terminated with success");
766
767 /* Flush the differences accumulated by the parent to the
768 * rewritten AOF. */
769 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
770 (int)server.bgrewritechildpid);
771 newfd = open(tmpfile,O_WRONLY|O_APPEND);
772 if (newfd == -1) {
773 redisLog(REDIS_WARNING,
774 "Unable to open the temporary AOF produced by the child: %s", strerror(errno));
775 goto cleanup;
776 }
777
778 nwritten = write(newfd,server.bgrewritebuf,sdslen(server.bgrewritebuf));
779 if (nwritten != (signed)sdslen(server.bgrewritebuf)) {
780 if (nwritten == -1) {
781 redisLog(REDIS_WARNING,
782 "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
783 } else {
784 redisLog(REDIS_WARNING,
785 "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
786 }
787 close(newfd);
788 goto cleanup;
789 }
790
791 redisLog(REDIS_NOTICE,
792 "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten);
793
794 /* The only remaining thing to do is to rename the temporary file to
795 * the configured file and switch the file descriptor used to do AOF
796 * writes. We don't want close(2) or rename(2) calls to block the
797 * server on old file deletion.
798 *
799 * There are two possible scenarios:
800 *
801 * 1) AOF is DISABLED and this was a one time rewrite. The temporary
802 * file will be renamed to the configured file. When this file already
803 * exists, it will be unlinked, which may block the server.
804 *
805 * 2) AOF is ENABLED and the rewritten AOF will immediately start
806 * receiving writes. After the temporary file is renamed to the
807 * configured file, the original AOF file descriptor will be closed.
808 * Since this will be the last reference to that file, closing it
809 * causes the underlying file to be unlinked, which may block the
810 * server.
811 *
812 * To mitigate the blocking effect of the unlink operation (either
813 * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we
814 * use a background thread to take care of this. First, we
815 * make scenario 1 identical to scenario 2 by opening the target file
816 * when it exists. The unlink operation after the rename(2) will then
817 * be executed upon calling close(2) for its descriptor. Everything to
818 * guarantee atomicity for this switch has already happened by then, so
819 * we don't care what the outcome or duration of that close operation
820 * is, as long as the file descriptor is released again. */
821 if (server.appendfd == -1) {
822 /* AOF disabled */
823
824 /* Don't care if this fails: oldfd will be -1 and we handle that.
825 * One notable case of -1 return is if the old file does
826 * not exist. */
827 oldfd = open(server.appendfilename,O_RDONLY|O_NONBLOCK);
828 } else {
829 /* AOF enabled */
830 oldfd = -1; /* We'll set this to the current AOF filedes later. */
831 }
832
833 /* Rename the temporary file. This will not unlink the target file if
834 * it exists, because we reference it with "oldfd". */
835 if (rename(tmpfile,server.appendfilename) == -1) {
836 redisLog(REDIS_WARNING,
837 "Error trying to rename the temporary AOF: %s", strerror(errno));
838 close(newfd);
839 if (oldfd != -1) close(oldfd);
840 goto cleanup;
841 }
842
843 if (server.appendfd == -1) {
844 /* AOF disabled, we don't need to set the AOF file descriptor
845 * to this new file, so we can close it. */
846 close(newfd);
847 } else {
848 /* AOF enabled, replace the old fd with the new one. */
849 oldfd = server.appendfd;
850 server.appendfd = newfd;
851 if (server.appendfsync == APPENDFSYNC_ALWAYS)
852 aof_fsync(newfd);
853 else if (server.appendfsync == APPENDFSYNC_EVERYSEC)
854 aof_background_fsync(newfd);
855 server.appendseldb = -1; /* Make sure SELECT is re-issued */
856 aofUpdateCurrentSize();
857 server.auto_aofrewrite_base_size = server.appendonly_current_size;
858
859 /* Clear regular AOF buffer since its contents was just written to
860 * the new AOF from the background rewrite buffer. */
861 sdsfree(server.aofbuf);
862 server.aofbuf = sdsempty();
863 }
864
865 redisLog(REDIS_NOTICE, "Background AOF rewrite successful");
866
867 /* Asynchronously close the overwritten AOF. */
868 if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL);
869
870 redisLog(REDIS_VERBOSE,
871 "Background AOF rewrite signal handler took %lldus", ustime()-now);
872 } else if (!bysignal && exitcode != 0) {
873 redisLog(REDIS_WARNING,
874 "Background AOF rewrite terminated with error");
875 } else {
876 redisLog(REDIS_WARNING,
877 "Background AOF rewrite terminated by signal %d", bysignal);
878 }
879
880cleanup:
881 sdsfree(server.bgrewritebuf);
882 server.bgrewritebuf = sdsempty();
883 aofRemoveTempFile(server.bgrewritechildpid);
884 server.bgrewritechildpid = -1;
885}