Redisソース分析:ディクショナリ実装(-)

12953 ワード

本論文では,Redisのバージョン4.0.11を解析した.
ディクショナリ
キー値データベース(key-value store)は辞書のデータ構造であり、キー(key)と値(value)を関連付けてキー値ペアを形成し、すべてのデータベースの操作はプライマリキー(primary key)によって実現される.ディクショナリは、データベースを表すために使用される以外は、Redisの最下位のハッシュ・キーの最下位実装である.
ディクショナリ構造定義
ディクショナリ構造定義はdict.hファイルにあります.まず、辞書構造の定義を見てみましょう.
/* Unused arguments generate annoying warnings... */
#define DICT_NOTUSED(V) ((void) V)

typedef struct dictEntry {
    void *key;
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    struct dictEntry *next;
} dictEntry;

typedef struct dictType {
    uint64_t (*hashFunction)(const void *key);
    void *(*keyDup)(void *privdata, const void *key);
    void *(*valDup)(void *privdata, const void *obj);
    int (*keyCompare)(void *privdata, const void *key1, const void *key2);
    void (*keyDestructor)(void *privdata, void *key);
    void (*valDestructor)(void *privdata, void *obj);
} dictType;

/* This is our hash table structure. Every dictionary has two of this as we
 * implement incremental rehashing, for the old to the new table. */
typedef struct dictht {
    dictEntry **table;
    unsigned long size;
    unsigned long sizemask;
    unsigned long used;
} dictht;

typedef struct dict {
    dictType *type;
    void *privdata;
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    unsigned long iterators; /* number of iterators currently running */
} dict;

辞書の実装は,ハッシュテーブルの構造を採用している.dictの構造では,ハッシュテーブルを含む構造dicthtの配列であり,配列の長さは2である.各ディクショナリには、主にインクリメンタル再ハッシュ機能を実現するために、大きなデータ量の場合、ハッシュ・テーブルが一度に拡張されるとパフォーマンスの問題を回避するための2つのハッシュ・テーブル構造が含まれています.下位ハッシュテーブルの実装は、典型的な配列+ファスナーの形式であり、dictht構造のメンバーtableが指すdictEntryポインタ配列であり、sizeはハッシュテーブルのサイズを表し、sizemaskはサイズマスクであり、usedはハッシュテーブルで実装されたハッシュバケツの個数を表す.dictは、次のマクロを提供します.
#define dictSlots(d) ((d)->ht[0].size+(d)->ht[1].size)
#define dictSize(d) ((d)->ht[0].used+(d)->ht[1].used)

dictEntryは、key、value、およびチェーンテーブルポインタを格納する辞書のエントリです.valueはunionの構造を使用しており,いずれも異なるタイプの値を格納するのに便利である.dictEntryでは、次のマクロを使用してメンバー変数へのアクセスと操作を容易にします.
#define dictSetSignedIntegerVal(entry, _val_) \
    do { (entry)->v.s64 = _val_; } while(0)

#define dictSetUnsignedIntegerVal(entry, _val_) \
    do { (entry)->v.u64 = _val_; } while(0)

#define dictSetDoubleVal(entry, _val_) \
    do { (entry)->v.d = _val_; } while(0)
    
#define dictGetKey(he) ((he)->key)
#define dictGetVal(he) ((he)->v.val)
#define dictGetSignedIntegerVal(he) ((he)->v.s64)
#define dictGetUnsignedIntegerVal(he) ((he)->v.u64)
#define dictGetDoubleVal(he) ((he)->v.d)

ディクショナリ構造のdictTypeメンバーは、ディクショナリの操作をカスタマイズし、関連する関数インタフェースを登録することができます.privdataは、カスタムインタフェースのパラメータを格納します.
#define dictFreeVal(d, entry) \
   if ((d)->type->valDestructor) \
       (d)->type->valDestructor((d)->privdata, (entry)->v.val)

#define dictSetVal(d, entry, _val_) do { \
   if ((d)->type->valDup) \
       (entry)->v.val = (d)->type->valDup((d)->privdata, _val_); \
   else \
       (entry)->v.val = (_val_); \
} while(0)

