redis3 using redis native sorted set

10 files changed, 776 insertions, 164 deletions

diff --git a/weed/filer/redis3/ItemList.go b/weed/filer/redis3/ItemList.go
new file mode 100644
index 000000000..ae4e61cfb
--- /dev/null
+++ b/weed/filer/redis3/ItemList.go
@@ -0,0 +1,483 @@
+package redis3
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/util/skiplist"
+	"github.com/go-redis/redis/v8"
+)
+
+type ItemList struct {
+	skipList  *skiplist.SkipList
+	batchSize int
+	client    redis.UniversalClient
+	prefix    string
+}
+
+func newItemList(client redis.UniversalClient, prefix string, store skiplist.ListStore, batchSize int) *ItemList {
+	return &ItemList{
+		skipList:  skiplist.New(store),
+		batchSize: batchSize,
+		client:    client,
+		prefix:    prefix,
+	}
+}
+
+/*
+Be reluctant to create new nodes. Try to fit into either previous node or next node.
+Prefer to add to previous node.
+
+There are multiple cases after finding the name for greater or equal node
+ 1. found and node.Key == name
+    The node contains a batch with leading key the same as the name
+    nothing to do
+ 2. no such node found or node.Key > name
+
+    if no such node found
+      prevNode = list.LargestNode
+
+	// case 2.1
+    if previousNode contains name
+      nothing to do
+
+    // prefer to add to previous node
+    if prevNode != nil {
+		// case 2.2
+		if prevNode has capacity
+			prevNode.add name, and save
+			return
+		// case 2.3
+		split prevNode by name
+    }
+
+	// case 2.4
+	// merge into next node. Avoid too many nodes if adding data in reverse order.
+	if nextNode is not nil and nextNode has capacity
+	  delete nextNode.Key
+      nextNode.Key = name
+      nextNode.batch.add name
+      insert nodeNode.Key
+	  return
+
+	// case 2.5
+    if prevNode is nil
+      insert new node with key = name, value = batch{name}
+      return
+
+*/
+func (nl *ItemList) WriteName(name string) error {
+
+	lookupKey := []byte(name)
+	prevNode, nextNode, found, err := nl.skipList.FindGreaterOrEqual(lookupKey)
+	if err != nil {
+		return err
+	}
+	// case 1: the name already exists as one leading key in the batch
+	if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
+		return nil
+	}
+
+	if !found {
+		prevNode, err = nl.skipList.GetLargestNode()
+		if err != nil {
+			return err
+		}
+	}
+
+	if nextNode != nil && prevNode == nil {
+		prevNode, err = nl.skipList.LoadElement(nextNode.Prev)
+		if err != nil {
+			return err
+		}
+	}
+
+	if prevNode != nil {
+		// case 2.1
+		if nl.NodeContainsItem(prevNode.Reference(), name) {
+			return nil
+		}
+
+		// case 2.2
+		nodeSize := nl.NodeSize(prevNode.Reference())
+		if nodeSize < nl.batchSize {
+			return nl.NodeAddMember(prevNode.Reference(), name)
+		}
+
+		// case 2.3
+		x := nl.NodeInnerPosition(prevNode.Reference(), name)
+		y := nodeSize - x
+		addToX := x <= y
+		// add to a new node
+		if x == 0 || y == 0 {
+			if err := nl.ItemAdd(lookupKey, 0, name); err != nil {
+				return err
+			}
+			return nil
+		}
+		if addToX {
+			// collect names before name, add them to X
+			namesToX, err := nl.NodeRangeBeforeExclusive(prevNode.Reference(), name)
+			if err != nil {
+				return nil
+			}
+			// delete skiplist reference to old node
+			if _, err := nl.skipList.DeleteByKey(prevNode.Key); err != nil {
+				return err
+			}
+			// add namesToY and name to a new X
+			namesToX = append(namesToX, name)
+			if err := nl.ItemAdd([]byte(namesToX[0]), 0, namesToX...); err != nil {
+				return nil
+			}
+			// remove names less than name from current Y
+			if err := nl.NodeDeleteBeforeExclusive(prevNode.Reference(), name); err != nil {
+				return nil
+			}
+
+			// point skip list to current Y
+			if err := nl.ItemAdd(lookupKey, prevNode.Id); err != nil {
+				return nil
+			}
+			return nil
+		} else {
+			// collect names after name, add them to Y
+			namesToY, err := nl.NodeRangeAfterExclusive(prevNode.Reference(), name)
+			if err != nil {
+				return nil
+			}
+			// add namesToY and name to a new Y
+			namesToY = append(namesToY, name)
+			if err := nl.ItemAdd(lookupKey, 0, namesToY...); err != nil {
+				return nil
+			}
+			// remove names after name from current X
+			if err := nl.NodeDeleteAfterExclusive(prevNode.Reference(), name); err != nil {
+				return nil
+			}
+			return nil
+		}
+
+	}
+
+	// case 2.4
+	if nextNode != nil {
+		nodeSize := nl.NodeSize(nextNode.Reference())
+		if nodeSize < nl.batchSize {
+			if id, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
+				return err
+			} else {
+				if err := nl.ItemAdd(lookupKey, id, name); err != nil {
+					return err
+				}
+			}
+			return nil
+		}
+	}
+
+	// case 2.5
+	// now prevNode is nil
+	return nl.ItemAdd(lookupKey, 0, name)
+}
+
+/*
+// case 1: exists in nextNode
+if nextNode != nil && nextNode.Key == name {
+	remove from nextNode, update nextNode
+	// TODO: merge with prevNode if possible?
+	return
+}
+if nextNode is nil
+	prevNode = list.Largestnode
+if prevNode == nil and nextNode.Prev != nil
+	prevNode = load(nextNode.Prev)
+
+// case 2: does not exist
+// case 2.1
+if prevNode == nil {
+	return
+}
+// case 2.2
+if prevNameBatch does not contain name {
+	return
+}
+
+// case 3
+delete from prevNameBatch
+if prevNameBatch + nextNode < capacityList
+	// case 3.1
+	merge
+else
+	// case 3.2
+	update prevNode
+
+
+*/
+func (nl *ItemList) DeleteName(name string) error {
+	lookupKey := []byte(name)
+	prevNode, nextNode, found, err := nl.skipList.FindGreaterOrEqual(lookupKey)
+	if err != nil {
+		return err
+	}
+
+	// case 1
+	if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
+		if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
+			return err
+		}
+		if err := nl.NodeDeleteMember(nextNode.Reference(), name); err != nil {
+			return err
+		}
+		minName := nl.NodeMin(nextNode.Reference())
+		if minName == "" {
+			return nl.NodeDelete(nextNode.Reference())
+		}
+		return nl.ItemAdd([]byte(minName), nextNode.Id)
+	}
+
+	if !found {
+		prevNode, err = nl.skipList.GetLargestNode()
+		if err != nil {
+			return err
+		}
+	}
+
+	if nextNode != nil && prevNode == nil {
+		prevNode, err = nl.skipList.LoadElement(nextNode.Prev)
+		if err != nil {
+			return err
+		}
+	}
+
+	// case 2
+	if prevNode == nil {
+		// case 2.1
+		return nil
+	}
+	if !nl.NodeContainsItem(prevNode.Reference(), name) {
+		return nil
+	}
+
+	// case 3
+	if err := nl.NodeDeleteMember(prevNode.Reference(), name); err != nil {
+		return err
+	}
+	prevSize := nl.NodeSize(prevNode.Reference())
+	if prevSize == 0 {
+		if _, err := nl.skipList.DeleteByKey(prevNode.Key); err != nil {
+			return err
+		}
+		return nil
+	}
+	nextSize := nl.NodeSize(nextNode.Reference())
+	if nextSize > 0 && prevSize + nextSize < nl.batchSize {
+		// case 3.1 merge nextNode and prevNode
+		if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
+			return err
+		}
+		nextNames, err := nl.NodeRangeBeforeExclusive(nextNode.Reference(), "")
+		if err != nil {
+			return err
+		}
+		if err := nl.NodeAddMember(prevNode.Reference(), nextNames...); err != nil {
+			return err
+		}
+		return nl.NodeDelete(nextNode.Reference())
+	} else {
+		// case 3.2 update prevNode
+		// no action to take
+		return nil
+	}
+
+	return nil
+}
+
+func (nl *ItemList) ListNames(startFrom string, visitNamesFn func(name string) bool) error {
+	lookupKey := []byte(startFrom)
+	prevNode, nextNode, found, err := nl.skipList.FindGreaterOrEqual(lookupKey)
+	if err != nil {
+		return err
+	}
+	if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
+		prevNode = nil
+	}
+	if !found {
+		prevNode, err = nl.skipList.GetLargestNode()
+		if err != nil {
+			return err
+		}
+	}
+
+	if prevNode != nil {
+		if !nl.NodeScanIncluseiveAfter(prevNode.Reference(), startFrom, visitNamesFn) {
+			return nil
+		}
+	}
+
+	for nextNode != nil {
+		if !nl.NodeScanIncluseiveAfter(nextNode.Reference(), startFrom, visitNamesFn) {
+			return nil
+		}
+		nextNode, err = nl.skipList.LoadElement(nextNode.Next[0])
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (nl *ItemList) RemoteAllListElement() error {
+
+	t := nl.skipList
+
+	nodeRef := t.StartLevels[0]
+	for nodeRef != nil {
+		node, err := t.LoadElement(nodeRef)
+		if err != nil {
+			return err
+		}
+		if node == nil {
+			return nil
+		}
+		if err := t.DeleteElement(node); err != nil {
+			return err
+		}
+		if err := nl.NodeDelete(node.Reference()); err != nil {
+			return err
+		}
+		nodeRef = node.Next[0]
+	}
+	return nil
+
+}
+
+func (nl *ItemList) NodeContainsItem(node *skiplist.SkipListElementReference, item string) bool {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	_, err := nl.client.ZScore(context.Background(), key, item).Result()
+	if err == redis.Nil {
+		return false
+	}
+	if err == nil {
+		return true
+	}
+	return false
+}
+
+func (nl *ItemList) NodeSize(node *skiplist.SkipListElementReference) int {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	return int(nl.client.ZLexCount(context.Background(), key, "-", "+").Val())
+}
+
+func (nl *ItemList) NodeAddMember(node *skiplist.SkipListElementReference, names ...string) error {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	var members []*redis.Z
+	for _, name := range names {
+		members = append(members, &redis.Z{
+			Score:  0,
+			Member: name,
+		})
+	}
+	return nl.client.ZAddNX(context.Background(), key, members...).Err()
+}
+func (nl *ItemList) NodeDeleteMember(node *skiplist.SkipListElementReference, name string) error {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	return nl.client.ZRem(context.Background(), key, name).Err()
+}
+
+func (nl *ItemList) NodeDelete(node *skiplist.SkipListElementReference) error {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	return nl.client.Del(context.Background(), key).Err()
+}
+
+func (nl *ItemList) NodeInnerPosition(node *skiplist.SkipListElementReference, name string) int {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	return int(nl.client.ZLexCount(context.Background(), key, "-", "("+name).Val())
+}
+
+func (nl *ItemList) NodeMin(node *skiplist.SkipListElementReference) string {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	slice := nl.client.ZPopMin(context.Background(), key).Val()
+	if len(slice)>0{
+		s := slice[0].Member.(string)
+		return s
+	}
+	return ""
+}
+
+func (nl *ItemList) NodeScanIncluseiveAfter(node *skiplist.SkipListElementReference, startFrom string, visitNamesFn func(name string) bool) bool {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	if startFrom == "" {
+		startFrom = "-"
+	} else {
+		startFrom = "[" + startFrom
+	}
+	names := nl.client.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
+		Min: startFrom,
+		Max: "+",
+	}).Val()
+	for _, n := range names {
+		if !visitNamesFn(n) {
+			return false
+		}
+	}
+	return true
+}
+
+func (nl *ItemList) NodeRangeBeforeExclusive(node *skiplist.SkipListElementReference, stopAt string) ([]string, error) {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	if stopAt == "" {
+		stopAt = "+"
+	} else {
+		stopAt = "(" + stopAt
+	}
+	return nl.client.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
+		Min: "-",
+		Max: stopAt,
+	}).Result()
+}
+func (nl *ItemList) NodeRangeAfterExclusive(node *skiplist.SkipListElementReference, startFrom string) ([]string, error) {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	if startFrom == "" {
+		startFrom = "-"
+	} else {
+		startFrom = "(" + startFrom
+	}
+	return nl.client.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
+		Min: startFrom,
+		Max: "+",
+	}).Result()
+}
+
+func (nl *ItemList) NodeDeleteBeforeExclusive(node *skiplist.SkipListElementReference, stopAt string) error {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	if stopAt == "" {
+		stopAt = "+"
+	} else {
+		stopAt = "(" + stopAt
+	}
+	return nl.client.ZRemRangeByLex(context.Background(), key, "-", stopAt).Err()
+}
+func (nl *ItemList) NodeDeleteAfterExclusive(node *skiplist.SkipListElementReference, startFrom string) error {
+	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
+	if startFrom == "" {
+		startFrom = "-"
+	} else {
+		startFrom = "(" + startFrom
+	}
+	return nl.client.ZRemRangeByLex(context.Background(), key, startFrom, "+").Err()
+}
+
+func (nl *ItemList) ItemAdd(lookupKey []byte, idIfKnown int64, names ...string) error {
+	if id, err := nl.skipList.InsertByKey(lookupKey, idIfKnown, nil); err != nil {
+		return err
+	} else {
+		if len(names) > 0 {
+			return nl.NodeAddMember(&skiplist.SkipListElementReference{
+				ElementPointer: id,
+				Key:            lookupKey,
+			}, names...)
+		}
+	}
+	return nil
+}
diff --git a/weed/filer/redis3/item_list_serde.go b/weed/filer/redis3/item_list_serde.go
new file mode 100644
index 000000000..d0310ce40
--- /dev/null
+++ b/weed/filer/redis3/item_list_serde.go
@@ -0,0 +1,75 @@
+package redis3
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/util/skiplist"
+	"github.com/go-redis/redis/v8"
+	"github.com/golang/protobuf/proto"
+)
+
+func LoadItemList(data []byte, prefix string, client redis.UniversalClient, store skiplist.ListStore, batchSize int) *ItemList {
+
+	nl := &ItemList{
+		skipList:  skiplist.New(store),
+		batchSize: batchSize,
+		client:    client,
+		prefix:    prefix,
+	}
+
+	if len(data) == 0 {
+		return nl
+	}
+
+	message := &skiplist.SkipListProto{}
+	if err := proto.Unmarshal(data, message); err != nil {
+		glog.Errorf("loading skiplist: %v", err)
+	}
+	nl.skipList.MaxNewLevel = int(message.MaxNewLevel)
+	nl.skipList.MaxLevel = int(message.MaxLevel)
+	for i, ref := range message.StartLevels {
+		nl.skipList.StartLevels[i] = &skiplist.SkipListElementReference{
+			ElementPointer: ref.ElementPointer,
+			Key:            ref.Key,
+		}
+	}
+	for i, ref := range message.EndLevels {
+		nl.skipList.EndLevels[i] = &skiplist.SkipListElementReference{
+			ElementPointer: ref.ElementPointer,
+			Key:            ref.Key,
+		}
+	}
+	return nl
+}
+
+func (nl *ItemList) HasChanges() bool {
+	return nl.skipList.HasChanges
+}
+
+func (nl *ItemList) ToBytes() []byte {
+	message := &skiplist.SkipListProto{}
+	message.MaxNewLevel = int32(nl.skipList.MaxNewLevel)
+	message.MaxLevel = int32(nl.skipList.MaxLevel)
+	for _, ref := range nl.skipList.StartLevels {
+		if ref == nil {
+			break
+		}
+		message.StartLevels = append(message.StartLevels, &skiplist.SkipListElementReference{
+			ElementPointer: ref.ElementPointer,
+			Key:            ref.Key,
+		})
+	}
+	for _, ref := range nl.skipList.EndLevels {
+		if ref == nil {
+			break
+		}
+		message.EndLevels = append(message.EndLevels, &skiplist.SkipListElementReference{
+			ElementPointer: ref.ElementPointer,
+			Key:            ref.Key,
+		})
+	}
+	data, err := proto.Marshal(message)
+	if err != nil {
+		glog.Errorf("marshal skiplist: %v", err)
+	}
+	return data
+}
diff --git a/weed/filer/redis3/kv_directory_children.go b/weed/filer/redis3/kv_directory_children.go
index 797e7797c..624d17374 100644
--- a/weed/filer/redis3/kv_directory_children.go
+++ b/weed/filer/redis3/kv_directory_children.go
@@ -4,11 +4,10 @@ import (
 	"context"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/util/skiplist"
 	"github.com/go-redis/redis/v8"
 )
 
