From 6010dfdf3de79a81e70adfad5cd4e0d2c678904b Mon Sep 17 00:00:00 2001 From: Evgenii Stratonikov Date: Thu, 8 Feb 2024 10:01:27 +0300 Subject: [PATCH] [#969] policer: Make error skip thread-safe Introduces in afd2ba9a66. Refs #914 Signed-off-by: Evgenii Stratonikov --- pkg/services/policer/process.go | 30 ++++++++++++++++++++++-------- 1 file changed, 22 insertions(+), 8 deletions(-) diff --git a/pkg/services/policer/process.go b/pkg/services/policer/process.go index df4f0c4c7..0f1198f8d 100644 --- a/pkg/services/policer/process.go +++ b/pkg/services/policer/process.go @@ -3,6 +3,7 @@ package policer import ( "context" "errors" + "sync" "time" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" @@ -35,9 +36,7 @@ func (p *Policer) shardPolicyWorker(ctx context.Context) { p.log.Warn(logs.PolicerFailureAtObjectSelectForReplication, zap.Error(err)) } - // contains all errors logged in this iteration for each container - cnrErrSkip := make(map[cid.ID][]error) - + skipMap := newSkipMap() for i := range addrs { select { case <-ctx.Done(): @@ -59,12 +58,10 @@ func (p *Policer) shardPolicyWorker(ctx context.Context) { if p.objsInWork.add(addr.Address) { err := p.processObject(ctx, addr) - if err != nil && !skipLog(cnrErrSkip[addr.Address.Container()], err) { + if err != nil && !skipMap.addSeenError(addr.Address.Container(), err) { p.log.Error(logs.PolicerUnableToProcessObj, zap.Stringer("object", addr.Address), zap.String("error", err.Error())) - - cnrErrSkip[addr.Address.Container()] = append(cnrErrSkip[addr.Address.Container()], err) } p.cache.Add(addr.Address, time.Now()) p.objsInWork.remove(addr.Address) @@ -79,11 +76,28 @@ func (p *Policer) shardPolicyWorker(ctx context.Context) { } } -func skipLog(errs []error, err error) bool { - for _, e := range errs { +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 }