redis源代码分析（5）——aof

前面几篇基本介绍了redis的主要功能、流程。接下来是一些相对独立的部分，首先看一下持久化。

redis持久化支持两种方式：RDB和AOF，我们首先看一下AOF的实现。

AOF（Append only file）实际上就是redis的redo log，在重新启动后，对redo log进行replay以便恢复数据。

正常情况下，为了保证一致性，对于每条命令都要保证其相应的log落地到磁盘。即每条命令相应的日志，要写到文件cache，然后再fsync落地磁盘，这样才干保证强一致性。仅仅要写日志失败。此条命令便运行失败。可是，redis本身是基于内存的，同一时候为了速度，在一致性上进行了折衷。AOF的sync策略分为：

（1）always：对于每条命令都运行fsync，速度慢，可是安全，不会丢数据。

（2）every second：每秒钟运行一次sync。足够快。仅仅会丢失1秒的数据。

（3）never：不进行fsync，全然由os实现数据刷到磁盘，最快，但不能保证数据安全。

随着时间的流逝。redis不断的服务请求，AOF会不断膨胀。一般的db的思路是：snapshot加redo log。定时进行快照，redo log记录当前时间点距上次快照的变化。在恢复时。先载入snapshot，然后再对redo log进行replay。而redis採取的方式稍微不同。它会对AOF进行rewrite。就是依据当前状态，生成一份新的AOF。保证每一个key仅仅会有一份数据，降低不必要的日志。

写AOF的总体流程是：redis会持有一个aof buffer，这个buffer会记录还没有写到文件的aof日志。在每一轮事件循环，运行更新命令时。都会将命令序列化然后追加到aof buffer。然后，在下一轮事件循环前，调用beforeSleep函数时。会将aof buffer写入到文件。

依据sync配置的策略，调用fsync或者调度一个后台job运行fsync。

之所以在beforeSleep中运行，是由于写AOF要在响应内容发送到客户端之前，在下一次事件循环会运行写事件处理函数发送响应内容。

AOF rewrite的总体流程是：rewrite流程的启动是在serverCron中，会创建一个子进程。遍历所有db，写到AOF中。同一时候，开启一个aof rewrite buffer，在命令写入aof buffer时。会推断是否开启rewrite。假设开启。则会同一时候追加到aof rewrite buffer。在子进程完毕rewrite后，会将aof rewrite buffer追加到AOF中，完毕aof rewrite。

1. AOF载入

AOF载入的流程要简单些。在启动后，读取AOF，然后将每条命令进行replay就可以。

以下看一下详细代码。

在redis.c的main函数中。完毕初始化后，会调用loadDataFromDisk()完毕数据的载入。

/* Function called at startup to load RDB or AOF file in memory. */
void loadDataFromDisk(void) {
    long long start = ustime();
    if (server.aof_state == REDIS_AOF_ON) {
        if (loadAppendOnlyFile(server.aof_filename) == REDIS_OK)
            redisLog(REDIS_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
    } else {
        if (rdbLoad(server.rdb_filename) == REDIS_OK) {
            redisLog(REDIS_NOTICE,"DB loaded from disk: %.3f seconds",
                (float)(ustime()-start)/1000000);
        } else if (errno != ENOENT) {
            redisLog(REDIS_WARNING,"Fatal error loading the DB: %s. Exiting.",strerror(errno));
            exit(1);
        }
    }
}

依据当前的持久化方式。分别运行aof或者rdb的数据载入。

以下看一下aof载入数据的函数loadAppendOnlyFile，主要是构建一个fake的client，然后从aof文件里解析并运行一条条命令。

    struct redisClient *fakeClient;
    FILE *fp = fopen(filename,"r");
    struct redis_stat sb;
    int old_aof_state = server.aof_state;
    long loops = 0;
    off_t valid_up_to = 0; /* Offset of the latest well-formed command loaded. */

    if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
        server.aof_current_size = 0;
        fclose(fp);
        return REDIS_ERR;
    }

    if (fp == NULL) {
        redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno));
        exit(1);
    }

    /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
     * to the same file we're about to read. */
    server.aof_state = REDIS_AOF_OFF;

