redis源码阅读之aof

本来今天想写点别的，但是心想之前一篇既然已经提到持久化这边了，而且也说了之后会讲到，索性，今天就说说这个。
所谓持久化，就是将内存中的内容同步到磁盘当中，redis提供了两种持久化机制：aof和rdb。今天的主角是aof。
aof持久化将被执行的命令写到AOF的末尾，以此来记录数据发生的变化，它一共有三个配置选项，在redis.conf这个配置文件中，具体如下：

点击(此处)折叠或打开

1. #
   
2. # If unsure, use "everysec".
   
3. 
   
4. # appendfsync always
   
5. appendfsync everysec
   
6. # appendfsync no

其意义也比较明显了，就不再多说了。
redis开启aof的开关代码如下：

点击(此处)折叠或打开

1. /* Called when the user switches from "appendonly no" to "appendonly yes"
   
2.  * at runtime using the CONFIG command. */
   
3. int startAppendOnly(void) {
   
4.     char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */
   
5.     int newfd;
   
6. 
   
7.     newfd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644);
   
8.     serverAssert(server.aof_state == AOF_OFF);
   
9.     if (newfd == -1) {
   
10.         char *cwdp = getcwd(cwd,MAXPATHLEN);
    
11. 
    
12.         serverLog(LL_WARNING,
    
13.             "Redis needs to enable the AOF but can't open the "
    
14.             "append only file %s (in server root dir %s): %s",
    
15.             server.aof_filename,
    
16.             cwdp ? cwdp : "unknown",
    
17.             strerror(errno));
    
18.         return C_ERR;
    
19.     }
    
20.     if (server.rdb_child_pid != -1) {
    
21.         server.aof_rewrite_scheduled = 1;
    
22.         serverLog(LL_WARNING,"AOF was enabled but there is already a child process saving an RDB file on disk. An AOF background was scheduled to start when possible.");
    
23.     } else {
    
24.         /* If there is a pending AOF rewrite, we need to switch it off and
    
25.          * start a new one: the old one cannot be reused becuase it is not
    
26.          * accumulating the AOF buffer. */
    
27.         if (server.aof_child_pid != -1) {
    
28.             serverLog(LL_WARNING,"AOF was enabled but there is already an AOF rewriting in background. Stopping background AOF and starting a rewrite now.");
    
29.             killAppendOnlyChild();
    
30.         }
    
31.         if (rewriteAppendOnlyFileBackground() == C_ERR) {
    
32.             close(newfd);
    
33.             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.");
    
34.             return C_ERR;
    
35.         }
    
36.     }
    
37.     /* We correctly switched on AOF, now wait for the rewrite to be complete
    
38.      * in order to append data on disk. */
    
39.     server.aof_state = AOF_WAIT_REWRITE;
    
40.     server.aof_last_fsync = server.unixtime;
    
41.     server.aof_fd = newfd;
    
42.     return C_OK;
    
43. }

aof写的代码如下：

点击(此处)折叠或打开

1. ssize_t aofWrite(int fd, const char *buf, size_t len) {
   
2.     ssize_t nwritten = 0, totwritten = 0;
   
3. 
   
4.     while(len) {
   
5.         nwritten = write(fd, buf, len);
   
6. 
   
7.         if (nwritten < 0) {
   
8.             if (errno == EINTR) {
   
9.                 continue;
   
10.             }
    
11.             return totwritten ? totwritten : -1;
    
12.         }
    
13. 
    
14.         len -= nwritten;
    
15.         buf += nwritten;
    
16.         totwritten += nwritten;
    
17.     }
    
18. 
    
19.     return totwritten;
    
20. }

将aof的缓冲区（将aof选项置为 everysec的时候，中间会有缓冲），写入磁盘的代码如下，其中需要判断后台是否有fsync正在执行（如果正在执行，会阻塞write调用），如果有，则会延迟，但是如果force参数被设置的话，就啥都不管不顾了，直接开整~

点击(此处)折叠或打开

1. #define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */
   
