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

[#377] metabase: Support batch Inhume operation

Browse source 
Replace single target address in `InhumePrm` with the list of addresses.
Rename `WithAddress` method to `WithAddresses` and change parameter to
variadic.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2021-02-15 14:50:21 +03:00 committed by Alex Vanin
parent 6ec7433e14
commit 0d6d195d0d
1 changed files with 33 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -7,16 +7,18 @@ import (
// InhumePrm encapsulates parameters for Inhume operation. // InhumePrm encapsulates parameters for Inhume operation.
type InhumePrm struct { type InhumePrm struct {
target, tomb *objectSDK.Address tomb *objectSDK.Address
target []*objectSDK.Address
} }
// InhumeRes encapsulates results of Inhume operation. // InhumeRes encapsulates results of Inhume operation.
type InhumeRes struct{} type InhumeRes struct{}
// WithAddress sets object address that should be inhumed. // WithAddresses sets list of object addresses that should be inhumed.
func (p *InhumePrm) WithAddress(addr *objectSDK.Address) *InhumePrm { func (p *InhumePrm) WithAddresses(addrs ...*objectSDK.Address) *InhumePrm {
if p != nil { if p != nil {
p.target = addr p.target = addrs
} }
return p return p
@ -50,7 +52,7 @@ func (p *InhumePrm) WithGCMark() *InhumePrm {
// tomb should not be nil. // tomb should not be nil.
func Inhume(db *DB, target, tomb *objectSDK.Address) error { func Inhume(db *DB, target, tomb *objectSDK.Address) error {
_, err := db.Inhume(new(InhumePrm). _, err := db.Inhume(new(InhumePrm).
WithAddress(target). WithAddresses(target).
WithTombstoneAddress(tomb), WithTombstoneAddress(tomb),
) )
@ -67,31 +69,38 @@ func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) {
return err return err
} }
obj, err := db.get(tx, prm.target, false, true) for i := range prm.target {
obj, err := db.get(tx, prm.target[i], false, true)
// if object is stored and it is regular object then update bucket // if object is stored and it is regular object then update bucket
// with container size estimations // with container size estimations
if err == nil && obj.Type() == objectSDK.TypeRegular { if err == nil && obj.Type() == objectSDK.TypeRegular {
err := changeContainerSize( err := changeContainerSize(
tx, tx,
obj.ContainerID(), obj.ContainerID(),
obj.PayloadSize(), obj.PayloadSize(),
false, false,
) )
if err != nil {
return err
}
}
var val []byte
if prm.tomb != nil {
val = addressKey(prm.tomb)
} else {
val = []byte(inhumeGCMarkValue)
}
// consider checking if target is already in graveyard?
err = graveyard.Put(addressKey(prm.target[i]), val)
if err != nil { if err != nil {
return err return err
} }
} }
var val []byte return nil
if prm.tomb != nil {
val = addressKey(prm.tomb)
} else {
val = []byte(inhumeGCMarkValue)
}
// consider checking if target is already in graveyard?
return graveyard.Put(addressKey(prm.target), val)
}) })
return return