打开aof文件并检查其大小。 

    // <MM>
    // fakeClient相应的文件描写叙述符为-1
    // 响应时。会据此推断是否须要发送响应内容
    // </MM>
    fakeClient = createFakeClient();
    startLoading(fp);

创建fake的client，相应的fd赋值为-1，在响应时。会推断假设fd不为-1，才会加入写事件处理函数。这里设为-1，避免产生响应内容。startLoading会设置状态信息，详细操作包含：

（1）将redisServer.loading置为1，表示当前正处于数据载入阶段。

此时有客户端訪问时，会依据loading状态返回“数据正在载入...”。

（2）将当前时间赋值给redisServer.loading_start_time，用以统计数据载入时间。

（3）将aof文件大小赋值给redisServer.loading_total_bytes，用以统计载入进度

接下来是一个while循环，不断的读取命令并运行。

以下看一下循环内部。

        int argc, j;
        unsigned long len;
        robj **argv;
        char buf[128];
        sds argsds;
        struct redisCommand *cmd;

        /* Serve the clients from time to time */
        if (!(loops++ % 1000)) {
            loadingProgress(ftello(fp));
            // <MM>
            // 处理部分事件
            // 在启动后，载入aof时，此时监听socket已准备好
            // 调用此函数。能够处理客户端的连接，之后也能够响应客户端的请求
            // </MM>
            processEventsWhileBlocked();
        }

loops记录循环次数，在每运行1000次循环时，会更新一下载入进度。同一时候，由于载入过程一般比較长，所以此处会调用processEventsWhileBlocked函数。处理文件io事件。避免客户端一直堵塞。这个函数能够完毕。客户端连接的建立，同一时候响应请求（数据正在载入。不完整。所以响应的内容都是返回错误，并提示“数据正在载入...”）。

接下来是读取aof文件并解析出命令。

        // <MM>
        // 读一行，遇到

        // </MM>
        if (fgets(buf,sizeof(buf),fp) == NULL) {
            // <MM>
            // 读到eof，载入完毕
            // </MM>
            if (feof(fp))
                break;
            else
                goto readerr;
        }
        // <MM>
        // 处理'*MULTI_BULK_LEN
'
        // </MM>
        if (buf[0] != '*') goto fmterr;
        if (buf[1] == '') goto readerr;
        argc = atoi(buf+1);
        if (argc < 1) goto fmterr;

        argv = zmalloc(sizeof(robj*)*argc);
        fakeClient->argc = argc;
        fakeClient->argv = argv;

读取multi bulk的长度，接下来是一个for循环，一次读取每一个bulk。

        // <MM>
        // 依次读取每一个bulk
        // </MM>
        for (j = 0; j < argc; j++) {
            // <MM>
            // 处理'$BULK_LEN
'
            // </MM>
            if (fgets(buf,sizeof(buf),fp) == NULL) {
                fakeClient->argc = j; /* Free up to j-1. */
                freeFakeClientArgv(fakeClient);
                goto readerr;
            }
            if (buf[0] != '$') goto fmterr;
            len = strtol(buf+1,NULL,10);
            // <MM>
            // 分配响应大小的buffer
            // </MM>
            argsds = sdsnewlen(NULL,len);
            // <MM>
            // 二进制读取len大小的buffer
            // </MM>
            if (len && fread(argsds,len,1,fp) == 0) {
                sdsfree(argsds);
                fakeClient->argc = j; /* Free up to j-1. */
                freeFakeClientArgv(fakeClient);
                goto readerr;
            }
            argv[j] = createObject(REDIS_STRING,argsds);
            // <MM>
            // 跳过

            // </MM>
            if (fread(buf,2,1,fp) == 0) {
                fakeClient->argc = j+1; /* Free up to j. */
                freeFakeClientArgv(fakeClient);
                goto readerr; /* discard CRLF */
            }
        }

