redis quicklist剖析

发表于 更新于 阅读次数：

说明

本篇文章详细剖析了redis的quicklist结构。前面我们已经介绍了两种redis数据结构：adlist和ziplist，这两中数据结构各有各的优缺点。

  • adlist: 双向链表，任意位置插入和删除非常方便，缺点也很显而易见，任意一个node都是独立的内存块，所以内存碎片化很严重
  • ziplist：压缩双向链表，一个链表就是一整块内存，同时如果元素值为整数的话可以进一步压缩，所以很省内存；缺点就是任意一次插入删除操作都会导致重新分配内存的操作，效率不高。

quicklist结构就是adlist和ziplist两种结构的中间体，整个结构分为两层：外层使用类似adlist的结构串联，内层使用ziplist来节约内存

为了进一步的优化内存，quicklist还有一个参数compress用来控制除了两头多少个节点外，中间的节点使用压缩算法进行压缩（quicklist使用lzf压缩）。例如如果compress==1，则除了头尾两个节点，中间的节点都将被压缩。

实现

为了实现这些功能，quicklist实现非常复杂，应该是redis中最复杂的一个结构体了。但是不考虑它的实现细节，它的一些实现这些功能的思想，可以很简单的通过它定义的一些结构体很容易的理解。

quicklistNode

1
2
3
4
5
6
7
8
9
10
11
12
typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *zl;
    unsigned int sz;             /* ziplist size in bytes */
    unsigned int count : 16;     /* count of items in ziplist */
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    unsigned int recompress : 1; /* was this node previous compressed? */
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode;

对应于adlist的listNode结构，用做一个双向链表的节点。不同的是该节点内部的值指向的是一个ziplist/quicklistLZF数据结构，具体指向哪个结构通过encoding变量控制。

  • prev, next：指向前后两个quicklistNode地址
  • zl：如果当前node的数据没有被压缩(encoding==RAW)，则zl指向一个ziplist结构，否则指向quicklistLZF结构
  • sz：ziplist内存大小
  • count：存放该节点的元素个数
  • encoding：表示当前节点的数据是以什么编码的，RAW：原始数据，LZF：通过LZF压缩数据
  • container：表示当前节点通过什么类型数据结构存储的，NONE：没有数据，ZIPLIST：通过ziplist存储，目前也就只有ziplist这一种
  • recompress：如果该值为true，表示只是临时将数据解压用于使用，需要再次被压缩
  • attempted_compress：用于测试用例
  • extra：保留，暂未使用

quicklistLZF

1
2
3
4
typedef struct quicklistLZF {
    unsigned int sz;        // 通过LZF压缩后的数据长度
    char compressed[];      // 压缩后的数据
} quicklistLZF;

quicklistNode中的ziplist压缩之后使用该结构体存储，quicklistNode中zl将会指向该结构体

quicklist

1
2
3
4
5
6
7
8
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：总元素个数（不是quicklistNode个数），所有node指向的ziplist的元素总数
  • len：quicklistNode个数
  • fill：控制每个quicklistNode中的ziplist大小，如果为正值，则表示以ziplist中元素个数为限制，但总的ziplist占用内存不能大于8K，如果未负值，则以ziplist占用内存大小来控制，-1：4k，-2：8k(redis默认)，-3：16k，-4：32k，-5，64k
  • compress：如果非0，表示头尾多少个节点不需要压缩，例如如果是2的话，表示除头尾各2个quicklistNode不被压缩外，其他都会被压缩，以减少内存

quicklistIter

1
2
3
4
5
6
7
typedef struct quicklistIter {
    const quicklist *quicklist;
    quicklistNode *current;
    unsigned char *zi;              // ziplist中当前的entry地址
    long offset;                    // ziplist中当前entry的偏移量
    int direction;                  // 迭代方向
} quicklistIter;

quicklist迭代器，不仅要知道当前指向哪个quicklistNode，还需要知道当前node中的ziplist的元素地址及偏移量

quicklistEntry

1
2
3
4
5
6
7
8
9
typedef struct quicklistEntry {
    const quicklist *quicklist;
    quicklistNode *node;        // 当前quicklistNode
    unsigned char *zi;          // ziplist当前的entry地址
    unsigned char *value;       // ziplist中字符串的value值
    long long longval;          // ziplist中整数值
    unsigned int sz;            // ziplist中字符串的value长度
    int offset;                 // ziplist中的offset
} quicklistEntry;

该结构体指向quicklist中具体的一个元素，quicklist中的get方法(quicklistIndex, quicklistNext)都是通过该结构体来返回元素的值。这里需要注意两点：

  1. 如果get方法返回正确，但是value指向null，则表示该元素被ziplist使用整形进行压缩了，返回的值存放在longval中。
  2. 如果value不为null，则value实际指向的是ziplist中的元素地址，所以应该假设这一块地址为只读的，不应该修改他，如果需要保存该value，则应该复制这一块的内存，以防止该指针会失效(ziplist插入/删除都会重新分配内存)

