Redis源码学习-AOF

struct redisServer{

    ...

    /* AOF persistence */

    int aof_state;                  /* REDIS_AOF_(ON|OFF|WAIT_REWRITE) */

    int aof_fsync;                  /* Kind of fsync() policy （每个操作|每秒|缓冲区满）*/

    char *aof_filename;             /* Name of the AOF file */

    int aof_no_fsync_on_rewrite;    /* Don't fsync if a rewrite is in prog. */

    int aof_rewrite_perc;           /* Rewrite AOF if % growth is > M and... */

    off_t aof_rewrite_min_size;     /* the AOF file is at least N bytes. */

    off_t aof_rewrite_base_size;    /* AOF size on latest startup or rewrite. */

    off_t aof_current_size;         /* AOF current size. */

    int aof_rewrite_scheduled;      /* Rewrite once BGSAVE terminates. 是否需要开启后台aof子进程*/

    pid_t aof_child_pid;            /* PID if rewriting process */

    list *aof_rewrite_buf_blocks;   /* Hold changes during an AOF rewrite. 在aof bgsave期间redis执行的命令将存储到aof_rewrite_buf_blocks，当然aof_buf还是要照常使用的，二者不冲突*/

    sds aof_buf;      /* AOF buffer, written before entering the event loop */

    int aof_fd;       /* File descriptor of currently selected AOF file */

    int aof_selected_db; /* Currently selected DB in AOF */

    time_t aof_flush_postponed_start; /* UNIX time of postponed AOF flush */

    time_t aof_last_fsync;            /* UNIX time of last fsync() */

    time_t aof_rewrite_time_last;   /* Time used by last AOF rewrite run. */

    time_t aof_rewrite_time_start;  /* Current AOF rewrite start time. */

    int aof_lastbgrewrite_status;   /* REDIS_OK or REDIS_ERR */

    unsigned long aof_delayed_fsync;  /* delayed AOF fsync() counter */

    int aof_rewrite_incremental_fsync;/* fsync incrementally while rewriting? */

    ...

}

/////////////////////////////////////////////////////////////////////////////////

/* Call() is the core of Redis execution of a command */

void call(redisClient *c, int flags) {

    long long dirty, start = ustime(), duration;

    int client_old_flags = c->flags;

    ...

     /* 执行用户命令 */

    c->flags &= ~(REDIS_FORCE_AOF|REDIS_FORCE_REPL);

    redisOpArrayInit(&server.also_propagate);

    dirty = server.dirty;

    c->cmd->proc(c);

    dirty = server.dirty-dirty;

    duration = ustime()-start;

    ...

    /* 将用户命令进行AOF备份 */

    if (flags & REDIS_CALL_PROPAGATE) {

        int flags = REDIS_PROPAGATE_NONE;

        if (c->flags & REDIS_FORCE_REPL) flags |= REDIS_PROPAGATE_REPL;

        if (c->flags & REDIS_FORCE_AOF) flags |= REDIS_PROPAGATE_AOF;

        if (dirty)

            flags |= (REDIS_PROPAGATE_REPL | REDIS_PROPAGATE_AOF);

        if (flags != REDIS_PROPAGATE_NONE)

            propagate(c->cmd,c->db->id,c->argv,c->argc,flags);

    }

}

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

}

void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {

    sds buf = sdsempty();

    robj *tmpargv[3];

    /* 如果当前操作的dict和前一次操作的dict不同，

    那么redis要在aof中添加一条：select命令，选择当前dict */

    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;

    }

    //依据不同的命令，进行字符画处理，并将结果写入临时的buff中

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

    }

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

    //如果用户开启的AOF，那么将当前命令的buff Append到server.aof_buf缓冲的尾部

    if (server.aof_state == REDIS_AOF_ON)

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

     //如果当前有子进程正在进行AOF日志的重构（即扫描redis数据库，依据数据构建日志）

     //那么将当前命令的buff添加到server.aof_rewrite_buf_blocks内存中（该部分内存

     //专门记录在重构AOF期间redis处理的操作）

    if (server.aof_child_pid != -1)

        aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));

    sdsfree(buf);

}

