redis源码学习之dict

参考《Redis 设计与实现》 (基于redis3.0.0) 作者:黄健宏
学习redis3.2.13

介绍
字典的结构
键值对节点 dictEntry
哈希表结构 dictht
字典结构 dict
hash与rehash
键冲突的原因与处理
rehash
rehash过程概览
扩容、缩容
为什么首先检查扩容、缩容条件
容量计算
何时扩容、缩容
rehash前奏:准备ht[1]
rehashing
渐进式rehash执行期间的哈希表操作
迭代器
迭代器的结构
迭代器的获取与释放
迭代器的游走方式
使用迭代器遍历
后记

toc

介绍

字典是一种存储键值对的抽象数据结构。其键与值相互关联，在字典中，通过键可以找到相应的值。
字典的实现方式是多种多样，可以是数组、也可以是哈希表、或者也可以是树。C++中的有序字典map与无序字典unordered_map就分别使用了红黑树与哈希表来实现。
redis实现的字典使用的是哈希表的方式。

字典的结构

键值对节点 dictEntry

typedef struct dictEntry {
    void *key;
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    //处理键冲突时使用，指向下个节点的地址
    struct dictEntry *next;
} dictEntry;

• 为了支持存储多种类型的value，同时节省空间，redis选用了联合来表示值。
• redis在处理键的冲突时，采用了链地址法，使用一个指针来记录的下个冲突节点的位置。

哈希表结构 dictht

typedef struct dictht {
    //存放dictEntry *的数组(里面的元素也叫bucket)
    dictEntry **table;
    //数组的大小，必须是2的N次方(初始情况为0)
    unsigned long size;
    //用于数组下标计算的掩码，总是等于size - 1
    unsigned long sizemask;
    //当前dictht 中已有节点的总和
    unsigned long used;
} dictht;

• 值得注意的是size和used没有任何关系，size是数组的大小，而used是dictht中已有的键值对节点数量，包括数组中使用的节点，以及发生冲突后由链表连接起来的节点
• 这里size必须是2的N次方， 且sizemask等于size - 1，是为了配合hash计算数组的下标，redis通过位操作来提高了性能

//下标的计算方式
index = hash % size
// 当size为2的N次方时，
hash % size == hash & (size - 1) <====> hash & (sizemask)

字典结构 dict