2. void flushAppendOnlyFile(int force) {
   
3.     ssize_t nwritten;
   
4.     int sync_in_progress = 0;
   
5.     mstime_t latency;
   
6. 
   
7.     if (sdslen(server.aof_buf) == 0) return;
   
8. 
   
9.     if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
   
10.         sync_in_progress = bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;//bio有讲过
    
11. 
    
12.     if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
    
13.         /* With this append fsync policy we do background fsyncing.
    
14.          * If the fsync is still in progress we can try to delay
    
15.          * the write for a couple of seconds. */
    
16.         if (sync_in_progress) {
    
17.             if (server.aof_flush_postponed_start == 0) {
    
18.                 /* No previous write postponing, remember that we are
    
19.                  * postponing the flush and return. */
    
20.                 server.aof_flush_postponed_start = server.unixtime;
    
21.                 return;
    
22.             } else if (server.unixtime - server.aof_flush_postponed_start < 2) {
    
23.                 /* We were already waiting for fsync to finish, but for less
    
24.                  * than two seconds this is still ok. Postpone again. */
    
25.                 return;
    
26.             }
    
27.             /* Otherwise fall trough, and go write since we can't wait
    
28.              * over two seconds. */
    
29.             server.aof_delayed_fsync++;
    
30.             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.");
    
31.         }
    
32.     }
    
33.     /* We want to perform a single write. This should be guaranteed atomic
    
34.      * at least if the filesystem we are writing is a real physical one.
    
35.      * While this will save us against the server being killed I don't think
    
36.      * there is much to do about the whole server stopping for power problems
    
37.      * or alike */
    
38. 
    
39.     latencyStartMonitor(latency);
    
40.     nwritten = aofWrite(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
    
41.     latencyEndMonitor(latency);
    
42.     /* We want to capture different events for delayed writes:
    
43.      * when the delay happens with a pending fsync, or with a saving child
    
44.      * active, and when the above two conditions are missing.
    
45.      * We also use an additional event name to save all samples which is
    
46.      * useful for graphing / monitoring purposes. */
    
47.     if (sync_in_progress) {
    
48.         latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
    
49.     } else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) {
    
50.         latencyAddSampleIfNeeded("aof-write-active-child",latency);
    
51.     } else {
    
52.         latencyAddSampleIfNeeded("aof-write-alone",latency);
    
53.     }
    
54.     latencyAddSampleIfNeeded("aof-write",latency);
    
55. 
    
56.     /* We performed the write so reset the postponed flush sentinel to zero. */
    
57.     server.aof_flush_postponed_start = 0;
    
58. 
    
59.     if (nwritten != (ssize_t)sdslen(server.aof_buf)) {
    
60.         static time_t last_write_error_log = 0;
    
61.         int can_log = 0;
    
62. 
    
63.         /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */
    
64.         if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) {
    
65.             can_log = 1;
    
66.             last_write_error_log = server.unixtime;
    
67.         }
    
68. 
    
69.         /* Log the AOF write error and record the error code. */
    
70.         if (nwritten == -1) {
    
71.             if (can_log) {
    
72.                 serverLog(LL_WARNING,"Error writing to the AOF file: %s",
    
73.                     strerror(errno));
    
74.                 server.aof_last_write_errno = errno;
    
75.             }
    
76.         } else {
    
77.             if (can_log) {
    
78.                 serverLog(LL_WARNING,"Short write while writing to "
    
79.                                        "the AOF file: (nwritten=%lld, "
    
80.                                        "expected=%lld)",
    
81.                                        (long long)nwritten,
    
82.                                        (long long)sdslen(server.aof_buf));
    
83.             }
    
84. 
    
85.             if (ftruncate(server.aof_fd, server.aof_current_size) == -1) {
    
86.                 if (can_log) {
    
87.                     serverLog(LL_WARNING, "Could not remove short write "
    
88.                              "from the append-only file. Redis may refuse "
    
89.                              "to load the AOF the next time it starts. "
    
90.                              "ftruncate: %s", strerror(errno));
    
91.                 }
    
92.             } else {
    
93.                 /* If the ftruncate() succeeded we can set nwritten to
    
94.                  * -1 since there is no longer partial data into the AOF. */
    
95.                 nwritten = -1;
    
96.             }
    
97.             server.aof_last_write_errno = ENOSPC;
    
98.         }
    
99. 
    