依次读取每一个bulk。解析出并赋值给fake client。

        /* Command lookup */
        cmd = lookupCommand(argv[0]->ptr);
        if (!cmd) {
            redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", (char*)argv[0]->ptr);
            exit(1);
        }

        /* Run the command in the context of a fake client */
        // <MM>
        // 运行命令的处理函数
        // </MM>
        cmd->proc(fakeClient);

解析出完整命令后，须要运行该命令，首先依据命令名。查找相应的command结构。最后回调命令处理函数。

        /* The fake client should not have a reply */
        // <MM>
        // fake client相应的socket fd为负数
        // 准备响应的函数prepareClientToWrite会据此作推断。不返回响应内容
        // </MM>
        redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
        /* The fake client should never get blocked */
        redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0);

        /* Clean up. Command code may have changed argv/argc so we use the
         * argv/argc of the client instead of the local variables. */
        freeFakeClientArgv(fakeClient);
        if (server.aof_load_truncated) valid_up_to = ftello(fp);

此处进行校验，由于fake client不可能有响应内容，最后清理fake client，以便下一个命令的运行。valid_up_to记录当前正确解析的日志长度。在数据不完整（提前读到eof）而且设置aof_load_truncated时，会将aof文件截断到valid_up_to字节。

最后是各种处理分支：

loaded_ok: /* DB loaded, cleanup and return REDIS_OK to the caller. */
    fclose(fp);
    freeFakeClient(fakeClient);
    server.aof_state = old_aof_state;
    stopLoading();
    aofUpdateCurrentSize();
    server.aof_rewrite_base_size = server.aof_current_size;
    return REDIS_OK;

数据载入正确的情况，会关闭aof文件。释放fake client，更新各种状态等。

readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */
    if (!feof(fp)) {
        redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno));
        exit(1);
    }

命令解析失败，直接退出。

uxeof: /* Unexpected AOF end of file. */
    if (server.aof_load_truncated) {
        redisLog(REDIS_WARNING,"!!! Warning: short read while loading the AOF file !!!");
        redisLog(REDIS_WARNING,"!!! Truncating the AOF at offset %llu !!!",
            (unsigned long long) valid_up_to);
        if (valid_up_to == -1 || truncate(filename,valid_up_to) == -1) {
            if (valid_up_to == -1) {
                redisLog(REDIS_WARNING,"Last valid command offset is invalid");
            } else {
                redisLog(REDIS_WARNING,"Error truncating the AOF file: %s",
                    strerror(errno));
            }
        } else {
            /* Make sure the AOF file descriptor points to the end of the
             * file after the truncate call. */
            if (server.aof_fd != -1 && lseek(server.aof_fd,0,SEEK_END) == -1) {
                redisLog(REDIS_WARNING,"Can't seek the end of the AOF file: %s",
                    strerror(errno));
            } else {
                redisLog(REDIS_WARNING,
                    "AOF loaded anyway because aof-load-truncated is enabled");
                goto loaded_ok;
            }
        }
    }
    redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file. You can: 1) Make a backup of your AOF file, then use ./redis-check-aof --fix <filename>. 2) Alternatively you can set the 'aof-load-truncated' configuration option to yes and restart the server.");
    exit(1);

读到非预期的eof。即最后一条命令不完整。假设设置了aof_load_truncated。会将aof文件截断到valid_up_to，否则，直接退出。

fmterr: /* Format error. */
    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>");
    exit(1);

最后是命令的格式不对，直接退出。

2. AOF序列化

AOF要记录每条命令对数据库的更改，所以须要记录每条更新命令。redis会持有一个aof buffer，用于在一轮事件循环中，记录多天命令，然后在调用一次write进行写入，避免一个命令一次write，提高效率。序列化的流程非常easy。对命令序列化，然后追加到aof buffer后面。

在介绍请求处理时。我们知道对于每条命令都会调用call函数处理。当中，会调用propagate函数处理主从复制和AOF。

