Fix queue processor throttling logic
@@ -70,6 +70,7 @@ type (
rescheduler queues.Rescheduler
lastPollTime time.Time
+ backoffTimerLock sync.Mutex
backoffTimer *time.Timer
readTaskRetrier backoff.Retrier
@@ -269,21 +270,23 @@ func (p *queueProcessorBase) processBatch() {
func (p *queueProcessorBase) verifyReschedulerSize() bool {
passed := p.rescheduler.Len() < p.options.MaxReschdulerSize()
- if passed && p.backoffTimer != nil {
- p.backoffTimer.Stop()
- p.backoffTimer = nil
- }
- if !passed && p.backoffTimer == nil {
+ if !passed {
p.throttle(p.options.PollBackoffInterval())
}
-
return passed
}
func (p *queueProcessorBase) throttle(duration time.Duration) {
+ p.backoffTimerLock.Lock()
+ defer p.backoffTimerLock.Unlock()
+
if p.backoffTimer == nil {
p.backoffTimer = time.AfterFunc(duration, func() {
+ p.backoffTimerLock.Lock()
+ defer p.backoffTimerLock.Unlock()
+
p.notifyNewTask() // re-enqueue the event
+ p.backoffTimer = nil
})
}
}
