关于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 搜寻十分相似，其中要害的差异在于

1. GE 搜寻返回的是 next 节点，LT 返回的是以后 node
2. 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…