凌云的博客

行胜于言

Redis 5 数据结构(三)quicklist

分类:Redis| 发布时间:2019-04-18 20:23:00


概述

本文是 Redis 5 数据结构系列的第三篇,主要讲解 quicklist 的结构和实现细节。

quciklist 是从 3.2 版本加入 Redis 的,用于取代普通的双向链表结构。 主要用于减少拥有大量节点的列表的内存使用量,具体可以看这里: https://github.com/antirez/redis/pull/2143

quicklist 和节点的实现

quicklist 的结构如下所示:

#define QUICKLIST_NOCOMPRESS 0

typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* 所有的 ziplists 中的所有节点的总数 */
    unsigned long len;          /* quicklistNodes 的个数 */
    int fill : 16;              /* 每个 quicklistNodes 的 fill factor */
    unsigned int compress : 16; /* 0 表示不进行压缩,否则这个值表示,表的头和尾不压缩的节点的个数  */
} quicklist;

在 Redis 内部,当创建一个列表对象时,实际上会使用 quicklist 来保存, 此时 quicklist 的 fill 和 compress 字段是通过选项 list-max-ziplist-size 和 list-compress-depth 来进行设置的。 此选项的具体含义可以查看 redis.conf 的 ADVANCED CONFIG 部分,或者查看我的另一篇文章: Redis 配置详解(五)

quicklist 节点(quicklistNode)的结构如下:

#define QUICKLIST_NODE_ENCODING_RAW 1
#define QUICKLIST_NODE_ENCODING_LZF 2

typedef struct quicklistNode {
    struct quicklistNode *prev;  /* 指向上一个节点 */
    struct quicklistNode *next;  /* 指向下一个节点 */
    unsigned char *zl;           /* 当节点保存的是压缩 ziplist 时,指向 quicklistLZF,否则指向 ziplist */
    unsigned int sz;             /* ziplist 的大小(字节为单位) */
    unsigned int count : 16;     /* ziplist 的项目个数 */
    unsigned int encoding : 2;   /* RAW==1 或者 LZF==2 */
    unsigned int container : 2;  /* NONE==1 或者 ZIPLIST==2 */
    unsigned int recompress : 1; /* 这个节点是否之前是被压缩的 */
    unsigned int attempted_compress : 1; /* 节点太小,不压缩 */
    unsigned int extra : 10; /* 未使用,保留字段 */
} quicklistNode;

quicklistNode 是一个 32 字节的结构体(64 位系统下)。其中使用了位域来让结构保持在 32 字节。 * count 16比特,最大 65536(ziplist 最大为 64K,由于 ziplist 中的每个项目最小要 2 字节,因此这个值实际上会小于 32k) * encoding 2比特,RAW = 1, LZF = 2 * container 2比特,NONE = 1, ZIPLIST = 2 * recompress 1比特,bool,1 表示此节点由于使用中而被临时解压 * attempted_compress 1比特,bool,用于测试 * extra 10比特,未使用,将来使用,同时保持结构体 4 字节对齐

当 quicklistNode 指向的 ziplist 被压缩的情况下,上述的 zl 成员会指向如下的 quicklistLZF 结构:

typedef struct quicklistLZF {
    unsigned int sz; /* LZF size in bytes*/
    char compressed[];
} quicklistLZF;
  • sz 为压缩后的 ziplist 的长度
  • compressed 为压缩后的 ziplist 的实际内容

注意:压缩前的长度保存在 quicklistNode 的 sz 成员中。

下面给出几个 quicklist 的例子。

下图是一个拥有三个 quicklistNode 无压缩的 quicklist 的例子:

下图是一个拥有三个 quicklistNode 压缩深度为 1 的 quicklist 的例子:

quicklist API

quicklist *quicklistCreate(void);

创建一个新的 quicklist。

quicklist *quicklistNew(int fill, int compress);

给定 fill 和 compress 创建一个新的 quicklist。

void quicklistSetCompressDepth(quicklist *quicklist, int depth);
void quicklistSetFill(quicklist *quicklist, int fill);
void quicklistSetOptions(quicklist *quicklist, int fill, int depth);

设置 quicklist 的 fill 和 depth(compress)。

void quicklistRelease(quicklist *quicklist);

