手写快速列表 quicklist
项目地址：xgzlucario/quicklist
前言
quicklist 是 Redis 为了优化列表（List）的内存使用而引入的一种数据结构。
在之前的版本，quicklist 是由多个 ziplist（压缩列表）组成的 linkedlist（双向链表），这样的设计结合了 ziplist 的内存高效性和 linkedlist 的快速插入与删除特性。因此，quicklist 在不同的使用场景下都能保持较好的性能与效率。
但由于 ziplist 中的每个 entry 都编码保存着前一个 entry 长度，在极端情况下，修改一个元素可能需要对后续所有元素进行移动，导致性能问题，这种情况也被称为 级联更新（cascade update）。为了解决 ziplist 的这一设计缺陷，Redis 在 7.0 版本正式引入 listpack 全面代替 ziplist。
本文是 手写系列 的扩展，将介绍 quicklist 内部设计思想，并实现一个 Golang 版本的 quicklist.
上一篇：手写缓存 GigaCache
性能
在 Golang 中，quicklist 更适合存储大量小元素，相比 []string 插入性能提升 25%，内存使用减少 70%，并且具有更小的 GC 延迟。
完整性能测试请查看 xgzlucario/quicklist。
[]string
entries: 20000000
alloc: 860 mb
gcsys: 10 mb
heap inuse: 860 mb
heap object: 19531 k
gc: 14
pause: 566.08µs
cost: 1.821569192s
quicklist
entries: 20000000
alloc: 258 mb
gcsys: 5 mb
heap inuse: 258 mb
heap object: 4274 k
gc: 14
pause: 508.364µs
cost: 1.447244873s
内部设计
listpack 是一种紧凑的、顺序的、以字节为单位的数据结构，针对 ziplist 进行了一些关键改进。下面是 quicklist 的架构图：
首先，listpack 中不再保存前一个 entry 的长度 previous_entry_length，而是在 entry 的末尾通过 逆序变长编码 保存当前 entry 的总长度 entry_len，在保证 listpack 反向遍历功能的同时，解决了级联更新问题。
当我们想要在一个字节数组 []byte 中编码多个元素，常见的方法就是以 [sizeData0, data0, sizeData1, data1, ...] 的方式，交替编码元素长度和元素内容。为了减少小整型的内存占用，编码 sizeData 时通常使用 可变长度编码，这也是 protobuf 中的编码格式。
当我们使用如上编码格式时，想要进行正序解码 [data0, data1, ..., dataN]，只需要交替解码长度和实际内容，便可读出所有元素。但想要支持逆序遍历元素，我们还需要存储额外的信息。
逆序变长编码 是 listpack 中的关键设计，这种编码方式存储 listpack 中的各个条目的总长度 entry_len，以支持反向遍历。逆序变长编码允许在不知道编码全长的情况下，从后向前 遍历元素 [dataN, ..., data1, data0]。
在解码时，首先指针移动到 listpack 末尾 (1)，逆序变长解码出 entry_len，并根据 entry_len 将指针偏移到 (2) 位置，正序变长解码出 data_len, 再将指针偏移到 (3)，获取实际内容 data，一个条目就解码完成，将指针移动到 (2) 进行下一个条目的解码。
关键代码
下面是 xgzlucario/quicklist 中的结构体定义，以及 listpack 正序遍历 和 逆序遍历 部分的关键代码。
结构体定义
QuickList：master/list.go
// +------------------------------ QuickList -----------------------------+
// | +-----------+ +-----------+ +-----------+ |
// head --- | listpack0 | <-> | listpack1 | <-> ... <-> | listpackN | --- tail
// +-----------+ +-----------+ +-----------+
//
// QuickList is double linked listpack.
type QuickList struct {
mu sync.RWMutex
head, tail *ListPack
}
ListPack：master/listpack.go
// ListPack is a lists of strings serialization format on Redis.
/*
ListPack data content:
+--------+--------+-----+--------+
| entry0 | entry1 | ... | entryN |
+--------+--------+-----+--------+
|
entry0 content:
+------------+--------------+---------------------+
| data_len | data | entry_len |
+------------+--------------+---------------------+
|<- varint ->|<- data_len ->|<- varint(reverse) ->|
|<------- entry_len ------->|
Using this structure, it is fast to iterate from both sides.
*/
type ListPack struct {
size uint32
data []byte
prev, next *ListPack
}
listpack 双向遍历
// lpIterator 迭代器
type lpIterator func(data []byte, entryStartPos, entryEndPos int) (stop bool)
// iterFront 正序遍历
func (lp *ListPack) iterFront(start, end int, f lpIterator) {
// ...
var index int
for i := 0; i < end && index < len(lp.data); i++ {
//
// index dataStartPos dataEndPos indexNext
// | | | |
// +------------+--------------+---------------------+-----+
// --> | data_len | data | entry_len | ... |
// +------------+--------------+---------------------+-----+
// |<--- n ---->|<- data_len ->|<-- size_entry_len ->|
//
dataLen, n := binary.Uvarint(lp.data[index:])
indexNext := index + n + int(dataLen) + +SizeUvarint(dataLen+uint64(n))
if i >= start {
dataStartPos := index + n
dataEndPos := dataStartPos + int(dataLen)
data := lp.data[dataStartPos:dataEndPos]
if f(data, index, indexNext) {
return
}
}
index = indexNext
}
}
// iterBack 逆序遍历
func (lp *ListPack) iterBack(start, end int, f lpIterator) {
// ...
var index = len(lp.data)
for i := 0; i < end && index > 0; i++ {
//
// indexNext dataStartPos dataEndPos index
// | | | |
// +-----+------------+--------------+---------------------+
// | ... | data_len | data | entry_len | <--
// +-----+------------+--------------+---------------------+
// |<--- n ---->|<- data_len ->|<-- size_entry_len ->|
// |<------ entry_len -------->|
//
entryLen, sizeEntryLen := uvarintReverse(lp.data[:index])
indexNext := index - int(entryLen) - sizeEntryLen
if i >= start {
dataLen, n := binary.Uvarint(lp.data[indexNext:])
dataStartPos := indexNext + n
dataEndPos := dataStartPos + int(dataLen)
data := lp.data[dataStartPos:dataEndPos]
if f(data, indexNext, index) {
return
}
}
index = indexNext
}
}
编解码
// 编码 reverse=true 表示逆序
func appendUvarint(b []byte, n int, reverse bool) []byte {
if !reverse {
return binary.AppendUvarint(b, uint64(n))
}
before := len(b)
b = binary.AppendUvarint(b, uint64(n))
after := len(b)
if after-before > 1 {
slices.Reverse(b[before:])
}
return b
}
// 逆序解码
// uvarintReverse is the reverse version from binary.Uvarint.
func uvarintReverse(buf []byte) (uint64, int) {
var x uint64
var s uint
for i := range buf {
b := buf[len(buf)-1-i]
if i == binary.MaxVarintLen64 {
return 0, -(i + 1) // overflow
}
if b < 0x80 {
if i == binary.MaxVarintLen64-1 && b > 1 {
return 0, -(i + 1) // overflow
}
return x | uint64(b)<<s, i + 1
}
x |= uint64(b&0x7f) << s
s += 7
}
return 0, 0
}