redis源码阅读之aof

11210阅读　0评论2019-05-27　stolennnxb
分类：NOSQL

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

点击(此处)折叠或打开

  1. #
  2. # If unsure, use "everysec".

  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;

  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);

  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;

  4.     while(len) {
  5.         nwritten = write(fd, buf, len);

  7.         if (nwritten < 0) {
  8.             if (errno == EINTR) {
  9.                 continue;
  10.             }
  11.             return totwritten ? totwritten : -1;
  12.         }

  14.         len -= nwritten;
  15.         buf += nwritten;
  16.         totwritten += nwritten;
  17.     }

  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;

  7.     if (sdslen(server.aof_buf) == 0) return;

  9.     if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
  10.         sync_in_progress = bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;//bio有讲过

  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 */

  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);

  56.     /* We performed the write so reset the postponed flush sentinel to zero. */
  57.     server.aof_flush_postponed_start = 0;

  59.     if (nwritten != (ssize_t)sdslen(server.aof_buf)) {
  60.         static time_t last_write_error_log = 0;
  61.         int can_log = 0;

  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.         }

  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.             }

  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.         }

  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;

  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;

  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.     }

  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;

  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 */

  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;

  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-
上一篇：redis源码阅读之rio
下一篇：redis源码阅读之ae_select