void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
               int flags)
{
    if (server.aof_state != REDIS_AOF_OFF && flags & REDIS_PROPAGATE_AOF)
        feedAppendOnlyFile(cmd,dbid,argv,argc);
    if (flags & REDIS_PROPAGATE_REPL)
        replicationFeedSlaves(server.slaves,dbid,argv,argc);
}

在redis开启aof，而且该命令须要记录aof时。会调用feedAppendOnlyFile函数用于生成并写入aof。以下看一下这个函数。

    sds buf = sdsempty();
    robj *tmpargv[3];

    /* The DB this command was targeting is not the same as the last command
     * we appendend. To issue a SELECT command is needed. */
    // <MM>
    // 当前操作的db与aof相应的db不同一时候，须要一个切换db的命令
    // </MM>
    if (dictid != server.aof_selected_db) {
        char seldb[64];

        snprintf(seldb,sizeof(seldb),"%d",dictid);
        buf = sdscatprintf(buf,"*2
$6
SELECT
$%lu
%s
",
            (unsigned long)strlen(seldb),seldb);
        server.aof_selected_db = dictid;
    }

在全局server结构中得aof_selected_db记录当前aof相应的数据库，假设当前命令操作的数据库与之不同的话。首先须要切换数据库。上述代码就是用于生产select db命令的。

    // <MM>
    // 将命令序列化。并保存到buf
    // </MM>
    if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
        cmd->proc == expireatCommand) {
        /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */
        buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
    } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) {
        /* Translate SETEX/PSETEX to SET and PEXPIREAT */
        tmpargv[0] = createStringObject("SET",3);
        tmpargv[1] = argv[1];
        tmpargv[2] = argv[3];
        buf = catAppendOnlyGenericCommand(buf,3,tmpargv);
        decrRefCount(tmpargv[0]);
        buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
    } else {
        /* All the other commands don't need translation or need the
         * same translation already operated in the command vector
         * for the replication itself. */
        buf = catAppendOnlyGenericCommand(buf,argc,argv);
    }

接下来。将命令序列化为aof，详细序列化过程再次不赘述。这里应该能够优化。在读取命令buffer时。保存此buffer。命令參数使用指针指向该buffer，便能够节省次数序列化的开销。

    // <MW>
    // 为什么不提前推断？这会浪费资源
    // </MW>
    /* Append to the AOF buffer. This will be flushed on disk just before
     * of re-entering the event loop, so before the client will get a
     * positive reply about the operation performed. */
    if (server.aof_state == REDIS_AOF_ON)
        // <MM>
        // 将命令buf追加到aof_buf
        // </MM>
        server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf));

推断是否开启aof，假设开启则将aof追加到aof buffer。此处，应该能够提前推断，避免关闭aof时的aof的序列化开销。

    /* If a background append only file rewriting is in progress we want to
     * accumulate the differences between the child DB and the current one
     * in a buffer, so that when the child process will do its work we
     * can append the differences to the new append only file. */
    // <MM>
    // 假设开启了aof rewrite进程，将命令也加入到aof rewrite buf中
    // 等rewrite完之后。在将rewrite buf的数据追加到文件里
    // </MM>
    if (server.aof_child_pid != -1)
        aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));

    sdsfree(buf);

aof_child_pid记录aof rewrite进程的pid，假设rewrite正在进行，这个值不为-1。假设当前正在进行aof rewrite，则将命令的aof追加到aof rewrite buffer，待rewrite结束后进行replay。

3. AOF写入

AOF的写入就是将aof buffer写入到aof文件里，write系统调用仅仅能保证写入page cache中，要落地到磁盘还须要调用fsync。所以，涉及到fsync的策略。这个函数会稍微复杂一些。在beforeSleep函数中，会调用flushAppendOnlyFile函数进行写入。

    /* Write the AOF buffer on disk */
    flushAppendOnlyFile(0);

之所以。在beforeSleep中。是为了在给客户端发送响应内容前进行。保证返回给客户端的内容都是写过aof的。

