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;
}