typedef struct dictType {
    //哈希函数的函数指针
    unsigned int (*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;
...
typedef struct dict {
    //自定义键值对操作的结构
    dictType *type;
    //私有数据，创建字典时传入，可配合结构dictType中的函数使用
    void *privdata;
    //两个哈希表，正常使用表0，rehash时才会用到表1
    dictht ht[2];
    //渐进式rehash所需，表示rehash进行到dictEntry *数组的哪个位置，-1表示没有在rehash(或是rehash完成)
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    //dict当前存在的迭代器的个数
    int iterators; /* number of iterators currently running */
} dict;

未发生rehash、键冲突时，字典的示例(还是取书中的例子):

hash与rehash

键冲突的原因与处理

1. hash算法具有冲突必然性： 导致即使两个键是不同的，得出的哈希值也可能是一样的。哈希冲突必然性见鸽巢原理。
2. 已有节点多于dictEntry *大小：即使计算出的hash值不一样，对数组长度取余后得到的下标就会重复，必然会有键冲突。

redis处理冲突的方法是链地址法。即是使用一个链表来存储该键所有冲突的节点。 为了能快速存入键值对，redisi直接将新的键值对插入链表头部。
但是，随着节点的增多，链表会越来越长，严重影响字典性能。需要一定的方法去处理这个问题。

rehash

rehash过程概览

1. rehas之前会根据已有的节点个数和dictEntry *数组的大小综合判断需要扩容还是缩容:

• 当向字典添加节点时，会判断是否符合扩容条件
• redis后台定时判断是否符合缩容条件

1. 满足条件则分配一个足够的空间给ht[1],此时字典同时拥有两个dictEntry *数组
2. 将字典中的rehashidx置0，表明开始rehash，将要迁移ht[0]数组中0位置的元素
   4.重算ht[0]中数组0位置元素里的全部节点在ht[1]的下标，并根据下标将节点放入ht[1]，并更新rehashidx以便下次rehash
3. 迁移完ht[0]中的全部节点，释放ht[0]中的数组，并用ht[1]替换ht[0]，最后重置ht[1]，并将rehashidx设为-1表示rehash结束

redis在迁移dictEntry *数组时，并不是一次全部迁移完成的。而是一部分一部分迁移：

• 查找、添加、更新、删除内部进行的是一步迁移，一次只迁移dictEntry *数组中一个元素/bucket(会迁移完其上的链表里的全部节点)

static void _dictRehashStep(dict *d) {
    if (d->iterators == 0) dictRehash(d,1);    //当不存在安全迭代器时才触发一步迁移
}

• redis后台定时通过函数dictRehashMilliseconds迁移，这种迁移方式里，迁移1毫秒，并在1毫秒内一次试图迁移100个元素/bucket(会迁移完其上的链表里的全部节点)

//dict.c
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;
}
//server.c
int incrementallyRehash(int dbid) {
...
dictRehashMilliseconds(server.db[dbid].dict,1);
dictRehashMilliseconds(server.db[dbid].expires,1);
...    
}

扩容、缩容

为什么首先检查扩容、缩容条件

扩容检查
前面说到了redis字典有键冲突，字典中节点越多，重复的概率越大，链表也就可能越长。
需要一个更大的数组，使得对现有节点重新计算hash并取余后，能尽量落到数组的空槽里，使时间复杂度从O(N)变为最初的O(1)。
缩容检查
由于节点数量一直在随着程序运行进行着动态增减，只有扩容没有缩容的话，势必会造成不必要的内存浪费。所以，需要对字典的空间进行缩调。

容量计算

为了使用位操作取余，容量(除数)的为2的N次方

static unsigned long _dictNextPower(unsigned long size)
{
    unsigned long i = DICT_HT_INITIAL_SIZE;    //#define DICT_HT_INITIAL_SIZE     4

    if (size >= LONG_MAX) return LONG_MAX + 1LU;
    while(1) {
        if (i >= size)
            return i;
        i *= 2;
    }
}

何时扩容、缩容

扩容条件

• 添加节点时，dictEntry *数组为空(字典刚创建)
• ht[0]已有节点数大于数组大小，同时，开启了允许resize标志或已有节点数与数组大小之商大于5，即负载因子大于5

static int _dictExpandIfNeeded(dict *d)
{
    /* Incremental rehashing already in progress. Return. */
    //dict处于渐进式rehash状态不用扩容，是因为进行渐进式rehash的前置条件是扩容完成
    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 ||                                                            //static int dict_can_resize = 1;
         d->ht[0].used/d->ht[0].size > dict_force_resize_ratio))    //static unsigned int dict_force_resize_ratio = 5;
    {
        return dictExpand(d, d->ht[0].used*2);    //指定已有节点的2倍扩容，ht[0].used*2的2的N次方扩容
    }
    return DICT_OK;
}

resize标志由server.c中updateDictResizePolicy控制

/* This function is called once a background process of some kind terminates,
 * as we want to avoid resizing the hash tables when there is a child in order
 * to play well with copy-on-write (otherwise when a resize happens lots of
 * memory pages are copied). The goal of this function is to update the ability
 * for dict.c to resize the hash tables accordingly to the fact we have o not
 * running childs. */
void updateDictResizePolicy(void) {
    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1)
        dictEnableResize();
    else
        dictDisableResize();
}

结合注释与代码可以知道，由于redis想利用好写时复制，所以，当后台进程开始生成/重写RDB/AOF文件或结束生成/重写RDB/AOF文件，会调用此函数来关闭/开启字典的扩容。
总结下来，也可以这样理解：
服务器目前没有在执行 BGSAVE 命令或者 BGREWRITEAOF 命令， 并且哈希表的负载因子大于等于 1
服务器目前正在执行 BGSAVE 命令或者 BGREWRITEAOF 命令， 并且哈希表的负载因子大于等于 5
缩容条件
已有节点数与数组大小之商大于10%，即负载因子小于0.1时发生缩容

