seaweedfs/weed/filer/redis3/ItemList.go

package redis3

import (
	"bytes"
	"context"
	"fmt"
	"github.com/go-redis/redis/v8"
	"github.com/seaweedfs/seaweedfs/weed/util/skiplist"
)

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) canAddMember(node *skiplist.SkipListElementReference, name string) (alreadyContains bool, nodeSize int, err error) {
	ctx := context.Background()
	pipe := nl.client.TxPipeline()
	key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
	countOperation := pipe.ZLexCount(ctx, key, "-", "+")
	scoreOperationt := pipe.ZScore(ctx, key, name)
	if _, err = pipe.Exec(ctx); err != nil && err != redis.Nil {
		return false, 0, err
	}
	if err == redis.Nil {
		err = nil
	}
	alreadyContains = scoreOperationt.Err() == nil
	nodeSize = int(countOperation.Val())
	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
	}

	var prevNodeReference *skiplist.SkipListElementReference
	if !found {
		prevNodeReference = nl.skipList.GetLargestNodeReference()
	}

	if nextNode != nil && prevNode == nil {
		prevNodeReference = nextNode.Prev
	}

	if prevNodeReference != nil {
		alreadyContains, nodeSize, err := nl.canAddMember(prevNodeReference, name)
		if err != nil {
			return err
		}
		if alreadyContains {
			// case 2.1
			return nil
		}

		// case 2.2
		if nodeSize < nl.batchSize {
			return nl.NodeAddMember(prevNodeReference, name)
		}

		// case 2.3
		x := nl.NodeInnerPosition(prevNodeReference, 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(prevNodeReference, name)
			if err != nil {
				return nil
			}
			// delete skiplist reference to old node
			if _, err := nl.skipList.DeleteByKey(prevNodeReference.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(prevNodeReference, name); err != nil {
				return nil
			}

			// point skip list to current Y
			if err := nl.ItemAdd(lookupKey, prevNodeReference.ElementPointer); err != nil {
				return nil
			}
			return nil
		} else {
			// collect names after name, add them to Y
			namesToY, err := nl.NodeRangeAfterExclusive(prevNodeReference, 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(prevNodeReference, 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.NodeScanInclusiveAfter(prevNode.Reference(), startFrom, visitNamesFn) {
			return nil
		}
	}

	for nextNode != nil {
		if !nl.NodeScanInclusiveAfter(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 {
	if node == nil {
		return 0
	}
	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.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
		Min:    "-",
		Max:    "+",
		Offset: 0,
		Count:  1,
	}).Val()
	if len(slice) > 0 {
		s := slice[0]
		return s
	}
	return ""
}

func (nl *ItemList) NodeScanInclusiveAfter(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
}