seaweedfs/weed/util/skiplist/name_list.go

327 lines
7 KiB
Go
Raw Normal View History

2021-10-04 00:54:25 +00:00
package skiplist
import (
"bytes"
)
type NameList struct {
skipList *SkipList
batchSize int
}
2021-10-04 08:01:31 +00:00
func newNameList(store ListStore, batchSize int) *NameList {
2021-10-04 00:54:25 +00:00
return &NameList{
skipList: New(store),
batchSize: batchSize,
}
}
/*
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 *NameList) WriteName(name string) error {
2021-10-04 08:01:31 +00:00
2021-10-04 00:54:25 +00:00
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 {
2021-10-07 01:18:24 +00:00
prevNode, err = nl.skipList.LoadElement(nextNode.Prev)
2021-10-04 00:54:25 +00:00
if err != nil {
return err
}
}
if prevNode != nil {
prevNameBatch := LoadNameBatch(prevNode.Value)
// case 2.1
if prevNameBatch.ContainsName(name) {
return nil
}
// case 2.2
if len(prevNameBatch.names) < nl.batchSize {
prevNameBatch.WriteName(name)
return nl.skipList.ChangeValue(prevNode, prevNameBatch.ToBytes())
}
// case 2.3
x, y := prevNameBatch.SplitBy(name)
addToX := len(x.names) <= len(y.names)
if len(x.names) != len(prevNameBatch.names) {
if addToX {
x.WriteName(name)
}
if x.key == prevNameBatch.key {
if err := nl.skipList.ChangeValue(prevNode, x.ToBytes()); err != nil {
return err
}
} else {
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.InsertByKey([]byte(x.key), 0, x.ToBytes()); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
}
}
if len(y.names) != len(prevNameBatch.names) {
if !addToX {
y.WriteName(name)
}
if y.key == prevNameBatch.key {
if err := nl.skipList.ChangeValue(prevNode, y.ToBytes()); err != nil {
return err
}
} else {
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.InsertByKey([]byte(y.key), 0, y.ToBytes()); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
}
}
return nil
}
// case 2.4
if nextNode != nil {
nextNameBatch := LoadNameBatch(nextNode.Value)
if len(nextNameBatch.names) < nl.batchSize {
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
nextNameBatch.WriteName(name)
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.InsertByKey([]byte(nextNameBatch.key), 0, nextNameBatch.ToBytes()); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
return nil
}
}
// case 2.5
// now prevNode is nil
newNameBatch := NewNameBatch()
newNameBatch.WriteName(name)
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.InsertByKey([]byte(newNameBatch.key), 0, newNameBatch.ToBytes()); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
return nil
}
/*
// 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 *NameList) DeleteName(name string) error {
lookupKey := []byte(name)
prevNode, nextNode, found, err := nl.skipList.FindGreaterOrEqual(lookupKey)
if err != nil {
return err
}
// case 1
var nextNameBatch *NameBatch
if nextNode != nil {
nextNameBatch = LoadNameBatch(nextNode.Value)
}
if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
nextNameBatch.DeleteName(name)
if len(nextNameBatch.names) > 0 {
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.InsertByKey([]byte(nextNameBatch.key), 0, nextNameBatch.ToBytes()); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
}
return nil
}
if !found {
prevNode, err = nl.skipList.GetLargestNode()
if err != nil {
return err
}
}
if nextNode != nil && prevNode == nil {
2021-10-07 01:18:24 +00:00
prevNode, err = nl.skipList.LoadElement(nextNode.Prev)
2021-10-04 00:54:25 +00:00
if err != nil {
return err
}
}
// case 2
if prevNode == nil {
// case 2.1
return nil
}
prevNameBatch := LoadNameBatch(prevNode.Value)
if !prevNameBatch.ContainsName(name) {
// case 2.2
return nil
}
// case 3
prevNameBatch.DeleteName(name)
if len(prevNameBatch.names) == 0 {
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.DeleteByKey(prevNode.Key); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
return nil
}
2021-10-08 04:13:31 +00:00
if nextNameBatch != nil && len(nextNameBatch.names)+len(prevNameBatch.names) < nl.batchSize {
2021-10-04 00:54:25 +00:00
// case 3.1 merge nextNode and prevNode
2021-10-07 01:18:24 +00:00
if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
2021-10-04 00:54:25 +00:00
return err
}
for nextName := range nextNameBatch.names {
prevNameBatch.WriteName(nextName)
}
return nl.skipList.ChangeValue(prevNode, prevNameBatch.ToBytes())
} else {
// case 3.2 update prevNode
return nl.skipList.ChangeValue(prevNode, prevNameBatch.ToBytes())
}
return nil
}
func (nl *NameList) 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 {
prevNameBatch := LoadNameBatch(prevNode.Value)
if !prevNameBatch.ListNames(startFrom, visitNamesFn) {
return nil
}
}
for nextNode != nil {
nextNameBatch := LoadNameBatch(nextNode.Value)
if !nextNameBatch.ListNames(startFrom, visitNamesFn) {
return nil
}
2021-10-07 01:18:24 +00:00
nextNode, err = nl.skipList.LoadElement(nextNode.Next[0])
2021-10-04 00:54:25 +00:00
if err != nil {
return err
}
}
2021-10-08 04:13:31 +00:00
return nil
2021-10-04 00:54:25 +00:00
}
2021-10-04 08:01:31 +00:00
func (nl *NameList) RemoteAllListElement() error {
t := nl.skipList
2021-10-07 01:18:24 +00:00
nodeRef := t.StartLevels[0]
2021-10-04 08:01:31 +00:00
for nodeRef != nil {
2021-10-07 01:18:24 +00:00
node, err := t.LoadElement(nodeRef)
2021-10-04 08:01:31 +00:00
if err != nil {
return err
}
if node == nil {
return nil
}
2021-10-07 01:18:24 +00:00
if err := t.DeleteElement(node); err != nil {
2021-10-04 08:01:31 +00:00
return err
}
nodeRef = node.Next[0]
}
return nil
2021-10-08 04:13:31 +00:00
}