目录
AOF日志
1. 持久化——命令写入到AOF文件
写到用户缓冲区
AOF的触发入口函数——propagate
具体的实现逻辑——feedAppendOnlyFile
从用户缓冲区写入到AOF文件(磁盘)
函数write、fsync、fdatasync
Redis的线程池
AOF文件的同步策略
触发的入口函数——flushAppendOnlyFile
2. AOF重写
AOF 重写的2个触发时机
用户发送 bgrewriteaof 命令
在定时函数serverCron中触发
父子进程使用pipe进行通信
两个缓冲区——重写缓冲 和 差异缓冲
重写缓冲
差异缓冲
rewriteAppendOnlyFileBackground的实现
执行重写过程的函数——rewriteAppendOnlyFile
父进程监听子进程结束, AOF 重写收尾
在定时函数serverCron中监听
主进程对AOF重写收尾——backgroundRewriteDoneHandler
3. Redis重启,AOF 文件加载
Redis是把数据储存在内存的键值数据库,而服务器一旦宕机,那内存中的数据将全部丢失。像MySQL那样,是有宕机后数据恢复机制的。那Redis也是有的,其有两种方式:AOF和RDB。该文章讲解AOF。
AOF日志
MySQL是使用redo log(重做日志)来进行宕机恢复的。其是使用了写前日志(Write Ahead Log,WAL),即是在实际写数据前,先把修改的数据写到日志文件中,方便出故障时候进行恢复。
而AOF正好相反,是写后日志,即是先执行命令把数据写到内存,之后再把该操作记录到日志中。这是个文本日志,不是二进制文件。
那该日志主要有3个操作:
- AOF持久化(同步):客户端向Redis服务器发送命令,这些命令会被存储到AOF缓冲区中,并随后会持久化到磁盘文件中
- AOF重写:随着写入的内容越来越多,就会占用大量的磁盘空间,并且Redis重启时候需要按照顺序执行AOF中的命令,这样时间就比较长,所以Redis 会定期重写 AOF 日志,以达到文件瘦身的效果和缩短重启恢复所需的时间。
- 重启数据恢复:Redis重启后,通过AOF来进行数据恢复
1. 持久化——命令写入到AOF文件
写到用户缓冲区
首先,写入到AOF的命令是先存储在一个AOF缓冲区。
struct redisServer {
.........
sds aof_buf; /* AOF buffer, written before entering the event loop */
};
客户端发送的命令转为RESP协议格式的字符串,然后追加到已有的字符串后面,这些都是存储在aof_buf中。
AOF的触发入口函数——propagate
单线程情况下,其函数被调用的流程:readQueryFromClient——>processInputBuffer——>processCommandAndResetClient——>processCommand——> call(client *c, int flags) ——>propagate。
void call(client *c, int flags) {
/* Call the command. */
c->cmd->proc(c);
........................
// 入参的 flags 设置了 CMD_CALL_PROPAGATE 标识, 表示当前的命令需要传播
// 同时对应的客户端内部的标识不是 CLIENT_PREVENT_PROP (客户端的命令阻止传播)
if (flags & CMD_CALL_PROPAGATE && (c->flags & CLIENT_PREVENT_PROP) != CLIENT_PREVENT_PROP)
{
int propagate_flags = PROPAGATE_NONE;
if (dirty) propagate_flags |= (PROPAGATE_AOF|PROPAGATE_REPL);
// 当前的客户端设置了需要强制同步传播,或者设置了 需要强制 AOF 传播
if (c->flags & CLIENT_FORCE_REPL) propagate_flags |= PROPAGATE_REPL;
if (c->flags & CLIENT_FORCE_AOF) propagate_flags |= PROPAGATE_AOF;
// 与客户端c的flags对比,若是符合条件,取消 命令传播标识的repl或者aof
if (c->flags & CLIENT_PREVENT_REPL_PROP || !(flags & CMD_CALL_PROPAGATE_REPL))
propagate_flags &= ~PROPAGATE_REPL;
if (c->flags & CLIENT_PREVENT_AOF_PROP || !(flags & CMD_CALL_PROPAGATE_AOF))
propagate_flags &= ~PROPAGATE_AOF;
// 命令传播标识 不为 none, 且当前的命令不是模块命令
if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE))
// 处理aof 和 复制给副本
propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags);
}
..................................
}
// 将命令写到aof 文件,并将命令发送给副本
void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
int flags)
{
if (!server.replication_allowed)
return;
// AOF 开启了, 同时命令传播标识为 需要 AOF 传播
if (server.aof_state != AOF_OFF && flags & PROPAGATE_AOF)
feedAppendOnlyFile(cmd,dbid,argv,argc); // 将当前的命令保存到 AOF 缓冲区
..................
}
具体的实现逻辑——feedAppendOnlyFile
该函数就是把命令写入到aof缓冲区。
- 创建一个SDS对象buf,用户把命令写入到该对象。判断该命令使用的数据库号是否是用户选择的数据库号,若不是就需要在aof文件中添加选择数据库。
- 把命令写入到buf。
- 对于
EXPIRE
、EXPIREAT
和PEXPIRE
将其转换为PEXPIREAT
特殊处理。 - 对于带
EX
、PX
参数的SET
命令特殊处理,主要涉及过期时间的处理; - 对于其它命令,调用
catAppendOnlyGenericCommand
按照 RESP 协议组装命令,并将其暂存至 buf;
- 对于
- 如果启用 AOF 日志,则将 buf 中暂存的命令追加到 AOF缓冲区
server.aof_buf。
- 如果存在正在重写 AOF 的子进程,则将命令追加到 AOF 重写缓冲区
server.aof_rewrite_buf_blocks
void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
sds buf = sdsempty();
//该命令写入的数据库和用户选择的数据库不一致的话,需要在aof文件添加一段选择数据库的记录
if (dictid != server.aof_selected_db) {
char seldb[64];
snprintf(seldb,sizeof(seldb),"%d",dictid);
// 拼接出一个 select 数据库号 的语句, 这个语句是遵守 RESP 协议
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;
}
//这三个命令, 在 AOF 保存的时候, 都会转为 expireat key 具体的过期时间 (单位毫秒) 的格式存入到 AOF 文件中
if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
cmd->proc == expireatCommand) {
// 转为过期对应的文本, 同时追加到 buf 中
buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
} else if (cmd->proc == setCommand && argc > 3) {
//带 EX、PX 参数的 SET 命令,特殊处理, set key value ex seconds, set key value px milliseconds
robj *pxarg = NULL;
if (!strcasecmp(argv[3]->ptr, "px")) { //过期时间是毫秒的
pxarg = argv[4];
}
if (pxarg) { //毫秒的
robj *millisecond = getDecodedObject(pxarg);
long long when = strtoll(millisecond->ptr,NULL,10);
when += mstime();
decrRefCount(millisecond);
robj *newargs[5];
newargs[0] = argv[0];
newargs[1] = argv[1];
newargs[2] = argv[2];
newargs[3] = shared.pxat;
newargs[4] = createStringObjectFromLongLong(when);
// 往 buf 中追加 set 命令
buf = catAppendOnlyGenericCommand(buf,5,newargs);
// 创建的对象手动修改引用计数, 便于内存回收
decrRefCount(newargs[4]);
} else { //秒过期的
buf = catAppendOnlyGenericCommand(buf,argc,argv);
}
} else {
// 其他的命令直接转为 RESP 协议的字符串进行追加
buf = catAppendOnlyGenericCommand(buf,argc,argv);
}
//将组装好的命令追加到 aof_buf
if (server.aof_state == AOF_ON)
server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf));
// CHILD_TYPE_AOF表明后台正在进行重写,那么将命令再追加一份到重写缓冲区中,以便我们记录重写时 AOF 文件和当前数据库的差异
if (server.child_type == CHILD_TYPE_AOF)
aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));
sdsfree(buf);
}
从用户缓冲区写入到AOF文件(磁盘)
函数write、fsync、fdatasync
- write只是将数据保存到系统缓冲区或者用户缓冲区,还没有真正落入到磁盘中的
- fsync是真正地把数据写入到磁盘,即是把缓冲区中的数据落入磁盘。
- POSIX 标准定义的
fsync
函数在文件元数据(metadata,例如st_size
、st_atime
、st_mtime
等)变脏时,会将所有元数据同步到磁盘。由于每次同步都必定导致时间戳的改变,而且文件内容和文件元数据通常存储在磁盘上的不同位置,因此每次调用fsync
至少需要两次随机磁盘 I/O。
- POSIX 标准定义的
- 为此,Linux 平台提供了一个
fdatasync
函数。该函数仅在必要时才将元数据同步到磁盘(文件读写时间戳等信息的改变不会实时落盘),大大降低了元数据同步的频率。- 举例来说,文件的尺寸(st_size)如果变化,是需要立即同步的,否则OS一旦崩溃,即使文件的数据部分已同步,由于metadata没有同步,依然读不到修改的内容。而最后访问时间(atime)/修改时间(mtime)是不需要每次都同步的,只要应用程序对这两个时间戳没有苛刻的要求,基本无伤大雅
Redis 通过条件编译,将 Linux 平台的 redis_fsync
定义成了 fdatasync
,而在其它类 Unix 平台上依旧是 fsync
:
#ifdef __linux__
#define redis_fsync fdatasync
#else
#define redis_fsync fsync
#endif
Redis的线程池
真正写入到磁盘的是使用fsync函数,那说明该函数是相对比较耗时的。Redis维护了一个线程池,就是用来处理一些比较耗时的操作。
那么,AOF缓冲区写入到AOF文件(存入到磁盘)过程中,会先通过write将数据写入到系统缓存,然后根据当前的AOF保存策略,决定是否需要把fsync函数的执行交给线程池。
AOF文件的同步策略
- no:不进行同步,每个写命令执行完后,只是先把记录写到AOF文件中的内存缓冲区中,由操作系统决定合适将缓冲区内存写回磁盘。
- always:每次write后,都会立即执行fsync,这种就是在主线程中执行fsync。
- everysec:每次write后,不会立即执行fsync,理论是每秒执行一次fsync,同时内部将fysnc的执行交给线程池去处理。
触发的入口函数——flushAppendOnlyFile
将缓冲区中的数据写入到aof文件的函数是flushAppendOnlyFile。
在Redis中有5处会调用该函数
- 通过命令动态关闭AOF功能时,会进行一次保存,即是发送命令将appendonly yes设置为appendonly no。
- 在Redis正常关闭之前,会执行该函数。
- 在事件循环中的beforesleep函数中会调用一次,这个是AOF功能的主要的保存入口
- Redis的定时函数serverCron(默认100毫秒执行一次)
- 定时函数serverCron,判断上次AOF写入状态,失败就执行一次该函数。
定时函数serverCron关于这部分的代码。
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
..................
//上次的 AOF 写文件时, 没有执行, 将 aof_flush_postponed_start 设置为true, 表示需要延迟处理
//存在延迟的AOF落盘操作,在这里完成
if (server.aof_state == AOF_ON && server.aof_flush_postponed_start)
flushAppendOnlyFile(0);
run_with_period(1000) {
//上次的写文件失败,即是fync失败
if (server.aof_state == AOF_ON && server.aof_last_write_status == C_ERR)
flushAppendOnlyFile(0);
}
...................
}
存在延迟的AOF落盘操作
比如:主线程在执行flushAppendOnlyFile中调用write后,提交一个任务给后台线程,假设此时数据量很大,fsync需要执行较长时间。而主线程又执行到了flushAppendOnlyFile,而上一次的fsync函数还没有执行完,Redis会选择延迟执行,将Server成员变量aof_flush_postponed_start设置为当前时间,就结束该函数。
所以在执行定时任务时候,会判断该变量是否>0,若是,会再执行flushAppendOnlyFile,这个就是AOF同步延迟到定时函数处执行。
但是,延迟到定时任务处触发, 还是无法保证后台线程一定执行完上次的 fsync。所以该函数会根据当前的时间和变量储存的时间进行判断,若是在2s内,就不做任何处理,退出该函数;而大于2s,立即执行AOF缓冲区写入文件的逻辑。
flushAppendOnlyFile的实现
- 如果AOF缓冲区为空, 并且AOF策略是everysec,同步到磁盘的内容大小不等于当前AOF文件的内容大小,当前时间 > 上次AOF fsync的时间,同时当前没有正在运行的bio后台任务,则尝试执行fsync。
- 如果策略是everysec,且后台存在正在同步的bio线程,则判断
aof_flush_postponed_start是否为0:
若是0,表示之前没有延迟落盘任务,所以就只记录当前的时间戳给aof_flush_postponed_start并退出。
若是不为0,但判断距离aof_flush_postponed_start是否已经过去2s,若是就增加server.aof_delayed_fsync
计数,强制后续的磁盘同步流程
- 调用aofWrite将AOF缓冲区中的数据写入到系统内核缓冲区(这时是还没有使用fsync),若是写入到系统的数据长度不等于当前 AOF 缓冲区的长度, 需要进行异常处理
- 如果aof_buf的总空间小于4kb,则清空buffer内容并重新使用该缓冲区,否则创建一个新的。
// AOF 缓冲区数据写入文件
// 当持久策略被设置为 everysec, 实际上会由后台线程进行处理, 那么当前这次刷新写入时, 后台可能有线程还在写入, 所以这时的操作会延迟写入
//参数force 1:表示无视后台的 fsync, 直接写入, 0: 表示可以延迟, 一般 AOF 过程都是 0
#define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */
void flushAppendOnlyFile(int force) {
ssize_t nwritten;
int sync_in_progress = 0;
mstime_t latency;
//表示aof缓冲区中没有数据, 就可以结束了,但是Redis中有一些极端情况,不会结束,当前学习可以不用了解,后序熟悉该代码了再回头看
if (sdslen(server.aof_buf) == 0) {
if (server.aof_fsync == AOF_FSYNC_EVERYSEC &&
server.aof_fsync_offset != server.aof_current_size &&
server.unixtime > server.aof_last_fsync &&
!(sync_in_progress = aofFsyncInProgress())) {
goto try_fsync;
} else {
return;
}
}
// 持久策略为每秒 fsync 一次, 判断后台的线程池是否有线程在执行 fsync
if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
sync_in_progress = aofFsyncInProgress();//该返回值为true,表示当前有BIO线程在执行 fsync
// 持久策略为每秒 fsync 一次, 同时不需要强制写入文件
if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
// 当前有 BIO 线程在执行 fsync
if (sync_in_progress) {
if (server.aof_flush_postponed_start == 0) { // 0 表示当前没有延迟执行
//当前有后台线程在执行 fsync, 那么先延长一下, 设置aof_flush_postponed_start 为当前时间
server.aof_flush_postponed_start = server.unixtime;
return;
//若之前是偶延迟执行,然后又进入了该函数(一般是执行定时函数触发),那这时后台还在执行fsync,但是当前时间和上一次设置的延迟时间小于2s,可以接受,就暂时不做处理
} else if (server.unixtime - server.aof_flush_postponed_start < 2) {
/* We were already waiting for fsync to finish, but for less
* than two seconds this is still ok. Postpone again. */
return;
}
//到了这一步表示线程池中有请求 fsync 的任务, 同时上次延迟距离当前时间超过 2 秒了
server.aof_delayed_fsync++; // 延迟 fsync 的次数 + 1
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.");
}
}
if (server.aof_flush_sleep && sdslen(server.aof_buf)) {
usleep(server.aof_flush_sleep);
}
latencyStartMonitor(latency);
//调用 write 函数将缓冲区中的数据写入到文件 (此时还在系统缓存, 还没写入到磁盘
nwritten = aofWrite(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
latencyEndMonitor(latency);
if (sync_in_progress) {
latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
} else if (hasActiveChildProcess()) {
latencyAddSampleIfNeeded("aof-write-active-child",latency);
} else {
latencyAddSampleIfNeeded("aof-write-alone",latency);
}
latencyAddSampleIfNeeded("aof-write",latency);
//将缓冲区中的数据 write 到系统后, 可以把延迟执行设置为 0
server.aof_flush_postponed_start = 0;
// 写入到系统的数据长度不等于当前 AOF 缓冲区的长度, 进入异常处理
if (nwritten != (ssize_t)sdslen(server.aof_buf)) {
static time_t last_write_error_log = 0;
if (nwritten == -1) { // -1, 没有写入任何数据, 就直接失败了
server.aof_last_write_errno = errno;
}
} else {// 大于 -1 但是不等于缓冲区的大小, 写入成功了一部分,
if (ftruncate(server.aof_fd, server.aof_current_size) == -1) {
//写错误日志..............
}
} else {
/* If the ftruncate() 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;
}
// 同步策略为 always
if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
// 这种情况无法处理了, 已经告知客户端写入成功了, 但是当前写入失败了, 直接退出程序。
serverLog(LL_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
exit(1);
} else {
// 设置上一次写入状态为异常, 在定时器中会判断这个状态, 再次触发 flushAppendOnlyFile
server.aof_last_write_status = C_ERR;
if (nwritten > 0) {
// 更新当前 aof 文件的大小, 同时将缓冲区中这部分大小的数据移除
// 表示这部分写入成功了, 剩余部分下次调用继续
server.aof_current_size += nwritten;
sdsrange(server.aof_buf,nwritten,-1);
}
return; /* We'll try again on the next call... */
}
} else { //写入成功
if (server.aof_last_write_status == C_ERR) {
server.aof_last_write_status = C_OK;
}
}
server.aof_current_size += nwritten; // 更新当前 AOF 文件的大小
//如果当前 AOF 缓冲区足够小,小于4K,那么重用这个缓存,否则释放 AOF 缓冲区, 然后重新分配一个
if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
sdsclear(server.aof_buf);
} else {
sdsfree(server.aof_buf);
server.aof_buf = sdsempty();
}
try_fsync:
//aof缓冲区中没有数据,但是有一些特例情况的需要处理的..........
}
try_fsync部分的代码:
void flushAppendOnlyFile(int force) {
if (sdslen(server.aof_buf) == 0) {
if (server.aof_fsync == AOF_FSYNC_EVERYSEC && server.aof_fsync_offset != server.aof_current_size &&
server.unixtime > server.aof_last_fsync && !(sync_in_progress = aofFsyncInProgress())) {
goto try_fsync;
}
.......................
}
..........................
try_fsync:
// no-appendfsync-on-rewrite (正在重写, 不执行 fsync) 被设置为 yes
//判断是否有运行中的 bio 线程
if (server.aof_no_fsync_on_rewrite && hasActiveChildProcess())
return;
if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
latencyStartMonitor(latency);
//如果 AOF 落盘策略为 always,直接同步
if (redis_fsync(server.aof_fd) == -1) {
serverLog(LL_WARNING,"Can't persist AOF for fsync error when the "
"AOF fsync policy is 'always': %s. Exiting...", strerror(errno));
exit(1);
}
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("aof-fsync-always",latency);
server.aof_fsync_offset = server.aof_current_size;//更新 aof_fsync_offset 为当前的 AOF文件大小
server.aof_last_fsync = server.unixtime; // 上次 fsync 为当前的时间
} else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
server.unixtime > server.aof_last_fsync)) {
// 当前没有请求 fsync 的任务在线程池中
if (!sync_in_progress) {
//提交一个任务,就是向线程池的任务链表中添加任务节点, 最终就是一个后台线程执行一次 redis_fsync 函数
aof_background_fsync(server.aof_fd);
server.aof_fsync_offset = server.aof_current_size;
}
server.aof_last_fsync = server.unixtime;
}
}
2. AOF重写
很容易想到的一个情况,文件越写越大。AOF文件是以追加形式逐一记录接受到的写命令的。当一个键值对被多条命令反复修改时,AOF文件会记录相应的多条命令。要是宕机后重启,对同一个key,就需要依次执行AOF文件中对该key的操作命令。但是我们只需要最新的对该key的操作。
所以就有了重写。重写的时候,是根据这个键值对的最新状态,为它生成对应的写入命令。这样一来,一个键值对在重写日志中只使用一条命令即可。在日志恢复时,只执行一条命令,就可以直接完成这个键值对的写入,也方便省时。
AOF 重写的2个触发时机
- bgrewriteaof 命令被执行。
- 定时器函数, 定时检查 AOF 文件, 如果满足配置文件里面设置的条件, 就触发AOF重写
用户发送 bgrewriteaof 命令
bgrewriteaof 命令方式对应的逻辑函数为 bgrewriteaofCommand。主要逻辑是:
- 如果正在执行重写中了,返回错误提示
- 如果正在执行RDB保存,就将 aof_rewrite_scheduled 属性设置为 true, 返回提示后, 结束。之后是通过定时器函数serverCron判断这个状态确定是否需要触发
- 否则,调用 rewriteAppendOnlyFileBackground 执行重写
struct redisCommand redisCommandTable[] = {
{"bgrewriteaof",bgrewriteaofCommand,1,
"admin no-script",
0,NULL,0,0,0,0,0,0},
.................
};
void bgrewriteaofCommand(client *c) {
if (server.child_type == CHILD_TYPE_AOF) {
addReplyError(c,"Background append only file rewriting already in progress");
} else if (hasActiveChildProcess()) {
server.aof_rewrite_scheduled = 1;
addReplyStatus(c,"Background append only file rewriting scheduled");
} else if (rewriteAppendOnlyFileBackground() == C_OK) {
addReplyStatus(c,"Background append only file rewriting started");
} else {
addReplyError(c,"Can't execute an AOF background rewriting. "
"Please check the server logs for more information.");
}
}
在定时函数serverCron中触发
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
.......................
// 后台没有进程在 RDB 和 AOF, 同时通过 bgrewriteaof 命令设置了定时刷新重写 AOF
if (!hasActiveChildProcess() && server.aof_rewrite_scheduled) {
rewriteAppendOnlyFileBackground();
}
if (hasActiveChildProcess() || ldbPendingChildren()){ // 后台有进程在 RDB 或者 AOF
.......................
} else { // 当前后台 没有进程在 RDB 或者 AOF
//省略了一些检查.............
//达到了AOF重写的条件:开启了AOF && 后台没有RDB和AOF重写进行 &&
// 配置了目前 AOF 文件大小超过上次重写的 AOF 文件的百分比 &&
//当前的 AOF 文件大小超过了配置的需要触发重写的最小大小
if (server.aof_state == AOF_ON && !hasActiveChildProcess() &&
server.aof_rewrite_perc && server.aof_current_size > server.aof_rewrite_min_size){
// 计算当前的文件增长的比例
long long base = server.aof_rewrite_base_size ?
server.aof_rewrite_base_size : 1;
long long growth = (server.aof_current_size*100/base) - 100;
// 超过了就调用 rewriteAppendOnlyFileBackground 进行重写
if (growth >= server.aof_rewrite_perc) {
rewriteAppendOnlyFileBackground();
}
}
}
...........................................
}
所以,都是集中到函数rewriteAppendOnlyFileBackground中处理的。
在某时刻,需要AOF文件重写:
- 那为了不阻塞主线程,那可以fork一个子进程来重写。fork出来的子进程,拥有了和父进程一样的内存数据,子进程就先把这些内存数据写入到一个AOF临时文件。
- 但是在这个过程中,父进程还是能接受客户端的命令的,所以父子进程需要通讯,而Redis中父子进程是使用管道pipe进行通讯的。
父子进程使用pipe进行通信
Redis是使用了3个管道。每个管道有2端,所以有6个fd。
struct redisServer {
/* AOF pipes used to communicate between parent and child during rewrite. */
int aof_pipe_write_data_to_child;
int aof_pipe_read_data_from_parent;
int aof_pipe_write_ack_to_parent;
int aof_pipe_read_ack_from_child;
int aof_pipe_write_ack_to_child;
int aof_pipe_read_ack_from_parent;
...................
};
int aofCreatePipes(void) {
int fds[6] = {-1, -1, -1, -1, -1, -1};
int j;
//int pipe(int pipefd[2]); 成功:0;失败:-1,设置errno,函数调用成功返回r/w两个文件描述符
if (pipe(fds) == -1) goto error; /* parent -> children data. */
if (pipe(fds+2) == -1) goto error; /* children -> parent ack. */
if (pipe(fds+4) == -1) goto error; /* parent -> children ack. */
/* Parent -> children data is non blocking. */
if (anetNonBlock(NULL,fds[0]) != ANET_OK) goto error;
if (anetNonBlock(NULL,fds[1]) != ANET_OK) goto error;
if (aeCreateFileEvent(server.el, fds[2], AE_READABLE, aofChildPipeReadable, NULL) == AE_ERR) goto error;
server.aof_pipe_write_data_to_child = fds[1];
server.aof_pipe_read_data_from_parent = fds[0];
server.aof_pipe_write_ack_to_parent = fds[3];
server.aof_pipe_read_ack_from_child = fds[2];
server.aof_pipe_write_ack_to_child = fds[5];
server.aof_pipe_read_ack_from_parent = fds[4];
server.aof_stop_sending_diff = 0;
return C_OK;
................
}
- aof_pipe_write_data_to_child 和 aof_pipe_read_data_from_parent, 主要是父进程将子进程重写过程中产生的命令同步给子进程(即是同步数据)
- aof_pipe_write_ack_to_parent 和 aof_pipe_read_ack_from_child, 主要是用于子进程通知父进程停止同步变更命令
- aof_pipe_write_ack_to_child 和 aof_pipe_read_ack_from_parent, 主要用于父进程响应子进程的停止同步变更命令的请求
我们要重点关注aof_pipe_write_data_to_child(写端) 和 aof_pipe_read_data_from_parent(读端)。这两个是传输Redis内存数据的管道fd。
两个缓冲区——重写缓冲 和 差异缓冲
重写缓冲
struct redisServer {
list *aof_rewrite_buf_blocks; //重写缓冲区,注意是个链表 /* Hold changes during an AOF rewrite. */
...................
};
#define AOF_RW_BUF_BLOCK_SIZE (1024*1024*10) /* 10 MB per block */
//AOF 重写缓存列表的节点定义
typedef struct aofrwblock {
unsigned long used, free; //used:已使用的空间,free:剩余的空阿金
char buf[AOF_RW_BUF_BLOCK_SIZE];
} aofrwblock;
void aofRewriteBufferReset(void) {
if (server.aof_rewrite_buf_blocks)
listRelease(server.aof_rewrite_buf_blocks);
server.aof_rewrite_buf_blocks = listCreate();
listSetFreeMethod(server.aof_rewrite_buf_blocks,zfree);
}
什么时候使用到重写缓冲区?那就是需要进行AOF重写的时候。
将缓冲区中的数据写入到aof的函数是flushAppendOnlyFile。那也是在该函数中,会使用到重写缓冲区。
void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
.............................
if (server.child_type == CHILD_TYPE_AOF) /// 如果后台正在进行重写
aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));//将命令写入到 AOF 重写缓冲区
}
/* Append data to the AOF rewrite buffer, allocating new blocks if needed. */
void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
//获取缓冲区列表,是添加在尾部,所以获取尾部
listNode *ln = listLast(server.aof_rewrite_buf_blocks);
aofrwblock *block = ln ? ln->value : NULL;
while(len) {
/* If we already got at least an allocated block, try appending
* at least some piece into it. */
if (block) { //表明重写缓冲列表已有数据
//计算当前节点的剩余空间是否够len长度的数据写入
unsigned long thislen = (block->free < len) ? block->free : len;
if (thislen) { /* The current block is not already full. */
memcpy(block->buf+block->used, s, thislen);
block->used += thislen;
block->free -= thislen;
s += thislen;
len -= thislen;
}
}
// len > 0, 说明还需要空间, 但是当前的节点没有空间了, 需要新建一个节点
if (len) { /* First block to allocate, or need another block. */
int numblocks;
// 分配新的缓存节点, 同时放到列表的尾部
block = zmalloc(sizeof(*block));
block->free = AOF_RW_BUF_BLOCK_SIZE;
block->used = 0;
listAddNodeTail(server.aof_rewrite_buf_blocks,block);
numblocks = listLength(server.aof_rewrite_buf_blocks);
if (((numblocks+1) % 10) == 0) {
int level = ((numblocks+1) % 100) == 0 ? LL_WARNING :
LL_NOTICE;
serverLog(level,"Background AOF buffer size: %lu MB",
aofRewriteBufferSize()/(1024*1024));
}
}
}
// 注册一个文件事件, 用来将缓冲区的数据写入到 aof_pipe_write_data_to_child 中,
//然后在 Pipe 的作用下, 可以同步到 aof_pipe_read_data_from_parent
if (!server.aof_stop_sending_diff &&
aeGetFileEvents(server.el,server.aof_pipe_write_data_to_child) == 0)
{
//这里注意:注册的是 写事件, 写事件就绪的条件是内核空间的缓冲有空,就可以写
aeCreateFileEvent(server.el, server.aof_pipe_write_data_to_child,
AE_WRITABLE, aofChildWriteDiffData, NULL);
}
}
接着来看看管道fd的写事件回调函数aofChildWriteDiffData。
那么当内核缓冲区空间有空闲,就会触发该管道fd的写事件,就会执行aofChildWriteDiffData。通过该函数就把重写缓存中的数据写到了管道中,供子进程读取到子进程的差异缓冲中。
//事件回调函数, 把当前的 AOF 缓冲区同步到 aof_pipe_write_data_to_child, 在 Pipe 的作用下间接同步到 aof_pipe_read_data_from_parent
void aofChildWriteDiffData(aeEventLoop *el, int fd, void *privdata, int mask) {
listNode *ln;
aofrwblock *block;
ssize_t nwritten;
while(1) {
ln = listFirst(server.aof_rewrite_buf_blocks);
block = ln ? ln->value : NULL;
// 停止同步 或者 重写缓冲区为空, 就需要删除这个 写事件
if (server.aof_stop_sending_diff || !block) {
aeDeleteFileEvent(server.el,server.aof_pipe_write_data_to_child,
AE_WRITABLE);
return;
}
if (block->used > 0) {
// 把 block 的数据写入到 aof_pipe_write_data_to_child
nwritten = write(server.aof_pipe_write_data_to_child,
block->buf,block->used);
if (nwritten <= 0) return;
memmove(block->buf,block->buf+nwritten,block->used-nwritten);
block->used -= nwritten;
block->free += nwritten;
}
if (block->used == 0) listDelNode(server.aof_rewrite_buf_blocks,ln);
}
}
差异缓冲
在子进程重写AOF过程中,子进程等待主进程把重写缓冲中的数据通过pipe发送到差异缓冲区。
struct redisServer {
sds aof_child_diff; //子进程的差异缓冲区 /* AOF diff accumulator child side. */te. */
...................
};
子进程通过pipe将重写缓冲区中的数据同步到差异缓冲区的函数是aofReadDiffFromParent。
ssize_t aofReadDiffFromParent(void) {
char buf[65536]; /* Default pipe buffer size on most Linux systems. */
ssize_t nread, total = 0;
// 将 aof_pipe_read_data_from_parent 中的数据读取到 buf 中
while ((nread =read(server.aof_pipe_read_data_from_parent,buf,sizeof(buf))) > 0) {
// 把buf的数据拼接到aof_child_diff 中
server.aof_child_diff = sdscatlen(server.aof_child_diff,buf,nread);
total += nread;
}
return total;
}
rewriteAppendOnlyFileBackground的实现
了解了Redis中关于AOF重写的两个缓冲区和父子进程通过pipe通讯,那对AOF重写的过程就好理解了。
其具体的细节步骤:
- 主进程fork出一个子进程,让子进程来进行AOF重写。fork出来的子进程,拥有了和父进程一样的内存数据
- 子进程将内存中的数据写入到一个AOF临时文件中
- 在子进程重写期间,主进程还是会继续将新到达的命令追加写到原AOF,并将这些命令拷贝到重写缓冲,然后通过pipe管道发送给子进程的差异缓冲中。
- 子进程处理完内存数据后,就把差异缓冲中的数据追加到临时AOF文件中,之后就禁止主进程发新数据。
- 这时,若主进程中的重写缓存中还剩余数据,就把该数据追加到临时AOF文件中,再用临时AOF文件替换旧的AOF,结束。
int rewriteAppendOnlyFileBackground(void) {
pid_t childpid;
if (hasActiveChildProcess()) return C_ERR; //判断当前没有RDB和aof重写
if (aofCreatePipes() != C_OK) return C_ERR; //创建 Pipe 通道, 用于父子进程之间通信
//创建 AOF 子进程
if ((childpid = redisFork(CHILD_TYPE_AOF)) == 0) {
char tmpfile[256];
/* Child */
redisSetProcTitle("redis-aof-rewrite");
//将自己绑定给某个cpu
redisSetCpuAffinity(server.aof_rewrite_cpulist);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
//这个是重点, 重写AOF
if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
//子进程重写完成的一些收尾工作, 基本不涉及主流程, 通知父进程过程中子进程修改了多少数据
sendChildCowInfo(CHILD_INFO_TYPE_AOF_COW_SIZE, "AOF rewrite");
exitFromChild(0);
} else {
exitFromChild(1);
}
} else {
/* Parent */
..............
server.aof_rewrite_scheduled = 0;
server.aof_rewrite_time_start = time(NULL);
server.aof_selected_db = -1;
// 清空 redisServer 的 repl_scriptcache_dict 字典和 repl_scriptcache_fifo 这个列表
// 和主从复制相关
replicationScriptCacheFlush();
return C_OK;
}
return C_OK; /* unreached */
}
执行重写过程的函数——rewriteAppendOnlyFile
子进程执行的rewriteAppendOnlyFile就是真正的AOF重写过程。
这个流程步骤有点多:
- 打开aof临时文件,并命名;初始化差异缓冲server.aof_child_diff
- 若是启用了混合持久化,则调用rdbSaveRio将 RDB 数据写入 aof 临时文件;否则,调用
rewriteAppendOnlyFileRio()
进行普通的 aof 重写。其内部会遍历字典快照,删除无效数据后,将其封装为 RESP 数据写入临时文件。在遍历的过程中,还会周期性地从管道中拉取增量数据到aof_child_diff
。 - 将I/O缓冲和内核缓冲中的剩余数据同步到磁盘
- 从管道中读取剩余的增量数据,持续一段时间
- 停止读取后,发送指令给管道让主进程停止向管道写入。然后等待主进程地 ACK;
- 此时父进程不会在同步差异命令过来了, 再做最后一次同步, 将 Pipe 通道中残留的数据同步过来,再次从管道中读取数据。
- 将差异缓冲中的数据追加到AOF临时文件中,并再次将AOF临时文件缓冲中的数据同步到磁盘中。
- 修改临时文件名,并确认写入成功
int rewriteAppendOnlyFile(char *filename) {
rio aof;
char tmpfile[256];
char byte;
// 1
snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
FILE *fp = fopen(tmpfile,"w");
//..................
// 清空 aof_child_diff 的数据, 这个就是 AOF 子进程差异缓冲区
server.aof_child_diff = sdsempty();
rioInitWithFile(&aof,fp); // 初始 rio 流, 也就是 IO 流, 用于写入数据到文件
// 设定 fsync 触发条件
if (server.aof_rewrite_incremental_fsync)
rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES);
startSaving(RDBFLAGS_AOF_PREAMBLE);
// 2
if (server.aof_use_rdb_preamble) {
int error;
//混合持久化
if (rdbSaveRio(&aof,&error,RDBFLAGS_AOF_PREAMBLE,NULL) == C_ERR) {
errno = error;
goto werr;
}
} else {
//普通持久化
if (rewriteAppendOnlyFileRio(&aof) == C_ERR) goto werr;
}
// 3
//fflush:是把C库中的缓冲调用write函数写到磁盘[其实是写到内核的缓冲区]。
//fsync:是把内核缓冲刷到磁盘上。
if (fflush(fp) == EOF) goto werr;
if (fsync(fileno(fp)) == -1) goto werr;
int nodata = 0;
mstime_t start = mstime();
// 4 .从管道中拉取剩余的增量数据,持续一段时间
while(mstime()-start < 1000 && nodata < 20) {
if (aeWait(server.aof_pipe_read_data_from_parent, AE_READABLE, 1) <= 0)
{
nodata++;
continue;
}
nodata = 0; /* Start counting from zero, we stop on N *contiguous*
timeouts. */
aofReadDiffFromParent(); //从管道读数据到 差异缓冲aof_child_diff
}
// 5 通知主进程 停止发送增量数据
if (write(server.aof_pipe_write_ack_to_parent,"!",1) != 1) goto werr;
if (anetNonBlock(NULL,server.aof_pipe_read_ack_from_parent) != ANET_OK)
goto werr;
// 等待主进程的 ACK,最多等 5s
if (syncRead(server.aof_pipe_read_ack_from_parent,&byte,1,5000) != 1 ||
byte != '!') goto werr;
// 此时父进程不会在同步差异命令过来了, 再做最后一次同步, 将 Pipe 通道中残留的数据同步过来
// 再次从管道中读取差异数据
aofReadDiffFromParent();
//获取差异缓冲数据的内容大小
size_t bytes_to_write = sdslen(server.aof_child_diff);
const char *buf = server.aof_child_diff;
long long cow_updated_time = mstime();
long long key_count = dbTotalServerKeyCount();
// 6 . 将差异缓冲数据写入 aof 文件
while (bytes_to_write) {
size_t chunk_size = bytes_to_write < (8<<20) ? bytes_to_write : (8<<20);
// 将 aof_child_diff 中的数据写入到 aof 文件中
if (rioWrite(&aof,buf,chunk_size) == 0)
goto werr;
bytes_to_write -= chunk_size;
buf += chunk_size;
/* Update COW info */
long long now = mstime();
if (now - cow_updated_time >= 1000) {
sendChildInfo(CHILD_INFO_TYPE_CURRENT_INFO, key_count, "AOF rewrite");
cow_updated_time = now;
}
}
// 7 .将 aof 文件缓冲中的数据,同步到磁盘
if (fflush(fp)) goto werr;
if (fsync(fileno(fp))) goto werr;
if (fclose(fp)) { fp = NULL; goto werr; }
fp = NULL;
//8 .重命名文件
if (rename(tmpfile,filename) == -1) {
unlink(tmpfile);
stopSaving(0);
return C_ERR;
}
stopSaving(1);
return C_OK;
werr:
if (fp) fclose(fp);
unlink(tmpfile);
stopSaving(0);
return C_ERR;
}
父进程监听子进程结束, AOF 重写收尾
在定时函数serverCron中监听
那是在哪进行监听呢?还是在定时函数serverCron中。定时地检查子进程的状态是否为结束了, 是的话, 执行结束逻辑。在下一次运行 serverCron
定时函数时,调用 checkChildrenDone()
完成 AOF 收尾工作。checkChildrenDone的核心内容
是backgroundRewriteDoneHandler函数。
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
.................
// 检查是否有 RDB 子进程或者 AOF 重写子进程结束了
if (hasActiveChildProcess() || ldbPendingChildren()){
run_with_period(1000) receiveChildInfo();
checkChildrenDone();
} else {
...........
}
............................
}
/* Receive info data from child. */
void receiveChildInfo(void) {
if (server.child_info_pipe[0] == -1) return;
size_t cow;
monotime cow_updated;
size_t keys;
double progress;
childInfoType information_type;
/* Drain the pipe and update child info so that we get the final message. */
while (readChildInfo(&information_type, &cow, &cow_updated, &keys, &progress)) {
updateChildInfo(information_type, cow, cow_updated, keys, progress);
}
}
void checkChildrenDone(void) {
int statloc = 0;
pid_t pid;
// wait3可以获取所有的进程是否有一个进程退出状态的, 有的话, 进行彻底的销毁,并返回其进程id
if ((pid = waitpid(-1, &statloc, WNOHANG)) != 0) {
int exitcode = WIFEXITED(statloc) ? WEXITSTATUS(statloc) : -1;
int bysignal = 0;
if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
if (exitcode == SERVER_CHILD_NOERROR_RETVAL) {
bysignal = SIGUSR1;
exitcode = 1;
}
if (pid == -1) {
//打印日志
} else if (pid == server.child_pid) {
if (server.child_type == CHILD_TYPE_RDB) {
backgroundSaveDoneHandler(exitcode, bysignal);
} else if (server.child_type == CHILD_TYPE_AOF) {
backgroundRewriteDoneHandler(exitcode, bysignal); //自己想哦买噶最终的清理逻辑
}
if (!bysignal && exitcode == 0) receiveChildInfo(); //获取子进程发送给父进程的信息
resetChildState();
} else {
if (!ldbRemoveChild(pid)) {
//打印日志
}
}
/* start any pending forks immediately. */
replicationStartPendingFork();
}
}
主进程对AOF重写收尾——backgroundRewriteDoneHandler
主进程的backgroundRewriteDoneHandler中主要是4步骤:
- 打开子进程刚刚处理完的 aof 临时文件
- 将停止发送增量数据期间积累的数据追加到 临时AOF文件
- 重命名,替换旧的aof文件
- 最后,进行清除工作
void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
if (!bysignal && exitcode == 0) {
int newfd, oldfd;
char tmpfile[256];
long long now = ustime();
mstime_t latency;
latencyStartMonitor(latency);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int)server.child_pid);
// 1 打开子进程刚刚处理完的 aof 临时文件
newfd = open(tmpfile,O_WRONLY|O_APPEND);
if (newfd == -1) { goto cleanup; }
// 2 将停止发送增量数据期间积累的数据追加到 临时AOF文件
if (aofRewriteBufferWrite(newfd) == -1) {
close(newfd); goto cleanup;
}
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("aof-rewrite-diff-write",latency);
if (server.aof_fsync == AOF_FSYNC_EVERYSEC) {
aof_background_fsync(newfd);
} else if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
latencyStartMonitor(latency);
if (redis_fsync(newfd) == -1) {
close(newfd);
goto cleanup;
}
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("aof-rewrite-done-fsync",latency);
}
// aof_fd 为当前的 AOF 文件的文件描述符, 等于 -1, 应该是 AOF 功能停用了
// 这时为了下面的流程能走下去, 从配置文件中获取到配置的文件名, 尝试打开禁用前的文件
if (server.aof_fd == -1) {
/* AOF disabled */
oldfd = open(server.aof_filename,O_RDONLY|O_NONBLOCK);
} else {
/* AOF enabled */
oldfd = -1; /* We'll set this to the current AOF filedes later. */
}
latencyStartMonitor(latency);
// 3 重命名,替换旧的aof文件
if (rename(tmpfile,server.aof_filename) == -1) {
close(newfd);
if (oldfd != -1) close(oldfd);
goto cleanup;
}
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("aof-rename",latency);
if (server.aof_fd == -1) {
/* AOF disabled, we don't need to set the AOF file descriptor
* to this new file, so we can close it. */
close(newfd);
} else {
/* AOF enabled, replace the old fd with the new one. */
oldfd = server.aof_fd;
server.aof_fd = newfd;
server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size;
server.aof_fsync_offset = server.aof_current_size;
server.aof_last_fsync = server.unixtime;
/* Clear regular AOF buffer since its contents was just written to
* the new AOF from the background rewrite buffer. */
sdsfree(server.aof_buf);
server.aof_buf = sdsempty();
}
server.aof_lastbgrewrite_status = C_OK;
/* Change state from WAIT_REWRITE to ON if needed */
if (server.aof_state == AOF_WAIT_REWRITE)
server.aof_state = AOF_ON;
/* Asynchronously close the overwritten AOF. */
if (oldfd != -1) bioCreateCloseJob(oldfd);
} else if (!bysignal && exitcode != 0) {
server.aof_lastbgrewrite_status = C_ERR;
} else {
if (bysignal != SIGUSR1)
server.aof_lastbgrewrite_status = C_ERR;
}
cleanup:
//清除工作
aofClosePipes();
aofRewriteBufferReset();
aofRemoveTempFile(server.child_pid);
server.aof_rewrite_time_last = time(NULL)-server.aof_rewrite_time_start;
server.aof_rewrite_time_start = -1;
/* Schedule a new rewrite if we are waiting for it to switch the AOF ON. */
if (server.aof_state == AOF_WAIT_REWRITE)
server.aof_rewrite_scheduled = 1;
}
重写失败的话,原来的AOF文件依然是可以使用的。在AOF重写过程中,新来的命令会被写入磁盘两次(主进程写入到旧AOF,子进程是追加到临时AOF),这就会浪费一定的磁盘空间(磁盘便宜大碗,没问题的)。只是在重写过程中,新的命令会被全部储存到子进程的差异缓冲区中,这可能会导致较高的内存占用。
3. Redis重启,AOF 文件加载
从main函数开始,其调用流程 main——>loadDataFromDisk——>loadAppendOnlyFile。
其主要流程:
- 打开AOF文件
- 创建一个虚拟客户端,用于执行AOF中的命令
- 根据aof文件中的前导码,判断若是REDIS开头,就调用rdbLoadRio加载RDB的数据;否则将文件指针归零;
- 开始循环处理RESP格式的字符串
- 按照RESP协议读取命令的参数的个数
- 读取命令的每个参数
- 根据第一个参数,查询命令表,得到命令
- 执行命令
int main(int argc, char **argv) {
.........
if (!server.sentinel_mode) { //非哨兵模式
loadDataFromDisk();
............
}
}
void loadDataFromDisk(void) {
if (server.aof_state == AOF_ON) {
loadAppendOnlyFile(server.aof_filename)
}
...................
}
int loadAppendOnlyFile(char *filename) {
struct client *fakeClient;
// 1 打开aof文件
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 latest well-formed command loaded. */
off_t valid_before_multi = 0; /* Offset before MULTI command loaded. */
if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
server.aof_current_size = 0;
server.aof_fsync_offset = server.aof_current_size;
fclose(fp);
return C_ERR;
}
/* Temporarily disable AOF, to prevent EXEC from feeding a MULTI
* to the same file we're about to read. */
server.aof_state = AOF_OFF;
// 2 创建虚拟客户端
fakeClient = createAOFClient();
startLoadingFile(fp, filename, RDBFLAGS_AOF_PREAMBLE);
// 3 根据是否有RDB前导码,再确定处理方式
char sig[5]; /* "REDIS" */
if (fread(sig,1,5,fp) != 5 || memcmp(sig,"REDIS",5) != 0) {
/* No RDB preamble, seek back at 0 offset. */
if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
} else {
/* RDB preamble. Pass loading the RDB functions. */
rio rdb;
if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
rioInitWithFile(&rdb,fp);
//加载rdb内容
if (rdbLoadRio(&rdb,RDBFLAGS_AOF_PREAMBLE,NULL) != C_OK) {
goto readerr;
}
}
// 4 循环处理Aof文件中剩下的所有命令
while(1) {
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));
processEventsWhileBlocked();
processModuleLoadingProgressEvent(1);
}
if (fgets(buf,sizeof(buf),fp) == NULL) {
if (feof(fp))
break;
else
goto readerr;
}
if (buf[0] != '*') goto fmterr;
if (buf[1] == '\0') goto readerr;
// 4.1 按照resp协议读取命令的参数数量
argc = atoi(buf+1);
if (argc < 1) goto fmterr;
argv = zmalloc(sizeof(robj*)*argc);
fakeClient->argc = argc;
fakeClient->argv = argv;
// 4.2 循环读取命令的每个参数
for (j = 0; j < argc; j++) {
/* Parse the argument len. */
char *readres = fgets(buf,sizeof(buf),fp);
if (readres == NULL || buf[0] != '$') {
fakeClient->argc = j; /* Free up to j-1. */
freeFakeClientArgv(fakeClient);
if (readres == NULL)
goto readerr;
else
goto fmterr;
}
len = strtol(buf+1,NULL,10);
/* Read it into a string object. */
argsds = sdsnewlen(SDS_NOINIT,len);
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(OBJ_STRING,argsds);
/* Discard CRLF. */
if (fread(buf,2,1,fp) == 0) {
fakeClient->argc = j+1; /* Free up to j. */
freeFakeClientArgv(fakeClient);
goto readerr;
}
}
// 4.3 根据第一个参数,查询命令表,获取命令
cmd = lookupCommand(argv[0]->ptr);
if (cmd == server.multiCommand) valid_before_multi = valid_up_to;
// 4.4 执行命令
fakeClient->cmd = fakeClient->lastcmd = cmd;
if (fakeClient->flags & CLIENT_MULTI &&
fakeClient->cmd->proc != execCommand)
{
queueMultiCommand(fakeClient);
} else {
cmd->proc(fakeClient);
}
/* 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);
fakeClient->cmd = NULL;
if (server.aof_load_truncated) valid_up_to = ftello(fp);
if (server.key_load_delay)
debugDelay(server.key_load_delay);
}
if (fakeClient->flags & CLIENT_MULTI) {
valid_up_to = valid_before_multi;
goto uxeof;
}
..........................
}
goto部分的代码:
int loadAppendOnlyFile(char *filename) {
...................................
loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */
fclose(fp);
freeFakeClient(fakeClient);
server.aof_state = old_aof_state;
stopLoading(1);
aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size;
server.aof_fsync_offset = server.aof_current_size;
return C_OK;
readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */
if (!feof(fp)) {
if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */
fclose(fp);
serverLog(LL_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno));
exit(1);
}
uxeof: /* Unexpected AOF end of file. */
if (server.aof_load_truncated) {
serverLog(LL_WARNING,"!!! Warning: short read while loading the AOF file !!!");
serverLog(LL_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) {
serverLog(LL_WARNING,"Last valid command offset is invalid");
} else {
serverLog(LL_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) {
serverLog(LL_WARNING,"Can't seek the end of the AOF file: %s",
strerror(errno));
} else {
serverLog(LL_WARNING,
"AOF loaded anyway because aof-load-truncated is enabled");
goto loaded_ok;
}
}
}
if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */
fclose(fp);
exit(1);
fmterr: /* Format error. */
if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */
fclose(fp);
exit(1);
}