Redis Source - Hash | HoleLin's Blog

参考文献

极客时间专栏 <<Redis 源码剖析与实战>> 第三讲如何实现一个性能优异的Hash表

`Hash`

源码基于Redis 6.2分支

`Redis`如何实现链式哈希

什么是哈希冲突？

由于键空间会大于Hash表空间,这就导致在用Hash函数把键映射到Hash表空间时,不可避免地会出现不同的键被映射到数组的同一个位置上.而如果同一个位置只能保存一个键值对,就会导致Hash表保存的数据非常有限,这就是常说的哈希冲突.

如何解决哈希冲突呢?

方案一: 链式哈希
- 局限: 链式哈希的链不能太长,否则会降低Hash表性能
方案二: 就是当链式哈希的链长达到一定长度时,可以使用rehash
- 改进方法: 渐进式rehash

结构体

// src/dict.h
typedef struct dict {
    // 指向 dictType 类型的指针,用于保存字典的类型信息.
    // 通常,dictType 结构体包含了一系列函数指针,用于定义字典的操作,比如插入、删除、查找等
    dictType *type;
    // 这是一个指向私有数据的指针,用于保存字典操作所需的任何附加数据.这个指针允许用户向字典中传递自定义的数据,以满足特定的需求
    void *privdata;
    // 两个Hash表,交替使用,用于rehash操作
    // 这种方式可以避免在 rehash 过程中对字典的操作阻塞
    dictht ht[2];
    // 这是一个标识值,用于表示是否正在进行 rehash 操作.
    // 如果 rehashidx 的值为 -1,则表示没有进行 rehash 操作.
    // 否则,它表示正在进行 rehash 操作,并且指示了当前正在 rehash 的索引位置
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    // 这是一个整数值,用于暂停 rehash 操作的标志.
    // 如果 pauserehash 的值大于 0,则表示正在暂停 rehash 操作.小于 0 的值通常表示编码错误
    int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
} dict;

typedef struct dictht {
    // 这是一个指向指针的指针,实际上是一个二维数组.它用来存储指向 dictEntry 结构体的指针,即字典条目.
    // 通过哈希函数,字典的键被映射到这个二维数组中的某个位置
    dictEntry **table;
    // 这是哈希表的大小,即它包含的槽（slot）的数量.哈希表的大小通常是一个质数,以提高哈希算法的效率
    unsigned long size;
    // 这是哈希表大小的掩码,用于快速计算键的哈希值对应的槽索引.掩码的值总是等于 size - 1,因为哈希表的大小必须是 2 的幂次方,这样才能确保掩码的二进制形式全为 1
    unsigned long sizemask;
    // 这是哈希表中当前使用的槽的数量.它表示已经存储了多少个字典条目
    unsigned long used;
} dictht;

typedef struct dictEntry {
    // 存储实际的建值
    void *key;
    // union { ... } v;：这是一个联合体（union）,用来存储值的数据.联合体允许在同一个内存位置存储不同的数据类型.
    // 在这里,联合体 v 可以存储四种类型的值：指针类型的 void *val、64 位无符号整数 uint64_t u64、64 位有符号整数 int64_t s64 和双精度浮点数 double d.
    // 这样设计的好处是可以根据具体情况选择存储哪种类型的值,节省内存空间
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    // 指向下一个哈希项的指针,用来实现链式哈希
    struct dictEntry *next;
} dictEntry;

`Redis`如何实现`rehash`？

`Redis`实现`rehash`的基本思路

首先,Redis准备了两个哈希表,用于rehash时交替保存数据.
其次,在正常服务请求阶段,所有的键值对写入哈希表ht[0].
接着,当进行rehash时,键值对被迁移到哈希表ht[1]中.
最后,当迁移完成后,ht[0]的空间会被释放,并把ht[1]的地址赋值给ht[0],ht[1]的表大小设置为0.这样一来,又回到了正常服务请求的阶段,ht[0]接收和服务请求,ht[1]作为下一次rehash时的迁移表.

什么时候触发`rehash`?

Redis用来判断是否触发rehash的函数是_dictExpandIfNeeded

