关于c++:leveldb-memdb源码分析上

前言
最近在钻研学习leveldb的源码，并且尝试用Rust进行重写leveldb-rs，leveldb中memdb模块是应用skiplist作为一个kv的内存存储，相干代码实现十分丑陋，所以有了这篇文章。 leveldb通过应用Arena模式来实现skiplist。简略来说，就是利用线性数组来模仿节点之间的关系，能够无效防止循环援用。

c++版本的leveldb尽管也是应用的arena模式，然而节点数据内存的申请和拜访进行了封装，skiplist的构造定义和实现跟传统意义上的skiplist的代码实现十分类似，如果如果大家之前理解过skiplist的话，c++版本的代码是非常容易看懂的。
golang版本leveldb 不足arena的封装，间接操作slice，如果对arena模式不相熟的话，了解起来就比拟麻烦。从软件工程角度上开，golang版本的memdb的代码写的不太好，能够进一步优化的和重构arena的操作。

在本文中将会解说上面内容：

比照c++和golang版本中查问、插入、删除的实现
剖析golang版本中能够优化的中央

而后在下一篇文章中将会介绍

基于golang版本应用rust重写memdb（arena版本）
应用rust重写一个非arena版本的memdb，也就是经典的链表构造实现形式

类型申明
首先咱们来比照C++和Golang的代码中的skiplist定义：

C++
https://github.com/google/lev…

这里次要列出要害的成员变量，具体的能够去看源码：

template <typename Key, class Comparator>
class SkipList {

...

  // Immutable after construction
  Comparator const compare_; 
  Arena* const arena_;  // Arena used for allocations of nodes

  Node* const head_;

  // Modified only by Insert().  Read racily by readers, but stale
  // values are ok.
  std::atomic<int> max_height_;  // Height of the entire list

  // Read/written only by Insert().
  Random rnd_;
};

Comparator const compare_; 用来在遍历skiplist进行节点key的比拟
Arena* const arena_; 应用Arena模式的内存治理
Node* const head_; 首节点
std::atomic max_height_; skiplist的层高，在插入的时候可能会变动
Random rnd_; 随机数生成器，用于在每次插入的时候生成新节点的层高

Golang
https://github.com/syndtr/gol…

type DB struct {
    cmp comparer.BasicComparer 
    rnd *rand.Rand

    mu     sync.RWMutex
    kvData []byte
    nodeData  []int
    prevNode  [tMaxHeight]int
    maxHeight int
    n         int
    kvSize    int
}

cmp comparer.BasicComparer :用来在遍历skiplist进行节点key的比拟
rnd *rand.Rand：随机数生成器，用于在每次插入的时候生成新节点的层高
kvData []byte: key和value理论数据寄存的中央
nodeData[]int: 存储各个节点的信息
prevNode [tMaxHeight]int：用于在遍历skiplist的时候，保留每一层的前一个节点
maxHeight int: skiplist的层高，在插入的时候可能会变动
n int: 节点的总个数
kvsize: skiplist中存储key和value的总字节数

golang版本外面最难了解的就是nodeData, 只有了解了nodeData的数据布局，前面代码就容易了解了。

kvData中存储的是key，value的实在的字节数据
kvNode中存储的是skiplist中的全副节点，然而节点不存储key和value的理论数据而是在Kvdata中的偏移以及key的长度，value的长度，在比拟的时候再依据偏移和长度到KvData中读取。另外KvNode中还存储了以后节点的层高，以及每一层的下一个节点在KvNode中的偏移量，在查问的时候，就能够依据偏移量跳到KvNode中下一个节点的地位，在从外面读取信息。

查问大于等于特定Key
首先看skiplist中的查问，leveldb中查问的实现是最要害的，插入和删除也都是基于查问实现，咱们先来简略回顾下查问的过程：

首先依据跳表的高度选取最高层的头节点；
若跳表中的节点内容小于查找节点的内容，则取该层的下一个节点持续比拟；
若跳表中的节点内容等于查找节点的内容，则间接返回；
若跳表中的节点内容大于查找节点的内容，且层高不为0，则升高层高，且从前一个节点开始，从新查找低一层中的节点信息；若层高为0，则返回以后节点，该节点的key大于所查找节点的key。

