redis数据结构---quicklist
程序员文章站
2022-05-19 16:00:02
...
quicklist是redis在3.2版本中加入的新的数据结构，用作redis对外提供的五种数据类型---list的底层实现。本文关于redis的讲解都是基于redis-4.0.1版本的源码。
黄建宏老师的《redis设计与实现》第二版是针对redis 3.0源码讲解的。书中讲redis的list数据结构的底层实现有两种，压缩列表（ziplist)和双向链表（adlist，以数据结构结构所在的.c文件命名以与list进行区分）。这里的adlist就是我们通常见到的双向链表结构；关于ziplist的结构请参考文章--ziplist。按照书中的解释，当列表中的元素的数目比较少（小于512个）并且元素的长度比较小（小于64字节）的时候，redis使用ziplist作为list的底层实现；当其中的任何一个条件不满足的时候，redis将ziplist编码的list转换成adlist编码。redis3.0对list采用以上两种实现是基于空间和时间效率的考虑。
1.当元素的长度小于64字节的时候（假设都是以字符串的形式进行存储，没有转换成整数），使用ziplist存储num个entry需要的字节长度为以下长度之和：total = zlbytes+zltail+zllen+zlend+sizeof(entry)*num = 4 + 4 +2 + 1 + sizeof(entry) * num = 11 + sizeof(entry) * num。如果长度小于64字节，那么一个entry的prev_entry_length最多需要5个字节，encoding域需要1个字节，再加上最多63字节数据，那么total <= (5 + 1 + 63) * num + 11 = 69 * num + 11。如果使用双端链表实现，那么在64位机器上，需要的长度为total = sizeof(struct list) + （sizeof(struct listNode) + 63）* num = 48 + （8 + 8 + 8 + 63 ）* num = 48 + 87 * num。所以，相比之下，当元素的长度比较小的时候，使用ziplist存储相同的数据，可以节省存储空间；
2.但是，为什么要限定压缩列表存储的元素个数上限是512呢？ziplist在内存中是占据一整块的连续内存空间，当元素的个数过多的时候，list占据的连续内存空间较大。即使是存储的是0-12之间的整数(每个entry需要占据2个字节），512个元素需要大约1M的内存空间（当然这是最少的情况，按照上面的分析，存储512个63字节长的字符串的时候可能需要35M左右的连续内存空间），那么随着程序的运行在内存中寻找这么大的连续内存空间变得比较困难。即使能够找到这么大的空间，那么每次修改list中元素的时候都需要重新分配这么大的内存空间，而重新分配内存空间属于比较耗时的操作，严重降低了redis的效率。因此当数据量比较大的时候采用adlist作为list的底层实现，这样每次插入或者删除的时候只需要为插入的元素(listNode)单独分配一小块的内存空间，并将listNode衔接到现存的list中就可以了。但是使用adlist，每次分配的内存空间比较小，如果这种操作比较频繁，那么容易产生内存碎片；
综合考虑redis的空间存储效率和时间效率，redis在3.2版本的时候引入了quicklist作为list的底层实现，quicklist同时使用adlist和ziplist两种数据结构作为list的底层实现，用于缓解单独使用ziplist或者adlist时的缺陷，quicklist在内存中的布局如图1所示。
图 1 quicklist在内存中的布局
quicklist的基本思想是将一个比较大的ziplist拆分成较小的ziplist，每个ziplist存储较少个数或者是较小连续内存的数据。既然是将一个很大的ziplist拆分成若干个较小的ziplist，每个ziplist应该保持多大才比较合适？
·如果ziplist占用的连续内存过大或者是ziplist中entry的数目过多，那么quicklist退化成单纯使用ziplist的情况。一个极端的情况就是，整个list使用单个quicklistNode指向单个ziplist。
·如果ziplist中占用的连续内存过小或者是ziplist中entry的数目过少，那么quicklist退化成单纯使用adlist的情况。一个极端的情况就是，每个ziplist中只保存一个占用空间很小的entry(保存0-12之间的整数)。
如果不能为每个quicklistNode指向的ziplist确定合适的大小，那么并不能充分挥发quicklist数据结构带来的优势。针对每个ziplist的大小，redis.conf文件中有两个配置选项是与之相关的，分别是list-max-ziplist-size和list-compress-depth。
list-max-ziplist-size指定每个ziplist占据的内存字节长度的上限（负整数）或者是每个ziplist中entry的个数（正整数），二者选其一。在redis.conf中关于指定每个ziplist占用的内存上限有以下几种选项。
· -5: 每个ziplist占用的内存不超过64kb;
· -4: 每个ziplist占用的内存不超过32kb;
· -3: 每个ziplist占用的内存不超过16kb;
· -2: 每个ziplist占用的内存不超过8kb;
· -1: 每个ziplist占用的内存不超过4kb;
其中，如果要使用以上几种选项的话，redis.conf文件中建议使用-1或者-2作为最佳选择（系统页的大小为4kb或者是8kb)，但是对于16kb、32kb或者64kb的大小，redis.conf不建议使用，应该是基于CPU效率的考虑。在quicklist.c文件中存在以下宏定义判断一个ziplist中是否还继续允许插入新的entry。
#define SIZE_SAFETY_LIMIT 8192
#define sizeMeetsSafetyLimit(sz) ((sz) <= SIZE_SAFETY_LIMIT)如果redis.conf中配置的list-max-ziplist-size的值为正整数，该数值指定了一个ziplist中允许存在的entry的最大数目。例如，redis.conf文件中默认指定的list-max-ziplist-size的值为3，在每个ziplist中允许存在的entry的最大数目为3个，且3个entry的长度之和不能超过8kb，需要满足sizeMeetsSafetyLimit函数的要求。
为了节省内存空间，quicklist的实现中允许对每个ziplist进行压缩，但是ziplist是否能够压缩，需要看是否能够满足压缩的条件，这个在介绍__quicklistCompressNode的时候解释。现在先解释redis.conf文件中关于quicklistNode的另一个配置参数---list-compress-depth。list-compress-depth指定在每个list中的两端没有被压缩的节点的个数。
·list-compress-depth = 1, quicklist的两端各有1个节点不被压缩；
·list-compress-depth = 2, quicklist的两端各有2个节点不被压缩;
·list-compress-depth = 3, quicklist的两端各有3个节点不被压缩;
·list-compress-depth = N, 依此类推;
作为一种链表结构，list经常执行的操作是在链表的头部或者尾部执行插入或者删除操作，因此list的头部节点和尾部节点总是不被压缩的。list-compress-depth = 0，代表整个quicklist都不被压缩。
quicklist主要由四种数据结构实现:quicklist、quicklistNode、quicklistLZF以及ziplist。下面分别介绍一下前三种数据结构。
typedef struct quicklist {
/* 整个链表的头部节点 */
quicklistNode *head;
/* 整个链表的尾部节点 */
quicklistNode *tail;
/* 整个链表中的所有的entry的数量 */
unsigned long count;
/* quicklistNode的数量 */
unsigned int len;
/* 用户存放配置文件中的list-max-ziplist-size的值 */
int fill : 16;
/* 压缩节点的深度，表示表头或者表尾有几个节点不被压缩;
* 存放list-compress-depth参数的值
*/
unsigned int compress : 16;
} quicklist;
typedef struct quicklistNode {
/* 指向前面的quicklistNode */
struct quicklistNode *prev;
/* 指向后面的quicklistNode */
struct quicklistNode *next;
/* 如果对数据进行压缩，zl指向ziplist；
* 如果不对数据进行压缩，zl指向quicklistLZF
*/
unsigned char *zl;
/* zl指向的ziplist的总大小(包括zlbytes,zltail,zllen,zlend和各个entry的长度之和);
* 如果ziplist被压缩了，那么这个sz值仍然是表示压缩前的ziplist的大小;
* ziplist没有被压缩时，sz表示的就是ziplist占据的字节数
*/
unsigned int sz;
/* ziplist内部entry的个数 */
unsigned int count : 16;
/* 是否可以对ziplist的压缩;
* encoding = QUICKLIST_NODE_ENCODING_LZF，表示可以压缩;
* encoding = QUICKLIST_NODE_ENCODING_RAW, 表示不压缩;
*/
unsigned int encoding : 2;
/* quicklistNode是否作为一个容器使用;
* 如果quicklistNode不作为容器使用，container = QUICKLIST_NODE_CONTAINER_NONE;
* 如果quicklistNode作为容器使用, container = QUICKLIST_NODE_CONTAINER_ZIPLIST;
* 设计quicklist的目的就是为了避免单独使用adlist和ziplist，所以quicklistNode一般
* 用作容器，指向一个包含少量entry的ziplist或者是quicklistLZF
*/
unsigned int container : 2; /* NONE==1 or ZIPLIST==2 */
/* 当使用lindex这样的命令查看某一项本来压缩的数据的时候，需要把数据暂时解压，
* 这时就设置recompress=1做一个标记，等有机会再把数据进行压缩
*/
unsigned int recompress : 1;
/* 只用于redis的自动化测试 */
unsigned int attempted_compress : 1;
unsigned int extra : 10;
} quicklistNode; typedef struct quicklistLZF {
/* ziplist压缩之后的存储在compressed中的数据长度；
* 压缩之前的数据的长度存储在quicklistNode的成员sz中
*/
unsigned int sz;
/* 柔性数组 */
char compressed[];
} quicklistLZF;其他的辅助性的结构体的定义分别为quicklistIter，用于在quicklist中迭代entry（从后向前或者是从前向后）；还有quicklistEntry，代表ziplist中某个entry的有关信息。
typedef struct quicklistIter {
const quicklist *quicklist;
/* 指向当前的zi所在的quicklistNode */
quicklistNode *current;
/* 指向quicklistNode中某个entry的开始地址 */
unsigned char *zi;
/* zi在当前quicklistNode中的偏移量 */
long offset;
/* 向前还是向后 */
int direction;
} quicklistIter;
typedef struct quicklistEntry {
const quicklist *quicklist;
/* 指向当前节点的指针 */
quicklistNode *node;
/* 指向在entry在ziplist中的开始地址,就是该entry的prev_entry_length在ziplist中的地址 */
unsigned char *zi;
/* 如果ziplist中的entry的编码格式是字符串,则value指向ziplist中entry的实际值 */
unsigned char *value;
/* 如果ziplist中，该entry的编码格式是整数的形式，longval被赋予entry中的实际值*/
long long longval;
/* sz代表在zi指向的entry中，字符串值的长度 */
unsigned int sz;
/* 当前zi指向的entry在当前ziplist的偏移量 */
int offset;
} quicklistEntry;在介绍了redis中实现quicklist的几种重要数据结构之后，针对自己在阅读redis的源码的过程中认为比较困难或者比较重要的函数加以注释解释，至于比较比较简单、或者不涉及到quicklist设计技巧的辅助性函数，请阅读redis源码的quicklist.c文件。
quicklist的思想是将一个很大的ziplist拆分成若干个较小的ziplist，那么当插入一个entry到现有的quicklistNode中的时候，必须根据list-max-ziplist-size的值判断quicklistNode指向的ziplist是否能够继续插入新的entry，由_quicklistNodeAllowInsert()完成该任务。
REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node,
const int fill, const size_t sz) {
if (unlikely(!node))
return 0;
int ziplist_overhead;
/* 这种估计方式即使不考虑连锁更新带来的影响，那么也同样存在缺陷之处：
* 后一个节点的prev_entry_length域的长度不仅仅包含了encoding和实际数据的长度，
* 同样也包含了本节点的prev_entry_length域的长度;
* 在本函数的估计方法中，只是用实际数据的长度估计后一个节点的prev_entry_length域的长度
*/
/* 计算prev_entry_length域需要的字节数 */
if (sz < 254)
ziplist_overhead = 1;
else
ziplist_overhead = 5;
/* encoding域需要的长度 */
if (sz < 64)
ziplist_overhead += 1;
else if (likely(sz < 16384))
ziplist_overhead += 2;
else
ziplist_overhead += 5;
/* 这只是一种大概的估计，node的已有长度+新增节点的长度;
* 这种估计忽略了级联更新带来的影响;
*/
unsigned int new_sz = node->sz + sz + ziplist_overhead;
/* fill为负数的时候，该节点已有数据的长度加上新插入的数据的长度是否能够继续存储在当前quicklistNode指向的ziplist结构中*/
if (likely(_quicklistNodeSizeMeetsOptimizationRequirement(new_sz, fill)))
return 1;
/* 在没有满足if的条件下，防止单个节点的长度过大 */
else if (!sizeMeetsSafetyLimit(new_sz))
return 0;
/* fill为正数的时候，防止节点的数目过多 */
else if ((int)node->count < fill)
return 1;
else
return 0;
}
其中，关于_quicklistNodeSizeMeetsOptimizationRequirement()主要是是根据要插入的元素的sz以及配置的list-max-ziplist-size确定是否允许插入，其实现如下。
REDIS_STATIC int
_quicklistNodeSizeMeetsOptimizationRequirement(const size_t sz,
const int fill) {
if (fill >= 0)
return 0;
/* 确定在限制ziplist大小的数组中的偏移量 */
size_t offset = (-fill) - 1;
if (offset < (sizeof(optimization_level) / sizeof(*optimization_level))) {
if (sz <= optimization_level[offset]) {
return 1;
} else {
return 0;
}
} else {
return 0;
}
} redis.conf文件中的redis-compress-depth配置参数决定是否以及如何对ziplist数据结构进行压缩，但是，redis究竟是否真的压缩一个quicklistNode指向的ziplist，还要看是否满足压缩的条件，具体由__quicklistCompressNode(quicklistNode *node)实现。
REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
#ifdef REDIS_TEST
node->attempted_compress = 1;
#endif
/* #define MIN_COMPRESS_BYTES 48
* 允许压缩的ziplist的最小长度，小于这个值不执行压缩
*/
if (node->sz < MIN_COMPRESS_BYTES)
return 0;
/* 为压缩之后的空间分配足够的大小 */
quicklistLZF *lzf = zmalloc(sizeof(*lzf) + node->sz);
/* 如果压缩失败(压缩过程本身出错或者是传入的压缩之后的空间大小不满足要求)
* 或者是压缩之后节省的空间小于MIN_COMPRESS_IMPROVE，不执行压缩
*/
if (((lzf->sz = lzf_compress(node->zl, node->sz, lzf->compressed,
node->sz)) == 0) ||
lzf->sz + MIN_COMPRESS_IMPROVE >= node->sz) {
zfree(lzf);
return 0;
}
/* 去掉压缩之后的lzf中多余的空间大小 */
lzf = zrealloc(lzf, sizeof(*lzf) + lzf->sz);
/* 释放掉原来的ziplist占据的空间大小 */
zfree(node->zl);
node->zl = (unsigned char *)lzf;
node->encoding = QUICKLIST_NODE_ENCODING_LZF;
node->recompress = 0;
return 1;
}
既然存在压缩，那么肯定存在解压，将quicklistLZF结构的数据解压到ziplist结构，__quicklistDecompressNode就是完成这个功能的，如下。
REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) {
#ifdef REDIS_TEST
node->attempted_compress = 0;
#endif
void *decompressed = zmalloc(node->sz);
quicklistLZF *lzf = (quicklistLZF *)node->zl;
if (lzf_decompress(lzf->compressed, lzf->sz, decompressed, node->sz) == 0) {
zfree(decompressed);
return 0;
}
zfree(lzf);
node->zl = decompressed;
node->encoding = QUICKLIST_NODE_ENCODING_RAW;
return 1;
}以上，是关于redis对quicklist设计思想以及redis.conf文件中对于list-max-ziplist-size和list-compress-depth参数的讲解。下面，剖析list的增、删、改、查调用的quicklist的接口。
在quicklist的头部增加entry。
int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
quicklistNode *orig_head = quicklist->head;
if (likely(
_quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
quicklist->head->zl =
ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
quicklistNodeUpdateSz(quicklist->head);
} else {
quicklistNode *node = quicklistCreateNode();
node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
quicklistNodeUpdateSz(node);
/* 将新增加的节点添加到列表的quicklist的头部 */
_quicklistInsertNodeBefore(quicklist, quicklist->head, node);
}
quicklist->count++;
quicklist->head->count++;
return (orig_head != quicklist->head);
}
在quicklist的尾部增加entry。
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->count++;
quicklist->tail->count++;
return (orig_tail != quicklist->tail);
}
在某个指定的quicklistEntry之前或者之后增加一个新的entry。after=0，表示在传入的entry之前插入新的entry；after=1表示在传入的entry之后插入新的entry。
REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
void *value, const size_t sz, int after) {
int full = 0, at_tail = 0, at_head = 0, full_next = 0, full_prev = 0;
int fill = quicklist->fill;
quicklistNode *node = entry->node;
quicklistNode *new_node = NULL;
if (!node) {
/* we have no reference node, so let's create only node in the list */
D("No node given!");
new_node = quicklistCreateNode();
new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
/* __quicklistInsertNode中在增加node的同时已经将node的数量增加过了 */
__quicklistInsertNode(quicklist, NULL, new_node, after);
/* 增加当前node中ziplist中的entry的数量 */
new_node->count++;
/* 增加整个quicklist中entry的数量 */
quicklist->count++;
return;
}
/* 插入的策略是，如果当前的节点允许进行插入，那么直接在当前的节点中进行插入;
* 否则根据after的值决定是在node->next中插入还是在node->prev中插入
*/
/* Populate accounting flags for easier boolean checks later */
/* 判断当前的节点是否允许插入 */
if (!_quicklistNodeAllowInsert(node, fill, sz)) {
D("Current node is full with count %d with requested fill %lu",
node->count, fill);
full = 1;
}
/* 一般情况下，entry->offset <= node->count - 1，但是在执行ziplistDelEntry之后，
* 可能offset的值为node->count，这种情况下执行quicklistNext的时候，
* 会跳转到下一个quicklistNode
*/
if (after && (entry->offset == node->count)) {
D("At Tail of current ziplist");
at_tail = 1;
if (!_quicklistNodeAllowInsert(node->next, fill, sz)) {
D("Next node is full too.");
full_next = 1;
}
}
/* 如果当前的entry为某个node中最前面的一个，并且新添加的entry要位于该entry之前，
* 那么需要判断前一个节点之中是否再次允许添加entry
*/
if (!after && (entry->offset == 0)) {
D("At Head");
at_head = 1;
if (!_quicklistNodeAllowInsert(node->prev, fill, sz)) {
D("Prev node is full too.");
full_prev = 1;
}
}
/* Now determine where and how to insert the new element */
/* 如果当前节点允许插入新的entry，则直接在entry的后面插入 */
if (!full && after) {
D("Not full, inserting after current position.");
quicklistDecompressNodeForUse(node);
/* 确定新的entry要插入的位置 */
unsigned char *next = ziplistNext(node->zl, entry->zi);
if (next == NULL) {
node->zl = ziplistPush(node->zl, value, sz, ZIPLIST_TAIL);
} else {
node->zl = ziplistInsert(node->zl, next, value, sz);
}
node->count++;
quicklistNodeUpdateSz(node);
quicklistRecompressOnly(quicklist, node);
/* 如果当前节点允许插入新的entry，那么当前entry所在的位置就应该是插入的位置 */
} else if (!full && !after) {
D("Not full, inserting before current position.");
quicklistDecompressNodeForUse(node);
node->zl = ziplistInsert(node->zl, entry->zi, value, sz);
node->count++;
quicklistNodeUpdateSz(node);
quicklistRecompressOnly(quicklist, node);
/* 如果满足以下条件，那么在当前node的下一个node中的ziplist的头部插入新的entry:
* 1.当前ziplist中不允许插入新的entry，也就是full为1;
* 2.要插入的位置为当前的node的最后一个位置，也就是at_tail == 1;
* 3.node的下一个节点存在,并且下一个节点允许插入新的entry;
* 4.是要在尾部插入，但是个人感觉这项条件不必放在这里，因为at_tail已经表明after为真;
*/
/* 只有要在当前节点的尾部进行插入，并且下一个节点允许插入的时候才在下一个node对应的
* ziplist的头部进行插入
*/
} else if (full && at_tail && node->next && !full_next && after) {
/* If we are: at tail, next has free space, and inserting after:
* - insert entry at head of next node. */
D("Full and tail, but next isn't full; inserting next node head");
new_node = node->next;
quicklistDecompressNodeForUse(new_node);
new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_HEAD);
new_node->count++;
quicklistNodeUpdateSz(new_node);
quicklistRecompressOnly(quicklist, new_node);
/* 与上一个一样，只不过这里是要在当前node对应的ziplist的头部进行插入的时候，
* 如果上一个node对应的ziplist中允许插入新的entry,在上一个节点对应的ziplist的
* 尾部插入新的entry
*/
} else if (full && at_head && node->prev && !full_prev && !after) {
/* If we are: at head, previous has free space, and inserting before:
* - insert entry at tail of previous node. */
D("Full and head, but prev isn't full, inserting prev node tail");
new_node = node->prev;
quicklistDecompressNodeForUse(new_node);
new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_TAIL);
new_node->count++;
quicklistNodeUpdateSz(new_node);
quicklistRecompressOnly(quicklist, new_node);
/* 如果满足以下条件，那么创建新的quicklistNode，将entry插入到新创建的quicklistNode中，
* 并根据after的值决定将新创建的quicklistNode链接到当前node的前面还是后面:
* 1.当前节点不允许插入要插入的entry;
* 2.如果要插入到当前node的最后位置，但是node->next存在并且已经不允许插入
* 或者要插入到当前node最开始的位置，但是node->prev存在并且已经不允许插入
*/
} else if (full && ((at_tail && node->next && full_next && after) ||
(at_head && node->prev && full_prev && !after))) {
/* If we are: full, and our prev/next is full, then:
* - create new node and attach to quicklist */
D("\tprovisioning new node...");
new_node = quicklistCreateNode();
new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
new_node->count++;
quicklistNodeUpdateSz(new_node);
__quicklistInsertNode(quicklist, node, new_node, after);
/* 如果要插入的位置在当前node的中间位置（不是最前面也不是最后面）且当前的node已满,
* 不允许插入新的entry,则在当前entry所在的位置将node分裂
*/
} else if (full) {
/* else, node is full we need to split it. */
/* covers both after and !after cases */
D("\tsplitting node...");
quicklistDecompressNodeForUse(node);
/* 根据当前entry的offset将node对应的ziplist分裂成两个quicklistNode,
* after=1的时候，返回的是[offset+1, end],此时在new_node对应的ziplist的头部进行插入;
* after=0的时候，返回的是[0, offset-1],此时在new_node对应的ziplist的尾部插入;
*/
new_node = _quicklistSplitNode(node, entry->offset, after);
new_node->zl = ziplistPush(new_node->zl, value, sz,
after ? ZIPLIST_HEAD : ZIPLIST_TAIL);
new_node->count++;
quicklistNodeUpdateSz(new_node);
/* 将分裂出的节点根据after的值，确定是放在输入节点的前面还是放在输入节点的后面 */
__quicklistInsertNode(quicklist, node, new_node, after);
/* else处理的是当前节点不能在entry的之前或者之后直接插入，并且要插入的位置不是在当前node对应的ziplist的最开始或者最后的时候，
* 此时要对ziplistNode进行分裂，分裂之后的new_node可能可以和node->prev或者node->next进行合并，
* 所以执行_quicklistMergeNodes
*/
_quicklistMergeNodes(quicklist, node);
}
quicklist->count++;
}
删除从偏移位置start开始的count个节点。
int quicklistDelRange(quicklist *quicklist, const long start,
const long count) {
if (count <= 0)
return 0;
unsigned long extent = count; /* range is inclusive of start position */
/* 对从start开始的超出ziplist的长度限制的情况进行约束 */
if (start >= 0 && extent > (quicklist->count - start)) {
/* if requesting delete more elements than exist, limit to list size. */
extent = quicklist->count - start;
} else if (start < 0 && extent > (unsigned long)(-start)) {
/* else, if at negative offset, limit max size to rest of list. */
extent = -start; /* c.f. LREM -29 29; just delete until end. */
}
quicklistEntry entry;
/* 如果start索引位置的entry并不存在，则quicklistIndex返回的值为0,表示没有这个entry;
* 如果start索引位置的entry存在，则quicklistIndex返回的值为1，表示这个entry是存在的，
* 这样能够保证以后对这个entry的所有操作都是合理的
*/
if (!quicklistIndex(quicklist, start, &entry))
return 0;
D("Quicklist delete request for start %ld, count %ld, extent: %ld", start,
count, extent);
quicklistNode *node = entry.node;
/* iterate over next nodes until everything is deleted. */
while (extent) {
quicklistNode *next = node->next;
unsigned long del;
int delete_entire_node = 0;
/* 删除当前的整个节点的情况;可能还需要删除掉其余节点中的entry */
if (entry.offset == 0 && extent >= node->count) {
/* If we are deleting more than the count of this node, we
* can just delete the entire node without ziplist math. */
delete_entire_node = 1;
del = node->count;
/* else if 中肯定不存在entry.offset == 0 并且 extent >= node->count 的情况;
因此，肯定不存在删除整个quicklistNode的情况
*/
/* 个人认为这里存在问题，正确的判断条件应该是（entry.offset >=0 && extent >= node->count - entry.offset。举个例子，假如每个quicklist Node中最多允许存在9个entry，现在一共有3个quicklistNode，且每个quicklistNode中都存在9个entry。当start=6,count=6时，直接进入了最后了else，del变
为6，在第一个quicklisNode指向的ziplist中从offset=6开始删除6个元素，但是此时很显然只能删除4个元素，ziplistDelteRange并不能分辨出这种错误
* 欢迎吐槽这个疑问！！！
*/
} else if (entry.offset >= 0 && extent >= node->count) {
/* If deleting more nodes after this one, calculate delete based
* on size of current node. */
del = node->count - entry.offset;
} else if (entry.offset < 0) {
/* If offset is negative, we are in the first run of this loop
* and we are deleting the entire range
* from this start offset to end of list. Since the Negative
* offset is the number of elements until the tail of the list,
* just use it directly as the deletion count. */
del = -entry.offset;
/* If the positive offset is greater than the remaining extent,
* we only delete the remaining extent, not the entire offset.
*/
if (del > extent)
del = extent;
} else {
/* else, we are deleting less than the extent of this node, so
* use extent directly. */
del = extent;
}
D("[%ld]: asking to del: %ld because offset: %d; (ENTIRE NODE: %d), "
"node count: %u",
extent, del, entry.offset, delete_entire_node, node->count);
if (delete_entire_node) {
__quicklistDelNode(quicklist, node);
} else {
/* 解压当前的quicklistNode用于删除quciklistNode指向的ziplist中指定的entry */
quicklistDecompressNodeForUse(node);
node->zl = ziplistDeleteRange(node->zl, entry.offset, del);
quicklistNodeUpdateSz(node);
node->count -= del;
quicklist->count -= del;
/* quicklistDeleteIfEmpty在释放掉node指向的内存空间之后，将node置为NULL，
* 因此在下面可以对node是否NULL进行判断；仅仅是free的话并不会将指针置为NULL
*/
quicklistDeleteIfEmpty(quicklist, node);
if (node)
quicklistRecompressOnly(quicklist, node);
}
extent -= del;
node = next;
entry.offset = 0;
}
return 1;
}对于链表而言，经常做的操作是从链表的头部或者是尾部进行删除操作。
/* 返回为0，表示没有没有找到相应的entry;返回1,表示存在相应的entry */
int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *slong) {
unsigned char *vstr;
unsigned int vlen;
long long vlong;
if (quicklist->count == 0)
return 0;
int ret = quicklistPopCustom(quicklist, where, &vstr, &vlen, &vlong,
_quicklistSaver);
if (data)
*data = vstr;
if (slong)
*slong = vlong;
if (sz)
*sz = vlen;
return ret;
}
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *sval,
void *(*saver)(unsigned char *data, unsigned int sz)) {
unsigned char *p;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
int pos = (where == QUICKLIST_HEAD) ? 0 : -1;
if (quicklist->count == 0)
return 0;
/* 对传入的参数进行初始化，data用来存储set，sz表示data的长度;
* 当只是数值的时候，直接用sval来存储
*/
if (data)
*data = NULL;
if (sz)
*sz = 0;
if (sval)
*sval = -123456789;
quicklistNode *node;
if (where == QUICKLIST_HEAD && quicklist->head) {
node = quicklist->head;
} else if (where == QUICKLIST_TAIL && quicklist->tail) {
node = quicklist->tail;
} else {
return 0;
}
p = ziplistIndex(node->zl, pos);
if (ziplistGet(p, &vstr, &vlen, &vlong)) {
if (vstr) {
if (data)
*data = saver(vstr, vlen);
if (sz)
*sz = vlen;
} else {
if (data)
*data = NULL;
if (sval)
*sval = vlong;
}
quicklistDelIndex(quicklist, node, &p);
return 1;
}
return 0;
}
改，quicklist允许在指定的偏移位置用新值代替旧值。
int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
int sz) {
quicklistEntry entry;
if (likely(quicklistIndex(quicklist, index, &entry))) {
/* quicklistIndex provides an uncompressed node */
entry.node->zl = ziplistDelete(entry.node->zl, &entry.zi);
entry.node->zl = ziplistInsert(entry.node->zl, entry.zi, data, sz);
quicklistNodeUpdateSz(entry.node);
/* 判断是否应该对entry.node进行压缩 */
quicklistCompress(quicklist, entry.node);
return 1;
} else {
return 0;
}
} 查，quicklist根据指定的偏移量确定该位置的entry。
int quicklistIndex(const quicklist *quicklist, const long long idx,
quicklistEntry *entry) {
quicklistNode *n;
unsigned long long accum = 0;
unsigned long long index;
int forward = idx < 0 ? 0 : 1; /* < 0 -> reverse, 0+ -> forward */
initEntry(entry);
entry->quicklist = quicklist;
if (!forward) {
index = (-idx) - 1;
n = quicklist->tail;
} else {
index = idx;
n = quicklist->head;
}
if (index >= quicklist->count)
return 0;
/* 已经将 index > quicklist->count 的情况排除在外 */
while (likely(n)) {
if ((accum + n->count) > index) {
break;
} else {
D("Skipping over (%p) %u at accum %lld", (void *)n, n->count,
accum);
accum += n->count;
n = forward ? n->next : n->prev;
}
}
/* quicklist中entry的数目不够的话，可能导致n变成了NULL指针;
* 经过上面的循环之后重新检查n是否为NULL。但是index >= quicklist->count的条件难道不是已经排除了这种情况了吗？
*/
if (!n)
return 0;
D("Found node: %p at accum %llu, idx %llu, sub+ %llu, sub- %llu", (void *)n,
accum, index, index - accum, (-index) - 1 + accum);
entry->node = n;
if (forward) {
/* forward = normal head-to-tail offset. */
entry->offset = index - accum;
} else {
/* reverse = need negative offset for tail-to-head, so undo
* the result of the original if (index < 0) above. */
/* 如果是从最后向前按照索引进行查找（也就是传入的参数中 idx < 0），
* 传入到ziplist中进行查找的时候也需要传入在当前的ziplist中的负的偏移量
*/
entry->offset = (-index) - 1 + accum;
}
/* 以下的操作中确定quicklistNode和quicklistNode中的ziplist的entry都是存在的 */
quicklistDecompressNodeForUse(entry->node);
entry->zi = ziplistIndex(entry->node->zl, entry->offset);
ziplistGet(entry->zi, &entry->value, &entry->sz, &entry->longval);
/* The caller will use our result, so we don't re-compress here.
* The caller can recompress or delete the node as needed. */
return 1;
}当然，quicklist提供了迭代获取每一个元素的方法,quicklistNext。
/* 获取quicklistIter中成员变量zi指向的entry的下一个entry，
* 当然quicklistNext是一个自身的递归调用
*/
int quicklistNext(quicklistIter *iter, quicklistEntry *entry) {
initEntry(entry);
if (!iter) {
D("Returning because no iter!");
return 0;
}
entry->quicklist = iter->quicklist;
entry->node = iter->current;
if (!iter->current) {
D("Returning because current node is NULL")
return 0;
}
unsigned char *(*nextFn)(unsigned char *, unsigned char *) = NULL;
int offset_update = 0;
/* 满足if条件的有以下几种情况:
* 1.初始化一个quicklistIter之后直接调用该函数;
* 2.调用quicklistDelEntry删除一个node的节点之后，iter->zi被赋值为NULL值,这里又划分为两种情况:
* 1.删除的是一个node的最后一个entry，那么iter->offset并没有发生变化（参考quicklistDelEntry的注释)
* ,此时通过ziplistIndex获取当前的entry仍然为NULL值;
* 2.删除的node中不是最后一个entry,此时iter->offset并没有发生也不用发生变化(参考quicklistDelEntry
* 的注释说明),那么此时通过iter->offset调用ziplistIndex可以获取相应的ziplist在iter->offset
* 位置的entry;
* 3. 函数发生递归调用的时候iter->zi被赋值为NULL;
*/
if (!iter->zi) {
/* If !zi, use current index. 即使是经过ziplistIndex获取iter->offset，iter->zi仍然可能为NULL值 */
quicklistDecompressNodeForUse(iter->current);
iter->zi = ziplistIndex(iter->current->zl, iter->offset);
/* 如果iter->zi指向的位置的entry是存在的，那么获取iter->zi的下一个位置的entry */
} else {
/* else, use existing iterator offset and get prev/next as necessary. */
/* 如果iter指向的是某个node中的最后一个entry，
* 那么ziplistNext返回的是NULL
*/
if (iter->direction == AL_START_HEAD) {
nextFn = ziplistNext;
offset_update = 1;
/* 如果iter指向的是某个node中的最开始的一个entry，
* 那么ziplistPrev返回的是NULL
*/
} else if (iter->direction == AL_START_TAIL) {
nextFn = ziplistPrev;
offset_update = -1;
}
iter->zi = nextFn(iter->current->zl, iter->zi);
iter->offset += offset_update;
}
entry->zi = iter->zi;
entry->offset = iter->offset;
/* 如果获取的iter->zi存在，那么直接获取获取对应的entry中的值 */
if (iter->zi) {
/* Populate value from existing ziplist position */
ziplistGet(entry->zi, &entry->value, &entry->sz, &entry->longval);
return 1;
/* 如果对应的iter->zi(为NULL的话),根据direction参数确定下一个quicklistNode */
} else {
/* We ran out of ziplist entries.
* Pick next node, update offset, then re-run retrieval. */
quicklistCompress(iter->quicklist, iter->current);
if (iter->direction == AL_START_HEAD) {
/* Forward traversal */
D("Jumping to start of next node");
iter->current = iter->current->next;
iter->offset = 0;
} else if (iter->direction == AL_START_TAIL) {
/* Reverse traversal */
D("Jumping to end of previous node");
iter->current = iter->current->prev;
iter->offset = -1;
}
/* 适用于最开始的if条件语句中判断iter->zi为NULL，此时可以通过以上的iter->offset设定
* 的值，确定新的iter->zi，并确定entry中响应的值;
*/
iter->zi = NULL;
return quicklistNext(iter, entry);
}
}
quicklist的操作中，针对quicklistNode的操作是双端链表的操作；针对每一个具体的entry的操作，直接调用redis中ziplist的实现。redis中关于双端链表的实现还包括以下的函数。
·quicklistAppendValuesFromZiplist():实现向前的兼容性，将原来3.2版本之前的ziplist可以转换成quicklist结构，用于读取rdb或者aof文件，将数据加载到内存中；
·_quicklistSplitNode():将一个节点分裂成两个节点，其实现如下：
REDIS_STATIC quicklistNode *_quicklistSplitNode(quicklistNode *node, int offset,
int after) {
size_t zl_sz = node->sz;
quicklistNode *new_node = quicklistCreateNode();
new_node->zl = zmalloc(zl_sz);
/* Copy original ziplist so we can split it */
memcpy(new_node->zl, node->zl, zl_sz);
/* -1 here means "continue deleting until the list ends" */
/* 这里之所以说-1表示删除掉从某个位置开始的entry中的左右数目，是因为ziplistDeleteRange(ziplist* ziplist, unsigned char *p, unsigned int num)中num的类型为unsigned int，但是如果传递-1，表
示传递的是int类型，根据数据在计算机中的"补码"表示形式，那么实际传递进去的应该是0xffffffff，这个值已经足可以和zlbytes成员的上限一致了，但是每个ziplist中的entry至少需要2个字节，所以总的ziplis t中总的数量是不可能达到0xffffffff的。所以，传递"-1"表示删除掉从开始位置开始的所有的元素
*/
/* 根据输入的参数after确定输入的节点node要删除的entry的起始位置以及数量 */
int orig_start = after ? offset + 1 : 0;
int orig_extent = after ? -1 : offset;
/* 根据输入的参数after确定新创建的节点中要删除的entry的起始位置和数量 */
int new_start = after ? 0 : offset;
int new_extent = after ? offset + 1 : -1;
D("After %d (%d); ranges: [%d, %d], [%d, %d]", after, offset, orig_start,
orig_extent, new_start, new_extent);
node->zl = ziplistDeleteRange(node->zl, orig_start, orig_extent);
node->count = ziplistLen(node->zl);
quicklistNodeUpdateSz(node);
new_node->zl = ziplistDeleteRange(new_node->zl, new_start, new_extent);
new_node->count = ziplistLen(new_node->zl);
quicklistNodeUpdateSz(new_node);
D("After split lengths: orig (%d), new (%d)", node->count, new_node->count);
return new_node;
}·_quicklistZiplistMerge():用于讲两个节点合并成一个节点；
以上是个人关于redis中的quicklist实现的理解，欢迎吐槽。
上一篇： 求高手解答， php 弱数据类型，IDE的智能提示问题
下一篇： PHP 七大优势分析_PHP教程