LevelDB源码剖析之SkipList

SkipList的基本原理及具体实现可参考：/article/6798166.html（此文中SkipList的实现方式是把多层单链表叠加成一个SkipList链表，SkipList的每个节点中包含一个指向下一个Node的指针数组成员）。如图1：



图1

另外一种实现方式可见：/article/3965793.html（这种方式中SkipList实际是由相对独立的多层单链表构成，每层单链表的节点除了有next指针指向自己的后续节点外，还有一个down指针用来指向自己的下一层链表的对应节点，并以此实现各层链表之间的联系。）如图2：



图2

尽管具体实现方式不一，但其基本思想是一致的：以空间（增加节点的指针数）换取时间。

LevelDB的SkipList实现采用第一种方式：每个节点包含一个前向指针数组，指向后继节点。 以内部类实现节点和迭代器。空间分配器为Arena

[cpp] view
plaincopy

template<typename Key, class Comparator>

class SkipList {

private:

struct Node;//链表节点

public:

<span style="font-size: 18px; font-family: Arial, Helvetica, sans-serif;"> //构造函数中 构造头节点，并设置头结点的前向指针数组</span>

[cpp] view
plaincopy

explicit SkipList(Comparator cmp, Arena* arena);//在*arena析构时才会释放分配的空间，

//而*arena作为SkipList的局部变量，因此在SkipList析构时自动析构，释放空间

void Insert(const Key& key);//要求：当前链表中没有与准备插入相等的key

bool Contains(const Key& key) const;//是否包含key的节点

[cpp] view
plaincopy

class Iterator;//迭代器类

private:

enum { kMaxHeight = 12 };

Comparator const compare_;//比较器

Arena* const arena_; // 空间分配器

Node* const head_; //头结点

port::AtomicPointer max_height_; // 链表节点最大层数

inline int GetMaxHeight() const {//返回允许最大层数

return reinterpret_cast<intptr_t>(max_height_.NoBarrier_Load());

}

Random rnd_;//随机数生成器

Node* NewNode(const Key& key, int height);//产生新节点

int RandomHeight();//生成随机层数，基数为1，以四分之一概率++

bool Equal(const Key& a, const Key& b) const { return (compare_(a, b) == 0); }

bool KeyIsAfterNode(const Key& key, Node* n) const;//key是否比note->key大

Node* FindGreaterOrEqual(const Key& key, Node** prev) const; //查找大等于key的最近值

Node* FindLessThan(const Key& key) const;//查找小于key的最大节点，没有则返回头节点

Node* FindLast() const;//返回尾节点

SkipList(const SkipList&);//禁止拷贝构造或复制操作

void operator=(const SkipList&);

};

迭代器实现代码：

[cpp] view
plaincopy

/****************************************************************/

//迭代器类 ：数据成员 包含一个要操作的SkipList指针，以及一个当前节点note指针

class Iterator {

public:

explicit Iterator(const SkipList* list);

bool Valid() const;//当前节点是否有效

const Key& key() const;//返回当前节点的key

void Next();//移动到后继节点

void Prev();//移动到前一节点（实际是通过查找*list中小于当前key的最大节点）

void Seek(const Key& target);//移动到第一个大等于target的节点处

void SeekToFirst();//移动到链表第一个节点（header的next）

void SeekToLast();//移动到链表尾节点

private:

const SkipList* list_; //迭代器操作的链表

Node* node_;//当前迭代器指向的节点

};

/****************************************************************/<pre name="code" class="cpp">/***********************************************************/

[cpp] view
plaincopy

<span style="font-family: Arial, Helvetica, sans-serif; font-size: 12px;">

</span>

[cpp] view
plaincopy

<span style="font-family: Arial, Helvetica, sans-serif; font-size: 12px;">

</span>

[cpp] view
plaincopy

<span style="font-family: Arial, Helvetica, sans-serif; font-size: 12px;">SkipList 方法实现代码：</span>

[cpp] view
plaincopy

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

return new (mem) Node(key);//布局new

}

template<typename Key, class Comparator>

inline SkipList<Key,Comparator>::Iterator::Iterator(const SkipList* list) {

list_ = list;

node_ = NULL;

}