/* 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. */
    // 条件一; 如果Hash表为空,将Hash表扩为初始大小
    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 (!dictTypeExpandAllowed(d))
        return DICT_OK;
    // 条件二: 如果Hash表承载的元素个数超过其当前大小,并且可以进行扩容
    if ((dict_can_resize == DICT_RESIZE_ENABLE &&
         d->ht[0].used >= d->ht[0].size) ||
         // 条件三: 或者Hash表承载的元素个数已是当前大小的5倍
        (dict_can_resize != DICT_RESIZE_FORBID &&
         d->ht[0].used / d->ht[0].size > dict_force_resize_ratio))
    {
        return dictExpand(d, d->ht[0].used + 1);
    }
    return DICT_OK;
}

_dictExpandIfNeeded函数中定义了三个扩容条件.
- 条件一：ht[0]的大小为0,此时Hash表是空的,所以Redis就需要将Hash表空间设置为初始大小,而这是初始化的工作,并不属于rehash操作.
- 条件二：ht[0]承载的元素个数已经超过了ht[0]的大小,同时Hash表可以进行扩容.
- 条件三：ht[0]承载的元素个数,是ht[0]的大小的dict_force_resize_ratio倍,其中,dict_force_resize_ratio的默认值是5.
- 当前承载的元素个数(d->ht[0].used)和Hash表当前设定的大小(d->ht[0].size).这两个值的比值一般称为负载因子（load factor）.也就是说,Redis判断是否进行rehash的条件,就是看load factor是否大于等于1和是否大于5

dict_can_resize控制是否开启rehash功能

// src/dict.h
typedef enum {
    DICT_RESIZE_ENABLE,
    DICT_RESIZE_AVOID,
    DICT_RESIZE_FORBID,
} dictResizeEnable;

// src/server.c
/* 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 or rehash the tables accordingly to the fact we have an
 * active fork child running. */
void updateDictResizePolicy(void) {
    // 如果当前进程是一个 fork 出的子进程（server.in_fork_child != CHILD_TYPE_NONE）,
    // 则禁止字典进行大小调整,因为在子进程中进行调整可能会影响到父进程的内存（由于写时复制机制）
    if (server.in_fork_child != CHILD_TYPE_NONE)
        dictSetResizeEnabled(DICT_RESIZE_FORBID);
    // 如果存在活跃的子进程（通过 hasActiveChildProcess() 函数检查）,则避免字典进行大小调整,以免影响到子进程的内存.
    else if (hasActiveChildProcess())
        dictSetResizeEnabled(DICT_RESIZE_AVOID);
    else
        // 以上条件都不满足时,开启rehash
        dictSetResizeEnabled(DICT_RESIZE_ENABLE);
}

在src/dict.c文件中,_dictExpandIfNeeded被调用关系


dictAdd

dictReplace				-->   dictAddRaw --> _dictKeyIndex  --> _dictExpandIfNeeded

dictAddOrFind

dictAdd: 用来往Hash表中添加一个键值对.

/* Add an element to the target hash table */
int dictAdd(dict *d, void *key, void *val)
{
    dictEntry *entry = dictAddRaw(d,key,NULL);

    if (!entry) return DICT_ERR;
    dictSetVal(d, entry, val);
    return DICT_OK;
}

dictReplace: 用来往Hash表中添加一个键值对,或者键值对存在时,修改键值对.

/* Add or Overwrite:
 * Add an element, discarding the old value if the key already exists.
 * Return 1 if the key was added from scratch, 0 if there was already an
 * element with such key and dictReplace() just performed a value update
 * operation. */
int dictReplace(dict *d, void *key, void *val)
{
    dictEntry *entry, *existing, auxentry;

    /* Try to add the element. If the key
     * does not exists dictAdd will succeed. */
    entry = dictAddRaw(d,key,&existing);
    if (entry) {
        dictSetVal(d, entry, val);
        return 1;
    }

    /* Set the new value and free the old one. Note that it is important
     * to do that in this order, as the value may just be exactly the same
     * as the previous one. In this context, think to reference counting,
     * you want to increment (set), and then decrement (free), and not the
     * reverse. */
    auxentry = *existing;
    dictSetVal(d, existing, val);
    dictFreeVal(d, &auxentry);
    return 0;
}

dictAddorFind: 直接调用dictAddRaw

/* Add or Find:
 * dictAddOrFind() is simply a version of dictAddRaw() that always
 * returns the hash entry of the specified key, even if the key already
 * exists and can't be added (in that case the entry of the already
 * existing key is returned.)
 *
 * See dictAddRaw() for more information. */
dictEntry *dictAddOrFind(dict *d, void *key) {
    dictEntry *entry, *existing;
    entry = dictAddRaw(d,key,&existing);
    return entry ? entry : existing;
}

