working skiplist

6 files changed, 1165 insertions, 0 deletions

diff --git a/weed/util/skiplist/Makefile b/weed/util/skiplist/Makefile
new file mode 100644
index 000000000..af4afe639
--- /dev/null
+++ b/weed/util/skiplist/Makefile
@@ -0,0 +1,6 @@
+all: gen
+
+.PHONY : gen
+
+gen:
+	protoc skiplist.proto --go_out=plugins=grpc:. --go_opt=paths=source_relative
diff --git a/weed/util/skiplist/serde.go b/weed/util/skiplist/serde.go
new file mode 100644
index 000000000..2337b4b19
--- /dev/null
+++ b/weed/util/skiplist/serde.go
@@ -0,0 +1,54 @@
+package skiplist
+
+import "bytes"
+
+func compareElement(a *SkipListElement, key []byte) int {
+	if len(a.Values) == 0 {
+		return -1
+	}
+	if bytes.Compare(a.Values[0], key) < 0 {
+		return -1
+	}
+	if bytes.Compare(a.Values[len(a.Values)-1], key) > 0 {
+		return 1
+	}
+	return 0
+}
+
+var (
+	memStore = make(map[int64]*SkipListElement)
+)
+
+func (node *SkipListElement) Reference() *SkipListElementReference {
+	if node == nil {
+		return nil
+	}
+	return &SkipListElementReference{
+		ElementPointer: node.Id,
+		Key:            node.Values[0],
+	}
+}
+func (node *SkipListElement) Save() {
+	if node == nil {
+		return
+	}
+	memStore[node.Id] = node
+	//println("++ node", node.Id, string(node.Values[0]))
+}
+
+func (node *SkipListElement) DeleteSelf() {
+	if node == nil {
+		return
+	}
+	delete(memStore, node.Id)
+	//println("++ node", node.Id, string(node.Values[0]))
+}
+
+func (ref *SkipListElementReference) Load() *SkipListElement {
+	if ref == nil {
+		return nil
+	}
+	//println("~ node", ref.ElementPointer, string(ref.Key))
+	return memStore[ref.ElementPointer]
+}
+
diff --git a/weed/util/skiplist/skiplist.go b/weed/util/skiplist/skiplist.go
new file mode 100644
index 000000000..a47cf4608
--- /dev/null
+++ b/weed/util/skiplist/skiplist.go
@@ -0,0 +1,480 @@
+package skiplist
+
+import (
+	"bytes"
+	"fmt"
+	"math/bits"
+	"math/rand"
+	"time"
+)
+
+const (
+	// maxLevel denotes the maximum height of the skiplist. This height will keep the skiplist
+	// efficient for up to 34m entries. If there is a need for much more, please adjust this constant accordingly.
+	maxLevel = 25
+)
+
+type SkipList struct {
+	startLevels  [maxLevel]*SkipListElementReference
+	endLevels    [maxLevel]*SkipListElementReference
+	maxNewLevel  int
+	maxLevel     int
+	elementCount int
+}
+
+// NewSeedEps returns a new empty, initialized Skiplist.
+// Given a seed, a deterministic height/list behaviour can be achieved.
+// Eps is used to compare keys given by the ExtractKey() function on equality.
+func NewSeed(seed int64) *SkipList {
+
+	// Initialize random number generator.
+	rand.Seed(seed)
+	//fmt.Printf("SkipList seed: %v\n", seed)
+
+	list := &SkipList{
+		maxNewLevel:  maxLevel,
+		maxLevel:     0,
+		elementCount: 0,
+	}
+
+	return list
+}
+
+// New returns a new empty, initialized Skiplist.
+func New() *SkipList {
+	return NewSeed(time.Now().UTC().UnixNano())
+}
+
+// IsEmpty checks, if the skiplist is empty.
+func (t *SkipList) IsEmpty() bool {
+	return t.startLevels[0] == nil
+}
+
+func (t *SkipList) generateLevel(maxLevel int) int {
+	level := maxLevel - 1
+	// First we apply some mask which makes sure that we don't get a level
+	// above our desired level. Then we find the first set bit.
+	var x = rand.Uint64() & ((1 << uint(maxLevel-1)) - 1)
+	zeroes := bits.TrailingZeros64(x)
+	if zeroes <= maxLevel {
+		level = zeroes
+	}
+
+	return level
+}
+
+func (t *SkipList) findEntryIndex(key []byte, level int) int {
+	// Find good entry point so we don't accidentally skip half the list.
+	for i := t.maxLevel; i >= 0; i-- {
+		if t.startLevels[i] != nil && bytes.Compare(t.startLevels[i].Key, key) < 0 || i <= level {
+			return i
+		}
+	}
+	return 0
+}
+
+func (t *SkipList) findExtended(key []byte, findGreaterOrEqual bool) (foundElem *SkipListElement, ok bool) {
+
+	foundElem = nil
+	ok = false
+
+	if t.IsEmpty() {
+		return
+	}
+
+	index := t.findEntryIndex(key, 0)
+	var currentNode *SkipListElement
+
+	currentNode = t.startLevels[index].Load()
+
+	// In case, that our first element is already greater-or-equal!
+	if findGreaterOrEqual && compareElement(currentNode, key) > 0 {
+		foundElem = currentNode
+		ok = true
+		return
+	}
+
+	for {
+		if compareElement(currentNode, key) == 0 {
+			foundElem = currentNode
+			ok = true
+			return
+		}
+
+		// Which direction are we continuing next time?
+		if currentNode.Next[index] != nil && bytes.Compare(currentNode.Next[index].Key, key) <= 0 {
+			// Go right
+			currentNode = currentNode.Next[index].Load()
+		} else {
+			if index > 0 {
+
+				// Early exit
+				if currentNode.Next[0] != nil && bytes.Compare(currentNode.Next[0].Key, key) == 0 {
+					currentNodeNext := currentNode.Next[0].Load()
+					foundElem = currentNodeNext
+					ok = true
+					return
+				}
+				// Go down
+				index--
+			} else {
+				// Element is not found and we reached the bottom.
+				if findGreaterOrEqual {
+					foundElem = currentNode.Next[index].Load()
+					ok = foundElem != nil
+				}
+
+				return
+			}
+		}
+	}
+}
+
+// Find tries to find an element in the skiplist based on the key from the given ListElement.
+// elem can be used, if ok is true.
+// Find runs in approx. O(log(n))
+func (t *SkipList) Find(key []byte) (elem *SkipListElement, ok bool) {
+
+	if t == nil || key == nil {
+		return
+	}
+
+	elem, ok = t.findExtended(key, false)
+	return
+}
+
+// FindGreaterOrEqual finds the first element, that is greater or equal to the given ListElement e.
+// The comparison is done on the keys (So on ExtractKey()).
+// FindGreaterOrEqual runs in approx. O(log(n))
+func (t *SkipList) FindGreaterOrEqual(key []byte) (elem *SkipListElement, ok bool) {
+
+	if t == nil || key == nil {
+		return
+	}
+
+	elem, ok = t.findExtended(key, true)
+	return
+}
+
+// Delete removes an element equal to e from the skiplist, if there is one.
+// If there are multiple entries with the same value, Delete will remove one of them
+// (Which one will change based on the actual skiplist layout)
+// Delete runs in approx. O(log(n))
+func (t *SkipList) Delete(key []byte) {
+
+	if t == nil || t.IsEmpty() || key == nil {
+		return
+	}
+
+	index := t.findEntryIndex(key, t.maxLevel)
+
+	var currentNode *SkipListElement
+	var nextNode *SkipListElement
+
+	for {
+
+		if currentNode == nil {
+			nextNode = t.startLevels[index].Load()
+		} else {
+			nextNode = currentNode.Next[index].Load()
+		}
+
+		// Found and remove!
+		if nextNode != nil && compareElement(nextNode, key) == 0 {
+
+			if currentNode != nil {
+				currentNode.Next[index] = nextNode.Next[index]
+				currentNode.Save()
+			}
+
+			if index == 0 {
+				if nextNode.Next[index] != nil {
+					nextNextNode := nextNode.Next[index].Load()
+					nextNextNode.Prev = currentNode.Reference()
+					nextNextNode.Save()
+				}
+				t.elementCount--
+				nextNode.DeleteSelf()
+			}
+
+			// Link from start needs readjustments.
+			startNextKey := t.startLevels[index].Key
+			if compareElement(nextNode, startNextKey) == 0 {
+				t.startLevels[index] = nextNode.Next[index]
+				// This was our currently highest node!
+				if t.startLevels[index] == nil {
+					t.maxLevel = index - 1
+				}
+			}
+
+			// Link from end needs readjustments.
+			if nextNode.Next[index] == nil {
+				t.endLevels[index] = currentNode.Reference()
+			}
+			nextNode.Next[index] = nil
+		}
+
+		if nextNode != nil && compareElement(nextNode, key) < 0 {
+			// Go right
+			currentNode = nextNode
+		} else {
+			// Go down
+			index--
+			if index < 0 {
+				break
+			}
+		}
+	}
+
+}
+
+// Insert inserts the given ListElement into the skiplist.
+// Insert runs in approx. O(log(n))
+func (t *SkipList) Insert(key []byte) {
+
+	if t == nil || key == nil {
+		return
+	}
+
+	level := t.generateLevel(t.maxNewLevel)
+
+	// Only grow the height of the skiplist by one at a time!
+	if level > t.maxLevel {
+		level = t.maxLevel + 1
+		t.maxLevel = level
+	}
+
+	elem := &SkipListElement{
+		Id:     rand.Int63(),
+		Next:   make([]*SkipListElementReference, t.maxNewLevel, t.maxNewLevel),
+		Level:  int32(level),
+		Values: [][]byte{key},
+	}
+
+	t.elementCount++
+
+	newFirst := true
+	newLast := true
+	if !t.IsEmpty() {
+		newFirst = compareElement(elem, t.startLevels[0].Key) < 0
+		newLast = compareElement(elem, t.endLevels[0].Key) > 0
+	}
+
+	normallyInserted := false
+	if !newFirst && !newLast {
+
+		normallyInserted = true
+
+		index := t.findEntryIndex(key, level)
+
+		var currentNode *SkipListElement
+		var nextNodeRef *SkipListElementReference
+
+		for {
+
+			if currentNode == nil {
+				nextNodeRef = t.startLevels[index]
+			} else {
+				nextNodeRef = currentNode.Next[index]
+			}
+
+			var nextNode *SkipListElement
+
+			// Connect node to next
+			if index <= level && (nextNodeRef == nil || bytes.Compare(nextNodeRef.Key, key) > 0) {
+				elem.Next[index] = nextNodeRef
+				if currentNode != nil {
+					currentNode.Next[index] = elem.Reference()
+					currentNode.Save()
+				}
+				if index == 0 {
+					elem.Prev = currentNode.Reference()
+					if nextNodeRef != nil {
+						nextNode = nextNodeRef.Load()
+						nextNode.Prev = elem.Reference()
+						nextNode.Save()
+					}
+				}
+			}
+
+			if nextNodeRef != nil && bytes.Compare(nextNodeRef.Key, key) <= 0 {
+				// Go right
+				if nextNode == nil {
+					// reuse nextNode when index == 0
+					nextNode = nextNodeRef.Load()
+				}
+				currentNode = nextNode
+			} else {
+				// Go down
+				index--
+				if index < 0 {
+					break
+				}
+			}
+		}
+	}
+
+	// Where we have a left-most position that needs to be referenced!
+	for i := level; i >= 0; i-- {
+
+		didSomething := false
+
+		if newFirst || normallyInserted {
+
+			if t.startLevels[i] == nil || bytes.Compare(t.startLevels[i].Key, key) > 0 {
+				if i == 0 && t.startLevels[i] != nil {
+					startLevelElement := t.startLevels[i].Load()
+					startLevelElement.Prev = elem.Reference()
+					startLevelElement.Save()
+				}
+				elem.Next[i] = t.startLevels[i]
+				t.startLevels[i] = elem.Reference()
+			}
+
+			// link the endLevels to this element!
+			if elem.Next[i] == nil {
+				t.endLevels[i] = elem.Reference()
+			}
+
+			didSomething = true
+		}
+
+		if newLast {
+			// Places the element after the very last element on this level!
+			// This is very important, so we are not linking the very first element (newFirst AND newLast) to itself!
+			if !newFirst {
+				if t.endLevels[i] != nil {
+					endLevelElement := t.endLevels[i].Load()
+					endLevelElement.Next[i] = elem.Reference()
+					endLevelElement.Save()
+				}
+				if i == 0 {
+					elem.Prev = t.endLevels[i]
+				}
+				t.endLevels[i] = elem.Reference()
+			}
+
+			// Link the startLevels to this element!
+			if t.startLevels[i] == nil || bytes.Compare(t.startLevels[i].Key, key) > 0 {
+				t.startLevels[i] = elem.Reference()
+			}
+
+			didSomething = true
+		}
+
+		if !didSomething {
+			break
+		}
+	}
+
+	elem.Save()
+
+}
+
+// GetValue extracts the ListElement value from a skiplist node.
+func (e *SkipListElement) GetValue() []byte {
+	return e.Values[0]
+}
+
+// GetSmallestNode returns the very first/smallest node in the skiplist.
+// GetSmallestNode runs in O(1)
+func (t *SkipList) GetSmallestNode() *SkipListElement {
+	return t.startLevels[0].Load()
+}
+
+// GetLargestNode returns the very last/largest node in the skiplist.
+// GetLargestNode runs in O(1)
+func (t *SkipList) GetLargestNode() *SkipListElement {
+	return t.endLevels[0].Load()
+}
+
+// Next returns the next element based on the given node.
+// Next will loop around to the first node, if you call it on the last!
+func (t *SkipList) Next(e *SkipListElement) *SkipListElement {
+	if e.Next[0] == nil {
+		return t.startLevels[0].Load()
+	}
+	return e.Next[0].Load()
+}
+
+// Prev returns the previous element based on the given node.
+// Prev will loop around to the last node, if you call it on the first!
+func (t *SkipList) Prev(e *SkipListElement) *SkipListElement {
+	if e.Prev == nil {
+		return t.endLevels[0].Load()
+	}
+	return e.Prev.Load()
+}
+
+// GetNodeCount returns the number of nodes currently in the skiplist.
+func (t *SkipList) GetNodeCount() int {
+	return t.elementCount
+}
+
+// String returns a string format of the skiplist. Useful to get a graphical overview and/or debugging.
+func (t *SkipList) println() {
+
+	print("start --> ")
+	for i, l := range t.startLevels {
+		if l == nil {
+			break
+		}
+		if i > 0 {
+			print(" -> ")
+		}
+		next := "---"
+		if l != nil {
+			next = string(l.Key)
+		}
+		print(fmt.Sprintf("[%v]", next))
+	}
+	println()
+
+	nodeRef := t.startLevels[0]
+	for nodeRef != nil {
+		print(fmt.Sprintf("%v: ", string(nodeRef.Key)))
+		node := nodeRef.Load()
+		for i := 0; i <= int(node.Level); i++ {
+
+			l := node.Next[i]
+
+			next := "---"
+			if l != nil {
+				next = string(l.Key)
+			}
+
+			if i == 0 {
+				prev := "---"
+
+				if node.Prev != nil {
+					prev = string(node.Prev.Key)
+				}
+				print(fmt.Sprintf("[%v|%v]", prev, next))
+			} else {
+				print(fmt.Sprintf("[%v]", next))
+			}
+			if i < int(node.Level) {
+				print(" -> ")
+			}
+
+		}
+		println()
+		nodeRef = node.Next[0]
+	}
+
+	print("end --> ")
+	for i, l := range t.endLevels {
+		if l == nil {
+			break
+		}
+		if i > 0 {
+			print(" -> ")
+		}
+		next := "---"
+		if l != nil {
+			next = string(l.Key)
+		}
+		print(fmt.Sprintf("[%v]", next))
+	}
+	println()
+}
diff --git a/weed/util/skiplist/skiplist.pb.go b/weed/util/skiplist/skiplist.pb.go
new file mode 100644
index 000000000..63b6c74a3
--- /dev/null
+++ b/weed/util/skiplist/skiplist.pb.go
@@ -0,0 +1,386 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.25.0
+// 	protoc        v3.12.3
+// source: skiplist.proto
+
+package skiplist
+
+import (
+	proto "github.com/golang/protobuf/proto"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// This is a compile-time assertion that a sufficiently up-to-date version
+// of the legacy proto package is being used.
+const _ = proto.ProtoPackageIsVersion4
+
+type SkipListProto struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	StartLevels  []*SkipListElementReference `protobuf:"bytes,1,rep,name=start_levels,json=startLevels,proto3" json:"start_levels,omitempty"`
+	EndLevels    []*SkipListElementReference `protobuf:"bytes,2,rep,name=end_levels,json=endLevels,proto3" json:"end_levels,omitempty"`
+	MaxNewLevel  int32                       `protobuf:"varint,3,opt,name=max_new_level,json=maxNewLevel,proto3" json:"max_new_level,omitempty"`
+	MaxLevel     int32                       `protobuf:"varint,4,opt,name=max_level,json=maxLevel,proto3" json:"max_level,omitempty"`
+	ElementCount int64                       `protobuf:"varint,5,opt,name=element_count,json=elementCount,proto3" json:"element_count,omitempty"`
+	Eps          float64                     `protobuf:"fixed64,7,opt,name=eps,proto3" json:"eps,omitempty"`
+}
+
+func (x *SkipListProto) Reset() {
+	*x = SkipListProto{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_skiplist_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *SkipListProto) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SkipListProto) ProtoMessage() {}
+
+func (x *SkipListProto) ProtoReflect() protoreflect.Message {
+	mi := &file_skiplist_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SkipListProto.ProtoReflect.Descriptor instead.
+func (*SkipListProto) Descriptor() ([]byte, []int) {
+	return file_skiplist_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *SkipListProto) GetStartLevels() []*SkipListElementReference {
+	if x != nil {
+		return x.StartLevels
+	}
+	return nil
+}
+
+func (x *SkipListProto) GetEndLevels() []*SkipListElementReference {
+	if x != nil {
+		return x.EndLevels
+	}
+	return nil
+}
+
+func (x *SkipListProto) GetMaxNewLevel() int32 {
+	if x != nil {
+		return x.MaxNewLevel
+	}
+	return 0
+}
+
+func (x *SkipListProto) GetMaxLevel() int32 {
+	if x != nil {
+		return x.MaxLevel
+	}
+	return 0
+}
+
+func (x *SkipListProto) GetElementCount() int64 {
+	if x != nil {
+		return x.ElementCount
+	}
+	return 0
+}
+
+func (x *SkipListProto) GetEps() float64 {
+	if x != nil {
+		return x.Eps
+	}
+	return 0
+}
+
+type SkipListElementReference struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	ElementPointer int64  `protobuf:"varint,1,opt,name=element_pointer,json=elementPointer,proto3" json:"element_pointer,omitempty"`
+	Key            []byte `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"`
+}
+
+func (x *SkipListElementReference) Reset() {
+	*x = SkipListElementReference{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_skiplist_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *SkipListElementReference) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SkipListElementReference) ProtoMessage() {}
+
+func (x *SkipListElementReference) ProtoReflect() protoreflect.Message {
+	mi := &file_skiplist_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SkipListElementReference.ProtoReflect.Descriptor instead.
+func (*SkipListElementReference) Descriptor() ([]byte, []int) {
+	return file_skiplist_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *SkipListElementReference) GetElementPointer() int64 {
+	if x != nil {
+		return x.ElementPointer
+	}
+	return 0
+}
+
+func (x *SkipListElementReference) GetKey() []byte {
+	if x != nil {
+		return x.Key
+	}
+	return nil
+}
+
+type SkipListElement struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Id     int64                       `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
+	Next   []*SkipListElementReference `protobuf:"bytes,2,rep,name=next,proto3" json:"next,omitempty"`
+	Level  int32                       `protobuf:"varint,3,opt,name=level,proto3" json:"level,omitempty"`
+	Values [][]byte                    `protobuf:"bytes,4,rep,name=values,proto3" json:"values,omitempty"`
+	Prev   *SkipListElementReference   `protobuf:"bytes,5,opt,name=prev,proto3" json:"prev,omitempty"`
+}
+
+func (x *SkipListElement) Reset() {
+	*x = SkipListElement{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_skiplist_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *SkipListElement) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SkipListElement) ProtoMessage() {}
+
+func (x *SkipListElement) ProtoReflect() protoreflect.Message {
+	mi := &file_skiplist_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SkipListElement.ProtoReflect.Descriptor instead.
+func (*SkipListElement) Descriptor() ([]byte, []int) {
+	return file_skiplist_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *SkipListElement) GetId() int64 {
+	if x != nil {
+		return x.Id
+	}
+	return 0
+}
+
+func (x *SkipListElement) GetNext() []*SkipListElementReference {
+	if x != nil {
+		return x.Next
+	}
+	return nil
+}
+
+func (x *SkipListElement) GetLevel() int32 {
+	if x != nil {
+		return x.Level
+	}
+	return 0
+}
+
+func (x *SkipListElement) GetValues() [][]byte {
+	if x != nil {
+		return x.Values
+	}
+	return nil
+}
+
+func (x *SkipListElement) GetPrev() *SkipListElementReference {
+	if x != nil {
+		return x.Prev
+	}
+	return nil
+}
+
+var File_skiplist_proto protoreflect.FileDescriptor
+
+var file_skiplist_proto_rawDesc = []byte{
+	0x0a, 0x0e, 0x73, 0x6b, 0x69, 0x70, 0x6c, 0x69, 0x73, 0x74, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x12, 0x08, 0x73, 0x6b, 0x69, 0x70, 0x6c, 0x69, 0x73, 0x74, 0x22, 0x91, 0x02, 0x0a, 0x0d, 0x53,
+	0x6b, 0x69, 0x70, 0x4c, 0x69, 0x73, 0x74, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x45, 0x0a, 0x0c,
+	0x73, 0x74, 0x61, 0x72, 0x74, 0x5f, 0x6c, 0x65, 0x76, 0x65, 0x6c, 0x73, 0x18, 0x01, 0x20, 0x03,
+	0x28, 0x0b, 0x32, 0x22, 0x2e, 0x73, 0x6b, 0x69, 0x70, 0x6c, 0x69, 0x73, 0x74, 0x2e, 0x53, 0x6b,
+	0x69, 0x70, 0x4c, 0x69, 0x73, 0x74, 0x45, 0x6c, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x66,
+	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x0b, 0x73, 0x74, 0x61, 0x72, 0x74, 0x4c, 0x65, 0x76,
+	0x65, 0x6c, 0x73, 0x12, 0x41, 0x0a, 0x0a, 0x65, 0x6e, 0x64, 0x5f, 0x6c, 0x65, 0x76, 0x65, 0x6c,
+	0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x22, 0x2e, 0x73, 0x6b, 0x69, 0x70, 0x6c, 0x69,
+	0x73, 0x74, 0x2e, 0x53, 0x6b, 0x69, 0x70, 0x4c, 0x69, 0x73, 0x74, 0x45, 0x6c, 0x65, 0x6d, 0x65,
+	0x6e, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x65, 0x6e, 0x64,
+	0x4c, 0x65, 0x76, 0x65, 0x6c, 0x73, 0x12, 0x22, 0x0a, 0x0d, 0x6d, 0x61, 0x78, 0x5f, 0x6e, 0x65,
+	0x77, 0x5f, 0x6c, 0x65, 0x76, 0x65, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, 0x6d,
+	0x61, 0x78, 0x4e, 0x65, 0x77, 0x4c, 0x65, 0x76, 0x65, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x6d, 0x61,
+	0x78, 0x5f, 0x6c, 0x65, 0x76, 0x65, 0x6c, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, 0x6d,
+	0x61, 0x78, 0x4c, 0x65, 0x76, 0x65, 0x6c, 0x12, 0x23, 0x0a, 0x0d, 0x65, 0x6c, 0x65, 0x6d, 0x65,
+	0x6e, 0x74, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x05, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0c,
+	0x65, 0x6c, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x10, 0x0a, 0x03,
+	0x65, 0x70, 0x73, 0x18, 0x07, 0x20, 0x01, 0x28, 0x01, 0x52, 0x03, 0x65, 0x70, 0x73, 0x22, 0x55,
+	0x0a, 0x18, 0x53, 0x6b, 0x69, 0x70, 0x4c, 0x69, 0x73, 0x74, 0x45, 0x6c, 0x65, 0x6d, 0x65, 0x6e,
+	0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x27, 0x0a, 0x0f, 0x65, 0x6c,
+	0x65, 0x6d, 0x65, 0x6e, 0x74, 0x5f, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x03, 0x52, 0x0e, 0x65, 0x6c, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x50, 0x6f, 0x69, 0x6e,
+	0x74, 0x65, 0x72, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c,
+	0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0xbf, 0x01, 0x0a, 0x0f, 0x53, 0x6b, 0x69, 0x70, 0x4c, 0x69,
+	0x73, 0x74, 0x45, 0x6c, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x02, 0x69, 0x64, 0x12, 0x36, 0x0a, 0x04, 0x6e, 0x65, 0x78,
+	0x74, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x22, 0x2e, 0x73, 0x6b, 0x69, 0x70, 0x6c, 0x69,
+	0x73, 0x74, 0x2e, 0x53, 0x6b, 0x69, 0x70, 0x4c, 0x69, 0x73, 0x74, 0x45, 0x6c, 0x65, 0x6d, 0x65,
+	0x6e, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x04, 0x6e, 0x65, 0x78,
+	0x74, 0x12, 0x14, 0x0a, 0x05, 0x6c, 0x65, 0x76, 0x65, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05,
+	0x52, 0x05, 0x6c, 0x65, 0x76, 0x65, 0x6c, 0x12, 0x16, 0x0a, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65,
+	0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0c, 0x52, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x12,
+	0x36, 0x0a, 0x04, 0x70, 0x72, 0x65, 0x76, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x22, 0x2e,
+	0x73, 0x6b, 0x69, 0x70, 0x6c, 0x69, 0x73, 0x74, 0x2e, 0x53, 0x6b, 0x69, 0x70, 0x4c, 0x69, 0x73,
+	0x74, 0x45, 0x6c, 0x65, 0x6d, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63,
+	0x65, 0x52, 0x04, 0x70, 0x72, 0x65, 0x76, 0x42, 0x33, 0x5a, 0x31, 0x67, 0x69, 0x74, 0x68, 0x75,
+	0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f,
+	0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x75,
+	0x74, 0x69, 0x6c, 0x2f, 0x73, 0x6b, 0x69, 0x70, 0x6c, 0x69, 0x73, 0x74, 0x62, 0x06, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_skiplist_proto_rawDescOnce sync.Once
+	file_skiplist_proto_rawDescData = file_skiplist_proto_rawDesc
+)
+
+func file_skiplist_proto_rawDescGZIP() []byte {
+	file_skiplist_proto_rawDescOnce.Do(func() {
+		file_skiplist_proto_rawDescData = protoimpl.X.CompressGZIP(file_skiplist_proto_rawDescData)
+	})
+	return file_skiplist_proto_rawDescData
+}
+
+var file_skiplist_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_skiplist_proto_goTypes = []interface{}{
+	(*SkipListProto)(nil),            // 0: skiplist.SkipListProto
+	(*SkipListElementReference)(nil), // 1: skiplist.SkipListElementReference
+	(*SkipListElement)(nil),          // 2: skiplist.SkipListElement
+}
+var file_skiplist_proto_depIdxs = []int32{
+	1, // 0: skiplist.SkipListProto.start_levels:type_name -> skiplist.SkipListElementReference
+	1, // 1: skiplist.SkipListProto.end_levels:type_name -> skiplist.SkipListElementReference
+	1, // 2: skiplist.SkipListElement.next:type_name -> skiplist.SkipListElementReference
+	1, // 3: skiplist.SkipListElement.prev:type_name -> skiplist.SkipListElementReference
+	4, // [4:4] is the sub-list for method output_type
+	4, // [4:4] is the sub-list for method input_type
+	4, // [4:4] is the sub-list for extension type_name
+	4, // [4:4] is the sub-list for extension extendee
+	0, // [0:4] is the sub-list for field type_name
+}
+
+func init() { file_skiplist_proto_init() }
+func file_skiplist_proto_init() {
+	if File_skiplist_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_skiplist_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*SkipListProto); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_skiplist_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*SkipListElementReference); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_skiplist_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*SkipListElement); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_skiplist_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   3,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_skiplist_proto_goTypes,
+		DependencyIndexes: file_skiplist_proto_depIdxs,
+		MessageInfos:      file_skiplist_proto_msgTypes,
+	}.Build()
+	File_skiplist_proto = out.File
+	file_skiplist_proto_rawDesc = nil
+	file_skiplist_proto_goTypes = nil
+	file_skiplist_proto_depIdxs = nil
+}
diff --git a/weed/util/skiplist/skiplist.proto b/weed/util/skiplist/skiplist.proto
new file mode 100644
index 000000000..ce84ed996
--- /dev/null
+++ b/weed/util/skiplist/skiplist.proto
@@ -0,0 +1,27 @@
+syntax = "proto3";
+
+package skiplist;
+
+option go_package = "github.com/chrislusf/seaweedfs/weed/util/skiplist";
+
+message SkipListProto {
+  repeated SkipListElementReference start_levels = 1;
+  repeated SkipListElementReference end_levels = 2;
+  int32 max_new_level = 3;
+  int32 max_level = 4;
+  int64 element_count = 5;
+  double eps = 7;
+}
+
+message SkipListElementReference {
+  int64 element_pointer = 1;
+  bytes key = 2;
+}
+
+message SkipListElement {
+  int64 id = 1;
+  repeated SkipListElementReference next = 2;
+  int32 level = 3;
+  repeated bytes values = 4;
+  SkipListElementReference prev = 5;
+}
diff --git a/weed/util/skiplist/skiplist_test.go b/weed/util/skiplist/skiplist_test.go
new file mode 100644
index 000000000..ca41e382a
--- /dev/null
+++ b/weed/util/skiplist/skiplist_test.go
@@ -0,0 +1,212 @@
+package skiplist
+
+import (
+	"bytes"
+	"math/rand"
+	"strconv"
+	"testing"
+)
+
+const (
+	maxN = 10000
+)
+
+func TestInsertAndFind(t *testing.T) {
+
+	k0 := []byte("0")
+	var list *SkipList
+
+	var listPointer *SkipList
+	listPointer.Insert(k0)
+	if _, ok := listPointer.Find(k0); ok {
+		t.Fail()
+	}
+
+	list = New()
+	if _, ok := list.Find(k0); ok {
+		t.Fail()
+	}
+	if !list.IsEmpty() {
+		t.Fail()
+	}
+
+	// Test at the beginning of the list.
+	for i := 0; i < maxN; i++ {
+		key := []byte(strconv.Itoa(maxN-i))
+		list.Insert(key)
+	}
+	for i := 0; i < maxN; i++ {
+		key := []byte(strconv.Itoa(maxN-i))
+		if _, ok := list.Find(key); !ok {
+			t.Fail()
+		}
+	}
+
+
+	list = New()
+	// Test at the end of the list.
+	for i := 0; i < maxN; i++ {
+		key := []byte(strconv.Itoa(i))
+		list.Insert(key)
+	}
+	for i := 0; i < maxN; i++ {
+		key := []byte(strconv.Itoa(i))
+		if _, ok := list.Find(key); !ok {
+			t.Fail()
+		}
+	}
+
+	list = New()
+	// Test at random positions in the list.
+	rList := rand.Perm(maxN)
+	for _, e := range rList {
+		key := []byte(strconv.Itoa(e))
+		println("insert", e)
+		list.Insert(key)
+	}
+	for _, e := range rList {
+		key := []byte(strconv.Itoa(e))
+		println("find", e)
+		if _, ok := list.Find(key); !ok {
+			t.Fail()
+		}
+	}
+	println("print list")
+	list.println()
+
+}
+
+func Element(x int) []byte {
+	return []byte(strconv.Itoa(x))
+}
+
+func TestDelete(t *testing.T) {
+
+	k0 := []byte("0")
+
+	var list *SkipList
+
+	// Delete on empty list
+	list.Delete(k0)
+
+	list = New()
+
+	list.Delete(k0)
+	if !list.IsEmpty() {
+		t.Fail()
+	}
+
+	list.Insert(k0)
+	list.Delete(k0)
+	if !list.IsEmpty() {
+		t.Fail()
+	}
+
+	// Delete elements at the beginning of the list.
+	for i := 0; i < maxN; i++ {
+		list.Insert(Element(i))
+	}
+	for i := 0; i < maxN; i++ {
+		list.Delete(Element(i))
+	}
+	if !list.IsEmpty() {
+		t.Fail()
+	}
+
+	list = New()
+	// Delete elements at the end of the list.
+	for i := 0; i < maxN; i++ {
+		list.Insert(Element(i))
+	}
+	for i := 0; i < maxN; i++ {
+		list.Delete(Element(maxN - i - 1))
+	}
+	if !list.IsEmpty() {
+		t.Fail()
+	}
+
+	list = New()
+	// Delete elements at random positions in the list.
+	rList := rand.Perm(maxN)
+	for _, e := range rList {
+		list.Insert(Element(e))
+	}
+	for _, e := range rList {
+		list.Delete(Element(e))
+	}
+	if !list.IsEmpty() {
+		t.Fail()
+	}
+}
+
+func TestNext(t *testing.T) {
+	list := New()
+
+	for i := 0; i < maxN; i++ {
+		list.Insert(Element(i))
+	}
+
+	smallest := list.GetSmallestNode()
+	largest := list.GetLargestNode()
+
+	lastNode := smallest
+	node := lastNode
+	for node != largest {
+		node = list.Next(node)
+		// Must always be incrementing here!
+		if bytes.Compare(node.Values[0], lastNode.Values[0]) <= 0 {
+			t.Fail()
+		}
+		// Next.Prev must always point to itself!
+		if list.Next(list.Prev(node)) != node {
+			t.Fail()
+		}
+		lastNode = node
+	}
+
+	if list.Next(largest) != smallest {
+		t.Fail()
+	}
+}
+
+func TestPrev(t *testing.T) {
+	list := New()
+
+	for i := 0; i < maxN; i++ {
+		list.Insert(Element(i))
+	}
+
+	smallest := list.GetSmallestNode()
+	largest := list.GetLargestNode()
+
+	lastNode := largest
+	node := lastNode
+	for node != smallest {
+		node = list.Prev(node)
+		// Must always be incrementing here!
+		if bytes.Compare(node.Values[0], lastNode.Values[0]) >= 0 {
+			t.Fail()
+		}
+		// Next.Prev must always point to itself!
+		if list.Prev(list.Next(node)) != node {
+			t.Fail()
+		}
+		lastNode = node
+	}
+
+	if list.Prev(smallest) != largest {
+		t.Fail()
+	}
+}
+
+func TestGetNodeCount(t *testing.T) {
+	list := New()
+
+	for i := 0; i < maxN; i++ {
+		list.Insert(Element(i))
+	}
+
+	if list.GetNodeCount() != maxN {
+		t.Fail()
+	}
+}