LevelDB 源代码阅读（四）：MemTable

在 LevelDB 源代码阅读（一）：写流程 中，我们在介绍写入时最终讲到写入的数据被编码成 internalKey 然后被插入到 SkipList 中。在本文中，我们将结合 LevelDB 的源代码，详细介绍 MemTable 的结构。

MemTable 基本结构

代码中有关于 MemTable 的定义和实现在 memtable.{h/cc} 中，我们先从 MemTable 的定义开始了解其基础结构和功能。

// memtable.h
class MemTable {
 public:
  explicit MemTable(const InternalKeyComparator& comparator);

  void Ref() { ++refs_; }

  void Unref() {
    --refs_;
    assert(refs_ >= 0);
    if (refs_ <= 0) {
      delete this;
    }
  }

  size_t ApproximateMemoryUsage();
  Iterator* NewIterator();
  void Add(SequenceNumber seq, ValueType type, const Slice& key,
           const Slice& value);
  bool Get(const LookupKey& key, std::string* value, Status* s);
  
private:
  struct KeyComparator {
    const InternalKeyComparator comparator;
    explicit KeyComparator(const InternalKeyComparator& c) : comparator(c) {}
    int operator()(const char* a, const char* b) const;
  };

  typedef SkipList<const char*, KeyComparator> Table;

  ~MemTable();

  KeyComparator comparator_;
  int refs_;
  Arena arena_;
  Table table_;
}

MemTable 是一个有引用计数的类型，当其引用计数为 0 时（除了刚创建时）就会从内存中删除。Ref 函数和 Unref 函数就是用以管理引用计数的函数。ApproximateMemoryUsage 函数会计算当前 MemTable 的内存占用值，但是这个值不是精确的。NewIterator 函数会返回一个 Iterator 对象，用以正序遍历 MemTable 中的数据。Add 函数和 Get 的函数则分别对应写数据和读数据，注意 Add 函数既包含了新增值或修改值，也包含删除值。在数据部分，MemTable 维护了四个私有变量：comparator_、refs_、arena_ 和 table_ 。comparator_ 的定义就在 MemTable 中，它重载了 () 运算符，可以像函数一样被调用，主要用以比较 MemTable 中 key 的大小，以便对 key 进行排序。refs_ 是一个整数类型的值，用以维护引用计数。arena_ 是一个 Arena 对象，主要用于管理内存。table_ 是一个 SkipList 类，是 MemTable 中实际存储数据的结构。

在 LevelDB 源代码阅读（一）：写流程 中我们已经简要介绍过了 MemTable 的写入过程：写入的 key 和 value（删除则没有 value）以及写入操作对应的 sequence number 会编码成一个如图所示的 entry ，然后这个 entry 会被插入到 table_ 中（即 SkipList 中）。关于 SkipList 的结构以及如何在 SkipList 中插入数据，我们在下面一节中描述。

MemTable 的读操作代码如下：

// memtable.cc
bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
  Slice memkey = key.memtable_key();
  Table::Iterator iter(&table_);
  iter.Seek(memkey.data());
  if (iter.Valid()) {
    // entry format is:
    //    klength  varint32
    //    userkey  char[klength]
    //    tag      uint64
    //    vlength  varint32
    //    value    char[vlength]
    // Check that it belongs to same user key.  We do not check the
    // sequence number since the Seek() call above should have skipped
    // all entries with overly large sequence numbers.
    const char* entry = iter.key();
    uint32_t key_length;
    const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
    if (comparator_.comparator.user_comparator()->Compare(
            Slice(key_ptr, key_length - 8), key.user_key()) == 0) {
      // Correct user key
      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
      switch (static_cast<ValueType>(tag & 0xff)) {
        case kTypeValue: {
          Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
          value->assign(v.data(), v.size());
          return true;
        }
        case kTypeDeletion:
          *s = Status::NotFound(Slice());
          return true;
      }
    }
  }
  return false;
}

代码中首先声明了一个 table_ 的迭代器 iter ，然后调用迭代器的 Seek 函数，寻找 SkipList 中有对应 Key 的节点。由于 SkipList 是有序的，这里 iter 找到的节点是所有拥有待搜索键且其 sequence number 大于此次操作的 sequence number 的节点中 sequence number 最大的一个，因此其数据也是最新的。找到这个节点之后，将这个节点进行解析，获取其 key 的值以及节点对应的操作类型。如果节点对应的操作类型是 insert ，那么则返回 insert 对应的 value ；如果是 delete ，那么则返回空。

