简单来说,Redis 7.0 以后的 quicklist 其实相当于是 linkedlist + listpack 的变形组合。listpack 的优势在于可以高效利用内存空间,缺陷是不适合在单个 listpack 中存放太多数据,为了优化这一问题,Redis 通过链表的形式将大量不连续的 listpack 串联起来,这种结构就是 quicklist 的基本框架。

quicklist

结构

quicklistNode 相当于链表中的数据节点,基本结构定义在 https://github.com/redis/redis/blob/7.2.4/src/quicklist.h#L46:

typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *entry;
    // 指向的 listpack 占用的字节数
    size_t sz;
    // 指向的 listpack 中的 entry 数量
    unsigned int count : 16;
    // 是否压缩,1表示不压缩,2表示通过LZF压缩
    unsigned int encoding : 2;
    unsigned int container : 2;  /* PLAIN==1 or PACKED==2 */
    unsigned int recompress : 1; /* was this node previous compressed? */
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    unsigned int dont_compress : 1; /* prevent compression of entry that will be used later */
    unsigned int extra : 9; /* more bits to steal for future usage */
} quicklistNode;

快速列表的节点构成继承了链表节点的一些特性,区别的地方在于 quicklistNode 支持对指向的数据以 Redis 提供的 LZF 算法进行压缩,如果节点指向 quicklistNode->entry 被压缩过,那么 entry 属性会指向 quicklistLZF 结构体:

typedef struct quicklistLZF {
    // 压缩后的字节数
    size_t sz;
    char compressed[];
} quicklistLZF;

没被压缩的节点,quicklistNode->entry 会指向一个正常 quicklistEntry 结构体:

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

其中 qucklistEntry->zi 指向的是存储数据的 listpack 结构。

多个 quicklistNode 统一由 quicklist 结构体来管理:

typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    // 全部节点持有的listpack中的数据节点总数
    unsigned long count;
    // quicklistNode数量
    unsigned long len;
    signed int fill : QL_FILL_BITS;
    // 节点是否全部压缩
    unsigned int compress : QL_COMP_BITS;
    unsigned int bookmark_count: QL_BM_BITS;
    quicklistBookmark bookmarks[];
} quicklist;

quicklist->fill 字段用来标识每个 quicklistNode 的最大容量(默认为-2),fill 字段为负数(-N)时候,表示每个 quicklistNode 持有的 listpack 所占字节数不超过 2^(N+1) KB 个字节,fill 值为任意正数时候表示 listpack 最多允许包含的 entry 数量。

quicklist 结构体还附带管理了一个 quicklistBookmark 结构体的列表:

typedef struct quicklistBookmark {
    quicklistNode *node;
    char *name;
} quicklistBookmark;

quicklistBookmark 通常在 quicklist 中持有的节点数量很多的时候启用,当列表中有成千上万个节点的时候,可以通过 quicklistBookmark 快速定位到特定的 quicklistNode,这样就不用去花费时间来遍历列表获取特定的节点,而通常相对了大量节点而言,少数 quicklistBookmark 占用的内存空间是可以忽略不计的。

操作

quicklist 的创建基本和链表初始化差不多,分配内存后初始化头尾指针即可:

quicklist *quicklistCreate(void) {
    struct quicklist *quicklist;

    quicklist = zmalloc(sizeof(*quicklist));
    quicklist->head = quicklist->tail = NULL;
    quicklist->len = 0;
    quicklist->count = 0;
    quicklist->compress = 0;
    quicklist->fill = -2;
    quicklist->bookmark_count = 0;
    return quicklist;
}

插入节点的操作也是基本的链表插入行为,还是比较容易理解的:

REDIS_STATIC void __quicklistInsertNode(quicklist *quicklist,
                                        quicklistNode *old_node,
                                        quicklistNode *new_node, int after) {
    if (after) {
        new_node->prev = old_node;
        if (old_node) {
            new_node->next = old_node->next;
            if (old_node->next)
                old_node->next->prev = new_node;
            old_node->next = new_node;
        }
        if (quicklist->tail == old_node)
            quicklist->tail = new_node;
    } else {
        new_node->next = old_node;
        if (old_node) {
            new_node->prev = old_node->prev;
            if (old_node->prev)
                old_node->prev->next = new_node;
            old_node->prev = new_node;
        }
        if (quicklist->head == old_node)
            quicklist->head = new_node;
    }
    /* If this insert creates the only element so far, initialize head/tail. */
    if (quicklist->len == 0) {
        quicklist->head = quicklist->tail = new_node;
    }

    /* Update len first, so in __quicklistCompress we know exactly len */
    quicklist->len++;

    if (old_node)
        quicklistCompress(quicklist, old_node);

    quicklistCompress(quicklist, new_node);
}

quicklistNode 节点的压缩行为主要通过 lzf_compress() 完成:

REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
#ifdef REDIS_TEST
    node->attempted_compress = 1;
#endif
    if (node->dont_compress) return 0;

    /* validate that the node is neither
     * tail nor head (it has prev and next)*/
    assert(node->prev && node->next);

    node->recompress = 0;
    /* Don't bother compressing small values */
    if (node->sz < MIN_COMPRESS_BYTES)
        return 0;

    quicklistLZF *lzf = zmalloc(sizeof(*lzf) + node->sz);

    /* Cancel if compression fails or doesn't compress small enough */
    if (((lzf->sz = lzf_compress(node->entry, node->sz, lzf->compressed,
                                 node->sz)) == 0) ||
        lzf->sz + MIN_COMPRESS_IMPROVE >= node->sz) {
        /* lzf_compress aborts/rejects compression if value not compressible. */
        zfree(lzf);
        return 0;
    }
    lzf = zrealloc(lzf, sizeof(*lzf) + lzf->sz);
    zfree(node->entry);
    node->entry = (unsigned char *)lzf;
    node->encoding = QUICKLIST_NODE_ENCODING_LZF;
    return 1;
}