TrueCloudLab/frostfs-node
19
1
Fork 27
Code Issues 92 Pull requests 9 Projects Releases 1 Activity Actions

metabase: optimize Inhume operation #1385

Merged
dstepanov-yadro merged 1 commit from dstepanov-yadro/frostfs-node:fix/metabase_inhume_hangs into master 2024-10-26 11:30:26 +00:00
Conversation 1 Commits 1 Files changed 1 +11 -18
1 changed files with 11 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
pkg/local_object_storage/metabase/inhume.go
View file

@ -377,11 +377,8 @@ func (db *DB) getInhumeTargetBucketAndValue(garbageBKT, graveyardBKT *bbolt.Buck
return targetBucket, value, nil
}
func (db *DB) markAsGC(graveyardBKT, garbageBKT *bbolt.Bucket, key []byte) (bool, error) {
targetIsTomb, err := isTomb(graveyardBKT, key)
if err != nil {
return false, err
}
func (db *DB) markAsGC(graveyardBKT, garbageBKT *bbolt.Bucket, addressKey []byte) (bool, error) {
targetIsTomb := isTomb(graveyardBKT, addressKey)
// do not add grave if target is a tombstone
if targetIsTomb {
@ -390,7 +387,7 @@ func (db *DB) markAsGC(graveyardBKT, garbageBKT *bbolt.Bucket, key []byte) (bool
// if tombstone appears object must be
// additionally marked with GC
return false, garbageBKT.Put(key, zeroValue)
return false, garbageBKT.Put(addressKey, zeroValue)
}
func (db *DB) updateDeleteInfo(tx *bbolt.Tx, garbageBKT, graveyardBKT *bbolt.Bucket, targetKey []byte, cnr cid.ID, obj *objectSDK.Object, res *InhumeRes) error {
@ -410,25 +407,21 @@ func (db *DB) updateDeleteInfo(tx *bbolt.Tx, garbageBKT, graveyardBKT *bbolt.Buc
return nil
}
func isTomb(graveyardBucket *bbolt.Bucket, key []byte) (bool, error) {
func isTomb(graveyardBucket *bbolt.Bucket, addressKey []byte) bool {
targetIsTomb := false
// iterate over graveyard and check if target address
// is the address of tombstone in graveyard.
err := graveyardBucket.ForEach(func(_, v []byte) error {
// tombstone must have the same container ID as key.
c := graveyardBucket.Cursor()
containerPrefix := addressKey[:cidSize]
for k, v := c.Seek(containerPrefix); k != nil && bytes.HasPrefix(k, containerPrefix); k, v = c.Next() {
// check if graveyard has record with key corresponding
// to tombstone address (at least one)
targetIsTomb = bytes.Equal(v, key)
targetIsTomb = bytes.Equal(v, addressKey)
if targetIsTomb {
// break bucket iterator
return errBreakBucketForEach
break
}
return nil
})
if err != nil && !errors.Is(err, errBreakBucketForEach) {
return false, err
}
return targetIsTomb, nil
return targetIsTomb
}