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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
|
/**
* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
* SPDX-License-Identifier: Apache-2.0.
*/
#include <aws/io/retry_strategy.h>
#include <aws/io/event_loop.h>
#include <aws/io/logging.h>
#include <aws/common/clock.h>
#include <aws/common/device_random.h>
#include <aws/common/mutex.h>
#include <aws/common/task_scheduler.h>
#include <inttypes.h>
struct exponential_backoff_strategy {
struct aws_retry_strategy base;
struct aws_exponential_backoff_retry_options config;
};
struct exponential_backoff_retry_token {
struct aws_retry_token base;
struct aws_atomic_var current_retry_count;
struct aws_atomic_var last_backoff;
size_t max_retries;
uint64_t backoff_scale_factor_ns;
enum aws_exponential_backoff_jitter_mode jitter_mode;
/* Let's not make this worse by constantly moving across threads if we can help it */
struct aws_event_loop *bound_loop;
uint64_t (*generate_random)(void);
struct aws_task retry_task;
struct {
struct aws_mutex mutex;
aws_retry_strategy_on_retry_token_acquired_fn *acquired_fn;
aws_retry_strategy_on_retry_ready_fn *retry_ready_fn;
void *user_data;
} thread_data;
};
static void s_exponential_retry_destroy(struct aws_retry_strategy *retry_strategy) {
if (retry_strategy) {
struct exponential_backoff_strategy *exponential_strategy = retry_strategy->impl;
struct aws_event_loop_group *el_group = exponential_strategy->config.el_group;
aws_mem_release(retry_strategy->allocator, exponential_strategy);
aws_ref_count_release(&el_group->ref_count);
}
}
static void s_exponential_retry_task(struct aws_task *task, void *arg, enum aws_task_status status) {
(void)task;
int error_code = AWS_ERROR_IO_OPERATION_CANCELLED;
if (status == AWS_TASK_STATUS_RUN_READY) {
error_code = AWS_OP_SUCCESS;
}
struct exponential_backoff_retry_token *backoff_retry_token = arg;
aws_retry_strategy_on_retry_token_acquired_fn *acquired_fn = NULL;
aws_retry_strategy_on_retry_ready_fn *retry_ready_fn = NULL;
void *user_data = NULL;
{ /***** BEGIN CRITICAL SECTION *********/
AWS_FATAL_ASSERT(
!aws_mutex_lock(&backoff_retry_token->thread_data.mutex) && "Retry token mutex acquisition failed");
acquired_fn = backoff_retry_token->thread_data.acquired_fn;
retry_ready_fn = backoff_retry_token->thread_data.retry_ready_fn;
user_data = backoff_retry_token->thread_data.user_data;
backoff_retry_token->thread_data.user_data = NULL;
backoff_retry_token->thread_data.retry_ready_fn = NULL;
backoff_retry_token->thread_data.acquired_fn = NULL;
AWS_FATAL_ASSERT(
!aws_mutex_unlock(&backoff_retry_token->thread_data.mutex) && "Retry token mutex release failed");
} /**** END CRITICAL SECTION ***********/
aws_retry_token_acquire(&backoff_retry_token->base);
if (acquired_fn) {
AWS_LOGF_DEBUG(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: Vending retry_token %p",
(void *)backoff_retry_token->base.retry_strategy,
(void *)&backoff_retry_token->base);
acquired_fn(backoff_retry_token->base.retry_strategy, error_code, &backoff_retry_token->base, user_data);
} else if (retry_ready_fn) {
AWS_LOGF_DEBUG(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: Invoking retry_ready for token %p",
(void *)backoff_retry_token->base.retry_strategy,
(void *)&backoff_retry_token->base);
retry_ready_fn(&backoff_retry_token->base, error_code, user_data);
/* it's acquired before being scheduled for retry */
aws_retry_token_release(&backoff_retry_token->base);
}
aws_retry_token_release(&backoff_retry_token->base);
}
static int s_exponential_retry_acquire_token(
struct aws_retry_strategy *retry_strategy,
const struct aws_byte_cursor *partition_id,
aws_retry_strategy_on_retry_token_acquired_fn *on_acquired,
void *user_data,
uint64_t timeout_ms) {
(void)partition_id;
/* no resource contention here so no timeouts. */
(void)timeout_ms;
struct exponential_backoff_retry_token *backoff_retry_token =
aws_mem_calloc(retry_strategy->allocator, 1, sizeof(struct exponential_backoff_retry_token));
if (!backoff_retry_token) {
return AWS_OP_ERR;
}
AWS_LOGF_DEBUG(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: Initializing retry token %p",
(void *)retry_strategy,
(void *)&backoff_retry_token->base);
backoff_retry_token->base.allocator = retry_strategy->allocator;
backoff_retry_token->base.retry_strategy = retry_strategy;
aws_atomic_init_int(&backoff_retry_token->base.ref_count, 1u);
aws_retry_strategy_acquire(retry_strategy);
backoff_retry_token->base.impl = backoff_retry_token;
struct exponential_backoff_strategy *exponential_backoff_strategy = retry_strategy->impl;
backoff_retry_token->bound_loop = aws_event_loop_group_get_next_loop(exponential_backoff_strategy->config.el_group);
backoff_retry_token->max_retries = exponential_backoff_strategy->config.max_retries;
backoff_retry_token->backoff_scale_factor_ns = aws_timestamp_convert(
exponential_backoff_strategy->config.backoff_scale_factor_ms, AWS_TIMESTAMP_MILLIS, AWS_TIMESTAMP_NANOS, NULL);
backoff_retry_token->jitter_mode = exponential_backoff_strategy->config.jitter_mode;
backoff_retry_token->generate_random = exponential_backoff_strategy->config.generate_random;
aws_atomic_init_int(&backoff_retry_token->current_retry_count, 0);
aws_atomic_init_int(&backoff_retry_token->last_backoff, 0);
backoff_retry_token->thread_data.acquired_fn = on_acquired;
backoff_retry_token->thread_data.user_data = user_data;
AWS_FATAL_ASSERT(
!aws_mutex_init(&backoff_retry_token->thread_data.mutex) && "Retry strategy mutex initialization failed");
aws_task_init(
&backoff_retry_token->retry_task,
s_exponential_retry_task,
backoff_retry_token,
"aws_exponential_backoff_retry_task");
aws_event_loop_schedule_task_now(backoff_retry_token->bound_loop, &backoff_retry_token->retry_task);
return AWS_OP_SUCCESS;
}
static inline uint64_t s_random_in_range(uint64_t from, uint64_t to, struct exponential_backoff_retry_token *token) {
uint64_t max = aws_max_u64(from, to);
uint64_t min = aws_min_u64(from, to);
uint64_t diff = max - min;
if (!diff) {
return 0;
}
uint64_t random = token->generate_random();
return min + random % (diff);
}
typedef uint64_t(compute_backoff_fn)(struct exponential_backoff_retry_token *token);
static uint64_t s_compute_no_jitter(struct exponential_backoff_retry_token *token) {
uint64_t retry_count = aws_min_u64(aws_atomic_load_int(&token->current_retry_count), 63);
return aws_mul_u64_saturating((uint64_t)1 << retry_count, token->backoff_scale_factor_ns);
}
static uint64_t s_compute_full_jitter(struct exponential_backoff_retry_token *token) {
uint64_t non_jittered = s_compute_no_jitter(token);
return s_random_in_range(0, non_jittered, token);
}
static uint64_t s_compute_deccorelated_jitter(struct exponential_backoff_retry_token *token) {
size_t last_backoff_val = aws_atomic_load_int(&token->last_backoff);
if (!last_backoff_val) {
return s_compute_full_jitter(token);
}
return s_random_in_range(token->backoff_scale_factor_ns, aws_mul_u64_saturating(last_backoff_val, 3), token);
}
static compute_backoff_fn *s_backoff_compute_table[] = {
[AWS_EXPONENTIAL_BACKOFF_JITTER_DEFAULT] = s_compute_full_jitter,
[AWS_EXPONENTIAL_BACKOFF_JITTER_NONE] = s_compute_no_jitter,
[AWS_EXPONENTIAL_BACKOFF_JITTER_FULL] = s_compute_full_jitter,
[AWS_EXPONENTIAL_BACKOFF_JITTER_DECORRELATED] = s_compute_deccorelated_jitter,
};
static int s_exponential_retry_schedule_retry(
struct aws_retry_token *token,
enum aws_retry_error_type error_type,
aws_retry_strategy_on_retry_ready_fn *retry_ready,
void *user_data) {
struct exponential_backoff_retry_token *backoff_retry_token = token->impl;
AWS_LOGF_DEBUG(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: Attempting retry on token %p with error type %d",
(void *)backoff_retry_token->base.retry_strategy,
(void *)token,
error_type);
uint64_t schedule_at = 0;
/* AWS_RETRY_ERROR_TYPE_CLIENT_ERROR does not count against your retry budget since you were responding to an
* improperly crafted request. */
if (error_type != AWS_RETRY_ERROR_TYPE_CLIENT_ERROR) {
size_t retry_count = aws_atomic_load_int(&backoff_retry_token->current_retry_count);
if (retry_count >= backoff_retry_token->max_retries) {
AWS_LOGF_WARN(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: token %p has exhausted allowed retries. Retry count %zu max retries %zu",
(void *)backoff_retry_token->base.retry_strategy,
(void *)token,
backoff_retry_token->max_retries,
retry_count);
return aws_raise_error(AWS_IO_MAX_RETRIES_EXCEEDED);
}
uint64_t backoff = s_backoff_compute_table[backoff_retry_token->jitter_mode](backoff_retry_token);
uint64_t current_time = 0;
aws_event_loop_current_clock_time(backoff_retry_token->bound_loop, ¤t_time);
schedule_at = backoff + current_time;
aws_atomic_init_int(&backoff_retry_token->last_backoff, (size_t)backoff);
aws_atomic_fetch_add(&backoff_retry_token->current_retry_count, 1u);
AWS_LOGF_DEBUG(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: Computed backoff value of %" PRIu64 "ns on token %p",
(void *)backoff_retry_token->base.retry_strategy,
backoff,
(void *)token);
}
bool already_scheduled = false;
{ /***** BEGIN CRITICAL SECTION *********/
AWS_FATAL_ASSERT(
!aws_mutex_lock(&backoff_retry_token->thread_data.mutex) && "Retry token mutex acquisition failed");
if (backoff_retry_token->thread_data.user_data) {
already_scheduled = true;
} else {
backoff_retry_token->thread_data.retry_ready_fn = retry_ready;
backoff_retry_token->thread_data.user_data = user_data;
/* acquire to hold until the task runs. */
aws_retry_token_acquire(token);
aws_task_init(
&backoff_retry_token->retry_task,
s_exponential_retry_task,
backoff_retry_token,
"aws_exponential_backoff_retry_task");
}
AWS_FATAL_ASSERT(
!aws_mutex_unlock(&backoff_retry_token->thread_data.mutex) && "Retry token mutex release failed");
} /**** END CRITICAL SECTION ***********/
if (already_scheduled) {
AWS_LOGF_ERROR(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: retry token %p is already scheduled.",
(void *)backoff_retry_token->base.retry_strategy,
(void *)token)
return aws_raise_error(AWS_ERROR_INVALID_STATE);
}
aws_event_loop_schedule_task_future(backoff_retry_token->bound_loop, &backoff_retry_token->retry_task, schedule_at);
return AWS_OP_SUCCESS;
}
static int s_exponential_backoff_record_success(struct aws_retry_token *token) {
/* we don't do book keeping in this mode. */
(void)token;
return AWS_OP_SUCCESS;
}
static void s_exponential_backoff_release_token(struct aws_retry_token *token) {
if (token) {
aws_retry_strategy_release(token->retry_strategy);
struct exponential_backoff_retry_token *backoff_retry_token = token->impl;
aws_mutex_clean_up(&backoff_retry_token->thread_data.mutex);
aws_mem_release(token->allocator, backoff_retry_token);
}
}
static struct aws_retry_strategy_vtable s_exponential_retry_vtable = {
.destroy = s_exponential_retry_destroy,
.acquire_token = s_exponential_retry_acquire_token,
.schedule_retry = s_exponential_retry_schedule_retry,
.record_success = s_exponential_backoff_record_success,
.release_token = s_exponential_backoff_release_token,
};
static uint64_t s_default_gen_rand(void) {
uint64_t res = 0;
aws_device_random_u64(&res);
return res;
}
struct aws_retry_strategy *aws_retry_strategy_new_exponential_backoff(
struct aws_allocator *allocator,
const struct aws_exponential_backoff_retry_options *config) {
AWS_PRECONDITION(config);
AWS_PRECONDITION(config->el_group)
AWS_PRECONDITION(config->jitter_mode <= AWS_EXPONENTIAL_BACKOFF_JITTER_DECORRELATED)
AWS_PRECONDITION(config->max_retries)
if (config->max_retries > 63 || !config->el_group ||
config->jitter_mode > AWS_EXPONENTIAL_BACKOFF_JITTER_DECORRELATED) {
aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
return NULL;
}
struct exponential_backoff_strategy *exponential_backoff_strategy =
aws_mem_calloc(allocator, 1, sizeof(struct exponential_backoff_strategy));
if (!exponential_backoff_strategy) {
return NULL;
}
AWS_LOGF_INFO(
AWS_LS_IO_EXPONENTIAL_BACKOFF_RETRY_STRATEGY,
"id=%p: Initializing exponential backoff retry strategy with scale factor: %" PRIu32
" jitter mode: %d and max retries %zu",
(void *)&exponential_backoff_strategy->base,
config->backoff_scale_factor_ms,
config->jitter_mode,
config->max_retries);
exponential_backoff_strategy->base.allocator = allocator;
exponential_backoff_strategy->base.impl = exponential_backoff_strategy;
exponential_backoff_strategy->base.vtable = &s_exponential_retry_vtable;
aws_atomic_init_int(&exponential_backoff_strategy->base.ref_count, 1);
exponential_backoff_strategy->config = *config;
exponential_backoff_strategy->config.el_group =
aws_ref_count_acquire(&exponential_backoff_strategy->config.el_group->ref_count);
if (!exponential_backoff_strategy->config.generate_random) {
exponential_backoff_strategy->config.generate_random = s_default_gen_rand;
}
if (!exponential_backoff_strategy->config.max_retries) {
exponential_backoff_strategy->config.max_retries = 5;
}
if (!exponential_backoff_strategy->config.backoff_scale_factor_ms) {
exponential_backoff_strategy->config.backoff_scale_factor_ms = 25;
}
return &exponential_backoff_strategy->base;
}
|