#define dictFreeKey(d, entry) \
   if ((d)->type->keyDestructor) \
       (d)->type->keyDestructor((d)->privdata, (entry)->key)

#define dictSetKey(d, entry, _key_) do { \
   if ((d)->type->keyDup) \
       (entry)->key = (d)->type->keyDup((d)->privdata, _key_); \
   else \
       (entry)->key = (_key_); \
} while(0)

#define dictCompareKeys(d, key1, key2) \
   (((d)->type->keyCompare) ? \
       (d)->type->keyCompare((d)->privdata, key1, key2) : \
       (key1) == (key2))
       
#define dictHashKey(d, key) (d)->type->hashFunction(key)

dict構造には、再ハッシュ・プロシージャが行われているかどうかを示す重要なメンバー変数rehashidxがあります.rehashidx=-1の場合、再プロシージャが行われていないことを示し、特定の関数に関連する場合、その値が再ハッシュでどのように変化するかを説明します.ヘッダ・ファイルには、再ハッシュ中かどうかのマクロが定義されています.
#define dictIsRehashing(d) ((d)->rehashidx != -1)

辞書構造の関数実装
このセクションでは、辞書操作のタイプに従って、インタフェースの実装を分析します.辞書の実装はdict.cファイルにあります.
アクションの作成
作成インタフェースdictCreate関数カスタム操作インタフェースのdictTypeタイプ変数と関連するデータポインタを入力し、dictタイプのポインタとして出力します.
dict *dictCreate(dictType *type, void *privDataPtr);

具体的な実装は比較的簡単で、dictのメモリ空間を割り当て、ハッシュテーブル構造と他のメンバー変数を初期化します.なお、dictCreate関数で作成されたdict辞書構造は、ハッシュ・テーブルのtableメンバーがNULLであるため、dictResize()を呼び出して空間割り当てを行う必要があるため、直接使用することはできません.
/* Reset a hash table already initialized with ht_init().
* NOTE: This function should only be called by ht_destroy(). */
static void _dictReset(dictht *ht)
{
   ht->table = NULL;
   ht->size = 0;
   ht->sizemask = 0;
   ht->used = 0;
}

/* Create a new hash table */
dict *dictCreate(dictType *type,
       void *privDataPtr)
{
   dict *d = zmalloc(sizeof(*d));

   _dictInit(d,type,privDataPtr);
   return d;
}

/* Initialize the hash table */
int _dictInit(dict *d, dictType *type,
       void *privDataPtr)
{
   _dictReset(&d->ht[0]);
   _dictReset(&d->ht[1]);
   d->type = type;
   d->privdata = privDataPtr;
   d->rehashidx = -1;
   d->iterators = 0;
   return DICT_OK;
}

アクションの削除
ディクショナリ・インプリメンテーションでは、ディクショナリ・インタフェースを消去する2つの操作が提供されます.ディクショナリEmpty()はコールバック関数のパラメータを提供します.
/* Clear & Release the hash table */
void dictRelease(dict *d);

void dictEmpty(dict *d, void(callback)(void*));

インタフェースの具体的な実現は以下の通りで、資源の回収作業を完成する.
/* Destroy an entire dictionary */
int _dictClear(dict *d, dictht *ht, void(callback)(void *)) {
    unsigned long i;

    /* Free all the elements */
    for (i = 0; i < ht->size && ht->used > 0; i++) {
        dictEntry *he, *nextHe;

        if (callback && (i & 65535) == 0) callback(d->privdata);

        if ((he = ht->table[i]) == NULL) continue;
        while(he) {
            nextHe = he->next;
            dictFreeKey(d, he);
            dictFreeVal(d, he);
            zfree(he);
            ht->used--;
            he = nextHe;
        }
    }
    /* Free the table and the allocated cache structure */
    zfree(ht->table);
    /* Re-initialize the table */
    _dictReset(ht);
    return DICT_OK; /* never fails */
}

