Redis源码学习(14)-Redis中快速链表的实现（下）

快速链表的插入

因为对于调用者来说，链表节点这一层对于他其实是透明，调用者主要进行的是数据节点的操作，而并不是太关心一个数据节点是以何种策略插入到哪个链表节点中，或者是从哪个链表节点中删除的。因此我们可以发现，在src/quicklist.h头文件中定义的通常是对于数据节点的操作函数，而对于链表节点的操作，则通常是定义在src/quicklist.c源文件中的静态函数。

插入条件的判断

因为快速链表定义了装载因子quicklist.fill
这个字段，用来对每个链表节点底层压缩链表的内存大小或者节点长度进行限制，因此在向某个链表节点中插入数据节点的时候，需要预先进行判断，新的数据节点是否可以进行插入。

如果快速链表的装载因子设置对于压缩链表内存大小的限制，那么Redis会通过下面这个函数来检查插入后新的内存大小是否满足装载因子的限制：

REDIS_STATIC int _quicklistNodeSizeMeetsOptimizationRequirement(const size_t sz, const int fill);

这个函数会首先判断fill
参数是否为负数，因为快速链表只有在装载因子为负数的情况下才会对压缩链表内存大小进行限制。如果fill
参数合法，那么会从optimization_level
数组中查询对应内存上限，然后判断新的内存大小是否符合要求。

那么对于插入条件的通用判断，Redis则是通过下面这个静态函数来实现的：

REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node, const int fill, const size_t sz);

给定一个待插入的链表节点，以及这个快速链表的装载因子和插入数据的内存大小，来判断插入操作是否可以执行，返回0，表示无法插入到给定的链表节点；返回1，则表示可以进行插入。整个判断过程分为三步：

1. 如果装载因子限制了压缩链表的内存的大小，那么通过_quicklistNodeSizeMeetsOptimizationRequirement
   函数检查内存大小是否符合要求。

2. 如果装载因子限制的是压缩链表的节点长度，那么对于这种情况，Redis也会定义其内存大小的限制#define SIZE_SAFETY_LIMIT 8192
   ，也就是这种情况下，整个压缩链表的内存大小不能超过8K字节，否则就需要一个单独的链表节点来存储数据。

3. 最后会判断node
   节点中数据节点的个数是否满足装载因子对于节点个数的限制。

在操作快速链表的过程中，为了提高内存使用率，会尝试对两个相邻的链表节点进行合并，这种合并操作也应该符合快速链表中装载因子对链表节点的限制，因此Redis还定义个下面的函数，用于判断两个链表节点是否可以合并为1个：

REDIS_STATIC int _quicklistNodeAllowMerge(const quicklistNode *a, const quicklistNode *b, const int fill);

这个函数的判断流程与_quicklistNodeAllowInsert
类似，也一样会从三个方面进行判断。

链表节点的插入

Redis定义了一个底层的链表节点的插入操作函数：

REDIS_STATIC void __quicklistInsertNode(quicklist *quicklist, quicklistNode *old_node, quicklistNode *new_node, int after);

这个函数执行的是一个经典的双端链表插入操作，

1. 如果after
   为1，那么new_node
   会被插入到old_node
   节点的后面。

2. 如果after
   为0，那么new_node
   会被插入到old_node
   节点的前面。

此外需要注意的一点，待插入的链表节点new_node
，应该是一个未被压缩的节点，如果被插入到快速链表的头部或者尾部的时候，便没有必要对该节点进行压缩，而插入到其他位置的时候，需要调用者在__quicklistInsertNode
之后，手动调用压缩函数，对节点进行压缩。

通过对__quicklistInsertNode
进行包装，Redis还定义了两个静态函数，用于实现先向某个已存在的链表节点前后进行插入新的链表节点的操作：

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

链表节点的拆分与合并

Redis会使用静态函数_quicklistZiplistMerge
来作为合并链表节点的操作基础，该函数会给定两个链表节点a
和b
，然后尝试将合并这两个链表节点的底层压缩链表。

REDIS_STATIC quicklistNode *_quicklistZiplistMerge(quicklist *quicklist, quicklistNode *a, quicklistNode *b){    ...	quicklistDecompressNode(a);    quicklistDecompressNode(b);    if ((ziplistMerge(&a->zl, &b->zl))) {    	quicklistNode *keep = NULL, *nokeep = NULL;        ...        __quicklistDelNode(quicklist, nokeep);        quicklistCompress(quicklist, keep);        return keep;    }    ...}

这个合并操作的流程为：

1. 调用quicklistDecompressNode
   将两个链表节点进行解压。