template<typename Key, class Comparator>

inline bool SkipList<Key,Comparator>::Iterator::Valid() const {

return node_ != NULL;

}

template<typename Key, class Comparator>

inline const Key& SkipList<Key,Comparator>::Iterator::key() const {

assert(Valid());

return node_->key;

}

template<typename Key, class Comparator>

inline void SkipList<Key,Comparator>::Iterator::Next() {

assert(Valid());

node_ = node_->Next(0);

}

template<typename Key, class Comparator>

inline void SkipList<Key,Comparator>::Iterator::Prev() {

assert(Valid());

node_ = list_->FindLessThan(node_->key);

if (node_ == list_->head_) {

node_ = NULL;

}

}

template<typename Key, class Comparator>

inline void SkipList<Key,Comparator>::Iterator::Seek(const Key& target) {

node_ = list_->FindGreaterOrEqual(target, NULL);//查找大等于key的最近值

}

template<typename Key, class Comparator>

inline void SkipList<Key,Comparator>::Iterator::SeekToFirst() {

node_ = list_->head_->Next(0);

}

template<typename Key, class Comparator>

inline void SkipList<Key,Comparator>::Iterator::SeekToLast() {

node_ = list_->FindLast();

if (node_ == list_->head_) {

node_ = NULL;

}

}

template<typename Key, class Comparator>

int SkipList<Key,Comparator>::RandomHeight() {

static const unsigned int kBranching = 4;

int height = 1;

while (height < kMaxHeight && ((rnd_.Next() % kBranching) == 0)) {

height++;

}

assert(height > 0);

assert(height <= kMaxHeight);

return height;

}

template<typename Key, class Comparator>

bool SkipList<Key,Comparator>::KeyIsAfterNode(const Key& key, Node* n) const {

return (n != NULL) && (compare_(n->key, key) < 0);

}

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)) {//如果key在该节点的后面

x = next;

} else {

if (prev != NULL) prev[level] = x;//把每层的最后转折节点指针放入prev

if (level == 0) {

return next;//最后需要返回的值在最底层时返回

} else {

// Switch to next list

level--;

}

}

}

}

template<typename Key, class Comparator>

typename SkipList<Key,Comparator>::Node*

SkipList<Key,Comparator>::FindLessThan(const Key& key) const {

Node* x = head_;

int level = GetMaxHeight() - 1;

while (true) {

assert(x == head_ || compare_(x->key, key) < 0);

Node* next = x->Next(level);

if (next == NULL || compare_(next->key, key) >= 0) {

if (level == 0) {

return x;

} else {

// Switch to next list

level--;

}

} else {

x = next;

}

}

}

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 == NULL) {

if (level == 0) {

return x;

} else {

// Switch to next list

level--;

}

} else {

x = next;

}

}

}

template<typename Key, class Comparator>

SkipList<Key,Comparator>::SkipList(Comparator cmp, Arena* arena)

: compare_(cmp),

arena_(arena),

head_(NewNode(0 /* any key will do */, kMaxHeight)),

max_height_(reinterpret_cast<void*>(1)),

rnd_(0xdeadbeef) {

for (int i = 0; i < kMaxHeight; i++) {

head_->SetNext(i, NULL);

}

}

template<typename Key, class Comparator>

void SkipList<Key,Comparator>::Insert(const Key& key) {

Node* prev[kMaxHeight];//前向指针数组 max_height_：当前最大层数

Node* x = FindGreaterOrEqual(key, prev);

// 不允许重复插入

assert(x == NULL || !Equal(key, x->key));

int height = RandomHeight();//随机产生插入节点的层数

if (height > GetMaxHeight()) {//插入节点层数大于允许的最大层数，把新节点的新增前向指针指向头结点

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 = 0; i < height; i++) {

x->NoBarrier_SetNext(i, prev[i]->NoBarrier_Next(i));

prev[i]->SetNext(i, x);

}

}

template<typename Key, class Comparator>

bool SkipList<Key,Comparator>::Contains(const Key& key) const {

Node* x = FindGreaterOrEqual(key, NULL);

if (x != NULL && Equal(key, x->key)) {

return true;

} else {

return false;

}

}

}