forked from smallrye/smallrye-mutiny
-
Notifications
You must be signed in to change notification settings - Fork 0
/
ExponentialBackoff.java
148 lines (128 loc) · 6.11 KB
/
ExponentialBackoff.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
package io.smallrye.mutiny.helpers;
import java.time.Duration;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
public class ExponentialBackoff {
public static final Duration MAX_BACKOFF = Duration.ofMillis(Long.MAX_VALUE);
public static final double DEFAULT_JITTER = 0.5;
private ExponentialBackoff() {
// avoid direct instantiation
}
/**
* Computes a method that would delay <em>ticks</em> using an exponential backoff.
*
* @param numRetries the max number of retries
* @param firstBackoff the delay of the first backoff
* @param maxBackoff the max backoff
* @param jitterFactor the jitter factor in [0, 1]
* @param executor the executor used for the delay
* @return the function
*/
public static Function<Multi<Throwable>, Publisher<Long>> randomExponentialBackoffFunction(
long numRetries, Duration firstBackoff, Duration maxBackoff,
double jitterFactor, ScheduledExecutorService executor) {
validate(firstBackoff, maxBackoff, jitterFactor, executor);
AtomicInteger index = new AtomicInteger();
return t -> t
.onItem().transformToUni(failure -> {
int iteration = index.getAndIncrement();
if (iteration >= numRetries) {
return Uni.createFrom().failure(
new IllegalStateException("Retries exhausted: " + iteration + "/" + numRetries,
failure));
}
Duration delay = getNextDelay(firstBackoff, maxBackoff, jitterFactor, iteration);
return Uni.createFrom().item((long) iteration).onItem().delayIt()
.onExecutor(executor).by(delay);
}).concatenate();
}
private static Duration getNextDelay(Duration firstBackoff, Duration maxBackoff, double jitterFactor, int iteration) {
Duration nextBackoff = getNextAttemptDelay(firstBackoff, maxBackoff, iteration);
// Compute the jitter
long jitterOffset = getJitter(jitterFactor, nextBackoff);
long lowBound = Math.max(firstBackoff.minus(nextBackoff).toMillis(), -jitterOffset);
long highBound = Math.min(maxBackoff.minus(nextBackoff).toMillis(), jitterOffset);
long jitter;
ThreadLocalRandom random = ThreadLocalRandom.current();
if (highBound == lowBound) {
if (highBound == 0) {
jitter = 0;
} else {
jitter = random.nextLong(highBound);
}
} else {
jitter = random.nextLong(lowBound, highBound);
}
return nextBackoff.plusMillis(jitter);
}
private static void validate(Duration firstBackoff, Duration maxBackoff, double jitterFactor,
ScheduledExecutorService executor) {
if (jitterFactor < 0 || jitterFactor > 1) {
throw new IllegalArgumentException("jitterFactor must be between 0 and 1 (default 0.5)");
}
ParameterValidation.nonNull(firstBackoff, "firstBackoff");
ParameterValidation.nonNull(maxBackoff, "maxBackoff");
ParameterValidation.nonNull(executor, "executor");
}
/**
* Computes a method that would delay <em>ticks</em> using an exponential backoff.
* Will keep retrying until an expiration time.
* The last attempt will start before the expiration time.
*
* @param expireAt absolute time in millis that specifies when to give up.
* @param firstBackoff the delay of the first backoff
* @param maxBackoff the max backoff
* @param jitterFactor the jitter factor in [0, 1]
* @param executor the executor used for the delay
* @return the function
*/
public static Function<Multi<Throwable>, Publisher<Long>> randomExponentialBackoffFunctionExpireAt(
long expireAt, Duration firstBackoff, Duration maxBackoff,
double jitterFactor, ScheduledExecutorService executor) {
validate(firstBackoff, maxBackoff, jitterFactor, executor);
AtomicInteger index = new AtomicInteger();
return t -> t
.onItem().transformToUni(failure -> {
int iteration = index.incrementAndGet();
Duration delay = getNextDelay(firstBackoff, maxBackoff, jitterFactor, iteration);
long checkTime = System.currentTimeMillis() + delay.toMillis();
if (checkTime > expireAt) {
return Uni.createFrom().failure(
new IllegalStateException(
"Retries exhausted : " + iteration + " attempts against " + checkTime + "/" + expireAt
+ " expiration",
failure));
}
return Uni.createFrom().item((long) iteration).onItem().delayIt()
.onExecutor(executor).by(delay);
}).concatenate();
}
private static long getJitter(double jitterFactor, Duration nextBackoff) {
long jitterOffset;
try {
jitterOffset = nextBackoff.multipliedBy((long) (100 * jitterFactor))
.dividedBy(100)
.toMillis();
} catch (ArithmeticException ae) {
jitterOffset = Math.round(Long.MAX_VALUE * jitterFactor);
}
return jitterOffset;
}
private static Duration getNextAttemptDelay(Duration firstBackoff, Duration maxBackoff, int iteration) {
Duration nextBackoff;
try {
nextBackoff = firstBackoff.multipliedBy((long) Math.pow(2, iteration));
if (nextBackoff.compareTo(maxBackoff) > 0) {
nextBackoff = maxBackoff;
}
} catch (ArithmeticException overflow) {
nextBackoff = maxBackoff;
}
return nextBackoff;
}
}