////////////////////////////////////////////////////////////////////////////////////////

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {

    int j;

    REDIS_NOTUSED(eventLoop);

    REDIS_NOTUSED(id);

    REDIS_NOTUSED(clientData);

    /* Software watchdog: deliver the SIGALRM that will reach the signal

     * handler if we don't return here fast enough. */

    if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period);

    /* We take a cached value of the unix time in the global state because

     * with virtual memory and aging there is to store the current time

     * in objects at every object access, and accuracy is not needed.

     * To access a global var is faster than calling time(NULL) */

     //缓存系统时间...

    server.unixtime = time(NULL);

    server.mstime = mstime();

    ...

    /* Start a scheduled AOF rewrite if this was requested by the user while

     * a BGSAVE was in progress. */

     //开启AOF日志重建的子进程（简化日志）

     //后台AOF子进程通过扫描redis.db[16]数据，生成可重建当前数据库的命令，

     //并写入临时文件tmpfile

    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&

        server.aof_rewrite_scheduled)

    {

        //AOF

        rewriteAppendOnlyFileBackground();

    }

    /* Check if a background saving or AOF rewrite in progress terminated. */

    //后台AOF进程结束：将在后台AOF子进程构建AOF日志期间redis执行的新命令

    //（记录于server.aof_rewrite_buf_blocks）append 到后台子进程构建的tmpfile中

    //最后将tmpfile重名为server.aof_filename 替换原有AOF

    if (server.rdb_child_pid != -1 || server.aof_child_pid != -1) {

        int statloc;

        pid_t pid;

        if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {

            int exitcode = WEXITSTATUS(statloc);

            int bysignal = 0;

            if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);

            if (pid == server.rdb_child_pid) {

                backgroundSaveDoneHandler(exitcode,bysignal);

            } else if (pid == server.aof_child_pid) {

                backgroundRewriteDoneHandler(exitcode,bysignal);

            } else {

                redisLog(REDIS_WARNING,

                    "Warning, detected child with unmatched pid: %ld",

                    (long)pid);

            }

            updateDictResizePolicy();

        }

    } else {

        /* If there is not a background saving/rewrite in progress check if

         * we have to save/rewrite now */

         //没有后台子进程在跑，那么检查是否要开启一个AOF或者RDB的子进程。。。

         ...

    }

    /* If we postponed an AOF buffer flush, let's try to do it every time the

     * cron function is called. */

     //将server.aof_buf（缓存redis最近执行过的命名）flush到磁盘AOF文件中

     //flush的策略有如下：

     //每个操作，调用fync将命令持久化

     //间隔1秒，调用fync将aof_buf持久化

     //从不调用fync，由系统自行安排时机

    if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);

    ...

    server.cronloops++;

    return 1000/server.hz;

}

/* This is how rewriting of the append only file in background works:

 *

 * 1) The user calls BGREWRITEAOF

 * 2) Redis calls this function, that forks():

 *    2a) the child rewrite the append only file in a temp file.

 *    2b) the parent accumulates differences in server.aof_rewrite_buf.

 * 3) When the child finished '2a' exists.

 * 4) The parent will trap the exit code, if it's OK, will append the

 *    data accumulated into server.aof_rewrite_buf into the temp file, and

 *    finally will rename(2) the temp file in the actual file name.

 *    The the new file is reopened as the new append only file. Profit!

 */