2. 通过ziplistMerge
   合并两个链表节点的底层压缩链表。

3. 调用__quicklistDelNode
   将被合掉的链表节点从快速链表之中删除。

4. 最后将保留的节点通过quicklistCompress
   进行重新压缩。

不过我们发现，前面我们介绍了_quicklistNodeAllowMerge
这个函数用于检查两个链表节点的合并是否可以满足装载因子的限制。但是_quicklistZiplistMerge
这个函数看起来只是执行了链表节点的合并操作，而并没有对合并的合法性进行校验。因此Redis在_quicklistZiplistMerge
的基础上，结合了合并合法性的校验函数_quicklistNodeAllowMerge
实现了另外一个更加高级的链表节点合并操作的接口：

REDIS_STATIC void _quicklistMergeNodes(quicklist *quicklist, quicklistNode *center);

这个函数会尝试对center
节点以及其两侧各两个链表节点一共五个节点尝试合并成一个链表节点，其合并的顺序为：

1. center->prev->prev
   与center->prev
   

2. center->next
   与center->next->next
   

3. center->prev
   与center
   

4. center
   与center->next
   

上述的四步合并都是通过先调用_quicklistNodeAllowMerge
检查合法性，然后调用_quicklistZiplistMerge
将两个节点进行合并。

在实现了链表节点的合并操作之后，Redis也为我们提供了一个链表节点的拆分功能，这个特性在执行数据节点的插入时，是非常有用的。想象一个场景：

1. 当我们期望在一个给定的索引位置进行数据节点插入，

2. 而这个给定的索引恰恰对于链表节点底层压缩链表的中间，

3. 同时这次插入的结果会导致对应的链表节点将不满足快速链表对于装载因子的限制

对于这种情况，便需要对对应的链表节点执行拆分操作，将一个较大的节点拆分成两个更小的节点，以满足数据节点的插入。因此Redis定义了一个静态函数用于实现链表节点的拆分：

REDIS_STATIC quicklistNode *_quicklistSplitNode(quicklistNode *node, int offset, int after);

这个函数会根据offset
以及after
来确定链表节点中低层压缩链表的拆分位置。如果拆分成功，那么这个函数会返回一个新的链表节点，其中保存了被拆分出来的数据节点，而剩下的数据节点则会被保留在node
节点之中。对于链表节点的拆分策略，如下所示：

1. 如果after
   为1，原node
   节点中保留了[0, offset]
   这些数据节点，返回的新节点存储了[offset+1, END]
   这些数据节点。

2. 如果after
   为0，原node
   节点中保留了[offset+1, END]
   这些数据节点，返回的新节点存储了[0, offset]
   这些数据节点。

数据节点的Push插入

Redis提供了两个接口函数，用于实现向快速链表的头部和尾部插入数据节点，这两个函数可以供调用者在外部进行调用：

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

上述两个函数实现方式基本上是一致的：

1. 通过调用前面提到的_quicklistNodeAllowInsert
   静态函数来判断，新数据是否可以插入到快速链表的头节点或者尾节点。

2. 如果可以插入，那么调用压缩链表的插入接口ziplistPush
   ，将数据插入到头节点或者尾节点的压缩链表中，值得注意的是，因为不论当前的快速链表的压缩深度是多少，其头节点和尾节点一定是未压缩的状态，因此执行压缩链表插入的时候，不需要对链表节点进行解压缩操作。

3. 如果头尾节点因为达到了装载因子的限制而无法插入新元素的话，或者当前的快读链表为空的时候，那么Redis会按照如下的步骤执行插入流程：

1. 调用quicklistCreateNode
   创建一个新的链表节点。

2. 调用压缩链表的插入接口ziplistPush
   ，将数据插入到新创建的链表节点的底层压缩链表之中。

3. 调用_quicklistInsertNodeBefore
   或者_quicklistInsertNodeAfter
   函数，将新创建的链表节点插入到快速链表之中。

上述两个函数的插入过程一定不会失败，当新数据被插入到一个已存在的链表节点之中，那么该函数会返回0；如果新创建一个链表节点，那么函数返回1。

同时Redis还将上述的两个函数包装成一个通用的函数接口：

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

通过将参数where
设置成QUICKLIST_HEAD
或者QUICKLIST_TAIL
来实现向快速链表的头部或者尾部进行数据节点的插入。

数据节点的随机插入

前面介绍的数据节点在快速链表两端的Push插入操作是快速链表中较为常用的操作，因为是在链表的两端执行插入，因此不需要考虑链表节点的合并与拆分的细节，通过前面的描述，我们也发现当在链表两端执行插入操作时，如果两端的链表节点不能满足装载因子的限制，那么Redis会直接为新数据创建一个新的链表节点，将之插入到快速链表中。但是对于将数据节点插入到指定位置索引的需求依然存在，因此Redis定义了一个向快速链表中的给定位置索引插入一个数据节点的函数：

REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, void *value, const size_t sz, int after);

这个函数向快速链表quicklist
中的数据节点entry
的前面或者后面根据value
以及sz
插入一个新的数据节点。在插入过程，Redis会执行如下的逻辑：

1. 定义了五个标记变量：

   标记变量	变量含义
   full	entry 对应的链表节点是否已经达到装载因子的限制
   at_tail	新数据的插入位置是否位于entry 对应链表节点底层压缩链表的尾部
   at_head	新数据的插入位置是否位于entry 对应链表节点底层压缩链表的头部
   full_next	entry 对应的链表节点的前序节点是否已满
   full_prev	entry 对应的链表节点的后序节点是否已满

2. 通过判断quicklistEntry.offset
   以及调用_quicklistNodeAllowInsert
   函数来为上述的5个标记赋值。

3. 根据前面定义的5个标记以及after
   参数，将插入过程划分为6种情况，根据情况的不同，执行不同的插入逻辑：

• !full && after
  ，插入到给定数据节点之后，对应的链表节点未满，那么直接调用ziplistPush
  或者ziplistInsert
  将新数据插入到当前链表节点的底层压缩链表中。

• !full && !after
  ，插入到给定数据节点之前，对应的链表节点未满，那么处理方式与上面一种情况类似，调用ziplistInsert
  将新数据插入到当前链表节点的底层压缩链表中。

• full && at_tail && node->next && !full_next && after
  ，插入对应链表节点的尾部，同时该节点已满，而其后序的链表节点未满，那么将新数据插入到其后续链表节点压缩链表的头部。

• full && at_head && node->prev && !full_prev && !after
  ，插入对应链表节点的头部，同时该节点已满，而其前序的链表节点未满，那么将数据插入到其前序链表节点压缩链表的尾部。

• full && ((at_tail && node->next && full_next && after) || (at_head && node->prev && full_prev && !after))
  ，在对应链表节点的头部或者尾部插入新的数据，但是该节点已满无法插入新数据，同时对应的前序或者后续节点也无法插入新数据，对于这种所有关联链表节点都无法进行插入的情况，只能为这个新的数据创建一个新的链表节点，调用__quicklistInsertNode
  将其插入到快速链表中。

• 最后一种情况，插入的位置正好是对应数据节点的中间位置，而该节点已满，那么将调用_quicklistSplitNode
  函数，通过quicklistEntry.offset
  ，以及after
  作为参数，将链表节点进行拆分，对拆分操作返回的新链表节点应用ziplistPush
  将数据插入其中，并将新链表节点插入快速链表，最后调用_quicklistMergeNodes
  尝试对链表节点进行合并。

基于这个_quicklistInsert
函数，Redis定义了两个接口函数：

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

这两个函数本质上是对_quicklistInsert
简单的封装，分别用于将数据插入到一个给定的数据节点之前，或者插入到给定的数据节点之后。

数据节点的批量插入

Redis为快速链表提供了三个函数，可以实现从一个压缩链表构造 一个快速链表，或者将压缩链表作为输入数据的集合，实现批量的数据节点插入。

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

这个函数通过给定的压缩链表参数zl
创建一个快速节点，然后调用_quicklistInsertNodeAfter
将这个新建的链表节点插入到快速链表的节点。

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

这个函数也是将一个压缩链表zl
中的数据插入到快速链表中，但是与quicklistAppendZiplist
不同的是，quicklistAppendValuesFromZiplist
不是将压缩链表作为一个整体插入到快速链表中的，而是遍历读取压缩链表中中的每一个节点，将其作为数据节点的输入，通过调用quicklistPushTail
将数据节点插入到快速链表的尾部。

quicklist *quicklistCreateFromZiplist(int fill, int compress, unsigned char *zl){    return quicklistAppendValuesFromZiplist(quicklistNew(fill, compress), zl);}

上述这个函数是通过压缩链表来创建一个快速链表，具体是通过函数quicklistNew
创建一个快速链表，然后调用quicklistAppendValuesFromZiplist
函数将压缩链表中的数据批量的插入到新创建的快速链表中。

快速链表的删除

链表节点的删除

Redis为快速链表中对于链表节点的删除给一个基础的操作函数：

REDIS_STATIC void __quicklistDelNode(quicklist *quicklist, quicklistNode *node){    ...    __quicklistCompress(quicklist, NULL);    ...}

