2 Copyright 2014 The Kubernetes Authors.
4 Licensed under the Apache License, Version 2.0 (the "License");
5 you may not use this file except in compliance with the License.
6 You may obtain a copy of the License at
8 http://www.apache.org/licenses/LICENSE-2.0
10 Unless required by applicable law or agreed to in writing, software
11 distributed under the License is distributed on an "AS IS" BASIS,
12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 See the License for the specific language governing permissions and
14 limitations under the License.
26 "k8s.io/apimachinery/pkg/util/runtime"
29 // For any test of the style:
31 // <- time.After(timeout):
32 // t.Errorf("Timed out")
33 // The value for timeout should effectively be "forever." Obviously we don't want our tests to truly lock up forever, but 30s
34 // is long enough that it is effectively forever for the things that can slow down a run on a heavily contended machine
35 // (GC, seeks, etc), but not so long as to make a developer ctrl-c a test run if they do happen to break that test.
36 var ForeverTestTimeout = time.Second * 30
38 // NeverStop may be passed to Until to make it never stop.
39 var NeverStop <-chan struct{} = make(chan struct{})
41 // Group allows to start a group of goroutines and wait for their completion.
46 func (g *Group) Wait() {
50 // StartWithChannel starts f in a new goroutine in the group.
51 // stopCh is passed to f as an argument. f should stop when stopCh is available.
52 func (g *Group) StartWithChannel(stopCh <-chan struct{}, f func(stopCh <-chan struct{})) {
58 // StartWithContext starts f in a new goroutine in the group.
59 // ctx is passed to f as an argument. f should stop when ctx.Done() is available.
60 func (g *Group) StartWithContext(ctx context.Context, f func(context.Context)) {
66 // Start starts f in a new goroutine in the group.
67 func (g *Group) Start(f func()) {
75 // Forever calls f every period for ever.
77 // Forever is syntactic sugar on top of Until.
78 func Forever(f func(), period time.Duration) {
79 Until(f, period, NeverStop)
82 // Until loops until stop channel is closed, running f every period.
84 // Until is syntactic sugar on top of JitterUntil with zero jitter factor and
85 // with sliding = true (which means the timer for period starts after the f
87 func Until(f func(), period time.Duration, stopCh <-chan struct{}) {
88 JitterUntil(f, period, 0.0, true, stopCh)
91 // NonSlidingUntil loops until stop channel is closed, running f every
94 // NonSlidingUntil is syntactic sugar on top of JitterUntil with zero jitter
95 // factor, with sliding = false (meaning the timer for period starts at the same
96 // time as the function starts).
97 func NonSlidingUntil(f func(), period time.Duration, stopCh <-chan struct{}) {
98 JitterUntil(f, period, 0.0, false, stopCh)
101 // JitterUntil loops until stop channel is closed, running f every period.
103 // If jitterFactor is positive, the period is jittered before every run of f.
104 // If jitterFactor is not positive, the period is unchanged and not jittered.
106 // If sliding is true, the period is computed after f runs. If it is false then
107 // period includes the runtime for f.
109 // Close stopCh to stop. f may not be invoked if stop channel is already
110 // closed. Pass NeverStop to if you don't want it stop.
111 func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding bool, stopCh <-chan struct{}) {
122 jitteredPeriod := period
123 if jitterFactor > 0.0 {
124 jitteredPeriod = Jitter(period, jitterFactor)
128 t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout)
132 defer runtime.HandleCrash()
137 t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout)
140 // NOTE: b/c there is no priority selection in golang
141 // it is possible for this to race, meaning we could
142 // trigger t.C and stopCh, and t.C select falls through.
143 // In order to mitigate we re-check stopCh at the beginning
144 // of every loop to prevent extra executions of f().
154 // Jitter returns a time.Duration between duration and duration + maxFactor *
157 // This allows clients to avoid converging on periodic behavior. If maxFactor
158 // is 0.0, a suggested default value will be chosen.
159 func Jitter(duration time.Duration, maxFactor float64) time.Duration {
160 if maxFactor <= 0.0 {
163 wait := duration + time.Duration(rand.Float64()*maxFactor*float64(duration))
167 // ErrWaitTimeout is returned when the condition exited without success.
168 var ErrWaitTimeout = errors.New("timed out waiting for the condition")
170 // ConditionFunc returns true if the condition is satisfied, or an error
171 // if the loop should be aborted.
172 type ConditionFunc func() (done bool, err error)
174 // Backoff holds parameters applied to a Backoff function.
175 type Backoff struct {
176 Duration time.Duration // the base duration
177 Factor float64 // Duration is multiplied by factor each iteration
178 Jitter float64 // The amount of jitter applied each iteration
179 Steps int // Exit with error after this many steps
182 // ExponentialBackoff repeats a condition check with exponential backoff.
184 // It checks the condition up to Steps times, increasing the wait by multiplying
185 // the previous duration by Factor.
187 // If Jitter is greater than zero, a random amount of each duration is added
188 // (between duration and duration*(1+jitter)).
190 // If the condition never returns true, ErrWaitTimeout is returned. All other
191 // errors terminate immediately.
192 func ExponentialBackoff(backoff Backoff, condition ConditionFunc) error {
193 duration := backoff.Duration
194 for i := 0; i < backoff.Steps; i++ {
197 if backoff.Jitter > 0.0 {
198 adjusted = Jitter(duration, backoff.Jitter)
201 duration = time.Duration(float64(duration) * backoff.Factor)
203 if ok, err := condition(); err != nil || ok {
207 return ErrWaitTimeout
210 // Poll tries a condition func until it returns true, an error, or the timeout
213 // Poll always waits the interval before the run of 'condition'.
214 // 'condition' will always be invoked at least once.
216 // Some intervals may be missed if the condition takes too long or the time
217 // window is too short.
219 // If you want to Poll something forever, see PollInfinite.
220 func Poll(interval, timeout time.Duration, condition ConditionFunc) error {
221 return pollInternal(poller(interval, timeout), condition)
224 func pollInternal(wait WaitFunc, condition ConditionFunc) error {
225 done := make(chan struct{})
227 return WaitFor(wait, condition, done)
230 // PollImmediate tries a condition func until it returns true, an error, or the timeout
233 // PollImmediate always checks 'condition' before waiting for the interval. 'condition'
234 // will always be invoked at least once.
236 // Some intervals may be missed if the condition takes too long or the time
237 // window is too short.
239 // If you want to immediately Poll something forever, see PollImmediateInfinite.
240 func PollImmediate(interval, timeout time.Duration, condition ConditionFunc) error {
241 return pollImmediateInternal(poller(interval, timeout), condition)
244 func pollImmediateInternal(wait WaitFunc, condition ConditionFunc) error {
245 done, err := condition()
252 return pollInternal(wait, condition)
255 // PollInfinite tries a condition func until it returns true or an error
257 // PollInfinite always waits the interval before the run of 'condition'.
259 // Some intervals may be missed if the condition takes too long or the time
260 // window is too short.
261 func PollInfinite(interval time.Duration, condition ConditionFunc) error {
262 done := make(chan struct{})
264 return PollUntil(interval, condition, done)
267 // PollImmediateInfinite tries a condition func until it returns true or an error
269 // PollImmediateInfinite runs the 'condition' before waiting for the interval.
271 // Some intervals may be missed if the condition takes too long or the time
272 // window is too short.
273 func PollImmediateInfinite(interval time.Duration, condition ConditionFunc) error {
274 done, err := condition()
281 return PollInfinite(interval, condition)
284 // PollUntil tries a condition func until it returns true, an error or stopCh is
287 // PollUntil always waits interval before the first run of 'condition'.
288 // 'condition' will always be invoked at least once.
289 func PollUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error {
290 return WaitFor(poller(interval, 0), condition, stopCh)
293 // PollImmediateUntil tries a condition func until it returns true, an error or stopCh is closed.
295 // PollImmediateUntil runs the 'condition' before waiting for the interval.
296 // 'condition' will always be invoked at least once.
297 func PollImmediateUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error {
298 done, err := condition()
307 return ErrWaitTimeout
309 return PollUntil(interval, condition, stopCh)
313 // WaitFunc creates a channel that receives an item every time a test
314 // should be executed and is closed when the last test should be invoked.
315 type WaitFunc func(done <-chan struct{}) <-chan struct{}
317 // WaitFor continually checks 'fn' as driven by 'wait'.
319 // WaitFor gets a channel from 'wait()'', and then invokes 'fn' once for every value
320 // placed on the channel and once more when the channel is closed.
322 // If 'fn' returns an error the loop ends and that error is returned, and if
323 // 'fn' returns true the loop ends and nil is returned.
325 // ErrWaitTimeout will be returned if the channel is closed without fn ever
327 func WaitFor(wait WaitFunc, fn ConditionFunc, done <-chan struct{}) error {
342 return ErrWaitTimeout
345 // poller returns a WaitFunc that will send to the channel every interval until
346 // timeout has elapsed and then closes the channel.
348 // Over very short intervals you may receive no ticks before the channel is
349 // closed. A timeout of 0 is interpreted as an infinity.
351 // Output ticks are not buffered. If the channel is not ready to receive an
352 // item, the tick is skipped.
353 func poller(interval, timeout time.Duration) WaitFunc {
354 return WaitFunc(func(done <-chan struct{}) <-chan struct{} {
355 ch := make(chan struct{})
360 tick := time.NewTicker(interval)
363 var after <-chan time.Time
365 // time.After is more convenient, but it
366 // potentially leaves timers around much longer
367 // than necessary if we exit early.
368 timer := time.NewTimer(timeout)
376 // If the consumer isn't ready for this signal drop it and
377 // check the other channels.
379 case ch <- struct{}{}:
394 // resetOrReuseTimer avoids allocating a new timer if one is already in use.
395 // Not safe for multiple threads.
396 func resetOrReuseTimer(t *time.Timer, d time.Duration, sawTimeout bool) *time.Timer {
398 return time.NewTimer(d)
400 if !t.Stop() && !sawTimeout {