int rewriteAppendOnlyFileBackground(void) {

    pid_t childpid;

    long long start;

    if (server.aof_child_pid != -1) return REDIS_ERR;

    start = ustime();

    if ((childpid = fork()) == 0) {

        char tmpfile[256];

        /* Child */

        closeListeningSockets(0);

        redisSetProcTitle("redis-aof-rewrite");

        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());

        if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {

            size_t private_dirty = zmalloc_get_private_dirty();

            if (private_dirty) {

                redisLog(REDIS_NOTICE,

                    "AOF rewrite: %zu MB of memory used by copy-on-write",

                    private_dirty/(1024*1024));

            }

            exitFromChild(0);

        } else {

            exitFromChild(1);

        }

    } else {

        /* Parent */

        server.stat_fork_time = ustime()-start;

        if (childpid == -1) {

            redisLog(REDIS_WARNING,

                "Can't rewrite append only file in background: fork: %s",

                strerror(errno));

            return REDIS_ERR;

        }

        redisLog(REDIS_NOTICE,

            "Background append only file rewriting started by pid %d",childpid);

        server.aof_rewrite_scheduled = 0;

        server.aof_rewrite_time_start = time(NULL);

        server.aof_child_pid = childpid;

        updateDictResizePolicy();

        /* We set appendseldb to -1 in order to force the next call to the

         * feedAppendOnlyFile() to issue a SELECT command, so the differences

         * accumulated by the parent into server.aof_rewrite_buf will start

         * with a SELECT statement and it will be safe to merge. */

        server.aof_selected_db = -1;

        replicationScriptCacheFlush();

        return REDIS_OK;

    }

    return REDIS_OK; /* unreached */

}

/* Write a sequence of commands able to fully rebuild the dataset into

 * "filename". Used both by REWRITEAOF and BGREWRITEAOF.

 *

 * In order to minimize the number of commands needed in the rewritten

 * log Redis uses variadic commands when possible, such as RPUSH, SADD

 * and ZADD. However at max REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time

 * are inserted using a single command. */

int rewriteAppendOnlyFile(char *filename) {

    dictIterator *di = NULL;

    dictEntry *de;

    rio aof;

    FILE *fp;

    char tmpfile[256];

    int j;

    long long now = mstime();

    /* Note that we have to use a different temp name here compared to the

     * one used by rewriteAppendOnlyFileBackground() function. */

    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());

    fp = fopen(tmpfile,"w");

    if (!fp) {

        redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));

        return REDIS_ERR;

    }

    rioInitWithFile(&aof,fp);

    if (server.aof_rewrite_incremental_fsync)

        rioSetAutoSync(&aof,REDIS_AOF_AUTOSYNC_BYTES);

    for (j = 0; j < server.dbnum; j++) {

        //添加一条定位dict的命令

        char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";

        redisDb *db = server.db+j;

        dict *d = db->dict;

        if (dictSize(d) == 0) continue;

        di = dictGetSafeIterator(d);

        if (!di) {

            fclose(fp);

            return REDIS_ERR;

        }

        /* SELECT the new DB */

        if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;

        if (rioWriteBulkLongLong(&aof,j) == 0) goto werr;

        /* Iterate this DB writing every entry */

        while((de = dictNext(di)) != NULL) {

            sds keystr;

            robj key, *o;

            long long expiretime;

            keystr = dictGetKey(de);

            o = dictGetVal(de);

            initStaticStringObject(key,keystr);

            expiretime = getExpire(db,&key);

            /* If this key is already expired skip it */

            if (expiretime != -1 && expiretime < now) continue;

            /* Save the key and associated value */

            if (o->type == REDIS_STRING) {

                /* Emit a SET command */

                char cmd[]="*3\r\n$3\r\nSET\r\n";

                if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;

                /* Key and value */

                if (rioWriteBulkObject(&aof,&key) == 0) goto werr;

                if (rioWriteBulkObject(&aof,o) == 0) goto werr;

            } else if (o->type == REDIS_LIST) {

                if (rewriteListObject(&aof,&key,o) == 0) goto werr;

            } else if (o->type == REDIS_SET) {

                if (rewriteSetObject(&aof,&key,o) == 0) goto werr;

            } else if (o->type == REDIS_ZSET) {

                if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;

            } else if (o->type == REDIS_HASH) {

                if (rewriteHashObject(&aof,&key,o) == 0) goto werr;

            } else {

                redisPanic("Unknown object type");

            }

            /* Save the expire time */

            if (expiretime != -1) {

                char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";

                if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;

                if (rioWriteBulkObject(&aof,&key) == 0) goto werr;

                if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;

            }

        }

        dictReleaseIterator(di);

    }

    /* Make sure data will not remain on the OS's output buffers */

    fflush(fp);

    aof_fsync(fileno(fp));

    fclose(fp);

    /* Use RENAME to make sure the DB file is changed atomically only

     * if the generate DB file is ok. */

    if (rename(tmpfile,filename) == -1) {

        redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));

        unlink(tmpfile);

        return REDIS_ERR;

    }

    redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");

    return REDIS_OK;

