redis ziplist压缩列表的源码分析

         ziplist是redis为了节约内存而开发的，由一系列特殊编码的连续内存块组成的顺序数据结构。一个

ziplist可以包含任意多个节点(entry)，每个节点可以保存一个字节数组或者一个整数值。

一、压缩列表的基本

       ziplist在连续内存块中基本结构如下所示

      <zlbytes><zltail><zllen><entry1><entry2>...<entryN><zlend>

zlbytes：记录整个压缩列表占用的内存字节数(4个字节)

zltail：记录压缩列表表尾节点距离压缩列表的起始位置有多少字节(4个字节)

zllen：记录压缩列表包含的节点数量(2个字节)

entryX：压缩列表中的各个节点(不定)

zlend：特殊值0xFF(十进制255)，用于标记压缩列表的末端(1个字节)

二、压缩列表节点的基本构成

      entry在内存中的顺序结构如下所示

      <prev_entry_len><encoding><content>

  prev_entry_len

       prev_entry_len属性记录了ziplist中前一个节点的长度。prev_entry_len属性的长度可以是1个字节或者5个字节：

1、当前一个节点的的长度小于254字节，prev_entry_len长度为1字节：长度保存在这一字节

2、当前一个节点的的长度大于等于254字节，prev_entry_len为5个字节：第一个字节设置为0xFE，后四个字节用

     于保存前一个字节的长度prev_entry_len用于根据当前节点计算前一节点的初始地址，而ziplist的从尾向头遍历

    操作就是基于此原理。

  encoding    

      encoding属性记录了节点content属性所保存数据的类型和长度：

1、一字节、二字节或者五字节，值的最高位为00，01或者10的字节编码，这样的编码表示content保存的是字节数组，

    数组的长度保存在剩余的其他位。

2、一字节长，值的最位为11，表示整数编码，具体的整数类型对应如下：

    11000000-int16_t,11010000-int32_t,11100000-int64_t,11110000-24位有符号整数

    11111110-8位有符号整数，1111xxxx——xxxx用于保存0-12的整数，因此无需content属性  

 content

     content记录保存节点的值，节点值可以是一个字节数组或者整数。

三、连锁更新

       ziplist中，在进行插入或者删除操作时，操作位置P存放的节点和后面连续若干节点(e1,e2,e3...)的长度介于

250～253字节之间并且p的prev_entr_len属性为一字节，新的前置节点大于254字节，需要对ziplist执行空间充分

配操作，将p的prev_entr_len从1字节扩展到5个字节才能存储前置节点的长度。这样一来，p的长度也大于254字

节，e1需要扩展，以此类推，需要一直扩展到节点的prev_entry_len符合ziplist的要求。

//由于p长度的变化，需要p开始往后遍历更新后面节点的prev_entry_len属性满足ziplist的要求
static unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p) {
size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), rawlen, rawlensize;
size_t offset, noffset, extra;
unsigned char *np;
zlentry cur, next;
while (p[0] != ZIP_END) {
zipEntry(p, &cur);
rawlen = cur.headersize + cur.len;//下一个节点相对当前节点的偏移量
rawlensize = zipPrevEncodeLength(NULL,rawlen);
/* Abort if there is no next entry. */
if (p[rawlen] == ZIP_END) break;
zipEntry(p+rawlen, &next);//下一个节点的entry
//p的长度不变，不需要进行连锁更新
if (next.prevrawlen == rawlen) break;
if (next.prevrawlensize < rawlensize) {
// 当下一个节点需要更多的字节来存储当前节点当长度，需要空间扩展
offset = p-zl;
extra = rawlensize-next.prevrawlensize;
zl = ziplistResize(zl,curlen+extra);
p = zl+offset;
/* Current pointer and offset for next element. */
np = p+rawlen;
noffset = np-zl;
/* Update tail offset when next element is not the tail element. */
if ((zl+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) != np) {
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+extra);
}
//扩充下一个节点的prevlen属性的字节数
memmove(np+rawlensize,
np+next.prevrawlensize,
curlen-noffset-next.prevrawlensize-1);
zipPrevEncodeLength(np,rawlen);//更新下一个节点的prevlen的值
//遍历到下个节点，更新压缩列表的长度
p += rawlen;
curlen += extra;
} else {
if (next.prevrawlensize > rawlensize) {
/*当下一个节点用来存储prev_entry_len的空间大于用于存储当前节点的长度的字节数，
不需要压缩prev_entry_len的字节数，只更新prev_entry_len的值，避免进一步连锁更新*/
zipPrevEncodeLengthForceLarge(p+rawlen,rawlen);
} else {
zipPrevEncodeLength(p+rawlen,rawlen);
}
break;
}
}
return zl;
}四、压缩列表的底层操作
     插入操作的实现