咱们举例来说，如果要在上面的skiplist中查问key为17节点

从最右边的head节点开始，以后层高是4；
head节点在第4层的next节点的key是6，因为 17 大于6，所以在以后节点的左边，就沿着以后层的链表走到下一节点，也就是key是6节点。
6节点在第4层的next节点是NIL，也就是前面没有节点了，那么就须要在以后节点往上层走，走到第3层。
6节点在第3层的next节点的key是25，因为 17 小于25，那么就须要在以后节点往上层走，走到第2层。
6节点在第2层的next节点的key是9，因为 17 大于9，那么就沿着以后层的链表走到下一节点，也就是key是9的节点。
9节点在第2层的nex节点的key是25，因为 17 小于25，那么就须要在以后节点往上层走，走到第1层。
8节点在第1层的next节点的key是12，因为 17 大于12，那么就沿着以后层的链表走到下一节点，也就是key是12的节点。
12节点在第1层的next节点的key是19，因为 17 小于19，原本应该要持续走到下一层，然而因为以后曾经是最初一层了，所以间接返回12的next节点，也就是19节点.

C++
https://github.com/google/lev…

在skiplist中查问大于等于key的最小节点的办法如下

template <typename Key, class Comparator>
typename SkipList<Key, Comparator>::Node*
SkipList<Key, Comparator>::FindGreaterOrEqual(const Key& key,
                                              Node** prev) const {
  Node* x = head_;  // head节点
  int level = GetMaxHeight() - 1;// 以后层高
  while (true) {
    Node* next = x->Next(level);

    if (KeyIsAfterNode(key, next)) {// 如果以后层中x的下一个节点的key小于key
      x = next; // 持续在以后层的list往后搜寻
    } else {
      if (prev != nullptr) prev[level] = x; // 如果要记录遍历过程中的pre节点，就记录
      if (level == 0) { // 搜寻到底了就返回
        return next;
      } else {
        // 如果以后层中x下一个节点的key大于key，往下一层进行搜寻
        level--;
      }
    }
  }
}

Go
https://github.com/syndtr/gol…

在skiplist中查问大于等于key的最小节点的办法如下

// Must hold RW-lock if prev == true, as it use shared prevNode slice.
func (p *DB) findGE(key []byte, prev bool) (int, bool) {

    node := 0  // head 节点
    h := p.maxHeight - 1  // 以后层高
    for {
        next := p.nodeData[node+nNext+h]
        cmp := 1
        if next != 0 {
            o := p.nodeData[next]
            cmp = p.cmp.Compare(p.kvData[o:o+p.nodeData[next+nKey]], key)
        }
        // 如果以后层中node的下一个节点的key小于key，持续在以后层的list往后搜寻
        if cmp < 0 {
            // Keep searching in this list
            node = next
        } else {

            if prev { // 对于插入或删除而进行的搜寻，即便遇到雷同的也要持续往下一层比拟
                p.prevNode[h] = node
            } else if cmp == 0 {
                return next, true
            }
            if h == 0 {
                return next, cmp == 0
            }
            // 如果以后层中node的下一个节点的key大于key，以后node往下一层进行搜寻
            h--
        }
    }
}

node + nNext + i 咱们能够当作成C++代码中链表构造skiplist中第node个节点在第i层，而后p.nodeData[node+nNext+i] 当成 node->Next(i)
p.nodeData[next+nKey]获取key的长度

查问小于等于
GE搜寻，和LT搜寻十分相似，其中要害的差异在于

GE搜寻返回的是next节点，LT返回的是以后node
GE搜寻过程中如果next和以后Key雷同就返回，LT如果next和以后key雷同，进入下一层，这样就将搜寻区间限度在小于key的范畴了。

C++

https://github.com/google/lev… 在skiplist中查问小于key的最大节点的办法如下

template <typename Key, class Comparator>
typename SkipList<Key, Comparator>::Node*
SkipList<Key, Comparator>::FindLessThan(const Key& key) const {
  Node* x = head_; // head节点 
  int level = GetMaxHeight() - 1; // 以后层高
  while (true) {
    assert(x == head_ || compare_(x->key, key) < 0);
    Node* next = x->Next(level);
    if (next == nullptr || compare_(next->key, key) >= 0) { // 如果next节点为空或 next节点大于等于申请key
      if (level == 0) { // 以后是最初一层了，返回以后节点
        return x;
      } else {
        level--; // 走到下一层
      }
    } else {
      x = next; // 如果next节点小于申请key，那么沿着以后层走到next节点
    }
  }
}

