ale64bit/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#1481] pilorama: Fix TreeApply

Browse source 
Current implementation prevents invalid operations to become valid at
some later point (consider adding a child to the non-existent parent and
then adding the parent). This seems to diverge from the paper algorithm
and complicates implementation. Make it simpler.

Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
Evgenii Stratonikov 2022-06-06 13:14:27 +03:00 committed by fyrchik
parent 5ffbeb76e6
commit 735931c842
3 changed files with 7 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
pkg/local_object_storage/pilorama/boltdb.go
View file

@ -270,9 +270,7 @@ func (t *boltForest) applyOperation(logBucket, treeBucket *bbolt.Bucket, lm *Log
} }
func (t *boltForest) do(lb *bbolt.Bucket, b *bbolt.Bucket, key []byte, op *LogMove) error { func (t *boltForest) do(lb *bbolt.Bucket, b *bbolt.Bucket, key []byte, op *LogMove) error {
shouldPut := !t.isAncestor(b, key, op.Child, op.Parent) && shouldPut := !t.isAncestor(b, key, op.Child, op.Parent)
!(op.Parent != 0 && op.Parent != TrashID && b.Get(timestampKey(key, op.Parent)) == nil)
shouldRemove := op.Parent == TrashID
currParent := b.Get(parentKey(key, op.Child)) currParent := b.Get(parentKey(key, op.Child))
if currParent != nil { // node is already in tree if currParent != nil { // node is already in tree
@ -292,16 +290,6 @@ func (t *boltForest) do(lb *bbolt.Bucket, b *bbolt.Bucket, key []byte, op *LogMo
return nil return nil
} }
if shouldRemove {
if currParent != nil {
p := binary.LittleEndian.Uint64(currParent)
if err := b.Delete(childrenKey(key, op.Child, p)); err != nil {
return err
}
}
return t.removeNode(b, key, op.Child, op.Parent)
}
if currParent == nil { if currParent == nil {
if err := b.Put(timestampKey(key, op.Child), toUint64(op.Time)); err != nil { if err := b.Put(timestampKey(key, op.Child), toUint64(op.Time)); err != nil {
return err return err

9
pkg/local_object_storage/pilorama/forest_test.go
View file

@ -354,16 +354,17 @@ func testForestTreeApply(t *testing.T, constructor func(t testing.TB) Forest) {
testMeta(t, s, cid, treeID, 11, 10, meta) testMeta(t, s, cid, treeID, 11, 10, meta)
testApply(t, s, 10, TrashID, Meta{Time: 2, Items: []KeyValue{{"parent", []byte{2}}}}) testApply(t, s, 10, TrashID, Meta{Time: 2, Items: []KeyValue{{"parent", []byte{2}}}})
testMeta(t, s, cid, treeID, 11, 0, Meta{}) testMeta(t, s, cid, treeID, 11, 10, meta)
}) })
t.Run("add a child to non-existent parent, then add a parent", func(t *testing.T) { t.Run("add a child to non-existent parent, then add a parent", func(t *testing.T) {
s := constructor(t) s := constructor(t)
testApply(t, s, 11, 10, Meta{Time: 1, Items: []KeyValue{{"child", []byte{3}}}}) meta := Meta{Time: 1, Items: []KeyValue{{"child", []byte{3}}}}
testMeta(t, s, cid, treeID, 11, 0, Meta{}) testApply(t, s, 11, 10, meta)
testMeta(t, s, cid, treeID, 11, 10, meta)
testApply(t, s, 10, 0, Meta{Time: 2, Items: []KeyValue{{"grand", []byte{1}}}}) testApply(t, s, 10, 0, Meta{Time: 2, Items: []KeyValue{{"grand", []byte{1}}}})
testMeta(t, s, cid, treeID, 11, 0, Meta{}) testMeta(t, s, cid, treeID, 11, 10, meta)
}) })
} }

14
pkg/local_object_storage/pilorama/inmemory.go
View file

@ -86,11 +86,7 @@ func (s *state) do(op *Move) LogMove {
}, },
} }
_, parentInTree := s.tree.infoMap[op.Parent] shouldPut := !s.tree.isAncestor(op.Child, op.Parent)
shouldPut := !s.tree.isAncestor(op.Child, op.Parent) &&
!(op.Parent != 0 && op.Parent != TrashID && !parentInTree)
shouldRemove := op.Parent == TrashID
p, ok := s.tree.infoMap[op.Child] p, ok := s.tree.infoMap[op.Child]
if ok { if ok {
lm.HasOld = true lm.HasOld = true
@ -101,14 +97,6 @@ func (s *state) do(op *Move) LogMove {
return lm return lm
} }
if shouldRemove {
if ok {
s.removeChild(op.Child, p.Parent)
}
delete(s.tree.infoMap, op.Child)
return lm
}
if !ok { if !ok {
p.Timestamp = op.Time p.Timestamp = op.Time
} else { } else {