Redis源码分析之QuickList篇

quicklist 是 Redis 对外暴露的 list 数据结构的内部实现，经常被当作队列或栈使用，我们可以从常用的一些 api 上先思考一下它的结构

最常用的就是 lpush、lpop、rpush、rpop，同时它也支持 lindex 查询某元素在 list 中的索引，linsert 在指定元素旁边插入新元素。

从头、尾节点的 push、pop 来看，这就是双向链表最优秀的设计，提供头尾节点的 O(1) 复杂度操作，但普通的链表每个节点都需要一个前驱和后向指针，空间利用率低；且在链表元素过多时查询缓慢，ziplist 让每一个 entry 都存了上一个 entry 和自己的长度，用来实现双向遍历，节省了一些空间，但缺点是增加、修改元素时可能会带来更大的内存拷贝。

有没有办法进行一个时间和空间的权衡呢？quicklist 就是这样设计的：

A doubly linked list of ziplists 翻译为一个 ziplist 的双向链表。

那么每个 ziplist 存多少数据合适呢？比如我有 20 个元素，可以 4 个 ziplist，每个 ziplist 里 5 个元素；也可以 5 个 ziplist，每个 ziplist 里 4 个元素：

• ziplist 的元素越多，进行操作时候需要内存拷贝的空间就越多，如果整个 quicklist 只有一个 ziplist 节点，那么其实就退化成了一个 ziplist 了；
• ziplist 的元素越少，就需要更多的节点，空间利用率低，更容易产生内存碎片，如果每个 ziplist 只有一个元素，quicklist 就退化成一个普通的双向链表了。

redis 提供了一个这样的配置项：

list-max-ziplist-size // 默认-2

当取正值的时候，表示按照数据项个数来限定每个 quicklist 节点上的 ziplist 长度。比如，当这个参数配置成 5的时候，表示每个 quicklist 节点的 ziplist 最多包含 5 个数据项。

当取负值的时候，表示按照占用字节数来限定每个 quicklist 节点上的 ziplist 长度。这时，它只能取 -1 到 -5 这五个值，每个值含义如下：

• -5: 每个 quicklist 节点上的 ziplist 大小不能超过 64Kb；
• -4: 每个 quicklist 节点上的 ziplist 大小不能超过 32Kb；
• -3: 每个 quicklist 节点上的 ziplist 大小不能超过 16Kb；
• -2: 每个 quicklist 节点上的 ziplist 大小不能超过 8Kb（Redis 默认值）；
• -1: 每个 quicklist 节点上的 ziplist 大小不能超过 4Kb。

quicklist

typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* total count of all entries in all ziplists */
    unsigned int len;           /* number of quicklistNodes */
    int fill : 16;              /* fill factor for individual nodes */
    unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
} quicklist;

和我们认知里的双向链表一样：

• head 头节点指针；
• tail 尾节点指针；
• count 整个 quicklist 中的元素数量；
• len ziplist 的数量；
• fill: 16bit，ziplist大小设置，存放list-max-ziplist-size参数的值；
• compress: 16bit，节点压缩深度设置，存放list-compress-depth参数的值。

list-max-ziplist-size 的值在最上面已经解释过含义了，list-compress-depth是出于这样的考虑：

当列表很长的时候，最容易被访问的很可能是两端的数据，中间的数据被访问的频率比较低，所以 redis 提供了该选项选择压缩的中间数据，取值含义如下：

• 0: 是个特殊值，表示都不压缩。这是 Redis 的默认值；
• 1: 表示 quicklist 两端各有1个节点不压缩，中间的节点压缩；
• 2: 表示 quicklist 两端各有2个节点不压缩，中间的节点压缩；
• 3: 表示 quicklist 两端各有3个节点不压缩，中间的节点压缩。

注意，两个端点（头、尾）是不会压缩的。

quicklistNode

quicklistNode 是每一个 quicklist 节点的数据结构

typedef struct quicklistNode {
    struct quicklistNode *prev;     // 前一个节点
    struct quicklistNode *next;     // 下一个节点
    unsigned char *zl;              // 数据指针。如果当前节点的数据没有压缩，那么它指向一个 ziplist 结构
                                    // 否则，它指向一个quicklistLZF 结构。
    unsigned int sz;                // zl 指向的 ziplist 的总大小（包括数据项和 zlbytes 等）
    unsigned int count: 16;         // ziplist 里面包含的数据项个数 2 字节
    unsigned int encoding: 2;       // 目前只有 1/2 取值 分别代表没压缩和 LZF 压缩
    unsigned int container: 2;      // 目前只有 2 取值，本意是取 1 代表节点直接存数据，取 2 代表使用 ziplist 存数据
    unsigned int recompress: 1;     // 当我们使用 index 等命令查看到了已经被压缩的数据，需要暂时解压，再设置标记为合适的时候压缩回去
    unsigned int attempted_compress: 1; // 自动化测试使用
    unsigned int extra: 10;         // 拓展字段，暂时没用上
} quicklistNode;

