xref: /redis-3.2.3/src/aof.c (revision 79d2f70f)
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 "redis.h"
31 #include "bio.h"
32 #include "rio.h"
33 
34 #include <signal.h>
35 #include <fcntl.h>
36 #include <sys/stat.h>
37 #include <sys/types.h>
38 #include <sys/time.h>
39 #include <sys/resource.h>
40 #include <sys/wait.h>
41 
42 void aofUpdateCurrentSize(void);
43 void aofClosePipes(void);
44 
45 /* ----------------------------------------------------------------------------
46  * AOF rewrite buffer implementation.
47  *
48  * The following code implement a simple buffer used in order to accumulate
49  * changes while the background process is rewriting the AOF file.
50  *
51  * We only need to append, but can't just use realloc with a large block
52  * because 'huge' reallocs are not always handled as one could expect
53  * (via remapping of pages at OS level) but may involve copying data.
54  *
55  * For this reason we use a list of blocks, every block is
56  * AOF_RW_BUF_BLOCK_SIZE bytes.
57  * ------------------------------------------------------------------------- */
58 
59 #define AOF_RW_BUF_BLOCK_SIZE (1024*1024*10)    /* 10 MB per block */
60 
61 typedef struct aofrwblock {
62     unsigned long used, free;
63     char buf[AOF_RW_BUF_BLOCK_SIZE];
64 } aofrwblock;
65 
66 /* This function free the old AOF rewrite buffer if needed, and initialize
67  * a fresh new one. It tests for server.aof_rewrite_buf_blocks equal to NULL
68  * so can be used for the first initialization as well. */
69 void aofRewriteBufferReset(void) {
70     if (server.aof_rewrite_buf_blocks)
71         listRelease(server.aof_rewrite_buf_blocks);
72 
73     server.aof_rewrite_buf_blocks = listCreate();
74     listSetFreeMethod(server.aof_rewrite_buf_blocks,zfree);
75 }
76 
77 /* Return the current size of the AOF rerwite buffer. */
78 unsigned long aofRewriteBufferSize(void) {
79     listNode *ln;
80     listIter li;
81     unsigned long size = 0;
82 
83     listRewind(server.aof_rewrite_buf_blocks,&li);
84     while((ln = listNext(&li))) {
85         aofrwblock *block = listNodeValue(ln);
86         size += block->used;
87     }
88     return size;
89 }
90 
91 /* Event handler used to send data to the child process doing the AOF
92  * rewrite. We send pieces of our AOF differences buffer so that the final
93  * write when the child finishes the rewrite will be small. */
94 void aofChildWriteDiffData(aeEventLoop *el, int fd, void *privdata, int mask) {
95     listNode *ln;
96     aofrwblock *block;
97     ssize_t nwritten;
98 
99     while(1) {
100         ln = listFirst(server.aof_rewrite_buf_blocks);
101         block = ln ? ln->value : NULL;
102         if (server.aof_stop_sending_diff || !block) {
103             aeDeleteFileEvent(server.el,server.aof_pipe_write_data_to_child,
104                               AE_WRITABLE);
105             return;
106         }
107         if (block->used > 0) {
108             nwritten = write(server.aof_pipe_write_data_to_child,
109                              block->buf,block->used);
110             if (nwritten <= 0) return;
111             memmove(block->buf,block->buf+nwritten,block->used-nwritten);
112             block->used -= nwritten;
113         }
114         if (block->used == 0) listDelNode(server.aof_rewrite_buf_blocks,ln);
115     }
116 }
117 
118 /* Append data to the AOF rewrite buffer, allocating new blocks if needed. */
119 void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
120     listNode *ln = listLast(server.aof_rewrite_buf_blocks);
121     aofrwblock *block = ln ? ln->value : NULL;
122 
123     while(len) {
124         /* If we already got at least an allocated block, try appending
125          * at least some piece into it. */
126         if (block) {
127             unsigned long thislen = (block->free < len) ? block->free : len;
128             if (thislen) {  /* The current block is not already full. */
129                 memcpy(block->buf+block->used, s, thislen);
130                 block->used += thislen;
131                 block->free -= thislen;
132                 s += thislen;
133                 len -= thislen;
134             }
135         }
136 
137         if (len) { /* First block to allocate, or need another block. */
138             int numblocks;
139 
140             block = zmalloc(sizeof(*block));
141             block->free = AOF_RW_BUF_BLOCK_SIZE;
142             block->used = 0;
143             listAddNodeTail(server.aof_rewrite_buf_blocks,block);
144 
145             /* Log every time we cross more 10 or 100 blocks, respectively
146              * as a notice or warning. */
147             numblocks = listLength(server.aof_rewrite_buf_blocks);
148             if (((numblocks+1) % 10) == 0) {
149                 int level = ((numblocks+1) % 100) == 0 ? REDIS_WARNING :
150                                                          REDIS_NOTICE;
151                 redisLog(level,"Background AOF buffer size: %lu MB",
152                     aofRewriteBufferSize()/(1024*1024));
153             }
154         }
155     }
156 
157     /* Install a file event to send data to the rewrite child if there is
158      * not one already. */
159     if (aeGetFileEvents(server.el,server.aof_pipe_write_data_to_child) == 0) {
160         aeCreateFileEvent(server.el, server.aof_pipe_write_data_to_child,
161             AE_WRITABLE, aofChildWriteDiffData, NULL);
162     }
163 }
164 
165 /* Write the buffer (possibly composed of multiple blocks) into the specified
166  * fd. If a short write or any other error happens -1 is returned,
167  * otherwise the number of bytes written is returned. */
168 ssize_t aofRewriteBufferWrite(int fd) {
169     listNode *ln;
170     listIter li;
171     ssize_t count = 0;
172 
173     listRewind(server.aof_rewrite_buf_blocks,&li);
174     while((ln = listNext(&li))) {
175         aofrwblock *block = listNodeValue(ln);
176         ssize_t nwritten;
177 
178         if (block->used) {
179             nwritten = write(fd,block->buf,block->used);
180             if (nwritten != block->used) {
181                 if (nwritten == 0) errno = EIO;
182                 return -1;
183             }
184             count += nwritten;
185         }
186     }
187     return count;
188 }
189 
190 /* ----------------------------------------------------------------------------
191  * AOF file implementation
192  * ------------------------------------------------------------------------- */
193 
194 /* Starts a background task that performs fsync() against the specified
195  * file descriptor (the one of the AOF file) in another thread. */
196 void aof_background_fsync(int fd) {
197     bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL);
198 }
199 
200 /* Called when the user switches from "appendonly yes" to "appendonly no"
201  * at runtime using the CONFIG command. */
202 void stopAppendOnly(void) {
203     redisAssert(server.aof_state != REDIS_AOF_OFF);
204     flushAppendOnlyFile(1);
205     aof_fsync(server.aof_fd);
206     close(server.aof_fd);
207 
208     server.aof_fd = -1;
209     server.aof_selected_db = -1;
210     server.aof_state = REDIS_AOF_OFF;
211     /* rewrite operation in progress? kill it, wait child exit */
212     if (server.aof_child_pid != -1) {
213         int statloc;
214 
215         redisLog(REDIS_NOTICE,"Killing running AOF rewrite child: %ld",
216             (long) server.aof_child_pid);
217         if (kill(server.aof_child_pid,SIGUSR1) != -1)
218             wait3(&statloc,0,NULL);
219         /* reset the buffer accumulating changes while the child saves */
220         aofRewriteBufferReset();
221         aofRemoveTempFile(server.aof_child_pid);
222         server.aof_child_pid = -1;
223         server.aof_rewrite_time_start = -1;
224         /* close pipes used for IPC between the two processes. */
225         aofClosePipes();
226     }
227 }
228 
229 /* Called when the user switches from "appendonly no" to "appendonly yes"
230  * at runtime using the CONFIG command. */
231 int startAppendOnly(void) {
232     server.aof_last_fsync = server.unixtime;
233     server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644);
234     redisAssert(server.aof_state == REDIS_AOF_OFF);
235     if (server.aof_fd == -1) {
236         redisLog(REDIS_WARNING,"Redis needs to enable the AOF but can't open the append only file: %s",strerror(errno));
237         return REDIS_ERR;
238     }
239     if (rewriteAppendOnlyFileBackground() == REDIS_ERR) {
240         close(server.aof_fd);
241         redisLog(REDIS_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.");
242         return REDIS_ERR;
243     }
244     /* We correctly switched on AOF, now wait for the rerwite to be complete
245      * in order to append data on disk. */
246     server.aof_state = REDIS_AOF_WAIT_REWRITE;
247     return REDIS_OK;
248 }
249 
250 /* Write the append only file buffer on disk.
251  *
252  * Since we are required to write the AOF before replying to the client,
253  * and the only way the client socket can get a write is entering when the
254  * the event loop, we accumulate all the AOF writes in a memory
255  * buffer and write it on disk using this function just before entering
256  * the event loop again.
257  *
258  * About the 'force' argument:
259  *
260  * When the fsync policy is set to 'everysec' we may delay the flush if there
261  * is still an fsync() going on in the background thread, since for instance
262  * on Linux write(2) will be blocked by the background fsync anyway.
263  * When this happens we remember that there is some aof buffer to be
264  * flushed ASAP, and will try to do that in the serverCron() function.
265  *
266  * However if force is set to 1 we'll write regardless of the background
267  * fsync. */
268 #define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */
269 void flushAppendOnlyFile(int force) {
270     ssize_t nwritten;
271     int sync_in_progress = 0;
272     mstime_t latency;
273 
274     if (sdslen(server.aof_buf) == 0) return;
275 
276     if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
277         sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0;
278 
279     if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
280         /* With this append fsync policy we do background fsyncing.
281          * If the fsync is still in progress we can try to delay
282          * the write for a couple of seconds. */
283         if (sync_in_progress) {
284             if (server.aof_flush_postponed_start == 0) {
285                 /* No previous write postponinig, remember that we are
286                  * postponing the flush and return. */
287                 server.aof_flush_postponed_start = server.unixtime;
288                 return;
289             } else if (server.unixtime - server.aof_flush_postponed_start < 2) {
290                 /* We were already waiting for fsync to finish, but for less
291                  * than two seconds this is still ok. Postpone again. */
292                 return;
293             }
294             /* Otherwise fall trough, and go write since we can't wait
295              * over two seconds. */
296             server.aof_delayed_fsync++;
297             redisLog(REDIS_NOTICE,"Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down Redis.");
298         }
299     }
300     /* We want to perform a single write. This should be guaranteed atomic
301      * at least if the filesystem we are writing is a real physical one.
302      * While this will save us against the server being killed I don't think
303      * there is much to do about the whole server stopping for power problems
304      * or alike */
305 
306     latencyStartMonitor(latency);
307     nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
308     latencyEndMonitor(latency);
309     /* We want to capture different events for delayed writes:
310      * when the delay happens with a pending fsync, or with a saving child
311      * active, and when the above two conditions are missing.
312      * We also use an additional event name to save all samples which is
313      * useful for graphing / monitoring purposes. */
314     if (sync_in_progress) {
315         latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
316     } else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) {
317         latencyAddSampleIfNeeded("aof-write-active-child",latency);
318     } else {
319         latencyAddSampleIfNeeded("aof-write-alone",latency);
320     }
321     latencyAddSampleIfNeeded("aof-write",latency);
322 
323     /* We performed the write so reset the postponed flush sentinel to zero. */
324     server.aof_flush_postponed_start = 0;
325 
326     if (nwritten != (signed)sdslen(server.aof_buf)) {
327         static time_t last_write_error_log = 0;
328         int can_log = 0;
329 
330         /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */
331         if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) {
332             can_log = 1;
333             last_write_error_log = server.unixtime;
334         }
335 
336         /* Lof the AOF write error and record the error code. */
337         if (nwritten == -1) {
338             if (can_log) {
339                 redisLog(REDIS_WARNING,"Error writing to the AOF file: %s",
340                     strerror(errno));
341                 server.aof_last_write_errno = errno;
342             }
343         } else {
344             if (can_log) {
345                 redisLog(REDIS_WARNING,"Short write while writing to "
346                                        "the AOF file: (nwritten=%lld, "
347                                        "expected=%lld)",
348                                        (long long)nwritten,
349                                        (long long)sdslen(server.aof_buf));
350             }
351 
352             if (ftruncate(server.aof_fd, server.aof_current_size) == -1) {
353                 if (can_log) {
354                     redisLog(REDIS_WARNING, "Could not remove short write "
355                              "from the append-only file.  Redis may refuse "
356                              "to load the AOF the next time it starts.  "
357                              "ftruncate: %s", strerror(errno));
358                 }
359             } else {
360                 /* If the ftrunacate() succeeded we can set nwritten to
361                  * -1 since there is no longer partial data into the AOF. */
362                 nwritten = -1;
363             }
364             server.aof_last_write_errno = ENOSPC;
365         }
366 
367         /* Handle the AOF write error. */
368         if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
369             /* We can't recover when the fsync policy is ALWAYS since the
370              * reply for the client is already in the output buffers, and we
371              * have the contract with the user that on acknowledged write data
372              * is synched on disk. */
373             redisLog(REDIS_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
374             exit(1);
375         } else {
376             /* Recover from failed write leaving data into the buffer. However
377              * set an error to stop accepting writes as long as the error
378              * condition is not cleared. */
379             server.aof_last_write_status = REDIS_ERR;
380 
381             /* Trim the sds buffer if there was a partial write, and there
382              * was no way to undo it with ftruncate(2). */
383             if (nwritten > 0) {
384                 server.aof_current_size += nwritten;
385                 sdsrange(server.aof_buf,nwritten,-1);
386             }
387             return; /* We'll try again on the next call... */
388         }
389     } else {
390         /* Successful write(2). If AOF was in error state, restore the
391          * OK state and log the event. */
392         if (server.aof_last_write_status == REDIS_ERR) {
393             redisLog(REDIS_WARNING,
394                 "AOF write error looks solved, Redis can write again.");
395             server.aof_last_write_status = REDIS_OK;
396         }
397     }
398     server.aof_current_size += nwritten;
399 
400     /* Re-use AOF buffer when it is small enough. The maximum comes from the
401      * arena size of 4k minus some overhead (but is otherwise arbitrary). */
402     if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
403         sdsclear(server.aof_buf);
404     } else {
405         sdsfree(server.aof_buf);
406         server.aof_buf = sdsempty();
407     }
408 
409     /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
410      * children doing I/O in the background. */
411     if (server.aof_no_fsync_on_rewrite &&
412         (server.aof_child_pid != -1 || server.rdb_child_pid != -1))
413             return;
414 
415     /* Perform the fsync if needed. */
416     if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
417         /* aof_fsync is defined as fdatasync() for Linux in order to avoid
418          * flushing metadata. */
419         latencyStartMonitor(latency);
420         aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */
421         latencyEndMonitor(latency);
422         latencyAddSampleIfNeeded("aof-fsync-always",latency);
423         server.aof_last_fsync = server.unixtime;
424     } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
425                 server.unixtime > server.aof_last_fsync)) {
426         if (!sync_in_progress) aof_background_fsync(server.aof_fd);
427         server.aof_last_fsync = server.unixtime;
428     }
429 }
430 
431 sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) {
432     char buf[32];
433     int len, j;
434     robj *o;
435 
436     buf[0] = '*';
437     len = 1+ll2string(buf+1,sizeof(buf)-1,argc);
438     buf[len++] = '\r';
439     buf[len++] = '\n';
440     dst = sdscatlen(dst,buf,len);
441 
442     for (j = 0; j < argc; j++) {
443         o = getDecodedObject(argv[j]);
444         buf[0] = '$';
445         len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr));
446         buf[len++] = '\r';
447         buf[len++] = '\n';
448         dst = sdscatlen(dst,buf,len);
449         dst = sdscatlen(dst,o->ptr,sdslen(o->ptr));
450         dst = sdscatlen(dst,"\r\n",2);
451         decrRefCount(o);
452     }
453     return dst;
454 }
455 
456 /* Create the sds representation of an PEXPIREAT command, using
457  * 'seconds' as time to live and 'cmd' to understand what command
458  * we are translating into a PEXPIREAT.
459  *
460  * This command is used in order to translate EXPIRE and PEXPIRE commands
461  * into PEXPIREAT command so that we retain precision in the append only
462  * file, and the time is always absolute and not relative. */
463 sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) {
464     long long when;
465     robj *argv[3];
466 
467     /* Make sure we can use strtol */
468     seconds = getDecodedObject(seconds);
469     when = strtoll(seconds->ptr,NULL,10);
470     /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */
471     if (cmd->proc == expireCommand || cmd->proc == setexCommand ||
472         cmd->proc == expireatCommand)
473     {
474         when *= 1000;
475     }
476     /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */
477     if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
478         cmd->proc == setexCommand || cmd->proc == psetexCommand)
479     {
480         when += mstime();
481     }
482     decrRefCount(seconds);
483 
484     argv[0] = createStringObject("PEXPIREAT",9);
485     argv[1] = key;
486     argv[2] = createStringObjectFromLongLong(when);
487     buf = catAppendOnlyGenericCommand(buf, 3, argv);
488     decrRefCount(argv[0]);
489     decrRefCount(argv[2]);
490     return buf;
491 }
492 
493 void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
494     sds buf = sdsempty();
495     robj *tmpargv[3];
496 
497     /* The DB this command was targeting is not the same as the last command
498      * we appendend. To issue a SELECT command is needed. */
499     if (dictid != server.aof_selected_db) {
500         char seldb[64];
501 
502         snprintf(seldb,sizeof(seldb),"%d",dictid);
503         buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
504             (unsigned long)strlen(seldb),seldb);
505         server.aof_selected_db = dictid;
506     }
507 
508     if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
509         cmd->proc == expireatCommand) {
510         /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */
511         buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
512     } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) {
513         /* Translate SETEX/PSETEX to SET and PEXPIREAT */
514         tmpargv[0] = createStringObject("SET",3);
515         tmpargv[1] = argv[1];
516         tmpargv[2] = argv[3];
517         buf = catAppendOnlyGenericCommand(buf,3,tmpargv);
518         decrRefCount(tmpargv[0]);
519         buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
520     } else {
521         /* All the other commands don't need translation or need the
522          * same translation already operated in the command vector
523          * for the replication itself. */
524         buf = catAppendOnlyGenericCommand(buf,argc,argv);
525     }
526 
527     /* Append to the AOF buffer. This will be flushed on disk just before
528      * of re-entering the event loop, so before the client will get a
529      * positive reply about the operation performed. */
530     if (server.aof_state == REDIS_AOF_ON)
531         server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf));
532 
533     /* If a background append only file rewriting is in progress we want to
534      * accumulate the differences between the child DB and the current one
535      * in a buffer, so that when the child process will do its work we
536      * can append the differences to the new append only file. */
537     if (server.aof_child_pid != -1)
538         aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));
539 
540     sdsfree(buf);
541 }
542 
543 /* ----------------------------------------------------------------------------
544  * AOF loading
545  * ------------------------------------------------------------------------- */
546 
547 /* In Redis commands are always executed in the context of a client, so in
548  * order to load the append only file we need to create a fake client. */
549 struct redisClient *createFakeClient(void) {
550     struct redisClient *c = zmalloc(sizeof(*c));
551 
552     selectDb(c,0);
553     c->fd = -1;
554     c->name = NULL;
555     c->querybuf = sdsempty();
556     c->querybuf_peak = 0;
557     c->argc = 0;
558     c->argv = NULL;
559     c->bufpos = 0;
560     c->flags = 0;
561     c->btype = REDIS_BLOCKED_NONE;
562     /* We set the fake client as a slave waiting for the synchronization
563      * so that Redis will not try to send replies to this client. */
564     c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
565     c->reply = listCreate();
566     c->reply_bytes = 0;
567     c->obuf_soft_limit_reached_time = 0;
568     c->watched_keys = listCreate();
569     c->peerid = NULL;
570     listSetFreeMethod(c->reply,decrRefCountVoid);
571     listSetDupMethod(c->reply,dupClientReplyValue);
572     initClientMultiState(c);
573     return c;
574 }
575 
576 void freeFakeClient(struct redisClient *c) {
577     sdsfree(c->querybuf);
578     listRelease(c->reply);
579     listRelease(c->watched_keys);
580     freeClientMultiState(c);
581     zfree(c);
582 }
583 
584 /* Replay the append log file. On error REDIS_OK is returned. On non fatal
585  * error (the append only file is zero-length) REDIS_ERR is returned. On
586  * fatal error an error message is logged and the program exists. */
587 int loadAppendOnlyFile(char *filename) {
588     struct redisClient *fakeClient;
589     FILE *fp = fopen(filename,"r");
590     struct redis_stat sb;
591     int old_aof_state = server.aof_state;
592     long loops = 0;
593 
594     if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
595         server.aof_current_size = 0;
596         fclose(fp);
597         return REDIS_ERR;
598     }
599 
600     if (fp == NULL) {
601         redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno));
602         exit(1);
603     }
604 
605     /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
606      * to the same file we're about to read. */
607     server.aof_state = REDIS_AOF_OFF;
608 
609     fakeClient = createFakeClient();
610     startLoading(fp);
611 
612     while(1) {
613         int argc, j;
614         unsigned long len;
615         robj **argv;
616         char buf[128];
617         sds argsds;
618         struct redisCommand *cmd;
619 
620         /* Serve the clients from time to time */
621         if (!(loops++ % 1000)) {
622             loadingProgress(ftello(fp));
623             processEventsWhileBlocked();
624         }
625 
626         if (fgets(buf,sizeof(buf),fp) == NULL) {
627             if (feof(fp))
628                 break;
629             else
630                 goto readerr;
631         }
632         if (buf[0] != '*') goto fmterr;
633         argc = atoi(buf+1);
634         if (argc < 1) goto fmterr;
635 
636         argv = zmalloc(sizeof(robj*)*argc);
637         for (j = 0; j < argc; j++) {
638             if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr;
639             if (buf[0] != '$') goto fmterr;
640             len = strtol(buf+1,NULL,10);
641             argsds = sdsnewlen(NULL,len);
642             if (len && fread(argsds,len,1,fp) == 0) goto fmterr;
643             argv[j] = createObject(REDIS_STRING,argsds);
644             if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */
645         }
646 
647         /* Command lookup */
648         cmd = lookupCommand(argv[0]->ptr);
649         if (!cmd) {
650             redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", (char*)argv[0]->ptr);
651             exit(1);
652         }
653         /* Run the command in the context of a fake client */
654         fakeClient->argc = argc;
655         fakeClient->argv = argv;
656         cmd->proc(fakeClient);
657 
658         /* The fake client should not have a reply */
659         redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
660         /* The fake client should never get blocked */
661         redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0);
662 
663         /* Clean up. Command code may have changed argv/argc so we use the
664          * argv/argc of the client instead of the local variables. */
665         for (j = 0; j < fakeClient->argc; j++)
666             decrRefCount(fakeClient->argv[j]);
667         zfree(fakeClient->argv);
668     }
669 
670     /* This point can only be reached when EOF is reached without errors.
671      * If the client is in the middle of a MULTI/EXEC, log error and quit. */
672     if (fakeClient->flags & REDIS_MULTI) goto readerr;
673 
674     fclose(fp);
675     freeFakeClient(fakeClient);
676     server.aof_state = old_aof_state;
677     stopLoading();
678     aofUpdateCurrentSize();
679     server.aof_rewrite_base_size = server.aof_current_size;
680     return REDIS_OK;
681 
682 readerr:
683     if (feof(fp)) {
684         redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file");
685     } else {
686         redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno));
687     }
688     exit(1);
689 fmterr:
690     redisLog(REDIS_WARNING,"Bad file format reading the append only file: make a backup of your AOF file, then use ./redis-check-aof --fix <filename>");
691     exit(1);
692 }
693 
694 /* ----------------------------------------------------------------------------
695  * AOF rewrite
696  * ------------------------------------------------------------------------- */
697 
698 /* Delegate writing an object to writing a bulk string or bulk long long.
699  * This is not placed in rio.c since that adds the redis.h dependency. */
700 int rioWriteBulkObject(rio *r, robj *obj) {
701     /* Avoid using getDecodedObject to help copy-on-write (we are often
702      * in a child process when this function is called). */
703     if (obj->encoding == REDIS_ENCODING_INT) {
704         return rioWriteBulkLongLong(r,(long)obj->ptr);
705     } else if (sdsEncodedObject(obj)) {
706         return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr));
707     } else {
708         redisPanic("Unknown string encoding");
709     }
710 }
711 
712 /* Emit the commands needed to rebuild a list object.
713  * The function returns 0 on error, 1 on success. */
714 int rewriteListObject(rio *r, robj *key, robj *o) {
715     long long count = 0, items = listTypeLength(o);
716 
717     if (o->encoding == REDIS_ENCODING_ZIPLIST) {
718         unsigned char *zl = o->ptr;
719         unsigned char *p = ziplistIndex(zl,0);
720         unsigned char *vstr;
721         unsigned int vlen;
722         long long vlong;
723 
724         while(ziplistGet(p,&vstr,&vlen,&vlong)) {
725             if (count == 0) {
726                 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
727                     REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
728 
729                 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
730                 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
731                 if (rioWriteBulkObject(r,key) == 0) return 0;
732             }
733             if (vstr) {
734                 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
735             } else {
736                 if (rioWriteBulkLongLong(r,vlong) == 0) return 0;
737             }
738             p = ziplistNext(zl,p);
739             if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
740             items--;
741         }
742     } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
743         list *list = o->ptr;
744         listNode *ln;
745         listIter li;
746 
747         listRewind(list,&li);
748         while((ln = listNext(&li))) {
749             robj *eleobj = listNodeValue(ln);
750 
751             if (count == 0) {
752                 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
753                     REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
754 
755                 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
756                 if (rioWriteBulkString(r,"RPUSH",5) == 0) return 0;
757                 if (rioWriteBulkObject(r,key) == 0) return 0;
758             }
759             if (rioWriteBulkObject(r,eleobj) == 0) return 0;
760             if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
761             items--;
762         }
763     } else {
764         redisPanic("Unknown list encoding");
765     }
766     return 1;
767 }
768 
769 /* Emit the commands needed to rebuild a set object.
770  * The function returns 0 on error, 1 on success. */
771 int rewriteSetObject(rio *r, robj *key, robj *o) {
772     long long count = 0, items = setTypeSize(o);
773 
774     if (o->encoding == REDIS_ENCODING_INTSET) {
775         int ii = 0;
776         int64_t llval;
777 
778         while(intsetGet(o->ptr,ii++,&llval)) {
779             if (count == 0) {
780                 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
781                     REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
782 
783                 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
784                 if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
785                 if (rioWriteBulkObject(r,key) == 0) return 0;
786             }
787             if (rioWriteBulkLongLong(r,llval) == 0) return 0;
788             if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
789             items--;
790         }
791     } else if (o->encoding == REDIS_ENCODING_HT) {
792         dictIterator *di = dictGetIterator(o->ptr);
793         dictEntry *de;
794 
795         while((de = dictNext(di)) != NULL) {
796             robj *eleobj = dictGetKey(de);
797             if (count == 0) {
798                 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
799                     REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
800 
801                 if (rioWriteBulkCount(r,'*',2+cmd_items) == 0) return 0;
802                 if (rioWriteBulkString(r,"SADD",4) == 0) return 0;
803                 if (rioWriteBulkObject(r,key) == 0) return 0;
804             }
805             if (rioWriteBulkObject(r,eleobj) == 0) return 0;
806             if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
807             items--;
808         }
809         dictReleaseIterator(di);
810     } else {
811         redisPanic("Unknown set encoding");
812     }
813     return 1;
814 }
815 
816 /* Emit the commands needed to rebuild a sorted set object.
817  * The function returns 0 on error, 1 on success. */
818 int rewriteSortedSetObject(rio *r, robj *key, robj *o) {
819     long long count = 0, items = zsetLength(o);
820 
821     if (o->encoding == REDIS_ENCODING_ZIPLIST) {
822         unsigned char *zl = o->ptr;
823         unsigned char *eptr, *sptr;
824         unsigned char *vstr;
825         unsigned int vlen;
826         long long vll;
827         double score;
828 
829         eptr = ziplistIndex(zl,0);
830         redisAssert(eptr != NULL);
831         sptr = ziplistNext(zl,eptr);
832         redisAssert(sptr != NULL);
833 
834         while (eptr != NULL) {
835             redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
836             score = zzlGetScore(sptr);
837 
838             if (count == 0) {
839                 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
840                     REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
841 
842                 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
843                 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
844                 if (rioWriteBulkObject(r,key) == 0) return 0;
845             }
846             if (rioWriteBulkDouble(r,score) == 0) return 0;
847             if (vstr != NULL) {
848                 if (rioWriteBulkString(r,(char*)vstr,vlen) == 0) return 0;
849             } else {
850                 if (rioWriteBulkLongLong(r,vll) == 0) return 0;
851             }
852             zzlNext(zl,&eptr,&sptr);
853             if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
854             items--;
855         }
856     } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
857         zset *zs = o->ptr;
858         dictIterator *di = dictGetIterator(zs->dict);
859         dictEntry *de;
860 
861         while((de = dictNext(di)) != NULL) {
862             robj *eleobj = dictGetKey(de);
863             double *score = dictGetVal(de);
864 
865             if (count == 0) {
866                 int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
867                     REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
868 
869                 if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
870                 if (rioWriteBulkString(r,"ZADD",4) == 0) return 0;
871                 if (rioWriteBulkObject(r,key) == 0) return 0;
872             }
873             if (rioWriteBulkDouble(r,*score) == 0) return 0;
874             if (rioWriteBulkObject(r,eleobj) == 0) return 0;
875             if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
876             items--;
877         }
878         dictReleaseIterator(di);
879     } else {
880         redisPanic("Unknown sorted zset encoding");
881     }
882     return 1;
883 }
884 
885 /* Write either the key or the value of the currently selected item of a hash.
886  * The 'hi' argument passes a valid Redis hash iterator.
887  * The 'what' filed specifies if to write a key or a value and can be
888  * either REDIS_HASH_KEY or REDIS_HASH_VALUE.
889  *
890  * The function returns 0 on error, non-zero on success. */
891 static int rioWriteHashIteratorCursor(rio *r, hashTypeIterator *hi, int what) {
892     if (hi->encoding == REDIS_ENCODING_ZIPLIST) {
893         unsigned char *vstr = NULL;
894         unsigned int vlen = UINT_MAX;
895         long long vll = LLONG_MAX;
896 
897         hashTypeCurrentFromZiplist(hi, what, &vstr, &vlen, &vll);
898         if (vstr) {
899             return rioWriteBulkString(r, (char*)vstr, vlen);
900         } else {
901             return rioWriteBulkLongLong(r, vll);
902         }
903 
904     } else if (hi->encoding == REDIS_ENCODING_HT) {
905         robj *value;
906 
907         hashTypeCurrentFromHashTable(hi, what, &value);
908         return rioWriteBulkObject(r, value);
909     }
910 
911     redisPanic("Unknown hash encoding");
912     return 0;
913 }
914 
915 /* Emit the commands needed to rebuild a hash object.
916  * The function returns 0 on error, 1 on success. */
917 int rewriteHashObject(rio *r, robj *key, robj *o) {
918     hashTypeIterator *hi;
919     long long count = 0, items = hashTypeLength(o);
920 
921     hi = hashTypeInitIterator(o);
922     while (hashTypeNext(hi) != REDIS_ERR) {
923         if (count == 0) {
924             int cmd_items = (items > REDIS_AOF_REWRITE_ITEMS_PER_CMD) ?
925                 REDIS_AOF_REWRITE_ITEMS_PER_CMD : items;
926 
927             if (rioWriteBulkCount(r,'*',2+cmd_items*2) == 0) return 0;
928             if (rioWriteBulkString(r,"HMSET",5) == 0) return 0;
929             if (rioWriteBulkObject(r,key) == 0) return 0;
930         }
931 
932         if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_KEY) == 0) return 0;
933         if (rioWriteHashIteratorCursor(r, hi, REDIS_HASH_VALUE) == 0) return 0;
934         if (++count == REDIS_AOF_REWRITE_ITEMS_PER_CMD) count = 0;
935         items--;
936     }
937 
938     hashTypeReleaseIterator(hi);
939 
940     return 1;
941 }
942 
943 /* This function is called by the child rewriting the AOF file to read
944  * the difference accumulated from the parent into a buffer, that is
945  * concatenated at the end of the rewrite. */
946 ssize_t aofReadDiffFromParent(void) {
947     char buf[65536]; /* Default pipe buffer size on most Linux sytems. */
948     ssize_t nread, total = 0;
949 
950     while ((nread =
951             read(server.aof_pipe_read_data_from_parent,buf,sizeof(buf))) > 0) {
952         server.aof_child_diff = sdscatlen(server.aof_child_diff,buf,nread);
953         total += nread;
954     }
955     return total;
956 }
957 
958 /* Write a sequence of commands able to fully rebuild the dataset into
959  * "filename". Used both by REWRITEAOF and BGREWRITEAOF.
960  *
961  * In order to minimize the number of commands needed in the rewritten
962  * log Redis uses variadic commands when possible, such as RPUSH, SADD
963  * and ZADD. However at max REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time
964  * are inserted using a single command. */
965 int rewriteAppendOnlyFile(char *filename) {
966     dictIterator *di = NULL;
967     dictEntry *de;
968     rio aof;
969     FILE *fp;
970     char tmpfile[256];
971     int j;
972     long long now = mstime();
973     char byte;
974     size_t processed = 0;
975 
976     /* Note that we have to use a different temp name here compared to the
977      * one used by rewriteAppendOnlyFileBackground() function. */
978     snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
979     fp = fopen(tmpfile,"w");
980     if (!fp) {
981         redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
982         return REDIS_ERR;
983     }
984 
985     server.aof_child_diff = sdsempty();
986     rioInitWithFile(&aof,fp);
987     if (server.aof_rewrite_incremental_fsync)
988         rioSetAutoSync(&aof,REDIS_AOF_AUTOSYNC_BYTES);
989     for (j = 0; j < server.dbnum; j++) {
990         char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
991         redisDb *db = server.db+j;
992         dict *d = db->dict;
993         if (dictSize(d) == 0) continue;
994         di = dictGetSafeIterator(d);
995         if (!di) {
996             fclose(fp);
997             return REDIS_ERR;
998         }
999 
1000         /* SELECT the new DB */
1001         if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
1002         if (rioWriteBulkLongLong(&aof,j) == 0) goto werr;
1003 
1004         /* Iterate this DB writing every entry */
1005         while((de = dictNext(di)) != NULL) {
1006             sds keystr;
1007             robj key, *o;
1008             long long expiretime;
1009 
1010             keystr = dictGetKey(de);
1011             o = dictGetVal(de);
1012             initStaticStringObject(key,keystr);
1013 
1014             expiretime = getExpire(db,&key);
1015 
1016             /* If this key is already expired skip it */
1017             if (expiretime != -1 && expiretime < now) continue;
1018 
1019             /* Save the key and associated value */
1020             if (o->type == REDIS_STRING) {
1021                 /* Emit a SET command */
1022                 char cmd[]="*3\r\n$3\r\nSET\r\n";
1023                 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
1024                 /* Key and value */
1025                 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
1026                 if (rioWriteBulkObject(&aof,o) == 0) goto werr;
1027             } else if (o->type == REDIS_LIST) {
1028                 if (rewriteListObject(&aof,&key,o) == 0) goto werr;
1029             } else if (o->type == REDIS_SET) {
1030                 if (rewriteSetObject(&aof,&key,o) == 0) goto werr;
1031             } else if (o->type == REDIS_ZSET) {
1032                 if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
1033             } else if (o->type == REDIS_HASH) {
1034                 if (rewriteHashObject(&aof,&key,o) == 0) goto werr;
1035             } else {
1036                 redisPanic("Unknown object type");
1037             }
1038             /* Save the expire time */
1039             if (expiretime != -1) {
1040                 char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
1041                 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
1042                 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
1043                 if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;
1044             }
1045             /* Read some diff from the parent process from time to time. */
1046             if (aof.processed_bytes > processed+1024*10) {
1047                 processed = aof.processed_bytes;
1048                 aofReadDiffFromParent();
1049             }
1050         }
1051         dictReleaseIterator(di);
1052     }
1053 
1054     /* Do an initial slow fsync here while the parent is still sending
1055      * data, in order to make the next final fsync faster. */
1056     if (fflush(fp) == EOF) goto werr;
1057     if (fsync(fileno(fp)) == -1) goto werr;
1058 
1059     /* Read again a few times to get more data from the parent.
1060      * We can't read forever (the server may receive data from clients
1061      * fater than it is able to send data to the child), so we try to read
1062      * some more data in a loop as soon as there is a good chance more data
1063      * will come. If it looks like we are wasting time, we abort (this
1064      * happens after 20 ms without new data). */
1065     int nodata = 0;
1066     mstime_t start = mstime();
1067     while(mstime()-start < 1000 && nodata < 20) {
1068         if (aeWait(server.aof_pipe_read_data_from_parent, AE_READABLE, 1) <= 0)
1069         {
1070             nodata++;
1071             continue;
1072         }
1073         nodata = 0; /* Start counting from zero, we stop on N *contiguous*
1074                        timeouts. */
1075         aofReadDiffFromParent();
1076     }
1077 
1078     /* Ask the master to stop sending diffs. */
1079     if (write(server.aof_pipe_write_ack_to_parent,"!",1) != 1) goto werr;
1080     if (anetNonBlock(NULL,server.aof_pipe_read_ack_from_parent) != ANET_OK)
1081         goto werr;
1082     /* We read the ACK from the server using a 10 seconds timeout. Normally
1083      * it should reply ASAP, but just in case we lose its reply, we are sure
1084      * the child will eventually get terminated. */
1085     if (syncRead(server.aof_pipe_read_ack_from_parent,&byte,1,5000) != 1 ||
1086         byte != '!') goto werr;
1087     redisLog(REDIS_NOTICE,"Parent agreed to stop sending diffs. Finalizing AOF...");
1088 
1089     /* Read the final diff if any. */
1090     aofReadDiffFromParent();
1091 
1092     /* Write the received diff to the file. */
1093     redisLog(REDIS_NOTICE,
1094         "Concatenating %.2f MB of AOF diff received from parent.",
1095         (double) sdslen(server.aof_child_diff) / (1024*1024));
1096     if (rioWrite(&aof,server.aof_child_diff,sdslen(server.aof_child_diff)) == 0)
1097         goto werr;
1098 
1099     /* Make sure data will not remain on the OS's output buffers */
1100     if (fflush(fp) == EOF) goto werr;
1101     if (fsync(fileno(fp)) == -1) goto werr;
1102     if (fclose(fp) == EOF) goto werr;
1103 
1104     /* Use RENAME to make sure the DB file is changed atomically only
1105      * if the generate DB file is ok. */
1106     if (rename(tmpfile,filename) == -1) {
1107         redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
1108         unlink(tmpfile);
1109         return REDIS_ERR;
1110     }
1111     redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");
1112     return REDIS_OK;
1113 
1114 werr:
1115     fclose(fp);
1116     unlink(tmpfile);
1117     redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
1118     if (di) dictReleaseIterator(di);
1119     return REDIS_ERR;
1120 }
1121 
1122 /* ----------------------------------------------------------------------------
1123  * AOF rewrite pipes for IPC
1124  * -------------------------------------------------------------------------- */
1125 
1126 /* This event handler is called when the AOF rewriting child sends us a
1127  * single '!' char to signal we should stop sending buffer diffs. The
1128  * parent sends a '!' as well to acknowledge. */
1129 void aofChildPipeReadable(aeEventLoop *el, int fd, void *privdata, int mask) {
1130     char byte;
1131 
1132     if (read(fd,&byte,1) == 1 && byte == '!') {
1133         redisLog(REDIS_NOTICE,"AOF rewrite child asks to stop sending diffs.");
1134         server.aof_stop_sending_diff = 1;
1135         if (write(server.aof_pipe_write_ack_to_child,"!",1) != 1) {
1136             /* If we can't send the ack, inform the user, but don't try again
1137              * since in the other side the children will use a timeout if the
1138              * kernel can't buffer our write, or, the children was
1139              * terminated. */
1140             redisLog(REDIS_WARNING,"Can't send ACK to AOF child: %s",
1141                 strerror(errno));
1142         }
1143     }
1144     /* Remove the handler since this can be called only one time during a
1145      * rewrite. */
1146     aeDeleteFileEvent(server.el,server.aof_pipe_read_ack_from_child,AE_READABLE);
1147 }
1148 
1149 /* Create the pipes used for parent - child process IPC during rewrite.
1150  * We have a data pipe used to send AOF incremental diffs to the child,
1151  * and two other pipes used by the children to signal it finished with
1152  * the rewrite so no more data should be written, and another for the
1153  * parent to acknowledge it understood this new condition. */
1154 int aofCreatePipes(void) {
1155     int fds[6] = {-1, -1, -1, -1, -1, -1};
1156     int j;
1157 
1158     if (pipe(fds) == -1) goto error; /* parent -> children data. */
1159     if (pipe(fds+2) == -1) goto error; /* children -> parent ack. */
1160     if (pipe(fds+4) == -1) goto error; /* children -> parent ack. */
1161     /* Parent -> children data is non blocking. */
1162     if (anetNonBlock(NULL,fds[0]) != ANET_OK) goto error;
1163     if (anetNonBlock(NULL,fds[1]) != ANET_OK) goto error;
1164     if (aeCreateFileEvent(server.el, fds[2], AE_READABLE, aofChildPipeReadable, NULL) == AE_ERR) goto error;
1165 
1166     server.aof_pipe_write_data_to_child = fds[1];
1167     server.aof_pipe_read_data_from_parent = fds[0];
1168     server.aof_pipe_write_ack_to_parent = fds[3];
1169     server.aof_pipe_read_ack_from_child = fds[2];
1170     server.aof_pipe_write_ack_to_child = fds[5];
1171     server.aof_pipe_read_ack_from_parent = fds[4];
1172     server.aof_stop_sending_diff = 0;
1173     return REDIS_OK;
1174 
1175 error:
1176     redisLog(REDIS_WARNING,"Error opening /setting AOF rewrite IPC pipes: %s",
1177         strerror(errno));
1178     for (j = 0; j < 6; j++) if(fds[j] != -1) close(fds[j]);
1179     return REDIS_ERR;
1180 }
1181 
1182 void aofClosePipes(void) {
1183     aeDeleteFileEvent(server.el,server.aof_pipe_read_ack_from_child,AE_READABLE);
1184     aeDeleteFileEvent(server.el,server.aof_pipe_write_data_to_child,AE_WRITABLE);
1185     close(server.aof_pipe_write_data_to_child);
1186     close(server.aof_pipe_read_data_from_parent);
1187     close(server.aof_pipe_write_ack_to_parent);
1188     close(server.aof_pipe_read_ack_from_child);
1189     close(server.aof_pipe_write_ack_to_child);
1190     close(server.aof_pipe_read_ack_from_parent);
1191 }
1192 
1193 /* ----------------------------------------------------------------------------
1194  * AOF backgorund rewrite
1195  * ------------------------------------------------------------------------- */
1196 
1197 /* This is how rewriting of the append only file in background works:
1198  *
1199  * 1) The user calls BGREWRITEAOF
1200  * 2) Redis calls this function, that forks():
1201  *    2a) the child rewrite the append only file in a temp file.
1202  *    2b) the parent accumulates differences in server.aof_rewrite_buf.
1203  * 3) When the child finished '2a' exists.
1204  * 4) The parent will trap the exit code, if it's OK, will append the
1205  *    data accumulated into server.aof_rewrite_buf into the temp file, and
1206  *    finally will rename(2) the temp file in the actual file name.
1207  *    The the new file is reopened as the new append only file. Profit!
1208  */
1209 int rewriteAppendOnlyFileBackground(void) {
1210     pid_t childpid;
1211     long long start;
1212 
1213     if (server.aof_child_pid != -1) return REDIS_ERR;
1214     if (aofCreatePipes() != REDIS_OK) return REDIS_ERR;
1215     start = ustime();
1216     if ((childpid = fork()) == 0) {
1217         char tmpfile[256];
1218 
1219         /* Child */
1220         closeListeningSockets(0);
1221         redisSetProcTitle("redis-aof-rewrite");
1222         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
1223         if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
1224             size_t private_dirty = zmalloc_get_private_dirty();
1225 
1226             if (private_dirty) {
1227                 redisLog(REDIS_NOTICE,
1228                     "AOF rewrite: %zu MB of memory used by copy-on-write",
1229                     private_dirty/(1024*1024));
1230             }
1231             exitFromChild(0);
1232         } else {
1233             exitFromChild(1);
1234         }
1235     } else {
1236         /* Parent */
1237         server.stat_fork_time = ustime()-start;
1238         server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
1239         latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
1240         if (childpid == -1) {
1241             redisLog(REDIS_WARNING,
1242                 "Can't rewrite append only file in background: fork: %s",
1243                 strerror(errno));
1244             return REDIS_ERR;
1245         }
1246         redisLog(REDIS_NOTICE,
1247             "Background append only file rewriting started by pid %d",childpid);
1248         server.aof_rewrite_scheduled = 0;
1249         server.aof_rewrite_time_start = time(NULL);
1250         server.aof_child_pid = childpid;
1251         updateDictResizePolicy();
1252         /* We set appendseldb to -1 in order to force the next call to the
1253          * feedAppendOnlyFile() to issue a SELECT command, so the differences
1254          * accumulated by the parent into server.aof_rewrite_buf will start
1255          * with a SELECT statement and it will be safe to merge. */
1256         server.aof_selected_db = -1;
1257         replicationScriptCacheFlush();
1258         return REDIS_OK;
1259     }
1260     return REDIS_OK; /* unreached */
1261 }
1262 
1263 void bgrewriteaofCommand(redisClient *c) {
1264     if (server.aof_child_pid != -1) {
1265         addReplyError(c,"Background append only file rewriting already in progress");
1266     } else if (server.rdb_child_pid != -1) {
1267         server.aof_rewrite_scheduled = 1;
1268         addReplyStatus(c,"Background append only file rewriting scheduled");
1269     } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) {
1270         addReplyStatus(c,"Background append only file rewriting started");
1271     } else {
1272         addReply(c,shared.err);
1273     }
1274 }
1275 
1276 void aofRemoveTempFile(pid_t childpid) {
1277     char tmpfile[256];
1278 
1279     snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
1280     unlink(tmpfile);
1281 }
1282 
1283 /* Update the server.aof_current_size filed explicitly using stat(2)
1284  * to check the size of the file. This is useful after a rewrite or after
1285  * a restart, normally the size is updated just adding the write length
1286  * to the current length, that is much faster. */
1287 void aofUpdateCurrentSize(void) {
1288     struct redis_stat sb;
1289     mstime_t latency;
1290 
1291     latencyStartMonitor(latency);
1292     if (redis_fstat(server.aof_fd,&sb) == -1) {
1293         redisLog(REDIS_WARNING,"Unable to obtain the AOF file length. stat: %s",
1294             strerror(errno));
1295     } else {
1296         server.aof_current_size = sb.st_size;
1297     }
1298     latencyEndMonitor(latency);
1299     latencyAddSampleIfNeeded("aof-fstat",latency);
1300 }
1301 
1302 /* A background append only file rewriting (BGREWRITEAOF) terminated its work.
1303  * Handle this. */
1304 void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
1305     if (!bysignal && exitcode == 0) {
1306         int newfd, oldfd;
1307         char tmpfile[256];
1308         long long now = ustime();
1309         mstime_t latency;
1310 
1311         redisLog(REDIS_NOTICE,
1312             "Background AOF rewrite terminated with success");
1313 
1314         /* Flush the differences accumulated by the parent to the
1315          * rewritten AOF. */
1316         latencyStartMonitor(latency);
1317         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
1318             (int)server.aof_child_pid);
1319         newfd = open(tmpfile,O_WRONLY|O_APPEND);
1320         if (newfd == -1) {
1321             redisLog(REDIS_WARNING,
1322                 "Unable to open the temporary AOF produced by the child: %s", strerror(errno));
1323             goto cleanup;
1324         }
1325 
1326         if (aofRewriteBufferWrite(newfd) == -1) {
1327             redisLog(REDIS_WARNING,
1328                 "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno));
1329             close(newfd);
1330             goto cleanup;
1331         }
1332         latencyEndMonitor(latency);
1333         latencyAddSampleIfNeeded("aof-rewrite-diff-write",latency);
1334 
1335         redisLog(REDIS_NOTICE,
1336             "Redidual parent diff successfully flushed to the rewritten AOF (%.2f MB)", (double) aofRewriteBufferSize() / (1024*1024));
1337 
1338         /* The only remaining thing to do is to rename the temporary file to
1339          * the configured file and switch the file descriptor used to do AOF
1340          * writes. We don't want close(2) or rename(2) calls to block the
1341          * server on old file deletion.
1342          *
1343          * There are two possible scenarios:
1344          *
1345          * 1) AOF is DISABLED and this was a one time rewrite. The temporary
1346          * file will be renamed to the configured file. When this file already
1347          * exists, it will be unlinked, which may block the server.
1348          *
1349          * 2) AOF is ENABLED and the rewritten AOF will immediately start
1350          * receiving writes. After the temporary file is renamed to the
1351          * configured file, the original AOF file descriptor will be closed.
1352          * Since this will be the last reference to that file, closing it
1353          * causes the underlying file to be unlinked, which may block the
1354          * server.
1355          *
1356          * To mitigate the blocking effect of the unlink operation (either
1357          * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we
1358          * use a background thread to take care of this. First, we
1359          * make scenario 1 identical to scenario 2 by opening the target file
1360          * when it exists. The unlink operation after the rename(2) will then
1361          * be executed upon calling close(2) for its descriptor. Everything to
1362          * guarantee atomicity for this switch has already happened by then, so
1363          * we don't care what the outcome or duration of that close operation
1364          * is, as long as the file descriptor is released again. */
1365         if (server.aof_fd == -1) {
1366             /* AOF disabled */
1367 
1368              /* Don't care if this fails: oldfd will be -1 and we handle that.
1369               * One notable case of -1 return is if the old file does
1370               * not exist. */
1371              oldfd = open(server.aof_filename,O_RDONLY|O_NONBLOCK);
1372         } else {
1373             /* AOF enabled */
1374             oldfd = -1; /* We'll set this to the current AOF filedes later. */
1375         }
1376 
1377         /* Rename the temporary file. This will not unlink the target file if
1378          * it exists, because we reference it with "oldfd". */
1379         latencyStartMonitor(latency);
1380         if (rename(tmpfile,server.aof_filename) == -1) {
1381             redisLog(REDIS_WARNING,
1382                 "Error trying to rename the temporary AOF file: %s", strerror(errno));
1383             close(newfd);
1384             if (oldfd != -1) close(oldfd);
1385             goto cleanup;
1386         }
1387         latencyEndMonitor(latency);
1388         latencyAddSampleIfNeeded("aof-rename",latency);
1389 
1390         if (server.aof_fd == -1) {
1391             /* AOF disabled, we don't need to set the AOF file descriptor
1392              * to this new file, so we can close it. */
1393             close(newfd);
1394         } else {
1395             /* AOF enabled, replace the old fd with the new one. */
1396             oldfd = server.aof_fd;
1397             server.aof_fd = newfd;
1398             if (server.aof_fsync == AOF_FSYNC_ALWAYS)
1399                 aof_fsync(newfd);
1400             else if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
1401                 aof_background_fsync(newfd);
1402             server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
1403             aofUpdateCurrentSize();
1404             server.aof_rewrite_base_size = server.aof_current_size;
1405 
1406             /* Clear regular AOF buffer since its contents was just written to
1407              * the new AOF from the background rewrite buffer. */
1408             sdsfree(server.aof_buf);
1409             server.aof_buf = sdsempty();
1410         }
1411 
1412         server.aof_lastbgrewrite_status = REDIS_OK;
1413 
1414         redisLog(REDIS_NOTICE, "Background AOF rewrite finished successfully");
1415         /* Change state from WAIT_REWRITE to ON if needed */
1416         if (server.aof_state == REDIS_AOF_WAIT_REWRITE)
1417             server.aof_state = REDIS_AOF_ON;
1418 
1419         /* Asynchronously close the overwritten AOF. */
1420         if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL);
1421 
1422         redisLog(REDIS_VERBOSE,
1423             "Background AOF rewrite signal handler took %lldus", ustime()-now);
1424     } else if (!bysignal && exitcode != 0) {
1425         server.aof_lastbgrewrite_status = REDIS_ERR;
1426 
1427         redisLog(REDIS_WARNING,
1428             "Background AOF rewrite terminated with error");
1429     } else {
1430         server.aof_lastbgrewrite_status = REDIS_ERR;
1431 
1432         redisLog(REDIS_WARNING,
1433             "Background AOF rewrite terminated by signal %d", bysignal);
1434     }
1435 
1436 cleanup:
1437     aofClosePipes();
1438     aofRewriteBufferReset();
1439     aofRemoveTempFile(server.aof_child_pid);
1440     server.aof_child_pid = -1;
1441     server.aof_rewrite_time_last = time(NULL)-server.aof_rewrite_time_start;
1442     server.aof_rewrite_time_start = -1;
1443     /* Schedule a new rewrite if we are waiting for it to switch the AOF ON. */
1444     if (server.aof_state == REDIS_AOF_WAIT_REWRITE)
1445         server.aof_rewrite_scheduled = 1;
1446 }
1447