// Copyright 2019 Yunion // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package wait import ( "errors" "math/rand" "time" "yunion.io/x/pkg/util/runtime" ) // For any test of the style: // ... // <- time.After(timeout): // t.Errorf("Timed out") // The value for timeout should effectively be "forever." Obviously we don't want our tests to truly lock up forever, but 30s // is long enough that it is effectively forever for the things that can slow down a run on a heavily contended machine // (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. var ForeverTestTimeout = time.Second * 30 // NeverStop may be passed to Until to make it never stop. var NeverStop <-chan struct{} = make(chan struct{}) // Forever calls f every period for ever. // // Forever is syntactic sugar on top of Until. func Forever(f func(), period time.Duration) { Until(f, period, NeverStop) } // Until loops until stop channel is closed, running f every period. // // Until is syntactic sugar on top of JitterUntil with zero jitter factor and // with sliding = true (which means the timer for period starts after the f // completes). func Until(f func(), period time.Duration, stopCh <-chan struct{}) { JitterUntil(f, period, 0.0, true, stopCh) } // NonSlidingUntil loops until stop channel is closed, running f every // period. // // NonSlidingUntil is syntactic sugar on top of JitterUntil with zero jitter // factor, with sliding = false (meaning the timer for period starts at the same // time as the function starts). func NonSlidingUntil(f func(), period time.Duration, stopCh <-chan struct{}) { JitterUntil(f, period, 0.0, false, stopCh) } // JitterUntil loops until stop channel is closed, running f every period. // // If jitterFactor is positive, the period is jittered before every run of f. // If jitterFactor is not positive, the period is unchanged and not jitterd. // // If sliding is true, the period is computed after f runs. If it is false then // period includes the runtime for f. // // Close stopCh to stop. f may not be invoked if stop channel is already // closed. Pass NeverStop to if you don't want it stop. func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding bool, stopCh <-chan struct{}) { var t *time.Timer var sawTimeout bool for { select { case <-stopCh: return default: } jitteredPeriod := period if jitterFactor > 0.0 { jitteredPeriod = Jitter(period, jitterFactor) } if !sliding { t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout) } func() { defer runtime.HandleCrash() f() }() if sliding { t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout) } // NOTE: b/c there is no priority selection in golang // it is possible for this to race, meaning we could // trigger t.C and stopCh, and t.C select falls through. // In order to mitigate we re-check stopCh at the beginning // of every loop to prevent extra executions of f(). select { case <-stopCh: return case <-t.C: sawTimeout = true } } } // Jitter returns a time.Duration between duration and duration + maxFactor * // duration. // // This allows clients to avoid converging on periodic behavior. If maxFactor // is 0.0, a suggested default value will be chosen. func Jitter(duration time.Duration, maxFactor float64) time.Duration { if maxFactor <= 0.0 { maxFactor = 1.0 } wait := duration + time.Duration(rand.Float64()*maxFactor*float64(duration)) return wait } // ErrWaitTimeout is returned when the condition exited without success. var ErrWaitTimeout = errors.New("timed out waiting for the condition") // ConditionFunc returns true if the condition is satisfied, or an error // if the loop should be aborted. type ConditionFunc func() (done bool, err error) // Backoff holds parameters applied to a Backoff function. type Backoff struct { Duration time.Duration // the base duration Factor float64 // Duration is multipled by factor each iteration Jitter float64 // The amount of jitter applied each iteration Steps int // Exit with error after this many steps } // ExponentialBackoff repeats a condition check with exponential backoff. // // It checks the condition up to Steps times, increasing the wait by multipling // the previous duration by Factor. // // If Jitter is greater than zero, a random amount of each duration is added // (between duration and duration*(1+jitter)). // // If the condition never returns true, ErrWaitTimeout is returned. All other // errors terminate immediately. func ExponentialBackoff(backoff Backoff, condition ConditionFunc) error { duration := backoff.Duration for i := 0; i < backoff.Steps; i++ { if i != 0 { adjusted := duration if backoff.Jitter > 0.0 { adjusted = Jitter(duration, backoff.Jitter) } time.Sleep(adjusted) duration = time.Duration(float64(duration) * backoff.Factor) } if ok, err := condition(); err != nil || ok { return err } } return ErrWaitTimeout } // Poll tries a condition func until it returns true, an error, or the timeout // is reached. // // Poll always waits the interval before the run of 'condition'. // 'condition' will always be invoked at least once. // // Some intervals may be missed if the condition takes too long or the time // window is too short. // // If you want to Poll something forever, see PollInfinite. func Poll(interval, timeout time.Duration, condition ConditionFunc) error { return pollInternal(poller(interval, timeout), condition) } func pollInternal(wait WaitFunc, condition ConditionFunc) error { return WaitFor(wait, condition, NeverStop) } // PollImmediate tries a condition func until it returns true, an error, or the timeout // is reached. // // Poll always checks 'condition' before waiting for the interval. 'condition' // will always be invoked at least once. // // Some intervals may be missed if the condition takes too long or the time // window is too short. // // If you want to Poll something forever, see PollInfinite. func PollImmediate(interval, timeout time.Duration, condition ConditionFunc) error { return pollImmediateInternal(poller(interval, timeout), condition) } func pollImmediateInternal(wait WaitFunc, condition ConditionFunc) error { done, err := condition() if err != nil { return err } if done { return nil } return pollInternal(wait, condition) } // PollInfinite tries a condition func until it returns true or an error // // PollInfinite always waits the interval before the run of 'condition'. // // Some intervals may be missed if the condition takes too long or the time // window is too short. func PollInfinite(interval time.Duration, condition ConditionFunc) error { done := make(chan struct{}) defer close(done) return PollUntil(interval, condition, done) } // PollImmediateInfinite tries a condition func until it returns true or an error // // PollImmediateInfinite runs the 'condition' before waiting for the interval. // // Some intervals may be missed if the condition takes too long or the time // window is too short. func PollImmediateInfinite(interval time.Duration, condition ConditionFunc) error { done, err := condition() if err != nil { return err } if done { return nil } return PollInfinite(interval, condition) } // PollUntil tries a condition func until it returns true, an error or stopCh is // closed. // // PolUntil always waits interval before the first run of 'condition'. // 'condition' will always be invoked at least once. func PollUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error { return WaitFor(poller(interval, 0), condition, stopCh) } // WaitFunc creates a channel that receives an item every time a test // should be executed and is closed when the last test should be invoked. type WaitFunc func(done <-chan struct{}) <-chan struct{} // WaitFor continually checks 'fn' as driven by 'wait'. // // WaitFor gets a channel from 'wait()'', and then invokes 'fn' once for every value // placed on the channel and once more when the channel is closed. // // If 'fn' returns an error the loop ends and that error is returned, and if // 'fn' returns true the loop ends and nil is returned. // // ErrWaitTimeout will be returned if the channel is closed without fn ever // returning true. func WaitFor(wait WaitFunc, fn ConditionFunc, done <-chan struct{}) error { c := wait(done) for { _, open := <-c ok, err := fn() if err != nil { return err } if ok { return nil } if !open { break } } return ErrWaitTimeout } // poller returns a WaitFunc that will send to the channel every interval until // timeout has elapsed and then closes the channel. // // Over very short intervals you may receive no ticks before the channel is // closed. A timeout of 0 is interpreted as an infinity. // // Output ticks are not buffered. If the channel is not ready to receive an // item, the tick is skipped. func poller(interval, timeout time.Duration) WaitFunc { return WaitFunc(func(done <-chan struct{}) <-chan struct{} { ch := make(chan struct{}) go func() { defer close(ch) tick := time.NewTicker(interval) defer tick.Stop() var after <-chan time.Time if timeout != 0 { // time.After is more convenient, but it // potentially leaves timers around much longer // than necessary if we exit early. timer := time.NewTimer(timeout) after = timer.C defer timer.Stop() } for { select { case <-tick.C: // If the consumer isn't ready for this signal drop it and // check the other channels. select { case ch <- struct{}{}: default: } case <-after: return case <-done: return } } }() return ch }) } // resetOrReuseTimer avoids allocating a new timer if one is already in use. // Not safe for multiple threads. func resetOrReuseTimer(t *time.Timer, d time.Duration, sawTimeout bool) *time.Timer { if t == nil { return time.NewTimer(d) } if !t.Stop() && !sawTimeout { <-t.C } t.Reset(d) return t }