Redis之ziplist数据结构

0.前言

redis初始创建hash表,有序集合，链表时, 存储结构采用一种ziplist的存储结构, 这种结构内存排列更紧密, 能提高访存性能. 本文介绍ziplist数据结构

1.ziplist存储结构

ziplist并没有定义明确的结构体, 根据存储结构我们可以定义ziplist如下, 只是进行演示使用.其中content字段存储实际的实体内容, 实体

typedef struct ziplist{
     /*ziplist分配的内存大小*/
     uint32_t bytes;
     /*达到尾部的偏移量*/
     uint32_t tail_offset;
     /*存储元素实体个数*/
     uint16_t length;
     /*存储内容实体元素*/
     unsigned char* content[];
     /*尾部标识*/
     unsigned char end;
}ziplist;

/*元素实体所有信息, 仅仅是描述使用, 内存中并非如此存储*/
typedef struct zlentry {
     /*前一个元素长度需要空间和前一个元素长度*/
    unsigned int prevrawlensize, prevrawlen;
     /*元素长度需要空间和元素长度*/
    unsigned int lensize, len;
     /*头部长度即prevrawlensize + lensize*/
    unsigned int headersize;
     /*元素内容编码*/
    unsigned char encoding;
     /*元素实际内容*/
    unsigned char *p;
}zlentry;

                                     ziplist内存布局
|-----------|-----------|----------|---------------------------------------------------|---|
    bytes      offset      length  content         {zlentry, zlentry ... ...}           end

2.1 zlentry之prevrawlen编码

zlentry中prevrawlen进行了压缩编码, 如果字段小于254, 则直接用一个字节保存, 如果大于254字节, 则使用5个字节进行保存, 第一个字节固定值254, 后四个字节保存实际字段值. zipPrevEncodeLength函数是对改字段编码的函数, 我们可以通过此函数看下编码格式.

/*prevrawlen字段进行编码函数*/
static unsigned int zipPrevEncodeLength(unsigned char *p, unsigned int len) {
     /*
     *ZIP_BIGLEN值为254, 返回值表示len所占用的空间大小, 要么1要么5
     */
    if (p == NULL) {
        return (len < ZIP_BIGLEN) ? 1 : sizeof(len)+1;
    } else {
          /*len小于254直接用一个字节保存*/
        if (len < ZIP_BIGLEN) {
            p[0] = len;
            return 1;
        } else {
               /*大于254,第一个字节赋值为254, 后四个字节保存值*/
            p[0] = ZIP_BIGLEN;
            memcpy(p+1,&len,sizeof(len));
            memrev32ifbe(p+1);
            return 1+sizeof(len);
        }
    }
}

2.2 zlentry之len编码

zlentry中len字段配合encoding字段进行了编码, 尽量压缩字段长度, 减少内存使用. 如果实体内容被编码成整数, 则长度默认为1, 如果实体内容被编码为字符串, 则会根据不同长度进行不同编码.编码原则是第一个字节前两个bit位标识占用空间长度, 分别有以下几种, 后面紧跟着存储实际值.

/*字符串编码标识使用了最高2bit位 */
#define ZIP_STR_06B (0 << 6)  //6bit
#define ZIP_STR_14B (1 << 6)  //14bit
#define ZIP_STR_32B (2 << 6)  //32bit

/*zlentry中len字段进行编码过程*/
static unsigned int zipEncodeLength(unsigned char *p, unsigned char encoding, unsigned int rawlen) {
    unsigned char len = 1, buf[5];

    if (ZIP_IS_STR(encoding)) {
        /*
          *6bit可以存储, 占用空间为1个字节, 值存储在字节后6bit中.
          */
        if (rawlen <= 0x3f) {
            if (!p) return len;
            buf[0] = ZIP_STR_06B | rawlen;
        } else if (rawlen <= 0x3fff) {
            len += 1;
            if (!p) return len;
               /*14bit可以存储, 置前两个bit位为ZIP_STR_14B标志 */
            buf[0] = ZIP_STR_14B | ((rawlen >> 8) & 0x3f);
            buf[1] = rawlen & 0xff;
        } else {
            len += 4;
            if (!p) return len;
            buf[0] = ZIP_STR_32B;
            buf[1] = (rawlen >> 24) & 0xff;
            buf[2] = (rawlen >> 16) & 0xff;
            buf[3] = (rawlen >> 8) & 0xff;
            buf[4] = rawlen & 0xff;
        }
    } else {
        /* 内容编码为整型, 长度默认为1*/
        if (!p) return len;
        buf[0] = encoding;
    }

    /* Store this length at p */
    memcpy(p,buf,len);
    return len;
}