不难看出，MemTable 无论是写还是读，其涉及到的核心组件就是 SkipList 。那么 SkipList 是一种怎么样的数据结构呢，又是如何实现的呢，下面让我们一起来看一看。

跳表 SkipList

跳表的英文名称是 SkipList ，最早由 William Pugh 在论文 SkipList: A Probabilistic Alternative to Balanaced Tree 中提出。在 跳表：平衡树的概率替代方案 一文中，笔者对 SkipList 这篇论文涉及到数据结构与算法的前半部分进行了翻译，有兴趣的读者可以自行阅读论文原文或者翻译。

简单来说，SkipList 就是有序链表加上索引。如图 1a ，这是一个顺序链表。虽然所有的元素已经排好序，但是如果我们想要从中查找一个元素，还是需要将所有元素遍历一遍，直到找到我们想要的元素为止。例如我们需要查找 19，那么总共需要遍历 7 个元素。

图 1

下面我们考虑给链表增加索引：如果隔一个元素，我们就增加一个索引，使其指向下一个具有索引的节点，那我们就获得了图 1b 所示的链表。在这个链表中进行查找，首先我们从高层指针开始查找，如果高层指针指向的下一个元素的值小于等于我们的预期值，那么我们就跳转到这个指针指向的元素；否则，我们就检查下一个层级的指针是否满足条件。如果我们已经在一级指针了，那么当前所在的元素就是我们要找的元素（如果元素存在）。如在图 1b 中我们要查找 19 ，那么经过的节点顺序是 头节点 -> 6 -> 9 -> 17（17 的二级指针指向的下一个元素为 21 ，大于 19 ，因此我们转换到一级指针进行搜索）-> 19 （19 的一级指针指向的下一个元素为 21 ，大于 19 ，那么此时所在的节点就是我们要搜索的节点）。不算头节点，总共遍历了 4 个元素。

那我们还可以继续优化吗？当然可以！我们先定义一个概念，一个拥有 n 个指针的链表节点称为 n 级节点。前面所述的情况有两种节点：一级节点和二级节点。我们可以在二级指针的基础上，增加一个三级指针，相当于三级索引。假如每隔 1 个元素，就有一个二级节点，其指向下一个级别大于等于二的节点；每隔三个元素，就有一个三级节点，其指向下一个为三的节点。这样，我们就得到了图 1c 的链表。搜索的顺序同样是从高级别指针开始向低级别指针搜索，在这个链表中查找 19 的路径为：头节点 -> 9（在这里先检查三级指针指向的下一个元素，其值为 21 ，大于我们要搜索的 19 ，因此切换到二级指针进行检查） -> 17 （在这里先检查二级指针，其指向的下一个元素的值为 21 ，大于我们要搜索的 19 ，因此切换到一级指针进行检查） -> 19 （在这里检查一级指针指向的下一个元素，其值为 21 ，大于我们要搜索的 19 ，因此这个节点就是我们要搜索的节点）。加了三级指针以后，我们查找只遍历了 3 个节点。

除了三级指针，我们还可以继续加上更多级别的指针，如图 1d 就有 4 级指针，查找到 19 要遍历 3 个节点。在以上这几种情况中，拥有 50% 的结点为 1 级节点，25% 的结点为 2 级节点，12.5% 的结点为 3 级节点，6.25% 的结点为 4 级节点。我们定义一个链表的级别为其中所有节点中最高级的那个节点的级别，图 1 a-d 的四个链表分别对应了一级到四级链表。随着链表的级别升高，各种级别的节点的比例是可计算的，为 $1/2^n$ ，其中 $n$ 为节点的级别。

图 2：跳表的插入流程

