Implement throttling #144
Implement throttling #144
Conversation
| if dps == 0 { | ||
| dps = rate.Inf | ||
| } | ||
| limiter := rate.NewLimiter(dps, int(m.MaxBurstSize)) |
There was a problem hiding this comment.
When MaxDispatchesPerSecond is empty, we set it to Inf, meaning no limit. Though MaxBurstSize is also empty in such case, we don't have to care it because it is ignored.
https://pkg.go.dev/golang.org/x/time/rate#pkg-constants
Inf is the infinite rate limit; it allows all events (even if burst is zero).
https://pkg.go.dev/golang.org/x/time/rate#Limiter.WaitN
The burst limit is ignored if the rate limit is Inf.
| @@ -155,8 +183,11 @@ Loop: | |||
| go func(job jobqueue.Job) { | |||
| defer wg.Done() | |||
| defer func() { <-d.sem }() | |||
| rslt := d.worker.Work(job) | |||
| d.jobqueue.Complete(job, rslt) | |||
| err := d.limiter.Wait(ctx) | |||
There was a problem hiding this comment.
https://pkg.go.dev/golang.org/x/time/rate#Limiter.WaitN
It returns an error if n exceeds the Limiter's burst size, the Context is canceled, or the expected wait time exceeds the Context's Deadline.
There was a problem hiding this comment.
[QUESTION]
If there are more than one gorutines waiting for limiter.Wait() here, can we assume that they will be unblocked in FIFO order?
There was a problem hiding this comment.
Yes, the order is preserved. Wait internally uses reserveN API, which returns delay taking early reservations into consideration. So, for example, if the first goroutine waited for 2 secs, the next would wait for 4 secs.
| @@ -155,8 +183,11 @@ Loop: | |||
| go func(job jobqueue.Job) { | |||
| defer wg.Done() | |||
| defer func() { <-d.sem }() | |||
| rslt := d.worker.Work(job) | |||
| d.jobqueue.Complete(job, rslt) | |||
| err := d.limiter.Wait(ctx) | |||
There was a problem hiding this comment.
[QUESTION]
If there are more than one gorutines waiting for limiter.Wait() here, can we assume that they will be unblocked in FIFO order?
| rslt := d.worker.Work(job) | ||
| d.jobqueue.Complete(job, rslt) | ||
| err := d.limiter.Wait(ctx) | ||
| if err == nil { |
There was a problem hiding this comment.
[QUESTION]
It looks that the job will be lost when it exceeded the burst size. Shouldn't we retry the job later?
There was a problem hiding this comment.
No, I think we shoudn't.
https://pkg.go.dev/golang.org/x/time/rate#Limiter.WaitN
It returns an error if n exceeds the Limiter's burst size
In this case, n is always 1 because we use Wait. Although we have to care for burst size = 0, this is a broken job queue basically (0 burst size allows no execution). We will never encounter the case.
However, as you worry about, some jobs would stay grabbed when the dispatcher is stopped. Should we do something? If so, is there a recommended way to recover?
There was a problem hiding this comment.
We will never encounter the case.
Ah, I see, I've misunderstood about this.
If so, is there a recommended way to recover?
We don't have to worry about this. We already have a mechanism to recover grabbed jobs in primary-backup job queue activation.
|
Thank you! |
Related issues: #138 #139 #141
This p-r implements throttled dispatcher using max_dispatches_per_second and max_busrt_size.