-const maxNameBatchSizeLimit = 1000
+const maxNameBatchSizeLimit = 1000000
 
 func insertChild(ctx context.Context, redisStore *UniversalRedis3Store, key string, name string) error {
 
@@ -29,7 +28,7 @@ func insertChild(ctx context.Context, redisStore *UniversalRedis3Store, key stri
 		}
 	}
 	store := newSkipListElementStore(key, client)
-	nameList := skiplist.LoadNameList([]byte(data), store, maxNameBatchSizeLimit)
+	nameList := LoadItemList([]byte(data), key, client, store, maxNameBatchSizeLimit)
 
 	if err := nameList.WriteName(name); err != nil {
 		glog.Errorf("add %s %s: %v", key, name, err)
@@ -64,7 +63,7 @@ func removeChild(ctx context.Context, redisStore *UniversalRedis3Store, key stri
 		}
 	}
 	store := newSkipListElementStore(key, client)
-	nameList := skiplist.LoadNameList([]byte(data), store, maxNameBatchSizeLimit)
+	nameList := LoadItemList([]byte(data), key, client, store, maxNameBatchSizeLimit)
 
 	if err := nameList.DeleteName(name); err != nil {
 		return err
@@ -97,7 +96,7 @@ func removeChildren(ctx context.Context, redisStore *UniversalRedis3Store, key s
 		}
 	}
 	store := newSkipListElementStore(key, client)
