redis quicklist

redis源码版本v5.0.4

为什么引入quicklist

v3.2以前，redis list对象中元素的长度比较小或者数量比较少的时候，采用ziplist来存储。当列表对象中元素的长度比较大或者数量比较多的时候，则会转而使用双向列表linkedlist来存储。
v3.2以后，引入了一个 quicklist 的数据结构，列表的底层都由quicklist实现, list对象使用quicklist。

linkedlist vs ziplist

linkedlist优缺点

• 优点: 插入，删除节点复杂度很低，简单
• 缺点: 除保存数据外还需要保存prev, next两个指针，内存利用率低，双向链表的各个节点是单独的内存块，地址不连续，节点多了容易产生内存碎片。

ziplist优缺点

• 优点: 存储在一段连续的内存上，不容易产生内存碎片，内存利用率高。
• 缺点: 插入和删除操作需要频繁的申请和释放内存, 同时会发生内存拷贝，数据量大时内存拷贝开销较大。

quicklist数据结构

typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *zl;
    unsigned int sz;             /* ziplist占用byte数*/
    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;

typedef struct quicklistLZF {
    unsigned int sz; /* LZF size in bytes*/
    char compressed[];
} quicklistLZF;

typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* 所有ziplist中entry的数量 */
    unsigned long len;          /* quicklistNode的数量 */
    int fill : 16;              /* fill factor for individual nodes */
    unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
} quicklist;

image

lpush的过程

lpush过程是从网络缓存区读到数据并将数据包装成client之后的处理过程。rpush过程与lpush类似，只是插入的位置有LIST_HEAD改为LIST_TAIL

/** lpushCommand, pushGenericCommand, listTypePush在t_list.c 文件中*/
void lpushCommand(client *c) {
    //插入的是LIST_HEAD
    pushGenericCommand(c,LIST_HEAD);
}

void pushGenericCommand(client *c, int where) {
    int j, pushed = 0;
    //从db数据机构里找到要插入的quicklist地址
    robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
    //判断编码格式
    if (lobj && lobj->type != OBJ_LIST) {
        addReply(c,shared.wrongtypeerr);
        return;
    }

    for (j = 2; j < c->argc; j++) {
        if (!lobj) {
           //db中不存在quicklist，初始化并存储到db
            lobj = createQuicklistObject();
            quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,
                                server.list_compress_depth);
            dbAdd(c->db,c->argv[1],lobj);
        }
        //进行push过程
        listTypePush(lobj,c->argv[j],where);
        pushed++;
    }
    addReplyLongLong(c, (lobj ? listTypeLength(lobj) : 0));
    if (pushed) {
        char *event = (where == LIST_HEAD) ? "lpush" : "rpush";

        signalModifiedKey(c->db,c->argv[1]);
        notifyKeyspaceEvent(NOTIFY_LIST,event,c->argv[1],c->db->id);
    }
    server.dirty += pushed;
}

//push入quicklist的过程
void listTypePush(robj *subject, robj *value, int where) {
    //编码检查
    if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
        //插入QUICKLIST_HEAD
        int pos = (where == LIST_HEAD) ? QUICKLIST_HEAD : QUICKLIST_TAIL;
        //从robj数据结构中提取出要插入的value，计算value的长度
        value = getDecodedObject(value);
        size_t len = sdslen(value->ptr);
        //quicklist的push操作
        quicklistPush(subject->ptr, value->ptr, len, pos);
        decrRefCount(value);
    } else {
        serverPanic("Unknown list encoding");
    }
}

/** quicklistPush, quicklistPushHead 在quicklist.c文件中 */
void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
                   int where) {
    //这里lpush是插入QUICKLIST_HEAD, rpush是插入QUICKLIST_TAIL
    if (where == QUICKLIST_HEAD) {
        quicklistPushHead(quicklist, value, sz);
    } else if (where == QUICKLIST_TAIL) {
        quicklistPushTail(quicklist, value, sz);
    }
}

int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_head = quicklist->head;
    //判断是否quicklist一个quicklistNode是否满了，不满直接找到quicklistNode的ziplist,使用ziplist的push操作， 满则新建quicklistNode，使用新quicklistNode的ziplist进行push
    if (likely(
            _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
        //调用ziplist的api
        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);
        //更新quicklistNode的sz属性(ziplist占用字节数)
        quicklistNodeUpdateSz(node);
        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
    }
    //更新quicklist 的entry数量， quicklist header的enty数量
    quicklist->count++;
    quicklist->head->count++;
    return (orig_head != quicklist->head);
}

/** ziplistPush 在ziplist.c */
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where) {
    unsigned char *p;
    //计算插入位置p
    p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
    //在redis ziplist中有__ziplistInsert的解释
    return __ziplistInsert(zl,p,s,slen);
}