同一时候，也保证一轮事件循环。对于多个客户端的请求处理仅仅写一次aof，提升性能（当然，这样做的缺点就是不能保证数据的一致性）。以下看一下flushAppendOnlyFile函数。

    ssize_t nwritten;
    int sync_in_progress = 0;
    mstime_t latency;

    // <MM>
    // 没有aof须要write，直接返回
    // </MM>
    if (sdslen(server.aof_buf) == 0) return;

检查aof buffer是否为空，空的话直接返回，不是必需进行flush。

    if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
        sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0;

fsync是堵塞操作。避免影响主线程的事件循环，fsync操作由后台线程完毕。

假设设置的fsync策略是everysec，获取是否有后台线程正在进行fsync。

    if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
        /* With this append fsync policy we do background fsyncing.
         * If the fsync is still in progress we can try to delay
         * the write for a couple of seconds. */
        if (sync_in_progress) {
            // <MM>
            // aof_flush_postponed_start记录从什么时候開始延迟flush
            // </MM>
            if (server.aof_flush_postponed_start == 0) {
                /* No previous write postponinig, remember that we are
                 * postponing the flush and return. */
                server.aof_flush_postponed_start = server.unixtime;
                return;
            } else if (server.unixtime - server.aof_flush_postponed_start < 2) {
                // <MM>
                // 尚未flush的aof buf不超过1s，没有违反every_sec策略。此次也不进行flush
                // </MM>
                /* We were already waiting for fsync to finish, but for less
                 * than two seconds this is still ok. Postpone again. */
                return;
            }
            /* Otherwise fall trough, and go write since we can't wait
             * over two seconds. */
            server.aof_delayed_fsync++;
            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.");
        }
    }

在fsync策略是everysec时，这段代码用于控制flush的频率。server.aof_flush_postponed_start记录上次延迟flush的时间戳，假设等于0，说明没有延迟。

假设是everysec的fsync策略。而且当前正在进行fsync，这里会设置aof_flush_postponed_start。假设当前时间戳server.unixtime与延迟flush的时间戳间隔小于2s。那么没有违反everysec策略，不进行flush，直接返回。

通过这段代码能够保证1秒内，不会flush多次。

假设没有当前没有进行fsync。或者当前时间戳server.unixtime与延迟flush的时间戳间隔大于2s，就会跳过这段代码，进行flush操作。

    latencyStartMonitor(latency);
    // <MM>
    // 将aof写入日志，此处仅仅是写入page cache。还须要fsync
    // </MM>
    nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
    latencyEndMonitor(latency);
    /* We want to capture different events for delayed writes:
     * when the delay happens with a pending fsync, or with a saving child
     * active, and when the above two conditions are missing.
     * We also use an additional event name to save all samples which is
     * useful for graphing / monitoring purposes. */
    if (sync_in_progress) {
        latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
    } else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) {
        latencyAddSampleIfNeeded("aof-write-active-child",latency);
    } else {
        latencyAddSampleIfNeeded("aof-write-alone",latency);
    }
    latencyAddSampleIfNeeded("aof-write",latency);

接下来，调用write进行写入。同一时候会记录各种延迟。write时。是将整个aof_buf进行写入。

这里能够看到，假设fsync的策略是everysec，那么write也是每秒钟调用一次。实际上，这存在一个缺陷：即在机器没有掉电的情况下，redis挂了，也会最多丢失1秒的数据。假设不限制每秒调用一次write，而是每轮事件循环都调用write，就能够保证数据已经写入page cache，仅仅要机器没挂，终于数据都会写入磁盘，就不会丢失数据。

本身write是写cache，不存在性能瓶颈，所以这里能够改进一下。

    /* We performed the write so reset the postponed flush sentinel to zero. */
    // <MM>
    // 清空。当前没有延迟flush aof
    // </MM>
    server.aof_flush_postponed_start = 0;

重置aof_flush_postponed_start。由于接下来会进行flush。