如果要遵循上面这种不同级别节点出现的位置都非常规律的规定（$n$ 级节点每隔 $2^n - 1$ 个节点出现一次），那么对链表进行插入和删除是非常麻烦的。跳表和这种索引链表的区别在于，跳表去掉了对不同级别的节点出现的位置的规定，仅要求它们的比例符合规定。如图 1e 就是一个跳表，其中不同级别节点出现的位置并不规定，但其总体分布遵循索引链表的分布。在插入时，一个新的节点的级别通过随机数进行生成，这个随机数的生成的分布要遵循索引链表的节点级别分布。生成节点后，我们通过搜索算法将这个节点插入，并设置好对应级别的指针，即完成了数据的插入。删除时先搜索到对应的节点并将其移除，然后设置好其他节点的指针即可。可以证明，跳表的搜索、插入和删除算法都是 $O(log(n))$ 的，其中 $n$ 为跳表中元素的数量。

跳表的理论分析就分析到这，下面让我们来看看在 LevelDB 中跳表是如何实现的。

LevelDB 中的跳表

我们先来看看 SkipList 的声明：

// skiplist.h 
template <typename Key, class Comparator>
class SkipList {
 private:
  struct Node;

 public:
  // Create a new SkipList object that will use "cmp" for comparing keys,
  // and will allocate memory using "*arena".  Objects allocated in the arena
  // must remain allocated for the lifetime of the skiplist object.
  explicit SkipList(Comparator cmp, Arena* arena);

  SkipList(const SkipList&) = delete;
  SkipList& operator=(const SkipList&) = delete;

  // Insert key into the list.
  // REQUIRES: nothing that compares equal to key is currently in the list.
  void Insert(const Key& key);

  // Returns true iff an entry that compares equal to key is in the list.
  bool Contains(const Key& key) const;

  // Iteration over the contents of a skip list
  class Iterator {
   public:
    // Initialize an iterator over the specified list.
    // The returned iterator is not valid.
    explicit Iterator(const SkipList* list);

    // Returns true iff the iterator is positioned at a valid node.
    bool Valid() const;

    // Returns the key at the current position.
    // REQUIRES: Valid()
    const Key& key() const;

    // Advances to the next position.
    // REQUIRES: Valid()
    void Next();

    // Advances to the previous position.
    // REQUIRES: Valid()
    void Prev();

    // Advance to the first entry with a key >= target
    void Seek(const Key& target);

    // Position at the first entry in list.
    // Final state of iterator is Valid() iff list is not empty.
    void SeekToFirst();

    // Position at the last entry in list.
    // Final state of iterator is Valid() iff list is not empty.
    void SeekToLast();

   private:
    const SkipList* list_;
    Node* node_;
    // Intentionally copyable
  };

 private:
  enum { kMaxHeight = 12 };

  inline int GetMaxHeight() const {
    return max_height_.load(std::memory_order_relaxed);
  }

  Node* NewNode(const Key& key, int height);
  int RandomHeight();
  bool Equal(const Key& a, const Key& b) const { return (compare_(a, b) == 0); }

  // Return true if key is greater than the data stored in "n"
  bool KeyIsAfterNode(const Key& key, Node* n) const;

  // Return the earliest node that comes at or after key.
  // Return nullptr if there is no such node.
  //
  // If prev is non-null, fills prev[level] with pointer to previous
  // node at "level" for every level in [0..max_height_-1].
  Node* FindGreaterOrEqual(const Key& key, Node** prev) const;

  // Return the latest node with a key < key.
  // Return head_ if there is no such node.
  Node* FindLessThan(const Key& key) const;

  // Return the last node in the list.
  // Return head_ if list is empty.
  Node* FindLast() const;

  // 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_;
};

首先我们看 SkipList 暴露给外界的接口，总共只有两个：Insert 和 Contains 。因此，SkipList 实际上并不支持删除操作。在所有对外暴露的接口中，自由 Insert 可以修改 SkipList 的结构。一个 Node 一旦被插入了 SkipList 中，那么其就无法再被移除了。并且，在插入时要求跳表中不存在与待插入键相同键的节点。注意，这里的键并不是用户在插入时给出的原始键，而是包含了 sequence number 的 internalKey ，理论上每个 internalKey 都是独一无二的。Iterator 类实际上提供了遍历 SkipList 和在 SkipList 中进行搜索的功能。在 MemTable 的 Get 函数中，就是通过 Iterator 来实现的，因此也可以算作 SkipList 对外暴露的接口。当然，这个接口也不会修改 SkipList 内部节点的数据。

