aboutsummaryrefslogtreecommitdiff
path: root/examples/redis-unstable/src/aof.c
diff options
context:
space:
mode:
Diffstat (limited to 'examples/redis-unstable/src/aof.c')
-rw-r--r--examples/redis-unstable/src/aof.c2921
1 files changed, 2921 insertions, 0 deletions
diff --git a/examples/redis-unstable/src/aof.c b/examples/redis-unstable/src/aof.c
new file mode 100644
index 0000000..3ace670
--- /dev/null
+++ b/examples/redis-unstable/src/aof.c
@@ -0,0 +1,2921 @@
1/*
2 * Copyright (c) 2009-Present, Redis Ltd.
3 * All rights reserved.
4 *
5 * Licensed under your choice of (a) the Redis Source Available License 2.0
6 * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
7 * GNU Affero General Public License v3 (AGPLv3).
8 */
9
10#include "server.h"
11#include "bio.h"
12#include "rio.h"
13#include "functions.h"
14#include "cluster_asm.h"
15
16#include <signal.h>
17#include <fcntl.h>
18#include <sys/stat.h>
19#include <sys/types.h>
20#include <sys/time.h>
21#include <sys/resource.h>
22#include <sys/wait.h>
23#include <sys/param.h>
24
25void freeClientArgv(client *c);
26off_t getAppendOnlyFileSize(sds filename, int *status);
27off_t getBaseAndIncrAppendOnlyFilesSize(aofManifest *am, int *status);
28int getBaseAndIncrAppendOnlyFilesNum(aofManifest *am);
29int aofFileExist(char *filename);
30int rewriteAppendOnlyFile(char *filename);
31aofManifest *aofLoadManifestFromFile(sds am_filepath);
32void aofManifestFreeAndUpdate(aofManifest *am);
33void aof_background_fsync_and_close(int fd);
34
35/* When we call 'startAppendOnly', we will create a temp INCR AOF, and rename
36 * it to the real INCR AOF name when the AOFRW is done, so if want to know the
37 * accurate start offset of the INCR AOF, we need to record it when we create
38 * the temp INCR AOF. This variable is used to record the start offset, and
39 * set the start offset of the real INCR AOF when the AOFRW is done. */
40static long long tempIncAofStartReplOffset = 0;
41
42/* ----------------------------------------------------------------------------
43 * AOF Manifest file implementation.
44 *
45 * The following code implements the read/write logic of AOF manifest file, which
46 * is used to track and manage all AOF files.
47 *
48 * Append-only files consist of three types:
49 *
50 * BASE: Represents a Redis snapshot from the time of last AOF rewrite. The manifest
51 * file contains at most a single BASE file, which will always be the first file in the
52 * list.
53 *
54 * INCR: Represents all write commands executed by Redis following the last successful
55 * AOF rewrite. In some cases it is possible to have several ordered INCR files. For
56 * example:
57 * - During an on-going AOF rewrite
58 * - After an AOF rewrite was aborted/failed, and before the next one succeeded.
59 *
60 * HISTORY: After a successful rewrite, the previous BASE and INCR become HISTORY files.
61 * They will be automatically removed unless garbage collection is disabled.
62 *
63 * The following is a possible AOF manifest file content:
64 *
65 * file appendonly.aof.2.base.rdb seq 2 type b
66 * file appendonly.aof.1.incr.aof seq 1 type h
67 * file appendonly.aof.2.incr.aof seq 2 type h
68 * file appendonly.aof.3.incr.aof seq 3 type h
69 * file appendonly.aof.4.incr.aof seq 4 type i
70 * file appendonly.aof.5.incr.aof seq 5 type i
71 * ------------------------------------------------------------------------- */
72
73/* Naming rules. */
74#define BASE_FILE_SUFFIX ".base"
75#define INCR_FILE_SUFFIX ".incr"
76#define RDB_FORMAT_SUFFIX ".rdb"
77#define AOF_FORMAT_SUFFIX ".aof"
78#define MANIFEST_NAME_SUFFIX ".manifest"
79#define TEMP_FILE_NAME_PREFIX "temp-"
80
81/* AOF manifest key. */
82#define AOF_MANIFEST_KEY_FILE_NAME "file"
83#define AOF_MANIFEST_KEY_FILE_SEQ "seq"
84#define AOF_MANIFEST_KEY_FILE_TYPE "type"
85#define AOF_MANIFEST_KEY_FILE_STARTOFFSET "startoffset"
86#define AOF_MANIFEST_KEY_FILE_ENDOFFSET "endoffset"
87
88/* Create an empty aofInfo. */
89aofInfo *aofInfoCreate(void) {
90 aofInfo *ai = zcalloc(sizeof(aofInfo));
91 ai->start_offset = -1;
92 ai->end_offset = -1;
93 return ai;
94}
95
96/* Free the aofInfo structure (pointed to by ai) and its embedded file_name. */
97void aofInfoFree(aofInfo *ai) {
98 serverAssert(ai != NULL);
99 if (ai->file_name) sdsfree(ai->file_name);
100 zfree(ai);
101}
102
103/* Deep copy an aofInfo. */
104aofInfo *aofInfoDup(aofInfo *orig) {
105 serverAssert(orig != NULL);
106 aofInfo *ai = aofInfoCreate();
107 ai->file_name = sdsdup(orig->file_name);
108 ai->file_seq = orig->file_seq;
109 ai->file_type = orig->file_type;
110 ai->start_offset = orig->start_offset;
111 ai->end_offset = orig->end_offset;
112 return ai;
113}
114
115/* Format aofInfo as a string and it will be a line in the manifest.
116 *
117 * When update this format, make sure to update redis-check-aof as well. */
118sds aofInfoFormat(sds buf, aofInfo *ai) {
119 sds filename_repr = NULL;
120
121 if (sdsneedsrepr(ai->file_name))
122 filename_repr = sdscatrepr(sdsempty(), ai->file_name, sdslen(ai->file_name));
123
124 sds ret = sdscatprintf(buf, "%s %s %s %lld %s %c",
125 AOF_MANIFEST_KEY_FILE_NAME, filename_repr ? filename_repr : ai->file_name,
126 AOF_MANIFEST_KEY_FILE_SEQ, ai->file_seq,
127 AOF_MANIFEST_KEY_FILE_TYPE, ai->file_type);
128
129 if (ai->start_offset != -1) {
130 ret = sdscatprintf(ret, " %s %lld", AOF_MANIFEST_KEY_FILE_STARTOFFSET, ai->start_offset);
131 if (ai->end_offset != -1) {
132 ret = sdscatprintf(ret, " %s %lld", AOF_MANIFEST_KEY_FILE_ENDOFFSET, ai->end_offset);
133 }
134 }
135
136 ret = sdscatlen(ret, "\n", 1);
137 sdsfree(filename_repr);
138
139 return ret;
140}
141
142/* Method to free AOF list elements. */
143void aofListFree(void *item) {
144 aofInfo *ai = (aofInfo *)item;
145 aofInfoFree(ai);
146}
147
148/* Method to duplicate AOF list elements. */
149void *aofListDup(void *item) {
150 return aofInfoDup(item);
151}
152
153/* Create an empty aofManifest, which will be called in `aofLoadManifestFromDisk`. */
154aofManifest *aofManifestCreate(void) {
155 aofManifest *am = zcalloc(sizeof(aofManifest));
156 am->incr_aof_list = listCreate();
157 am->history_aof_list = listCreate();
158 listSetFreeMethod(am->incr_aof_list, aofListFree);
159 listSetDupMethod(am->incr_aof_list, aofListDup);
160 listSetFreeMethod(am->history_aof_list, aofListFree);
161 listSetDupMethod(am->history_aof_list, aofListDup);
162 return am;
163}
164
165/* Free the aofManifest structure (pointed to by am) and its embedded members. */
166void aofManifestFree(aofManifest *am) {
167 if (am->base_aof_info) aofInfoFree(am->base_aof_info);
168 if (am->incr_aof_list) listRelease(am->incr_aof_list);
169 if (am->history_aof_list) listRelease(am->history_aof_list);
170 zfree(am);
171}
172
173sds getAofManifestFileName(void) {
174 return sdscatprintf(sdsempty(), "%s%s", server.aof_filename,
175 MANIFEST_NAME_SUFFIX);
176}
177
178sds getTempAofManifestFileName(void) {
179 return sdscatprintf(sdsempty(), "%s%s%s", TEMP_FILE_NAME_PREFIX,
180 server.aof_filename, MANIFEST_NAME_SUFFIX);
181}
182
183sds appendAofInfoFromList(sds buf, list *aofList) {
184 listNode *ln;
185 listIter li;
186
187 listRewind(aofList, &li);
188 while ((ln = listNext(&li)) != NULL) {
189 aofInfo *ai = (aofInfo*)ln->value;
190 buf = aofInfoFormat(buf, ai);
191 }
192
193 return buf;
194}
195
196/* Returns the string representation of aofManifest pointed to by am.
197 *
198 * The string is multiple lines separated by '\n', and each line represents
199 * an AOF file.
200 *
201 * Each line is space delimited and contains 6 fields, as follows:
202 * "file" [filename] "seq" [sequence] "type" [type]
203 *
204 * Where "file", "seq" and "type" are keywords that describe the next value,
205 * [filename] and [sequence] describe file name and order, and [type] is one
206 * of 'b' (base), 'h' (history) or 'i' (incr).
207 *
208 * The base file, if exists, will always be first, followed by history files,
209 * and incremental files.
210 */
211sds getAofManifestAsString(aofManifest *am) {
212 serverAssert(am != NULL);
213
214 sds buf = sdsempty();
215
216 /* 1. Add BASE File information, it is always at the beginning
217 * of the manifest file. */
218 if (am->base_aof_info) {
219 buf = aofInfoFormat(buf, am->base_aof_info);
220 }
221
222 /* 2. Add HISTORY type AOF information. */
223 buf = appendAofInfoFromList(buf, am->history_aof_list);
224
225 /* 3. Add INCR type AOF information. */
226 buf = appendAofInfoFromList(buf, am->incr_aof_list);
227
228 return buf;
229}
230
231/* Load the manifest information from the disk to `server.aof_manifest`
232 * when the Redis server start.
233 *
234 * During loading, this function does strict error checking and will abort
235 * the entire Redis server process on error (I/O error, invalid format, etc.)
236 *
237 * If the AOF directory or manifest file do not exist, this will be ignored
238 * in order to support seamless upgrades from previous versions which did not
239 * use them.
240 */
241void aofLoadManifestFromDisk(void) {
242 server.aof_manifest = aofManifestCreate();
243 if (!dirExists(server.aof_dirname)) {
244 serverLog(LL_DEBUG, "The AOF directory %s doesn't exist", server.aof_dirname);
245 return;
246 }
247
248 sds am_name = getAofManifestFileName();
249 sds am_filepath = makePath(server.aof_dirname, am_name);
250 if (!fileExist(am_filepath)) {
251 serverLog(LL_DEBUG, "The AOF manifest file %s doesn't exist", am_name);
252 sdsfree(am_name);
253 sdsfree(am_filepath);
254 return;
255 }
256
257 aofManifest *am = aofLoadManifestFromFile(am_filepath);
258 if (am) aofManifestFreeAndUpdate(am);
259 sdsfree(am_name);
260 sdsfree(am_filepath);
261}
262
263/* Generic manifest loading function, used in `aofLoadManifestFromDisk` and redis-check-aof tool. */
264#define MANIFEST_MAX_LINE 1024
265aofManifest *aofLoadManifestFromFile(sds am_filepath) {
266 const char *err = NULL;
267 long long maxseq = 0;
268
269 aofManifest *am = aofManifestCreate();
270 FILE *fp = fopen(am_filepath, "r");
271 if (fp == NULL) {
272 serverLog(LL_WARNING, "Fatal error: can't open the AOF manifest "
273 "file %s for reading: %s", am_filepath, strerror(errno));
274 exit(1);
275 }
276
277 char buf[MANIFEST_MAX_LINE+1];
278 sds *argv = NULL;
279 int argc;
280 aofInfo *ai = NULL;
281
282 sds line = NULL;
283 int linenum = 0;
284
285 while (1) {
286 if (fgets(buf, MANIFEST_MAX_LINE+1, fp) == NULL) {
287 if (feof(fp)) {
288 if (linenum == 0) {
289 err = "Found an empty AOF manifest";
290 goto loaderr;
291 } else {
292 break;
293 }
294 } else {
295 err = "Read AOF manifest failed";
296 goto loaderr;
297 }
298 }
299
300 linenum++;
301
302 /* Skip comments lines */
303 if (buf[0] == '#') continue;
304
305 if (strchr(buf, '\n') == NULL) {
306 err = "The AOF manifest file contains too long line";
307 goto loaderr;
308 }
309
310 line = sdstrim(sdsnew(buf), " \t\r\n");
311 if (!sdslen(line)) {
312 err = "Invalid AOF manifest file format";
313 goto loaderr;
314 }
315
316 argv = sdssplitargs(line, &argc);
317 /* 'argc < 6' was done for forward compatibility. */
318 if (argv == NULL || argc < 6 || (argc % 2)) {
319 err = "Invalid AOF manifest file format";
320 goto loaderr;
321 }
322
323 ai = aofInfoCreate();
324 for (int i = 0; i < argc; i += 2) {
325 if (!strcasecmp(argv[i], AOF_MANIFEST_KEY_FILE_NAME)) {
326 ai->file_name = sdsnew(argv[i+1]);
327 if (!pathIsBaseName(ai->file_name)) {
328 err = "File can't be a path, just a filename";
329 goto loaderr;
330 }
331 } else if (!strcasecmp(argv[i], AOF_MANIFEST_KEY_FILE_SEQ)) {
332 ai->file_seq = atoll(argv[i+1]);
333 } else if (!strcasecmp(argv[i], AOF_MANIFEST_KEY_FILE_TYPE)) {
334 ai->file_type = (argv[i+1])[0];
335 } else if (!strcasecmp(argv[i], AOF_MANIFEST_KEY_FILE_STARTOFFSET)) {
336 ai->start_offset = atoll(argv[i+1]);
337 } else if (!strcasecmp(argv[i], AOF_MANIFEST_KEY_FILE_ENDOFFSET)) {
338 ai->end_offset = atoll(argv[i+1]);
339 }
340 /* else if (!strcasecmp(argv[i], AOF_MANIFEST_KEY_OTHER)) {} */
341 }
342
343 /* We have to make sure we load all the information. */
344 if (!ai->file_name || !ai->file_seq || !ai->file_type) {
345 err = "Invalid AOF manifest file format";
346 goto loaderr;
347 }
348
349 sdsfreesplitres(argv, argc);
350 argv = NULL;
351
352 if (ai->file_type == AOF_FILE_TYPE_BASE) {
353 if (am->base_aof_info) {
354 err = "Found duplicate base file information";
355 goto loaderr;
356 }
357 am->base_aof_info = ai;
358 am->curr_base_file_seq = ai->file_seq;
359 } else if (ai->file_type == AOF_FILE_TYPE_HIST) {
360 listAddNodeTail(am->history_aof_list, ai);
361 } else if (ai->file_type == AOF_FILE_TYPE_INCR) {
362 if (ai->file_seq <= maxseq) {
363 err = "Found a non-monotonic sequence number";
364 goto loaderr;
365 }
366 listAddNodeTail(am->incr_aof_list, ai);
367 am->curr_incr_file_seq = ai->file_seq;
368 maxseq = ai->file_seq;
369 } else {
370 err = "Unknown AOF file type";
371 goto loaderr;
372 }
373
374 sdsfree(line);
375 line = NULL;
376 ai = NULL;
377 }
378
379 fclose(fp);
380 return am;
381
382loaderr:
383 /* Sanitizer suppression: may report a false positive if we goto loaderr
384 * and exit(1) without freeing these allocations. */
385 if (argv) sdsfreesplitres(argv, argc);
386 if (ai) aofInfoFree(ai);
387
388 serverLog(LL_WARNING, "\n*** FATAL AOF MANIFEST FILE ERROR ***\n");
389 if (line) {
390 serverLog(LL_WARNING, "Reading the manifest file, at line %d\n", linenum);
391 serverLog(LL_WARNING, ">>> '%s'\n", line);
392 }
393 serverLog(LL_WARNING, "%s\n", err);
394 exit(1);
395}
396
397/* Deep copy an aofManifest from orig.
398 *
399 * In `backgroundRewriteDoneHandler` and `openNewIncrAofForAppend`, we will
400 * first deep copy a temporary AOF manifest from the `server.aof_manifest` and
401 * try to modify it. Once everything is modified, we will atomically make the
402 * `server.aof_manifest` point to this temporary aof_manifest.
403 */
404aofManifest *aofManifestDup(aofManifest *orig) {
405 serverAssert(orig != NULL);
406 aofManifest *am = zcalloc(sizeof(aofManifest));
407
408 am->curr_base_file_seq = orig->curr_base_file_seq;
409 am->curr_incr_file_seq = orig->curr_incr_file_seq;
410 am->dirty = orig->dirty;
411
412 if (orig->base_aof_info) {
413 am->base_aof_info = aofInfoDup(orig->base_aof_info);
414 }
415
416 am->incr_aof_list = listDup(orig->incr_aof_list);
417 am->history_aof_list = listDup(orig->history_aof_list);
418 serverAssert(am->incr_aof_list != NULL);
419 serverAssert(am->history_aof_list != NULL);
420 return am;
421}
422
423/* Change the `server.aof_manifest` pointer to 'am' and free the previous
424 * one if we have. */
425void aofManifestFreeAndUpdate(aofManifest *am) {
426 serverAssert(am != NULL);
427 if (server.aof_manifest) aofManifestFree(server.aof_manifest);
428 server.aof_manifest = am;
429}
430
431/* Called in `backgroundRewriteDoneHandler` to get a new BASE file
432 * name, and mark the previous (if we have) BASE file as HISTORY type.
433 *
434 * BASE file naming rules: `server.aof_filename`.seq.base.format
435 *
436 * for example:
437 * appendonly.aof.1.base.aof (server.aof_use_rdb_preamble is no)
438 * appendonly.aof.1.base.rdb (server.aof_use_rdb_preamble is yes)
439 */
440sds getNewBaseFileNameAndMarkPreAsHistory(aofManifest *am) {
441 serverAssert(am != NULL);
442 if (am->base_aof_info) {
443 serverAssert(am->base_aof_info->file_type == AOF_FILE_TYPE_BASE);
444 am->base_aof_info->file_type = AOF_FILE_TYPE_HIST;
445 listAddNodeHead(am->history_aof_list, am->base_aof_info);
446 }
447
448 char *format_suffix = server.aof_use_rdb_preamble ?
449 RDB_FORMAT_SUFFIX:AOF_FORMAT_SUFFIX;
450
451 aofInfo *ai = aofInfoCreate();
452 ai->file_name = sdscatprintf(sdsempty(), "%s.%lld%s%s", server.aof_filename,
453 ++am->curr_base_file_seq, BASE_FILE_SUFFIX, format_suffix);
454 ai->file_seq = am->curr_base_file_seq;
455 ai->file_type = AOF_FILE_TYPE_BASE;
456 am->base_aof_info = ai;
457 am->dirty = 1;
458 return am->base_aof_info->file_name;
459}
460
461/* Get a new INCR type AOF name.
462 *
463 * INCR AOF naming rules: `server.aof_filename`.seq.incr.aof
464 *
465 * for example:
466 * appendonly.aof.1.incr.aof
467 */
468sds getNewIncrAofName(aofManifest *am, long long start_reploff) {
469 aofInfo *ai = aofInfoCreate();
470 ai->file_type = AOF_FILE_TYPE_INCR;
471 ai->file_name = sdscatprintf(sdsempty(), "%s.%lld%s%s", server.aof_filename,
472 ++am->curr_incr_file_seq, INCR_FILE_SUFFIX, AOF_FORMAT_SUFFIX);
473 ai->file_seq = am->curr_incr_file_seq;
474 ai->start_offset = start_reploff;
475 listAddNodeTail(am->incr_aof_list, ai);
476 am->dirty = 1;
477 return ai->file_name;
478}
479
480/* Get temp INCR type AOF name. */
481sds getTempIncrAofName(void) {
482 return sdscatprintf(sdsempty(), "%s%s%s", TEMP_FILE_NAME_PREFIX, server.aof_filename,
483 INCR_FILE_SUFFIX);
484}
485
486/* Get the last INCR AOF name or create a new one. */
487sds getLastIncrAofName(aofManifest *am) {
488 serverAssert(am != NULL);
489
490 /* If 'incr_aof_list' is empty, just create a new one. */
491 if (!listLength(am->incr_aof_list)) {
492 return getNewIncrAofName(am, server.master_repl_offset);
493 }
494
495 /* Or return the last one. */
496 listNode *lastnode = listIndex(am->incr_aof_list, -1);
497 aofInfo *ai = listNodeValue(lastnode);
498 return ai->file_name;
499}
500
501/* Called in `backgroundRewriteDoneHandler`. when AOFRW success, This
502 * function will change the AOF file type in 'incr_aof_list' from
503 * AOF_FILE_TYPE_INCR to AOF_FILE_TYPE_HIST, and move them to the
504 * 'history_aof_list'.
505 */
506void markRewrittenIncrAofAsHistory(aofManifest *am) {
507 serverAssert(am != NULL);
508 if (!listLength(am->incr_aof_list)) {
509 return;
510 }
511
512 listNode *ln;
513 listIter li;
514
515 listRewindTail(am->incr_aof_list, &li);
516
517 /* "server.aof_fd != -1" means AOF enabled, then we must skip the
518 * last AOF, because this file is our currently writing. */
519 if (server.aof_fd != -1) {
520 ln = listNext(&li);
521 serverAssert(ln != NULL);
522 }
523
524 /* Move aofInfo from 'incr_aof_list' to 'history_aof_list'. */
525 while ((ln = listNext(&li)) != NULL) {
526 aofInfo *ai = (aofInfo*)ln->value;
527 serverAssert(ai->file_type == AOF_FILE_TYPE_INCR);
528
529 aofInfo *hai = aofInfoDup(ai);
530 hai->file_type = AOF_FILE_TYPE_HIST;
531 listAddNodeHead(am->history_aof_list, hai);
532 listDelNode(am->incr_aof_list, ln);
533 }
534
535 am->dirty = 1;
536}
537
538/* Write the formatted manifest string to disk. */
539int writeAofManifestFile(sds buf) {
540 int ret = C_OK;
541 ssize_t nwritten;
542 int len;
543
544 sds am_name = getAofManifestFileName();
545 sds am_filepath = makePath(server.aof_dirname, am_name);
546 sds tmp_am_name = getTempAofManifestFileName();
547 sds tmp_am_filepath = makePath(server.aof_dirname, tmp_am_name);
548
549 int fd = open(tmp_am_filepath, O_WRONLY|O_TRUNC|O_CREAT, 0644);
550 if (fd == -1) {
551 serverLog(LL_WARNING, "Can't open the AOF manifest file %s: %s",
552 tmp_am_name, strerror(errno));
553
554 ret = C_ERR;
555 goto cleanup;
556 }
557
558 len = sdslen(buf);
559 while(len) {
560 nwritten = write(fd, buf, len);
561
562 if (nwritten < 0) {
563 if (errno == EINTR) continue;
564
565 serverLog(LL_WARNING, "Error trying to write the temporary AOF manifest file %s: %s",
566 tmp_am_name, strerror(errno));
567
568 ret = C_ERR;
569 goto cleanup;
570 }
571
572 len -= nwritten;
573 buf += nwritten;
574 }
575
576 if (redis_fsync(fd) == -1) {
577 serverLog(LL_WARNING, "Fail to fsync the temp AOF file %s: %s.",
578 tmp_am_name, strerror(errno));
579
580 ret = C_ERR;
581 goto cleanup;
582 }
583
584 if (rename(tmp_am_filepath, am_filepath) != 0) {
585 serverLog(LL_WARNING,
586 "Error trying to rename the temporary AOF manifest file %s into %s: %s",
587 tmp_am_name, am_name, strerror(errno));
588
589 ret = C_ERR;
590 goto cleanup;
591 }
592
593 /* Also sync the AOF directory as new AOF files may be added in the directory */
594 if (fsyncFileDir(am_filepath) == -1) {
595 serverLog(LL_WARNING, "Fail to fsync AOF directory %s: %s.",
596 am_filepath, strerror(errno));
597
598 ret = C_ERR;
599 goto cleanup;
600 }
601
602cleanup:
603 if (fd != -1) close(fd);
604 sdsfree(am_name);
605 sdsfree(am_filepath);
606 sdsfree(tmp_am_name);
607 sdsfree(tmp_am_filepath);
608 return ret;
609}
610
611/* Persist the aofManifest information pointed to by am to disk. */
612int persistAofManifest(aofManifest *am) {
613 if (am->dirty == 0) {
614 return C_OK;
615 }
616
617 sds amstr = getAofManifestAsString(am);
618 int ret = writeAofManifestFile(amstr);
619 sdsfree(amstr);
620 if (ret == C_OK) am->dirty = 0;
621 return ret;
622}
623
624/* Called in `loadAppendOnlyFiles` when we upgrade from a old version redis.
625 *
626 * 1) Create AOF directory use 'server.aof_dirname' as the name.
627 * 2) Use 'server.aof_filename' to construct a BASE type aofInfo and add it to
628 * aofManifest, then persist the manifest file to AOF directory.
629 * 3) Move the old AOF file (server.aof_filename) to AOF directory.
630 *
631 * If any of the above steps fails or crash occurs, this will not cause any
632 * problems, and redis will retry the upgrade process when it restarts.
633 */
634void aofUpgradePrepare(aofManifest *am) {
635 serverAssert(!aofFileExist(server.aof_filename));
636
637 /* Create AOF directory use 'server.aof_dirname' as the name. */
638 if (dirCreateIfMissing(server.aof_dirname) == -1) {
639 serverLog(LL_WARNING, "Can't open or create append-only dir %s: %s",
640 server.aof_dirname, strerror(errno));
641 exit(1);
642 }
643
644 /* Manually construct a BASE type aofInfo and add it to aofManifest. */
645 if (am->base_aof_info) aofInfoFree(am->base_aof_info);
646 aofInfo *ai = aofInfoCreate();
647 ai->file_name = sdsnew(server.aof_filename);
648 ai->file_seq = 1;
649 ai->file_type = AOF_FILE_TYPE_BASE;
650 am->base_aof_info = ai;
651 am->curr_base_file_seq = 1;
652 am->dirty = 1;
653
654 /* Persist the manifest file to AOF directory. */
655 if (persistAofManifest(am) != C_OK) {
656 exit(1);
657 }
658
659 /* Move the old AOF file to AOF directory. */
660 sds aof_filepath = makePath(server.aof_dirname, server.aof_filename);
661 if (rename(server.aof_filename, aof_filepath) == -1) {
662 serverLog(LL_WARNING,
663 "Error trying to move the old AOF file %s into dir %s: %s",
664 server.aof_filename,
665 server.aof_dirname,
666 strerror(errno));
667 sdsfree(aof_filepath);
668 exit(1);
669 }
670 sdsfree(aof_filepath);
671
672 serverLog(LL_NOTICE, "Successfully migrated an old-style AOF file (%s) into the AOF directory (%s).",
673 server.aof_filename, server.aof_dirname);
674}
675
676/* When AOFRW success, the previous BASE and INCR AOFs will
677 * become HISTORY type and be moved into 'history_aof_list'.
678 *
679 * The function will traverse the 'history_aof_list' and submit
680 * the delete task to the bio thread.
681 */
682int aofDelHistoryFiles(void) {
683 if (server.aof_manifest == NULL ||
684 server.aof_disable_auto_gc == 1 ||
685 !listLength(server.aof_manifest->history_aof_list))
686 {
687 return C_OK;
688 }
689
690 listNode *ln;
691 listIter li;
692
693 listRewind(server.aof_manifest->history_aof_list, &li);
694 while ((ln = listNext(&li)) != NULL) {
695 aofInfo *ai = (aofInfo*)ln->value;
696 serverAssert(ai->file_type == AOF_FILE_TYPE_HIST);
697 serverLog(LL_NOTICE, "Removing the history file %s in the background", ai->file_name);
698 sds aof_filepath = makePath(server.aof_dirname, ai->file_name);
699 bg_unlink(aof_filepath);
700 sdsfree(aof_filepath);
701 listDelNode(server.aof_manifest->history_aof_list, ln);
702 }
703
704 server.aof_manifest->dirty = 1;
705 return persistAofManifest(server.aof_manifest);
706}
707
708/* Used to clean up temp INCR AOF when AOFRW fails. */
709void aofDelTempIncrAofFile(void) {
710 sds aof_filename = getTempIncrAofName();
711 sds aof_filepath = makePath(server.aof_dirname, aof_filename);
712 serverLog(LL_NOTICE, "Removing the temp incr aof file %s in the background", aof_filename);
713 bg_unlink(aof_filepath);
714 sdsfree(aof_filepath);
715 sdsfree(aof_filename);
716 return;
717}
718
719/* Called after `loadDataFromDisk` when redis start. If `server.aof_state` is
720 * 'AOF_ON', It will do three things:
721 * 1. Force create a BASE file when redis starts with an empty dataset
722 * 2. Open the last opened INCR type AOF for writing, If not, create a new one
723 * 3. Synchronously update the manifest file to the disk
724 *
725 * If any of the above steps fails, the redis process will exit.
726 */
727void aofOpenIfNeededOnServerStart(void) {
728 if (server.aof_state != AOF_ON) {
729 return;
730 }
731
732 serverAssert(server.aof_manifest != NULL);
733 serverAssert(server.aof_fd == -1);
734
735 if (dirCreateIfMissing(server.aof_dirname) == -1) {
736 serverLog(LL_WARNING, "Can't open or create append-only dir %s: %s",
737 server.aof_dirname, strerror(errno));
738 exit(1);
739 }
740
741 /* If we start with an empty dataset, we will force create a BASE file. */
742 size_t incr_aof_len = listLength(server.aof_manifest->incr_aof_list);
743 if (!server.aof_manifest->base_aof_info && !incr_aof_len) {
744 sds base_name = getNewBaseFileNameAndMarkPreAsHistory(server.aof_manifest);
745 sds base_filepath = makePath(server.aof_dirname, base_name);
746 if (rewriteAppendOnlyFile(base_filepath) != C_OK) {
747 exit(1);
748 }
749 sdsfree(base_filepath);
750 serverLog(LL_NOTICE, "Creating AOF base file %s on server start",
751 base_name);
752 }
753
754 /* Because we will 'exit(1)' if open AOF or persistent manifest fails, so
755 * we don't need atomic modification here. */
756 sds aof_name = getLastIncrAofName(server.aof_manifest);
757
758 /* Here we should use 'O_APPEND' flag. */
759 sds aof_filepath = makePath(server.aof_dirname, aof_name);
760 server.aof_fd = open(aof_filepath, O_WRONLY|O_APPEND|O_CREAT, 0644);
761 sdsfree(aof_filepath);
762 if (server.aof_fd == -1) {
763 serverLog(LL_WARNING, "Can't open the append-only file %s: %s",
764 aof_name, strerror(errno));
765 exit(1);
766 }
767
768 /* Persist our changes. */
769 int ret = persistAofManifest(server.aof_manifest);
770 if (ret != C_OK) {
771 exit(1);
772 }
773
774 server.aof_last_incr_size = getAppendOnlyFileSize(aof_name, NULL);
775 server.aof_last_incr_fsync_offset = server.aof_last_incr_size;
776
777 if (incr_aof_len) {
778 serverLog(LL_NOTICE, "Opening AOF incr file %s on server start", aof_name);
779 } else {
780 serverLog(LL_NOTICE, "Creating AOF incr file %s on server start", aof_name);
781 }
782}
783
784int aofFileExist(char *filename) {
785 sds file_path = makePath(server.aof_dirname, filename);
786 int ret = fileExist(file_path);
787 sdsfree(file_path);
788 return ret;
789}
790
791/* Called in `rewriteAppendOnlyFileBackground`. If `server.aof_state`
792 * is 'AOF_ON', It will do two things:
793 * 1. Open a new INCR type AOF for writing
794 * 2. Synchronously update the manifest file to the disk
795 *
796 * The above two steps of modification are atomic, that is, if
797 * any step fails, the entire operation will rollback and returns
798 * C_ERR, and if all succeeds, it returns C_OK.
799 *
800 * If `server.aof_state` is 'AOF_WAIT_REWRITE', It will open a temporary INCR AOF
801 * file to accumulate data during AOF_WAIT_REWRITE, and it will eventually be
802 * renamed in the `backgroundRewriteDoneHandler` and written to the manifest file.
803 * */
804int openNewIncrAofForAppend(void) {
805 serverAssert(server.aof_manifest != NULL);
806 int newfd = -1;
807 aofManifest *temp_am = NULL;
808 sds new_aof_name = NULL;
809
810 /* Only open new INCR AOF when AOF enabled. */
811 if (server.aof_state == AOF_OFF) return C_OK;
812
813 /* Open new AOF. */
814 if (server.aof_state == AOF_WAIT_REWRITE) {
815 /* Use a temporary INCR AOF file to accumulate data during AOF_WAIT_REWRITE. */
816 new_aof_name = getTempIncrAofName();
817 tempIncAofStartReplOffset = server.master_repl_offset;
818 } else {
819 /* Dup a temp aof_manifest to modify. */
820 temp_am = aofManifestDup(server.aof_manifest);
821 new_aof_name = sdsdup(getNewIncrAofName(temp_am, server.master_repl_offset));
822 }
823 sds new_aof_filepath = makePath(server.aof_dirname, new_aof_name);
824 newfd = open(new_aof_filepath, O_WRONLY|O_TRUNC|O_CREAT, 0644);
825 sdsfree(new_aof_filepath);
826 if (newfd == -1) {
827 serverLog(LL_WARNING, "Can't open the append-only file %s: %s",
828 new_aof_name, strerror(errno));
829 goto cleanup;
830 }
831
832 if (temp_am) {
833 /* Persist AOF Manifest. */
834 if (persistAofManifest(temp_am) == C_ERR) {
835 goto cleanup;
836 }
837 }
838
839 serverLog(LL_NOTICE, "Creating AOF incr file %s on background rewrite",
840 new_aof_name);
841 sdsfree(new_aof_name);
842
843 /* If reaches here, we can safely modify the `server.aof_manifest`
844 * and `server.aof_fd`. */
845
846 /* fsync and close old aof_fd if needed. In fsync everysec it's ok to delay
847 * the fsync as long as we grantee it happens, and in fsync always the file
848 * is already synced at this point so fsync doesn't matter. */
849 if (server.aof_fd != -1) {
850 aof_background_fsync_and_close(server.aof_fd);
851 server.aof_last_fsync = server.mstime;
852 }
853 server.aof_fd = newfd;
854
855 /* Reset the aof_last_incr_size. */
856 server.aof_last_incr_size = 0;
857 /* Reset the aof_last_incr_fsync_offset. */
858 server.aof_last_incr_fsync_offset = 0;
859 /* Update `server.aof_manifest`. */
860 if (temp_am) aofManifestFreeAndUpdate(temp_am);
861 return C_OK;
862
863cleanup:
864 if (new_aof_name) sdsfree(new_aof_name);
865 if (newfd != -1) close(newfd);
866 if (temp_am) aofManifestFree(temp_am);
867 return C_ERR;
868}
869
870/* When we close gracefully the AOF file, we have the chance to persist the
871 * end replication offset of current INCR AOF. */
872void updateCurIncrAofEndOffset(void) {
873 if (server.aof_state != AOF_ON) return;
874 serverAssert(server.aof_manifest != NULL);
875
876 if (listLength(server.aof_manifest->incr_aof_list) == 0) return;
877 aofInfo *ai = listNodeValue(listLast(server.aof_manifest->incr_aof_list));
878 ai->end_offset = server.master_repl_offset;
879 server.aof_manifest->dirty = 1;
880 /* It doesn't matter if the persistence fails since this information is not
881 * critical, we can get an approximate value by start offset plus file size. */
882 persistAofManifest(server.aof_manifest);
883}
884
885/* After loading AOF data, we need to update the `server.master_repl_offset`
886 * based on the information of the last INCR AOF, to avoid the rollback of
887 * the start offset of new INCR AOF. */
888void updateReplOffsetAndResetEndOffset(void) {
889 if (server.aof_state != AOF_ON) return;
890 serverAssert(server.aof_manifest != NULL);
891
892 /* If the INCR file has an end offset, we directly use it, and clear it
893 * to avoid the next time we load the manifest file, we will use the same
894 * offset, but the real offset may have advanced. */
895 if (listLength(server.aof_manifest->incr_aof_list) == 0) return;
896 aofInfo *ai = listNodeValue(listLast(server.aof_manifest->incr_aof_list));
897 if (ai->end_offset != -1) {
898 server.master_repl_offset = ai->end_offset;
899 ai->end_offset = -1;
900 server.aof_manifest->dirty = 1;
901 /* We must update the end offset of INCR file correctly, otherwise we
902 * may keep wrong information in the manifest file, since we continue
903 * to append data to the same INCR file. */
904 if (persistAofManifest(server.aof_manifest) != AOF_OK)
905 exit(1);
906 } else {
907 /* If the INCR file doesn't have an end offset, we need to calculate
908 * the replication offset by the start offset plus the file size. */
909 server.master_repl_offset = (ai->start_offset == -1 ? 0 : ai->start_offset) +
910 getAppendOnlyFileSize(ai->file_name, NULL);
911 }
912}
913
914/* Whether to limit the execution of Background AOF rewrite.
915 *
916 * At present, if AOFRW fails, redis will automatically retry. If it continues
917 * to fail, we may get a lot of very small INCR files. so we need an AOFRW
918 * limiting measure.
919 *
920 * We can't directly use `server.aof_current_size` and `server.aof_last_incr_size`,
921 * because there may be no new writes after AOFRW fails.
922 *
923 * So, we use time delay to achieve our goal. When AOFRW fails, we delay the execution
924 * of the next AOFRW by 1 minute. If the next AOFRW also fails, it will be delayed by 2
925 * minutes. The next is 4, 8, 16, the maximum delay is 60 minutes (1 hour).
926 *
927 * During the limit period, we can still use the 'bgrewriteaof' command to execute AOFRW
928 * immediately.
929 *
930 * Return 1 means that AOFRW is limited and cannot be executed. 0 means that we can execute
931 * AOFRW, which may be that we have reached the 'next_rewrite_time' or the number of INCR
932 * AOFs has not reached the limit threshold.
933 * */
934#define AOF_REWRITE_LIMITE_THRESHOLD 3
935#define AOF_REWRITE_LIMITE_MAX_MINUTES 60 /* 1 hour */
936int aofRewriteLimited(void) {
937 static int next_delay_minutes = 0;
938 static time_t next_rewrite_time = 0;
939
940 if (server.stat_aofrw_consecutive_failures < AOF_REWRITE_LIMITE_THRESHOLD) {
941 /* We may be recovering from limited state, so reset all states. */
942 next_delay_minutes = 0;
943 next_rewrite_time = 0;
944 return 0;
945 }
946
947 /* if it is in the limiting state, then check if the next_rewrite_time is reached */
948 if (next_rewrite_time != 0) {
949 if (server.unixtime < next_rewrite_time) {
950 return 1;
951 } else {
952 next_rewrite_time = 0;
953 return 0;
954 }
955 }
956
957 next_delay_minutes = (next_delay_minutes == 0) ? 1 : (next_delay_minutes * 2);
958 if (next_delay_minutes > AOF_REWRITE_LIMITE_MAX_MINUTES) {
959 next_delay_minutes = AOF_REWRITE_LIMITE_MAX_MINUTES;
960 }
961
962 next_rewrite_time = server.unixtime + next_delay_minutes * 60;
963 serverLog(LL_WARNING,
964 "Background AOF rewrite has repeatedly failed and triggered the limit, will retry in %d minutes", next_delay_minutes);
965 return 1;
966}
967
968/* ----------------------------------------------------------------------------
969 * AOF file implementation
970 * ------------------------------------------------------------------------- */
971
972/* Return true if an AOf fsync is currently already in progress in a
973 * BIO thread. */
974int aofFsyncInProgress(void) {
975 /* Note that we don't care about aof_background_fsync_and_close because
976 * server.aof_fd has been replaced by the new INCR AOF file fd,
977 * see openNewIncrAofForAppend. */
978 return bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;
979}
980
981/* Starts a background task that performs fsync() against the specified
982 * file descriptor (the one of the AOF file) in another thread. */
983void aof_background_fsync(int fd) {
984 bioCreateFsyncJob(fd, server.master_repl_offset, 1);
985}
986
987/* Close the fd on the basis of aof_background_fsync. */
988void aof_background_fsync_and_close(int fd) {
989 bioCreateCloseAofJob(fd, server.master_repl_offset, 1);
990}
991
992/* Kills an AOFRW child process if exists */
993void killAppendOnlyChild(void) {
994 int statloc;
995 /* No AOFRW child? return. */
996 if (server.child_type != CHILD_TYPE_AOF) return;
997 /* Kill AOFRW child, wait for child exit. */
998 serverLog(LL_NOTICE,"Killing running AOF rewrite child: %ld",
999 (long) server.child_pid);
1000 if (kill(server.child_pid,SIGUSR1) != -1) {
1001 while(waitpid(-1, &statloc, 0) != server.child_pid);
1002 }
1003 aofRemoveTempFile(server.child_pid);
1004 resetChildState();
1005 server.aof_rewrite_time_start = -1;
1006}
1007
1008/* Called when the user switches from "appendonly yes" to "appendonly no"
1009 * at runtime using the CONFIG command. */
1010void stopAppendOnly(void) {
1011 serverAssert(server.aof_state != AOF_OFF);
1012 flushAppendOnlyFile(1);
1013 if (redis_fsync(server.aof_fd) == -1) {
1014 serverLog(LL_WARNING,"Fail to fsync the AOF file: %s",strerror(errno));
1015 } else {
1016 server.aof_last_fsync = server.mstime;
1017 }
1018 close(server.aof_fd);
1019 updateCurIncrAofEndOffset();
1020
1021 server.aof_fd = -1;
1022 server.aof_selected_db = -1;
1023 server.aof_state = AOF_OFF;
1024 server.aof_rewrite_scheduled = 0;
1025 server.aof_last_incr_size = 0;
1026 server.aof_last_incr_fsync_offset = 0;
1027 server.fsynced_reploff = -1;
1028 atomicSet(server.fsynced_reploff_pending, 0);
1029 killAppendOnlyChild();
1030 sdsfree(server.aof_buf);
1031 server.aof_buf = sdsempty();
1032}
1033
1034/* Called when the user switches from "appendonly no" to "appendonly yes"
1035 * at runtime using the CONFIG command. */
1036int startAppendOnly(void) {
1037 serverAssert(server.aof_state == AOF_OFF);
1038
1039 server.aof_state = AOF_WAIT_REWRITE;
1040 if (hasActiveChildProcess() && server.child_type != CHILD_TYPE_AOF) {
1041 server.aof_rewrite_scheduled = 1;
1042 serverLog(LL_NOTICE,"AOF was enabled but there is already another background operation. An AOF background was scheduled to start when possible.");
1043 } else if (server.in_exec){
1044 server.aof_rewrite_scheduled = 1;
1045 serverLog(LL_NOTICE,"AOF was enabled during a transaction. An AOF background was scheduled to start when possible.");
1046 } else {
1047 /* If there is a pending AOF rewrite, we need to switch it off and
1048 * start a new one: the old one cannot be reused because it is not
1049 * accumulating the AOF buffer. */
1050 if (server.child_type == CHILD_TYPE_AOF) {
1051 serverLog(LL_NOTICE,"AOF was enabled but there is already an AOF rewriting in background. Stopping background AOF and starting a rewrite now.");
1052 killAppendOnlyChild();
1053 }
1054
1055 if (rewriteAppendOnlyFileBackground() == C_ERR) {
1056 server.aof_state = AOF_OFF;
1057 serverLog(LL_WARNING,"Redis needs to enable the AOF but can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.");
1058 return C_ERR;
1059 }
1060 }
1061 server.aof_last_fsync = server.mstime;
1062 /* If AOF fsync error in bio job, we just ignore it and log the event. */
1063 int aof_bio_fsync_status;
1064 atomicGet(server.aof_bio_fsync_status, aof_bio_fsync_status);
1065 if (aof_bio_fsync_status == C_ERR) {
1066 serverLog(LL_WARNING,
1067 "AOF reopen, just ignore the AOF fsync error in bio job");
1068 atomicSet(server.aof_bio_fsync_status,C_OK);
1069 }
1070
1071 /* If AOF was in error state, we just ignore it and log the event. */
1072 if (server.aof_last_write_status == C_ERR) {
1073 serverLog(LL_WARNING,"AOF reopen, just ignore the last error.");
1074 server.aof_last_write_status = C_OK;
1075 }
1076 return C_OK;
1077}
1078
1079void startAppendOnlyWithRetry(void) {
1080 unsigned int tries, max_tries = 10;
1081 for (tries = 0; tries < max_tries; ++tries) {
1082 if (startAppendOnly() == C_OK)
1083 break;
1084 serverLog(LL_WARNING, "Failed to enable AOF! Trying it again in one second.");
1085 sleep(1);
1086 }
1087 if (tries == max_tries) {
1088 serverLog(LL_WARNING, "FATAL: AOF can't be turned on. Exiting now.");
1089 exit(1);
1090 }
1091}
1092
1093/* Called after "appendonly" config is changed. */
1094void applyAppendOnlyConfig(void) {
1095 if (!server.aof_enabled && server.aof_state != AOF_OFF) {
1096 stopAppendOnly();
1097 } else if (server.aof_enabled && server.aof_state == AOF_OFF) {
1098 startAppendOnlyWithRetry();
1099 }
1100}
1101
1102/* This is a wrapper to the write syscall in order to retry on short writes
1103 * or if the syscall gets interrupted. It could look strange that we retry
1104 * on short writes given that we are writing to a block device: normally if
1105 * the first call is short, there is a end-of-space condition, so the next
1106 * is likely to fail. However apparently in modern systems this is no longer
1107 * true, and in general it looks just more resilient to retry the write. If
1108 * there is an actual error condition we'll get it at the next try. */
1109ssize_t aofWrite(int fd, const char *buf, size_t len) {
1110 ssize_t nwritten = 0, totwritten = 0;
1111
1112 while(len) {
1113 nwritten = write(fd, buf, len);
1114
1115 if (nwritten < 0) {
1116 if (errno == EINTR) continue;
1117 return totwritten ? totwritten : -1;
1118 }
1119
1120 len -= nwritten;
1121 buf += nwritten;
1122 totwritten += nwritten;
1123 }
1124
1125 return totwritten;
1126}
1127
1128/* Write the append only file buffer on disk.
1129 *
1130 * Since we are required to write the AOF before replying to the client,
1131 * and the only way the client socket can get a write is entering when
1132 * the event loop, we accumulate all the AOF writes in a memory
1133 * buffer and write it on disk using this function just before entering
1134 * the event loop again.
1135 *
1136 * About the 'force' argument:
1137 *
1138 * When the fsync policy is set to 'everysec' we may delay the flush if there
1139 * is still an fsync() going on in the background thread, since for instance
1140 * on Linux write(2) will be blocked by the background fsync anyway.
1141 * When this happens we remember that there is some aof buffer to be
1142 * flushed ASAP, and will try to do that in the serverCron() function.
1143 *
1144 * However if force is set to 1 we'll write regardless of the background
1145 * fsync. */
1146#define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */
1147void flushAppendOnlyFile(int force) {
1148 ssize_t nwritten;
1149 int sync_in_progress = 0;
1150 mstime_t latency;
1151
1152 if (sdslen(server.aof_buf) == 0) {
1153 if (server.aof_last_incr_fsync_offset == server.aof_last_incr_size) {
1154 /* All data is fsync'd already: Update fsynced_reploff_pending just in case.
1155 * This is needed to avoid a WAITAOF hang in case a module used RM_Call
1156 * with the NO_AOF flag, in which case master_repl_offset will increase but
1157 * fsynced_reploff_pending won't be updated (because there's no reason, from
1158 * the AOF POV, to call fsync) and then WAITAOF may wait on the higher offset
1159 * (which contains data that was only propagated to replicas, and not to AOF) */
1160 if (!aofFsyncInProgress())
1161 atomicSet(server.fsynced_reploff_pending, server.master_repl_offset);
1162 } else {
1163 /* Check if we need to do fsync even the aof buffer is empty,
1164 * because previously in AOF_FSYNC_EVERYSEC mode, fsync is
1165 * called only when aof buffer is not empty, so if users
1166 * stop write commands before fsync called in one second,
1167 * the data in page cache cannot be flushed in time. */
1168 if (server.aof_fsync == AOF_FSYNC_EVERYSEC &&
1169 server.mstime - server.aof_last_fsync >= 1000 &&
1170 !(sync_in_progress = aofFsyncInProgress()))
1171 goto try_fsync;
1172
1173 /* Check if we need to do fsync even the aof buffer is empty,
1174 * the reason is described in the previous AOF_FSYNC_EVERYSEC block,
1175 * and AOF_FSYNC_ALWAYS is also checked here to handle a case where
1176 * aof_fsync is changed from everysec to always. */
1177 if (server.aof_fsync == AOF_FSYNC_ALWAYS)
1178 goto try_fsync;
1179 }
1180 return;
1181 }
1182
1183 if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
1184 sync_in_progress = aofFsyncInProgress();
1185
1186 if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
1187 /* With this append fsync policy we do background fsyncing.
1188 * If the fsync is still in progress we can try to delay
1189 * the write for a couple of seconds. */
1190 if (sync_in_progress) {
1191 if (server.aof_flush_postponed_start == 0) {
1192 /* No previous write postponing, remember that we are
1193 * postponing the flush and return. */
1194 server.aof_flush_postponed_start = server.mstime;
1195 return;
1196 } else if (server.mstime - server.aof_flush_postponed_start < 2000) {
1197 /* We were already waiting for fsync to finish, but for less
1198 * than two seconds this is still ok. Postpone again. */
1199 return;
1200 }
1201 /* Otherwise fall through, and go write since we can't wait
1202 * over two seconds. */
1203 server.aof_delayed_fsync++;
1204 serverLog(LL_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.");
1205 }
1206 }
1207 /* We want to perform a single write. This should be guaranteed atomic
1208 * at least if the filesystem we are writing is a real physical one.
1209 * While this will save us against the server being killed I don't think
1210 * there is much to do about the whole server stopping for power problems
1211 * or alike */
1212
1213 if (server.aof_flush_sleep && sdslen(server.aof_buf)) {
1214 usleep(server.aof_flush_sleep);
1215 }
1216
1217 latencyStartMonitor(latency);
1218 nwritten = aofWrite(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
1219 latencyEndMonitor(latency);
1220 /* We want to capture different events for delayed writes:
1221 * when the delay happens with a pending fsync, or with a saving child
1222 * active, and when the above two conditions are missing.
1223 * We also use an additional event name to save all samples which is
1224 * useful for graphing / monitoring purposes. */
1225 if (sync_in_progress) {
1226 latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
1227 } else if (hasActiveChildProcess()) {
1228 latencyAddSampleIfNeeded("aof-write-active-child",latency);
1229 } else {
1230 latencyAddSampleIfNeeded("aof-write-alone",latency);
1231 }
1232 latencyAddSampleIfNeeded("aof-write",latency);
1233
1234 /* We performed the write so reset the postponed flush sentinel to zero. */
1235 server.aof_flush_postponed_start = 0;
1236
1237 if (nwritten != (ssize_t)sdslen(server.aof_buf)) {
1238 static time_t last_write_error_log = 0;
1239 int can_log = 0;
1240
1241 /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */
1242 if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) {
1243 can_log = 1;
1244 last_write_error_log = server.unixtime;
1245 }
1246
1247 /* Log the AOF write error and record the error code. */
1248 if (nwritten == -1) {
1249 if (can_log) {
1250 serverLog(LL_WARNING,"Error writing to the AOF file: %s",
1251 strerror(errno));
1252 }
1253 server.aof_last_write_errno = errno;
1254 } else {
1255 if (can_log) {
1256 serverLog(LL_WARNING,"Short write while writing to "
1257 "the AOF file: (nwritten=%lld, "
1258 "expected=%lld)",
1259 (long long)nwritten,
1260 (long long)sdslen(server.aof_buf));
1261 }
1262
1263 if (ftruncate(server.aof_fd, server.aof_last_incr_size) == -1) {
1264 if (can_log) {
1265 serverLog(LL_WARNING, "Could not remove short write "
1266 "from the append-only file. Redis may refuse "
1267 "to load the AOF the next time it starts. "
1268 "ftruncate: %s", strerror(errno));
1269 }
1270 } else {
1271 /* If the ftruncate() succeeded we can set nwritten to
1272 * -1 since there is no longer partial data into the AOF. */
1273 nwritten = -1;
1274 }
1275 server.aof_last_write_errno = ENOSPC;
1276 }
1277
1278 /* Handle the AOF write error. */
1279 if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
1280 /* We can't recover when the fsync policy is ALWAYS since the reply
1281 * for the client is already in the output buffers (both writes and
1282 * reads), and the changes to the db can't be rolled back. Since we
1283 * have a contract with the user that on acknowledged or observed
1284 * writes are is synced on disk, we must exit. */
1285 serverLog(LL_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
1286 exit(1);
1287 } else {
1288 /* Recover from failed write leaving data into the buffer. However
1289 * set an error to stop accepting writes as long as the error
1290 * condition is not cleared. */
1291 server.aof_last_write_status = C_ERR;
1292
1293 /* Trim the sds buffer if there was a partial write, and there
1294 * was no way to undo it with ftruncate(2). */
1295 if (nwritten > 0) {
1296 server.aof_current_size += nwritten;
1297 server.aof_last_incr_size += nwritten;
1298 sdsrange(server.aof_buf,nwritten,-1);
1299 }
1300 return; /* We'll try again on the next call... */
1301 }
1302 } else {
1303 /* Successful write(2). If AOF was in error state, restore the
1304 * OK state and log the event. */
1305 if (server.aof_last_write_status == C_ERR) {
1306 serverLog(LL_NOTICE,
1307 "AOF write error looks solved, Redis can write again.");
1308 server.aof_last_write_status = C_OK;
1309 }
1310 }
1311 server.aof_current_size += nwritten;
1312 server.aof_last_incr_size += nwritten;
1313
1314 /* Re-use AOF buffer when it is small enough. The maximum comes from the
1315 * arena size of 4k minus some overhead (but is otherwise arbitrary). */
1316 if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
1317 sdsclear(server.aof_buf);
1318 } else {
1319 sdsfree(server.aof_buf);
1320 server.aof_buf = sdsempty();
1321 }
1322
1323try_fsync:
1324 /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
1325 * children doing I/O in the background. */
1326 if (server.aof_no_fsync_on_rewrite && hasActiveChildProcess())
1327 return;
1328
1329 /* Perform the fsync if needed. */
1330 if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
1331 /* redis_fsync is defined as fdatasync() for Linux in order to avoid
1332 * flushing metadata. */
1333 latencyStartMonitor(latency);
1334 /* Let's try to get this data on the disk. To guarantee data safe when
1335 * the AOF fsync policy is 'always', we should exit if failed to fsync
1336 * AOF (see comment next to the exit(1) after write error above). */
1337 if (redis_fsync(server.aof_fd) == -1) {
1338 serverLog(LL_WARNING,"Can't persist AOF for fsync error when the "
1339 "AOF fsync policy is 'always': %s. Exiting...", strerror(errno));
1340 exit(1);
1341 }
1342 latencyEndMonitor(latency);
1343 latencyAddSampleIfNeeded("aof-fsync-always",latency);
1344 server.aof_last_incr_fsync_offset = server.aof_last_incr_size;
1345 server.aof_last_fsync = server.mstime;
1346 atomicSet(server.fsynced_reploff_pending, server.master_repl_offset);
1347 } else if (server.aof_fsync == AOF_FSYNC_EVERYSEC &&
1348 server.mstime - server.aof_last_fsync >= 1000) {
1349 if (!sync_in_progress) {
1350 aof_background_fsync(server.aof_fd);
1351 server.aof_last_incr_fsync_offset = server.aof_last_incr_size;
1352 }
1353 server.aof_last_fsync = server.mstime;
1354 }
1355}
1356
1357sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) {
1358 char buf[32];
1359 int len, j;
1360 robj *o;
1361
1362 buf[0] = '*';
1363 len = 1+ll2string(buf+1,sizeof(buf)-1,argc);
1364 buf[len++] = '\r';
1365 buf[len++] = '\n';
1366 dst = sdscatlen(dst,buf,len);
1367
1368 for (j = 0; j < argc; j++) {
1369 o = getDecodedObject(argv[j]);
1370 buf[0] = '$';
1371 len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr));
1372 buf[len++] = '\r';
1373 buf[len++] = '\n';
1374 dst = sdscatlen(dst,buf,len);
1375 dst = sdscatlen(dst,o->ptr,sdslen(o->ptr));
1376 dst = sdscatlen(dst,"\r\n",2);
1377 decrRefCount(o);
1378 }
1379 return dst;
1380}
1381
1382/* Generate a piece of timestamp annotation for AOF if current record timestamp
1383 * in AOF is not equal server unix time. If we specify 'force' argument to 1,
1384 * we would generate one without check, currently, it is useful in AOF rewriting
1385 * child process which always needs to record one timestamp at the beginning of
1386 * rewriting AOF.
1387 *
1388 * Timestamp annotation format is "#TS:${timestamp}\r\n". "TS" is short of
1389 * timestamp and this method could save extra bytes in AOF. */
1390sds genAofTimestampAnnotationIfNeeded(int force) {
1391 sds ts = NULL;
1392
1393 if (force || server.aof_cur_timestamp < server.unixtime) {
1394 server.aof_cur_timestamp = force ? time(NULL) : server.unixtime;
1395 ts = sdscatfmt(sdsempty(), "#TS:%I\r\n", server.aof_cur_timestamp);
1396 serverAssert(sdslen(ts) <= AOF_ANNOTATION_LINE_MAX_LEN);
1397 }
1398 return ts;
1399}
1400
1401/* Write the given command to the aof file.
1402 * dictid - dictionary id the command should be applied to,
1403 * this is used in order to decide if a `select` command
1404 * should also be written to the aof. Value of -1 means
1405 * to avoid writing `select` command in any case.
1406 * argv - The command to write to the aof.
1407 * argc - Number of values in argv
1408 */
1409void feedAppendOnlyFile(int dictid, robj **argv, int argc) {
1410 sds buf = sdsempty();
1411
1412 serverAssert(dictid == -1 || (dictid >= 0 && dictid < server.dbnum));
1413
1414 /* Feed timestamp if needed */
1415 if (server.aof_timestamp_enabled) {
1416 sds ts = genAofTimestampAnnotationIfNeeded(0);
1417 if (ts != NULL) {
1418 buf = sdscatsds(buf, ts);
1419 sdsfree(ts);
1420 }
1421 }
1422
1423 /* The DB this command was targeting is not the same as the last command
1424 * we appended. To issue a SELECT command is needed. */
1425 if (dictid != -1 && dictid != server.aof_selected_db) {
1426 char seldb[64];
1427
1428 snprintf(seldb,sizeof(seldb),"%d",dictid);
1429 buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
1430 (unsigned long)strlen(seldb),seldb);
1431 server.aof_selected_db = dictid;
1432 }
1433
1434 /* All commands should be propagated the same way in AOF as in replication.
1435 * No need for AOF-specific translation. */
1436 buf = catAppendOnlyGenericCommand(buf,argc,argv);
1437
1438 /* Append to the AOF buffer. This will be flushed on disk just before
1439 * of re-entering the event loop, so before the client will get a
1440 * positive reply about the operation performed. */
1441 if (server.aof_state == AOF_ON ||
1442 (server.aof_state == AOF_WAIT_REWRITE && server.child_type == CHILD_TYPE_AOF))
1443 {
1444 server.aof_buf = sdscatlen(server.aof_buf, buf, sdslen(buf));
1445 }
1446
1447 sdsfree(buf);
1448}
1449
1450/* ----------------------------------------------------------------------------
1451 * AOF loading
1452 * ------------------------------------------------------------------------- */
1453
1454/* In Redis commands are always executed in the context of a client, so in
1455 * order to load the append only file we need to create a fake client. */
1456struct client *createAOFClient(void) {
1457 struct client *c = createClient(NULL);
1458
1459 c->id = CLIENT_ID_AOF; /* So modules can identify it's the AOF client. */
1460
1461 /*
1462 * The AOF client should never be blocked (unlike master
1463 * replication connection).
1464 * This is because blocking the AOF client might cause
1465 * deadlock (because potentially no one will unblock it).
1466 * Also, if the AOF client will be blocked just for
1467 * background processing there is a chance that the
1468 * command execution order will be violated.
1469 */
1470 c->flags = CLIENT_DENY_BLOCKING;
1471
1472 /* We set the fake client as a slave waiting for the synchronization
1473 * so that Redis will not try to send replies to this client. */
1474 c->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
1475 return c;
1476}
1477
1478static int truncateAppendOnlyFile(char *filename, off_t valid_up_to) {
1479 if (valid_up_to == -1) {
1480 serverLog(LL_WARNING,"Last valid command offset is invalid");
1481 return 0;
1482 }
1483
1484 if (truncate(filename, valid_up_to) == -1) {
1485 serverLog(LL_WARNING,"Error truncating the AOF file %s: %s",
1486 filename, strerror(errno));
1487 return 0;
1488 }
1489
1490 /* Make sure the AOF file descriptor points to the end of the
1491 * file after the truncate call. */
1492 if (server.aof_fd != -1 && lseek(server.aof_fd, 0, SEEK_END) == -1) {
1493 serverLog(LL_WARNING,"Can't seek the end of the AOF file %s: %s",
1494 filename, strerror(errno));
1495 return 0;
1496 }
1497
1498 return 1; /* Success */
1499}
1500
1501/* Replay an append log file. On success AOF_OK or AOF_TRUNCATED is returned,
1502 * otherwise, one of the following is returned:
1503 * AOF_OPEN_ERR: Failed to open the AOF file.
1504 * AOF_NOT_EXIST: AOF file doesn't exist.
1505 * AOF_EMPTY: The AOF file is empty (nothing to load).
1506 * AOF_FAILED: Failed to load the AOF file. */
1507int loadSingleAppendOnlyFile(char *filename) {
1508 struct client *fakeClient;
1509 struct redis_stat sb;
1510 int old_aof_state = server.aof_state;
1511 long loops = 0;
1512 off_t valid_up_to = 0; /* Offset of latest well-formed command loaded. */
1513 off_t valid_before_multi = 0; /* Offset before MULTI command loaded. */
1514 off_t last_progress_report_size = 0;
1515 int ret = AOF_OK;
1516
1517 sds aof_filepath = makePath(server.aof_dirname, filename);
1518 FILE *fp = fopen(aof_filepath, "r");
1519 if (fp == NULL) {
1520 int en = errno;
1521 if (redis_stat(aof_filepath, &sb) == 0 || errno != ENOENT) {
1522 serverLog(LL_WARNING,"Fatal error: can't open the append log file %s for reading: %s", filename, strerror(en));
1523 sdsfree(aof_filepath);
1524 return AOF_OPEN_ERR;
1525 } else {
1526 serverLog(LL_WARNING,"The append log file %s doesn't exist: %s", filename, strerror(errno));
1527 sdsfree(aof_filepath);
1528 return AOF_NOT_EXIST;
1529 }
1530 }
1531
1532 if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
1533 fclose(fp);
1534 sdsfree(aof_filepath);
1535 return AOF_EMPTY;
1536 }
1537
1538 /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
1539 * to the same file we're about to read. */
1540 server.aof_state = AOF_OFF;
1541
1542 client *old_cur_client = server.current_client;
1543 client *old_exec_client = server.executing_client;
1544 fakeClient = createAOFClient();
1545 server.current_client = server.executing_client = fakeClient;
1546
1547 /* Check if the AOF file is in RDB format (it may be RDB encoded base AOF
1548 * or old style RDB-preamble AOF). In that case we need to load the RDB file
1549 * and later continue loading the AOF tail if it is an old style RDB-preamble AOF. */
1550 char sig[5]; /* "REDIS" */
1551 if (fread(sig,1,5,fp) != 5 || memcmp(sig,"REDIS",5) != 0) {
1552 /* Not in RDB format, seek back at 0 offset. */
1553 if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
1554 } else {
1555 /* RDB format. Pass loading the RDB functions. */
1556 rio rdb;
1557 int old_style = !strcmp(filename, server.aof_filename);
1558 if (old_style)
1559 serverLog(LL_NOTICE, "Reading RDB preamble from AOF file...");
1560 else
1561 serverLog(LL_NOTICE, "Reading RDB base file on AOF loading...");
1562
1563 if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
1564 rioInitWithFile(&rdb,fp);
1565 if (rdbLoadRio(&rdb,RDBFLAGS_AOF_PREAMBLE,NULL) != C_OK) {
1566 if (old_style)
1567 serverLog(LL_WARNING, "Error reading the RDB preamble of the AOF file %s, AOF loading aborted", filename);
1568 else
1569 serverLog(LL_WARNING, "Error reading the RDB base file %s, AOF loading aborted", filename);
1570
1571 ret = AOF_FAILED;
1572 goto cleanup;
1573 } else {
1574 loadingAbsProgress(ftello(fp));
1575 last_progress_report_size = ftello(fp);
1576 if (old_style) serverLog(LL_NOTICE, "Reading the remaining AOF tail...");
1577 }
1578 }
1579
1580 /* Read the actual AOF file, in REPL format, command by command. */
1581 while(1) {
1582 int argc, j;
1583 unsigned long len;
1584 robj **argv;
1585 char buf[AOF_ANNOTATION_LINE_MAX_LEN];
1586 sds argsds;
1587 struct redisCommand *cmd;
1588
1589 /* Serve the clients from time to time */
1590 if (!(loops++ % 1024)) {
1591 off_t progress_delta = ftello(fp) - last_progress_report_size;
1592 loadingIncrProgress(progress_delta);
1593 last_progress_report_size += progress_delta;
1594 processEventsWhileBlocked();
1595 processModuleLoadingProgressEvent(1);
1596 }
1597 if (fgets(buf,sizeof(buf),fp) == NULL) {
1598 if (feof(fp)) {
1599 break;
1600 } else {
1601 goto readerr;
1602 }
1603 }
1604 if (buf[0] == '#') continue; /* Skip annotations */
1605 if (buf[0] != '*') goto fmterr;
1606 if (buf[1] == '\0') goto readerr;
1607 argc = atoi(buf+1);
1608 if (argc < 1) goto fmterr;
1609 if ((size_t)argc > SIZE_MAX / sizeof(robj*)) goto fmterr;
1610
1611 /* Load the next command in the AOF as our fake client
1612 * argv. */
1613 argv = zmalloc(sizeof(robj*)*argc);
1614 fakeClient->argc = argc;
1615 fakeClient->argv = argv;
1616 fakeClient->argv_len = argc;
1617
1618 for (j = 0; j < argc; j++) {
1619 /* Parse the argument len. */
1620 char *readres = fgets(buf,sizeof(buf),fp);
1621 if (readres == NULL || buf[0] != '$') {
1622 fakeClient->argc = j; /* Free up to j-1. */
1623 freeClientArgv(fakeClient);
1624 if (readres == NULL)
1625 goto readerr;
1626 else
1627 goto fmterr;
1628 }
1629 len = strtol(buf+1,NULL,10);
1630
1631 /* Read it into a string object. */
1632 argsds = sdsnewlen(SDS_NOINIT,len);
1633 if (len && fread(argsds,len,1,fp) == 0) {
1634 sdsfree(argsds);
1635 fakeClient->argc = j; /* Free up to j-1. */
1636 freeClientArgv(fakeClient);
1637 goto readerr;
1638 }
1639 argv[j] = createObject(OBJ_STRING,argsds);
1640
1641 /* Discard CRLF. */
1642 if (fread(buf,2,1,fp) == 0) {
1643 fakeClient->argc = j+1; /* Free up to j. */
1644 freeClientArgv(fakeClient);
1645 goto readerr;
1646 }
1647 }
1648
1649 /* Command lookup */
1650 cmd = lookupCommand(argv,argc);
1651 if (!cmd) {
1652 serverLog(LL_WARNING,
1653 "Unknown command '%s' reading the append only file %s",
1654 (char*)argv[0]->ptr, filename);
1655 freeClientArgv(fakeClient);
1656 ret = AOF_FAILED;
1657 goto cleanup;
1658 }
1659
1660 if (cmd->proc == multiCommand) valid_before_multi = valid_up_to;
1661
1662 /* Run the command in the context of a fake client */
1663 fakeClient->cmd = fakeClient->lastcmd = cmd;
1664 if (fakeClient->flags & CLIENT_MULTI &&
1665 fakeClient->cmd->proc != execCommand)
1666 {
1667 /* queueMultiCommand requires a pendingCommand, so we create a "fake" one here
1668 * for it to consume */
1669 pendingCommand *pcmd = zmalloc(sizeof(pendingCommand));
1670 initPendingCommand(pcmd);
1671 addPendingCommand(&fakeClient->pending_cmds, pcmd);
1672
1673 pcmd->argc = argc;
1674 pcmd->argv_len = argc;
1675 pcmd->argv = argv;
1676 pcmd->cmd = cmd;
1677
1678 /* Note: we don't have to attempt calling evalGetCommandFlags,
1679 * since this is AOF, the checks in processCommand are not made
1680 * anyway.*/
1681 queueMultiCommand(fakeClient, cmd->flags);
1682 } else {
1683 cmd->proc(fakeClient);
1684 fakeClient->all_argv_len_sum = 0; /* Otherwise no one cleans this up and we reach cleanup with it non-zero */
1685 }
1686
1687 /* The fake client should not have a reply */
1688 serverAssert(fakeClient->bufpos == 0 &&
1689 listLength(fakeClient->reply) == 0);
1690
1691 /* The fake client should never get blocked */
1692 serverAssert((fakeClient->flags & CLIENT_BLOCKED) == 0);
1693
1694 /* Clean up. Command code may have changed argv/argc so we use the
1695 * argv/argc of the client instead of the local variables. */
1696 freeClientArgv(fakeClient);
1697 if (server.aof_load_truncated || server.aof_load_corrupt_tail_max_size) valid_up_to = ftello(fp);
1698 if (server.key_load_delay)
1699 debugDelay(server.key_load_delay);
1700 }
1701
1702 /* This point can only be reached when EOF is reached without errors.
1703 * If the client is in the middle of a MULTI/EXEC, handle it as it was
1704 * a short read, even if technically the protocol is correct: we want
1705 * to remove the unprocessed tail and continue. */
1706 if (fakeClient->flags & CLIENT_MULTI) {
1707 serverLog(LL_WARNING,
1708 "Revert incomplete MULTI/EXEC transaction in AOF file %s", filename);
1709 valid_up_to = valid_before_multi;
1710 goto uxeof;
1711 }
1712
1713loaded_ok: /* DB loaded, cleanup and return success (AOF_OK or AOF_TRUNCATED). */
1714 loadingIncrProgress(ftello(fp) - last_progress_report_size);
1715 server.aof_state = old_aof_state;
1716 goto cleanup;
1717
1718readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */
1719 if (!feof(fp)) {
1720 serverLog(LL_WARNING,"Unrecoverable error reading the append only file %s: %s", filename, strerror(errno));
1721 ret = AOF_FAILED;
1722 goto cleanup;
1723 }
1724
1725uxeof: /* Unexpected AOF end of file. */
1726 if (server.aof_load_truncated) {
1727 serverLog(LL_WARNING,"!!! Warning: short read while loading the AOF file %s!!!", filename);
1728 serverLog(LL_WARNING,"!!! Truncating the AOF %s at offset %llu !!!",
1729 filename, (unsigned long long) valid_up_to);
1730 if (truncateAppendOnlyFile(aof_filepath, valid_up_to)) {
1731 serverLog(LL_WARNING, "AOF %s loaded anyway because aof-load-truncated is enabled", aof_filepath);
1732 ret = AOF_TRUNCATED;
1733 goto loaded_ok;
1734 }
1735 }
1736 serverLog(LL_WARNING, "Unexpected end of file reading the append only file %s. You can: "
1737 "1) Make a backup of your AOF file, then use ./redis-check-aof --fix <filename.manifest>. "
1738 "2) Alternatively you can set the 'aof-load-truncated' configuration option to yes and restart the server.", filename);
1739 ret = AOF_FAILED;
1740 goto cleanup;
1741
1742fmterr: /* Format error. */
1743 /* fmterr may be caused by accidentally machine shutdown, so if the broken tail
1744 * is less than a specified size, try to recover it automatically */
1745 if (server.aof_load_corrupt_tail_max_size && sb.st_size - valid_up_to < server.aof_load_corrupt_tail_max_size) {
1746 serverLog(LL_WARNING,"!!! Warning: corrupt AOF file tail!!!");
1747 serverLog(LL_WARNING,"!!! Truncating the AOF %s at offset %llu (remaining %llu) !!!",
1748 aof_filepath, (unsigned long long) valid_up_to, (unsigned long long) sb.st_size - valid_up_to);
1749 if (truncateAppendOnlyFile(aof_filepath, valid_up_to)) {
1750 serverLog(LL_WARNING, "AOF %s loaded anyway because aof-load-corrupt-tail-max-size is enabled", aof_filepath);
1751 ret = AOF_BROKEN_RECOVERED;
1752 goto loaded_ok;
1753 }
1754 }
1755 serverLog(LL_WARNING, "Bad file format reading the append only file %s at offset %llu. \
1756 make a backup of your AOF file, then use ./redis-check-aof --fix <filename.manifest>. \
1757 Alternatively you can set the 'aof-load-corrupt-tail-max-size' configuration option to %llu and restart the server.",
1758 aof_filepath, (unsigned long long)valid_up_to, (unsigned long long) sb.st_size - valid_up_to);
1759 ret = AOF_FAILED;
1760 /* fall through to cleanup. */
1761
1762cleanup:
1763 if (fakeClient) freeClient(fakeClient);
1764 server.current_client = old_cur_client;
1765 server.executing_client = old_exec_client;
1766 int fd = dup(fileno(fp));
1767 fclose(fp);
1768 /* Reclaim page cache memory used by the AOF file in background. */
1769 if (fd >= 0) bioCreateCloseJob(fd, 0, 1);
1770 sdsfree(aof_filepath);
1771 return ret;
1772}
1773
1774/* Load the AOF files according the aofManifest pointed by am. */
1775int loadAppendOnlyFiles(aofManifest *am) {
1776 serverAssert(am != NULL);
1777 int status, ret = AOF_OK;
1778 long long start;
1779 off_t total_size = 0, base_size = 0;
1780 sds aof_name;
1781 int total_num, aof_num = 0, last_file;
1782
1783 /* If the 'server.aof_filename' file exists in dir, we may be starting
1784 * from an old redis version. We will use enter upgrade mode in three situations.
1785 *
1786 * 1. If the 'server.aof_dirname' directory not exist
1787 * 2. If the 'server.aof_dirname' directory exists but the manifest file is missing
1788 * 3. If the 'server.aof_dirname' directory exists and the manifest file it contains
1789 * has only one base AOF record, and the file name of this base AOF is 'server.aof_filename',
1790 * and the 'server.aof_filename' file not exist in 'server.aof_dirname' directory
1791 * */
1792 if (fileExist(server.aof_filename)) {
1793 if (!dirExists(server.aof_dirname) ||
1794 (am->base_aof_info == NULL && listLength(am->incr_aof_list) == 0) ||
1795 (am->base_aof_info != NULL && listLength(am->incr_aof_list) == 0 &&
1796 !strcmp(am->base_aof_info->file_name, server.aof_filename) && !aofFileExist(server.aof_filename)))
1797 {
1798 aofUpgradePrepare(am);
1799 }
1800 }
1801
1802 if (am->base_aof_info == NULL && listLength(am->incr_aof_list) == 0) {
1803 return AOF_NOT_EXIST;
1804 }
1805
1806 total_num = getBaseAndIncrAppendOnlyFilesNum(am);
1807 serverAssert(total_num > 0);
1808
1809 /* Here we calculate the total size of all BASE and INCR files in
1810 * advance, it will be set to `server.loading_total_bytes`. */
1811 total_size = getBaseAndIncrAppendOnlyFilesSize(am, &status);
1812 if (status != AOF_OK) {
1813 /* If an AOF exists in the manifest but not on the disk, we consider this to be a fatal error. */
1814 if (status == AOF_NOT_EXIST) status = AOF_FAILED;
1815
1816 return status;
1817 } else if (total_size == 0) {
1818 return AOF_EMPTY;
1819 }
1820
1821 startLoading(total_size, RDBFLAGS_AOF_PREAMBLE, 0);
1822
1823 /* Load BASE AOF if needed. */
1824 if (am->base_aof_info) {
1825 serverAssert(am->base_aof_info->file_type == AOF_FILE_TYPE_BASE);
1826 aof_name = (char*)am->base_aof_info->file_name;
1827 updateLoadingFileName(aof_name);
1828 base_size = getAppendOnlyFileSize(aof_name, NULL);
1829 last_file = ++aof_num == total_num;
1830 start = ustime();
1831 ret = loadSingleAppendOnlyFile(aof_name);
1832 if (ret == AOF_OK || ((ret == AOF_TRUNCATED || ret == AOF_BROKEN_RECOVERED) && last_file)) {
1833 serverLog(LL_NOTICE, "DB loaded from base file %s: %.3f seconds",
1834 aof_name, (float)(ustime()-start)/1000000);
1835 }
1836
1837 /* If the truncated file is not the last file, we consider this to be a fatal error. */
1838 if ((ret == AOF_TRUNCATED || ret == AOF_BROKEN_RECOVERED) && !last_file) {
1839 ret = AOF_FAILED;
1840 serverLog(LL_WARNING, "Fatal error: the truncated file is not the last file");
1841 }
1842
1843 if (ret == AOF_OPEN_ERR || ret == AOF_FAILED) {
1844 goto cleanup;
1845 }
1846 }
1847
1848 /* Load INCR AOFs if needed. */
1849 if (listLength(am->incr_aof_list)) {
1850 listNode *ln;
1851 listIter li;
1852
1853 listRewind(am->incr_aof_list, &li);
1854 while ((ln = listNext(&li)) != NULL) {
1855 aofInfo *ai = (aofInfo*)ln->value;
1856 serverAssert(ai->file_type == AOF_FILE_TYPE_INCR);
1857 aof_name = (char*)ai->file_name;
1858 updateLoadingFileName(aof_name);
1859 last_file = ++aof_num == total_num;
1860 start = ustime();
1861 ret = loadSingleAppendOnlyFile(aof_name);
1862 if (ret == AOF_OK || ((ret == AOF_TRUNCATED || ret == AOF_BROKEN_RECOVERED) && last_file)) {
1863 serverLog(LL_NOTICE, "DB loaded from incr file %s: %.3f seconds",
1864 aof_name, (float)(ustime()-start)/1000000);
1865 }
1866
1867 /* We know that (at least) one of the AOF files has data (total_size > 0),
1868 * so empty incr AOF file doesn't count as a AOF_EMPTY result */
1869 if (ret == AOF_EMPTY) ret = AOF_OK;
1870
1871 /* If the truncated file is not the last file, we consider this to be a fatal error. */
1872 if ((ret == AOF_TRUNCATED || ret == AOF_BROKEN_RECOVERED) && !last_file) {
1873 ret = AOF_FAILED;
1874 serverLog(LL_WARNING, "Fatal error: the truncated file is not the last file");
1875 }
1876
1877 if (ret == AOF_OPEN_ERR || ret == AOF_FAILED) {
1878 goto cleanup;
1879 }
1880 }
1881 }
1882
1883 server.aof_current_size = total_size;
1884 /* Ideally, the aof_rewrite_base_size variable should hold the size of the
1885 * AOF when the last rewrite ended, this should include the size of the
1886 * incremental file that was created during the rewrite since otherwise we
1887 * risk the next automatic rewrite to happen too soon (or immediately if
1888 * auto-aof-rewrite-percentage is low). However, since we do not persist
1889 * aof_rewrite_base_size information anywhere, we initialize it on restart
1890 * to the size of BASE AOF file. This might cause the first AOFRW to be
1891 * executed early, but that shouldn't be a problem since everything will be
1892 * fine after the first AOFRW. */
1893 server.aof_rewrite_base_size = base_size;
1894
1895cleanup:
1896 stopLoading(ret == AOF_OK || ret == AOF_TRUNCATED);
1897 return ret;
1898}
1899
1900/* ----------------------------------------------------------------------------
1901 * AOF rewrite
1902 * ------------------------------------------------------------------------- */
1903
1904/* Delegate writing an object to writing a bulk string or bulk long long.
1905 * This is not placed in rio.c since that adds the server.h dependency. */
1906int rioWriteBulkObject(rio *r, robj *obj) {
1907 /* Avoid using getDecodedObject to help copy-on-write (we are often
1908 * in a child process when this function is called). */
1909 if (obj->encoding == OBJ_ENCODING_INT) {
1910 return rioWriteBulkLongLong(r,(long)obj->ptr);
1911 } else if (sdsEncodedObject(obj)) {
1912 return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr));
1913 } else {
1914 serverPanic("Unknown string encoding");
1915 }
1916}
1917
1918/* Emit the commands needed to rebuild a list object.
1919 * The function returns 0 on error, 1 on success. */
1920int rewriteListObject(rio *r, robj *key, robj *o) {
1921 long long count = 0, items = listTypeLength(o);
1922
1923 listTypeIterator li;
1924 listTypeEntry entry;
1925 listTypeInitIterator(&li, o, 0, LIST_TAIL);
1926 while (listTypeNext(&li, &entry)) {
1927 if (count == 0) {
1928 int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
1929 AOF_REWRITE_ITEMS_PER_CMD : items;
1930 if (!rioWriteBulkCount(r,'*',2+cmd_items) ||
1931 !rioWriteBulkString(r,"RPUSH",5) ||
1932 !rioWriteBulkObject(r,key))
1933 {
1934 listTypeResetIterator(&li);
1935 return 0;
1936 }
1937 }
1938
1939 unsigned char *vstr;
1940 size_t vlen;
1941 long long lval;
1942 vstr = listTypeGetValue(&entry,&vlen,&lval);
1943 if (vstr) {
1944 if (!rioWriteBulkString(r,(char*)vstr,vlen)) {
1945 listTypeResetIterator(&li);
1946 return 0;
1947 }
1948 } else {
1949 if (!rioWriteBulkLongLong(r,lval)) {
1950 listTypeResetIterator(&li);
1951 return 0;
1952 }
1953 }
1954 if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
1955 items--;
1956 }
1957 listTypeResetIterator(&li);
1958 return 1;
1959}
1960
1961/* Emit the commands needed to rebuild a set object.
1962 * The function returns 0 on error, 1 on success. */
1963int rewriteSetObject(rio *r, robj *key, robj *o) {
1964 long long count = 0, items = setTypeSize(o);
1965 setTypeIterator si;
1966 char *str;
1967 size_t len;
1968 int64_t llval;
1969 setTypeInitIterator(&si, o);
1970 while (setTypeNext(&si, &str, &len, &llval) != -1) {
1971 if (count == 0) {
1972 int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
1973 AOF_REWRITE_ITEMS_PER_CMD : items;
1974 if (!rioWriteBulkCount(r,'*',2+cmd_items) ||
1975 !rioWriteBulkString(r,"SADD",4) ||
1976 !rioWriteBulkObject(r,key))
1977 {
1978 setTypeResetIterator(&si);
1979 return 0;
1980 }
1981 }
1982 size_t written = str ?
1983 rioWriteBulkString(r, str, len) : rioWriteBulkLongLong(r, llval);
1984 if (!written) {
1985 setTypeResetIterator(&si);
1986 return 0;
1987 }
1988 if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
1989 items--;
1990 }
1991 setTypeResetIterator(&si);
1992 return 1;
1993}
1994
1995/* Emit the commands needed to rebuild a sorted set object.
1996 * The function returns 0 on error, 1 on success. */
1997int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
1998 long long count = 0, items = zsetLength(o);
1999
2000 if (o->encoding == OBJ_ENCODING_LISTPACK) {
2001 unsigned char *zl = o->ptr;
2002 unsigned char *eptr, *sptr;
2003 unsigned char *vstr;
2004 unsigned int vlen;
2005 long long vll;
2006 double score;
2007
2008 eptr = lpSeek(zl,0);
2009 serverAssert(eptr != NULL);
2010 sptr = lpNext(zl,eptr);
2011 serverAssert(sptr != NULL);
2012
2013 while (eptr != NULL) {
2014 vstr = lpGetValue(eptr,&vlen,&vll);
2015 score = zzlGetScore(sptr);
2016
2017 if (count == 0) {
2018 int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
2019 AOF_REWRITE_ITEMS_PER_CMD : items;
2020
2021 if (!rioWriteBulkCount(r,'*',2+cmd_items*2) ||
2022 !rioWriteBulkString(r,"ZADD",4) ||
2023 !rioWriteBulkObject(r,key))
2024 {
2025 return 0;
2026 }
2027 }
2028 if (!rioWriteBulkDouble(r,score)) return 0;
2029 if (vstr != NULL) {
2030 if (!rioWriteBulkString(r,(char*)vstr,vlen)) return 0;
2031 } else {
2032 if (!rioWriteBulkLongLong(r,vll)) return 0;
2033 }
2034 zzlNext(zl,&eptr,&sptr);
2035 if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
2036 items--;
2037 }
2038 } else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
2039 zset *zs = o->ptr;
2040 dictIterator di;
2041 dictEntry *de;
2042
2043 dictInitIterator(&di, zs->dict);
2044 while((de = dictNext(&di)) != NULL) {
2045 zskiplistNode *znode = dictGetKey(de);
2046 sds ele = zslGetNodeElement(znode);
2047 double score = znode->score;
2048
2049 if (count == 0) {
2050 int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
2051 AOF_REWRITE_ITEMS_PER_CMD : items;
2052
2053 if (!rioWriteBulkCount(r,'*',2+cmd_items*2) ||
2054 !rioWriteBulkString(r,"ZADD",4) ||
2055 !rioWriteBulkObject(r,key))
2056 {
2057 dictResetIterator(&di);
2058 return 0;
2059 }
2060 }
2061 if (!rioWriteBulkDouble(r,score) ||
2062 !rioWriteBulkString(r,ele,sdslen(ele)))
2063 {
2064 dictResetIterator(&di);
2065 return 0;
2066 }
2067 if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
2068 items--;
2069 }
2070 dictResetIterator(&di);
2071 } else {
2072 serverPanic("Unknown sorted zset encoding");
2073 }
2074 return 1;
2075}
2076
2077/* Write either the key or the value of the currently selected item of a hash.
2078 * The 'hi' argument passes a valid Redis hash iterator.
2079 * The 'what' filed specifies if to write a key or a value and can be
2080 * either OBJ_HASH_KEY or OBJ_HASH_VALUE.
2081 *
2082 * The function returns 0 on error, non-zero on success. */
2083static int rioWriteHashIteratorCursor(rio *r, hashTypeIterator *hi, int what) {
2084 if ((hi->encoding == OBJ_ENCODING_LISTPACK) || (hi->encoding == OBJ_ENCODING_LISTPACK_EX)) {
2085 unsigned char *vstr = NULL;
2086 unsigned int vlen = UINT_MAX;
2087 long long vll = LLONG_MAX;
2088
2089 hashTypeCurrentFromListpack(hi, what, &vstr, &vlen, &vll, NULL);
2090 if (vstr)
2091 return rioWriteBulkString(r, (char*)vstr, vlen);
2092 else
2093 return rioWriteBulkLongLong(r, vll);
2094 } else if (hi->encoding == OBJ_ENCODING_HT) {
2095 char *str;
2096 size_t len;
2097 hashTypeCurrentFromHashTable(hi, what, &str, &len, NULL);
2098 return rioWriteBulkString(r, str, len);
2099 }
2100
2101 serverPanic("Unknown hash encoding");
2102 return 0;
2103}
2104
2105/* Emit the commands needed to rebuild a hash object.
2106 * The function returns 0 on error, 1 on success. */
2107int rewriteHashObject(rio *r, robj *key, robj *o) {
2108 int res = 0; /*fail*/
2109
2110 hashTypeIterator hi;
2111 long long count = 0, items = hashTypeLength(o, 0);
2112
2113 int isHFE = hashTypeGetMinExpire(o, 0) != EB_EXPIRE_TIME_INVALID;
2114 hashTypeInitIterator(&hi, o);
2115
2116 if (!isHFE) {
2117 while (hashTypeNext(&hi, 0) != C_ERR) {
2118 if (count == 0) {
2119 int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
2120 AOF_REWRITE_ITEMS_PER_CMD : items;
2121 if (!rioWriteBulkCount(r, '*', 2 + cmd_items * 2) ||
2122 !rioWriteBulkString(r, "HMSET", 5) ||
2123 !rioWriteBulkObject(r, key))
2124 goto reHashEnd;
2125 }
2126
2127 if (!rioWriteHashIteratorCursor(r, &hi, OBJ_HASH_KEY) ||
2128 !rioWriteHashIteratorCursor(r, &hi, OBJ_HASH_VALUE))
2129 goto reHashEnd;
2130
2131 if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
2132 items--;
2133 }
2134 } else {
2135 while (hashTypeNext(&hi, 0) != C_ERR) {
2136
2137 char hmsetCmd[] = "*4\r\n$5\r\nHMSET\r\n";
2138 if ( (!rioWrite(r, hmsetCmd, sizeof(hmsetCmd) - 1)) ||
2139 (!rioWriteBulkObject(r, key)) ||
2140 (!rioWriteHashIteratorCursor(r, &hi, OBJ_HASH_KEY)) ||
2141 (!rioWriteHashIteratorCursor(r, &hi, OBJ_HASH_VALUE)) )
2142 goto reHashEnd;
2143
2144 if (hi.expire_time != EB_EXPIRE_TIME_INVALID) {
2145 char cmd[] = "*6\r\n$10\r\nHPEXPIREAT\r\n";
2146 if ( (!rioWrite(r, cmd, sizeof(cmd) - 1)) ||
2147 (!rioWriteBulkObject(r, key)) ||
2148 (!rioWriteBulkLongLong(r, hi.expire_time)) ||
2149 (!rioWriteBulkString(r, "FIELDS", 6)) ||
2150 (!rioWriteBulkString(r, "1", 1)) ||
2151 (!rioWriteHashIteratorCursor(r, &hi, OBJ_HASH_KEY)) )
2152 goto reHashEnd;
2153 }
2154 }
2155 }
2156
2157 res = 1; /* success */
2158
2159reHashEnd:
2160 hashTypeResetIterator(&hi);
2161 return res;
2162}
2163
2164/* Helper for rewriteStreamObject() that generates a bulk string into the
2165 * AOF representing the ID 'id'. */
2166int rioWriteBulkStreamID(rio *r,streamID *id) {
2167 int retval;
2168
2169 sds replyid = sdscatfmt(sdsempty(),"%U-%U",id->ms,id->seq);
2170 retval = rioWriteBulkString(r,replyid,sdslen(replyid));
2171 sdsfree(replyid);
2172 return retval;
2173}
2174
2175/* Helper for rewriteStreamObject(): emit the XCLAIM needed in order to
2176 * add the message described by 'nack' having the id 'rawid', into the pending
2177 * list of the specified consumer. All this in the context of the specified
2178 * key and group. */
2179int rioWriteStreamPendingEntry(rio *r, robj *key, const char *groupname, size_t groupname_len, streamConsumer *consumer, unsigned char *rawid, streamNACK *nack) {
2180 /* XCLAIM <key> <group> <consumer> 0 <id> TIME <milliseconds-unix-time>
2181 RETRYCOUNT <count> JUSTID FORCE. */
2182 streamID id;
2183 streamDecodeID(rawid,&id);
2184 if (rioWriteBulkCount(r,'*',12) == 0) return 0;
2185 if (rioWriteBulkString(r,"XCLAIM",6) == 0) return 0;
2186 if (rioWriteBulkObject(r,key) == 0) return 0;
2187 if (rioWriteBulkString(r,groupname,groupname_len) == 0) return 0;
2188 if (rioWriteBulkString(r,consumer->name,sdslen(consumer->name)) == 0) return 0;
2189 if (rioWriteBulkString(r,"0",1) == 0) return 0;
2190 if (rioWriteBulkStreamID(r,&id) == 0) return 0;
2191 if (rioWriteBulkString(r,"TIME",4) == 0) return 0;
2192 if (rioWriteBulkLongLong(r,nack->delivery_time) == 0) return 0;
2193 if (rioWriteBulkString(r,"RETRYCOUNT",10) == 0) return 0;
2194 if (rioWriteBulkLongLong(r,nack->delivery_count) == 0) return 0;
2195 if (rioWriteBulkString(r,"JUSTID",6) == 0) return 0;
2196 if (rioWriteBulkString(r,"FORCE",5) == 0) return 0;
2197 return 1;
2198}
2199
2200/* Helper for rewriteStreamObject(): emit the XGROUP CREATECONSUMER is
2201 * needed in order to create consumers that do not have any pending entries.
2202 * All this in the context of the specified key and group. */
2203int rioWriteStreamEmptyConsumer(rio *r, robj *key, const char *groupname, size_t groupname_len, streamConsumer *consumer) {
2204 /* XGROUP CREATECONSUMER <key> <group> <consumer> */
2205 if (rioWriteBulkCount(r,'*',5) == 0) return 0;
2206 if (rioWriteBulkString(r,"XGROUP",6) == 0) return 0;
2207 if (rioWriteBulkString(r,"CREATECONSUMER",14) == 0) return 0;
2208 if (rioWriteBulkObject(r,key) == 0) return 0;
2209 if (rioWriteBulkString(r,groupname,groupname_len) == 0) return 0;
2210 if (rioWriteBulkString(r,consumer->name,sdslen(consumer->name)) == 0) return 0;
2211 return 1;
2212}
2213
2214/* Emit the commands needed to rebuild a stream object.
2215 * The function returns 0 on error, 1 on success. */
2216int rewriteStreamObject(rio *r, robj *key, robj *o) {
2217 stream *s = o->ptr;
2218 streamIterator si;
2219 streamIteratorStart(&si,s,NULL,NULL,0);
2220 streamID id;
2221 int64_t numfields;
2222
2223 if (s->length) {
2224 /* Reconstruct the stream data using XADD commands. */
2225 while(streamIteratorGetID(&si,&id,&numfields)) {
2226 /* Emit a two elements array for each item. The first is
2227 * the ID, the second is an array of field-value pairs. */
2228
2229 /* Emit the XADD <key> <id> ...fields... command. */
2230 if (!rioWriteBulkCount(r,'*',3+numfields*2) ||
2231 !rioWriteBulkString(r,"XADD",4) ||
2232 !rioWriteBulkObject(r,key) ||
2233 !rioWriteBulkStreamID(r,&id))
2234 {
2235 streamIteratorStop(&si);
2236 return 0;
2237 }
2238 while(numfields--) {
2239 unsigned char *field, *value;
2240 int64_t field_len, value_len;
2241 streamIteratorGetField(&si,&field,&value,&field_len,&value_len);
2242 if (!rioWriteBulkString(r,(char*)field,field_len) ||
2243 !rioWriteBulkString(r,(char*)value,value_len))
2244 {
2245 streamIteratorStop(&si);
2246 return 0;
2247 }
2248 }
2249 }
2250 } else {
2251 /* Use the XADD MAXLEN 0 trick to generate an empty stream if
2252 * the key we are serializing is an empty string, which is possible
2253 * for the Stream type. */
2254 id.ms = 0; id.seq = 1;
2255 if (!rioWriteBulkCount(r,'*',7) ||
2256 !rioWriteBulkString(r,"XADD",4) ||
2257 !rioWriteBulkObject(r,key) ||
2258 !rioWriteBulkString(r,"MAXLEN",6) ||
2259 !rioWriteBulkString(r,"0",1) ||
2260 !rioWriteBulkStreamID(r,&id) ||
2261 !rioWriteBulkString(r,"x",1) ||
2262 !rioWriteBulkString(r,"y",1))
2263 {
2264 streamIteratorStop(&si);
2265 return 0;
2266 }
2267 }
2268
2269 /* Append XSETID after XADD, make sure lastid is correct,
2270 * in case of XDEL lastid. */
2271 if (!rioWriteBulkCount(r,'*',7) ||
2272 !rioWriteBulkString(r,"XSETID",6) ||
2273 !rioWriteBulkObject(r,key) ||
2274 !rioWriteBulkStreamID(r,&s->last_id) ||
2275 !rioWriteBulkString(r,"ENTRIESADDED",12) ||
2276 !rioWriteBulkLongLong(r,s->entries_added) ||
2277 !rioWriteBulkString(r,"MAXDELETEDID",12) ||
2278 !rioWriteBulkStreamID(r,&s->max_deleted_entry_id))
2279 {
2280 streamIteratorStop(&si);
2281 return 0;
2282 }
2283
2284
2285 /* Create all the stream consumer groups. */
2286 if (s->cgroups) {
2287 raxIterator ri;
2288 raxStart(&ri,s->cgroups);
2289 raxSeek(&ri,"^",NULL,0);
2290 while(raxNext(&ri)) {
2291 streamCG *group = ri.data;
2292 /* Emit the XGROUP CREATE in order to create the group. */
2293 if (!rioWriteBulkCount(r,'*',7) ||
2294 !rioWriteBulkString(r,"XGROUP",6) ||
2295 !rioWriteBulkString(r,"CREATE",6) ||
2296 !rioWriteBulkObject(r,key) ||
2297 !rioWriteBulkString(r,(char*)ri.key,ri.key_len) ||
2298 !rioWriteBulkStreamID(r,&group->last_id) ||
2299 !rioWriteBulkString(r,"ENTRIESREAD",11) ||
2300 !rioWriteBulkLongLong(r,group->entries_read))
2301 {
2302 raxStop(&ri);
2303 streamIteratorStop(&si);
2304 return 0;
2305 }
2306
2307 /* Generate XCLAIMs for each consumer that happens to
2308 * have pending entries. Empty consumers would be generated with
2309 * XGROUP CREATECONSUMER. */
2310 raxIterator ri_cons;
2311 raxStart(&ri_cons,group->consumers);
2312 raxSeek(&ri_cons,"^",NULL,0);
2313 while(raxNext(&ri_cons)) {
2314 streamConsumer *consumer = ri_cons.data;
2315 /* If there are no pending entries, just emit XGROUP CREATECONSUMER */
2316 if (raxSize(consumer->pel) == 0) {
2317 if (rioWriteStreamEmptyConsumer(r,key,(char*)ri.key,
2318 ri.key_len,consumer) == 0)
2319 {
2320 raxStop(&ri_cons);
2321 raxStop(&ri);
2322 streamIteratorStop(&si);
2323 return 0;
2324 }
2325 continue;
2326 }
2327 /* For the current consumer, iterate all the PEL entries
2328 * to emit the XCLAIM protocol. */
2329 raxIterator ri_pel;
2330 raxStart(&ri_pel,consumer->pel);
2331 raxSeek(&ri_pel,"^",NULL,0);
2332 while(raxNext(&ri_pel)) {
2333 streamNACK *nack = ri_pel.data;
2334 if (rioWriteStreamPendingEntry(r,key,(char*)ri.key,
2335 ri.key_len,consumer,
2336 ri_pel.key,nack) == 0)
2337 {
2338 raxStop(&ri_pel);
2339 raxStop(&ri_cons);
2340 raxStop(&ri);
2341 streamIteratorStop(&si);
2342 return 0;
2343 }
2344 }
2345 raxStop(&ri_pel);
2346 }
2347 raxStop(&ri_cons);
2348 }
2349 raxStop(&ri);
2350 }
2351
2352 streamIteratorStop(&si);
2353 return 1;
2354}
2355
2356/* Call the module type callback in order to rewrite a data type
2357 * that is exported by a module and is not handled by Redis itself.
2358 * The function returns 0 on error, 1 on success. */
2359int rewriteModuleObject(rio *r, robj *key, robj *o, int dbid) {
2360 RedisModuleIO io;
2361 moduleValue *mv = o->ptr;
2362 moduleType *mt = mv->type;
2363 moduleInitIOContext(&io, &mt->entity, r, key, dbid);
2364 mt->aof_rewrite(&io,key,mv->value);
2365 if (io.ctx) {
2366 moduleFreeContext(io.ctx);
2367 zfree(io.ctx);
2368 }
2369 return io.error ? 0 : 1;
2370}
2371
2372static int rewriteFunctions(rio *aof) {
2373 dict *functions = functionsLibGet();
2374 dictIterator iter;
2375 dictEntry *entry = NULL;
2376 dictInitIterator(&iter, functions);
2377 while ((entry = dictNext(&iter))) {
2378 functionLibInfo *li = dictGetVal(entry);
2379 if (rioWrite(aof, "*3\r\n", 4) == 0) goto werr;
2380 char function_load[] = "$8\r\nFUNCTION\r\n$4\r\nLOAD\r\n";
2381 if (rioWrite(aof, function_load, sizeof(function_load) - 1) == 0) goto werr;
2382 if (rioWriteBulkString(aof, li->code, sdslen(li->code)) == 0) goto werr;
2383 }
2384 dictResetIterator(&iter);
2385 return 1;
2386
2387werr:
2388 dictResetIterator(&iter);
2389 return 0;
2390}
2391
2392int rewriteObject(rio *r, robj *key, robj *o, int dbid, long long expiretime) {
2393 /* Save the key and associated value */
2394 if (o->type == OBJ_STRING) {
2395 /* Emit a SET command */
2396 static const char cmd[]="*3\r\n$3\r\nSET\r\n";
2397 if (rioWrite(r,cmd,sizeof(cmd)-1) == 0) return C_ERR;
2398 /* Key and value */
2399 if (rioWriteBulkObject(r,key) == 0) return C_ERR;
2400 if (rioWriteBulkObject(r,o) == 0) return C_ERR;
2401 } else if (o->type == OBJ_LIST) {
2402 if (rewriteListObject(r,key,o) == 0) return C_ERR;
2403 } else if (o->type == OBJ_SET) {
2404 if (rewriteSetObject(r,key,o) == 0) return C_ERR;
2405 } else if (o->type == OBJ_ZSET) {
2406 if (rewriteSortedSetObject(r,key,o) == 0) return C_ERR;
2407 } else if (o->type == OBJ_HASH) {
2408 if (rewriteHashObject(r,key,o) == 0) return C_ERR;
2409 } else if (o->type == OBJ_STREAM) {
2410 if (rewriteStreamObject(r,key,o) == 0) return C_ERR;
2411 } else if (o->type == OBJ_MODULE) {
2412 if (rewriteModuleObject(r,key,o,dbid) == 0) return C_ERR;
2413 } else {
2414 serverPanic("Unknown object type");
2415 }
2416
2417 /* Save the expire time */
2418 if (expiretime != -1) {
2419 static const char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
2420 if (rioWrite(r,cmd,sizeof(cmd)-1) == 0) return C_ERR;
2421 if (rioWriteBulkObject(r,key) == 0) return C_ERR;
2422 if (rioWriteBulkLongLong(r,expiretime) == 0) return C_ERR;
2423 }
2424
2425 /* If modules metadata is available */
2426 if ((getModuleMetaBits(o->metabits)) && (keyMetaOnAof(r, key, o, dbid) == 0))
2427 return C_ERR;
2428
2429 return C_OK;
2430}
2431
2432int rewriteAppendOnlyFileRio(rio *aof) {
2433 dictEntry *de;
2434 int j;
2435 long key_count = 0;
2436 long long updated_time = 0;
2437 unsigned long long skipped = 0;
2438 kvstoreIterator kvs_it;
2439
2440 /* Record timestamp at the beginning of rewriting AOF. */
2441 if (server.aof_timestamp_enabled) {
2442 sds ts = genAofTimestampAnnotationIfNeeded(1);
2443 if (rioWrite(aof,ts,sdslen(ts)) == 0) { sdsfree(ts); goto werr; }
2444 sdsfree(ts);
2445 }
2446
2447 if (rewriteFunctions(aof) == 0) goto werr;
2448
2449 for (j = 0; j < server.dbnum; j++) {
2450 char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
2451 redisDb *db = server.db + j;
2452 if (kvstoreSize(db->keys) == 0) continue;
2453
2454 /* SELECT the new DB */
2455 if (rioWrite(aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
2456 if (rioWriteBulkLongLong(aof,j) == 0) goto werr;
2457
2458 kvstoreIteratorInit(&kvs_it, db->keys);
2459 /* Iterate this DB writing every entry */
2460 while((de = kvstoreIteratorNext(&kvs_it)) != NULL) {
2461 long long expiretime;
2462 size_t aof_bytes_before_key = aof->processed_bytes;
2463
2464 /* Get the value object (of type kvobj) */
2465 kvobj *o = dictGetKV(de);
2466
2467 /* Get the expire time */
2468 expiretime = kvobjGetExpire(o);
2469
2470 /* Skip keys that are being trimmed */
2471 if (server.cluster_enabled) {
2472 int curr_slot = kvstoreIteratorGetCurrentDictIndex(&kvs_it);
2473 if (isSlotInTrimJob(curr_slot)) {
2474 skipped++;
2475 continue;
2476 }
2477 }
2478
2479 /* Set on stack string object for key */
2480 robj key;
2481 initStaticStringObject(key, kvobjGetKey(o));
2482
2483 if (rewriteObject(aof, &key, o, j, expiretime) == C_ERR) goto werr2;
2484
2485 /* In fork child process, we can try to release memory back to the
2486 * OS and possibly avoid or decrease COW. We give the dismiss
2487 * mechanism a hint about an estimated size of the object we stored. */
2488 size_t dump_size = aof->processed_bytes - aof_bytes_before_key;
2489 if (server.in_fork_child) dismissObject(o, dump_size);
2490
2491 /* Update info every 1 second (approximately).
2492 * in order to avoid calling mstime() on each iteration, we will
2493 * check the diff every 1024 keys */
2494 if ((key_count++ & 1023) == 0) {
2495 long long now = mstime();
2496 if (now - updated_time >= 1000) {
2497 sendChildInfo(CHILD_INFO_TYPE_CURRENT_INFO, key_count, "AOF rewrite");
2498 updated_time = now;
2499 }
2500 }
2501
2502 /* Delay before next key if required (for testing) */
2503 if (server.rdb_key_save_delay)
2504 debugDelay(server.rdb_key_save_delay);
2505 }
2506 kvstoreIteratorReset(&kvs_it);
2507 }
2508 serverLog(LL_NOTICE, "AOF rewrite done, %ld keys saved, %llu keys skipped.", key_count, skipped);
2509 return C_OK;
2510
2511werr2:
2512 kvstoreIteratorReset(&kvs_it);
2513werr:
2514 return C_ERR;
2515}
2516
2517/* Write a sequence of commands able to fully rebuild the dataset into
2518 * "filename". Used both by REWRITEAOF and BGREWRITEAOF.
2519 *
2520 * In order to minimize the number of commands needed in the rewritten
2521 * log Redis uses variadic commands when possible, such as RPUSH, SADD
2522 * and ZADD. However at max AOF_REWRITE_ITEMS_PER_CMD items per time
2523 * are inserted using a single command. */
2524int rewriteAppendOnlyFile(char *filename) {
2525 rio aof;
2526 FILE *fp = NULL;
2527 char tmpfile[256];
2528
2529 /* Note that we have to use a different temp name here compared to the
2530 * one used by rewriteAppendOnlyFileBackground() function. */
2531 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
2532 fp = fopen(tmpfile,"w");
2533 if (!fp) {
2534 serverLog(LL_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
2535 return C_ERR;
2536 }
2537
2538 rioInitWithFile(&aof,fp);
2539
2540 if (server.aof_rewrite_incremental_fsync) {
2541 rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES);
2542 rioSetReclaimCache(&aof,1);
2543 }
2544
2545 startSaving(RDBFLAGS_AOF_PREAMBLE);
2546
2547 if (server.aof_use_rdb_preamble) {
2548 int error;
2549 if (rdbSaveRio(SLAVE_REQ_NONE,&aof,&error,RDBFLAGS_AOF_PREAMBLE,NULL) == C_ERR) {
2550 errno = error;
2551 goto werr;
2552 }
2553 } else {
2554 if (rewriteAppendOnlyFileRio(&aof) == C_ERR) goto werr;
2555 }
2556
2557 /* Make sure data will not remain on the OS's output buffers */
2558 if (fflush(fp)) goto werr;
2559 if (fsync(fileno(fp))) goto werr;
2560 if (reclaimFilePageCache(fileno(fp), 0, 0) == -1) {
2561 /* A minor error. Just log to know what happens */
2562 serverLog(LL_NOTICE,"Unable to reclaim page cache: %s", strerror(errno));
2563 }
2564 if (fclose(fp)) { fp = NULL; goto werr; }
2565 fp = NULL;
2566
2567 /* Use RENAME to make sure the DB file is changed atomically only
2568 * if the generate DB file is ok. */
2569 if (rename(tmpfile,filename) == -1) {
2570 serverLog(LL_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
2571 unlink(tmpfile);
2572 stopSaving(0);
2573 return C_ERR;
2574 }
2575 stopSaving(1);
2576
2577 return C_OK;
2578
2579werr:
2580 serverLog(LL_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
2581 if (fp) fclose(fp);
2582 unlink(tmpfile);
2583 stopSaving(0);
2584 return C_ERR;
2585}
2586/* ----------------------------------------------------------------------------
2587 * AOF background rewrite
2588 * ------------------------------------------------------------------------- */
2589
2590/* This is how rewriting of the append only file in background works:
2591 *
2592 * 1) The user calls BGREWRITEAOF
2593 * 2) Redis calls this function, that forks():
2594 * 2a) the child rewrite the append only file in a temp file.
2595 * 2b) the parent open a new INCR AOF file to continue writing.
2596 * 3) When the child finished '2a' exists.
2597 * 4) The parent will trap the exit code, if it's OK, it will:
2598 * 4a) get a new BASE file name and mark the previous (if we have) as the HISTORY type
2599 * 4b) rename(2) the temp file in new BASE file name
2600 * 4c) mark the rewritten INCR AOFs as history type
2601 * 4d) persist AOF manifest file
2602 * 4e) Delete the history files use bio
2603 */
2604int rewriteAppendOnlyFileBackground(void) {
2605 pid_t childpid;
2606
2607 if (hasActiveChildProcess()) return C_ERR;
2608
2609 if (dirCreateIfMissing(server.aof_dirname) == -1) {
2610 serverLog(LL_WARNING, "Can't open or create append-only dir %s: %s",
2611 server.aof_dirname, strerror(errno));
2612 server.aof_lastbgrewrite_status = C_ERR;
2613 return C_ERR;
2614 }
2615
2616 /* We set aof_selected_db to -1 in order to force the next call to the
2617 * feedAppendOnlyFile() to issue a SELECT command. */
2618 server.aof_selected_db = -1;
2619 flushAppendOnlyFile(1);
2620 if (openNewIncrAofForAppend() != C_OK) {
2621 server.aof_lastbgrewrite_status = C_ERR;
2622 return C_ERR;
2623 }
2624
2625 if (server.aof_state == AOF_WAIT_REWRITE) {
2626 /* Wait for all bio jobs related to AOF to drain. This prevents a race
2627 * between updates to `fsynced_reploff_pending` of the worker thread, belonging
2628 * to the previous AOF, and the new one. This concern is specific for a full
2629 * sync scenario where we don't wanna risk the ACKed replication offset
2630 * jumping backwards or forward when switching to a different master. */
2631 bioDrainWorker(BIO_AOF_FSYNC);
2632
2633 /* Set the initial repl_offset, which will be applied to fsynced_reploff
2634 * when AOFRW finishes (after possibly being updated by a bio thread) */
2635 atomicSet(server.fsynced_reploff_pending, server.master_repl_offset);
2636 server.fsynced_reploff = 0;
2637 }
2638
2639 server.stat_aof_rewrites++;
2640
2641 if ((childpid = redisFork(CHILD_TYPE_AOF)) == 0) {
2642 char tmpfile[256];
2643
2644 /* Child */
2645 redisSetProcTitle("redis-aof-rewrite");
2646 redisSetCpuAffinity(server.aof_rewrite_cpulist);
2647 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
2648 if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
2649 serverLog(LL_NOTICE,
2650 "Successfully created the temporary AOF base file %s", tmpfile);
2651 sendChildCowInfo(CHILD_INFO_TYPE_AOF_COW_SIZE, "AOF rewrite");
2652 exitFromChild(0, 0);
2653 } else {
2654 exitFromChild(1, 0);
2655 }
2656 } else {
2657 /* Parent */
2658 if (childpid == -1) {
2659 server.aof_lastbgrewrite_status = C_ERR;
2660 serverLog(LL_WARNING,
2661 "Can't rewrite append only file in background: fork: %s",
2662 strerror(errno));
2663 return C_ERR;
2664 }
2665 serverLog(LL_NOTICE,
2666 "Background append only file rewriting started by pid %ld",(long) childpid);
2667 server.aof_rewrite_scheduled = 0;
2668 server.aof_rewrite_time_start = time(NULL);
2669 return C_OK;
2670 }
2671 return C_OK; /* unreached */
2672}
2673
2674void bgrewriteaofCommand(client *c) {
2675 if (server.child_type == CHILD_TYPE_AOF) {
2676 addReplyError(c,"Background append only file rewriting already in progress");
2677 } else if (hasActiveChildProcess() || server.in_exec) {
2678 server.aof_rewrite_scheduled = 1;
2679 /* When manually triggering AOFRW we reset the count
2680 * so that it can be executed immediately. */
2681 server.stat_aofrw_consecutive_failures = 0;
2682 addReplyStatus(c,"Background append only file rewriting scheduled");
2683 } else if (rewriteAppendOnlyFileBackground() == C_OK) {
2684 addReplyStatus(c,"Background append only file rewriting started");
2685 } else {
2686 addReplyError(c,"Can't execute an AOF background rewriting. "
2687 "Please check the server logs for more information.");
2688 }
2689}
2690
2691void aofRemoveTempFile(pid_t childpid) {
2692 char tmpfile[256];
2693
2694 snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
2695 bg_unlink(tmpfile);
2696
2697 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) childpid);
2698 bg_unlink(tmpfile);
2699}
2700
2701/* Get size of an AOF file.
2702 * The status argument is an optional output argument to be filled with
2703 * one of the AOF_ status values. */
2704off_t getAppendOnlyFileSize(sds filename, int *status) {
2705 struct redis_stat sb;
2706 off_t size;
2707 mstime_t latency;
2708
2709 sds aof_filepath = makePath(server.aof_dirname, filename);
2710 latencyStartMonitor(latency);
2711 if (redis_stat(aof_filepath, &sb) == -1) {
2712 if (status) *status = errno == ENOENT ? AOF_NOT_EXIST : AOF_OPEN_ERR;
2713 serverLog(LL_WARNING, "Unable to obtain the AOF file %s length. stat: %s",
2714 filename, strerror(errno));
2715 size = 0;
2716 } else {
2717 if (status) *status = AOF_OK;
2718 size = sb.st_size;
2719 }
2720 latencyEndMonitor(latency);
2721 latencyAddSampleIfNeeded("aof-fstat", latency);
2722 sdsfree(aof_filepath);
2723 return size;
2724}
2725
2726/* Get size of all AOF files referred by the manifest (excluding history).
2727 * The status argument is an output argument to be filled with
2728 * one of the AOF_ status values. */
2729off_t getBaseAndIncrAppendOnlyFilesSize(aofManifest *am, int *status) {
2730 off_t size = 0;
2731 listNode *ln;
2732 listIter li;
2733
2734 if (am->base_aof_info) {
2735 serverAssert(am->base_aof_info->file_type == AOF_FILE_TYPE_BASE);
2736
2737 size += getAppendOnlyFileSize(am->base_aof_info->file_name, status);
2738 if (*status != AOF_OK) return 0;
2739 }
2740
2741 listRewind(am->incr_aof_list, &li);
2742 while ((ln = listNext(&li)) != NULL) {
2743 aofInfo *ai = (aofInfo*)ln->value;
2744 serverAssert(ai->file_type == AOF_FILE_TYPE_INCR);
2745 size += getAppendOnlyFileSize(ai->file_name, status);
2746 if (*status != AOF_OK) return 0;
2747 }
2748
2749 return size;
2750}
2751
2752int getBaseAndIncrAppendOnlyFilesNum(aofManifest *am) {
2753 int num = 0;
2754 if (am->base_aof_info) num++;
2755 if (am->incr_aof_list) num += listLength(am->incr_aof_list);
2756 return num;
2757}
2758
2759/* A background append only file rewriting (BGREWRITEAOF) terminated its work.
2760 * Handle this. */
2761void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
2762 if (!bysignal && exitcode == 0) {
2763 char tmpfile[256];
2764 long long now = ustime();
2765 sds new_base_filepath = NULL;
2766 sds new_incr_filepath = NULL;
2767 aofManifest *temp_am;
2768 mstime_t latency;
2769
2770 serverLog(LL_NOTICE,
2771 "Background AOF rewrite terminated with success");
2772
2773 snprintf(tmpfile, 256, "temp-rewriteaof-bg-%d.aof",
2774 (int)server.child_pid);
2775
2776 serverAssert(server.aof_manifest != NULL);
2777
2778 /* Dup a temporary aof_manifest for subsequent modifications. */
2779 temp_am = aofManifestDup(server.aof_manifest);
2780
2781 /* Get a new BASE file name and mark the previous (if we have)
2782 * as the HISTORY type. */
2783 sds new_base_filename = getNewBaseFileNameAndMarkPreAsHistory(temp_am);
2784 serverAssert(new_base_filename != NULL);
2785 new_base_filepath = makePath(server.aof_dirname, new_base_filename);
2786
2787 /* Rename the temporary aof file to 'new_base_filename'. */
2788 latencyStartMonitor(latency);
2789 if (rename(tmpfile, new_base_filepath) == -1) {
2790 serverLog(LL_WARNING,
2791 "Error trying to rename the temporary AOF base file %s into %s: %s",
2792 tmpfile,
2793 new_base_filepath,
2794 strerror(errno));
2795 aofManifestFree(temp_am);
2796 sdsfree(new_base_filepath);
2797 server.aof_lastbgrewrite_status = C_ERR;
2798 server.stat_aofrw_consecutive_failures++;
2799 goto cleanup;
2800 }
2801 latencyEndMonitor(latency);
2802 latencyAddSampleIfNeeded("aof-rename", latency);
2803 serverLog(LL_NOTICE,
2804 "Successfully renamed the temporary AOF base file %s into %s", tmpfile, new_base_filename);
2805
2806 /* Rename the temporary incr aof file to 'new_incr_filename'. */
2807 if (server.aof_state == AOF_WAIT_REWRITE) {
2808 /* Get temporary incr aof name. */
2809 sds temp_incr_aof_name = getTempIncrAofName();
2810 sds temp_incr_filepath = makePath(server.aof_dirname, temp_incr_aof_name);
2811 /* Get next new incr aof name. */
2812 sds new_incr_filename = getNewIncrAofName(temp_am, tempIncAofStartReplOffset);
2813 new_incr_filepath = makePath(server.aof_dirname, new_incr_filename);
2814 latencyStartMonitor(latency);
2815 if (rename(temp_incr_filepath, new_incr_filepath) == -1) {
2816 serverLog(LL_WARNING,
2817 "Error trying to rename the temporary AOF incr file %s into %s: %s",
2818 temp_incr_filepath,
2819 new_incr_filepath,
2820 strerror(errno));
2821 bg_unlink(new_base_filepath);
2822 sdsfree(new_base_filepath);
2823 aofManifestFree(temp_am);
2824 sdsfree(temp_incr_filepath);
2825 sdsfree(new_incr_filepath);
2826 sdsfree(temp_incr_aof_name);
2827 server.aof_lastbgrewrite_status = C_ERR;
2828 server.stat_aofrw_consecutive_failures++;
2829 goto cleanup;
2830 }
2831 latencyEndMonitor(latency);
2832 latencyAddSampleIfNeeded("aof-rename", latency);
2833 serverLog(LL_NOTICE,
2834 "Successfully renamed the temporary AOF incr file %s into %s", temp_incr_aof_name, new_incr_filename);
2835 sdsfree(temp_incr_filepath);
2836 sdsfree(temp_incr_aof_name);
2837 }
2838
2839 /* Change the AOF file type in 'incr_aof_list' from AOF_FILE_TYPE_INCR
2840 * to AOF_FILE_TYPE_HIST, and move them to the 'history_aof_list'. */
2841 markRewrittenIncrAofAsHistory(temp_am);
2842
2843 /* Persist our modifications. */
2844 if (persistAofManifest(temp_am) == C_ERR) {
2845 bg_unlink(new_base_filepath);
2846 aofManifestFree(temp_am);
2847 sdsfree(new_base_filepath);
2848 if (new_incr_filepath) {
2849 bg_unlink(new_incr_filepath);
2850 sdsfree(new_incr_filepath);
2851 }
2852 server.aof_lastbgrewrite_status = C_ERR;
2853 server.stat_aofrw_consecutive_failures++;
2854 goto cleanup;
2855 }
2856 sdsfree(new_base_filepath);
2857 if (new_incr_filepath) sdsfree(new_incr_filepath);
2858
2859 /* We can safely let `server.aof_manifest` point to 'temp_am' and free the previous one. */
2860 aofManifestFreeAndUpdate(temp_am);
2861
2862 if (server.aof_state != AOF_OFF) {
2863 /* AOF enabled. */
2864 server.aof_current_size = getAppendOnlyFileSize(new_base_filename, NULL) + server.aof_last_incr_size;
2865 server.aof_rewrite_base_size = server.aof_current_size;
2866 }
2867
2868 /* We don't care about the return value of `aofDelHistoryFiles`, because the history
2869 * deletion failure will not cause any problems. */
2870 aofDelHistoryFiles();
2871
2872 server.aof_lastbgrewrite_status = C_OK;
2873 server.stat_aofrw_consecutive_failures = 0;
2874
2875 serverLog(LL_NOTICE, "Background AOF rewrite finished successfully");
2876 /* Change state from WAIT_REWRITE to ON if needed */
2877 if (server.aof_state == AOF_WAIT_REWRITE) {
2878 server.aof_state = AOF_ON;
2879
2880 /* Update the fsynced replication offset that just now become valid.
2881 * This could either be the one we took in startAppendOnly, or a
2882 * newer one set by the bio thread. */
2883 long long fsynced_reploff_pending;
2884 atomicGet(server.fsynced_reploff_pending, fsynced_reploff_pending);
2885 server.fsynced_reploff = fsynced_reploff_pending;
2886 }
2887
2888 serverLog(LL_VERBOSE,
2889 "Background AOF rewrite signal handler took %lldus", ustime()-now);
2890 } else if (!bysignal && exitcode != 0) {
2891 server.aof_lastbgrewrite_status = C_ERR;
2892 server.stat_aofrw_consecutive_failures++;
2893
2894 serverLog(LL_WARNING,
2895 "Background AOF rewrite terminated with error");
2896 } else {
2897 /* SIGUSR1 is whitelisted, so we have a way to kill a child without
2898 * triggering an error condition. */
2899 if (bysignal != SIGUSR1) {
2900 server.aof_lastbgrewrite_status = C_ERR;
2901 server.stat_aofrw_consecutive_failures++;
2902 }
2903
2904 serverLog(LL_WARNING,
2905 "Background AOF rewrite terminated by signal %d", bysignal);
2906 }
2907
2908cleanup:
2909 aofRemoveTempFile(server.child_pid);
2910 /* Clear AOF buffer and delete temp incr aof for next rewrite. */
2911 if (server.aof_state == AOF_WAIT_REWRITE) {
2912 sdsfree(server.aof_buf);
2913 server.aof_buf = sdsempty();
2914 aofDelTempIncrAofFile();
2915 }
2916 server.aof_rewrite_time_last = time(NULL)-server.aof_rewrite_time_start;
2917 server.aof_rewrite_time_start = -1;
2918 /* Schedule a new rewrite if we are waiting for it to switch the AOF ON. */
2919 if (server.aof_state == AOF_WAIT_REWRITE)
2920 server.aof_rewrite_scheduled = 1;
2921}
generated by cgit v1.2.3 (git 2.46.0) at 2026-04-30 00:56:07 +0000