总结: Redis中触发rehash操作的关键,就是 _dictExpandIfNeeded 函数和 updateDictResizePolicy函数._dictExpandIfNeeded函数会根据Hash表的负载因子以及能否进行 rehash 的标识,判断是否进行 rehash,而 updateDictResizePolicy 函数会根据 RDB 和 AOF 的执行情况,启用或禁用 rehash.

`rehash`扩容括多大

在Redis中,rehash对 Hash 表空间的扩容是通过调用dictExpand函数来完成的.dictExpand函数的参数有两个,一个是要扩容的 Hash 表,另一个是要扩到的容量.
1
2
3
4
5
// src/dict.c
/* return DICT_ERR if expand was not performed */
int dictExpand(dict *d, unsigned long size) {
return _dictExpand(d, size, NULL);
}
从2020/12/6日Redis 6.2版本开始扩容大小由原来的d->ht[0].used*2变为了d->ht[0].used+1
Had a discuss this with @yossigo @guybe7 @MeirShpilraien @inbaryuval @yaacovhazan-Redislabs, and here’s the conclusions:
- It’s hard to predict the impact of such a mechanism when enabled on all dicts, we feel we should focus on just solving the major known problem of the main dict, in which is easier to realize all the implications of this change. for that, a new field can be added in the dictType struct, so it doesn’t have a cost per instance of the dictionary.
- this is essentially a balance between memory and execution performance, currently redis aims to hit a dict saturation ratio between 0.5 and 1.0 (as soon at it reaches 1.0 it bring it back to 0.5), but these two constants can in theory be set to different ones (0.8 and 1.2 or anything else).
- the code that does both used*2 and later _dictNextPower seems like a bug. since the check is that used >= size, the outcome of this will usually be the same as _dictNextPower(used+1) (no need for the *2). the only case where it’ll differ is when dict_can_resize is false during fork, so that later the _dictNextPower(used*2) will cause the dict to jump to *4 (i.e. _dictNextPower(1025*2) will give you 4096).
proposals:
1. replace the *2 with +1 (will only affect wakeup after fork)
2. when there is enough memory, keep the logic like it is (saturation between 0.5 and 1.0), but when the memory is in shortage (rehash is going to cause eviction), we can relax the saturation threshold, but only up to a limit. let’s say up to saturation of 1.61803 (after that we will prefer to rehash, despite the eviction it’ll cause).
3. we can probably revise the rehashing code to be a little bit less memory hungry during rehashing (i’ll open another issue to describe the design).
- 具体变更原因,可查看Limit the main db and expires dictionaries to expand #7954

`rehash`如何执行

/* return DICT_ERR if expand was not performed */
int dictExpand(dict *d, unsigned long size) {
    return _dictExpand(d, size, NULL);
}

/* Expand or create the hash table,
 * when malloc_failed is non-NULL, it'll avoid panic if malloc fails (in which case it'll be set to 1).
 * Returns DICT_OK if expand was performed, and DICT_ERR if skipped. */
/*
 * 创建一个新的哈希表，并根据字典的情况，选择以下其中一个动作来进行：
 *
 * 1) 如果字典的 0 号哈希表为空，那么将新哈希表设置为 0 号哈希表
 * 2) 如果字典的 0 号哈希表非空，那么将新哈希表设置为 1 号哈希表，
 *    并打开字典的 rehash 标识，使得程序可以开始对字典进行 rehash
 *
 * size 参数不够大，或者 rehash 已经在进行时，返回 DICT_ERR 。
 *
 * 成功创建 0 号哈希表，或者 1 号哈希表时，返回 DICT_OK 。
 *
 * T = O(N)
 */