100.         /* Handle the AOF write error. */
     
101.         if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
     
102.             /* We can't recover when the fsync policy is ALWAYS since the
     
103.              * reply for the client is already in the output buffers, and we
     
104.              * have the contract with the user that on acknowledged write data
     
105.              * is synced on disk. */
     
106.             serverLog(LL_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
     
107.             exit(1);
     
108.         } else {
     
109.             /* Recover from failed write leaving data into the buffer. However
     
110.              * set an error to stop accepting writes as long as the error
     
111.              * condition is not cleared. */
     
112.             server.aof_last_write_status = C_ERR;
     
113. 
     
114.             /* Trim the sds buffer if there was a partial write, and there
     
115.              * was no way to undo it with ftruncate(2). */
     
116.             if (nwritten > 0) {
     
117.                 server.aof_current_size += nwritten;
     
118.                 sdsrange(server.aof_buf,nwritten,-1);
     
119.             }
     
120.             return; /* We'll try again on the next call... */
     
121.         }
     
122.     } else {
     
123.         /* Successful write(2). If AOF was in error state, restore the
     
124.          * OK state and log the event. */
     
125.         if (server.aof_last_write_status == C_ERR) {
     
126.             serverLog(LL_WARNING,
     
127.                 "AOF write error looks solved, Redis can write again.");
     
128.             server.aof_last_write_status = C_OK;
     
129.         }
     
130.     }
     
131.     server.aof_current_size += nwritten;
     
132. 
     
133.     /* Re-use AOF buffer when it is small enough. The maximum comes from the
     
134.      * arena size of 4k minus some overhead (but is otherwise arbitrary). */
     
135.     if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
     
136.         sdsclear(server.aof_buf);
     
137.     } else {
     
138.         sdsfree(server.aof_buf);
     
139.         server.aof_buf = sdsempty();
     
140.     }
     
141. 
     
142.     /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
     
143.      * children doing I/O in the background. */
     
144.     if (server.aof_no_fsync_on_rewrite &&
     
145.         (server.aof_child_pid != -1 || server.rdb_child_pid != -1))
     
146.             return;
     
147. 
     
148.     /* Perform the fsync if needed. */
     
149.     if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
     
150.         /* aof_fsync is defined as fdatasync() for Linux in order to avoid
     
151.          * flushing metadata. */
     
152.         latencyStartMonitor(latency);
     
153.         aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */
     
154.         latencyEndMonitor(latency);
     
155.         latencyAddSampleIfNeeded("aof-fsync-always",latency);
     
156.         server.aof_last_fsync = server.unixtime;
     
157.     } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
     
158.                 server.unixtime > server.aof_last_fsync)) {
     
159.         if (!sync_in_progress) aof_background_fsync(server.aof_fd);
     
160.         server.aof_last_fsync = server.unixtime;
     
161.     }
     
162. }

另一个需要注意的点就是BGREWRITEAOF命令会重写AOF文件，使AOF文件尽可能的小，其中的大部分操作都是尽可能的使用占用空间小的内存类型，在此不再赘述了。在此期间的命令缓存是通过如下机制实现的。
1.  使用多个缓存block而非一整块大缓存，每个block10M，如下所示

点击(此处)折叠或打开

1. #define AOF_RW_BUF_BLOCK_SIZE (1024*1024*10) /* 10 MB per block */
   
2. 
   
3. typedef struct aofrwblock {
   
4.     unsigned long used, free;
   
5.     char buf[AOF_RW_BUF_BLOCK_SIZE];
   
6. } aofrwblock;

2. 向缓存中写数据的时候是先找到当前链表的最后一个元素，若满足，直接写，若不满足，填上空缺，重新建，继续写

点击(此处)折叠或打开

1. /* Append data to the AOF rewrite buffer, allocating new blocks if needed. */
   
2. void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
   
3.     listNode *ln = listLast(server.aof_rewrite_buf_blocks);
   
4.     aofrwblock *block = ln ? ln->value : NULL;
   
5. 
   
6.     while(len) {
   
7.         /* If we already got at least an allocated block, try appending
   
8.          * at least some piece into it. */
   
9.         if (block) {
   
10.             unsigned long thislen = (block->free < len) ? block-