-	nameList := skiplist.LoadNameList([]byte(data), store, maxNameBatchSizeLimit)
+	nameList := LoadItemList([]byte(data), key, client, store, maxNameBatchSizeLimit)
 
 	if err = nameList.ListNames("", func(name string) bool {
 		if err := onDeleteFn(name); err != nil {
@@ -126,7 +125,7 @@ func listChildren(ctx context.Context, redisStore *UniversalRedis3Store, key str
 		}
 	}
 	store := newSkipListElementStore(key, client)
-	nameList := skiplist.LoadNameList([]byte(data), store, maxNameBatchSizeLimit)
+	nameList := LoadItemList([]byte(data), key, client, store, maxNameBatchSizeLimit)
 
 	if err = nameList.ListNames(startFileName, func(name string) bool {
 		return eachFn(name)
diff --git a/weed/filer/redis3/kv_directory_children_test.go b/weed/filer/redis3/kv_directory_children_test.go
index 5c1cff2bb..77988e1a3 100644
--- a/weed/filer/redis3/kv_directory_children_test.go
+++ b/weed/filer/redis3/kv_directory_children_test.go
@@ -2,11 +2,12 @@ package redis3
 
 import (
 	"context"
-	"github.com/chrislusf/seaweedfs/weed/util/skiplist"
+	"fmt"
 	"github.com/go-redis/redis/v8"
 	"github.com/stvp/tempredis"
 	"strconv"
 	"testing"
+	"time"
 )
 
 var names = []string{
@@ -53,20 +54,21 @@ func TestNameList(t *testing.T) {
 	store := newSkipListElementStore("/yyy/bin", client)
 	var data []byte
 	for _, name := range names {
-		nameList := skiplist.LoadNameList(data, store, maxNameBatchSizeLimit)
+		nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
 		nameList.WriteName(name)
 
 		nameList.ListNames("", func(name string) bool {
+			// println(name)
 			return true
 		})
 
 		if nameList.HasChanges() {
 			data = nameList.ToBytes()
 		}
-		println()
+		// println()
 	}
 
-	nameList := skiplist.LoadNameList(data, store, maxNameBatchSizeLimit)
+	nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
 	nameList.ListNames("", func(name string) bool {
 		println(name)
 		return true
@@ -74,7 +76,7 @@ func TestNameList(t *testing.T) {
 
 }
 
-func BenchmarkNameList(b *testing.B) {
+func xBenchmarkNameList(b *testing.B) {
 
 	server, err := tempredis.Start(tempredis.Config{})
 	if err != nil {
@@ -90,9 +92,9 @@ func BenchmarkNameList(b *testing.B) {
 	store := newSkipListElementStore("/yyy/bin", client)
 	var data []byte
 	for i := 0; i < b.N; i++ {
-		nameList := skiplist.LoadNameList(data, store, maxNameBatchSizeLimit)
+		nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
 
-		nameList.WriteName("name"+strconv.Itoa(i))
+		nameList.WriteName(strconv.Itoa(i)+"namexxxxxxxxxxxxxxxxxxx")
 
 		if nameList.HasChanges() {
 			data = nameList.ToBytes()
@@ -100,7 +102,7 @@ func BenchmarkNameList(b *testing.B) {
 	}
 }
 
-func BenchmarkRedis(b *testing.B) {
+func xBenchmarkRedis(b *testing.B) {
 
 	server, err := tempredis.Start(tempredis.Config{})
 	if err != nil {
@@ -114,12 +116,60 @@ func BenchmarkRedis(b *testing.B) {
 	})
 
 	for i := 0; i < b.N; i++ {
-		client.ZAddNX(context.Background(),"/yyy/bin", &redis.Z{Score: 0, Member: "name"+strconv.Itoa(i)})
+		client.ZAddNX(context.Background(),"/yyy/bin", &redis.Z{Score: 0, Member: strconv.Itoa(i)+"namexxxxxxxxxxxxxxxxxxx"})
 	}
 }
 
+func TestNameListAdd(t *testing.T) {
 
-func xBenchmarkNameList(b *testing.B) {
+	server, err := tempredis.Start(tempredis.Config{})
+	if err != nil {
+		panic(err)
+	}
+	defer server.Term()
+
+	client := redis.NewClient(&redis.Options{
+		Addr:     "localhost:6379",
+		Password: "",
+		DB:       0,
+	})
+
+	client.FlushAll(context.Background())
+
+	N := 364800
+
+	ts0 := time.Now()
+	store := newSkipListElementStore("/y", client)
+	var data []byte
+	nameList := LoadItemList(data, "/y", client, store, 100000)
+	for i := 0; i < N; i++ {
+		nameList.WriteName(fmt.Sprintf("%8d", i))
+	}
+
+	ts1 := time.Now()
+
+	for i := 0; i < N; i++ {
+		client.ZAddNX(context.Background(),"/x", &redis.Z{Score: 0, Member: fmt.Sprintf("name %8d", i)})
+	}
+	ts2 := time.Now()
+
+	fmt.Printf("%v %v", ts1.Sub(ts0), ts2.Sub(ts1))
+
+	/*
+	keys := client.Keys(context.Background(), "/*m").Val()
+	for _, k := range keys {
+		println("key", k)
+		for i, v := range client.ZRangeByLex(context.Background(), k, &redis.ZRangeBy{
+			Min:    "-",
+			Max:    "+",
+		}).Val() {
+			println(" ", i, v)
+		}
+	}
+	 */
+}
+
+func BenchmarkNameList(b *testing.B) {
 
 	server, err := tempredis.Start(tempredis.Config{})
 	if err != nil {
@@ -136,9 +186,9 @@ func xBenchmarkNameList(b *testing.B) {
 	store := newSkipListElementStore("/yyy/bin", client)
 	var data []byte
 	for i := 0; i < b.N; i++ {
-		nameList := skiplist.LoadNameList(data, store, maxNameBatchSizeLimit)
+		nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit)
 
-		nameList.WriteName("name"+strconv.Itoa(i))
+		nameList.WriteName(fmt.Sprintf("name %8d", i))
 
 		if nameList.HasChanges() {
 			data = nameList.ToBytes()
@@ -146,7 +196,7 @@ func xBenchmarkNameList(b *testing.B) {
 	}
 }
 
-func xBenchmarkRedis(b *testing.B) {
+func BenchmarkRedis(b *testing.B) {
 
 	client := redis.NewClient(&redis.Options{
 		Addr:     "localhost:6379",
@@ -155,6 +205,6 @@ func xBenchmarkRedis(b *testing.B) {
 	})
 
 	for i := 0; i < b.N; i++ {
-		client.ZAddNX(context.Background(),"/xxx/bin", &redis.Z{Score: 0, Member: "name"+strconv.Itoa(i)})
+		client.ZAddNX(context.Background(),"/xxx/bin", &redis.Z{Score: 0, Member: fmt.Sprintf("name %8d", i)})
 	}
 }
diff --git a/weed/filer/redis3/skiplist_element_store.go b/weed/filer/redis3/skiplist_element_store.go
index 66a5408d6..bcad356dd 100644
--- a/weed/filer/redis3/skiplist_element_store.go
+++ b/weed/filer/redis3/skiplist_element_store.go
@@ -10,7 +10,7 @@ import (
 )
 
 type SkipListElementStore struct {
-	prefix string
+	Prefix string
 	client redis.UniversalClient
 }
 
@@ -18,13 +18,13 @@ var _ = skiplist.ListStore(&SkipListElementStore{})
 
 func newSkipListElementStore(prefix string, client redis.UniversalClient) *SkipListElementStore {
 	return &SkipListElementStore{
-		prefix: prefix,
+		Prefix: prefix,
 		client: client,
 	}
 }
 
 func (m *SkipListElementStore) SaveElement(id int64, element *skiplist.SkipListElement) error {
-	key := fmt.Sprintf("%s%d", m.prefix, id)
+	key := fmt.Sprintf("%s%d", m.Prefix, id)
 	data, err := proto.Marshal(element)
 	if err != nil {
 		glog.Errorf("marshal %s: %v", key, err)
@@ -33,12 +33,12 @@ func (m *SkipListElementStore) SaveElement(id int64, element *skiplist.SkipListE
 }
 
 func (m *SkipListElementStore) DeleteElement(id int64) error {
-	key := fmt.Sprintf("%s%d", m.prefix, id)
+	key := fmt.Sprintf("%s%d", m.Prefix, id)
 	return m.client.Del(context.Background(), key).Err()
 }
 
 func (m *SkipListElementStore) LoadElement(id int64) (*skiplist.SkipListElement, error) {
-	key := fmt.Sprintf("%s%d", m.prefix, id)
+	key := fmt.Sprintf("%s%d", m.Prefix, id)
 	data, err := m.client.Get(context.Background(), key).Result()
 	if err != nil {
 		if err == redis.Nil {
diff --git a/weed/util/skiplist/name_list.go b/weed/util/skiplist/name_list.go
index c5cbf3f87..19e8e6b49 100644
--- a/weed/util/skiplist/name_list.go
+++ b/weed/util/skiplist/name_list.go
@@ -78,7 +78,7 @@ func (nl *NameList) WriteName(name string) error {
 	}
 
 	if nextNode != nil && prevNode == nil {
-		prevNode, err = nl.skipList.loadElement(nextNode.Prev)
+		prevNode, err = nl.skipList.LoadElement(nextNode.Prev)
 		if err != nil {
 			return err
 		}
@@ -109,7 +109,7 @@ func (nl *NameList) WriteName(name string) error {
 					return err
 				}
 			} else {
-				if err := nl.skipList.Insert([]byte(x.key), x.ToBytes()); err != nil {
+				if _, err := nl.skipList.InsertByKey([]byte(x.key), 0, x.ToBytes()); err != nil {
 					return err
 				}
 			}
@@ -123,7 +123,7 @@ func (nl *NameList) WriteName(name string) error {
 					return err
 				}
 			} else {
-				if err := nl.skipList.Insert([]byte(y.key), y.ToBytes()); err != nil {
+				if _, err := nl.skipList.InsertByKey([]byte(y.key), 0, y.ToBytes()); err != nil {
 					return err
 				}
 			}
@@ -136,11 +136,11 @@ func (nl *NameList) WriteName(name string) error {
 	if nextNode != nil {
 		nextNameBatch := LoadNameBatch(nextNode.Value)
 		if len(nextNameBatch.names) < nl.batchSize {
-			if err := nl.skipList.Delete(nextNode.Key); err != nil {
+			if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
 				return err
 			}
 			nextNameBatch.WriteName(name)
-			if err := nl.skipList.Insert([]byte(nextNameBatch.key), nextNameBatch.ToBytes()); err != nil {
+			if _, err := nl.skipList.InsertByKey([]byte(nextNameBatch.key), 0, nextNameBatch.ToBytes()); err != nil {
 				return err
 			}
 			return nil
@@ -151,7 +151,7 @@ func (nl *NameList) WriteName(name string) error {
 	// now prevNode is nil
 	newNameBatch := NewNameBatch()
 	newNameBatch.WriteName(name)
-	if err := nl.skipList.Insert([]byte(newNameBatch.key), newNameBatch.ToBytes()); err != nil {
+	if _, err := nl.skipList.InsertByKey([]byte(newNameBatch.key), 0, newNameBatch.ToBytes()); err != nil {
 		return err
 	}
 
@@ -204,12 +204,12 @@ func (nl *NameList) DeleteName(name string) error {
 		nextNameBatch = LoadNameBatch(nextNode.Value)
 	}
 	if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
-		if err := nl.skipList.Delete(nextNode.Key); err != nil {
+		if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
 			return err
 		}
 		nextNameBatch.DeleteName(name)
 		if len(nextNameBatch.names) > 0 {
-			if err := nl.skipList.Insert([]byte(nextNameBatch.key), nextNameBatch.ToBytes()); err != nil {
+			if _, err := nl.skipList.InsertByKey([]byte(nextNameBatch.key), 0, nextNameBatch.ToBytes()); err != nil {
 				return err
 			}
 		}
@@ -224,7 +224,7 @@ func (nl *NameList) DeleteName(name string) error {
 	}
 
 	if nextNode != nil && prevNode == nil {
-		prevNode, err = nl.skipList.loadElement(nextNode.Prev)
+		prevNode, err = nl.skipList.LoadElement(nextNode.Prev)
 		if err != nil {
 			return err
 		}
@@ -244,14 +244,14 @@ func (nl *NameList) DeleteName(name string) error {
 	// case 3
 	prevNameBatch.DeleteName(name)
 	if len(prevNameBatch.names) == 0 {
-		if err := nl.skipList.Delete(prevNode.Key); err != nil {
+		if _, err := nl.skipList.DeleteByKey(prevNode.Key); err != nil {
 			return err
 		}
 		return nil
 	}
 	if nextNameBatch != nil && len(nextNameBatch.names) + len(prevNameBatch.names) < nl.batchSize {
 		// case 3.1 merge nextNode and prevNode
-		if err := nl.skipList.Delete(nextNode.Key); err != nil {
+		if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
 			return err
 		}
 		for nextName := range nextNameBatch.names {
@@ -294,7 +294,7 @@ func (nl *NameList) ListNames(startFrom string, visitNamesFn func(name string) b
 		if !nextNameBatch.ListNames(startFrom, visitNamesFn) {
 			return nil
 		}
-		nextNode, err = nl.skipList.loadElement(nextNode.Next[0])
+		nextNode, err = nl.skipList.LoadElement(nextNode.Next[0])
 		if err != nil {
 			return err
 		}
@@ -307,16 +307,16 @@ func (nl *NameList) RemoteAllListElement() error {
 
 	t := nl.skipList
 
-	nodeRef := t.startLevels[0]
+	nodeRef := t.StartLevels[0]
 	for nodeRef != nil {
-		node, err := t.loadElement(nodeRef)
+		node, err := t.LoadElement(nodeRef)
 		if err != nil {
 			return err
 		}
 		if node == nil {
 			return nil
 		}
-		if err := t.deleteElement(node); err != nil {
+		if err := t.DeleteElement(node); err != nil {
 			return err
 		}
 		nodeRef = node.Next[0]
diff --git a/weed/util/skiplist/name_list_serde.go b/weed/util/skiplist/name_list_serde.go
index be9f06698..397dfd432 100644
--- a/weed/util/skiplist/name_list_serde.go
+++ b/weed/util/skiplist/name_list_serde.go
@@ -20,16 +20,16 @@ func LoadNameList(data []byte, store ListStore, batchSize int) *NameList {
 	if err := proto.Unmarshal(data, message); err != nil {
 		glog.Errorf("loading skiplist: %v", err)
 	}
-	nl.skipList.maxNewLevel = int(message.MaxNewLevel)
-	nl.skipList.maxLevel = int(message.MaxLevel)
+	nl.skipList.MaxNewLevel = int(message.MaxNewLevel)
+	nl.skipList.MaxLevel = int(message.MaxLevel)
 	for i, ref := range message.StartLevels {
-		nl.skipList.startLevels[i] = &SkipListElementReference{
+		nl.skipList.StartLevels[i] = &SkipListElementReference{
 			ElementPointer: ref.ElementPointer,
 			Key:            ref.Key,
 		}
 	}
 	for i, ref := range message.EndLevels {
-		nl.skipList.endLevels[i] = &SkipListElementReference{
+		nl.skipList.EndLevels[i] = &SkipListElementReference{
 			ElementPointer: ref.ElementPointer,
 			Key:            ref.Key,
 		}
@@ -38,14 +38,14 @@ func LoadNameList(data []byte, store ListStore, batchSize int) *NameList {
 }
 
 func (nl *NameList) HasChanges() bool {
-	return nl.skipList.hasChanges
+	return nl.skipList.HasChanges
 }
 
 func (nl *NameList) ToBytes() []byte {
 	message := &SkipListProto{}
-	message.MaxNewLevel = int32(nl.skipList.maxNewLevel)
-	message.MaxLevel = int32(nl.skipList.maxLevel)
-	for _, ref := range nl.skipList.startLevels {
+	message.MaxNewLevel = int32(nl.skipList.MaxNewLevel)
+	message.MaxLevel = int32(nl.skipList.MaxLevel)
+	for _, ref := range nl.skipList.StartLevels {
 		if ref == nil {
 			break
 		}
@@ -54,7 +54,7 @@ func (nl *NameList) ToBytes() []byte {
 			Key:            ref.Key,
 		})
 	}
-	for _, ref := range nl.skipList.endLevels {
+	for _, ref := range nl.skipList.EndLevels {
 		if ref == nil {
 			break
 		}
diff --git a/weed/util/skiplist/skiplist.go b/weed/util/skiplist/skiplist.go
index 52e6c606a..f42ec23cd 100644
--- a/weed/util/skiplist/skiplist.go
+++ b/weed/util/skiplist/skiplist.go
@@ -17,12 +17,12 @@ const (
 )
 
 type SkipList struct {
-	startLevels [maxLevel]*SkipListElementReference
-	endLevels   [maxLevel]*SkipListElementReference
-	maxNewLevel int
-	maxLevel    int
-	listStore   ListStore
-	hasChanges  bool
+	StartLevels [maxLevel]*SkipListElementReference
+	EndLevels   [maxLevel]*SkipListElementReference
+	MaxNewLevel int
+	MaxLevel    int
+	ListStore   ListStore
+	HasChanges  bool
 	// elementCount int
 }
 
@@ -36,9 +36,9 @@ func NewSeed(seed int64, listStore ListStore) *SkipList {
 	//fmt.Printf("SkipList seed: %v\n", seed)
 
 	list := &SkipList{
-		maxNewLevel: maxLevel,
-		maxLevel:    0,
-		listStore:   listStore,
+		MaxNewLevel: maxLevel,
+		MaxLevel:    0,
+		ListStore:   listStore,
 		// elementCount: 0,
 	}
 
@@ -52,7 +52,7 @@ func New(listStore ListStore) *SkipList {
 
 // IsEmpty checks, if the skiplist is empty.
 func (t *SkipList) IsEmpty() bool {
-	return t.startLevels[0] == nil
+	return t.StartLevels[0] == nil
 }
 
 func (t *SkipList) generateLevel(maxLevel int) int {
@@ -70,8 +70,8 @@ func (t *SkipList) generateLevel(maxLevel int) int {
 
 func (t *SkipList) findEntryIndex(key []byte, minLevel 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 <= minLevel {
+	for i := t.MaxLevel; i >= 0; i-- {
+		if t.StartLevels[i] != nil && bytes.Compare(t.StartLevels[i].Key, key) < 0 || i <= minLevel {
 			return i
 		}
 	}
@@ -90,7 +90,7 @@ func (t *SkipList) findExtended(key []byte, findGreaterOrEqual bool) (prevElemen
 	index := t.findEntryIndex(key, 0)
 	var currentNode *SkipListElement
 
-	currentNode, err = t.loadElement(t.startLevels[index])
+	currentNode, err = t.LoadElement(t.StartLevels[index])
 	if err != nil {
 		return
 	}
@@ -115,7 +115,7 @@ func (t *SkipList) findExtended(key []byte, findGreaterOrEqual bool) (prevElemen
 		// Which direction are we continuing next time?
 		if currentNode.Next[index] != nil && bytes.Compare(currentNode.Next[index].Key, key) <= 0 {
 			// Go right
-			currentNode, err = t.loadElement(currentNode.Next[index])
+			currentNode, err = t.LoadElement(currentNode.Next[index])
 			if err != nil {
 				return
 			}
@@ -129,7 +129,7 @@ func (t *SkipList) findExtended(key []byte, findGreaterOrEqual bool) (prevElemen
 				if currentNode.Next[0] != nil && bytes.Compare(currentNode.Next[0].Key, key) == 0 {
 					prevElementIfVisited = currentNode
 					var currentNodeNext *SkipListElement
-					currentNodeNext, err = t.loadElement(currentNode.Next[0])
+					currentNodeNext, err = t.LoadElement(currentNode.Next[0])
 					if err != nil {
 						return
 					}
@@ -145,7 +145,7 @@ func (t *SkipList) findExtended(key []byte, findGreaterOrEqual bool) (prevElemen
 			} else {
 				// Element is not found and we reached the bottom.
 				if findGreaterOrEqual {
-					foundElem, err = t.loadElement(currentNode.Next[index])
+					foundElem, err = t.LoadElement(currentNode.Next[index])
 					if err != nil {
 						return
 					}
@@ -188,13 +188,13 @@ func (t *SkipList) FindGreaterOrEqual(key []byte) (prevIfVisited *SkipListElemen
 // 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) (err error) {
+func (t *SkipList) DeleteByKey(key []byte) (id int64, err error) {
 
 	if t == nil || t.IsEmpty() || key == nil {
 		return
 	}
 
-	index := t.findEntryIndex(key, t.maxLevel)
+	index := t.findEntryIndex(key, t.MaxLevel)
 
 	var currentNode *SkipListElement
 	var nextNode *SkipListElement
@@ -202,12 +202,12 @@ func (t *SkipList) Delete(key []byte) (err error) {
 	for {
 
 		if currentNode == nil {
-			nextNode, err = t.loadElement(t.startLevels[index])
+			nextNode, err = t.LoadElement(t.StartLevels[index])
 		} else {
-			nextNode, err = t.loadElement(currentNode.Next[index])
+			nextNode, err = t.LoadElement(currentNode.Next[index])
 		}
 		if err != nil {
-			return err
+			return id, err
 		}
 
 		// Found and remove!
@@ -215,45 +215,46 @@ func (t *SkipList) Delete(key []byte) (err error) {
 
 			if currentNode != nil {
 				currentNode.Next[index] = nextNode.Next[index]
-				if err = t.saveElement(currentNode); err != nil {
-					return err
+				if err = t.SaveElement(currentNode); err != nil {
+					return id, err
 				}
 			}
 
 			if index == 0 {
 				if nextNode.Next[index] != nil {
-					nextNextNode, err := t.loadElement(nextNode.Next[index])
+					nextNextNode, err := t.LoadElement(nextNode.Next[index])
 					if err != nil {
-						return err
+						return id, err
 					}
 					if nextNextNode != nil {
 						nextNextNode.Prev = currentNode.Reference()
-						if err = t.saveElement(nextNextNode); err != nil {
-							return err
+						if err = t.SaveElement(nextNextNode); err != nil {
+							return id, err
 						}
 					}
 				}
 				// t.elementCount--
-				if err = t.deleteElement(nextNode); err != nil {
-					return err
+				id = nextNode.Id
+				if err = t.DeleteElement(nextNode); err != nil {
+					return id, err
 				}
 			}
 
 			// Link from start needs readjustments.
-			startNextKey := t.startLevels[index].Key
+			startNextKey := t.StartLevels[index].Key
 			if compareElement(nextNode, startNextKey) == 0 {
-				t.hasChanges = true
-				t.startLevels[index] = nextNode.Next[index]
+				t.HasChanges = true
+				t.StartLevels[index] = nextNode.Next[index]
 				// This was our currently highest node!
-				if t.startLevels[index] == nil {
-					t.maxLevel = index - 1
+				if t.StartLevels[index] == nil {
+					t.MaxLevel = index - 1
 				}
 			}
 
 			// Link from end needs readjustments.
 			if nextNode.Next[index] == nil {
-				t.endLevels[index] = currentNode.Reference()
-				t.hasChanges = true
+				t.EndLevels[index] = currentNode.Reference()
+				t.HasChanges = true
 			}
 			nextNode.Next[index] = nil
 		}
@@ -274,24 +275,28 @@ func (t *SkipList) Delete(key []byte) (err error) {
 
 // Insert inserts the given ListElement into the skiplist.
 // Insert runs in approx. O(log(n))
-func (t *SkipList) Insert(key, value []byte) (err error) {
+func (t *SkipList) InsertByKey(key []byte, idIfKnown int64, value []byte) (id int64, err error) {
 
 	if t == nil || key == nil {
 		return
 	}
 
-	level := t.generateLevel(t.maxNewLevel)
+	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
-		t.hasChanges = true
+	if level > t.MaxLevel {
+		level = t.MaxLevel + 1
+		t.MaxLevel = level
+		t.HasChanges = true
 	}
 
+	id = idIfKnown
+	if id == 0 {
+		id = rand.Int63()
+	}
 	elem := &SkipListElement{
-		Id:    rand.Int63(),
-		Next:  make([]*SkipListElementReference, t.maxNewLevel, t.maxNewLevel),
+		Id:    id,
+		Next:  make([]*SkipListElementReference, t.MaxNewLevel, t.MaxNewLevel),
 		Level: int32(level),
 		Key:   key,
 		Value: value,
@@ -302,8 +307,8 @@ func (t *SkipList) Insert(key, value []byte) (err error) {
 	newFirst := true
 	newLast := true
 	if !t.IsEmpty() {
-		newFirst = compareElement(elem, t.startLevels[0].Key) < 0
-		newLast = compareElement(elem, t.endLevels[0].Key) > 0
+		newFirst = compareElement(elem, t.StartLevels[0].Key) < 0
+		newLast = compareElement(elem, t.EndLevels[0].Key) > 0
 	}
 
 	normallyInserted := false
@@ -319,7 +324,7 @@ func (t *SkipList) Insert(key, value []byte) (err error) {
 		for {
 
 			if currentNode == nil {
-				nextNodeRef = t.startLevels[index]
+				nextNodeRef = t.StartLevels[index]
 			} else {
 				nextNodeRef = currentNode.Next[index]
 			}
@@ -331,19 +336,19 @@ func (t *SkipList) Insert(key, value []byte) (err error) {
 				elem.Next[index] = nextNodeRef
 				if currentNode != nil {
 					currentNode.Next[index] = elem.Reference()
-					if err = t.saveElement(currentNode); err != nil {
+					if err = t.SaveElement(currentNode); err != nil {
 						return
 					}
 				}
 				if index == 0 {
 					elem.Prev = currentNode.Reference()
 					if nextNodeRef != nil {
-						if nextNode, err = t.loadElement(nextNodeRef); err != nil {
+						if nextNode, err = t.LoadElement(nextNodeRef); err != nil {
 							return
 						}
 						if nextNode != nil {
 							nextNode.Prev = elem.Reference()
-							if err = t.saveElement(nextNode); err != nil {
+							if err = t.SaveElement(nextNode); err != nil {
 								return
 							}
 						}
@@ -355,7 +360,7 @@ func (t *SkipList) Insert(key, value []byte) (err error) {
 				// Go right
 				if nextNode == nil {
 					// reuse nextNode when index == 0
-					if nextNode, err = t.loadElement(nextNodeRef); err != nil {
+					if nextNode, err = t.LoadElement(nextNodeRef); err != nil {
 						return
 					}
 				}
@@ -380,28 +385,28 @@ func (t *SkipList) Insert(key, value []byte) (err error) {
 
 		if newFirst || normallyInserted {
 
-			if t.startLevels[i] == nil || bytes.Compare(t.startLevels[i].Key, key) > 0 {
-				if i == 0 && t.startLevels[i] != nil {
-					startLevelElement, err := t.loadElement(t.startLevels[i])
+			if t.StartLevels[i] == nil || bytes.Compare(t.StartLevels[i].Key, key) > 0 {
+				if i == 0 && t.StartLevels[i] != nil {
+					startLevelElement, err := t.LoadElement(t.StartLevels[i])
 					if err != nil {
-						return err
+						return id, err
 					}
 					if startLevelElement != nil {
 						startLevelElement.Prev = elem.Reference()
-						if err = t.saveElement(startLevelElement); err != nil {
-							return err
+						if err = t.SaveElement(startLevelElement); err != nil {
+							return id, err
 						}
 					}
 				}
-				elem.Next[i] = t.startLevels[i]
-				t.startLevels[i] = elem.Reference()
-				t.hasChanges = true
+				elem.Next[i] = t.StartLevels[i]
+				t.StartLevels[i] = elem.Reference()
+				t.HasChanges = true
 			}
 
-			// link the endLevels to this element!
+			// link the EndLevels to this element!
 			if elem.Next[i] == nil {
-				t.endLevels[i] = elem.Reference()
-				t.hasChanges = true
+				t.EndLevels[i] = elem.Reference()
+				t.HasChanges = true
 			}
 
 			didSomething = true
@@ -411,29 +416,29 @@ func (t *SkipList) Insert(key, value []byte) (err error) {
 			// 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, err := t.loadElement(t.endLevels[i])
+				if t.EndLevels[i] != nil {
+					endLevelElement, err := t.LoadElement(t.EndLevels[i])
 					if err != nil {
-						return err
+						return id, err
 					}
 					if endLevelElement != nil {
 						endLevelElement.Next[i] = elem.Reference()
-						if err = t.saveElement(endLevelElement); err != nil {
-							return err
+						if err = t.SaveElement(endLevelElement); err != nil {
+							return id, err
 						}
 					}
 				}
 				if i == 0 {
-					elem.Prev = t.endLevels[i]
+					elem.Prev = t.EndLevels[i]
 				}
-				t.endLevels[i] = elem.Reference()
-				t.hasChanges = true
+				t.EndLevels[i] = elem.Reference()
+				t.HasChanges = true
 			}
 
 			// Link the startLevels to this element!
-			if t.startLevels[i] == nil || bytes.Compare(t.startLevels[i].Key, key) > 0 {
-				t.startLevels[i] = elem.Reference()
-				t.hasChanges = true
+			if t.StartLevels[i] == nil || bytes.Compare(t.StartLevels[i].Key, key) > 0 {
+				t.StartLevels[i] = elem.Reference()
+				t.HasChanges = true
 			}
 
 			didSomething = true
@@ -444,41 +449,41 @@ func (t *SkipList) Insert(key, value []byte) (err error) {
 		}
 	}
 
-	if err = t.saveElement(elem); err != nil {
-		return err
+	if err = t.SaveElement(elem); err != nil {
+		return id, err
 	}
-	return nil
+	return id, nil
 
 }
 
 // GetSmallestNode returns the very first/smallest node in the skiplist.
 // GetSmallestNode runs in O(1)
 func (t *SkipList) GetSmallestNode() (*SkipListElement, error) {
-	return t.loadElement(t.startLevels[0])
+	return t.LoadElement(t.StartLevels[0])
 }
 
 // GetLargestNode returns the very last/largest node in the skiplist.
 // GetLargestNode runs in O(1)
 func (t *SkipList) GetLargestNode() (*SkipListElement, error) {
-	return t.loadElement(t.endLevels[0])
+	return t.LoadElement(t.EndLevels[0])
 }
 
 // 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, error) {
 	if e.Next[0] == nil {
-		return t.loadElement(t.startLevels[0])
+		return t.LoadElement(t.StartLevels[0])
 	}
-	return t.loadElement(e.Next[0])
+	return t.LoadElement(e.Next[0])
 }
 
 // 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, error) {
 	if e.Prev == nil {
-		return t.loadElement(t.endLevels[0])
+		return t.LoadElement(t.EndLevels[0])
 	}
-	return t.loadElement(e.Prev)
+	return t.LoadElement(e.Prev)
 }
 
 // ChangeValue can be used to change the actual value of a node in the skiplist
@@ -488,14 +493,14 @@ func (t *SkipList) Prev(e *SkipListElement) (*SkipListElement, error) {
 func (t *SkipList) ChangeValue(e *SkipListElement, newValue []byte) (err error) {
 	// The key needs to stay correct, so this is very important!
 	e.Value = newValue
-	return t.saveElement(e)
+	return t.SaveElement(e)
 }
 
 // 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 {
+	for i, l := range t.StartLevels {
 		if l == nil {
 			break
 		}
@@ -510,10 +515,10 @@ func (t *SkipList) println() {
 	}
 	println()
 
-	nodeRef := t.startLevels[0]
+	nodeRef := t.StartLevels[0]
 	for nodeRef != nil {
 		print(fmt.Sprintf("%v: ", string(nodeRef.Key)))
-		node, _ := t.loadElement(nodeRef)
+		node, _ := t.LoadElement(nodeRef)
 		if node == nil {
 			break
 		}
@@ -546,7 +551,7 @@ func (t *SkipList) println() {
 	}
 
 	print("end --> ")
-	for i, l := range t.endLevels {
+	for i, l := range t.EndLevels {
 		if l == nil {
 			break
 		}
diff --git a/weed/util/skiplist/skiplist_serde.go b/weed/util/skiplist/skiplist_serde.go
index e528b8a3d..f3b81b7fc 100644
--- a/weed/util/skiplist/skiplist_serde.go
+++ b/weed/util/skiplist/skiplist_serde.go
@@ -19,25 +19,25 @@ func (node *SkipListElement) Reference() *SkipListElementReference {
 	}
 }
 
-func (t *SkipList) saveElement(element *SkipListElement) error {
+func (t *SkipList) SaveElement(element *SkipListElement) error {
 	if element == nil {
 		return nil
 	}
-	return t.listStore.SaveElement(element.Id, element)
+	return t.ListStore.SaveElement(element.Id, element)
 }
 
-func (t *SkipList) deleteElement(element *SkipListElement) error {
+func (t *SkipList) DeleteElement(element *SkipListElement) error {
 	if element == nil {
 		return nil
 	}
-	return t.listStore.DeleteElement(element.Id)
+	return t.ListStore.DeleteElement(element.Id)
 }
 
-func (t *SkipList) loadElement(ref *SkipListElementReference) (*SkipListElement, error) {
+func (t *SkipList) LoadElement(ref *SkipListElementReference) (*SkipListElement, error) {
 	if ref.IsNil() {
 		return nil, nil
 	}
-	return t.listStore.LoadElement(ref.ElementPointer)
+	return t.ListStore.LoadElement(ref.ElementPointer)
 }
 
 func (ref *SkipListElementReference) IsNil() bool {
diff --git a/weed/util/skiplist/skiplist_test.go b/weed/util/skiplist/skiplist_test.go
index a35bef6f3..69b38012f 100644
--- a/weed/util/skiplist/skiplist_test.go
+++ b/weed/util/skiplist/skiplist_test.go
@@ -19,10 +19,10 @@ var (
 func TestReverseInsert(t *testing.T) {
 	list := NewSeed(100, memStore)
 
-	list.Insert([]byte("zzz"), []byte("zzz"))
-	list.Delete([]byte("zzz"))
+	list.InsertByKey([]byte("zzz"), 0, []byte("zzz"))
+	list.DeleteByKey([]byte("zzz"))
 
-	list.Insert([]byte("aaa"), []byte("aaa"))
+	list.InsertByKey([]byte("aaa"), 0, []byte("aaa"))
 
 	if list.IsEmpty() {
 		t.Fail()
@@ -37,7 +37,7 @@ func TestInsertAndFind(t *testing.T) {
 	var list *SkipList
 
 	var listPointer *SkipList
-	listPointer.Insert(k0, k0)
+	listPointer.InsertByKey(k0, 0, k0)
 	if _, _, ok, _ := listPointer.Find(k0); ok {
 		t.Fail()
 	}
@@ -53,7 +53,7 @@ func TestInsertAndFind(t *testing.T) {
 	// Test at the beginning of the list.
 	for i := 0; i < maxN; i++ {
 		key := []byte(strconv.Itoa(maxN - i))
-		list.Insert(key, key)
+		list.InsertByKey(key, 0, key)
 	}
 	for i := 0; i < maxN; i++ {
 		key := []byte(strconv.Itoa(maxN - i))
@@ -66,7 +66,7 @@ func TestInsertAndFind(t *testing.T) {
 	// Test at the end of the list.
 	for i := 0; i < maxN; i++ {
 		key := []byte(strconv.Itoa(i))
-		list.Insert(key, key)
+		list.InsertByKey(key, 0, key)
 	}
 	for i := 0; i < maxN; i++ {
 		key := []byte(strconv.Itoa(i))
@@ -81,7 +81,7 @@ func TestInsertAndFind(t *testing.T) {
 	for _, e := range rList {
 		key := []byte(strconv.Itoa(e))
 		// println("insert", e)
-		list.Insert(key, key)
+		list.InsertByKey(key, 0, key)
 	}
 	for _, e := range rList {
 		key := []byte(strconv.Itoa(e))
@@ -106,27 +106,27 @@ func TestDelete(t *testing.T) {
 	var list *SkipList
 
 	// Delete on empty list
-	list.Delete(k0)
+	list.DeleteByKey(k0)
 
 	list = New(memStore)
 
-	list.Delete(k0)
+	list.DeleteByKey(k0)
 	if !list.IsEmpty() {
 		t.Fail()
 	}
 
-	list.Insert(k0, k0)
-	list.Delete(k0)
+	list.InsertByKey(k0, 0, k0)
+	list.DeleteByKey(k0)
 	if !list.IsEmpty() {
 		t.Fail()
 	}
 
 	// Delete elements at the beginning of the list.
 	for i := 0; i < maxN; i++ {
-		list.Insert(Element(i), Element(i))
+		list.InsertByKey(Element(i), 0, Element(i))
 	}
 	for i := 0; i < maxN; i++ {
-		list.Delete(Element(i))
+		list.DeleteByKey(Element(i))
 	}
 	if !list.IsEmpty() {
 		t.Fail()
@@ -135,10 +135,10 @@ func TestDelete(t *testing.T) {
 	list = New(memStore)
 	// Delete elements at the end of the list.
 	for i := 0; i < maxN; i++ {
-		list.Insert(Element(i), Element(i))
+		list.InsertByKey(Element(i), 0, Element(i))
 	}
 	for i := 0; i < maxN; i++ {
-		list.Delete(Element(maxN - i - 1))
+		list.DeleteByKey(Element(maxN - i - 1))
 	}
 	if !list.IsEmpty() {
 		t.Fail()
@@ -148,10 +148,10 @@ func TestDelete(t *testing.T) {
 	// Delete elements at random positions in the list.
 	rList := rand.Perm(maxN)
 	for _, e := range rList {
-		list.Insert(Element(e), Element(e))
+		list.InsertByKey(Element(e), 0, Element(e))
 	}
 	for _, e := range rList {
-		list.Delete(Element(e))
+		list.DeleteByKey(Element(e))
 	}
 	if !list.IsEmpty() {
 		t.Fail()
@@ -162,7 +162,7 @@ func TestNext(t *testing.T) {
 	list := New(memStore)
 
 	for i := 0; i < maxN; i++ {
-		list.Insert(Element(i), Element(i))
+		list.InsertByKey(Element(i), 0, Element(i))
 	}
 
 	smallest, _ := list.GetSmallestNode()
@@ -194,7 +194,7 @@ func TestPrev(t *testing.T) {
 	list := New(memStore)
 
 	for i := 0; i < maxN; i++ {
-		list.Insert(Element(i), Element(i))
+		list.InsertByKey(Element(i), 0, Element(i))
 	}
 
 	smallest, _ := list.GetSmallestNode()
@@ -237,7 +237,7 @@ func TestFindGreaterOrEqual(t *testing.T) {
 	list = New(memStore)
 
 	for i := 0; i < maxN; i++ {
-		list.Insert(Element(rand.Intn(maxNumber)), Element(i))
+		list.InsertByKey(Element(rand.Intn(maxNumber)), 0, Element(i))
 	}
 
 	for i := 0; i < maxN; i++ {
@@ -271,7 +271,7 @@ func TestChangeValue(t *testing.T) {
 	list := New(memStore)
 
 	for i := 0; i < maxN; i++ {
-		list.Insert(Element(i), []byte("value"))
+		list.InsertByKey(Element(i), 0, []byte("value"))
 	}
 
 	for i := 0; i < maxN; i++ {