int htNeedsResize(dict *dict) {
    long long size, used;

    size = dictSlots(dict);
    used = dictSize(dict);
    return (size > DICT_HT_INITIAL_SIZE &&
            (used*100/size < HASHTABLE_MIN_FILL));    //#define HASHTABLE_MIN_FILL        10      /* Minimal hash table fill 10% */
}

缩容判断函数由redis定期调用

rehash前奏:准备ht[1]

ht[1]中数组空间的准备、以及rehash开启的标志 通过dictExpand来处理

#define dictIsRehashing(d) ((d)->rehashidx != -1)
...
int dictExpand(dict *d, unsigned long size)
{
    dictht n; /* the new hash table */
    unsigned long realsize = _dictNextPower(size);    //从4开始找大于等于size的最小2的N次方做为新大小

    /* the size is invalid if it is smaller than the number of
     * elements already inside the hash table */
    //字典正在rehash时，ht[0]与ht[1]都有存在节点的可能，后面的赋值操作可能导致节点丢失，不允许扩容
    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;    //ht[1]已准备好，可以从ht[0]的d->rehashidx处的bucket移动到ht[1]
    return DICT_OK;
}

如果是新创建的字典，dictEntry *数组是不会有任何容量的，扩容函数也是根据该数组是否为空，来确定是处理新字典还是准备rehash

rehashing

字典的rehash操作由dictRehash实现，此函数执行N步渐进式rehash，N决定了函数一次处理几个bucket(dictEntry *数组中的元素)

int dictRehash(dict *d, int n) {
    //一次rehash只会访问最多10个空桶便会返回，empty_visits用于记录已访问空桶个数
    int empty_visits = n*10; /* Max number of empty buckets to visit. */
    //没有准备准备好ht[1]不能rehash
    if (!dictIsRehashing(d)) return 0;
    //n步rehash
    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;    //一次最多访问10个空桶
        }
        de = d->ht[0].table[d->rehashidx];
        /* Move all the keys in this bucket from the old to the new hash HT */
        while(de) {
            unsigned int h;

            nextde = de->next;
            /* Get the index in the new hash table */
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            //将节点放入新哈希表数组table的h位置(如果形成了链表为头插)
            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++;    //下个将要rehash的位置
    }

    /* 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;    //rehash完成
        return 0;
    }

    /* More to rehash... */
    return 1;
}

渐进式rehash执行期间的哈希表操作

因为在进行渐进式 rehash 的过程中， 字典会同时使用 ht[0] 和 ht[1] 两个哈希表， 所以在渐进式 rehash 进行期间， 字典的删除（delete）、查找（find）、更新（update）等操作会在两个哈希表上进行： 比如说， 要在字典里面查找一个键的话， 程序会先在 ht[0] 里面进行查找， 如果没找到的话， 就会继续到 ht[1] 里面进行查找， 诸如此类。
另外， 在渐进式 rehash 执行期间， 新添加到字典的键值对一律会被保存到 ht[1] 里面， 而 ht[0] 则不再进行任何添加操作： 这一措施保证了 ht[0] 包含的键值对数量会只减不增， 并随着 rehash 操作的执行而最终变成空表。

迭代器

dict中的迭代器有两种，一种是安全迭代器，另一种是非安全迭代器

• 安全迭代器存在时，dict的查找、添加、更新、删除操作不会进行rehash，这避免了rehash造成的迭代顺序混乱。安全迭代器存在时，可以对dict进行增加、更新、查找、删除操作
• 非安全迭代器存在时，只能对字典进行迭代，如果对字典进行了修改，会导迭代器的指纹前后不一致而触发断言

迭代器的结构