Golang
https://github.com/syndtr/gol… 在skiplist中查问小于key的最大节点的办法如下

func (p *DB) findLT(key []byte) int {
    node := 0 // head节点
    h := p.maxHeight - 1 // 以后层高
    for {
        next := p.nodeData[node+nNext+h]  // 求以后节点的下一个节点
        o := p.nodeData[next]   // next节点在nodeData中的数据偏移
        if next == 0 || p.cmp.Compare(p.kvData[o:o+p.nodeData[next+nKey]], key) >= 0 {// 如果next节点为空或 next节点大于等于申请key
            if h == 0 {// 以后是最初一层了，返回以后节点
                break
            }
            h-- // 走到下一层
        } else {
            node = next // 如果next节点小于申请key，那么沿着以后层走到next节点
        }
    }
    return node
}

node + nNext + i 咱们能够当作成C++代码中链表构造skiplist中第node个节点在第i层，而后p.nodeData[node+nNext+i] 当成 node->Next(i)，
p.nodeData[next+nKey]获取key的长度，从而p.kvData[o:o+p.nodeData[next+nKey]] 获取key对应的实在数据

这里获取key的操作能够封装为一个独自的办法，进步代码的可读性

查问最初一个节点
从高层到底层，判断next是否是空（即以后层list曾经是否到最初一个节点了），

如果不为空，就持续跳到下一个节点
如果为空，挪动到下一层

C++
https://github.com/google/lev…

template <typename Key, class Comparator>
typename SkipList<Key, Comparator>::Node* SkipList<Key, Comparator>::FindLast()
    const {
  Node* x = head_;
  int level = GetMaxHeight() - 1;
  while (true) {
    Node* next = x->Next(level);
    if (next == nullptr) { // 如果next节点是空，就移到下一层
      if (level == 0) {
        return x;
      } else {
        // Switch to next list
        level--;
      }
    } else { // 如果以后next不是空，移到next
      x = next;
    }
  }
}

Golang
https://github.com/syndtr/gol…

func (p *DB) findLast() int {
    node := 0
    h := p.maxHeight - 1
    for {
        next := p.nodeData[node+nNext+h] // 获取 next
        if next == 0 { // next是空，就走到下一层
            if h == 0 {
                break
            }
            h--
        } else {
            node = next  // 挪动到next 
        }
    }
    return node
}

node + nNext + i 咱们能够当作成C++代码中链表构造skiplist中第node个节点在第i层，而后p.nodeData[node+nNext+i] 当成 node->Next(i)，

插入
这里借用 https://www.bookstack.cn/read… 的示例图

在查找的过程中，一直记录每一层的后任节点，如图中红色圆圈所示意的；
为新插入的节点随机产生层高（随机产生层高的算法较为简单，依赖最高层数和概率值p，可见下文中的代码实现）；
在适合的地位插入新节点（例如图中节点12与节点19之间），并根据查找时记录的后任节点信息，在每一层中，以链表插入的形式，将该节点插入到每一层的链接中。

C++
https://github.com/google/lev…