//节点插入到节点p前面
static unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p,
unsigned char *s, unsigned int slen) {
size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen;
unsigned int prevlensize, prevlen = 0;
size_t offset;
int nextdiff = 0;
unsigned char encoding = 0;
long long value = 123456789;
zlentry tail;
if (p[0] != ZIP_END) {//获取前一个节点的长度
ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
} else {
unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
if (ptail[0] != ZIP_END) {
prevlen = zipRawEntryLength(ptail);
}
}
//查看是否可以被编码成整数
if (zipTryEncoding(s,slen,&value,&encoding)) {
reqlen = zipIntSize(encoding);
} else {
reqlen = slen;
}
//加上prev_entry_len属性和encoding属性的长度
reqlen += zipPrevEncodeLength(NULL,prevlen);
reqlen += zipEncodeLength(NULL,encoding,slen);
//计算p的prev_entry_len属性需要字节数的变化
nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
//realloc zl,并记录p的位置
offset = p-zl;
zl = ziplistResize(zl,curlen+reqlen+nextdiff);
p = zl+offset;
//进行内存迁移
if (p[0] != ZIP_END) {
//p的prev_entry_len的属性内存大小和值的修改
memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);
zipPrevEncodeLength(p+reqlen,reqlen);
ZIPLIST_TAIL_OFFSET(zl) =intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
zipEntry(p+reqlen, &tail);
if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
} else {
/* This element will be the new tail. */
ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
}
//当p的前置节点的长度发送变化,根据p的长度变化，进行连锁更新
if (nextdiff != 0) {
offset = p-zl;
zl = __ziplistCascadeUpdate(zl,p+reqlen);
p = zl+offset;
}
//写入新插入的节点的内容
p += zipPrevEncodeLength(p,prevlen);
p += zipEncodeLength(p,encoding,slen);
if (ZIP_IS_STR(encoding)) {
memcpy(p,s,slen);
} else {
zipSaveInteger(p,value,encoding);
}
ZIPLIST_INCR_LENGTH(zl,1);
return zl;
}
     删除操作的实现
//从地址p开始删除num个节点
static unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsigned int num) {
unsigned int i, totlen, deleted = 0;
size_t offset;
int nextdiff = 0;
zlentry first, tail;
zipEntry(p, &first);
for (i = 0; p[0] != ZIP_END && i < num; i++) {
p += zipRawEntryLength(p);
deleted++;
}
//p为删除的num个节点的后一个节点的起始地址
totlen = p-first.p;
if (totlen > 0) {
if (p[0] != ZIP_END) {
//计算p的prev_entry_len属性需要字节数的变化
nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
p -= nextdiff;
//修改p的prev_entry_len
zipPrevEncodeLength(p,first.prevrawlen);
//更新尾节点的偏移量
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);
//根据nextdiff更新尾节点的偏移量
zipEntry(p, &tail);
if (p[tail.headersize+tail.len] != ZIP_END) {
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
//将删除的节点后面的节点复制到开始删除的的起始位置
memmove(first.p,p,intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1);
} else {
/* The entire tail was deleted. No need to move memory. */
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe((first.p-zl)-first.prevrawlen);
}
/* Resize and update length */
offset = first.p-zl;
zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
ZIPLIST_INCR_LENGTH(zl,-deleted);
p = zl+offset;
if (nextdiff != 0)
//当p的前置节点的长度发送变化,根据p的长度变化，进行连锁更新
zl = __ziplistCascadeUpdate(zl,p);
}
return zl;
}

五、压缩列表的API
unsigned char *ziplistNew(void);
//创建一个ziplist
unsigned char *ziplistMerge(unsigned char **first, unsigned char **second);
//合并连个压缩列表
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where);
//创建一个包含给定值的节点，添加到表头或表尾
unsigned char *ziplistIndex(unsigned char *zl, int index);
//返回指定的索引上的节点的地址
unsigned char *ziplistNext(unsigned char *zl, unsigned char *p);
//返回指定地址的下一个节点
unsigned char *ziplistPrev(unsigned char *zl, unsigned char *p);
//返回指定地址的前一个节点
unsigned int ziplistGet(unsigned char *p, unsigned char **sval,
unsigned int *slen, long long *lval);//获取指定地址上的节点的值
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p,
unsigned char *s, unsigned int slen);//创建一个包含给定值的节点，添加到指定地址后面
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p);
//删除指定地址上的节点
unsigned char *ziplistDeleteRange(unsigned char *zl, int index, unsigned int num);
//删除指定索引上连续多个节点
unsigned int ziplistCompare(unsigned char *p, unsigned char *s, unsigned int slen);
//地址p的节点是字符串，与*s进行字符串比较；地址p的节点是整数，将*s转成整数比较
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr,
unsigned int vlen, unsigned int skip);//
unsigned int ziplistLen(unsigned char *zl);
//返回压缩列表目前的节点数
size_t ziplistBlobLen(unsigned char *zl);
//返回目前压缩列表占用的内存字节数