2.3 zlentry之encoding和p编码

zlentry中encoding和p表示元素编码和内容, 下面分析下具体编码规则, 可以看到这里对内存节省真是到了魔性的地步. encoding是保存在len字段第一个字节中, 第一个字节最高2bit标识字符串编码, 5和6bit位标识是整数编码, 解码时直接从第一个字节中获取编码信息.

/* 整数编码标识使用了5和6bit位 */
#define ZIP_INT_16B (0xc0 | 0<<4)  //16bit整数
#define ZIP_INT_32B (0xc0 | 1<<4)  //32bit整数
#define ZIP_INT_64B (0xc0 | 2<<4)  //64bit整数
#define ZIP_INT_24B (0xc0 | 3<<4)  //24bit整数
#define ZIP_INT_8B 0xfe            //8bit整数

#define ZIP_INT_IMM_MASK 0x0f
#define ZIP_INT_IMM_MIN 0xf1    /* 11110001 */
#define ZIP_INT_IMM_MAX 0xfd    /* 11111101 */

static int zipTryEncoding(unsigned char *entry, unsigned int entrylen, long long *v, unsigned char *encoding) {
    long long value;
    if (entrylen >= 32 || entrylen == 0) return 0;
   
    if (string2ll((char*)entry,entrylen,&value)) {
        /* 0-12之间的值, 直接在保存在了encoding字段中, 其他根据值大小, 直接设置为相应的编码*/
        if (value >= 0 && value <= 12) {
            *encoding = ZIP_INT_IMM_MIN+value;
        } else if (value >= INT8_MIN && value <= INT8_MAX) {
            *encoding = ZIP_INT_8B;
        } else if (value >= INT16_MIN && value <= INT16_MAX) {
            *encoding = ZIP_INT_16B;
        } else if (value >= INT24_MIN && value <= INT24_MAX) {
            *encoding = ZIP_INT_24B;
        } else if (value >= INT32_MIN && value <= INT32_MAX) {
            *encoding = ZIP_INT_32B;
        } else {
            *encoding = ZIP_INT_64B;
        }
        *v = value;
        return 1;
    }
    return 0;
}

3.添加元素

添加元素分为两种方式,可以使用ziplistPush函数向头部或尾部追加元素, 可以使用ziplistInsert向指定位置插入元素

/*push元素, 添加到ziplist头部或者添加到尾部*/
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where) {
    unsigned char *p;
    p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
    return __ziplistInsert(zl,p,s,slen);
}

/* 插入元素, 向指定的位置p插入元素*/
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {
    return __ziplistInsert(zl,p,s,slen);
}

/* 向指定位置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; /* initialized to avoid warning. Using a value
                                    that is easy to see if for some reason
                                    we use it uninitialized. */
    zlentry tail;

    /* 判断是否是在尾部插入*/
    if (p[0] != ZIP_END) {
          /*取出prevlensize和prevlen值, 编码格式上面已经讲过*/
        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;
    }
    /* reqlen是元素需要分配内存空间大小, 需要加上前置元素长度占用长度, 当前元素长度字段*/
    reqlen += zipPrevEncodeLength(NULL,prevlen);
    reqlen += zipEncodeLength(NULL,encoding,slen);

    /* 插入位置不是最后位置, 则需要计算出下一个元素保存本元素prevlen字段空间是否足够, 不够时计算出欠缺的差值 */
    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;

    /* realloc重新分配内存 */
    offset = p-zl;
    zl = ziplistResize(zl,curlen+reqlen+nextdiff);
    p = zl+offset;

    /*更新tailoffset字段值*/
    if (p[0] != ZIP_END) {
        /* 移动p原有位置和后面的内容到新的位置 */
        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);

        /* 修改下一个元素中保存待插入元素的长度prevlen字段*/
        zipPrevEncodeLength(p+reqlen,reqlen);

        /* 更新尾部位置字段 */
        ZIPLIST_TAIL_OFFSET(zl) =
            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);

        /* 假如p后面存在元素, 则需要将尾部位置增加nextdiff */
        tail = zipEntry(p+reqlen);
        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);
    }

    /*
     *nextdiff值非0, 说明下一个元素需要扩展空间存放prevlen字段, 由于下一个元素空间变大, 有可能引起下下一个元素空间需要扩展, 下面函数检测后面元素, 并在需要时重置元素prevlen长度
     */
    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;
}