int _dictExpand(dict *d, unsigned long size, int* malloc_failed)
{
    if (malloc_failed) *malloc_failed = 0;

    /* the size is invalid if it is smaller than the number of
     * elements already inside the hash table */
    /* 如果正在rehash或者新哈希表的大小小于现已使用,则返回error */
    if (dictIsRehashing(d) || d->ht[0].used > size)
        return DICT_ERR;
    // 新哈希表
    dictht n; /* the new hash table */
    // 计算扩展或缩放新哈希表的大小(调用下面函数_dictNextPower())
    unsigned long realsize = _dictNextPower(size);

    /* Detect overflows */
    if (realsize < size || realsize * sizeof(dictEntry*) < realsize)
        return DICT_ERR;

    /* Rehashing to the same table size is not useful. */
    /* 如果计算出哈希表size与现哈希表大小一样,也返回error */
    if (realsize == d->ht[0].size) return DICT_ERR;

    /* Allocate the new hash table and initialize all pointers to NULL */
    // 为哈希表分配空间，并将所有指针指向 NULL
    n.size = realsize;
    n.sizemask = realsize-1;
    if (malloc_failed) {
        // 为table指向dictEntry 分配内存
        n.table = ztrycalloc(realsize*sizeof(dictEntry*));
        *malloc_failed = n.table == NULL;
        if (*malloc_failed)
            return DICT_ERR;
    } else
        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. */
    // 如果0号哈希表为空，那么这是一次初始化：
    // 程序将新哈希表赋给0号哈希表的指针，然后字典就可以开始处理键值对了。
    if (d->ht[0].table == NULL) {
        d->ht[0] = n;
        return DICT_OK;
    }

    /* Prepare a second hash table for incremental rehashing */
    /* 如果ht[0]不为空,则初始化ht[1]为当前键值对的哈希表,并开启渐进式rehash模式 */
    d->ht[1] = n;
  	// 此次d->rehashidx赋值为0,为后续渐进式rehash模式,遍历每个bucket做准备
    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;

    // 如果要扩容的大小已经超过最大值,则返回最大值加1
    if (size >= LONG_MAX) return LONG_MAX + 1LU;
    // 扩容大小没有超过最大值
    while(1) {
        // 如果扩容大小大于等于最大值,就返回截至当前扩到的大小
        if (i >= size)
            return i;
        // 每一步扩容都在现有大小基础上乘以2
        i *= 2;
    }
}

渐进式`rehash`如何实现？

为什么要实现渐进式`rehash`？

其实这是因为,Hash表在执行rehash时,由于Hash表空间扩大,原本映射到某一位置的键可能会被映射到一个新的位置上,因此,很多键就需要从原来的位置拷贝到新的位置.而在键拷贝时,由于Redis主线程无法执行其他请求,所以键拷贝会阻塞主线程,这样就会产生**rehash开销**
简单来说,渐进式 rehash 的意思就是 Redis 并不会一次性把当前 Hash 表中的所有键,都拷贝到新位置,而是会分批拷贝,每次的键拷贝只拷贝 Hash 表中一个 bucket 中的哈希项.这样一来,每次键拷贝的时长有限,对主线程的影响也就有限了

渐进式`rehash`在代码层面是如何实现的呢？

两个关键函数： _dictRehashStep和dictRehash

dictAddRaw

dictGenericDelete

dictFind								--> _dictRehashStep -->  dictRehash

dictGetRandomKey

dictGetSomeKeys

/* This function performs just a step of rehashing, and only if hashing has
 * not been paused for our hash table. When we have iterators in the
 * middle of a rehashing we can't mess with the two hash tables otherwise
 * some element can be missed or duplicated.
 * 在字典不存在安全迭代器的情况下，对字典进行单步 rehash 。
 *
 * 字典有安全迭代器的情况下不能进行 rehash ，
 * 因为两种不同的迭代和修改操作可能会弄乱字典。
 
 * This function is called by common lookup or update operations in the
 * dictionary so that the hash table automatically migrates from H1 to H2
 * while it is actively used.
 * 这个函数被多个通用的查找、更新操作调用，
 * 它可以让字典在被使用的同时进行 rehash 。
 *
 * T = O(1)
 */
static void _dictRehashStep(dict *d) {
    // 给dictRehash传入的循环次数参数为1,表明每迁移完一个bucket,就执行正常操作
    if (d->pauserehash == 0) dictRehash(d,1);
}


/* 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.
 *
 * 执行 N 步渐进式 rehash 。
 *
 * 返回 1 表示仍有键需要从 0 号哈希表移动到 1 号哈希表，
 * 返回 0 则表示所有键都已经迁移完毕。
 * 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. 
 
 * 注意，每步 rehash 都是以一个哈希表索引（桶）作为单位的，
 * 一个桶里可能会有多个节点，
 * 被 rehash 的桶里的所有节点都会被移动到新哈希表。
 *
 * T = O(N)
 */
