leveldb memdb源码分析（上）

本文主要是介绍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/leveldb/blob/master/db/skiplist.h#L41

这里主要列出关键的成员变量，详细的可以去看源码：

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/goleveldb/blob/master/leveldb/memdb/memdb.go#L183

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/leveldb/blob/master/db/skiplist.h#L260

在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/goleveldb/blob/master/leveldb/memdb/memdb.go#L211
在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/leveldb/blob/master/db/skiplist.h#L283
在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/goleveldb/blob/master/leveldb/memdb/memdb.go#L238在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++
http://github.com/google/leveldb/blob/master/db/skiplist.h#L305

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/goleveldb/blob/master/leveldb/memdb/memdb.go#L256

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/Leveldb-handbook/spilt.1.2630b6c305deecd4.md 的示例图

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

C++

https://github.com/google/leveldb/blob/master/db/skiplist.h#L337

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/goleveldb/blob/master/leveldb/memdb/memdb.go#L277

// 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/goleveldb/blob/master/leveldb/memdb/memdb.go#L303 是执行插入的核心代码

// 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/goleveldb/blob/master/leveldb/memdb/memdb.go#L321

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         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部分。

参考资料

跳跃表
跳跃链表

这篇关于leveldb memdb源码分析（上）的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！