quicklist.c

Redis的quicklist是一种基于ziplist实现的可压缩（quicklistLZF）的双向链表，结合了链表和ziplist的优点组成的。

传统的双向链表在内存空间上较为分散，而基于ziplist使得空间更为连续。

quicklist的数据结构如下：

/* 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;
    //所有quicklistNode中的ziplist中的entry个数的总和
    unsigned long count;        /* total count of all entries in all ziplists */
    //quicklistNode的个数
    unsigned long len;          /* number of quicklistNodes */
    //单个节点的填充因子，ziplist的最大长度，list-max-ziplist-size，entry的最大个数
    int fill : 16;              /* fill factor for individual nodes */
    //节点的压缩深度，list-compress-depth
    unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
} quicklist;

quicklist包含了:

• 首尾节点head、tail
• 所有quicklistNode中包含的Entry的总个数count
• quicklist中quicklistNode的个数len
• 填充因子，每个ziplist中可以容纳的entry的个数fill
• 压缩深度compress

quicklistNode的数据结构如下：

/* 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 byt es.
 * 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: 10 bits, free for future use; pads out the remainder of 32 bits */
//快速列表节点，双端
typedef struct quicklistNode {
    //前置节点
    struct quicklistNode *prev;
    //后置节点
    struct quicklistNode *next;
    //基于ziplist，如果没有被压缩，则指向ziplist，否则指向quicklistLZF
    unsigned char *zl;
    //zl指向的ziplist的字节大小
    unsigned int sz;             /* ziplist size in bytes */
    //ziplist中的entry个数
    unsigned int count : 16;     /* count of items in ziplist */
    //RAW(ziplist)或者LZF(lzf)
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    //标记当前存储数据的是ziplist还是其它，默认是ziplist
    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;

quicklistNode包含了：

• 前后指针prev、next
• zl指针zl，如果没有被压缩，则结构为ziplist，否则为quicklistLZF
• 压缩列表的字节大小sz
• entry的总个数count
• 编码encoding
• 数据容器container，默认是ziplist
• 解压标记recompress
• 测试专用字段attempted_compress
• 预留字段extra

quicklistLZF的结构如下：

/* 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 */
//LZF无损压缩算法，压缩过的ziplist
typedef struct quicklistLZF {
    //未压缩之前的大小
    unsigned int sz; /* LZF size in bytes*/
    //存放压缩过的ziplist数组
    char compressed[];
} quicklistLZF;

quicklist、quicklistNode、ziplist、quicklistLZF的关系图如下：

quicklist.png

quicklist也存在着一个类似ziplist的entry的结构，称为quicklistEntry。
在内存空间组织上也是连续的。

quicklistEntry的结构如下：

//快速列表entry对 <zi,value>
typedef struct quicklistEntry {
    //指向所属快速列表
    const quicklist *quicklist;
    //指向所属快速节点
    quicklistNode *node;
    //指向所属ziplist
    unsigned char *zi;
    //指向ziplist的value
    unsigned char *value;
    //指向ziplist的整数value
    long long longval;
    //保存当前ziplist的字节数大小
    unsigned int sz;
    //保存相对偏移量大小
    int offset;
} quicklistEntry;

1. 创建quicklist

/* Create a new quicklist.
 * Free with quicklistRelease(). */
quicklist *quicklistCreate(void) {
    struct quicklist *quicklist;

    //分配内存
    quicklist = zmalloc(sizeof(*quicklist));
    //首尾节点
    quicklist->head = quicklist->tail = NULL;
    //quicklistNode个数
    quicklist->len = 0;
    //entry的总个数
    quicklist->count = 0;
    //压缩深度
    quicklist->compress = 0;
    //填充因子
    quicklist->fill = -2;
    return quicklist;
}

/* Create a new quicklist with some default parameters. */
quicklist *quicklistNew(int fill, int compress) {
    quicklist *quicklist = quicklistCreate();
    //设置fill\compress
    quicklistSetOptions(quicklist, fill, compress);
    return quicklist;
}

2. 新增节点

/* Wrapper to allow argument-based switching between HEAD/TAIL pop */
/**
 * 新增节点
 * @param quicklist 目标快速列表
 * @param value 插入值
 * @param sz  字节大小
 * @param where 插入的位置 0和-1
 */
void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
                   int where) {
    if (where == QUICKLIST_HEAD) {
        //头插
        quicklistPushHead(quicklist, value, sz);
    } else if (where == QUICKLIST_TAIL) {
        //尾插
        quicklistPushTail(quicklist, value, sz);
    }
}

先看看quicklistPushHead的实现