werr:

    fclose(fp);

    unlink(tmpfile);

    redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));

    if (di) dictReleaseIterator(di);

    return REDIS_ERR;

}

/* Write the append only file buffer on disk.

 *

 * Since we are required to write the AOF before replying to the client,

 * and the only way the client socket can get a write is entering when the

 * the event loop, we accumulate all the AOF writes in a memory

 * buffer and write it on disk using this function just before entering

 * the event loop again.

 *

 * About the 'force' argument:

 *

 * When the fsync policy is set to 'everysec' we may delay the flush if there

 * is still an fsync() going on in the background thread, since for instance

 * on Linux write(2) will be blocked by the background fsync anyway.

 * When this happens we remember that there is some aof buffer to be

 * flushed ASAP, and will try to do that in the serverCron() function.

 *

 * However if force is set to 1 we'll write regardless of the background

 * fsync. */

void flushAppendOnlyFile(int force) {

    ssize_t nwritten;

    int sync_in_progress = 0;

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

    if (server.aof_fsync == AOF_FSYNC_EVERYSEC)

        sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0;

    //判定是否该开始将server.aof_buff中缓存的命令flush到server.aof_fd文件的写缓冲中

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

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

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

        }

    }

    /* If you are following this code path, then we are going to write so

     * set reset the postponed flush sentinel to zero. */

    server.aof_flush_postponed_start = 0;  

    /* We want to perform a single write. This should be guaranteed atomic

     * at least if the filesystem we are writing is a real physical one.

     * While this will save us against the server being killed I don't think

     * there is much to do about the whole server stopping for power problems

     * or alike */

     //将redis最近执行的一些命令（存于server.aof_buf）写入文件（server.aof_fd）

     //注意，写入文件并不能保证马上写入磁盘，因为这是带缓冲的写。关于何时将

     //文件写缓冲中的命令fync到磁盘，这就要看用户的设置：（见下文）

    nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));

    if (nwritten != (signed)sdslen(server.aof_buf)) {

        /* Ooops, we are in troubles. The best thing to do for now is

         * aborting instead of giving the illusion that everything is

         * working as expected. */

        ...

        exit(1);

    }

    server.aof_current_size += nwritten;   

    /* 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_no_fsync_on_rewrite ： 该标志位表示当有aof或rdb子进程时，不进行fsync操作

    if (server.aof_no_fsync_on_rewrite &&

        (server.aof_child_pid != -1 || server.rdb_child_pid != -1))

            return;

    //fsync...

    //每个操作，调用fync将命令持久化 [1]

    //间隔1秒，调用fync将aof_buf持久化 [2]

    //从不调用fync，由系统自行安排时机（fd的写缓冲区满了）[3]

    //【1】

    //每个操作都需要将文件缓冲区的写 buff sync到磁盘。从而保证每个redis操作在

    //被redis执行后，都能马上持久化，安全性很高，就是磁盘写的系统开销有点大大

    if (server.aof_fsync == AOF_FSYNC_ALWAYS) {

        /* aof_fsync is defined as fdatasync() for Linux in order to avoid

         * flushing metadata. */

        aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */

        server.aof_last_fsync = server.unixtime;

    }

    //【2】

    //每隔1s将文件缓冲区的写缓冲区sync到磁盘

    else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&

                server.unixtime > server.aof_last_fsync)) {

        if (!sync_in_progress) aof_background_fsync(server.aof_fd);

        server.aof_last_fsync = server.unixtime;

    }

    //【3】

    //else fd的写缓冲满后会由系统安排执行（听天由命）

}