Redis源碼閱讀筆記-快速列表

快速列表

快速列表(quicklist)是由壓縮列表(ziplist)組成的一個雙向鏈表，鏈表中，每個節點都是以壓縮列表(ziplist)的結構保存。

在 Redis3.2 後加入的新數據結構，在列表鍵中取代了雙向鏈表的做用。

特色

• 雙向鏈表（list）在插入節點刪除節點上效率高，可是每一個節點不連續，容易產生內存碎片。
• 壓縮列表（ziplist）是一段連續的內存，但不利於修改，插入刪除麻煩，複雜度高，頻繁申請釋放內存。

快速列表綜合了雙向列表和壓縮列表的優勢，既有鏈表頭尾插入相對便捷，又有連續內存存儲的優勢。

代碼結構

快速列表結構

/* quicklist is a 40 byte struct (on 64-bit systems) describing a quicklist.
 * 'count' is the number of total entries.
 * 'len' is the number of quicklist nodes.
 * 'compress' is: -1 if compression disabled, otherwise it's the number
 *                of quicklistNodes to leave uncompressed at ends of quicklist.
 * 'fill' is the user-requested (or default) fill factor. */
typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* total count of all entries in all ziplists */
    unsigned long len;          /* number of quicklistNodes */
    int fill : 16;              /* fill factor for individual nodes */
    unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
} quicklist;

快速列表的結構:

• *head: 指向快速列表(quicklist)頭節點
• *tail: 指向快速列表(quicklist)尾節點
• count : 列表中，數據項的個數
• len: 列表中，節點(quicklistNode)的個數
• fill: 每一個節點中壓縮列表(ziplist)的大小限制，能夠經過list-max-ziplist-size設定
• compress: 節點的壓縮深度設置，能夠經過list-compress-depth設定

list-max-ziplist-size

list-max-ziplist-size的默認配置是OBJ_LIST_MAX_ZIPLIST_SIZE -2，參數的範圍是（1 ~ 2^15）和（-1 ~ -5）：

• 當參數爲正數，表示按照數據項個數限制每一個節點的元素個數。
• -1 每一個節點的ziplist字節大小不能超過4kb
• -2 每一個節點的ziplist字節大小不能超過8kb
• -3 每一個節點的ziplist字節大小不能超過16kb
• -4 每一個節點的ziplist字節大小不能超過32kb
• -5 每一個節點的ziplist字節大小不能超過64kb

PS: 具體代碼參看quicklist.c中的int _quicklistNodeSizeMeetsOptimizationRequirement(const size_t sz, const int fill)函數，sz爲節點中ziplist的字節數，fill爲快速列表的屬性fill。

list-compress-depth

list-compress-depth的默認配置是OBJ_LIST_COMPRESS_DEPTH 0，取值範圍是（0 ~ 2^16）:

• 0 特殊值，表示不壓縮
• 1 表示quicklist兩端各有1個節點不壓縮，中間的節點壓縮
• 2 表示quicklist兩端各有2個節點不壓縮，中間的節點壓縮
• 3 表示quicklist兩端各有3個節點不壓縮，中間的節點壓縮
• 以此類推

快速列表節點結構

/* Node, quicklist, and Iterator are the only data structures used currently. */

/* quicklistNode is a 32 byte struct describing a ziplist for a quicklist.
 * We use bit fields keep the quicklistNode at 32 bytes.
 * count: 16 bits, max 65536 (max zl bytes is 65k, so max count actually < 32k).
 * encoding: 2 bits, RAW=1, LZF=2.
 * container: 2 bits, NONE=1, ZIPLIST=2.
 * recompress: 1 bit, bool, true if node is temporarry decompressed for usage.
 * attempted_compress: 1 bit, boolean, used for verifying during testing.
 * extra: 12 bits, free for future use; pads out the remainder of 32 bits */
typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *zl;
    unsigned int sz;             /* ziplist size in bytes */
    unsigned int count : 16;     /* count of items in ziplist */
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    unsigned int recompress : 1; /* was this node previous compressed? */
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode;

快速列表的節點結構:

• *prev：指向上一個節點
• *next：指向下一個節點
• *zl： 指向數據的指針，若是沒有被壓縮則是指向壓縮列表(ziplist)；若是被壓縮了，則是指向quicklistLZF
• sz： zl中指向的結構所佔用的字節數
• count： 節點中的元素項個數，最大爲65536
• encoding： 編碼方式 RAW=1, LZF=2 （1表示壓縮列表，2表示quicklistLZF）
• container： 預留字段，存放數據的方式，1-NONE，2-ziplist(代碼中並無使用到，僅僅設置了默認值)
• recompress： 標識位，1標識臨時解壓中，須要從新壓縮
• attempted_compress： 僅用於測試中
• extra： 預留字段

LZF壓縮數據結構

/* quicklistLZF is a 4+N byte struct holding 'sz' followed by 'compressed'.
 * 'sz' is byte length of 'compressed' field.
 * 'compressed' is LZF data with total (compressed) length 'sz'
 * NOTE: uncompressed length is stored in quicklistNode->sz.
 * When quicklistNode->zl is compressed, node->zl points to a quicklistLZF */
typedef struct quicklistLZF {
    unsigned int sz; /* LZF size in bytes*/
    char compressed[];
} quicklistLZF;

quicklistLZF 通過LZF算法壓縮後數據保存的結構:

• sz： LZF壓縮後佔用的字節數
• compressed[]：壓縮後的數據

PS: 在quicklist.c中的int __quicklistCompressNode(quicklistNode *node)函數能夠看到，當節點中的壓縮列表的大小小於#define MIN_COMPRESS_BYTES 48時，不會執行LZF壓縮。

表示數據項結構

typedef struct quicklistEntry {
    const quicklist *quicklist;
    quicklistNode *node;
    unsigned char *zi;
    unsigned char *value;
    long long longval;
    unsigned int sz;
    int offset;
} quicklistEntry;

表示quicklist節點中ziplist裏的一個數據項結構:

• *quicklist: 指向所在的快速列表
• *node: 指向所在的節點
• *zi: 指向所在的壓縮列表
• *value: 當前壓縮列表中的節點的字符串值
• longval: 當前壓縮列表中的節點的整數值
• sz: 當前壓縮列表中的節點的字節大小
• offset: 當前壓縮列表中的節點 相對於 壓縮列表 的偏移量

部分代碼解析

• int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) 將大小爲sz的新選項value添加到quicklist的頭中，若是函數返回0表示使用已經存在的節點，若是返回1表示新建頭節點：

  /* Add new entry to head node of quicklist.
  	 *
  	 * Returns 0 if used existing head.
  	 * Returns 1 if new head created. */
  	int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
  	    // orig_head 指向列表頭節點
  	    quicklistNode *orig_head = quicklist->head;
  
  	    // 來之博客: https://blog.csdn.net/terence1212/article/details/53770882 的解析
  	    // likely()是linux提供給程序員的編譯優化方法
  	    // 目的是將「分支轉移」的信息提供給編譯器，這樣編譯器能夠對代碼進行優化，以減小指令跳轉帶來的性能降低
  	    // 此處表示節點沒有滿發生的機率比較大，也就是數據項直接插入到當前節點的可能性大，
  	    // likely()屬於編譯器級別的優化
  	    if (likely(
  	        // 判斷value是否能直接插入到頭節點中，
  	        // 會經過設置的屬性 fill 判斷，
  	        // fill爲正數時，主要判斷 quicklist->head 中元素項的個數
  	        // fill爲負數時，主要判斷 quicklist->head 中的數據大小和新數據項value的大小
  	        _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
  
  	        // 直接插入到 quicklist->head 的壓縮列表中
  	        quicklist->head->zl =
  	            ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
  	        // 更新壓縮列表的大小
  	        quicklistNodeUpdateSz(quicklist->head);
  	    } else {
  	        // 須要新建節點
  
  	        quicklistNode *node = quicklistCreateNode();
  	        // 指向新建壓縮列表
  	        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
  
  	        // 更新節點的大小
  	        quicklistNodeUpdateSz(node);
  	        // 將node插入爲quicklist的表頭
  	        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
  	    }
  	    // 元素項 總數 + 1
  	    quicklist->count++;
  	    quicklist->head->count++;
  	    return (orig_head != quicklist->head);
  	}
  
  	// 判斷長度爲sz的值，是否能插入節點node
  	REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node,
  	                                           const int fill, const size_t sz) {
  	    if (unlikely(!node))
  	        return 0;
  
  	    // ziplist_overhead爲估算插入元素項後，壓縮列表該元素項的頭要佔用的字節數
  
  	    // 計算更新下一個節點的值，「上一個節點長度」所佔用的字節數
  	    int ziplist_overhead;
  	    /* size of previous offset */
  	    if (sz < 254)
  	        ziplist_overhead = 1;
  	    else
  	        ziplist_overhead = 5;
  
  	    // 計算編碼「encoding」所佔用的字節數
  	    /* size of forward offset */
  	    if (sz < 64)
  	        ziplist_overhead += 1;
  	    else if (likely(sz < 16384))
  	        ziplist_overhead += 2;
  	    else
  	        ziplist_overhead += 5;
  
  	    /* new_sz overestimates if 'sz' encodes to an integer type */
  	    unsigned int new_sz = node->sz + sz + ziplist_overhead;
  	    // new_sz爲加入插入後，該壓縮列表的大小
  	    // _quicklistNodeSizeMeetsOptimizationRequirement() 是判斷list-max-ziplist-size的設置的
  	    if (likely(_quicklistNodeSizeMeetsOptimizationRequirement(new_sz, fill)))
  	        return 1;
  	    else if (!sizeMeetsSafetyLimit(new_sz))
  	        return 0;
  	    else if ((int)node->count < fill)
  	        return 1;
  	    else
  	        return 0;
  	}