/* 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) {
    quicklistNode *orig_head = quicklist->head;
    //如果当前quicklistNode的ziplist中的entry个数没有超过限制，则插入当前的ziplist中
    if (likely(
            _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
        //插入节点
        quicklist->head->zl =
            ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
        //更新字节总数
        quicklistNodeUpdateSz(quicklist->head);
    } else {
        //否则，创建新的quicklistNode
        quicklistNode *node = quicklistCreateNode();
        //创建ziplist
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        //更新字节总数
        quicklistNodeUpdateSz(node);
        //插入节点
        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
    }
    //总数+1
    quicklist->count++;
    //个数+1
    quicklist->head->count++;
    return (orig_head != quicklist->head);
}

再看看quicklistPushTail的实现：

/* Add new entry to tail node of quicklist.
 *
 * Returns 0 if used existing tail.
 * Returns 1 if new tail created. */
int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_tail = quicklist->tail;
    if (likely(
            _quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) {
        quicklist->tail->zl =
            ziplistPush(quicklist->tail->zl, value, sz, ZIPLIST_TAIL);
        quicklistNodeUpdateSz(quicklist->tail);
    } else {
        quicklistNode *node = quicklistCreateNode();
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL);

        quicklistNodeUpdateSz(node);
        _quicklistInsertNodeAfter(quicklist, quicklist->tail, node);
    }
    quicklist->count++;
    quicklist->tail->count++;
    return (orig_tail != quicklist->tail);
}

和quicklistPushHead的实现类似。

3. 新增quicklistEntry

/* 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'. */
//插入entry节点，并压缩
REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
                                   void *value, const size_t sz, int after) {
    int full = 0, at_tail = 0, at_head = 0, full_next = 0, full_prev = 0;
    int fill = quicklist->fill;
    quicklistNode *node = entry->node;
    quicklistNode *new_node = NULL;

    //node为空
    if (!node) {
        /* we have no reference node, so let's create only node in the list */
        D("No node given!");
        //如果entry所属quicklistNode为空，那就创建一个
        new_node = quicklistCreateNode();
        //创建quicklistNode所属ziplist
        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        //在quicklist上插入new_node
        __quicklistInsertNode(quicklist, NULL, new_node, after);
        //entry个数+1
        new_node->count++;
        //总entry数+1
        quicklist->count++;
        return;
    }

    //node不为空
    //判断当前quicklistNode中的ziplist的entry个数是否满了
    if (!_quicklistNodeAllowInsert(node, fill, sz)) {
        D("Current node is full with count %d with requested fill %lu",
          node->count, fill);
        full = 1;
    }

    //如果是在队尾插入
    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;
        }
    }

    //如果是在队首插入
    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) {
        D("Not full, inserting after current position.");
        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++;
        quicklistNodeUpdateSz(node);
        quicklistRecompressOnly(quicklist, node);
    }
    //如果没有满且在队首
    else if (!full && !after) {
        D("Not full, inserting before current position.");
        quicklistDecompressNodeForUse(node);
        node->zl = ziplistInsert(node->zl, entry->zi, value, sz);
        node->count++;
        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. */
        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. */
        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 */
        D("\tprovisioning new node...");
        new_node = quicklistCreateNode();
        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        __quicklistInsertNode(quicklist, node, new_node, after);
    }
    //如果满了
    else if (full) {
        /* else, node is full we need to split it. */
        /* covers both after and !after cases */
        D("\tsplitting node...");
        quicklistDecompressNodeForUse(node);
        //分离node到2个地方存储
        new_node = _quicklistSplitNode(node, entry->offset, after);
        new_node->zl = ziplistPush(new_node->zl, value, sz,
                                   after ? ZIPLIST_HEAD : ZIPLIST_TAIL);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        __quicklistInsertNode(quicklist, node, new_node, after);
        _quicklistMergeNodes(quicklist, node);
    }

    quicklist->count++;
}

4. 移除节点

/* Default pop function
 *
 * Returns malloc'd value from quicklist */
/**
 * 移除quicklist节点
 * @param quicklist 目标快速列表
 * @param where  移除的位置
 * @param data  数据
 * @param sz  大小
 * @param slong 返回值
 * @return
 */
int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
                 unsigned int *sz, long long *slong) {
    unsigned char *vstr;
    unsigned int vlen;
    long long vlong;
    if (quicklist->count == 0)
        return 0;
    int ret = quicklistPopCustom(quicklist, where, &vstr, &vlen, &vlong,
                                 _quicklistSaver);
    if (data)
        *data = vstr;
    if (slong)
        *slong = vlong;
    if (sz)
        *sz = vlen;
    return ret;
}