int dictRehash(dict *d, int n) {
    int empty_visits = n*10; /* Max number of empty buckets to visit. */
    unsigned long s0 = d->ht[0].size;
    unsigned long s1 = d->ht[1].size;
    if (dict_can_resize == DICT_RESIZE_FORBID || !dictIsRehashing(d)) return 0;
    if (dict_can_resize == DICT_RESIZE_AVOID && 
        ((s1 > s0 && s1 / s0 < dict_force_resize_ratio) ||
         (s1 < s0 && s0 / s1 < dict_force_resize_ratio)))
    {
        return 0;
    }
    // 主循环,根据要拷贝的bucket数量n,循环n次后停止或ht[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 */
        // rehashidx 变量表示的是当前 rehash 在对哪个 bucket 做数据迁移.
        // 比如,当 rehashidx 等于 0 时,表示对 ht[0]中的第一个 bucket 进行数据迁移；
        // 当 rehashidx 等于 1 时,表示对 ht[0]中的第二个 bucket 进行数据迁移,以此类推.
        assert(d->ht[0].size > (unsigned long)d->rehashidx);
        while(d->ht[0].table[d->rehashidx] == NULL) {
            // 判断 rehashidx 指向的 bucket 是否为空,如果为空,那就将 rehashidx 的值加 1,检查下一个 bucket
            d->rehashidx++;
            // 渐进式 rehash 在执行时设置了一个变量 empty_visits,用来表示已经检查过的空 bucket,当检查了一定数量的空 bucket 后,这一轮的 rehash 就停止执行,转而继续处理外来请求,避免了对 Redis 性能的影响.
            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 */
        // 如果rehashidx指向的bucket不为空
        while(de) {
            uint64_t h;
            // 获得同一个bucket中下一个哈希项
            nextde = de->next;
            // 根据扩容后的哈希表ht[1]大小,计算当前哈希项在扩容后哈希表中的bucket位置
            /* Get the index in the new hash table */
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            // 将当前哈希项添加到扩容后的哈希表ht[1]中
            de->next = d->ht[1].table[h];
            d->ht[1].table[h] = de;
            // 减少当前哈希表的哈希项个数
            d->ht[0].used--;
            // 增加扩容后哈希表的哈希项个数
            d->ht[1].used++;
            // 指向下一个哈希项
            de = nextde;
        }
        // 如果当前bucket中已经没有哈希项了,将该bucket置为NULL
        d->ht[0].table[d->rehashidx] = NULL;
        // 将rehash加1,下一次将迁移下一个bucket中的元素
        d->rehashidx++;
    }

    /* Check if we already rehashed the whole table... */
    // 判断ht[0]的数据是否迁移完成
    if (d->ht[0].used == 0) {
        // ht[0]迁移完后,释放ht[0]内存空间
        zfree(d->ht[0].table);
        // 让ht[0]指向ht[1],以便接受正常的请求
        d->ht[0] = d->ht[1];
        // 重置ht[1]的大小为0
        _dictReset(&d->ht[1]);
        //设置全局哈希表的rehashidx标识为-1,表示rehash结束
        d->rehashidx = -1;
        // 返回0,表示ht[0]中所有元素都迁移完
        return 0;
    }

    /* More to rehash... */
    // 返回1,表示ht[0]中仍然有元素没有迁移完
    return 1;
}

dictRehash函数的整体逻辑包括两部分：
- 首先,该函数会执行一个循环,根据要进行键拷贝的bucket数量n,依次完成这些bucket内部所有键的迁移.当然,如果ht[0]哈希表中的数据已经都迁移完成了,键拷贝的循环也会停止执行.
- 其次,在完成了n个bucket拷贝后,dictRehash函数的第二部分逻辑,就是判断ht[0]表中数据是否都已迁移完.如果都迁移完了,那么ht[0]的空间会被释放.因为Redis在处理请求时,代码逻辑中都是使用ht[0],所以当rehash执行完成后,虽然数据都在ht[1]中了,但Redis仍然会把ht[1]赋值给ht[0],以便其他部分的代码逻辑正常使用.
- 而在ht[1]赋值给ht[0]后,它的大小就会被重置为0,等待下一次rehash.与此同时,全局哈希表中的rehashidx变量会被标为-1,表示rehash结束了

附录

src/dict.c完整版注释

参考文献

Hash

Redis如何实现链式哈希

什么是哈希冲突？

如何解决哈希冲突呢?

结构体

Redis如何实现rehash？

Redis实现rehash的基本思路

什么时候触发rehash?

rehash扩容括多大

rehash如何执行

渐进式rehash如何实现？

为什么要实现渐进式rehash？

渐进式rehash在代码层面是如何实现的呢？

附录