接下来，是对write调用进行错误检验。会有一个if分支进行

    if (nwritten != (signed)sdslen(server.aof_buf)) {
        // 错误分支
    } else {
        // 正常分支
    }

首先看一下错误分支。

        static time_t last_write_error_log = 0;
        int can_log = 0;

        /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */
        // <MM>
        // 限制记录错误日志的频率
        // </MM>
        if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) {
            can_log = 1;
            last_write_error_log = server.unixtime;
        }

last_write_error_log是static类型。记录上一次记录错误日志的时间戳，这段代码就是用于控制记录日志的频率，避免日志刷屏。

        /* Lof the AOF write error and record the error code. */
        if (nwritten == -1) {
            if (can_log) {
                redisLog(REDIS_WARNING,"Error writing to the AOF file: %s",
                    strerror(errno));
                server.aof_last_write_errno = errno;
            }
        } else {

write返回值是-1，说明调用错误。仅仅记录日志。接下来是处理部分写的情况。

        } else {
            if (can_log) {
                redisLog(REDIS_WARNING,"Short write while writing to "
                                       "the AOF file: (nwritten=%lld, "
                                       "expected=%lld)",
                                       (long long)nwritten,
                                       (long long)sdslen(server.aof_buf));
            }

            // <MM>
            // aof_current_size记录当前正确写入的aof的长度
            // 当前write仅仅写入部分数据，此处保证完整性，将写入的部分数据删掉
            // </MM>
            if (ftruncate(server.aof_fd, server.aof_current_size) == -1) {
                if (can_log) {
                    redisLog(REDIS_WARNING, "Could not remove short write "
                             "from the append-only file.  Redis may refuse "
                             "to load the AOF the next time it starts.  "
                             "ftruncate: %s", strerror(errno));
                }
            } else {
                /* If the ftrunacate() succeeded we can set nwritten to
                 * -1 since there is no longer partial data into the AOF. */
                nwritten = -1;
            }
            server.aof_last_write_errno = ENOSPC;
        }

在部分写的情况发生时，会将部分写入的内容截掉，保证aof中的是完整的。

server.aof_current_size记录当前正确写入的aof的长度。后面会对这个值进行更新。

假设ftruncate成功，会设置nwritten为-1。假设失败的话，后面代码会将aof_current_size加上部分写的数据长度，同一时候将aof_buf中截取已写入部分。

接下来处理aof write失败。

        /* Handle the AOF write error. */
        if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
            /* We can't recover when the fsync policy is ALWAYS since the
             * reply for the client is already in the output buffers, and we
             * have the contract with the user that on acknowledged write data
             * is synched on disk. */
            redisLog(REDIS_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
            // <MM>
            // fsync策略是always。write失败后，不能恢复。直接退出
            // </MM>
            exit(1);
        } else {
            /* Recover from failed write leaving data into the buffer. However
             * set an error to stop accepting writes as long as the error
             * condition is not cleared. */
            server.aof_last_write_status = REDIS_ERR;

            /* Trim the sds buffer if there was a partial write, and there
             * was no way to undo it with ftruncate(2). */
            if (nwritten > 0) {
                // <MM>
                // 在ftruncate失败时，走这个分支，会把已写的aof从buffer中清空
                // </MM>
                server.aof_current_size += nwritten;
                sdsrange(server.aof_buf,nwritten,-1);
            }

            // <MM>
            // write失败，下次会进行重试
            // </MM>
            return; /* We'll try again on the next call... */
        }

假设fsync策略是always。那么write失败，就表示整个操作失败，保证强一致性。此处进程退出。

假设是其它策略，会依据nwritten，更新aof_current_size并调整aof_buf。

上面就是write的错误分支，以下看一下正常分支。

        /* Successful write(2). If AOF was in error state, restore the
         * OK state and log the event. */
        if (server.aof_last_write_status == REDIS_ERR) {
            redisLog(REDIS_WARNING,
                "AOF write error looks solved, Redis can write again.");
            server.aof_last_write_status = REDIS_OK;
        }