主要函数分析

时间复杂度不太好分析，但是肯定是优于ziplist，但差于adlist。所以直接看每个函数的声明及作用，具体细节可以查看我注释的redis源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
// 创建一个空的quicklist，需要通过quicklistRelease释放
quicklist *quicklistCreate(void);

// 通过自定义fill和compress创建quicklist，同样需要通过quicklistRelease释放
quicklist *quicklistNew(int fill, int compress);

// 设置quicklist depth属性
void quicklistSetCompressDepth(quicklist *quicklist, int depth);

// 设置quicklist fill属性
void quicklistSetFill(quicklist *quicklist, int fill);

// 设置quiicklist fill和depth属性
void quicklistSetOptions(quicklist *quicklist, int fill, int depth);

// 释放整个quicklist内存
void quicklistRelease(quicklist *quicklist);

// 将value插入到quicklist头部
// 如果创建新的quicklistNode返回1，否则0
int quicklistPushHead(quicklist *quicklist, void *value, const size_t sz);

// 将value插入到quicklist尾部
int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz);

// 插入value到头部/尾部，通过where控制，QUICKLIST_HEAD/QUICKLIST_TAIL
void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
                   int where);

// 创建新的quicklistNode来存放追加的ziplist，在rdb中使用
void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl);

// 将ziplist所有元素追加到quicklist中，并释放ziplist内存
quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist,
                                            unsigned char *zl);

// 通过ziplist所有元素创建一个新的quicklist，会释放原来的ziplist
quicklist *quicklistCreateFromZiplist(int fill, int compress,
                                      unsigned char *zl);

// 在元素之后插入value
void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *node,
                          void *value, const size_t sz);

// 在元素之前插入value
void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *node,
                           void *value, const size_t sz);

// 删除指定元素，iter按照迭代方向返回下一个元素
void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry);

// 替换quicklist中index下标的元素，替换成功返回1，否则返回0
int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
                            int sz);

// 删除quicklist中一定范围的所有元素，删除成功返回1，失败返回0
int quicklistDelRange(quicklist *quicklist, const long start, const long stop);

// 创建一个迭代器，该迭代器的方向通过direction指定，AL_START_HEAD/AL_START_TAIL
quicklistIter *quicklistGetIterator(const quicklist *quicklist, int direction);

// 从index开始创建一个迭代器，方向由direction指定，AL_START_HEAD/AL_START_TAIL，如果index无效返回NULL
quicklistIter *quicklistGetIteratorAtIdx(const quicklist *quicklist,
                                         int direction, const long long idx);

// 获取迭代器下一个元素，返回0表示没有元素，否则获取下个元素成功
int quicklistNext(quicklistIter *iter, quicklistEntry *node);

// 释放迭代器内存，如果当前迭代器指向了一个quicklistNode，则尝试重新压缩他
void quicklistReleaseIterator(quicklistIter *iter);

// 从orig中复制整一个quicklist
quicklist *quicklistDup(quicklist *orig);

// 获取quicklist中index为下标的元素的值，存储在entry中返回
// index>=0表示从头开始，<0表示从尾部开始
// 找到返回1，否则返回0
int quicklistIndex(const quicklist *quicklist, const long long index,
                   quicklistEntry *entry);

// 下面两个函数没有被实现
void quicklistRewind(quicklist *quicklist, quicklistIter *li);
void quicklistRewindTail(quicklist *quicklist, quicklistIter *li);

// 旋转quicklist，将最后一个元素插入到头部
void quicklistRotate(quicklist *quicklist);

// 从quicklist头部/尾部pop元素，头部/尾部通过where控制，返回值根据是不是整形，通过data/sval返回。
// 如果pop的元素不是整形，则会调用saver函数用来存储该元素的内容，返回值将会赋值给data，因为pop之后quicklist中该元素的内容将会被清除
// 所以需要调用用户自定义的函数进行保存该内容
// 返回0表示没有元素，返回1表示pop元素成功，可以使用data/sval
// 如果返回1，且data为NULL，表示该元素是个整形，可以使用sval，否则的话data和sz表示当前的元素内容
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                       unsigned int *sz, long long *sval,
                       void *(*saver)(unsigned char *data, unsigned int sz));

// 默认quicklistpop函数（不可以自定义saver，其他跟quicklistPopCustom一样）
int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
                 unsigned int *sz, long long *slong);

// 返回quicklist总的元素个数
unsigned int quicklistCount(quicklist *ql);

// 比较p指向的ziplist的entry的内容与指定值相等，不等返回0，否则返回1
int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len);

// 获取quicklistNode中的lzf原始数据
// data存储lzf压缩后的数据，返回值为压缩后数据的长度
size_t quicklistGetLzf(const quicklistNode *node, void **data);