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