forked from TrueCloudLab/frostfs-node
[#1450] engine: Group object by shard before Inhume
``` goos: linux goarch: amd64 pkg: git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine cpu: 12th Gen Intel(R) Core(TM) i5-1235U │ old.txt │ new.txt │ │ sec/op │ sec/op vs base │ InhumeMultipart/objects=1-12 11.42m ± 1% 10.71m ± 0% -6.27% (p=0.000 n=10) InhumeMultipart/objects=10-12 113.5m ± 0% 100.9m ± 3% -11.08% (p=0.000 n=10) InhumeMultipart/objects=100-12 1135.4m ± 1% 681.3m ± 2% -40.00% (p=0.000 n=10) InhumeMultipart/objects=1000-12 11.358 ± 0% 1.089 ± 1% -90.41% (p=0.000 n=10) InhumeMultipart/objects=10000-12 113.251 ± 0% 1.645 ± 1% -98.55% (p=0.000 n=10) geomean 1.136 265.5m -76.63% ``` Signed-off-by: Aleksey Savchuk <a.savchuk@yadro.com>
This commit is contained in:
parent
b348b20289
commit
281d65435e
1 changed files with 110 additions and 74 deletions
|
@ -81,110 +81,146 @@ func (e *StorageEngine) Inhume(ctx context.Context, prm InhumePrm) (res InhumeRe
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *StorageEngine) inhume(ctx context.Context, prm InhumePrm) (InhumeRes, error) {
|
func (e *StorageEngine) inhume(ctx context.Context, prm InhumePrm) (InhumeRes, error) {
|
||||||
|
addrsPerShard, err := e.groupObjectsByShard(ctx, prm.addrs, !prm.forceRemoval)
|
||||||
|
if err != nil {
|
||||||
|
return InhumeRes{}, err
|
||||||
|
}
|
||||||
|
|
||||||
var shPrm shard.InhumePrm
|
var shPrm shard.InhumePrm
|
||||||
if prm.forceRemoval {
|
if prm.forceRemoval {
|
||||||
shPrm.ForceRemoval()
|
shPrm.ForceRemoval()
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := range prm.addrs {
|
var errLocked *apistatus.ObjectLocked
|
||||||
if !prm.forceRemoval {
|
|
||||||
locked, err := e.IsLocked(ctx, prm.addrs[i])
|
|
||||||
if err != nil {
|
|
||||||
e.log.Warn(ctx, logs.EngineRemovingAnObjectWithoutFullLockingCheck,
|
|
||||||
zap.Error(err),
|
|
||||||
zap.Stringer("addr", prm.addrs[i]),
|
|
||||||
zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
|
|
||||||
} else if locked {
|
|
||||||
return InhumeRes{}, new(apistatus.ObjectLocked)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
|
for shardID, addrs := range addrsPerShard {
|
||||||
if prm.tombstone != nil {
|
if prm.tombstone != nil {
|
||||||
shPrm.SetTarget(*prm.tombstone, prm.addrs[i])
|
shPrm.SetTarget(*prm.tombstone, addrs...)
|
||||||
} else {
|
} else {
|
||||||
shPrm.MarkAsGarbage(prm.addrs[i])
|
shPrm.MarkAsGarbage(addrs...)
|
||||||
}
|
}
|
||||||
|
|
||||||
ok, err := e.inhumeAddr(ctx, prm.addrs[i], shPrm, true)
|
sh, exists := e.shards[shardID]
|
||||||
if err != nil {
|
if !exists {
|
||||||
return InhumeRes{}, err
|
e.log.Warn(ctx, logs.EngineCouldNotInhumeObjectInShard,
|
||||||
}
|
zap.Error(errors.New("this shard was expected to exist")),
|
||||||
if !ok {
|
zap.String("shard_id", shardID),
|
||||||
ok, err := e.inhumeAddr(ctx, prm.addrs[i], shPrm, false)
|
zap.String("trace_id", tracingPkg.GetTraceID(ctx)),
|
||||||
if err != nil {
|
)
|
||||||
return InhumeRes{}, err
|
|
||||||
} else if !ok {
|
|
||||||
return InhumeRes{}, errInhumeFailure
|
return InhumeRes{}, errInhumeFailure
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if _, err := sh.Inhume(ctx, shPrm); err != nil {
|
||||||
|
switch {
|
||||||
|
case errors.As(err, &errLocked):
|
||||||
|
case errors.Is(err, shard.ErrLockObjectRemoval):
|
||||||
|
case errors.Is(err, shard.ErrReadOnlyMode):
|
||||||
|
case errors.Is(err, shard.ErrDegradedMode):
|
||||||
|
default:
|
||||||
|
e.reportShardError(ctx, sh, "couldn't inhume object in shard", err)
|
||||||
|
}
|
||||||
|
return InhumeRes{}, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return InhumeRes{}, nil
|
return InhumeRes{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Returns ok if object was inhumed during this invocation or before.
|
// groupObjectsByShard groups objects based on the shard(s) they are stored on.
|
||||||
func (e *StorageEngine) inhumeAddr(ctx context.Context, addr oid.Address, prm shard.InhumePrm, checkExists bool) (bool, error) {
|
//
|
||||||
root := false
|
// If checkLocked is set, [apistatus.ObjectLocked] will be returned if any of
|
||||||
var existPrm shard.ExistsPrm
|
// the objects are locked.
|
||||||
var retErr error
|
func (e *StorageEngine) groupObjectsByShard(ctx context.Context, addrs []oid.Address, checkLocked bool) (map[string][]oid.Address, error) {
|
||||||
var ok bool
|
groups := make(map[string][]oid.Address)
|
||||||
|
|
||||||
|
for _, addr := range addrs {
|
||||||
|
ids, err := e.findShards(ctx, addr, checkLocked)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
for _, id := range ids {
|
||||||
|
groups[id] = append(groups[id], addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return groups, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// findShards determines the shard(s) where the object is stored.
|
||||||
|
//
|
||||||
|
// If the object is a root object, multiple shards will be returned.
|
||||||
|
//
|
||||||
|
// If checkLocked is set, [apistatus.ObjectLocked] will be returned if any of
|
||||||
|
// the objects are locked.
|
||||||
|
func (e *StorageEngine) findShards(ctx context.Context, addr oid.Address, checkLocked bool) ([]string, error) {
|
||||||
|
var (
|
||||||
|
ids []string
|
||||||
|
retErr error
|
||||||
|
|
||||||
|
prm shard.ExistsPrm
|
||||||
|
|
||||||
|
siErr *objectSDK.SplitInfoError
|
||||||
|
ecErr *objectSDK.ECInfoError
|
||||||
|
|
||||||
|
isRootObject bool
|
||||||
|
objectExists bool
|
||||||
|
)
|
||||||
|
|
||||||
e.iterateOverSortedShards(addr, func(_ int, sh hashedShard) (stop bool) {
|
e.iterateOverSortedShards(addr, func(_ int, sh hashedShard) (stop bool) {
|
||||||
defer func() {
|
objectExists = false
|
||||||
// if object is root we continue since information about it
|
|
||||||
// can be presented in other shards
|
|
||||||
if checkExists && root {
|
|
||||||
stop = false
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
if checkExists {
|
prm.Address = addr
|
||||||
existPrm.Address = addr
|
switch res, err := sh.Exists(ctx, prm); {
|
||||||
exRes, err := sh.Exists(ctx, existPrm)
|
case client.IsErrObjectAlreadyRemoved(err) || shard.IsErrObjectExpired(err):
|
||||||
if err != nil {
|
// NOTE(@a-savchuk): there were some considerations that we can stop
|
||||||
if client.IsErrObjectAlreadyRemoved(err) || shard.IsErrObjectExpired(err) {
|
// immediately if the object is already removed or expired. However,
|
||||||
// inhumed once - no need to be inhumed again
|
// the previous method behavior was:
|
||||||
ok = true
|
// - keep iterating if it's a root object and already removed,
|
||||||
return true
|
// - stop iterating if it's not a root object and removed.
|
||||||
|
//
|
||||||
|
// Since my task was only improving method speed, let's keep the
|
||||||
|
// previous method behavior. Continue if it's a root object.
|
||||||
|
return !isRootObject
|
||||||
|
case errors.As(err, &siErr) || errors.As(err, &ecErr):
|
||||||
|
isRootObject = true
|
||||||
|
objectExists = true
|
||||||
|
case err != nil:
|
||||||
|
e.reportShardError(
|
||||||
|
ctx, sh, "couldn't check for presence in shard",
|
||||||
|
err, zap.Stringer("address", addr),
|
||||||
|
)
|
||||||
|
case res.Exists():
|
||||||
|
objectExists = true
|
||||||
|
default:
|
||||||
}
|
}
|
||||||
|
|
||||||
var siErr *objectSDK.SplitInfoError
|
if !objectExists {
|
||||||
var ecErr *objectSDK.ECInfoError
|
|
||||||
if !(errors.As(err, &siErr) || errors.As(err, &ecErr)) {
|
|
||||||
e.reportShardError(ctx, sh, "could not check for presents in shard", err, zap.Stringer("address", addr))
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
root = true
|
if checkLocked {
|
||||||
} else if !exRes.Exists() {
|
if isLocked, err := sh.IsLocked(ctx, addr); err != nil {
|
||||||
return
|
e.log.Warn(ctx, logs.EngineRemovingAnObjectWithoutFullLockingCheck,
|
||||||
}
|
zap.Error(err),
|
||||||
}
|
zap.Stringer("address", addr),
|
||||||
|
zap.String("trace_id", tracingPkg.GetTraceID(ctx)),
|
||||||
_, err := sh.Inhume(ctx, prm)
|
)
|
||||||
if err != nil {
|
} else if isLocked {
|
||||||
var errLocked *apistatus.ObjectLocked
|
|
||||||
switch {
|
|
||||||
case errors.As(err, &errLocked):
|
|
||||||
retErr = new(apistatus.ObjectLocked)
|
retErr = new(apistatus.ObjectLocked)
|
||||||
return true
|
return true
|
||||||
case errors.Is(err, shard.ErrLockObjectRemoval):
|
}
|
||||||
retErr = shard.ErrLockObjectRemoval
|
|
||||||
return true
|
|
||||||
case errors.Is(err, shard.ErrReadOnlyMode) || errors.Is(err, shard.ErrDegradedMode):
|
|
||||||
retErr = err
|
|
||||||
return true
|
|
||||||
}
|
}
|
||||||
|
|
||||||
e.reportShardError(ctx, sh, "could not inhume object in shard", err, zap.Stringer("address", addr))
|
ids = append(ids, sh.ID().String())
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
ok = true
|
// Continue if it's a root object.
|
||||||
return true
|
return !isRootObject
|
||||||
})
|
})
|
||||||
|
|
||||||
return ok, retErr
|
if retErr != nil {
|
||||||
|
return nil, retErr
|
||||||
|
}
|
||||||
|
return ids, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// IsLocked checks whether an object is locked according to StorageEngine's state.
|
// IsLocked checks whether an object is locked according to StorageEngine's state.
|
||||||
|
|
Loading…
Reference in a new issue