frostfs-node/pkg/services/policer/process.go
Alejandro Lopez 26acf5689e
All checks were successful
ci/woodpecker/push/pre-commit Pipeline was successful
[#92] Ensure policer objects cannot be worked on concurrently
Signed-off-by: Alejandro Lopez <a.lopez@yadro.com>
2023-07-03 07:05:31 +00:00

90 lines
2 KiB
Go

package policer
import (
"context"
"errors"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
"go.uber.org/zap"
)
func (p *Policer) Run(ctx context.Context) {
go p.poolCapacityWorker(ctx)
p.shardPolicyWorker(ctx)
p.log.Info(logs.PolicerRoutineStopped)
}
func (p *Policer) shardPolicyWorker(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
default:
}
addrs, err := p.keySpaceIterator.Next(ctx, 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(logs.PolicerFailureAtObjectSelectForReplication, 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) < p.evictDuration {
return
}
if p.objsInWork.add(addr.Address) {
p.processObject(ctx, addr)
p.cache.Add(addr.Address, time.Now())
p.objsInWork.remove(addr.Address)
}
})
if err != nil {
p.log.Warn(logs.PolicerPoolSubmission, zap.Error(err))
}
}
}
}
}
func (p *Policer) poolCapacityWorker(ctx context.Context) {
ticker := time.NewTicker(p.rebalanceFreq)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
frostfsSysLoad := p.loader.ObjectServiceLoad()
newCapacity := int((1.0 - frostfsSysLoad) * float64(p.maxCapacity))
if newCapacity == 0 {
newCapacity++
}
if p.taskPool.Cap() != newCapacity {
p.taskPool.Tune(newCapacity)
p.log.Debug(logs.PolicerTuneReplicationCapacity,
zap.Float64("system_load", frostfsSysLoad),
zap.Int("new_capacity", newCapacity))
}
}
}
}