TrueCloudLab/frostfs-node
20
1
Fork 27
Code Issues 110 Pull requests 6 Projects Releases 2 Activity Actions

[#1445] policer: Refactor (*Policer).shardPolicyWorker

Browse source 
It's because of warnings from `gocognit` linter.

Signed-off-by: Aleksey Savchuk <a.savchuk@yadro.com>
This commit is contained in:
Aleksey Savchuk 2024-11-08 11:30:30 +03:00
parent 4e316367f1
commit 165dd6d545
Signed by: a-savchuk
GPG key ID: 70C0A7FF6F9C4639
1 changed files with 35 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
pkg/services/policer/process.go
View file

@ -7,6 +7,7 @@ import (
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"go.uber.org/zap"
@ -18,10 +19,11 @@ func (p *Policer) Run(ctx context.Context) {
}
func (p *Policer) shardPolicyWorker(ctx context.Context) {
defer p.taskPool.Release()
for {
select {
case <-ctx.Done():
p.taskPool.Release()
return
default:
}
@ -37,45 +39,48 @@ func (p *Policer) shardPolicyWorker(ctx context.Context) {
}
skipMap := newSkipMap()
for i := range addrs {
for _, addr := range addrs {
select {
case <-ctx.Done():
p.taskPool.Release()
return
default:
addr := addrs[i]
if p.objsInWork.inWork(addr.Address) {
// do not process an object
// that is in work
continue
}
}
if p.objsInWork.inWork(addr.Address) {
// do not process an object that is in work
continue
}
err := p.taskPool.Submit(func() {
v, ok := p.cache.Get(addr.Address)
if ok && time.Since(v) < p.evictDuration {
return
}
if p.objsInWork.add(addr.Address) {
err := p.processObject(ctx, addr)
if err != nil && !skipMap.addSeenError(addr.Address.Container(), err) {
p.log.Error(logs.PolicerUnableToProcessObj,
zap.Stringer("object", addr.Address),
zap.String("error", err.Error()))
}
p.cache.Add(addr.Address, time.Now())
p.objsInWork.remove(addr.Address)
p.metrics.IncProcessedObjects()
}
})
if err != nil {
p.log.Warn(logs.PolicerPoolSubmission, zap.Error(err))
}
if err := p.submitPolicerTask(ctx, addr, skipMap); err != nil {
p.log.Warn(logs.PolicerPoolSubmission, zap.Error(err))
}
}
}
}
func (p *Policer) submitPolicerTask(ctx context.Context, addr object.Info, skipMap *errMap) error {
return p.taskPool.Submit(func() {
v, ok := p.cache.Get(addr.Address)
if ok && time.Since(v) < p.evictDuration {
return
}
if !p.objsInWork.add(addr.Address) {
return
}
err := p.processObject(ctx, addr)
if err != nil && !skipMap.addSeenError(addr.Address.Container(), err) {
p.log.Error(logs.PolicerUnableToProcessObj,
zap.Stringer("object", addr.Address),
zap.String("error", err.Error()))
}
p.cache.Add(addr.Address, time.Now())
p.objsInWork.remove(addr.Address)
p.metrics.IncProcessedObjects()
})
}
type errMap struct {
sync.Mutex
skipMap map[cid.ID][]error