template <typename Key, class Comparator>
void SkipList<Key, Comparator>::Insert(const Key& key) {
  // TODO(opt): We can use a barrier-free variant of FindGreaterOrEqual()
  // here since Insert() is externally synchronized.
  Node* prev[kMaxHeight];// 用于记录遍历过程中每一层的前一个节点
  Node* x = FindGreaterOrEqual(key, prev); // 找到插入点

  // Our data structure does not allow duplicate insertion
  assert(x == nullptr || !Equal(key, x->key));

// 为要插入的点生成一个随机层高
  int height = RandomHeight();
  if (height > GetMaxHeight()) {

   // 如果新生成的层高比以后skiplist的最大层高还要高，
   // 那么head节点之前在 (GetMaxHeight(), heigth] 是没有next节点的，所以要弥补下来
    for (int i = GetMaxHeight(); i < height; i++) {
      prev[i] = head_;
    }
    // It is ok to mutate max_height_ without any synchronization
    // with concurrent readers.  A concurrent reader that observes
    // the new value of max_height_ will see either the old value of
    // new level pointers from head_ (nullptr), or a new value set in
    // the loop below.  In the former case the reader will
    // immediately drop to the next level since nullptr sorts after all
    // keys.  In the latter case the reader will use the new node.
    max_height_.store(height, std::memory_order_relaxed);
  }

  x = NewNode(key, height);
  for (int i = 0; i < height; i++) {
    // NoBarrier_SetNext() suffices since we will add a barrier when
    // we publish a pointer to "x" in prev[i].
    // 将新节点所穿过的每一层链表，进行插入
    // 也就是将后面记录的每一个前向节点preNode,   x->next指向preNode->next,  而后preNode->next指向x
    x->NoBarrier_SetNext(i, prev[i]->NoBarrier_Next(i));
    prev[i]->SetNext(i, x);
  }
}

Golang
https://github.com/syndtr/gol…

// Put sets the value for the given key. It overwrites any previous value
// for that key; a DB is not a multi-map.
//
// It is safe to modify the contents of the arguments after Put returns.
func (p *DB) Put(key []byte, value []byte) error {
    p.mu.Lock()
    defer p.mu.Unlock()

    // 找到插入地位
    if node, exact := p.findGE(key, true); exact {
        // 如果key曾经存在了，
        kvOffset := len(p.kvData) // 因为上面key和value都是追加到kvData前面，所以这里记录kvData的长度，就是新节点的offset
        p.kvData = append(p.kvData, key...)//追加key的数据
        p.kvData = append(p.kvData, value...)// 追加value的数据
        p.nodeData[node] = kvOffset // 更新 offset
        m := p.nodeData[node+nVal]  // 之前 value的长度
        p.nodeData[node+nVal] = len(value) // 更新value的长度,因为key不变，所以key的长度不须要更新
        p.kvSize += len(value) - m //更新数据总长度
        return nil
    }

    // 插入新节点，获取以后节点的层高
    h := p.randHeight()

    if h > p.maxHeight {

   // 如果新生成的层高比以后skiplist的最大层高还要高，
   // 那么head节点之前在 (p.maxHeight, h] 是没有next节点的，所以要弥补下来
        for i := p.maxHeight; i < h; i++ {
            p.prevNode[i] = 0
        }
        p.maxHeight = h
    }


    kvOffset := len(p.kvData)  // 记录以后kvData的长度作为新节点的offset
    p.kvData = append(p.kvData, key...) // 追加key数据
    p.kvData = append(p.kvData, value...)// 追加value数据
    // Node
    node := len(p.nodeData)
    p.nodeData = append(p.nodeData, kvOffset, len(key), len(value), h)// 追加节点信息
    for i, n := range p.prevNode[:h] {
        m := n + nNext + i
        p.nodeData = append(p.nodeData, p.nodeData[m])  // 以后节点每一层的next指向前向节点每一层的next
        p.nodeData[m] = node // 前向节点的next指向以后节点
    }

    p.kvSize += len(key) + len(value) // 总长度
    p.n++
    return nil
}

其中这一段比拟重要，https://github.com/syndtr/gol… 是执行插入的外围代码

// Node
    node := len(p.nodeData)
    p.nodeData = append(p.nodeData, kvOffset, len(key), len(value), h)
    for i, n := range p.prevNode[:h] {
        m := n + nNext + i
        p.nodeData = append(p.nodeData, p.nodeData[m])
        p.nodeData[m] = node
    }

一开始不太容易看懂和解释，咱们把代码略微批改下,改成这个

// Node
    node := len(p.nodeData)
    p.nodeData = append(p.nodeData, kvOffset, len(key), len(value), h)
    // 增加 h 个 nextNode 的索引
    p.nodeData = append(p.nodeData, make([]int,h)...)
    for i, n := range p.prevNode[:h] {
        m := n + nNext + i
        p.nodeData[node+nNext+i] = p.nodeData[m]
        p.nodeData[m] = node
    }

这样就比拟容易解释了:

首先， m := n + nNext + i 咱们能够当作成C++代码中链表构造skiplist中第n个节点在第i层，而后p.nodeData[n+nNext+i] 当成 n->Next(i)，p.nodeData[node+nNext+i] 当成 node->Next(i)，那么下面循环中的局部就变成了

