frostfs-node/pkg/services/policer/process.go
Evgenii Stratonikov 6010dfdf3d [#969] policer: Make error skip thread-safe
Introduces in afd2ba9a66.
Refs #914

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
2024-02-08 10:10:41 +03:00

103 lines
2.3 KiB
Go

package policer
import (
"context"
"errors"
"sync"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"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"
)
func (p *Policer) Run(ctx context.Context) {
p.shardPolicyWorker(ctx)
p.log.Info(logs.PolicerRoutineStopped)
}
func (p *Policer) shardPolicyWorker(ctx context.Context) {
for {
select {
case <-ctx.Done():
p.taskPool.Release()
return
default:
}
addrs, err := p.keySpaceIterator.Next(ctx, p.batchSize)
if err != nil {
if errors.Is(err, engine.ErrEndOfListing) {
p.keySpaceIterator.Rewind()
time.Sleep(p.sleepDuration) // finished whole cycle, sleep a bit
continue
}
p.log.Warn(logs.PolicerFailureAtObjectSelectForReplication, zap.Error(err))
}
skipMap := newSkipMap()
for i := 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
}
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))
}
}
}
}
}
type errMap struct {
sync.Mutex
skipMap map[cid.ID][]error
}
func newSkipMap() *errMap {
return &errMap{
skipMap: make(map[cid.ID][]error),
}
}
// addSeenError marks err as seen error for the container.
// Returns true is the error has already been added.
func (m *errMap) addSeenError(cnr cid.ID, err error) bool {
m.Lock()
defer m.Unlock()
for _, e := range m.skipMap[cnr] {
if errors.Is(err, e) {
return true
}
}
m.skipMap[cnr] = append(m.skipMap[cnr], err)
return false
}