LevelDB 源码分析（七）：Skiplist

template<typename Key, class Comparator>
class SkipList {
    private:
        struct Node;

    public:
        // 创建一个skiplist。cmp是比较函数。利用arena分配内存
        explicit SkipList(Comparator cmp, Arena* arena);

        // 插入新值到skiplist中
        // REQUIRES: nothing that compares equal to key is currently in the list.
        void Insert(const Key& key);

        // 判断key是否存在
        bool Contains(const Key& key) const;

    // skiplist的迭代器
    class Iterator {
        public:
            // 在特定的skiplist上初始化一个迭代器
            // The returned iterator is not valid.
            explicit Iterator(const SkipList* list);

            // 迭代器是否定位在一个有效的节点上
            bool Valid() const;

            // 返回当前位置的key
            const Key& key() const;

            // 进行到下一个位置
            void Next();

            // 进行到前一个位置
            void Prev();

            // 进行到 key >= target的第一个位置
            void Seek(const Key& target);

            // 定位到头节点
            void SeekToFirst();

            // 定位到尾节点
            void SeekToLast();

            private:
                const SkipList* list_;
                Node* node_;
            // Intentionally copyable
    };
    private:
        enum { kMaxHeight = 12 };   // skiplist的最大层数，新增节点的随机层数不能大于此值

        Comparator const compare_; // key值的比较函数，一旦初始化就不能变化了（当插入一些数据后，改变key，状态不可控）
        Arena* const arena_;       // Arena 用来给结点分配内存

        Node* const head_;         // skiplist的头结点

        // 只会被 Insert() 修改。Read racily by readers, but stale values are ok.
        // AtomicPointer是leveldb定义的一个原子指针，
        // 它的读取和写入都设立了内存屏障，保证读取的值是即时的、最新的。
        // 这里直接将int型转化为指针保存，因为不会对其取地址。
        port::AtomicPointer max_height_;   // 当前skiplist的最大层数

        // 返回Skiplist的层数 
        inline int GetMaxHeight() const {   
            return static_cast<int>(reinterpret_cast<intptr_t>(max_height_.NoBarrier_Load()));
        }

        // 产生随机的level层数，只有执行Insert()时才会被调用
        Random rnd_;

        Node* NewNode(const Key& key, int height);// 新建一个level=height，值为key的节点  
        int RandomHeight();                       // 随机产生一个level层数
      
        // 比较两个key是否相等
        bool Equal(const Key& a, const Key& b) const { return (compare_(a, b) == 0); }

        // key值是否比节点n的key值大
        bool KeyIsAfterNode(const Key& key, Node* n) const;

        // If prev is non-NULL, fills prev[level] with pointer to previous node at "level" for every level in [0..max_height_-1].
        // 找到key对应的Node或是key后面紧邻的Node  
        Node* FindGreaterOrEqual(const Key& key, Node** prev) const;

        // 找到key前面紧邻的Node，如果skiplist为空，则返回头节点
        Node* FindLessThan(const Key& key) const;

        // Skiplist最后一个Node，如果skiplist为空，则返回头节点
        Node* FindLast() const;

        // 拷贝构造和赋值构造操作不允许  
        SkipList(const SkipList&);
        void operator=(const SkipList&);
};

template<typename Key, class Comparator>
struct SkipList<Key,Comparator>::Node {    // Skiplist节点Node定义 
    explicit Node(const Key& k) : key(k) { }

    Key const key;

    // Accessors/mutators for links.  Wrapped in methods so we can
    // add the appropriate barriers as necessary.
    // 返回本节点第n层的下一个节点指针
    Node* Next(int n) {
        assert(n >= 0);
        // Use an 'acquire load' so that we observe a fully initialized
        // version of the returned Node.
        return reinterpret_cast<Node*>(next_[n].Acquire_Load());
    }
    // 重设n层的next指针
    void SetNext(int n, Node* x) {
        assert(n >= 0);
        // Use a 'release store' so that anybody who reads through this pointer observes a fully initialized version of the inserted node.
        next_[n].Release_Store(x);
    }

    // No-barrier variants that can be safely used in a few locations.
    Node* NoBarrier_Next(int n) {
        assert(n >= 0);
        return reinterpret_cast<Node*>(next_[n].NoBarrier_Load());
    }
    void NoBarrier_SetNext(int n, Node* x) {
        assert(n >= 0);
        next_[n].NoBarrier_Store(x);
    }

    private:
        // 数组的长度等于节点的层数。next_[0] 表示最低的一层。
        port::AtomicPointer next_[1];
};

//建一个Node节点（指定key及level层数）
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);
}

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() {
    // 不需要使用 prev 链接，只要寻找 key 的前一个节点就行了
    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);
}

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() {
    // Increase height with probability 1 in kBranching
    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;
}

// key值是否比节点n的值大。如果节点n是NULL，则值表示无穷大
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;  // 首先获得跳表的最高层数，减一是数组next最大下标
    while (true) {
        Node* next = x->Next(level);
        if (KeyIsAfterNode(key, next)) {   // 如果key在本节点node之后，继续前进  
            x = next;    // 继续找
        } else { 
            // 如果小于本节点node，把本节点的level层上的前节点指针记录进数组prev中，并跳向第一层的链表，
            // 重复上述过程，直至来到最底层
            if (prev != NULL) 
                prev[level] = x; // 当前level上，x为高度>=key节点高度，且正好排在其前面，插入和删除时使用  
            if (level == 0) {      
                return next;
            } else {
                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) { // 从最高level尽可能向后移动更远的距离
            if (level == 0) {     // 后面key>查找的key时，或next为空时，level--，直到level=0  
                return x;
            } else {
                level--;   // 从最高层往下后续查找 
            }
        } else {
            x = next;
        }
    }
}

template<typename Key, class Comparator>
typename SkipList<Key,Comparator>::Node* SkipList<Key,Comparator>::FindLast()const {
    //从最高level走到头，然后减少level继续走到头，一直到level=0  
    Node* x = head_;
    int level = GetMaxHeight() - 1;
    while (true) {
        Node* next = x->Next(level);
        if (next == NULL) {
            if (level == 0) {
                return x;
            } else {
                level--;
            }
        } else {
            x = next;
        }
    }
}

template<typename Key, class Comparator>
SkipList<Key,Comparator>::SkipList(Comparator cmp, Arena* arena):compare_(cmp),arena_(arena),
                                   head_(NewNode(0,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) {
   // 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);   //查找待插入的位置

   // 不允许插入相同的数据
   assert(x == NULL || !Equal(key, x->key));

    int height = RandomHeight();
    if (height > GetMaxHeight()) {
        for (int i = GetMaxHeight(); i < height; i++) {
            prev[i] = head_;
        }
        //fprintf(stderr, "Change height from %d to %d\n", max_height_, height);

        // 并发的读操作可以在没有同步的情况下修改 max_height_。
        // 如果一个并发的读要观察 max_height_ 的新值，会看到两种情况中的一种：
        // (1)来自head_的新level指针的旧值，也就是NULL
        // (2)或者是下面的循环设置的新值
        // 在第一种情况中，读操作会直接下降到下一层，因为NULL排在所有key的后面
        // 在第二种情况中，读操作会使用这个新的节点
        max_height_.NoBarrier_Store(reinterpret_cast<void*>(height));
    }

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

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