4.查找元素

查找元素直接从指定位置开始,一个一个查找, 直到找到或者到达尾部.

/* 从位置p开始查找元素, skip表示每查找一次跳过的元素个数*/
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip) {
    int skipcnt = 0;
    unsigned char vencoding = 0;
    long long vll = 0;

    while (p[0] != ZIP_END) {
        unsigned int prevlensize, encoding, lensize, len;
        unsigned char *q;
        
          /*取出元素中元素内容放入q中*/
        ZIP_DECODE_PREVLENSIZE(p, prevlensize);
        ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
        q = p + prevlensize + lensize;

        if (skipcnt == 0) {
            /* 如果元素是字符串编码, */
            if (ZIP_IS_STR(encoding)) {
                if (len == vlen && memcmp(q, vstr, vlen) == 0) {
                    return p;
                }
            } else {
                /*元素是整数编码, 按照整型进行比较*/
                if (vencoding == 0) {
                    if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) {
                        /* 如果无法进行整数编码, 则直接赋值为UCHAR_MAX以后不会在进行整数类型比较*/
                        vencoding = UCHAR_MAX;
                    }
                    assert(vencoding);
                }

                /*如果待查元素是整型编码, 直接进行比较*/
                if (vencoding != UCHAR_MAX) {
                    long long ll = zipLoadInteger(q, encoding);
                    if (ll == vll) {
                        return p;
                    }
                }
            }

            /* 重置跳过元素值 */
            skipcnt = skip;
        } else {
            /* Skip entry */
            skipcnt--;
        }

        /* 移动到下个元素位置 */
        p = q + len;
    }

    return NULL;
}

5.删除元素

删除元素主要通过ziplistDelete和ziplistDeleteRange来进行

/* 删除一个元素*/
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p) {
    size_t offset = *p-zl;
    zl = __ziplistDelete(zl,*p,1);
    *p = zl+offset;
    return zl;
}

/* 删除一段数据 */
unsigned char *ziplistDeleteRange(unsigned char *zl, unsigned int index, unsigned int num) {
     /*根据索引查找出元素位置，下面介绍该函数*/
    unsigned char *p = ziplistIndex(zl,index);
    return (p == NULL) ? zl : __ziplistDelete(zl,p,num);
}

unsigned char *ziplistIndex(unsigned char *zl, int index) {
    unsigned char *p;
    unsigned int prevlensize, prevlen = 0;
     /*传入索引与零比较，比零大则从头部开始查找，比零小则从尾部开始查找*/
    if (index < 0) {
        index = (-index)-1;
        p = ZIPLIST_ENTRY_TAIL(zl);
        if (p[0] != ZIP_END) {
               /*不断取出prevlen值，从后向前开始查找*/
            ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
            while (prevlen > 0 && index--) {
                p -= prevlen;
                ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
            }
        }
    } else {
        p = ZIPLIST_ENTRY_HEAD(zl);
        while (p[0] != ZIP_END && index--) {
            p += zipRawEntryLength(p);
        }
    }
    return (p[0] == ZIP_END || index > 0) ? NULL : p;
}

/* 真正执行删除操作函数*/
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;

    first = zipEntry(p);
    for (i = 0; p[0] != ZIP_END && i < num; i++) {
        p += zipRawEntryLength(p);
        deleted++;
    }

    totlen = p-first.p;
    if (totlen > 0) {
        if (p[0] != ZIP_END) {
            /* 如果删除元素没有到尾部，则需要重新计算删除元素后面元素中prevlen字段占用空间，类似插入时进行的操作 */
            nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
            p -= nextdiff;
            zipPrevEncodeLength(p,first.prevrawlen);

            /* 重置尾部偏移量 */
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);

            /* 如果删除元素没有到尾部，尾部偏移量需要加上nextdiff偏移量 */
            tail = zipEntry(p);
            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 {
            /* 如果删除的元素到达尾部，则不需要移动*/
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe((first.p-zl)-first.prevrawlen);
        }

        /* 重置ziplist空间 */
        offset = first.p-zl;
        zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
        ZIPLIST_INCR_LENGTH(zl,-deleted);
        p = zl+offset;

        /* 同样和插入时一样，需要遍历检测删除元素后面的元素prevlen空间是否足够，不足时进行扩展*/
        if (nextdiff != 0)
            zl = __ziplistCascadeUpdate(zl,p);
    }
    return zl;
}