redis源码分析（5）——aof

标签：源码   redis   c   

1. AOF加载

/* 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加载数据的函数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;

    // <MM>
    // fakeClient对应的文件描述符为-1
    // 响应时，会据此判断是否需要发送响应内容
    // </MM>
    fakeClient = createFakeClient();
    startLoading(fp);

（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();
        }

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

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

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

        // <MM>
        // 依次读取每个bulk
        // </MM>
        for (j = 0; j < argc; j++) {
            // <MM>
            // 处理‘$BULK_LEN\r\n‘
            // </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>
            // 跳过\r\n
            // </MM>
            if (fread(buf,2,1,fp) == 0) {
                fakeClient->argc = j+1; /* Free up to j. */
                freeFakeClientArgv(fakeClient);
                goto readerr; /* discard CRLF */
            }
        }

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

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

最后是各种处理分支：

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;

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

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，提高效率。序列化的流程很简单，对命令序列化，然后追加到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);
}

    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\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
            (unsigned long)strlen(seldb),seldb);
        server.aof_selected_db = dictid;
    }

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

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

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

3. AOF写入

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

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

    ssize_t nwritten;
    int sync_in_progress = 0;
    mstime_t latency;

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

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

    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;

接下来，是对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;
        }

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

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

write调用的错误校验完成，接下来主要是后续的flush策略相关。

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

    // <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();
    }

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

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

上面就是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 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的序列化、写入以及sync的过程，rewrite放到下一篇再写。

redis源码分析（5）——aof

标签：源码   redis   c   

原文地址：http://blog.csdn.net/chosen0ne/article/details/44035453