redis skiplist性能实验

背景

    跳表：跳表是多个有序链表平行连接组成的结构，由于链表没有额外的信息和操作方式来执行快速的定位，所以跳表用了多个链表分摊了数据查询的复杂度；但是用了多个链表是怎么加快的？因为额外增加的链表中间是有空隙的，所以可以用来跳过步骤。理想情况下，底层的间隙为1，依次往上，分别是间隔2，4，8。

    但是实际实现中怎么知道跳表的间隔？如果有了增加和删除，那间隔是怎么确定的？实际上，跳表可能是很多人接触到的第一种随机数据结构。每个新值新建的时候会用一个随机值判断要不要增加上层节点，概率是p，所以新值节点有i层的概率是，i=1,2,3…

根据概率公式求和，每个值的平均层高为。例如，概率为0.25,则平均层高为1/(1-0.25)≈1.333。

实验环境

    ubuntu，2C4G机器，4GHz

前置知识

    最原始的跳表是未经优化的：

typedef struct tree_node_t {
    char* key;
    double score;
    struct tree_node_t* next;
    struct tree_node_t* down;
    /* possibly other information */
} tree_node_t;

typedef struct zskiplist {
    struct tree_node_t* header;
    unsigned long length;
    unsigned int level;
} zskiplist;

       redis里会有另一种结构的实现，先不表

关联知识

    zset是redis里的有序集合，对于每一个zset集合，可以存储多个value和对应的score（作为排序的依据，如果score相同，默认不排序，用sort进行排序）

    zset的底层实现包括了压缩列表和跳表，这里只讲跳表

redis提供了多个关于zset的命令，如zadd，zrange，zrem

    其中zadd和zrem就是添加、删除，zrange是范围查询。看到zrange可能会让你明白，为什么要用跳表而不用红黑树之类的数据结构--因为跳表可以作范围查询而树结构一般比较难。此外，跳表在代码层面相比树结构也更加容易debug。

    本文的源码参考来自6.0.12版本的redis

实验步骤

1. 好了，我们开始把redis的源码clone下来。

    git clone -b 6.0.12 git@github.com:redis/redis.git

找到src/t_zset.c和src/server.h，能看到跳表结构的定义

typedef struct zskiplistNode {
    sds ele;
    double score;
    struct zskiplistNode *backward;
    struct zskiplistLevel {
        struct zskiplistNode *forward;
        unsigned long span;
    } level[];
} zskiplistNode;

    对于这个实现，跳表的图能稍微帮我们理解一下。

其中的成员变量，score和level应该看得懂，分别是redis有序集里排序用到的分数和节点的高度；

    ele应该是element，也就是一个key，但是类型是sds，这个是什么类型呢？跳到定义：

typedef char *sds;

没啥信息，应该是个通用的类型；不过可以看到怎么生成和释放一个sds变量：

/* Create a new hisds string starting from a null terminated C string. */
hisds hi_sdsnew(const char *init) {
    size_t initlen = (init == NULL) ? 0 : strlen(init);
    return hi_sdsnewlen(init, initlen);
}

/* Free an hisds string. No operation is performed if 's' is NULL. */
void hi_sdsfree(hisds s) {
    if (s == NULL) return;
    hi_s_free((char*)s-hi_sdsHdrSize(s[-1]));
}

    这个先记下来，后面要用到

3.

抽取zslCreate，zslInsert，zslDelete

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;

    zsl = zmalloc(sizeof(*zsl));
    zsl->level = 1;
    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;
    }
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}


zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned long rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    // serverAssert(!isnan(score));
    x = zsl->header;
    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;
    }
    /* 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 = zslRandomLevel();
    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;
    }
    x = zslCreateNode(level,score,ele);
    for (i = 0; i < level; 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 */
        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 */
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }

    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;
    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. */
    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 */
}

3. 先写个main函数测试一下

参数：key 为64，按照<i> what's your name? <i>的格式生成

测试10万条的插入删除

用sdsnewlen搭配sds构造字符串类型的数据吧:

void test_redis_skiplist() {
    srandom(10000);
    zskiplist* zsl = zslCreate();
    const int LEN = 64;
    char** c = (char**)malloc(MAX_N * sizeof(char*));
    struct timespec st, ed;
    clock_gettime(CLOCK_MONOTONIC, &st);

    for (int i = 0; i < MAX_N; i++) {
        char* s = zmalloc(LEN + 1);
        sds ele;
        snprintf(s, LEN, "%d what's your name? %d", MAX_N - i, MAX_N - i);
        s[LEN] = '';
        ele = sdsnewlen(s, LEN);
        c[i] = ele;
        zfree(s);
        zslInsert(zsl, i, ele);
    }

    clock_gettime(CLOCK_MONOTONIC, &ed);
    report_time(st, ed);
    st = ed;
    printf("level length header: %d %d 0x%x
",
           zsl->level, zsl->length, zsl->header);

    for (int i = MAX_N - 1; i >= 0; i--) {
        sds ele = c[i];
        zslDelete(zsl, i, ele, NULL);
    }
    clock_gettime(CLOCK_MONOTONIC, &ed);
    report_time(st, ed);
    printf("level length header: %d %d 0x%x
",
           zsl->level, zsl->length, zsl->header);
}


int main() {
    test_redis_skiplist();
    return 0;
}

写个Makefile2，make，发现缺少zmalloc相关函数，参考该目录下的原Makefile，加上链接库：

../deps/jemalloc/lib/libjemalloc.a

# Makefile

default: zskip
MALLOC=jemalloc

DEPENDENCY_TARGETS+= jemalloc
FINAL_CFLAGS+= -DUSE_JEMALLOC -I../deps/jemalloc/include
FINAL_LIBS := ../deps/jemalloc/lib/libjemalloc.a $(FINAL_LIBS)

FINAL_LIBS+= -lm -ldl -pthread -lrt

zskip: zskip.c sds.o zmalloc.o
    gcc -std=c11 -g -o $@ $^ $(FINAL_LIBS)

sds.o: sds.c
    gcc -std=c11 -g -c $<

zmalloc.o: zmalloc.c
    gcc -std=c11 -g -c $<

jemalloc.o: jemalloc.c
    gcc -std=c11 -g -c $<

clean:
    rm zskip

.PHONY: clean

make -f Makefile2 && ./zskip

cost: 0s 038180193ns

avg: 381ns

per second: 2619159

level length header: 11 100000 0x7e80b000

cost: 0s 020009406ns

avg: 200ns

per second: 4997649

level length header: 1 0 0x7e80b000

key的长度：64

插入的平均耗时：381ns；

删除平均耗时：200ns

每秒插入260万，删除500万。

总结

    1.关于redis源码的分析已经有很多资料了，redis里的zskiplistNode的特点在于不是按原始的定义实现，而是每个节点已经包含了所有的层级；明确计算高度来判断要更新的节点以及用backward保存前向的节点也是一个比较简洁的处理，避免使用栈；

    2.关于header的细节：怎么判断一个节点是header？已经存在了zskiplist类型里，直接比较就可以了；

    3.高度上升，查询的复杂度就会变高，所以本来就限制了高度的上限；

    4.zrange为什么可以用start,end这种下标型的参数？想必从结构体的参数里可以看到span可以用来计算步长，在查找对应结果的时候已经可以很方便地计算了；

    5.redis的库里已经有很多针对性的优化，包括用了jemalloc做内存malloc；sds这种紧凑的类型在redis的其他地方也用到了，它甚至有5bit大小的类型；

    6.除了第一层外的层，其实也经常被叫做索引；增删节点的时候，可以看到有多个索引被对应更新，删掉节点会对应更新zskiplist的高度；

    7.本来是要比较redis的skiplist的算法复杂度和原始的类型next+down的速度的，但是测了一下基准已经差了很多倍，就不再比较了。

References

[1]redis 5设计与源码分析，陈雷

其他：写这篇博客的时候发现word又出BUG了，注册博客账户失败，后面看下要不要迁移

https://answers.microsoft.com/zh-hans/msoffice/forum/msoffice_word-mso_win10/%E6%9B%B4%E6%96%B0offices/b0f47f10-fd19-4de4-aad3-87a64ecc60fe