for i, n := range p.prevNode[:h] {
        m := n + nNext + i
        node.Next(i) = n->Next(i)//以后节点指向pre节点的下一个节点
        n->Next(i) = node // pre节点的下一个节点变成以后节点
    }

删除
删除只有goleveldb提供了这个办法，也比较简单

Golang
https://github.com/syndtr/gol…

func (p *DB) Delete(key []byte) error {
    p.mu.Lock()
    defer p.mu.Unlock()

    node, exact := p.findGE(key, true)// 查找删除点
    if !exact {
        return ErrNotFound
    }

    h := p.nodeData[node+nHeight]
    for i, n := range p.prevNode[:h] {
        m := n + nNext + i
        p.nodeData[m] = p.nodeData[p.nodeData[m]+nNext+i] // 每一层前向节点的next指向 以后节点的next
    }

    p.kvSize -= p.nodeData[node+nKey] + p.nodeData[node+nVal]
    p.n--
    return nil
}

n + nNext + i 咱们能够当作成C++代码中链表构造skiplist中第node个节点在第i层，而后p.nodeData[node+nNext+i] 当成 node->Next(i)，
p.nodeData[m] = p.nodeData[p.nodeData[m]+nNext+i] 就能够了解为 node->Next(i) = node->Next(i)->next(i)，实现删除

Golang版本优化
在钻研goleveldb的时候，发现了一些能够优化的中央，如下：

删除优化
在删除节点的时候，goleveldb中遍历每一层的前向节点preNode，而后让 preNode.next = preNode.next.next 链表删除 , 其实前向节点的next节点就是以后节点，所以能够改为 preNode.next = cur.next 也就是

p.nodeData[m] = p.nodeData[node+nNext+i]

插入优化
插入节点的时候，不论key是不是曾经存在了，都是间接在kvData中追加数据，其实这里能够进一步优化，如果新的value长度等于原先的value，或者新的value的长度小于原先的value的长度，都能够复用之前的内存空间。

原goleveldb的性能测试代码中插入的value都是nil，所以这里咱们新写一个测试函数性能测试代码

func BenchmarkPutRandomKV(b *testing.B) {
    buf := make([][4]byte, b.N)
    value := make([][]byte, b.N)
    for i := range buf {
        tmp := uint32(rand.Int()) % 100000
        binary.LittleEndian.PutUint32(buf[i][:], tmp)
        value[i] = make([]byte, tmp)
    }

    b.ResetTimer()
    p := New(comparer.DefaultComparer, 0)
    for i := range buf {
        p.Put(buf[i][:], value[i][:])
    }
}

优化之前

goos: linux
goarch: amd64
pkg: github.com/syndtr/goleveldb/leveldb/memdb
cpu: Intel(R) Core(TM) i7-9700 CPU @ 3.00GHz
BenchmarkPutRandomKV-8         20180         53760 ns/op      273344 B/op          0 allocs/op
PASS
ok      github.com/syndtr/goleveldb/leveldb/memdb   2.000s

优化之后

goos: linux
goarch: amd64
pkg: github.com/syndtr/goleveldb/leveldb/memdb
cpu: Intel(R) Core(TM) i7-9700 CPU @ 3.00GHz
BenchmarkPutRandomKV-8         29634         51464 ns/op      232687 B/op          0 allocs/op
PASS
ok      github.com/syndtr/goleveldb/leveldb/memdb   2.264s

这个性能的晋升次要跟key反复的概率，以及新插入的value小于之前value的概率无关。

进一步思考
goleveldb 对于删除节点在 KvData 中占用的空间是始终占用没有失去复用的，这里也能够想方法进行复用，就留给读者思考了。

序幕
本文到这里基本上把leveldb中memdb的外围代码讲完了，如果有Rust背景的同学，能够期待下篇用Rust重写skiplist局部。

参考资料

跳跃表 https://www.bookstack.cn/read…
跳跃链表 https://www.cnblogs.com/s-lis…

关于c++:leveldb-memdb源码分析上

评论

发表回复取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

关于c++:leveldb-memdb源码分析上

评论

发表回复 取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

发表回复取消回复