redis skiplist

skiplist数据结构

typedef struct zskiplistNode {
    sds ele;
    double score;
    struct zskiplistNode *backward;
    struct zskiplistLevel {
        struct zskiplistNode *forward;
        unsigned long span;    //span表示从current到forward的跨度，查找过程中计算rank
    } level[];
} zskiplistNode;

typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    unsigned long length;
    int level;
} zskiplist;

image.png

查找操作

// 在插入，删除操作中都需要先查找，直到查找到目标zskiplistNode前一个节点为止，并且记录过程中的所有需要变更的节点。
for (i = zsl->level-1; i >= 0; i--) {
       /* store rank that is crossed to reach the insert position */
       rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
       while (x->level[i].forward &&
               (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                    sdscmp(x->level[i].forward->ele,ele) < 0)))
       {
            rank[i] += x->level[i].span; 
            x = x->level[i].forward;
        }
       update[i] = x;
}

插入操作

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    // update数组表示需要发生变更的zskiplistNode, 同时是查找路径上的所有的zskiplistNode，x表示查询游标
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    // rank数组表示查找路径上所有节点的排名
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!isnan(score));
    x = zsl->header;
    // 计算的rank值的逻辑：
    // 如果只是向下查找，路径上的节点rank不变。
    // 如果向前查找，路径上的节点节点rank值 = 前一个节点rank + span 
    // 向下查找
    for (i = zsl->level-1; i >= 0; i--) {
        //向下查找时，计算rank值
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 两种情况向前查找：
        // 1. 当forward的score值小于查找score时向前查找
        // 2. forward的score值等于查找score值，但是forward的sds小于查找的sds时
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                    sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            // 向前查找时，计算rank值
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        update[i] = x;
    }
    /* we assume the element is not already inside, since we allow duplicated
     * scores, reinserting the same element should never happen since the
     * caller of zslInsert() should test in the hash table if the element is
     * already inside or not. */
    // 计算level
    level = zslRandomLevel();
    // 如果level大于skiplist的level, 变更路径要加上超出的部分，全部变更为header。
    if (level > zsl->level) {
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        zsl->level = level;
    }
    // 通过level, score, ele 创建要插入的zskiplistNode
    x = zslCreateNode(level,score,ele);
    for (i = 0; i < level; i++) {
       // 将zskiplistNode加入到skiplist中
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;

        // 因为加入了一个zskiplistNode，所以查找路径上zskiplistNode对应level的span都需要发生变更。
        //  这里需要画一下图来理解
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

    /* 如果level小于skiplist level, 查找路径上高于level的zskiplistLevel的span都需要加1 */
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    // 更新新加入节点的backward
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    zsl->length++;
    return x;
}

删除操作

int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;

    x = zsl->header;
    // 跟插入时的查找几乎一样，遍历结束后游标x指向要删除的前一个zskiplistNode
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                     sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            x = x->level[i].forward;
        }
        update[i] = x;
    }
    /* We may have multiple elements with the same score, what we need
     * is to find the element with both the right score and object. */
   // 找到真正要删除的zskiplistNode，并且进行比较
    x = x->level[0].forward;
    if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
        zslDeleteNode(zsl, x, update);
        if (!node)
            zslFreeNode(x);
        else
            *node = x;
        return 1;
    }
    return 0; /* not found */
}

// zslDeleteNode函数正在zslDelete, zslDeleteRangeByScore和zslDeleteRangeByRank中都使用到了，只是查找和比较的方式不一样
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    for (i = 0; i < zsl->level; i++) {
        // 如果查找路径上的节点对应的level的forward指向要删除的节点，需要同时变更span和forward。 新的span = 当前span + x的span - 1
        // 如果查找路径上的节点对应的level的forward不指向要删除的节点，span = span -1
        if (update[i]->level[i].forward == x) {
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {
            update[i]->level[i].span -= 1;
        }
    }
    // 变更x节点forward节点的backward
    if (x->level[0].forward) {
        x->level[0].forward->backward = x->backward;
    } else {
        zsl->tail = x->backward;
    }
    // 如果要删除的节点是唯一一个跟header同level的节点，需要进行降level操作
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    zsl->length--;
}