假设fsync策略不是always，在write出错时。会有server.aof_last_write_status记录错误状态。

假设兴许的write操作正常，此处仅仅是打印日志。表示错误恢复正常。

write调用的错误校验完毕，接下来主要是兴许的flush策略相关。

    // <MM>
    // write成功。更新aof文件的大小
    // </MM>
    server.aof_current_size += nwritten;

write成功时。更新aof_current_size。

    // <MM>
    // aof buf已成功write，此处序清空buffer
    // </MM>
    /* Re-use AOF buffer when it is small enough. The maximum comes from the
     * arena size of 4k minus some overhead (but is otherwise arbitrary). */
    if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
        sdsclear(server.aof_buf);
    } else {
        sdsfree(server.aof_buf);
        server.aof_buf = sdsempty();
    }

aof_buf已成功写入文件，能够清空。为避免频繁分配、释放内存，此处保证在buf小于4K时，会一直重用该buf。

假设大于4K。就会释放旧的buf，分配新的。

    /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
     * children doing I/O in the background. */
    if (server.aof_no_fsync_on_rewrite &&
        (server.aof_child_pid != -1 || server.rdb_child_pid != -1))
            return;

假设配置了no-appendfsync-on-rewrite。即在有aof rewrite或者是rdb save的子进程时不进行fsync，主要是避免对磁盘产生过大压力，这里会直接返回，不进行fsync。

    /* Perform the fsync if needed. */
    // <MM>
    // 1) always策略：每次write，都会调用fsync
    // 2) everysec策略：当大于上次fsync的时间（秒数）时，才会调度后台线程运行
    // </MM>
    if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
        /* aof_fsync is defined as fdatasync() for Linux in order to avoid
         * flushing metadata. */
        latencyStartMonitor(latency);
        aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */
        latencyEndMonitor(latency);
        latencyAddSampleIfNeeded("aof-fsync-always",latency);
        server.aof_last_fsync = server.unixtime;
    } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
                // <MM>
                // 通过当前时间戳和上次aof sync的时间比較，
                // 仅仅有比上次sync大时。才会启动后台sync操作
                // </MM>
                server.unixtime > server.aof_last_fsync)) {
        if (!sync_in_progress) aof_background_fsync(server.aof_fd);
        server.aof_last_fsync = server.unixtime;
    }

接下来，就是fsync相关。假设策略是always。直接进行fsync。记录延迟，同一时候更新aof_last_fsync。假设是everysec策略，而且server.unixtime > server.aof_last_fsync（保证一秒内不进行多次fsync），而且没有后台线程运行fsync。则调度后台线程进行fsync。

上面就是flush的所有流程。这个函数除了在beforeSleep中调用，在定时器事件处理函数serverCron中也会调用。

    /* AOF postponed flush: Try at every cron cycle if the slow fsync
     * completed. */
    // <MW>
    // 在有延迟flush aof的情况下，才会调用。主要是在fsync完毕后
    // 尽快进行下一次write aof
    // 可是。serverCron运行后，立马就会运行beforeSleep，有这个必要在这运行么？
    // </MW>
    if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);

在aof_flush_postponed_start不为0时调用。即存在延迟flush的情况。主要是保证fsync完毕之后。能够高速的进入下一次flush。尽量保证fsync策略是everysec时。每秒都能够进行fsync。同一时候缩短两次fsync的间隔，降低影响。

    /* AOF write errors: in this case we have a buffer to flush as well and
     * clear the AOF error in case of success to make the DB writable again,
     * however to try every second is enough in case of 'hz' is set to
     * an higher frequency. */
    run_with_period(1000) {
        if (server.aof_last_write_status == REDIS_ERR)
            flushAppendOnlyFile(0);
    }

另一处调用，是保证aof出错时，尽快运行下一次flush，以便从错误恢复。

上面便是aof的序列化、写入以及sync的过程，rewrite放到下一篇再写。