/* Clear & Release the hash table */
void dictRelease(dict *d)
{
    _dictClear(d,&d->ht[0],NULL);
    _dictClear(d,&d->ht[1],NULL);
    zfree(d);
}

void dictEmpty(dict *d, void(callback)(void*)) {
    _dictClear(d,&d->ht[0],callback);
    _dictClear(d,&d->ht[1],callback);
    d->rehashidx = -1;
    d->iterators = 0;
}

疑問なのは、なぜ満足しているのか(i&65535)==0の場合、dictClear才呼び出し回掉関数???
ハッシュ表の拡張
ハッシュ・テーブルを拡張する場合は、次のような質問が必要です.
  • 拡張ハッシュ・テーブルのサイズをどのように選択しますか?新しいハッシュ・テーブルのサイズが2のべき乗次数、サイズが現在のハッシュ・テーブルのサイズより大きい最小の2のべき乗次数の値、具体的な値は関数_dictNextPower()で計算します.
  • ハッシュテーブルの拡張と収縮操作はいつ行われますか?次のいずれかの条件が満たされると、プログラムはハッシュ・テーブルを自動的に拡張します.
  • サーバは現在、BGSAVEコマンドまたはBGREWRITEAOFコマンドを実行しておらず、ハッシュテーブルの負荷係数は1以上である.
  • サーバは現在、BGSAVEコマンドまたはBGREWRITEAOFコマンドを実行しており、ハッシュテーブルの負荷係数は5以上である.
  • /* Expand the hash table if needed */
    static int _dictExpandIfNeeded(dict *d)
    {
        /* Incremental rehashing already in progress. Return. */
        if (dictIsRehashing(d)) return DICT_OK;
    
        /* If the hash table is empty expand it to the initial size. */
        if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);
    
        /* If we reached the 1:1 ratio, and we are allowed to resize the hash
         * table (global setting) or we should avoid it but the ratio between
         * elements/buckets is over the "safe" threshold, we resize doubling
         * the number of buckets. */
        if (d->ht[0].used >= d->ht[0].size &&
            (dict_can_resize ||
             d->ht[0].used/d->ht[0].size > dict_force_resize_ratio))
        {
            return dictExpand(d, d->ht[0].used*2);
        }
        return DICT_OK;
    }
    
  • 新しいハッシュ・テーブルの移行作業はどのように行いますか?この質問はdictRehash()関数を解析するときに答えます.

  • 次のコードはハッシュ・テーブル拡張に関するコードです.注意すべき点は次のとおりです.
  • ハッシュ・テーブルの最小サイズはDICT_です.HT_INITIAL_SIZEは、その値が4である.
  • 時に最初に初期化された場合、再ハッシュのプロセスは関係なく、作成したハッシュテーブルをd->ht[0]にコピーするだけでよい.そうでない場合は、rehashidxを0に設定する再ハッシュプロセスを開始する必要があります.
  • /* Resize the table to the minimal size that contains all the elements,
    * but with the invariant of a USED/BUCKETS ratio near to <= 1 */
    int dictResize(dict *d)
    {
       int minimal;
    
       if (!dict_can_resize || dictIsRehashing(d)) return DICT_ERR;
       minimal = d->ht[0].used;
       if (minimal < DICT_HT_INITIAL_SIZE)
           minimal = DICT_HT_INITIAL_SIZE;
       return dictExpand(d, minimal);
    }
    
    /* Expand or create the hash table */
    int dictExpand(dict *d, unsigned long size)
    {
       dictht n; /* the new hash table */
       unsigned long realsize = _dictNextPower(size);
    
       /* the size is invalid if it is smaller than the number of
        * elements already inside the hash table */
       if (dictIsRehashing(d) || d->ht[0].used > size)
           return DICT_ERR;
    
       /* Rehashing to the same table size is not useful. */
       if (realsize == d->ht[0].size) return DICT_ERR;
    
       /* Allocate the new hash table and initialize all pointers to NULL */
       n.size = realsize;
       n.sizemask = realsize-1;
       n.table = zcalloc(realsize*sizeof(dictEntry*));
       n.used = 0;
    
       /* Is this the first initialization? If so it's not really a rehashing
        * we just set the first hash table so that it can accept keys. */
       if (d->ht[0].table == NULL) {
           d->ht[0] = n;
           return DICT_OK;
       }
    
       /* Prepare a second hash table for incremental rehashing */
       d->ht[1] = n;
       d->rehashidx = 0;
       return DICT_OK;
    }
    
    /* Our hash table capability is a power of two */
    static unsigned long _dictNextPower(unsigned long size)
    {
       unsigned long i = DICT_HT_INITIAL_SIZE;
    
       if (size >= LONG_MAX) return LONG_MAX + 1LU;
       while(1) {
           if (i >= size)
               return i;
           i *= 2;
       }
    }
    

    ハッシュ・テーブルの再ハッシュのプロセスは、実際には古いテーブル・データを新しいテーブルに移行するプロセスであり、古いテーブルのある位置のデータを古いテーブルのチェーン・テーブルから削除し、keyに基づいて新しいテーブルのハッシュ位置を計算し、マウントされた対応する新しいテーブル・チェーン・テーブル構造に格納します.rehashidxは、古いテーブルで移行処理が行われたインデックスの場所を記録します.再ハッシュを回避するためには、ブロックが長すぎるため、関数入力パラメータに今回最大再ハッシュのドレインバケツ数を指定するとともに、処理する空きバケツの数を制限します.
    /* Performs N steps of incremental rehashing. Returns 1 if there are still
    * keys to move from the old to the new hash table, otherwise 0 is returned.
    *
    * Note that a rehashing step consists in moving a bucket (that may have more
    * than one key as we use chaining) from the old to the new hash table, however
    * since part of the hash table may be composed of empty spaces, it is not
    * guaranteed that this function will rehash even a single bucket, since it
    * will visit at max N*10 empty buckets in total, otherwise the amount of
    * work it does would be unbound and the function may block for a long time. */
    int dictRehash(dict *d, int n) {
       int empty_visits = n*10; /* Max number of empty buckets to visit. */
       if (!dictIsRehashing(d)) return 0;
    
       while(n-- && d->ht[0].used != 0) {
           dictEntry *de, *nextde;
    
           /* Note that rehashidx can't overflow as we are sure there are more
            * elements because ht[0].used != 0 */
           assert(d->ht[0].size > (unsigned long)d->rehashidx);
           while(d->ht[0].table[d->rehashidx] == NULL) {
               d->rehashidx++;
               if (--empty_visits == 0) return 1;
           }
           de = d->ht[0].table[d->rehashidx];
           /* Move all the keys in this bucket from the old to the new hash HT */
           while(de) {
               uint64_t h;
    
               nextde = de->next;
               /* Get the index in the new hash table */
               h = dictHashKey(d, de->key) & d->ht[1].sizemask;
               de->next = d->ht[1].table[h];
               d->ht[1].table[h] = de;
               d->ht[0].used--;
               d->ht[1].used++;
               de = nextde;
           }
           d->ht[0].table[d->rehashidx] = NULL;
           d->rehashidx++;
       }
    
       /* Check if we already rehashed the whole table... */
       if (d->ht[0].used == 0) {
           zfree(d->ht[0].table);
           d->ht[0] = d->ht[1];
           _dictReset(&d->ht[1]);
           d->rehashidx = -1;
           return 0;
       }
    
       /* More to rehash... */
       return 1;
    }
    

    上記の関数に基づいて、内部には再ハッシュ時間を指定するインタフェースが用意されています.
    long long timeInMilliseconds(void) {
       struct timeval tv;
    
       gettimeofday(&tv,NULL);
       return (((long long)tv.tv_sec)*1000)+(tv.tv_usec/1000);
    }
    
    /* Rehash for an amount of time between ms milliseconds and ms+1 milliseconds */
    int dictRehashMilliseconds(dict *d, int ms) {
       long long start = timeInMilliseconds();
       int rehashes = 0;
    
       while(dictRehash(d,100)) {
           rehashes += 100;
           if (timeInMilliseconds()-start > ms) break;
       }
       return rehashes;
    }