frostfs-node/pkg/services/policer/process.go
Evgenii Stratonikov 1779664644 [#2058] services/policer: Fix panic in shardPolicyWorker
```
2022/11/15 08:40:56 worker exits from a panic: runtime error: index out of range [0] with length 0
2022/11/15 08:40:56 worker exits from panic: goroutine 1188 [running]:
github.com/panjf2000/ants/v2.(*goWorker).run.func1.1()
	github.com/panjf2000/ants/v2@v2.4.0/worker.go:58 +0x10c
panic({0x1042b60, 0xc0015ae018})
	runtime/panic.go:1038 +0x215
github.com/nspcc-dev/neofs-node/pkg/services/policer.(*Policer).shardPolicyWorker.func1()
	github.com/nspcc-dev/neofs-node/pkg/services/policer/process.go:65 +0x366
github.com/panjf2000/ants/v2.(*goWorker).run.func1()
	github.com/panjf2000/ants/v2@v2.4.0/worker.go:68 +0x97
created by github.com/panjf2000/ants/v2.(*goWorker).run
	github.com/panjf2000/ants/v2@v2.4.0/worker.go:48 +0x68
```

Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>
2022-11-19 11:01:04 +03:00

100 lines
2 KiB
Go

package policer
import (
"context"
"errors"
"time"
objectcore "github.com/nspcc-dev/neofs-node/pkg/core/object"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine"
"go.uber.org/zap"
)
func (p *Policer) Run(ctx context.Context) {
defer func() {
p.log.Info("routine stopped")
}()
go p.poolCapacityWorker(ctx)
p.shardPolicyWorker(ctx)
}
func (p *Policer) shardPolicyWorker(ctx context.Context) {
var (
addrs []objectcore.AddressWithType
cursor *engine.Cursor
err error
)
for {
select {
case <-ctx.Done():
return
default:
}
addrs, cursor, err = p.jobQueue.Select(cursor, p.batchSize)
if err != nil {
if errors.Is(err, engine.ErrEndOfListing) {
time.Sleep(time.Second) // finished whole cycle, sleep a bit
continue
}
p.log.Warn("failure at object select for replication", zap.Error(err))
}
for i := range addrs {
select {
case <-ctx.Done():
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.(time.Time)) < p.evictDuration {
return
}
p.objsInWork.add(addr.Address)
p.processObject(ctx, addr)
p.cache.Add(addr, time.Now())
p.objsInWork.remove(addr.Address)
})
if err != nil {
p.log.Warn("pool submission", zap.Error(err))
}
}
}
}
}
func (p *Policer) poolCapacityWorker(ctx context.Context) {
ticker := time.NewTicker(p.rebalanceFreq)
for {
select {
case <-ctx.Done():
ticker.Stop()
return
case <-ticker.C:
neofsSysLoad := p.loader.ObjectServiceLoad()
newCapacity := int((1.0 - neofsSysLoad) * float64(p.maxCapacity))
if newCapacity == 0 {
newCapacity++
}
if p.taskPool.Cap() != newCapacity {
p.taskPool.Tune(newCapacity)
p.log.Debug("tune replication capacity",
zap.Float64("system_load", neofsSysLoad),
zap.Int("new_capacity", newCapacity))
}
}
}
}