306
var reconsiderFwdTimerC <-chan time.Time
307
if fwdTokenC != nil && !tm.isBacklogNegligible() {
308
>
// If there is a non-negligible backlog, we stop forwarding to make sure
matcher.go
309
>
// root and leaf partitions are treated equally and can process their
310
>
// backlog at the same rate. Stopping task forwarding, prevent poll
311
>
// forwarding as well (in presence of a backlog). This ensures all partitions
312
>
// receive polls and tasks at the same rate.
313
>
314
>
// Exception: we allow forward if this partition has not got any polls
315
>
// recently. This is helpful when there are very few pollers and they
316
>
// and they are all stuck in the wrong (root) partition. (Note that since
317
>
// frontend balanced the number of pending pollers per partition this only
318
>
// becomes an issue when the pollers are fewer than the partitions)
319
>
lp := tm.timeSinceLastPoll()
320
>
maxWaitForLocalPoller := tm.config.MaxWaitForPollerBeforeFwd()
321
>
if lp < maxWaitForLocalPoller {
322
>
fwdTokenC = nil
323
>
reconsiderFwdTimer = time.NewTimer(maxWaitForLocalPoller - lp)
324
>
reconsiderFwdTimerC = reconsiderFwdTimer.C
325
>
}
326
}
327