redisソースの浅い分析--10-AOF(append only file)の持続化

環境説明:redisソースバージョン5.0.3;ソースコードを読む過程でコメントをしましたgitアドレス:https://gitee.com/xiaoangg/redis_annotation参考書:『redisの設計と実現』
RDB持続化機能に加えて、redisはAOF持続化機能を提供した.
RDB持計画機能は、キー値を符号化してRDBファイルキーに保存し、AOFは実行されたコマンドをAOFファイルに保存する.

一.AOF機能の実現

aof機能は、コマンド書き込み(append)、ファイル同期(sync)、ファイル書き換え(rewrite)、ロードの再起動(load)に分けることができる.
1.命令の追加
aofが開いている場合、サーバは書き込みコマンドを実行すると、aofキャッシュにプロトコル形式で書き込まれます.
aofファイル追撃の入り口はserviceにある.c/call()呼び出しのpropagate()
aofファイルフォーマットプロトコルはaofにある.c/catAppendOnlyGenericCommand:

/*
  aof    
argc      
argv robj      
(
set abc 1
  argc =3, argv =["set", "abc", "1"] 
)
*/
sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) {
    char buf[32];
    int len, j;
    robj *o;

    buf[0] = '*'; //         *  ；     buf="*"
    len = 1 + ll2string(buf + 1, sizeof(buf) - 1, argc); //   buf= "*"+       ；（ argc=3    buf=“*3”）
    buf[len++] = '\r';                                   //buf=“*3\r“
    buf[len++] = '
';                                   //buf=“*3\r
“
    dst = sdscatlen(dst, buf, len);                      // buf        dst

    //                   set abc 123 
    for (j = 0; j < argc; j++) {
        o = getDecodedObject(argv[j]);
        buf[0] = '$';   //buf ="$"
        len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr)); //buf="$3" ;      set      
        buf[len++] = '\r';                                    
        buf[len++] = '
';                                     //buf=“buf="$3\r
"
        dst = sdscatlen(dst, buf, len);                        //buf    dst 
        dst = sdscatlen(dst,o->ptr,sdslen(o->ptr));
        dst = sdscatlen(dst,"\r
",2);
        decrRefCount(o);
    }
    return dst;
}

2.AOFファイルの書き込みと同期
redisのサーバプロセスは、ファイルイベントがクライアントのコマンド要求を受信し、クライアントにコマンド返信を送信するイベントサイクルである.
時間イベントは、サービスがタイミング実行を必要とする関数を実行する.
サーバはイベントループを終了するたびにaofを呼び出す.c/flushAppendOnlyFile関数は、AOFキャッシュ領域の内容をAOFファイルに保存するかどうかを考慮します.
flushAppendOnlyFileの動作は、サーバ構成のappendfsyncによって決定されます.
#define AOF_FSYNC_NO 0 #define AOF_FSYNC_ALWAYS 1 #define AOF_FSYNC_EVERYSEC 2

AOF_FSYNC_ALWAYS=1 aofバッファのすべてのコンテンツをAOFファイル

に書き込み同期する

 AOF_FSYNC_EVERYSEC=2 aofバッファのすべての内容をAOFファイルに書き込み、最後にaofファイルを同期した時間が1秒を超えた場合、aofファイルを同期操作する.この操作はスレッドによって専門的に実行されます.

AOF_FSYNC_NO=0 aofバッファ内のすべてのコンテンツをAOFファイルに書き込みますが、aofファイルは同期されません.同期はオペレーティングシステムによって解決されます.
ファイルの書き込みと同期:ファイルの書き込み効率を高めるために、現代のオペレーティングシステムでは、ユーザーがwrite関数を呼び出してファイルを書くと、オペレーティングシステムは通常、メモリのバッファにデータを保存し、バッファが満たされたり、指定された時間を超えたりして、バッファのデータをディスクに書き込む.
この方法では、ファイルの書き込み効率が向上しますが、コンピュータがダウンタイムになると、バッファを保存するデータが失われ、セキュリティの問題も発生します.
このシステムにはfsyncとfdatasyncの2つの関数があり、バッファ内のデータをディスクに強制的に書き込むことができます.

二.AOFファイルの読み込みとデータ復元

aofファイルにはデータベースを再構築するすべての書き込みコマンドが含まれているので、aofファイルの内容を読み込んで、片側を再実行するだけで、データベースの状態を復元することができます.
redisはaofファイルを読み出して復元する具体的なステップは以下の通りである.