此处对链表节点的删除，与链表节点的插入类似，应用经典的双端链表的删除方式。但是执行删除操作时，如果这个节点刚好处于快速链表的压缩深度之内，那么我们需要调用__quicklistCompress(quicklist, NULL);
来对某一个处于压缩状态，但是因为删除了node
节点而落入压缩深度范围内的节点进行解压。

指定数据节点的删除

对于数据节点的删除操作，Redis同样定义了一个基础的操作函数：

REDIS_STATIC int quicklistDelIndex(quicklist *quicklist, quicklistNode *node, unsigned char **p);

这个函数用于从给定的快速链表quicklist
中，删除链表节点node
里由p
所指向的数据节点，这里有两点需要注意：

1. 因为这是一个内部调用的静态函数，因此需要在调用的外部，自行对node
   节点进行解压。

2. 如果node
   节点底层的压缩链表在调用ziplistDelete
   删除p
   之后，变成空的压缩链表，那么需要调用__quicklistDelNode
   将这个链表节点从链表中删去。

基于上面的静态函数，Redis定义了一个用户节点函数，用于从快速链表中删除一个由quicklistEntry
描述的数据节点：

void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry) {    ...    int deleted_node = quicklistDelIndex((quicklist *)entry->quicklist, entry->node, &entry->zi);    iter->zi = NULL;    ...}

对于链表而言，除了在链表两端执行的Pop删除之外，其他对给定数据节点的删除大概率都是在遍历的过程中进行，因此quicklistDelEntry
这个函数除了接受一个quicklistEntry
变量作为参数之后，选择使用一个quicklistIter
迭代器作为参数，而不是传入一个快速链表指针作为参数的。其常用的遍历删除范式为：

while (quicklistNext(iter, &entry)){    if (quicklistCompare(entry.zi, data, size))    {        quicklistDelEntry(iter, &entry);    }    i++}

那么Redis是如何保证迭代器在一边迭代一边删除的时候，不会失效的呢？我们可以在回头看一下quicklistNext
函数的实现细节：

int quicklistNext(quicklistIter *iter, quicklistEntry *entry){    ...    if (!iter->zi)    {        quicklistDecompressNodeForUse(iter->current);        iter->zi = ziplistIndex(iter->current->zl, iter->offset);    }}

我们可以看到，Redis在执行quicklistDelEntry
对数据节点进行删除操作后，会将quicklistIter.zi
字段设置为NULL
，

• 如果链表节点没有被删除的话，那么quicklistIter.offset
  不会变化，在下一轮迭代quicklistNext
  中，如果发现quicklistIter.zi
  为NULL
  的时候，会调用ziplistIndex
  根据quicklistIter.offset
  来将迭代器移动到下一个位置。

• 如果链表节点被删除的话，那么quicklistDelEntry
  会更新迭代器quicklistIter.current
  和quicklistIter.offset
  这两个字段，用于将迭代器移动到下一个链表节点，最后在quicklistNext
  调用中，会将迭代器移动到下一个位置。

数据节点的Pop删除

Redis除了提供若干个在指定位置执行删除数据节点的函数之外，还提供了两个更加定制化的，用于从链表两端执行删除操作的函数。

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

这个函数是快速链表的Pop操作的基础，会从quicklist
中弹出元素，将弹出的数据节点的数据通过函数指针saver
存储在data
之中，如果数据节点中的数据是一个整形数的话，那么会将数据存储在sval
之中。如果弹出成功，函数返回1，否则的话，函数返回0。

除了上述这个基础Pop函数之外，Redis还基于这个函数，定义了一个默认的Pop函数：

int quicklistPop(quicklist *quicklist, int where, unsigned char **data, unsigned int *sz, long long *slong);

这个函数本质上是对quicklistPopCustom
的简单封装，但是这个函数只在测试代码中有调用，而在外部则没有对应调用的地方。

数据节点的批量删除

int quicklistDelRange(quicklist *quicklist, const long start, const long count);

这个函数会从快速链表中删除从start
开始连续count
个数的数据节点，需要注意的是，被删除的多个数据节点有可能会跨越多个链表节点。

快速链表的其他操作

void quicklistRotate(quicklist *quicklist);

quicklistRotate
用于实现对快速链表的一次反转操作，也就是将快速链表尾部的数据节点弹出，然后插入到快速链表的头部。

quicklist *quicklistDup(quicklist *orig);

这个函数则用于拷贝一个给定的快速链表 ，并返回新建的快速链表的指针。

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

这个函数t通过调用ziplistCompare
来判断一个底层数据节点的数据是否等于p2
和p2_len
所表示的数据。