目录
概述
跳表是一种有序数据结构,它通过在每个节点维持多个指向其他节点的指针,从而达到快速访问节点的目的。
跳表基于并联的链表,其效率可比拟于二叉查找树(对于大多数操作需要O(log n)平均时间)。
插入新节点时,该节点层数随机。
更详尽的解释参见维基百科或自行google。贴两张维基百科的图片,便于大家理解。
大家有兴趣也可以去看redis的skiplist实现(zskiplist)。
SkipList定义
// Thread safety // ------------- // // Writes require external synchronization, most likely a mutex. // Reads require a guarantee that the SkipList will not be destroyed // while the read is in progress. Apart from that, reads progress // without any internal locking or synchronization. // // Invariants: // // (1) Allocated nodes are never deleted until the SkipList is // destroyed. This is trivially guaranteed by the code since we // never delete any skip list nodes. // // (2) The contents of a Node except for the next/prev pointers are // immutable after the Node has been linked into the SkipList. // Only Insert() modifies the list, and it is careful to initialize // a node and use release-stores to publish the nodes in one or // more lists. // // ... prev vs. next pointer ordering ... //线程安全:读不需要外部同步,写需要外部同步(如mutex) class SkipList { private: struct Node; 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. //skiplist用Arena做内存管理 //Comparator用来比较key,自行定义,例子可参考skiplist_test.cc 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; // Iteration over the contents of a skip list //迭代器,在此省去 class Iterator { ..... }; private: //最高12层 //《efficetive C++》尽量不要使用宏定义定义常量,取代的办法是const常量和enum类型 enum { kMaxHeight = 12 }; // Immutable after construction //const 构造后不可变 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_; // Height of the entire list inline int GetMaxHeight() const { return static_cast<int>( reinterpret_cast<intptr_t>(max_height_.NoBarrier_Load())); } // Read/written only by Insert(). //随机数生成器 Random rnd_; Node* NewNode(const Key& key, int height); int RandomHeight(); bool Equal(const Key& a, const Key& b) const { return (compare_(a, b) == 0); } // Return true if key is greater than the data stored in "n" bool KeyIsAfterNode(const Key& key, Node* n) const; // Return the earliest node that comes at or after key. // Return NULL if there is no such node. // // If prev is non-NULL, fills prev[level] with pointer to previous // node at "level" for every level in [0..max_height_-1]. Node* FindGreaterOrEqual(const Key& key, Node** prev) const; // Return the latest node with a key < key. // Return head_ if there is no such node. Node* FindLessThan(const Key& key) const; // Return the last node in the list. // Return head_ if list is empty. Node* FindLast() const; // No copying allowed SkipList(const SkipList&); void operator=(const SkipList&); };
节点定义
// Implementation details follow template<typename Key, class Comparator> struct SkipList<Key,Comparator>::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. 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()); } 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: // Array of length equal to the node height. next_[0] is lowest level link. //节点的层指针数组(弹性数组) port::AtomicPointer next_[1]; };
创建新节点
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)); //定位构造 //placement new就是在用户指定的内存位置上构建新的对象, //这个构建过程不需要额外分配内存,只需要调用对象的构造函数即可 return new (mem) Node(key); }
placement new(定位构造)的用法可参考手册,也可参考以下网址:
http://blog.csdn.net/songthin/article/details/1703966
节点插入
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; } template<typename Key, class Comparator> bool SkipList<Key,Comparator>::KeyIsAfterNode(const Key& key, Node* n) const { // NULL n is considered infinite 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)) { // Keep searching in this list x = next; } else { if (prev != NULL) prev[level] = x; if (level == 0) { return next; } else { // Switch to next list level--; } } } } 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. //TODO:因为插入需要外部同步,所以此处可以不用内存栅栏 Node* prev[kMaxHeight]; //找插入位置,prev记录每一层的前一指针 Node* x = FindGreaterOrEqual(key, prev); // Our data structure does not allow duplicate insertion assert(x == NULL || !Equal(key, x->key)); //随机插入节点层数,底层数为上层数的4倍(0层3/4 1层(1/4)*(3/4) ....) 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); // 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<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); } }
读redis的源码其random level函数写的也蛮漂亮
#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^32 elements */ #define ZSKIPLIST_P 0.25 /* Skiplist P = 1/4 */ /* Returns a random level for the new skiplist node we are going to create. * The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL * (both inclusive), with a powerlaw-alike distribution where higher * levels are less likely to be returned. */ int zslRandomLevel(void) { int level = 1; while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF)) level += 1; return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL; }
OK,插入有了(其中包含查询步骤),查询就不必写了。
卿若有意,可读redis的skiplist(zskiplist),其skiplist的定义
typedef struct zskiplistNode { robj *obj; double score; struct zskiplistNode *backward; struct zskiplistLevel { struct zskiplistNode *forward; unsigned int span; } level[]; } zskiplistNode; typedef struct zskiplist { struct zskiplistNode *header, *tail; unsigned long length; int level; } zskiplist;
redis的level不仅有前进指针,还有一个步长(unsigned int span),这使得根据rank(排名)做操作时更加简单效率。
参考
《redis设计与实现》