SkipList 内部的数据部分包含了六个对象：随机生成节点层级时的最大值 kMaxHeight 、用于比较不同键大小的比较器 compare_ 、用于管理内存的 arena_ 、跳表的头节点 head_ 、当前跳表中所有节点的最高层级 max_height_ 以及随机数生成器 rnd_ 。在第二节中我们分析过，SkipList 的插入、删除（当然 LevelDB 中不包含删除）都是基于其查找功能实现的，因此我们就从 SkipList 查找的实现开始分析其源代码。

Search

图 3：跳表搜索算法伪代码

// skiplist.h
template <typename Key, class Comparator>
typename SkipList<Key, Comparator>::Node*
SkipList<Key, Comparator>::FindGreaterOrEqual(const Key& key,
                                              Node** prev) const {
  Node* x = head_;
  int level = GetMaxHeight() - 1;
  while (true) {
    Node* next = x->Next(level);
    if (KeyIsAfterNode(key, next)) {
      // Keep searching in this list
      x = next;
    } else {
      if (prev != nullptr) prev[level] = x;
      if (level == 0) {
        return next;
      } else {
        // Switch to next list
        level--;
      }
    }
  }
}

对比 SkipList 论文中给出的搜索算法的伪代码和 LevelDB 中 SkipList 的搜索算法，两者基本上是一致的。算法首先从头节点开始，从高级别指针开始搜索，如果当前指针指向的下一个元素小于等于待搜索的键，那么就移动到下一个节点中，继续检查这一级别的指针；如果下一个节点的值大于待搜索键，那么就搜索当前节点的下一个级别的指针。如果没有下一个级别的指针了，即 level 等于 0 ，那么当前节点就是我们要搜索的节点。如果跳表中有一个节点的键与待搜索键相同，那么这就是我们查找的结果；如果不存在这样的节点，那么最终搜索到的节点的键的值会大于待搜索键。和原始算法不一样的一点是，LevelDB 的实现中加入了一个 Node** 类型的 prev 变量。如果我们将最终搜索到的节点称为 target_node ，那么 prev[i] 就会指向跳表中位于 target_node 之前、距离 target_node 最近的级别大于等于 $i+1$ 的节点。如果在 target_node 之前没有级别大于 prev[i] 的节点，那么就会指向跳表的头节点。

Insert

看完了 SkipList 的搜索过程，下面我们一起看看插入过程：

图 4：跳表插入算法伪代码

// skiplist.h
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()) {
    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].
    x->NoBarrier_SetNext(i, prev[i]->NoBarrier_Next(i));
    prev[i]->SetNext(i, x);
  }
}

对比 SkipList 论文中给出的插入算法和 LevelDB 的实现，还是可以比较清晰地看明白的。首先找到插入的位置，并且通过 prev（伪代码中为 update ）记录各个级别的前驱节点。在 LevelDB 的实现中，不允许对同一个键进行两次插入，在断言中否定了这种情况，而在 SkipList 的伪代码中是允许这种情况的。找到待插入的位置后，为该节点分配一个级别，如果这个级别高于了当前跳表的级别，那么就需要对 prev 中高于跳表级别的指针进行处理，使其指向 head_ 。完成好这些预备工作后，为当前键生成一个节点实例，然后将新节点的 $i$ 级指针设置为 $prev[i]$ 的 $i$ 级指针的值，而 $prev[i]$ 的 $i$ 级指针则更新为新生成的这个节点。处理好这些指针后，新的节点插入就算完成了。

Contains

最后我们来看看 Contains 是如何实现的：

// skiplist.h
template <typename Key, class Comparator>
bool SkipList<Key, Comparator>::Contains(const Key& key) const {
  Node* x = FindGreaterOrEqual(key, nullptr);
  if (x != nullptr && Equal(key, x->key)) {
    return true;
  } else {
    return false;
  }
}

这个实际上就比较简单了，就是调用了搜索过程的函数，然后判断搜索到的节点的键是否等于搜索键即可。

到这里，SkipList 的基本内容就讲完了。SkipList 这个数据结构实际上就是在有序链表的基础上通过增加索引形成的一个数据结构，LevelDB 中的实现基本也是按照论文中的思路进行的。建议希望深入了解 SkipList 的读者去阅读其论文原文以及相关的研究论文。