ネットワーク接続を持たない擬似クライアントを作成する:redisコマンドはクライアントコンテキストでしか実行できないため、aofファイルのコマンドを指定するために擬似クライアント(aof.c/createFakeClient)

が作成された.

aofでコマンドを読み取り、解析する.

擬似クライアントを使用して書き込みコマンドを実行する.

aofファイルの処理が完了するまで2、3サイクルします.(aof.c/loadAppendOnlyFile)

 

三.AOF書き換え

aof永続化は、実行される書き込みコマンドを保存することによってデータベースの状態を記録するため、aofファイルの体積はサーバの実行時間が長くなるにつれてますます大きくなる.
この問題を解決するために、redisはAOF書き換え機能を提供する. 
1.AOF書き換えの実現
AOFの書き換えは既存のAOFファイルの読み取り、分析または書き込み操作を必要とせず、この機能は現在のデータベース状態を読み取ることによって実現される.
AOF書き換えの入口はaofにある.c/rewriteAppendOnlyFileRio

/*
 * aof    
 */
int rewriteAppendOnlyFileRio(rio *aof) {
    dictIterator *di = NULL;
    dictEntry *de;
    size_t processed = 0;
    int j;

    //        
    for (j = 0; j < server.dbnum; j++) {
        char selectcmd[] = "*2\r
$6\r
SELECT\r
";
        redisDb *db = server.db+j;
        dict *d = db->dict;
        if (dictSize(d) == 0) continue;
        di = dictGetSafeIterator(d);

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

            /* Save the key and associated value */
            if (o->type == OBJ_STRING) {
                /* Emit a SET command */
                char cmd[]="*3\r
$3\r
SET\r
";
                if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                /* Key and value */
                //  key
                if (rioWriteBulkObject(aof,&key) == 0) goto werr;
                //  value
                if (rioWriteBulkObject(aof,o) == 0) goto werr;
            } else if (o->type == OBJ_LIST) { //list 
                if (rewriteListObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_SET) {
                if (rewriteSetObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_ZSET) {
                if (rewriteSortedSetObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_HASH) {
                if (rewriteHashObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_STREAM) {
                if (rewriteStreamObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_MODULE) {
                if (rewriteModuleObject(aof,&key,o) == 0) goto werr;
            } else {
                serverPanic("Unknown object type");
            }
            /* Save the expire time */
            if (expiretime != -1) {
                char cmd[]="*3\r
$9\r
PEXPIREAT\r
";
                if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                if (rioWriteBulkObject(aof,&key) == 0) goto werr;
                if (rioWriteBulkLongLong(aof,expiretime) == 0) goto werr;
            }
            /* Read some diff from the parent process from time to time. */
            if (aof->processed_bytes > processed+AOF_READ_DIFF_INTERVAL_BYTES) {
                processed = aof->processed_bytes;
                aofReadDiffFromParent();
            }
        }
        dictReleaseIterator(di);
        di = NULL;
    }
    return C_OK;

werr:
    if (di) dictReleaseIterator(di);
    return C_ERR;
}

2.AOFバックグラウンド書き換え
質問:AOF書き換えは大量の書き込み操作を行うため、この関数を呼び出すプロセスは長時間にわたって詰まってしまう.redisサーバはコマンドリクエストを単一のスレッドで処理するため、サーバが書き換えを呼び出すと、サーバはクライアントのコマンドリクエストを処理できません.
解決:この問題を解決するために、redisはaofの書き換えプログラムをサブプロセスで実行する.
 
サブプロセスの使用による新しい問題:サブプロセスはaof書き換え中にサーバプロセスがコマンド要求を処理し続け、新しいコマンドは既存のデータベース状態を変更し、データベースの現在のデータ状態とaofファイルのデータが一致しない.
解決:redisサーバは、サーバがサブプロセスを作成した後に使用を開始する書き換えバッファを設定します.サーバは、AOFバッファとAOF書き換えバッファに書き込みコマンドを同時に送信します.サブプロセスがAOF書き換えを完了すると、親プロセスに信号が送信され、親プロセスが信号を受信すると、処理関数が呼び出されます.

aof書き換えバッファ内のすべてのコンテンツをAOFファイルに書き込む.

古いAOFファイルを新しいAOFで置き換えます.

aofバックグラウンド全体の書き換え過程は、信号処理関数だけがサーバに渋滞をもたらし、AOF書き換えがサーバに与える影響を最小限に抑えた.