Vihren Atlas

Atlas › Test

TestExponentialBackoff

go.temporal.io/server/common/backoff/TestRetryPolicySuite/TestExponentialBackoff

Package
go.temporal.io/server/common/backoff
Suite / test hierarchy
TestRetryPolicySuite/TestExponentialBackoff
Test
TestExponentialBackoff
Introduced at
TestMaximumInterval, TestExponentialBackoff Test frontier
Covered ranges
28
Covered lines
112
Covered files
2

Co-introduced tests

1 other test enter at the same concept.

Covered source

Expand a file to inspect source; the > gutter marks covered lines.

go.temporal.io/server/common/backoff/retrypolicy.go 89 covered LOC · 23 ranges

Open complete file

80
81 // NewExponentialRetryPolicy returns an instance of ExponentialRetryPolicy using the provided initialInterval
82 > func NewExponentialRetryPolicy(initialInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go
83 > p := &ExponentialRetryPolicy{
84 > initialInterval: initialInterval,
85 > backoffCoefficient: defaultBackoffCoefficient,
86 > maximumInterval: defaultMaximumInterval,
87 > expirationInterval: defaultExpirationInterval,
88 > maximumAttempts: defaultMaximumAttempts,
89 > }
90 >
91 > return p
92 > }
93
94 // NewRetrier is used for creating a new instance of Retrier
95 > func NewRetrier(policy RetryPolicy, timeSource clock.TimeSource) Retrier { retrypolicy.go
96 > return &retrierImpl{
97 > policy: policy,
98 > timeSource: timeSource,
99 > startTime: timeSource.Now(),
100 > currentAttempt: 1,
101 > }
102 > }
103
104 // WithInitialInterval sets the initial interval used by ExponentialRetryPolicy for the very first retry
113 // All retries are computed using the following formula:
114 // initialInterval * math.Pow(backoffCoefficient, currentAttempt)
115 > func (p *ExponentialRetryPolicy) WithBackoffCoefficient(backoffCoefficient float64) *ExponentialRetryPolicy { retrypolicy.go
116 > p.backoffCoefficient = backoffCoefficient
117 > return p
118 > }
119
120 // WithMaximumInterval sets the maximum interval for each retry.
121 // This does *not* cause the policy to stop retrying when the interval between retries reaches the supplied duration.
122 // That is what WithExpirationInterval does. Instead, this prevents the interval from exceeding maximumInterval.
123 > func (p *ExponentialRetryPolicy) WithMaximumInterval(maximumInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go
124 > p.maximumInterval = maximumInterval
125 > return p
126 > }
127
128 // WithExpirationInterval sets the absolute expiration interval for all retries
129 > func (p *ExponentialRetryPolicy) WithExpirationInterval(expirationInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go
130 > p.expirationInterval = expirationInterval
131 > return p
132 > }
133
134 // WithMaximumAttempts sets the maximum number of retry attempts
135 > func (p *ExponentialRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ExponentialRetryPolicy { retrypolicy.go
136 > p.maximumAttempts = maximumAttempts
137 > return p
138 > }
139
140 // ComputeNextDelay returns the next delay interval. This is used by Retrier to delay calling the operation again
141 > func (p *ExponentialRetryPolicy) ComputeNextDelay(elapsedTime time.Duration, numAttempts int, _ error) time.Duration { retrypolicy.go
142 > // Check to see if we ran out of maximum number of attempts
143 > // NOTE: if maxAttempts is X, return done when numAttempts == X, otherwise there will be attempt X+1
144 > if p.maximumAttempts != noMaximumAttempts && numAttempts >= p.maximumAttempts {
145 return done
146 }
147
148 // Stop retrying after expiration interval is elapsed
149 > if p.expirationInterval != NoInterval && elapsedTime > p.expirationInterval { retrypolicy.go
150 return done
151 }
152
153 > nextInterval := float64(p.initialInterval) * math.Pow(p.backoffCoefficient, float64(numAttempts-1)) retrypolicy.go
154 > // Disallow retries if initialInterval is negative or nextInterval overflows
155 > if nextInterval <= 0 {
156 return done
157 }
158 > if p.maximumInterval != NoInterval { retrypolicy.go
159 > nextInterval = math.Min(nextInterval, float64(p.maximumInterval)) retrypolicy.go
160 > }
161
162 > if p.expirationInterval != NoInterval { retrypolicy.go
163 remainingTime := float64(math.Max(0, float64(p.expirationInterval-elapsedTime)))
164 nextInterval = math.Min(remainingTime, nextInterval)
166
167 // Bail out if the next interval is smaller than initial retry interval
168 > nextDuration := time.Duration(nextInterval) retrypolicy.go
169 > if nextDuration < p.initialInterval {
170 return done
171 }
172
173 > nextInterval = p.addJitter(nextInterval) retrypolicy.go
174 >
175 > return time.Duration(nextInterval)
176 }
177
178 > func (p *ExponentialRetryPolicy) addJitter(nextInterval float64) float64 { retrypolicy.go
179 > // add jitter to avoid global synchronization
180 > jitterPortion := max(
181 > // Prevent overflow
182 > int(0.2*nextInterval), 1)
183 > nextInterval = nextInterval*0.8 + float64(getJitterRand().Intn(jitterPortion))
184 > return nextInterval
185 > }
186
187 func (r *disabledRetryPolicyImpl) ComputeNextDelay(_ time.Duration, _ int, _ error) time.Duration {
225
226 // NextBackOff returns the next delay interval. This is used by Retry to delay calling the operation again
227 > func (r *retrierImpl) NextBackOff(err error) time.Duration { retrypolicy.go
228 > nextInterval := r.policy.ComputeNextDelay(r.getElapsedTime(), r.currentAttempt, err)
229 >
230 > // Now increment the current attempt
231 > r.currentAttempt++
232 > return nextInterval
233 > }
234
235 > func (r *retrierImpl) getElapsedTime() time.Duration { retrypolicy.go
236 > return r.timeSource.Now().Sub(r.startTime)
237 > }
238
239 var _ RetryPolicy = (*ErrorDependentRetryPolicy)(nil)
297 }
298
299 > func getJitterRand() *rand.Rand { retrypolicy.go
300 > if r := jitterRand.Load(); r != nil {
301 > return r retrypolicy.go
302 > }
303 > r := rand.New(NewRetryLockedSource()) retrypolicy.go
304 >
305 > if !jitterRand.CompareAndSwap(nil, r) {
306 // Two different goroutines called some top-level
307 // function at the same time. While the results in
330 }
331
332 > func (r *RetryLockedSource) Int63() int64 { retrypolicy.go
333 > r.lk.Lock()
334 > defer r.lk.Unlock()
335 > return r.s.Int63()
336 > }
337
338 func (r *RetryLockedSource) Seed(seed int64) {
340 }
341
342 > func NewRetryLockedSource() *RetryLockedSource { retrypolicy.go
343 > return &RetryLockedSource{
344 > lk: sync.Mutex{},
345 > s: rand.NewSource(time.Now().UnixNano()),
346 > }
347 > }
go.temporal.io/server/common/clock/event_time_source.go 23 covered LOC · 5 ranges

Open complete file

39
40 // NewEventTimeSource returns a EventTimeSource with the current time set to Unix zero: 1970-01-01 00:00:00 +0000 UTC.
41 > func NewEventTimeSource() *EventTimeSource { event_time_source.go
42 > return &EventTimeSource{
43 > now: time.Unix(0, 0),
44 > }
45 > }
46
47 // Some clients depend on the fact that the runtime's timers do _not_ run synchronously.
55
56 // Now return the current time.
57 > func (ts *EventTimeSource) Now() time.Time { event_time_source.go
58 > ts.mu.RLock()
59 > defer ts.mu.RUnlock()
60 >
61 > return ts.now
62 > }
63
64 func (ts *EventTimeSource) Since(t time.Time) time.Duration {
106 // Update the fake current time. It returns the timeSource so that you can chain calls like this:
107 // timeSource := NewEventTimeSource().Update(time.Now())
108 > func (ts *EventTimeSource) Update(now time.Time) *EventTimeSource { event_time_source.go
109 > ts.mu.Lock()
110 > defer ts.mu.Unlock()
111 >
112 > ts.now = now
113 > ts.fireTimers()
114 > return ts
115 > }
116
117 // Advance the timer by the specified duration.
156
157 // fireTimers fires all timers that are ready.
158 > func (ts *EventTimeSource) fireTimers() { event_time_source.go
159 > n := 0
160 > for _, t := range ts.timers {
161 if t.deadline.After(ts.now) {
162 ts.timers[n] = t