listpack
listpack 源码分析(紧凑列表,取代 ziplist)
listpack 是一块连续内存里顺序摆放多个元素的序列化结构,没有结构体、没有指针,Redis 7 起全面取代 ziplist,作为 Hash/ZSet/List/Stream 在「元素少且短」时的底层编码。
它对 ziplist 的核心修正只有一处:每个 entry 尾部记「自己的长度」(backlen)而非「前驱的长度」(prevlen),一条 entry 变长绝不牵连邻居,从根上消除了 ziplist 最坏 O(N^2) 的连锁更新。
整体布局:<tot-bytes:4B> <num-elements:2B> <entry>...<entry> <0xFF>,头部固定 6 字节,尾部 1 字节 EOF。
单个 entry 布局:<encoding> <data> <backlen:1~5B>,其中 backlen 记录 encoding+data 的总字节数。
布局与编码常量
代码块收起展开
// 基于本地 Redis 仓 (unstable), src/listpack.c
#define LP_HDR_SIZE 6 /* 32 bit total len + 16 bit number of elements. */ // 头部比 ziplist 的 11 字节还小：没有 tail 偏移量，找尾靠 EOF 前的 backlen 反推
#define LP_HDR_NUMELE_UNKNOWN UINT16_MAX // 元素数字段只有 2 字节，计数涨到 65535 即变 UNKNOWN，lpLength 退化为全扫
#define LP_MAX_INT_ENCODING_LEN 9
#define LP_MAX_BACKLEN_SIZE 5
// ...
#define LP_ENCODING_7BIT_UINT 0 // 首字节 0xxxxxxx：0~127 小整数，数值直接就是 encoding 字节，data 长度为 0
// ...
#define LP_ENCODING_6BIT_STR 0x80 // 10xxxxxx：长度 <64 的短串，长度塞进首字节低 6 位，只花 1 字节头
// ...
#define LP_ENCODING_13BIT_INT 0xC0 // 110xxxxx + 1B：13 位整数
// ...
#define LP_ENCODING_12BIT_STR 0xE0 // 1110xxxx + 1B：长度 <4096 的串
// ...
#define LP_ENCODING_16BIT_INT 0xF1 // 0xF1~0xF4：16/24/32/64 位整数，按数值大小逐级升格
#define LP_ENCODING_24BIT_INT 0xF2
#define LP_ENCODING_32BIT_INT 0xF3
#define LP_ENCODING_64BIT_INT 0xF4
// ...
#define LP_ENCODING_32BIT_STR 0xF0 // 0xF0 + 4B 长度：大字符串兜底
// ...
#define LP_EOF 0xFF // 结束哨兵，正向遍历撞到 0xFF 即结束
#define lpGetTotalBytes(p) (((uint32_t)(p)[0]<<0) | \
((uint32_t)(p)[1]<<8) | \
((uint32_t)(p)[2]<<16) | \
((uint32_t)(p)[3]<<24)) // 手工按小端拼字节：不依赖机器字节序，这块内存可以原样进 RDB/复制流
#define lpGetNumElements(p) (((uint32_t)(p)[4]<<0) | \
((uint32_t)(p)[5]<<8))
// ...
unsigned char *lpNew(size_t capacity) {
unsigned char *lp = lp_malloc(capacity > LP_HDR_SIZE+1 ? capacity : LP_HDR_SIZE+1);
if (lp == NULL) return NULL;
lpSetTotalBytes(lp,LP_HDR_SIZE+1);
lpSetNumElements(lp,0);
lp[LP_HDR_SIZE] = LP_EOF;
return lp; // 空 listpack 一共 7 字节；对比 Java 一个空 HashMap 光对象头+table 引用就不止这个数
}元素编码与 backlen
写入前先「试算」:元素能不能编成整数、编码后多少字节,全部算清了才动内存。
代码块收起展开
// 基于本地 Redis 仓 (unstable), src/listpack.c
static inline int lpEncodeGetType(unsigned char *ele, uint32_t size, unsigned char *intenc, uint64_t *enclen) {
int64_t v;
if (lpStringToInt64((const char*)ele, size, &v)) { // 严格解析：前导零/空格都不算数字，保证整数<->字符串能无损往返
lpEncodeIntegerGetType(v, intenc, enclen);
return LP_ENCODING_INT; // 内容是 "123" 的字符串会被自动降成整数编码存
} else {
if (size < 64) *enclen = 1+size;
else if (size < 4096) *enclen = 2+size;
else *enclen = 5+(uint64_t)size;
return LP_ENCODING_STRING;
}
}
static inline void lpEncodeIntegerGetType(int64_t v, unsigned char *intenc, uint64_t *enclen) {
if (v >= 0 && v <= 127) {
/* Single byte 0-127 integer. */
if (intenc != NULL) intenc[0] = v;
if (enclen != NULL) *enclen = 1; // 整条 entry 数据部分就 1 字节
} else if (v >= -4096 && v <= 4095) {
/* 13 bit integer. */
if (v < 0) v = ((int64_t)1<<13)+v; // 负数按 2^13 折叠成无符号存，解码端超过一半再减回去
if (intenc != NULL) {
intenc[0] = (v>>8)|LP_ENCODING_13BIT_INT;
intenc[1] = v&0xff;
}
if (enclen != NULL) *enclen = 2;
} else if (v >= -32768 && v <= 32767) {
// ... 16/24/32 位分支同一套路，按数值范围逐级升格
} else {
/* 64 bit integer. */
uint64_t uv = v;
if (intenc != NULL) {
intenc[0] = LP_ENCODING_64BIT_INT;
intenc[1] = uv&0xff;
// ... 小端逐字节写满 8 字节
}
if (enclen != NULL) *enclen = 9;
}
}
static inline unsigned long lpEncodeBacklen(unsigned char *buf, uint64_t l) {
if (l <= 127) {
if (buf) buf[0] = l; // 最高位 0 表示「到头了」，绝大多数小 entry 一字节搞定
return 1;
} else if (l < 16383) { // 真源就是 <而非 <=：16383 本可用 2 字节却走 3 字节，无害 off-by-one，因为 lpEncodeBacklenBytes 用同一套条件，两端自洽
if (buf) {
buf[0] = l>>7; // 高位字节在左、不带标志位
buf[1] = (l&127)|128; // 低位字节在右、置 128 位表示「左边还有」——这套标志位方向是为从右往左读设计的
}
return 2;
} else if (l < 2097151) {
// ... 3/4/5 字节分支同理，每字节 7 位有效载荷
}
}
static inline uint64_t lpDecodeBacklen(unsigned char *p) {
uint64_t val;
/* Fast path: single byte (most common for small entries <= 127 bytes) */
if (likely(!(p[0] & 128))) { // p 指向 entry 的最后一个字节，从右往左解析
return p[0] & 127;
}
/* Two bytes */
val = (uint64_t)(p[0] & 127);
if (!(p[-1] & 128)) {
return val | ((uint64_t)(p[-1] & 127) << 7);
}
// ... 三/四/五字节继续向左取；超过 5 字节说明数据已损坏，返回 UINT64_MAX
}双向遍历
正向靠 encoding 自描述长度,反向靠 backlen,两个方向都不需要额外索引。
代码块收起展开
// 基于本地 Redis 仓 (unstable), src/listpack.c
static inline uint32_t lpCurrentEncodedSizeUnsafe(unsigned char *p) {
if (LP_ENCODING_IS_7BIT_UINT(p[0])) return 1; // 看首字节就能算出整条 entry（不含 backlen）的长度
if (LP_ENCODING_IS_6BIT_STR(p[0])) return 1+LP_ENCODING_6BIT_STR_LEN(p);
// ...
if (p[0] == LP_EOF) return 1;
return 0;
}
static inline unsigned char *lpSkip(unsigned char *p) {
unsigned long entrylen = lpCurrentEncodedSizeUnsafe(p);
entrylen += lpEncodeBacklenBytes(entrylen); // backlen 自身占几字节不用存：由 entrylen 反推，编解码两端算出来必然一致
p += entrylen;
return p;
}
unsigned char *lpNext(unsigned char *lp, unsigned char *p) {
assert(p);
p = lpSkip(p);
if (p[0] == LP_EOF) return NULL;
lpAssertValidEntry(lp, lpBytes(lp), p); // 每步都校验不越界：listpack 可能来自 RDB/主从流，不能信任内容
return p;
}
unsigned char *lpPrev(unsigned char *lp, unsigned char *p) {
assert(p);
if (p-lp == LP_HDR_SIZE) return NULL; // 已是第一个元素
p--; /* Seek the first backlen byte of the last element. */ // 退一步正好落在前一条 entry 的 backlen 末字节上
uint64_t prevlen = lpDecodeBacklen(p);
prevlen += lpEncodeBacklenBytes(prevlen);
p -= prevlen-1; /* Seek the first byte of the previous entry. */
lpAssertValidEntry(lp, lpBytes(lp), p);
return p;
}
unsigned char *lpLast(unsigned char *lp) {
unsigned char *p = lp+lpGetTotalBytes(lp)-1; /* Seek EOF element. */ // 头部没存 tail 偏移，跳到 EOF 再 lpPrev 一步到位
return lpPrev(lp,p); /* Will return NULL if EOF is the only element. */
}
unsigned long lpLength(unsigned char *lp) {
uint32_t numele = lpGetNumElements(lp);
if (numele != LP_HDR_NUMELE_UNKNOWN) return numele; // 计数器没溢出就是 O(1)
// ... 超过 65535 时全扫计数；扫完若数量回落到范围内，顺手把缓存写回头部
}增删改统一入口 lpInsert
插入、替换、删除三种操作共用这一个函数:删除被建模为「用零长元素替换」,LP_AFTER 被归一成「在下一条之前 LP_BEFORE」,主体只处理两种情况。
代码块收起展开
// 基于本地 Redis 仓 (unstable), src/listpack.c
unsigned char *lpInsert(unsigned char *lp, unsigned char *elestr, unsigned char *eleint,
uint32_t size, unsigned char *p, int where, unsigned char **newp)
{
unsigned char intenc[LP_MAX_INT_ENCODING_LEN];
unsigned char backlen[LP_MAX_BACKLEN_SIZE];
uint64_t enclen; /* The length of the encoded element. */
int delete = (elestr == NULL && eleint == NULL); // 两个元素指针都不传 = 删除
if (delete) where = LP_REPLACE; // 删除 = 用零长元素替换，操作归一
if (where == LP_AFTER) {
p = lpSkip(p); // AFTER = 在下一条（可能是 EOF）之前插入，再归一
where = LP_BEFORE;
ASSERT_INTEGRITY(lp, p);
}
/* Store the offset of the element 'p', so that we can obtain its
* address again after a reallocation. */
unsigned long poff = p-lp; // realloc 可能把整块内存搬家，裸指针会悬空，必须换算成偏移量
int enctype;
if (elestr) {
enctype = lpEncodeGetType(elestr,size,intenc,&enclen); // 试算阶段：只求类型和长度，暂不写入
if (enctype == LP_ENCODING_INT) eleint = intenc;
} else if (eleint) {
enctype = LP_ENCODING_INT;
enclen = size; /* 'size' is the length of the encoded integer element. */
} else {
enctype = -1;
enclen = 0;
}
unsigned long backlen_size = (!delete) ? lpEncodeBacklen(backlen,enclen) : 0;
uint64_t old_listpack_bytes = lpGetTotalBytes(lp);
uint32_t replaced_len = 0;
if (where == LP_REPLACE) {
replaced_len = lpCurrentEncodedSizeUnsafe(p); // 被替换（或被删）的旧 entry 总长
replaced_len += lpEncodeBacklenBytes(replaced_len);
ASSERT_INTEGRITY_LEN(lp, p, replaced_len);
}
uint64_t new_listpack_bytes = old_listpack_bytes + enclen + backlen_size
- replaced_len;
if (new_listpack_bytes > UINT32_MAX) return NULL; // 头部 tot-bytes 只有 4 字节；实际上调用方还有 lpSafeToAdd 的 1GB 安全线
unsigned char *dst = lp + poff; /* May be updated after reallocation. */
/* Realloc before: we need more room. */
if (new_listpack_bytes > old_listpack_bytes &&
new_listpack_bytes > lp_malloc_size(lp)) { // 问分配器要真实容量，jemalloc 的富余空间够用就省一次 realloc
if ((lp = lp_realloc(lp,new_listpack_bytes)) == NULL) return NULL;
dst = lp + poff;
}
if (where == LP_BEFORE) {
memmove(dst+enclen+backlen_size,dst,old_listpack_bytes-poff); // 插入点之后整体右移腾位，O(N) 成本就在这一行
} else { /* LP_REPLACE. */
memmove(dst+enclen+backlen_size,
dst+replaced_len,
old_listpack_bytes-poff-replaced_len); // 按新旧长度差挪动；删除时 enclen=0，等于把后半段拉回来盖掉旧 entry
}
/* Realloc after: we need to free space. */
if (new_listpack_bytes < old_listpack_bytes) { // 缩容必须放在 memmove 之后，先缩会把还没挪走的尾部数据截掉
if ((lp = lp_realloc(lp,new_listpack_bytes)) == NULL) return NULL;
dst = lp + poff;
}
/* Store the entry. */
if (newp) {
*newp = dst;
if (delete && dst[0] == LP_EOF) *newp = NULL;
}
if (!delete) {
if (enctype == LP_ENCODING_INT) {
memcpy(dst,eleint,enclen);
} else if (elestr) {
lpEncodeString(dst,elestr,size);
} else {
redis_unreachable();
}
dst += enclen;
memcpy(dst,backlen,backlen_size); // 只写这一条自己的 backlen，前后邻居完全不动——ziplist 的连锁更新在这里被终结
dst += backlen_size;
}
/* Update header. */
if (where != LP_REPLACE || delete) {
uint32_t num_elements = lpGetNumElements(lp);
if (num_elements != LP_HDR_NUMELE_UNKNOWN) { // 计数器已溢出成 UNKNOWN 就不再维护，反正查长度要全扫
if (!delete)
lpSetNumElements(lp,num_elements+1);
else
lpSetNumElements(lp,num_elements-1);
}
}
lpSetTotalBytes(lp,new_listpack_bytes);
// ... (#if 0 包裹的调试代码略：强制每次修改都换新地址，用来揪出忘记更新 lp 指针的调用方)
return lp;
}原理串讲
以小 Hash 上一次 HSET 的落点为例。t_hash 层确认还在 listpack 编码后,经 lpAppend 追加 field 和 value:lpAppend 用 lpGetTotalBytes(lp)-1 直接定位到 EOF 字节,然后调 lpInsert(lp, ele, NULL, size, eofptr, LP_BEFORE, NULL)——「追加」就是「插在 EOF 之前」,不需要任何特殊分支。
进入 lpInsert 后第一步是 lpEncodeGetType:先用 lpStringToInt64 严格解析,“1024” 这样的字符串会被编成 13 位整数只占 2 字节,而不是按字符串编码的 1+4=5 字节;解析失败才按 6/12/32 位长度前缀的字符串编码。
这一步只算不写,拿到 enclen 后再用 lpEncodeBacklen 算出 backlen 占几字节,于是新 entry 的总尺寸在动内存之前就完全确定了。
为什么坚持「先试算后落笔」?因为 listpack 是一整块连续内存,扩容(realloc)和腾位(memmove)都是重操作,只有预先知道精确尺寸,才能保证每次插入至多一次 realloc、恰好一次 memmove,而不是边写边发现不够再挪第二次。
接下来是连续内存最容易踩的坑:lp_realloc 可能把整块内存搬到新地址,所以插入点指针 p 在 realloc 前先换算成偏移量 poff 保存,realloc 后用 lp + poff 重建。
扩容与缩容的时机也刻意不对称——变大时「先 realloc 再 memmove」(不然没地方挪),变小时「先 memmove 再 realloc」(不然尾部数据先被截掉)。
腾好位后依次写入 encoding+data(整数走 memcpy 预编码好的 intenc,字符串走 lpEncodeString)和这一条自己的 backlen,最后更新头部计数与总字节数。
为什么 backlen 记「自己的长度」就能消除连锁更新?对比 ziplist:它的每个 entry 头部存前驱长度 prevlen,前驱从小于 254 字节涨到不小于 254 字节时,本 entry 的 prevlen 字段要从 1 字节扩成 5 字节,本 entry 因此变长,又可能触发下一个 entry 扩 prevlen…一次插入最坏引发整条链的重写。
listpack 里每条 entry 只描述自己,任何一条的膨胀收缩都被 lpInsert 的一次 memmove 吸收(整体平移不改变各 entry 内容),编码层面互不耦合。
那把长度放在 entry 尾部,反向遍历怎么知道 backlen 本身占几个字节?看 lpPrev:p-- 落在前一条 entry 的最后一个字节上,lpDecodeBacklen 从这个字节开始向左逐字节取 7 位载荷,靠每字节的最高位判断「左边还有没有」——lpEncodeBacklen 写入时故意让最左(最高位)字节不带 128 标志、其余字节都带,这个方向和常规 varint 相反,正是为了让解析器能从右往左走。
拿到 entry 数据长度 prevlen 后,再用 lpEncodeBacklenBytes(prevlen) 反推 backlen 自身的字节数——因为长度到字节数的映射是确定函数,编码端与解码端算出的结果必然一致,这个「自身长度」就无需存储。
正向遍历则完全不碰 backlen:lpSkip 读首字节 encoding 即知 entry 数据多长,同样反推出 backlen 字节数后跨过去。
一份数据、零冗余索引,支撑了双向 O(1) 步进。
设计取舍
- 连续内存 + 无指针:内存省到极致且缓存友好,代价是定位第 N 个元素 O(N)、增删要 memmove。
所以只在小集合用,由hash-max-listpack-entries(默认 128)、hash-max-listpack-value(默认 64 字节)这类阈值把关,超限即转 hashtable/skiplist,且这个转换是单向的。 - num-elements 只给 2 字节看似抠门,实则自洽:listpack 本来就只服务小集合,65535 的上限在正常配置下永远碰不到;真溢出就置
LP_HDR_NUMELE_UNKNOWN退化为全扫。 - 数字型字符串自动整数化(
lpEncodeGetType里的lpStringToInt64)是纯存储优化,读取时lpGet可用 intbuf 还原成字符串,上层无感知。 - List 数据大了也没有完全放弃 listpack:quicklist 把多个 listpack 用双向链表串起来,单节点内紧凑、节点间用指针,是「省内存」和「改动局部化」的折中。
- ziplist 头部有 tail 偏移字段(11 字节头),listpack 砍掉它(6 字节头),
lpLast改为「跳到 EOF、lpPrev一步回退」——反正 backlen 机制让反向步进本来就是 O(1),专门存 tail 偏移是冗余。