• void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *entry, void *value, const size_t sz) 將大小爲sz的值value插入到快速列表quicklist指定數據項節點node後:

  void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *entry,
  	                          void *value, const size_t sz) {
  	    // 調用_quicklistInsert()函數插入
  	    // 最後參數1表示，要將 value 插入到 entry後
  	    _quicklistInsert(quicklist, entry, value, sz, 1);
  	}
  
  	/* Insert a new entry before or after existing entry 'entry'.
  	 *
  	 * If after==1, the new value is inserted after 'entry', otherwise
  	 * the new value is inserted before 'entry'. */
  	// 要將大小爲sz的value, 插入到元素項entry的前或後
  	// 1表示插入在後
  	// 其餘表示插入在前
  	REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
  	                                   void *value, const size_t sz, int after) {
  	    // full 是判斷entry所在的節點node中，是否能夠直接插入value，不能夠則置爲1
  	    // full_next 是判斷entry所在的節點的下一個節點，是否能夠直接插入value，不能夠則置爲1
  	    // full_prev 是判斷entry所在的節點的上一個節點，是否能夠直接插入value，不能夠則置爲1
  	    // at_tail 表示entry是否在node中是最後一個元素項
  	    // at_head 表示entry是否在node中是第一個元素項
  	    int full = 0, at_tail = 0, at_head = 0, full_next = 0, full_prev = 0;
  	    int fill = quicklist->fill;
  	    // 獲取entry所在的快速列表節點
  	    quicklistNode *node = entry->node;
  	    quicklistNode *new_node = NULL;
  
  	    if (!node) {
  	        // 若是指定的entry 並無節點，那麼建立一個並插入
  	        // 這裏應該是默認爲，當傳入的`node`爲NULL時，quicklist中是沒有節點的
  	        // 能夠看__quicklistInsertNode()函數中的處理，它並無考慮quicklist->len > 0 且 node 爲 NULL的狀態
  	        /* we have no reference node, so let's create only node in the list */
  	        D("No node given!");
  	        new_node = quicklistCreateNode();
  	        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
  	        __quicklistInsertNode(quicklist, NULL, new_node, after);
  	        new_node->count++;
  	        quicklist->count++;
  	        return;
  	    }
  
  	    /* Populate accounting flags for easier boolean checks later */
  	    // 判斷節點node是否可讓value插入到它的壓縮列表中
  	    if (!_quicklistNodeAllowInsert(node, fill, sz)) {
  	        D("Current node is full with count %d with requested fill %lu",
  	          node->count, fill);
  	        full = 1;
  	    }
  
  	    // 若是插在節點後，並且元素項entry剛好是最後一個元素
  	    // 那麼檢查節點node的下一個節點，是否可讓value插入到它的壓縮列表中
  	    if (after && (entry->offset == node->count)) {
  	        D("At Tail of current ziplist");
  	        at_tail = 1;
  	        if (!_quicklistNodeAllowInsert(node->next, fill, sz)) {
  	            D("Next node is full too.");
  	            full_next = 1;
  	        }
  	    }
  
  	    // 若是插在節點前，並且元素項entry剛好是第一個元素
  	    // 那麼檢查節點node的上一個節點，是否可讓value插入到它的壓縮列表中
  	    if (!after && (entry->offset == 0)) {
  	        D("At Head");
  	        at_head = 1;
  	        if (!_quicklistNodeAllowInsert(node->prev, fill, sz)) {
  	            D("Prev node is full too.");
  	            full_prev = 1;
  	        }
  	    }
  
  	    /* Now determine where and how to insert the new element */
  	    if (!full && after) {
  	        // 壓縮列表能夠插入，且插入entry以後
  	        D("Not full, inserting after current position.");
  	        // 嘗試解壓縮LZF的數據(該函數會判斷node中的數據是否已壓縮，因此直接調用)
  	        quicklistDecompressNodeForUse(node);
  
  	        // 壓縮列表插入
  	        unsigned char *next = ziplistNext(node->zl, entry->zi);
  	        if (next == NULL) {
  	            node->zl = ziplistPush(node->zl, value, sz, ZIPLIST_TAIL);
  	        } else {
  	            node->zl = ziplistInsert(node->zl, next, value, sz);
  	        }
  	        node->count++;
  	        // 更新node的字節數
  	        quicklistNodeUpdateSz(node);
  	        // 從新壓縮
  	        quicklistRecompressOnly(quicklist, node);
  	    } else if (!full && !after) {
  	        // 壓縮列表能夠插入，且插入entry以前
  	        D("Not full, inserting before current position.");
  	        // 嘗試解壓縮LZF的數據(該函數會判斷node中的數據是否已壓縮，因此直接調用)
  	        quicklistDecompressNodeForUse(node);
  
  	        // 壓縮列表插入
  	        node->zl = ziplistInsert(node->zl, entry->zi, value, sz);
  	        node->count++;
  	        // 更新node的字節數
  	        quicklistNodeUpdateSz(node);
  	        // 從新壓縮
  	        quicklistRecompressOnly(quicklist, node);
  	    } else if (full && at_tail && node->next && !full_next && after) {
  	        /* If we are: at tail, next has free space, and inserting after:
  	         *   - insert entry at head of next node. */
  	        // 若是node中壓縮列表沒法插入
  	        // 且 entry 是尾元素項
  	        // 且 node 存在下一個節點
  	        // 且 下一個節點能夠插入
  	        // 且 新元素項是插入到 entry 以後的
  	        D("Full and tail, but next isn't full; inserting next node head");
  	        new_node = node->next;
  	        quicklistDecompressNodeForUse(new_node);
  	        new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_HEAD);
  	        new_node->count++;
  	        quicklistNodeUpdateSz(new_node);
  	        quicklistRecompressOnly(quicklist, new_node);
  	    } else if (full && at_head && node->prev && !full_prev && !after) {
  	        /* If we are: at head, previous has free space, and inserting before:
  	         *   - insert entry at tail of previous node. */
  	        // 若是node中壓縮列表沒法插入
  	        // 且 entry 是頭元素項
  	        // 且 node 存在上一個節點
  	        // 且 上一個節點能夠插入
  	        // 且 新元素項是插入到 entry 以前的
  	        D("Full and head, but prev isn't full, inserting prev node tail");
  	        new_node = node->prev;
  	        quicklistDecompressNodeForUse(new_node);
  	        new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_TAIL);
  	        new_node->count++;
  	        quicklistNodeUpdateSz(new_node);
  	        quicklistRecompressOnly(quicklist, new_node);
  	    } else if (full && ((at_tail && node->next && full_next && after) ||
  	                        (at_head && node->prev && full_prev && !after))) {
  	        /* If we are: full, and our prev/next is full, then:
  	         *   - create new node and attach to quicklist */
  	        // 若是 node 中壓縮列表沒法插入
  	        // （entry在對列尾 且 node存在下一個節點 且 下一個節點壓縮列表沒法插入 且 是插入entry以後）
  	        // 或者
  	        // （entry在對列頭 且 node存在上一個節點 且 上一個節點壓縮列表沒法插入 且 是插入entry以前）
  	        D("\tprovisioning new node...");
  
  	        // 新建一個節點
  	        new_node = quicklistCreateNode();
  	        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
  	        new_node->count++;
  	        quicklistNodeUpdateSz(new_node);
  	        // 新節點插入到node節點的前或者後(由after決定)
  	        __quicklistInsertNode(quicklist, node, new_node, after);
  	    } else if (full) {
  	        /* else, node is full we need to split it. */
  	        /* covers both after and !after cases */
  	        // 以上都不是，並且node中的壓縮列表滿了，則須要將node拆分爲2個node
  	        D("\tsplitting node...");
  	        quicklistDecompressNodeForUse(node);
  	        // 拆分爲2個node
  	        new_node = _quicklistSplitNode(node, entry->offset, after);
  	        // 將vale添加到new_node中
  	        new_node->zl = ziplistPush(new_node->zl, value, sz,
  	                                   after ? ZIPLIST_HEAD : ZIPLIST_TAIL);
  	        new_node->count++;
  	        quicklistNodeUpdateSz(new_node);
  	        // 將new_node插入到node對應文職
  	        __quicklistInsertNode(quicklist, node, new_node, after);
  	        _quicklistMergeNodes(quicklist, node);
  	    }
  
  	    quicklist->count++;
  	}