释放整个 quicklist。

int quicklistPushHead(quicklist *quicklist, void *value, const size_t sz);

插入一个项目到 quicklist 的头。

int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz);

插入一个项目到 quicklist 的尾。

#define QUICKLIST_HEAD 0
#define QUICKLIST_TAIL -1

void quicklistPush(quicklist *quicklist, void *value, const size_t sz, int where);

根据 where 的值选择插入项目到 quicklist 的头或者尾。

void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl);

创建一个新的 quicklistNode,将其指向预先创建的 ziplist(zl)。 然后将这个节点追加到 quicklist 的末尾。

这个函数通常用于将数据从 RDB 文件加载到内存的情况。

quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist, unsigned char *zl);

将 zl 指向的 ziplist 的所有项目一个一个地追加到 quicklist。

这个方法用于将之前保存在 RDB 文件里的 ziplists 保存到新创建的 quicklist 中 (这个 quicklist 设置的 ziplist 的大小要比保存在 RDB 的 ziplist 要小)。

返回传进来的 quicklist,传进来的 ziplist 会在追加到 quicklist 后释放。

quicklist *quicklistCreateFromZiplist(int fill, int compress, unsigned char *zl);

根据传进来的 ziplist 创建 quicklist(可能会有多个 quicklistNode)。

返回新的 quicklist。释放传进来的 ziplist(zl)。

typedef struct quicklistEntry {
    const quicklist *quicklist;
    quicklistNode *node;
    unsigned char *zi;
    unsigned char *value;
    long long longval;
    unsigned int sz;
    int offset;
} quicklistEntry;

int quicklistIndex(const quicklist *quicklist, const long long index,
                   quicklistEntry *entry);

根据以 0 为开始的索引获取 ‘quicklistEntry’ 信息。 index 大于等于 0 时表示从头往后找,当 index 小于 0 时表示从后往前找, 比如 -1 表示最后一个项目,-2 表示倒数第二个项目等等。

返回值: * 1 表示对应索引的项目 * 0 表示未找到对应索引的项目 返回 quicklist 指定的项目的信息和内容。

quicklistEntry 的各字段含义如下: * quicklist 指向关联的 quicklist * node 指向相应的 quicklistNode * zi 指向 ziplist 中的具体项目 * value,sz 和 longval 对应 ziplistGet 的后面三个参数 * offset 为项目在 ziplist 的索引

void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *node,
                          void *value, const size_t sz);

插入到指定位置的后面

void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *node,
                           void *value, const size_t sz);

插入到指定位置的前面

#define AL_START_HEAD 0
#define AL_START_TAIL 1

typedef struct quicklistIter {
    const quicklist *quicklist;
    quicklistNode *current;
    unsigned char *zi;
    long offset; /* offset in current ziplist */
    int direction;
} quicklistIter;

quicklistIter *quicklistGetIterator(const quicklist *quicklist, int direction);

返回一个 quicklist 迭代器,创建迭代器后需要调用 quicklistNext 对 quicklist 进行迭代。 * direction 为 AL_START_HEAD 表示从头往后迭代,AL_START_TAIL 表示从后往前迭代。

quicklistIter *quicklistGetIteratorAtIdx(const quicklist *quicklist,
                                         int direction, const long long idx);

和 quicklistGetIterator 类似,只是多了一个 idx 参数用于指示迭代器的起始位置。

int quicklistNext(quicklistIter *iter, quicklistEntry *node);

获取迭代其器的下一个项目。 返回 1 表示获取成功,返回 0 表示到达列表的末尾,此时 node 的值是无效的。

注意:当你正在对一个 quicklist 进行迭代时,不能对 quicklist 进行插入操作。 但是你 可以 使用 quicklistDelEntry() 函数删除列表中的项目。

如果你在迭代过程中进行了插入操作,在插入操作后你应该重新创建迭代器。

iter = quicklistGetIterator(quicklist,<direction>);
quicklistEntry entry;
while (quicklistNext(iter, &entry)) {
    if (entry.value)
         [[ use entry.value with entry.sz ]]
    else
         [[ use entry.longval ]]
}

上述是一个使用 quicklist 迭代器的典型例子。

void quicklistReleaseIterator(quicklistIter *iter);

