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应该看得懂，
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]));
}
这个先记下来，后面要用到
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] = '\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