Redis源码解析之跳跃表（一）

127.0.0.1:6379> ZADD worker-language 90 Java

(integer) 1

127.0.0.1:6379> ZADD worker-language 20 C

(integer) 1

127.0.0.1:6379> ZADD worker-language 57 Python

(integer) 1

127.0.0.1:6379> ZADD worker-language 82 Go

(integer) 1

127.0.0.1:6379> ZADD worker-language 61 PHP

(integer) 1

127.0.0.1:6379> ZADD worker-language 28 Scala

(integer) 1

127.0.0.1:6379> ZADD worker-language 33 C++

(integer) 1

127.0.0.1:6379> ZCARD worker-language

(integer) 7

127.0.0.1:6379> ZRANGE worker-language 0 -1 WITHSCORES

 1) "C"

 2) "20"

 3) "Scala"

 4) "28"

 5) "C++"

 6) "33"

 7) "Python"

 8) "57"

 9) "PHP"

10) "61"

11) "Go"

12) "82"

13) "Java"

14) "90"

127.0.0.1:6379> ZRANGE worker-language 2 5 WITHSCORES

1) "C++"

2) "33"

3) "Python"

4) "57"

5) "PHP"

6) "61"

7) "Go"

8) "82"

127.0.0.1:6379> ZRANGEBYSCORE worker-language 25 85 WITHSCORES

 1) "Scala"

 2) "28"

 3) "C++"

 4) "33"

 5) "Python"

 6) "57"

 7) "PHP"

 8) "61"

 9) "Go"

10) "82"

#[25,85]区间的元素有[Scala,C++,Python,PHP,Go]，加上LIMIT选项仅查询从偏移1开始（包含偏移1）往后的3个元素

127.0.0.1:6379> ZRANGEBYSCORE worker-language 25 85 WITHSCORES LIMIT 1 3

1) "C++"

2) "33"

3) "Python"

4) "57"

5) "PHP"

6) "61"

typedef char *sds;

/* ZSETs use a specialized version of Skiplists */

typedef struct zskiplistNode {

    sds ele;

    double score;

    struct zskiplistNode *backward;

    struct zskiplistLevel {

        struct zskiplistNode *forward;

        unsigned long span;

    } level[];

} zskiplistNode;

typedef struct zskiplist {

    struct zskiplistNode *header, *tail;

    unsigned long length;

    int level;

} zskiplist;

typedef struct zset {

    dict *dict;

    zskiplist *zsl;

} zset;

//server.h

#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^64 elements */

/* t_zset.c

 * Create a new skiplist.

 创建并返回一个跳跃表

*/

zskiplist *zslCreate(void) {

    int j;

    zskiplist *zsl;

    //分配空间

    zsl = zmalloc(sizeof(*zsl));

    //设置起始层次

    zsl->level = 1;

    //初始元素个数为0

    zsl->length = 0;

    //初始化表头

    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);

    //初始化层高

    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {

        zsl->header->level[j].forward = NULL;

        zsl->header->level[j].span = 0;

    }

    //设置表头后继指针为NULL

    zsl->header->backward = NULL;

    //初始表尾指针为NULL

    zsl->tail = NULL;

    return zsl;

}

//创建跳跃表元素

zskiplistNode *zslCreateNode(int level, double score, sds ele) {

    /**

     * zskiplistNode最后一个字段为弹性数组，在计算内存占用时不占空间，需要

     * 额外分配level*sizeof(struct zskiplistLevel)个字节空间用于存放层高

     */

    zskiplistNode *zn =

        zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));

    zn->score = score;

    zn->ele = ele;

    return zn;

}

