天天看點

leveldb(五):SkipList跳表

跳表原理的簡單介紹

跳表是平衡樹的一種替代的資料結構,但是和紅黑樹不相同的是,跳表對于樹的平衡的實作是基于一種随機化的算法的,這樣也就是說跳表的插入和删除的工作是比較簡單的。

下面來研究一下跳表的核心思想:

先從連結清單開始,如果是一個簡單的連結清單,那麼我們知道在連結清單中查找一個元素I的話,需要将整個連結清單周遊一次。

如果是說連結清單是排序的,并且節點中還存儲了指向前面第二個節點的指針的話,那麼在查找一個節點時,僅僅需要周遊N/2個節點即可。

leveldb(五):SkipList跳表

跳表的核心思想,其實也是一種通過“空間來換取時間”的一個算法,通過在每個節點中增加了向前的指針,進而提升查找的效率。

跳表的資料存儲模型

我們定義:

如果一個基點存在k個向前的指針的話,那麼陳該節點是k層的節點。

一個跳表的層MaxLevel義為跳表中所有節點中最大的層數。

下面給出一個完整的跳表的圖示:

leveldb(五):SkipList跳表

那麼我們該如何将該資料結構使用二進制存儲呢?通過上面的跳表的很容易設計這樣的資料結構:

定義每個節點類型:

// 這裡僅僅是一個指針

typedef struct nodeStructure *node;

typedef struct nodeStructure

{

keyType key; // key值

valueType value; // value值

// 向前指針數組,根據該節點層數的

// 不同指向不同大小的數組,指向的是不同層的後面節點

node forward[1];

};

我們在來看看leveldb中跳表節點的定義:

template<typename Key, class Comparator>
struct SkipList<Key,Comparator>::Node {
  explicit Node(const Key& k) : key(k) { }

  Key const key;

  Node* Next(int n) {
    return reinterpret_cast<Node*>(next_[n].Acquire_Load());
  }
  void SetNext(int n, Node* x) {
    next_[n].Release_Store(x);
  }

  Node* NoBarrier_Next(int n) {
    return reinterpret_cast<Node*>(next_[n].NoBarrier_Load());
  }
  void NoBarrier_SetNext(int n, Node* x) {
    next_[n].NoBarrier_Store(x);
  }

 private:
  port::AtomicPointer next_[];
}
           

總體上與我們上面所說的存儲模型結構一樣,隻不過這裡的next_用了AtomicPointer ,關于AtomicPointer 請參考前面寫的部落格,主要起到同步的作用。Next的函數是為了讀取第n層的下一個節點,而SetNext則是設定第n層的下一個節點是誰。

再來看下leveldb跳表的定義:

class SkipList {
 private:
  struct Node;
 private:
  enum { kMaxHeight =  };//跳表最高多少層
  Comparator const compare_;//比較節點key大小的
  Arena* const arena_;    //為節點配置設定空間用的
  Node* const head_;  //指向表頭

  // Modified only by Insert().  Read racily by readers, but stale
  // values are ok.
  port::AtomicPointer max_height_;   //目前跳表最高多少層
  Random rnd_; //用來生成每個節點層數的随機數生成器
  ...
}
           

跳表的代碼實作分析

template<typename Key, class Comparator>
SkipList<Key,Comparator>::SkipList(Comparator cmp, Arena* arena)
    : compare_(cmp),
      arena_(arena),
      head_(NewNode( , kMaxHeight)),//頭部節點初始化為具有最高層數的節點
      max_height_(reinterpret_cast<void*>()),//但目前跳表高度初始化為1,就是隻有最下面一層
      rnd_() {
  for (int i = ; i < kMaxHeight; i++) {
    head_->SetNext(i, NULL);
  }
}

template<typename Key, class Comparator>
typename SkipList<Key,Comparator>::Node*
SkipList<Key,Comparator>::NewNode(const Key& key, int height) {
  char* mem = arena_->AllocateAligned(
      sizeof(Node) + sizeof(port::AtomicPointer) * (height - ));
  //這種配置設定方式很有特色
  return new (mem) Node(key);
}
           

插入操作

template<typename Key, class Comparator>
void SkipList<Key,Comparator>::Insert(const Key& key) {
  Node* prev[kMaxHeight];
  //從最高層往下依次尋找各層大于或等于key的節點,并把各層前一個節點儲存在prev中便于插入
  Node* x = FindGreaterOrEqual(key, prev);
  //随機生成幾點層數,RandomHeight保證不會超過跳表規定的最高層kMaxHeight
  int height = RandomHeight();
  if (height > GetMaxHeight()) {
  //把大于原來跳表高度的層的prev節點設為頭,最後更新跳表高度
    for (int i = GetMaxHeight(); i < height; i++) {
      prev[i] = head_;
    }
    max_height_.NoBarrier_Store(reinterpret_cast<void*>(height));
  }

  x = NewNode(key, height);
  for (int i = ; 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));
    //這裡使用了記憶體屏障,保證了下面這條setnext一定在上面那個setnext後發生,
    //想想如果後面這個先發生會發生什麼,先将prev的下一個節點設為了newNode,但
    //newNode此時的next為null,無法繼續讀後面的節點,是以一定是先修改了newNode
    //的next後在把newNode插入prev。
    prev[i]->SetNext(i, x);
  }
  }
           

因為leveldb的跳表是作為memtable使用的,上層隻有插入沒有删除操作,因為上層的删除操作其實也就是插入操作而以,是以這裡就不分析delete操作,其實原理與insert差不多。

其他操作也比較簡單,就不分析了。

跳表的疊代器

class Iterator 
{
   public:
    explicit Iterator(const SkipList* list);
    bool Valid() const;
    const Key& key() const; //傳回疊代器所處位置節點的值
    void Next();
    void Prev();
    void Seek(const Key& target);
    void SeekToFirst();//到跳表頭
    void SeekToLast();//到尾

   private:
    const SkipList* list_;//疊代器所疊代的跳表
    Node* node_;//目前疊代器所處位置的節點
  }
           

實作也比較簡單,看名字就知道啥意思了。

template<typename Key, class Comparator>
inline void SkipList<Key,Comparator>::Iterator::Seek(const Key& target) {
  node_ = list_->FindGreaterOrEqual(target, NULL);
}
           

seek就是在跳表中找到第一個key大于或等于target的元素。