-
Notifications
You must be signed in to change notification settings - Fork 4.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Change the router reload suppression so that it doesn't block updates
This changes the locking so that a reload doesn't hold a lock of the router object for the duration of the reload so that updates that happen while the router is reloading can be processed immediately and batched up, then included when the next reload occurs. Before this, if a reload ran longer than the reload interval, only one event would be processed before triggering a new reload. Which would then lock out other event processing. This caused the router to not make any meaningful progress consuming events. A new module to do the rate limiting has been added. The module has have a top and bottom half. The top half simply calls the bottom half with a flag indicating the user has made a change. The flag simply tells the bottom half to register the desire to reload (so we can do it under a single lock acquisition). The bottom half is in charge of determining if it can immediately reload or if it has to wait. If it must wait, then it works out the earliest time it can reload and schedules a callback to itself for that time. If it determines it can reload, then it runs the reload code immediately. When the reload is complete, it calls itself again to make sure there was no other pending reload that had come in while the reload was running. Whenever the bottom half calls itself, it does it without the flag indicating the user made a change. Fixes bug 1471899 -- https://bugzilla.redhat.com/show_bug.cgi?id=1471899
- Loading branch information
Showing
5 changed files
with
234 additions
and
25 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| package limiter | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/golang/glog" | ||
| utilruntime "k8s.io/apimachinery/pkg/util/runtime" | ||
| ) | ||
|
|
||
| // HandlerFunc defines function signature for a CoalescingSerializingRateLimiter. | ||
| type HandlerFunc func() error | ||
|
|
||
| // CoalescingSerializingRateLimiter guarantees that calls will not happen to the given function | ||
| // more frequently than the given interval, and it guarantees that only one call will happen at a time. | ||
| // The calls are not queued, i.e. if you make 5 calls to RegisterChange(), it does not guarantee that the | ||
| // handler will be invoked 5 times, it merely guarantees it will be invoked once, and no more often than | ||
| // the rate. | ||
| // The calls to the handler will happen in the background and are expected to do their own locking if needed. | ||
| type CoalescingSerializingRateLimiter struct { | ||
| // handlerFunc is the function to rate limit and seriaize calls to. | ||
| handlerFunc HandlerFunc | ||
|
|
||
| // callInterval is the minimum time between the starts of handler calls. | ||
| callInterval time.Duration | ||
|
|
||
| // lastStart is the time the last run of the handler started. | ||
| lastStart time.Time | ||
|
|
||
| // changeReqTime is nil if no change has been registered since the last handler run completed, otherwise it is the | ||
| // time last change was registered. | ||
| changeReqTime *time.Time | ||
|
|
||
| // handlerRunning indicates whether the Handler is actively running. | ||
| handlerRunning bool | ||
|
|
||
| // lock protects the CoalescingSerializingRateLimiter structure from multiple threads manipulating it at once. | ||
| lock sync.Mutex | ||
|
|
||
| // callbackTimer is the timer we use to make callbacks to re-run the function to decide if we need to do work. | ||
| callbackTimer *time.Timer | ||
| } | ||
|
|
||
| func NewCoalescingSerializingRateLimiter(interval time.Duration, handlerFunc HandlerFunc) *CoalescingSerializingRateLimiter { | ||
| limiter := &CoalescingSerializingRateLimiter{ | ||
| handlerFunc: handlerFunc, | ||
| callInterval: interval, | ||
| lastStart: time.Time{}, | ||
| changeReqTime: nil, | ||
| handlerRunning: false, | ||
| } | ||
|
|
||
| return limiter | ||
| } | ||
|
|
||
| // RegisterChange() indicates that the rate limited function should be called. It may not immediately run it, but it will cause it to run within | ||
| // the ReloadInterval. It will always immediately return, the function will be run in the background. Not every call to RegisterChange() will | ||
| // result in the function getting called. If it is called repeatedly while it is still within the ReloadInterval since the last run, it will | ||
| // only run once when the time allows it. | ||
| func (csrl *CoalescingSerializingRateLimiter) RegisterChange() { | ||
| glog.V(8).Infof("RegisterChange called") | ||
|
|
||
| csrl.changeWorker(true) | ||
| } | ||
|
|
||
| func (csrl *CoalescingSerializingRateLimiter) changeWorker(userChanged bool) { | ||
| csrl.lock.Lock() | ||
| defer csrl.lock.Unlock() | ||
|
|
||
| glog.V(8).Infof("changeWorker called") | ||
|
|
||
| if userChanged && csrl.changeReqTime == nil { | ||
| // They just registered a change manually (and we aren't in the middle of a change) | ||
| now := time.Now() | ||
| csrl.changeReqTime = &now | ||
| } | ||
|
|
||
| if csrl.handlerRunning { | ||
| // We don't need to do anything else... there's a run in progress, and when it is done it will re-call this function at which point the work will then happen | ||
| glog.V(8).Infof("The handler was already running (%v) started at %s, returning from the worker", csrl.handlerRunning, csrl.lastStart.String()) | ||
| return | ||
| } | ||
|
|
||
| if csrl.changeReqTime == nil { | ||
| // There's no work queued so we have nothing to do. We should only get here when | ||
| // the function is re-called after a reload | ||
| glog.V(8).Infof("No invoke requested time, so there's no queued work. Nothing to do.") | ||
| return | ||
| } | ||
|
|
||
| // There is no handler running, let's see if we should run yet, or schedule a callback | ||
| now := time.Now() | ||
| sinceLastRun := now.Sub(csrl.lastStart) | ||
| untilNextCallback := csrl.callInterval - sinceLastRun | ||
| glog.V(8).Infof("Checking reload; now: %v, lastStart: %v, sinceLast %v, limit %v, remaining %v", now, csrl.lastStart, sinceLastRun, csrl.callInterval, untilNextCallback) | ||
|
|
||
| if untilNextCallback > 0 { | ||
| // We want to reload... but can't yet because some window is not satisfied | ||
| if csrl.callbackTimer == nil { | ||
| csrl.callbackTimer = time.AfterFunc(untilNextCallback, func() { csrl.changeWorker(false) }) | ||
| } else { | ||
| // While we are resetting the timer, it should have fired and be stopped. | ||
| // The first time the worker is called it will know the precise duration | ||
| // until when a run would be valid and has scheduled a timer for that point | ||
| csrl.callbackTimer.Reset(untilNextCallback) | ||
| } | ||
|
|
||
| glog.V(8).Infof("Can't invoke the handler yet, need to delay %s, callback scheduled", untilNextCallback.String()) | ||
|
|
||
| return | ||
| } | ||
|
|
||
| // Otherwise we can reload immediately... let's do it! | ||
| glog.V(8).Infof("Calling the handler function (for invoke time %v)", csrl.changeReqTime) | ||
| csrl.handlerRunning = true | ||
| csrl.changeReqTime = nil | ||
| csrl.lastStart = now | ||
|
|
||
| // Go run the handler so we don't block the caller | ||
| go csrl.runHandler() | ||
|
|
||
| return | ||
| } | ||
|
|
||
| func (csrl *CoalescingSerializingRateLimiter) runHandler() { | ||
| // Call the handler, but do it in its own function so we can cleanup in case the handler panics | ||
| runHandler := func() error { | ||
| defer func() { | ||
| csrl.lock.Lock() | ||
| csrl.handlerRunning = false | ||
| csrl.lock.Unlock() | ||
| }() | ||
|
|
||
| return csrl.handlerFunc() | ||
| } | ||
| if err := runHandler(); err != nil { | ||
| utilruntime.HandleError(err) | ||
| } | ||
|
|
||
| // Re-call the commit in case there is work waiting that came in while we were working | ||
| // we want to call the top level commit in case the state has not changed | ||
| glog.V(8).Infof("Re-Calling the worker after a reload in case work came in") | ||
| csrl.changeWorker(false) | ||
| } |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package limiter | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| type handler struct { | ||
| count int | ||
| sync.Mutex | ||
| } | ||
|
|
||
| func (h *handler) handle() error { | ||
| h.Lock() | ||
| defer h.Unlock() | ||
| h.count += 1 | ||
| return nil | ||
| } | ||
|
|
||
| func (h *handler) counter() int { | ||
| h.Lock() | ||
| defer h.Unlock() | ||
| return h.count | ||
| } | ||
|
|
||
| func TestCoalescingSerializingRateLimiter(t *testing.T) { | ||
|
|
||
| fmt.Println("start") | ||
|
|
||
| tests := []struct { | ||
| Name string | ||
| Interval time.Duration | ||
| Times int | ||
| }{ | ||
| { | ||
| Name: "3PO", | ||
| Interval: 3 * time.Second, | ||
| Times: 10, | ||
| }, | ||
| { | ||
| Name: "five-fer", | ||
| Interval: 5 * time.Second, | ||
| Times: 20, | ||
| }, | ||
| { | ||
| Name: "longjob", | ||
| Interval: 2 * time.Second, | ||
| Times: 20, | ||
| }, | ||
| } | ||
|
|
||
| for _, tc := range tests { | ||
| h := &handler{} | ||
| rlf := NewCoalescingSerializingRateLimiter(tc.Interval, h.handle) | ||
|
|
||
| for i := 0; i < tc.Times; i++ { | ||
| fmt.Println("start") | ||
| rlf.RegisterChange() | ||
| fmt.Println("end") | ||
| } | ||
|
|
||
| select { | ||
| case <-time.After(tc.Interval + 2*time.Second): | ||
| fmt.Println("after") | ||
|
|
||
| counter := h.counter() | ||
| if tc.Interval > 0 && counter >= tc.Times/2 { | ||
| t.Errorf("For coalesced calls, expected number of invocations to be at least half. Expected: < %v Got: %v", | ||
| tc.Times/2, counter) | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains 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
This file contains 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