[#1616] engine: Do not use batches in delete

Use a simple loop at the callsite. This way we remove as much as we can.
Also, `Delete` metrics is more meaningful now.

Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>
This commit is contained in:
Evgenii Stratonikov 2022-09-14 17:18:11 +03:00 committed by fyrchik
parent cda8f9df2e
commit b064fb24d8
3 changed files with 46 additions and 41 deletions

View file

@ -11,7 +11,7 @@ import (
// DeletePrm groups the parameters of Delete operation.
type DeletePrm struct {
addr []oid.Address
addr oid.Address
forceRemoval bool
}
@ -19,12 +19,12 @@ type DeletePrm struct {
// DeleteRes groups the resulting values of Delete operation.
type DeleteRes struct{}
// WithAddresses is a Delete option to set the addresses of the objects to delete.
// WithAddress is a Delete option to set the addresses of the objects to delete.
//
// Option is required.
func (p *DeletePrm) WithAddresses(addr ...oid.Address) {
func (p *DeletePrm) WithAddress(addr oid.Address) {
if p != nil {
p.addr = append(p.addr, addr...)
p.addr = addr
}
}
@ -66,43 +66,41 @@ func (e *StorageEngine) delete(prm DeletePrm) (DeleteRes, error) {
err apistatus.ObjectLocked
}
for i := range prm.addr {
e.iterateOverSortedShards(prm.addr[i], func(_ int, sh hashedShard) (stop bool) {
var existsPrm shard.ExistsPrm
existsPrm.SetAddress(prm.addr[i])
e.iterateOverSortedShards(prm.addr, func(_ int, sh hashedShard) (stop bool) {
var existsPrm shard.ExistsPrm
existsPrm.SetAddress(prm.addr)
resExists, err := sh.Exists(existsPrm)
if err != nil {
_, ok := err.(*objectSDK.SplitInfoError)
if ok || shard.IsErrRemoved(err) || shard.IsErrObjectExpired(err) {
return true
}
if !shard.IsErrNotFound(err) {
e.reportShardError(sh, "could not check object existence", err)
}
return false
} else if !resExists.Exists() {
return false
resExists, err := sh.Exists(existsPrm)
if err != nil {
_, ok := err.(*objectSDK.SplitInfoError)
if ok || shard.IsErrRemoved(err) || shard.IsErrObjectExpired(err) {
return true
}
var shPrm shard.InhumePrm
shPrm.MarkAsGarbage(prm.addr[i])
if prm.forceRemoval {
shPrm.ForceRemoval()
if !shard.IsErrNotFound(err) {
e.reportShardError(sh, "could not check object existence", err)
}
return false
} else if !resExists.Exists() {
return false
}
_, err = sh.Inhume(shPrm)
if err != nil {
e.reportShardError(sh, "could not inhume object in shard", err)
var shPrm shard.InhumePrm
shPrm.MarkAsGarbage(prm.addr)
if prm.forceRemoval {
shPrm.ForceRemoval()
}
locked.is = errors.As(err, &locked.err)
_, err = sh.Inhume(shPrm)
if err != nil {
e.reportShardError(sh, "could not inhume object in shard", err)
return locked.is
}
locked.is = errors.As(err, &locked.err)
return true
})
}
return locked.is
}
return true
})
if locked.is {
return DeleteRes{}, locked.err

View file

@ -299,7 +299,7 @@ func TestLockForceRemoval(t *testing.T) {
// 4.
var deletePrm DeletePrm
deletePrm.WithAddresses(objectcore.AddressOf(lock))
deletePrm.WithAddress(objectcore.AddressOf(lock))
deletePrm.WithForceRemoval()
_, err = e.Delete(deletePrm)

View file

@ -35,13 +35,20 @@ func (s *Server) DropObjects(_ context.Context, req *control.DropObjectsRequest)
}
}
var prm engine.DeletePrm
prm.WithAddresses(addrList...)
prm.WithForceRemoval()
var firstErr error
for i := range addrList {
var prm engine.DeletePrm
prm.WithForceRemoval()
prm.WithAddress(addrList[i])
_, err := s.s.Delete(prm)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
_, err := s.s.Delete(prm)
if err != nil && firstErr == nil {
firstErr = err
}
}
if firstErr != nil {
return nil, status.Error(codes.Internal, firstErr.Error())
}
// create and fill response