LevelDB源码分析：skiplist

LevelDB中的skiplist实现方式基本上和中的实现方式类似。它向外暴露接口非常简单，如下：

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);

  // 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

private成员变量：

private:
  enum { kMaxHeight = 12 };

  // 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.
  port::AtomicPointer max_height_;   

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

一.构造函数

templateSkipList::SkipList(Comparator cmp, Arena* arena)
    : compare_(cmp),
      arena_(arena),
      head_(NewNode(0 /* any key will do */, kMaxHeight)),
      max_height_(reinterpret_cast(1)),
      rnd_(0xdeadbeef) {
  for (int i = 0; i < kMaxHeight; i++) {
    head_->SetNext(i, NULL);
  }
}

重点注意下head_和rnd_的初始化，NewNode方法如下。

templatetypename SkipList::Node*
SkipList::NewNode(const Key& key, int height) {
  char* mem = arena_->AllocateAligned(
      sizeof(Node) + sizeof(port::AtomicPointer) * (height - 1));
  return new (mem) Node(key);
}

这里为什么是“height-1”详见：LevelDb源码分析之五：skiplist（1）。

new (mem) Node(key)使用了placement new技巧，详见：C++中使用placement new

rnd_是一个Random类型的变量，使用0xdeadbeef进行初始化，Random详见LevelDB源码分析之七：Random

二.插入函数

templatevoid SkipList::Insert(const Key& key) {
  // TODO(opt): We can use a barrier-free variant of FindGreaterOrEqual()
  // here since Insert() is externally synchronized.
  // prev记录的是查询路径，下面需要使用prev来修改新生成结点的指针  
  // prev相当于LevelDb源码分析之五：skiplist（1）中的update
  // 整个流程与LevelDb源码分析之五：skiplist（1）相似，这里不再详细解释
  Node* prev[kMaxHeight];
  // 返回大于等于key的结点或者NULL，原因详见FindGreaterOrEqual的分析
  Node* x = FindGreaterOrEqual(key, prev);

  // Our data structure does not allow duplicate insertion
  // 不允许插入重复的值  
  assert(x == NULL || !Equal(key, x->key));
  // 产生一个随机层数height
  int height = RandomHeight();
  // 如果height大于原最大层数，则更新prev，并更新最大层数
  if (height > GetMaxHeight()) {
    for (int i = GetMaxHeight(); i < height; i++) {
      prev[i] = head_;
    }
    //fprintf(stderr, "Change height from %d to %dn", max_height_, height);

    // 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_ (NULL), or a new value set in
    // the loop below.  In the former case the reader will
    // immediately drop to the next level since NULL sorts after all
    // keys.  In the latter case the reader will use the new node.
    max_height_.NoBarrier_Store(reinterpret_cast(height));
  }
  // 创建一个待插入的结点x，从低到高一层层插入
  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);
  }
}

插入函数里调用了私有函数FindGreaterOrEqual。FindGreaterOrEqual中完成查询操作，就是向下（level控制）和向右（x控制）移动过程，并不断将经过路径保存到参数prev中。

templatetypename SkipList::Node* SkipList::FindGreaterOrEqual(const Key& key, Node** prev)
    const {
  Node* x = head_;
  int level = GetMaxHeight() - 1;
  // 从最高层往下，每层都查找插入位置，当遍历到该层的尾部（x->next[level]==NULL）  
  // 也没有找到比key大的结点时，将该层的最后一个结点的指针放到prev[level]中。  
  // 如果在某层中找到了比key大或等于key的结点时，将该结点之前的那个比key小的结点的指针  
  // 放到prev[level]中。  
  while (true) {
    Node* next = x->Next(level);
    if (KeyIsAfterNode(key, next)) {
      // Keep searching in this list
      x = next;
    } else {
      if (prev != NULL) prev[level] = x;
	  // 当查到第一层时，有两种情况：
	  // 1.第一层中有满足要求的结点，此时next刚好是不小于key的那个结点
	  // 2.第一层中没有满足要求的结点，此时实际上到了尾部，next=NULL
      if (level == 0) {
        return next;
      } else {
        // Switch to next list
        level--;
      }
    }
  }
}

三.查找函数

查找操作基本上就是调用函数上面的函数FindGreaterOrEqual实现。

templatebool SkipList::Contains(const Key& key) const {
  Node* x = FindGreaterOrEqual(key, NULL);
  if (x != NULL && Equal(key, x->key)) {
    return true;
  } else {
    return false;
  }
}

需要注意的是，LevelDB中没有提供显式的删除节点操作，但实际上是可以删除的，因为当我们插入数据时，key的形式为key:value，当删除数据时，则插入key:deleted类似删除的标记，等到Compaction再删除。