删除迭代器,如果迭代器正指向一个合法的节点,此函数会对节点重新编码(检查此节点之前是否是压缩状态的,若是则将其压缩)。

void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry);

删除指定的项目,可以在迭代过程中使用。

int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
                            int sz);

将索引为 index 的项目替换为 data 和 sz 指向的数据。 函数内部会使用 ziplistDelete 先删除已有项目再用 ziplistInsert 插入新的项目。

  • 返回 1,表示替换成功
  • 返回 0,表示替换失败,quicklist 未发生变化
int quicklistDelRange(quicklist *quicklist, const long start, const long count);

删除 quicklist 指定范围的项目。

  • 返回 1,表示删除成功
  • 返回 0,表示未删除(count 为负数等非法情况)
quicklist *quicklistDup(quicklist *orig);

复制一个 quicklist。

void quicklistRotate(quicklist *quicklist);

将 quicklist 末尾的一个项目移到列表开头。

int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                       unsigned int *sz, long long *sval,
                       void *(*saver)(unsigned char *data, unsigned int sz));

获取 quicklist 开头(where 为 QUICKLIST_HEAD)或者末尾(where 为 QUICKLIST_TAIL)的项目的内容(列表中的项目不会被删除,不要被函数名误导了), 若项目类型是字节数组类型则会将得到的数据调用 “saver 回调函数” 然后将回调函数的返回值设置给 data。

  • 返回 0 表示列表为空
  • 返回 1 表示成功,如果 data 未设置表示值在 sval)
int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
                 unsigned int *sz, long long *slong);

调用 quicklistPopCustom,若项目类型为字节数组则使用 zmalloc 复制一份副本返回给 data,调用者负责释放这个内存。

unsigned long quicklistCount(const quicklist *ql);

返回 quicklist 的项目总数(直接返回 quicklist 结构中的 count)

int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len);

内部直接调用 ziplistCompare。

size_t quicklistGetLzf(const quicklistNode *node, void **data);

返回 quicklistNode 中的 LZF 数据,返回值为 LZF 数据的长度。

quicklistNode 的分裂与合并

当插入数据到 quicklist 中间某个节点的时候,若果此节点的 ziplist 已经达到限值,则会发生 quicklistNode 的分裂,分裂点在要插入的位置。

假设我们有如下 quicklist,插入位置为 p 指向的位置。

假设我们要插入的 ziplist 已经到达限值,节点需要分裂。

可以看到分裂是以插入点进行分割的,分裂后我们要插入的数据会插入分裂后的前一个节点的末尾

插入后,为了避免由于分裂导致 quicklist 的节点碎片化,会调用 _quicklistMergeNodes(…) 对节点进行合并。

以分裂后的前一个节点作为中心节点(下面简称为 center)尝试进行如下合并: * (center->prev->prev, center->prev) * (center->next, center->next->next) * (center->prev, center) * (center, center->next)

在节点的两两合并前会先检查合并后的节点是否符合 quicklist 的 fill 限制,若符合则进行合并,否则不合并。 在我们的例子中,假设只有 (center->next, center->next->next) 这一步是符合合并条件的。 合并后的 quicklist 如下所示:

总结

  • quicklist 不会对过小的 ziplist 进行压缩,这个值为 MIN_COMPRESS_BYTES,目前是 48
  • quicklist 在对 ziplist 进行压缩后,会对比压缩后的长度和未压缩的长度,若压缩后的长度 - 未压缩的长度 < 8(MIN_COMPRESS_IMPROVE),则使用未压缩的数据
  • quicklist 的头节点和尾节点在任何时候都不会被压缩,因此可以保证将数据插入列表的头或者尾是高效的。
  • 插入一个数据到压缩的节点时,需要先对节点的 ziplist 整个进行解压,插入后再次进行压缩
  • 对中间某个已满的节点进行插入操作会导致此节点(假设此节点为 node)的 ziplist 在插入点进行分裂,此时这个 ziplist 会变成两个 ziplist(新建一个 quicklistNode 保存多出来的 ziplist,假设此节点为 newnode),然后将数据插入 node 指向的 ziplist 的末尾,然后对 node 进行 _quicklistMergeNodes 操作

参考