typedef struct dictIterator {
    //被迭代的字典
    dict *d;                   
    //bucket的位置(dictEntry *数组的下标)    
    long index;             
    //table ht的下标(创建迭代器前可能已经处于rehash状态，所以两个ht都需要遍历)   
    //safe表明当前迭代器的种类(安全或非安全)
    int table, safe;
    //当前迭代的节点 与 将迭代的节点
    //在迭代器游走函数dictNext中，当前的节点entry会被返回给用户，并可能被用户删除，保留nextEntry避免指针丢失
    dictEntry *entry, *nextEntry;    
    /* unsafe iterator fingerprint for misuse detection. */
    long long fingerprint;    //非安全迭代器使用的，用于验证的指纹
} dictIterator;

迭代器的获取与释放

获取

dictIterator *dictGetIterator(dict *d)
{
    dictIterator *iter = zmalloc(sizeof(*iter));

    iter->d = d;
    iter->table = 0;
    iter->index = -1;
    iter->safe = 0;
    iter->entry = NULL;
    iter->nextEntry = NULL;
    return iter;
}

dictIterator *dictGetSafeIterator(dict *d) {
    dictIterator *i = dictGetIterator(d);

    i->safe = 1;
    return i;
}

释放

void dictReleaseIterator(dictIterator *iter)
{
    if (!(iter->index == -1 && iter->table == 0)) {
        if (iter->safe)
            iter->d->iterators--;
        else
            assert(iter->fingerprint == dictFingerprint(iter->d));    //迭代完成后，释放迭代器时校验指纹
    }
    zfree(iter);

迭代器的游走方式

dictEntry *dictNext(dictIterator *iter)
{
    while (1) {
        //iter是一个全新迭代器或已迭代完bucket中的一个链表
        if (iter->entry == NULL) {
            //指向正在迭代的ht
            dictht *ht = &iter->d->ht[iter->table];
            //iter是个全新迭代器，是安全迭代器，增加被迭代dict上迭代器数量，否则字典计算指纹
            if (iter->index == -1 && iter->table == 0) {
                if (iter->safe)
                    iter->d->iterators++;
                else
                    iter->fingerprint = dictFingerprint(iter->d);
            }
            //当前bucket中的链表迭代完了，应该迭代下一个bucket，所以增加index，指向下一个bucket
            //若是新迭代器，则应该迭代首个bucket了，也需增加index，使其指向首个bucket
            iter->index++;
            //即将被迭代的bucket的下标大于当前ht的下标，分情况讨论
            if (iter->index >= (long) ht->size) {
                //1、正在rehash时会有两个ht，当前迭代完的是ht[0]，需要再迭代下ht[1]
                if (dictIsRehashing(iter->d) && iter->table == 0) {
                    iter->table++;
                    iter->index = 0;
                    ht = &iter->d->ht[1];
                } else {
                    //2、不在rehash状态，已经完成了迭代
                    break;    
                }
            }
            //迭代到下个bucket(或首个bucket)内的链表头部
            iter->entry = ht->table[iter->index];
        } else {
            //在某个bucket中的链表内迭代
            iter->entry = iter->nextEntry;
        }
        //没有到链表位部，先记录下个节点的位置，再返回迭代到的节点，因为返回的迭代器可能被用户删除
        if (iter->entry) {
            /* We need to save the 'next' here, the iterator user
             * may delete the entry we are returning. */
            iter->nextEntry = iter->entry->next;
            return iter->entry;
        }
    }
    return NULL;
}

使用迭代器遍历

while((de = dictNext(di)) != NULL) {
    //   doSomethingWith(de);    
}

后记

• 由于受当前哈希表空间的限制，节点数量增加到多于哈希表空间时，必定会发生键冲突，链地址法可以解决键冲突
• 链地址法虽然可以解决键冲突，同时也增加了时间复杂度，需要通过rehash来处理这个问题
• rehash也可以避免空间的浪费
• 渐进式rehash可以避免一次性迁移太多节点而造成的的阻塞
• 定时rehash可以避免字典同时拥有两个哈希表太久造成的性能损失
• 合适的时候使用位运算会得到更好的性能
• 通过linux系统的COW机制来在两个进程间共享内存时，避免修改太多内存，可减少内存复制量，从而更好的使用COW

来自为知笔记(Wiz)