[#1450] engine: Make Inhume operation handle objects in parallel
All checks were successful
DCO action / DCO (pull_request) Successful in 1m14s
Tests and linters / Run gofumpt (pull_request) Successful in 1m51s
Pre-commit hooks / Pre-commit (pull_request) Successful in 2m52s
Vulncheck / Vulncheck (pull_request) Successful in 2m55s
Build / Build Components (pull_request) Successful in 3m13s
Tests and linters / Staticcheck (pull_request) Successful in 3m26s
Tests and linters / gopls check (pull_request) Successful in 3m29s
Tests and linters / Lint (pull_request) Successful in 4m21s
Tests and linters / Tests with -race (pull_request) Successful in 5m23s
Tests and linters / Tests (pull_request) Successful in 5m37s

Add a worker pool for `Inhume` operation and use it to handle objects
in parallel. Since metabase `Inhume` uses `bbolt.Batch`, handling many
objects one by one may be inefficient.

```
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.45m ± 0%   11.43m ±  0%        ~ (p=0.315 n=10)
InhumeMultipart/objects=10-12       113.94m ± 1%   11.60m ±  0%  -89.82% (p=0.000 n=10)
InhumeMultipart/objects=100-12     1144.08m ± 1%   30.17m ± 12%  -97.36% (p=0.000 n=10)
InhumeMultipart/objects=1000-12    11367.5m ± 0%   265.7m ±  2%  -97.66% (p=0.000 n=10)
InhumeMultipart/objects=10000-12    113.505 ± 0%    2.436 ±  1%  -97.85% (p=0.000 n=10)
geomean                               1.140        76.32m        -93.31%
```

Signed-off-by: Aleksey Savchuk <a.savchuk@yadro.com>
This commit is contained in:
Aleksey Savchuk 2024-11-13 18:23:21 +03:00
parent 6e21fd350f
commit 59821fb271
Signed by: a-savchuk
GPG key ID: 70C0A7FF6F9C4639
3 changed files with 97 additions and 30 deletions

View file

@ -11,6 +11,7 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
"github.com/panjf2000/ants/v2"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
) )
@ -71,10 +72,14 @@ func (e *StorageEngine) open(ctx context.Context) error {
} }
// Init initializes all StorageEngine's components. // Init initializes all StorageEngine's components.
func (e *StorageEngine) Init(ctx context.Context) error { func (e *StorageEngine) Init(ctx context.Context) (err error) {
e.mtx.Lock() e.mtx.Lock()
defer e.mtx.Unlock() defer e.mtx.Unlock()
if e.inhumePool, err = ants.NewPool(int(e.inhumePoolSize)); err != nil {
return fmt.Errorf("could not create pool: %w", err)
}
errCh := make(chan shardInitError, len(e.shards)) errCh := make(chan shardInitError, len(e.shards))
var eg errgroup.Group var eg errgroup.Group
if e.cfg.lowMem && e.anyShardRequiresRefill() { if e.cfg.lowMem && e.anyShardRequiresRefill() {
@ -92,7 +97,7 @@ func (e *StorageEngine) Init(ctx context.Context) error {
return nil return nil
}) })
} }
err := eg.Wait() err = eg.Wait()
close(errCh) close(errCh)
if err != nil { if err != nil {
return fmt.Errorf("failed to initialize shards: %w", err) return fmt.Errorf("failed to initialize shards: %w", err)
@ -161,6 +166,7 @@ func (e *StorageEngine) close(ctx context.Context, releasePools bool) error {
for _, p := range e.shardPools { for _, p := range e.shardPools {
p.Release() p.Release()
} }
e.inhumePool.Release()
} }
for id, sh := range e.shards { for id, sh := range e.shards {

View file

@ -30,6 +30,8 @@ type StorageEngine struct {
shardPools map[string]util.WorkerPool shardPools map[string]util.WorkerPool
inhumePool util.WorkerPool
closeCh chan struct{} closeCh chan struct{}
setModeCh chan setModeRequest setModeCh chan setModeRequest
wg sync.WaitGroup wg sync.WaitGroup
@ -191,6 +193,8 @@ type cfg struct {
shardPoolSize uint32 shardPoolSize uint32
inhumePoolSize uint32
lowMem bool lowMem bool
containerSource atomic.Pointer[containerSource] containerSource atomic.Pointer[containerSource]
@ -200,6 +204,7 @@ func defaultCfg() *cfg {
res := &cfg{ res := &cfg{
log: logger.NewLoggerWrapper(zap.L()), log: logger.NewLoggerWrapper(zap.L()),
shardPoolSize: 20, shardPoolSize: 20,
inhumePoolSize: 50,
metrics: noopMetrics{}, metrics: noopMetrics{},
} }
res.containerSource.Store(&containerSource{}) res.containerSource.Store(&containerSource{})
@ -244,6 +249,13 @@ func WithShardPoolSize(sz uint32) Option {
} }
} }
// WithInhumePoolSize returns option to specify size of worker pool for Inhume operation.
func WithInhumePoolSize(sz uint32) Option {
return func(c *cfg) {
c.inhumePoolSize = sz
}
}
// WithErrorThreshold returns an option to specify size amount of errors after which // WithErrorThreshold returns an option to specify size amount of errors after which
// shard is moved to read-only mode. // shard is moved to read-only mode.
func WithErrorThreshold(sz uint32) Option { func WithErrorThreshold(sz uint32) Option {

View file

@ -3,6 +3,7 @@ package engine
import ( import (
"context" "context"
"errors" "errors"
"sync"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase" meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
@ -81,45 +82,93 @@ 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) {
defer elapsed("Inhume", e.metrics.AddMethodDuration)()
var retErr error
wg := sync.WaitGroup{}
errOnce := sync.Once{}
ctx, cancel := context.WithCancel(ctx)
defer cancel()
var shPrm shard.InhumePrm var shPrm shard.InhumePrm
if prm.forceRemoval { if prm.forceRemoval {
shPrm.ForceRemoval() shPrm.ForceRemoval()
} }
for i := range prm.addrs { loop:
if !prm.forceRemoval { for _, addr := range prm.addrs {
locked, err := e.IsLocked(ctx, prm.addrs[i]) select {
if err != nil { case <-ctx.Done():
e.log.Warn(ctx, logs.EngineRemovingAnObjectWithoutFullLockingCheck, break loop
zap.Error(err), default:
zap.Stringer("addr", prm.addrs[i]),
zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
} else if locked {
return InhumeRes{}, new(apistatus.ObjectLocked)
}
} }
if prm.tombstone != nil { wg.Add(1)
shPrm.SetTarget(*prm.tombstone, prm.addrs[i]) if err := e.inhumePool.Submit(func() {
} else { defer wg.Done()
shPrm.MarkAsGarbage(prm.addrs[i])
}
ok, err := e.inhumeAddr(ctx, prm.addrs[i], shPrm, true) err := e.handleInhumeTask(ctx, addr, prm.tombstone, prm.forceRemoval)
if err != nil { if err != nil {
return InhumeRes{}, err errOnce.Do(func() {
retErr = err
cancel()
})
} }
if !ok { }); err != nil {
ok, err := e.inhumeAddr(ctx, prm.addrs[i], shPrm, false) wg.Done()
if err != nil { cancel()
return InhumeRes{}, err wg.Wait()
} else if !ok {
e.log.Warn(ctx, logs.EngineCouldNotInhumeObjectInShard, zap.Error(err))
return InhumeRes{}, errInhumeFailure return InhumeRes{}, errInhumeFailure
} }
} }
wg.Wait()
return InhumeRes{}, retErr
}
func (e *StorageEngine) handleInhumeTask(ctx context.Context, addr oid.Address, tombstone *oid.Address, forceRemoval bool) error {
if !forceRemoval {
locked, err := e.IsLocked(ctx, addr)
if err != nil {
e.log.Warn(ctx, logs.EngineRemovingAnObjectWithoutFullLockingCheck,
zap.Error(err),
zap.Stringer("addr", addr),
zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
} else if locked {
return new(apistatus.ObjectLocked)
}
} }
return InhumeRes{}, nil var prm shard.InhumePrm
if tombstone != nil {
prm.SetTarget(*tombstone, addr)
} else {
prm.MarkAsGarbage(addr)
}
if forceRemoval {
prm.ForceRemoval()
}
ok, err := e.inhumeAddr(ctx, addr, prm, true)
if err != nil {
return err
}
if !ok {
ok, err := e.inhumeAddr(ctx, addr, prm, false)
if err != nil {
return err
}
if !ok {
return errInhumeFailure
}
}
return nil
} }
// Returns ok if object was inhumed during this invocation or before. // Returns ok if object was inhumed during this invocation or before.