/* pop from quicklist and return result in 'data' ptr.  Value of 'data'
 * is the return value of 'saver' function pointer if the data is NOT a number.
 *
 * If the quicklist element is a long long, then the return value is returned in
 * 'sval'.
 *
 * Return value of 0 means no elements available.
 * Return value of 1 means check 'data' and 'sval' for values.
 * If 'data' is set, use 'data' and 'sz'.  Otherwise, use 'sval'. */
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                       unsigned int *sz, long long *sval,
                       void *(*saver)(unsigned char *data, unsigned int sz)) {
    unsigned char *p;
    unsigned char *vstr;
    unsigned int vlen;
    long long vlong;
    int pos = (where == QUICKLIST_HEAD) ? 0 : -1;

    if (quicklist->count == 0)
        return 0;

    if (data)
        *data = NULL;
    if (sz)
        *sz = 0;
    if (sval)
        *sval = -123456789;

    quicklistNode *node;
    if (where == QUICKLIST_HEAD && quicklist->head) {
        node = quicklist->head;
    } else if (where == QUICKLIST_TAIL && quicklist->tail) {
        node = quicklist->tail;
    } else {
        return 0;
    }

    //找出节点
    p = ziplistIndex(node->zl, pos);
    if (ziplistGet(p, &vstr, &vlen, &vlong)) {
        if (vstr) {
            if (data)
                *data = saver(vstr, vlen);
            if (sz)
                *sz = vlen;
        } else {
            if (data)
                *data = NULL;
            if (sval)
                *sval = vlong;
        }
        //删除
        quicklistDelIndex(quicklist, node, &p);
        return 1;
    }
    return 0;
}

/* Delete one entry from list given the node for the entry and a pointer
 * to the entry in the node.
 *
 * Note: quicklistDelIndex() *requires* uncompressed nodes because you
 *       already had to get *p from an uncompressed node somewhere.
 *
 * Returns 1 if the entire node was deleted, 0 if node still exists.
 * Also updates in/out param 'p' with the next offset in the ziplist. */
REDIS_STATIC int quicklistDelIndex(quicklist *quicklist, quicklistNode *node,
                                   unsigned char **p) {
    int gone = 0;

    //删除ziplist中的p
    node->zl = ziplistDelete(node->zl, p);
    node->count--;
    //如果entry个数为0了，则删除此node
    if (node->count == 0) {
        gone = 1;
        __quicklistDelNode(quicklist, node);
    } else {
        quicklistNodeUpdateSz(node);
    }
    quicklist->count--;
    /* If we deleted the node, the original node is no longer valid */
    return gone ? 1 : 0;
}


REDIS_STATIC void __quicklistDelNode(quicklist *quicklist,
                                     quicklistNode *node) {
    if (node->next)
        node->next->prev = node->prev;
    if (node->prev)
        node->prev->next = node->next;

    if (node == quicklist->tail) {
        quicklist->tail = node->prev;
    }

    if (node == quicklist->head) {
        quicklist->head = node->next;
    }

    /* If we deleted a node within our compress depth, we
     * now have compressed nodes needing to be decompressed. */
    __quicklistCompress(quicklist, NULL);

    quicklist->count -= node->count;

    zfree(node->zl);
    zfree(node);
    quicklist->len--;
}

5. 移除entry

/* Delete one element represented by 'entry'
 *
 * 'entry' stores enough metadata to delete the proper position in
 * the correct ziplist in the correct quicklist node. */
//移除entry
void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry) {
    quicklistNode *prev = entry->node->prev;
    quicklistNode *next = entry->node->next;
    //和移除节点类似
    int deleted_node = quicklistDelIndex((quicklist *)entry->quicklist,
                                         entry->node, &entry->zi);

    /* after delete, the zi is now invalid for any future usage. */
    iter->zi = NULL;

    /* If current node is deleted, we must update iterator node and offset. */
    if (deleted_node) {
        if (iter->direction == AL_START_HEAD) {
            iter->current = next;
            iter->offset = 0;
        } else if (iter->direction == AL_START_TAIL) {
            iter->current = prev;
            iter->offset = -1;
        }
    }
    /* else if (!deleted_node), no changes needed.
     * we already reset iter->zi above, and the existing iter->offset
     * doesn't move again because:
     *   - [1, 2, 3] => delete offset 1 => [1, 3]: next element still offset 1
     *   - [1, 2, 3] => delete offset 0 => [2, 3]: next element still offset 0
     *  if we deleted the last element at offet N and now
     *  length of this ziplist is N-1, the next call into
     *  quicklistNext() will jump to the next node. */
}