/* Insert a new node in the skiplist. Assumes the element does not already

 * exist (up to the caller to enforce that). The skiplist takes ownership

 * of the passed SDS string 'ele'.

 * 将一个新节点插入到跳跃表。

 * */

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {

    /*

     * update数组会记录每一层最靠近新插入节点的后继节点，rank数组

     * 最靠近新插入节点的后继节点索引。

     */

    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;

    unsigned int rank[ZSKIPLIST_MAXLEVEL];

    int i, level;

    serverAssert(!isnan(score));

    x = zsl->header;//<1>

    //从头节点最高层出发，查找每一层最靠近新节点的后继节点

    for (i = zsl->level - 1; i >= 0; i--) {

        /* store rank that is crossed to reach the insert position

         * rank数组记录每一层最靠近新节点的后继节点索引，从最高层开始查找时初始索引为0，

         * 即初始rank[zsl->level-1]=0，如果能从当前节点前进到下一个节点，则把当前节点的跨度

         * 加到索引值上。当处于i层前进到最靠近新节点的后继节点不能再前进时，则下沉一层，

         * 基于当前前进到的后继节点继续尝试往i-1层前进，而i-1层的初始索引值为i层最后统计的

         * 索引值，因为i-1层不会从header节点逐个递进，而是基于上一层前进到的节点递进，所以

         * 要弥补i-1层的索引要弥补header前进到当前节点的索引值。

        */

        rank[i] = i == (zsl->level - 1) ? 0 : rank[i + 1];

        /*

         * x初始值为header(见<1>处)，如果x在i层的前驱节点不为NULL，则x有可能前进到前驱节点，

         * 即x在i层指向的前驱节点有可能成为新节点的后继节点，但x能否前进要进一步判断，这里有

         * 两个条件决定前进：

         * 1.如果x前驱节点的分值小于新节点的分值，则可以前进(x->level[i].forward->score < score)

         * 2.前驱节点的分值与新节点分值相同，且前驱节点的字符串小于新节点的字符串，则允许前进。注意：

         * 这里比较字符串并非比较字符串长度，sdscmp(x->level[i].forward->ele,ele)会进一步调用

         * 函数memcmp(s1,s2,n)，比较两个字符串的s1和s2的前n个字节。

         * (x->level[i].forward->score == score &&

                    sdscmp(x->level[i].forward->ele,ele) < 0)

         */

        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指向的内存区域更新为下一个节点。

             */

            rank[i] += x->level[i].span;

            x = x->level[i].forward;

        }

        //x一直前进到不能再前进，即代表x所指向的节点为i层最靠近新节点的后继节点，记录后继节点

        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.

     *

     * */

    /*

     * 这里会用随机算法生成一个层高，越往上的层高越难生成，

     * 并且保证层高一定<=ZSKIPLIST_MAXLEVEL，下面会附上

     * 生成层高的代码。

     */

    level = zslRandomLevel();

    /*

     * 如果当前生成的层高大于跳跃表最高层高，还需要更新大于等于zsl->level

     * 层的后继节点和节点索引。大于等于zsl->level层的后继节点只能是头节点，

     * 因此后继节点索引都为0，同时更新这些后继节点的跨度为跳跃表节点数，便于

     * 后续更新跨度。

     */

    if (level > zsl->level) {//<2>

        for (i = zsl->level; i < level; i++) {

            rank[i] = 0;

            update[i] = zsl->header;

            update[i]->level[i].span = zsl->length;

        }

        //更新跳跃表层高为目前生成的层高

        zsl->level = level;

    }

    //根据层高、分支和元素创建一个跳跃表节点

    x = zslCreateNode(level, score, ele);

    //将新节点插入到跳跃表中，并更新跨度

    for (i = 0; i < level; i++) {

        /*

         * update[i]为新节点在i层的后继节点，将新节点在i层指向的前驱更新

         * 为后继节点指向的前驱。再将后继节点在i层的前驱更新为新节点。如此，

         * 新节点便插入到跳跃表中。

         */

        x->level[i].forward = update[i]->level[i].forward;

        update[i]->level[i].forward = x;

        /* update span covered by update[i] as x is inserted here

         * (rank[0] - rank[i])为新节点和第i层后继节点中间隔了多少个节点，

         * 在i层x到前驱节点的跨度(x->level[i].span)为原先后继节点的跨度减去

         * 新节点和第i层后继节点中间的节点个数:

         * (update[i]->level[i].span - (rank[0] - rank[i]))

         * 后继节点在新节点的跨度为:(rank[0] - rank[i]) + 1

         * 如果程序之前曾进入到分支<2>，即执行level = zslRandomLevel()生成的层高

         * 高大于原先跳跃表的层高，比如原先zsl->level为3，但level生成为10，则新节点L3~L9

         * 的后继节点为头节点，后继节点索引值为0，新节点L3~L9的前驱为NULL，跨度为新节点之后

         * 的节点个数，而zsl->length - (rank[0] - rank[i])为新节点之后的节点个数。

        */

        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);

        update[i]->level[i].span = (rank[0] - rank[i]) + 1;

    }

    /* increment span for untouched levels

     * 如果level = zslRandomLevel()生成的层高大于原先zsl->level的

     * 层高，则不会进入此循环。只有当生成的层高小于原先zsl->level的层高，

     * 才需要对level<=i<zsl->level的后继节点中的跨度+1，因为新插入一个节点。

     */

    for (i = level; i < zsl->level; i++) {

        update[i]->level[i].span++;

    }

    /*

     * L0层相当于普通链表，如果新插入节点在L0层的后继节点为header，则新节点的backward指向NULL，

     * 否则指向跳跃表中最靠近新节点的后继节点。

     */

    x->backward = (update[0] == zsl->header) ? NULL : update[0];

    /*

     * 如果新节点不是末尾节点，则将新节点的前驱指点的backward指针指向新节点，

     * 否则将跳跃表的末尾指针指向新节点。

     */

    if (x->level[0].forward)

        x->level[0].forward->backward = x;

    else

        zsl->tail = x;

    //插入新节点后，对跳跃表的元素个数+1

    zsl->length++;

    return x;

}

/* Returns a random level for the new skiplist node we are going to create.

 * The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL

 * (both inclusive), with a powerlaw-alike distribution where higher

 * levels are less likely to be returned.

 *

 *              Skiplist P = 1/4

 *              while 循环中

 *              循环为真的概率为 P(1/4)

 *              循环为假的概率为 (1 - P)

 *

 *              层高为1 的概率为           (1 -P)

 *              层高为2 的概率为        P *(1 -P)

 *              层高为3 的概率为      P^2 *(1 -P)

 *              层高为4 的概率为      P^3 *(1 -P)

 *              层高为n 的概率为  P^(n-1) *(1 -P)

 *

 *             因此平均层高为

 *             E = 1*( 1-P) + 2*P( 1-P) + 3*(P^2) ( 1-P) + ...

 *               = (1-P) ∑ +∞ iP^(i-1)

 *                         i=1

 *

 *               =1/(1-P)

 *

 * */

int zslRandomLevel(void) {

    int level = 1;

    while ((random() & 0xFFFF) < (ZSKIPLIST_P * 0xFFFF))

        level += 1;

    return (level < ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;

}