// 一个被压缩过的 ziplist
typedef struct quicklistLZF {
    unsigned int sz;    // 压缩后的 ziplist 大小
    char compressed[];  // 柔性数组（flexible array member），存放压缩后的 ziplist 字节数组
} quicklistLZF;

以下是某一时刻一个 quicklist 的示意图：

list-max-ziplist-size 3
list-compress-depth 2

• 两端各有 2 个橙黄色节点没有被压缩，对应 list-compress-depth 为 2；
• 第二个节点和倒数第二个节点，都是元素数量为 3 的 ziplist，对应 list-max-ziplist-size。

代码分析

1. 创建函数

quicklist *quicklistNew(int fill, int compress) {
    quicklist *quicklist = quicklistCreate();
    // 将两个 option 设置到 quicklist 中
    quicklistSetOptions(quicklist, fill, compress);
    return quicklist;
}

quicklist *quicklistCreate(void) {
    struct quicklist *quicklist;
    quicklist = zmalloc(sizeof(*quicklist));
    // 初始化时 头尾节点都为 null
    quicklist->head = quicklist->tail = NULL;
    quicklist->len = 0;
    quicklist->count = 0;
    quicklist->compress = 0;		// 压缩深度
    quicklist->fill = -2;			// 单个 ziplist 最多 8 字节
    return quicklist;
}

创建函数其实很简单，只是按照默认值初始化，并把配置项的值放入到 quicklist 对应的 compress、fill 字段上

其他操作复杂的原因在于，比如插入操作过程中要判断是否达到了配置的值，达到的话需要创建新的 node 节点（包括 ziplist ）；删除操作如果 ziplist 为空了，那么对应的头部或尾部节点也要删除，还可能涉及到里面节点的解压缩问题……

2. 在指定节点插入（lpush、rpush 最终都会调用到这个函数）

// 在 quicklist 的指定位置插入元素，根据 after 判断在前还是在后
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;
        }
        // 如果插入在之后且 old_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;
        }
        // 如果插入在之前且 old_node 是头节点,更新头节点
        if (quicklist->head == old_node)
            quicklist->head = new_node;
    }
    
    // 为空时初始化头尾指针
    if (quicklist->len == 0) {
        quicklist->head = quicklist->tail = new_node;
    }

    if (old_node)
        quicklistCompress(quicklist, old_node);

    quicklist->len++;
}

3. push 操作

在调用这个函数前，会根据操作是 lpush 还是 rpush 确定 where 是头还是尾

void quicklistPush(quicklist *quicklist, void *value, const size_t sz, int where) {
    if (where == QUICKLIST_HEAD) {
        quicklistPushHead(quicklist, value, sz);
    } else if (where == QUICKLIST_TAIL) {
        quicklistPushTail(quicklist, value, sz);
    }
}

头插

int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_head = quicklist->head;
    // 可以直接插入
    if (likely(
            _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
        // 当前节点插入到头节点对应的 ziplist 的头部
        quicklist->head->zl =
                ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
        // 更新新节点指向的 ziplist 的总大小
        quicklistNodeUpdateSz(quicklist->head);
    } else {
        // 创建新的 node 节点
        quicklistNode *node = quicklistCreateNode();
        // 创建对应的 ziplist，并把新元素插入到头
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        // 更新新节点指向的 ziplist 的总大小
        quicklistNodeUpdateSz(node);
        // 把当前节点插入到原头节点的前面
        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
    }
    // 总数据项 + 1
    quicklist->count++;
    // 头节点的里的元素数量 + 1
    quicklist->head->count++;
    return (orig_head != quicklist->head);
}

REDIS_STATIC void _quicklistInsertNodeBefore(quicklist *quicklist,
                                             quicklistNode *old_node,
                                             quicklistNode *new_node) {
    __quicklistInsertNode(quicklist, old_node, new_node, 0);
}

尾插

int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_tail = quicklist->tail;
    // 是否可以直接插入
    if (likely(
            _quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) {
        quicklist->tail->zl =
                ziplistPush(quicklist->tail->zl, value, sz, ZIPLIST_TAIL);
        quicklistNodeUpdateSz(quicklist->tail);
    } else {
        quicklistNode *node = quicklistCreateNode();
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL);

        quicklistNodeUpdateSz(node);
        // 把当前节点插入到原尾节点的后面
        _quicklistInsertNodeAfter(quicklist, quicklist->tail, node);
    }
    // 更新当前 quicklist 中的元素总数和尾节点对应 ziplist 中的元素数量
    quicklist->count++;
    quicklist->tail->count++;
    return (orig_tail != quicklist->tail);
}

REDIS_STATIC void _quicklistInsertNodeAfter(quicklist *quicklist,
                                            quicklistNode *old_node,
                                            quicklistNode *new_node) {
    __quicklistInsertNode(quicklist, old_node, new_node, 1);
}

lpush 和 rpush 的逻辑几乎是一样的，而对其进行压缩，是在 __quicklistInsertNode 函数的 quicklistCompress 函数中……