LevelDB源碼分析之六：skiplist（2）

閱讀本文可參考：

LevelDB源碼分析之一：coding

LevelDB源碼分析之二：comparator

LevelDB源碼分析之三：arena

LevelDB源碼分析之四：AtomicPointer

LevelDb源碼分析之五：skiplist（1）

LevelDB源碼分析之七：Random

LevelDB中的skiplist實(shí)現(xiàn)方式基本上和中的實(shí)現(xiàn)方式類似。它向外暴露接口非常簡(jiǎn)單，如下：

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

一.構(gòu)造函數(shù)

template
SkipList::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);
  }
}

重點(diǎn)注意下head_和rnd_的初始化，NewNode方法如下。

template
typename 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_是一個(gè)Random類型的變量，使用0xdeadbeef進(jìn)行初始化，Random詳見LevelDB源碼分析之七：Random

二.插入函數(shù)

template
void SkipList::Insert(const Key& key) {
  // TODO(opt): We can use a barrier-free variant of FindGreaterOrEqual()
  // here since Insert() is externally synchronized.
  // prev記錄的是查詢路徑，下面需要使用prev來修改新生成結(jié)點(diǎn)的指針  
  // prev相當(dāng)于LevelDb源碼分析之五：skiplist（1）中的update
  // 整個(gè)流程與LevelDb源碼分析之五：skiplist（1）相似，這里不再詳細(xì)解釋
  Node* prev[kMaxHeight];
  // 返回大于等于key的結(jié)點(diǎn)或者NULL，原因詳見FindGreaterOrEqual的分析
  Node* x = FindGreaterOrEqual(key, prev);

  // Our data structure does not allow duplicate insertion
  // 不允許插入重復(fù)的值  
  assert(x == NULL || !Equal(key, x->key));
  // 產(chǎn)生一個(gè)隨機(jī)層數(shù)height
  int height = RandomHeight();
  // 如果height大于原最大層數(shù)，則更新prev，并更新最大層數(shù)
  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));
  }
  // 創(chuàng)建一個(gè)待插入的結(jié)點(diǎn)x，從低到高一層層插入
  x = NewNode(key, height);
  // 逐層更新結(jié)點(diǎn)的指針，和普通鏈表插入一樣 
  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);
  }
}

插入函數(shù)里調(diào)用了私有函數(shù)FindGreaterOrEqual。FindGreaterOrEqual中完成查詢操作，就是向下（level控制）和向右（x控制）移動(dòng)過程，并不斷將經(jīng)過路徑保存到參數(shù)prev中。

template
typename SkipList::Node* SkipList::FindGreaterOrEqual(const Key& key, Node** prev)
    const {
  Node* x = head_;
  int level = GetMaxHeight() - 1;
  // 從最高層往下，每層都查找插入位置，當(dāng)遍歷到該層的尾部（x->next[level]==NULL）  
  // 也沒有找到比key大的結(jié)點(diǎn)時(shí)，將該層的最后一個(gè)結(jié)點(diǎn)的指針放到prev[level]中。  
  // 如果在某層中找到了比key大或等于key的結(jié)點(diǎn)時(shí)，將該結(jié)點(diǎn)之前的那個(gè)比key小的結(jié)點(diǎn)的指針  
  // 放到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;
	  // 當(dāng)查到第一層時(shí)，有兩種情況：
	  // 1.第一層中有滿足要求的結(jié)點(diǎn)，此時(shí)next剛好是不小于key的那個(gè)結(jié)點(diǎn)
	  // 2.第一層中沒有滿足要求的結(jié)點(diǎn)，此時(shí)實(shí)際上到了尾部，next=NULL
      if (level == 0) {
        return next;
      } else {
        // Switch to next list
        level--;
      }
    }
  }
}

三.查找函數(shù)

查找操作基本上就是調(diào)用函數(shù)上面的函數(shù)FindGreaterOrEqual實(shí)現(xiàn)。

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

需要注意的是，LevelDB中沒有提供顯式的刪除節(jié)點(diǎn)操作，但實(shí)際上是可以刪除的，因?yàn)楫?dāng)我們插入數(shù)據(jù)時(shí)，key的形式為key:value，當(dāng)刪除數(shù)據(jù)時(shí)，則插入key:deleted類似刪除的標(biāo)記，等到Compaction再刪除。

參考鏈接：http://blog.csdn.net/xuqianghit/article/details/6948554