快速列表API

函數	做用
quicklist *quicklistCreate(void)	建立一個新的快速列表。
quicklist *quicklistNew(int fill, int compress)	使用指定的fill和compress建立一個新的快速列表。
void quicklistSetCompressDepth(quicklist *quicklist, int depth)	設置quicklist的depth，設置後並不會執行壓縮/解壓縮操做，只改變quicklist中的depth屬性值。
void quicklistSetFill(quicklist *quicklist, int fill)	設置quicklist的fill屬性值，設置後並不會對列表進行調整。
void quicklistSetOptions(quicklist *quicklist, int fill, int depth)	設置quicklist的fill和depth 屬性值，設置後並不會對列表進行調整。
void quicklistRelease(quicklist *quicklist)	釋放整個quicklist列表
int quicklistPushHead(quicklist *quicklist, void *value, const size_t sz)	將大小爲sz的新選項value添加到quicklist的頭中，若是函數返回0表示使用已經存在的節點，若是返回1表示新建頭節點。
int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz)	將大小爲sz的新選項value添加到quicklist的尾中，若是函數返回0表示使用已經存在的節點，若是返回1表示新建尾節點。
void quicklistPush(quicklist *quicklist, void *value, const size_t sz,int where)	將大小爲sz的新選項value添加到quicklist的最前或最末，由參數where決定位置，0表示添加到頭，-1表示添加到尾。
void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl)	爲壓縮列表*zl建立一個新的節點，並將這個節點添加到quicklist的尾。
quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist, unsigned char *zl)	在壓縮列表zl中讀取數據項，並將這些數據項添加到quicklist的尾。
quicklist *quicklistCreateFromZiplist(int fill, int compress, unsigned char *zl)	從指定的壓縮列表*zl中，建立一個新的quicklist。
void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *node, void *value, const size_t sz)	將大小爲sz的值value插入到快速列表quicklist指定數據項節點node後
void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *node, void *value, const size_t sz)	將大小爲sz的值value插入到快速列表quicklist指定數據項節點node前
int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data, int sz)	將長度爲sz的數據data插入到快速列表quicklist中的指定位置index
int quicklistDelRange(quicklist *quicklist, const long start, const long stop)	刪除quicklist中指定start ~ end範圍的數據項
quicklist *quicklistDup(quicklist *orig)	複製一個新的快速列表並返回
int quicklistIndex(const quicklist *quicklist, const long long index, quicklistEntry *entry)	返回quicklist中指定位置index數據項的值，並寫入*entry中，若是找到元素則返回1，不然返回0
void quicklistRotate(quicklist *quicklist)	旋轉快速列表quicklist
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data, unsigned int *sz, long long *sval, void *(*saver)(unsigned char *data, unsigned int sz))	在quicklist中彈出(pop)一個元素項，並寫入data中，若是這個元素是long long則寫入sval中，where爲0表示在頭彈出，其餘表示在尾彈出。
int quicklistPop(quicklist *quicklist, int where, unsigned char **data, unsigned int *sz, long long *slong)	在quicklist中彈出(pop)一個元素項，並寫入data中，where爲0表示在頭彈出，其餘表示在尾彈出。
unsigned long quicklistCount(const quicklist *ql)	返回快速列表ql中元素項的個數
int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len)	對比壓縮列表的內容。
size_t quicklistGetLzf(const quicklistNode *node, void **data)	從*node中獲取LZF壓縮數據，並寫入data中，返回值爲data的長度。