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