feat/throttled-dispatcher#8
Closed
mborders wants to merge 2 commits into
Closed
Conversation
There was a problem hiding this comment.
Pull request overview
Adds a new “throttled dispatcher” wrapper to rate-limit job dispatching (requested in Issue #6: Throttling) so callers can constrain how many jobs are queued over a time interval.
Changes:
- Introduces
ThrottleDispatcherwith a token-bucket style throttling mechanism around the existingDispatcher. - Adds a unit test validating that dispatching is limited to a configured number of jobs per time window.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
| throttled_dispatcher.go | New throttling wrapper around Dispatcher using a ticker-based token bucket and stop channel. |
| throttled_dispatcher_test.go | Adds a timing-based test to verify throttling behavior. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+5
to
+12
| // ThrottleDispatcher wraps a Dispatcher with rate-limiting behavior, | ||
| // allowing at most maxJobs to be dispatched per interval. | ||
| type ThrottleDispatcher struct { | ||
| *Dispatcher | ||
| maxJobs int | ||
| interval time.Duration | ||
| tokens chan struct{} | ||
| quit chan bool |
Comment on lines
+38
to
+42
| // Stop halts the throttled dispatcher and stops token replenishment. | ||
| func (td *ThrottleDispatcher) Stop() { | ||
| td.quit <- true | ||
| td.Dispatcher.Stop() | ||
| } |
Comment on lines
+65
to
+69
| // Dispatch waits for a throttle token before queuing the job, | ||
| // ensuring no more than maxJobs are dispatched per interval. | ||
| func (td *ThrottleDispatcher) Dispatch(run func()) error { | ||
| <-td.tokens // Block until a token is available | ||
| return td.Dispatcher.Dispatch(run) |
Comment on lines
+15
to
+22
| // NewThrottledDispatcher creates a new dispatcher that limits job execution | ||
| // to at most maxJobs per interval (e.g., 5 jobs per time.Minute). | ||
| func NewThrottledDispatcher(maxWorkers int, maxQueue int, maxJobs int, interval time.Duration) *ThrottleDispatcher { | ||
| tokens := make(chan struct{}, maxJobs) | ||
| // Pre-fill the token bucket | ||
| for i := 0; i < maxJobs; i++ { | ||
| tokens <- struct{}{} | ||
| } |
Comment on lines
+44
to
+67
| // replenish refills the token bucket at the configured rate. | ||
| // Tokens are added in batches of maxJobs every interval. | ||
| func (td *ThrottleDispatcher) replenish() { | ||
| ticker := time.NewTicker(td.interval) | ||
| defer ticker.Stop() | ||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| for i := 0; i < td.maxJobs; i++ { | ||
| select { | ||
| case td.tokens <- struct{}{}: | ||
| default: | ||
| // Bucket is full; discard extra tokens | ||
| } | ||
| } | ||
| case <-td.quit: | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Dispatch waits for a throttle token before queuing the job, | ||
| // ensuring no more than maxJobs are dispatched per interval. | ||
| func (td *ThrottleDispatcher) Dispatch(run func()) error { |
Comment on lines
+33
to
+37
| // Wait for second window to open | ||
| time.Sleep(600 * time.Millisecond) | ||
| total := atomic.LoadInt32(&count) | ||
| if total != 6 { | ||
| t.Errorf("expected 6 total jobs, got %d", total) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.