首页 > 数据库> 文章详细

redis skiplist性能实验

2021-09-22 22:37:36 阅读：160 来源： 互联网

标签：level header redis ele forward score 实验 zsl 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应该看得懂，

ele应该是element吧？但是类型是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]));
}

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

抽取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] = '\0';
        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\n",
           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\n",
           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设计与源码分析，陈雷

标签：level,header,redis,ele,forward,score,实验,zsl,skiplist
来源： https://www.cnblogs.com/wangzming/p/15322048.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9