1 /*
2 *
3 * Copyright 2017 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #include <grpc/support/port_platform.h>
20
21 #include "src/core/ext/filters/client_channel/retry_throttle.h"
22
23 #include <limits.h>
24 #include <string.h>
25
26 #include <string>
27
28 #include <grpc/support/alloc.h>
29 #include <grpc/support/atm.h>
30 #include <grpc/support/string_util.h>
31 #include <grpc/support/sync.h>
32
33 #include "src/core/lib/avl/avl.h"
34
35 namespace grpc_core {
36 namespace internal {
37
38 //
39 // ServerRetryThrottleData
40 //
41
ServerRetryThrottleData(intptr_t max_milli_tokens,intptr_t milli_token_ratio,ServerRetryThrottleData * old_throttle_data)42 ServerRetryThrottleData::ServerRetryThrottleData(
43 intptr_t max_milli_tokens, intptr_t milli_token_ratio,
44 ServerRetryThrottleData* old_throttle_data)
45 : max_milli_tokens_(max_milli_tokens),
46 milli_token_ratio_(milli_token_ratio) {
47 intptr_t initial_milli_tokens = max_milli_tokens;
48 // If there was a pre-existing entry for this server name, initialize
49 // the token count by scaling proportionately to the old data. This
50 // ensures that if we're already throttling retries on the old scale,
51 // we will start out doing the same thing on the new one.
52 if (old_throttle_data != nullptr) {
53 double token_fraction =
54 static_cast<intptr_t>(
55 gpr_atm_acq_load(&old_throttle_data->milli_tokens_)) /
56 static_cast<double>(old_throttle_data->max_milli_tokens_);
57 initial_milli_tokens =
58 static_cast<intptr_t>(token_fraction * max_milli_tokens);
59 }
60 gpr_atm_rel_store(&milli_tokens_, static_cast<gpr_atm>(initial_milli_tokens));
61 // If there was a pre-existing entry, mark it as stale and give it a
62 // pointer to the new entry, which is its replacement.
63 if (old_throttle_data != nullptr) {
64 Ref().release(); // Ref held by pre-existing entry.
65 gpr_atm_rel_store(&old_throttle_data->replacement_,
66 reinterpret_cast<gpr_atm>(this));
67 }
68 }
69
~ServerRetryThrottleData()70 ServerRetryThrottleData::~ServerRetryThrottleData() {
71 ServerRetryThrottleData* replacement =
72 reinterpret_cast<ServerRetryThrottleData*>(
73 gpr_atm_acq_load(&replacement_));
74 if (replacement != nullptr) {
75 replacement->Unref();
76 }
77 }
78
GetReplacementThrottleDataIfNeeded(ServerRetryThrottleData ** throttle_data)79 void ServerRetryThrottleData::GetReplacementThrottleDataIfNeeded(
80 ServerRetryThrottleData** throttle_data) {
81 while (true) {
82 ServerRetryThrottleData* new_throttle_data =
83 reinterpret_cast<ServerRetryThrottleData*>(
84 gpr_atm_acq_load(&(*throttle_data)->replacement_));
85 if (new_throttle_data == nullptr) return;
86 *throttle_data = new_throttle_data;
87 }
88 }
89
RecordFailure()90 bool ServerRetryThrottleData::RecordFailure() {
91 // First, check if we are stale and need to be replaced.
92 ServerRetryThrottleData* throttle_data = this;
93 GetReplacementThrottleDataIfNeeded(&throttle_data);
94 // We decrement milli_tokens by 1000 (1 token) for each failure.
95 const intptr_t new_value =
96 static_cast<intptr_t>(gpr_atm_no_barrier_clamped_add(
97 &throttle_data->milli_tokens_, static_cast<gpr_atm>(-1000),
98 static_cast<gpr_atm>(0),
99 static_cast<gpr_atm>(throttle_data->max_milli_tokens_)));
100 // Retries are allowed as long as the new value is above the threshold
101 // (max_milli_tokens / 2).
102 return new_value > throttle_data->max_milli_tokens_ / 2;
103 }
104
RecordSuccess()105 void ServerRetryThrottleData::RecordSuccess() {
106 // First, check if we are stale and need to be replaced.
107 ServerRetryThrottleData* throttle_data = this;
108 GetReplacementThrottleDataIfNeeded(&throttle_data);
109 // We increment milli_tokens by milli_token_ratio for each success.
110 gpr_atm_no_barrier_clamped_add(
111 &throttle_data->milli_tokens_,
112 static_cast<gpr_atm>(throttle_data->milli_token_ratio_),
113 static_cast<gpr_atm>(0),
114 static_cast<gpr_atm>(throttle_data->max_milli_tokens_));
115 }
116
117 //
118 // avl vtable for string -> server_retry_throttle_data map
119 //
120
121 namespace {
122
copy_server_name(void * key,void *)123 void* copy_server_name(void* key, void* /*unused*/) {
124 return gpr_strdup(static_cast<const char*>(key));
125 }
126
compare_server_name(void * key1,void * key2,void *)127 long compare_server_name(void* key1, void* key2, void* /*unused*/) {
128 return strcmp(static_cast<const char*>(key1), static_cast<const char*>(key2));
129 }
130
destroy_server_retry_throttle_data(void * value,void *)131 void destroy_server_retry_throttle_data(void* value, void* /*unused*/) {
132 ServerRetryThrottleData* throttle_data =
133 static_cast<ServerRetryThrottleData*>(value);
134 throttle_data->Unref();
135 }
136
copy_server_retry_throttle_data(void * value,void *)137 void* copy_server_retry_throttle_data(void* value, void* /*unused*/) {
138 ServerRetryThrottleData* throttle_data =
139 static_cast<ServerRetryThrottleData*>(value);
140 return throttle_data->Ref().release();
141 }
142
destroy_server_name(void * key,void *)143 void destroy_server_name(void* key, void* /*unused*/) { gpr_free(key); }
144
145 const grpc_avl_vtable avl_vtable = {
146 destroy_server_name, copy_server_name, compare_server_name,
147 destroy_server_retry_throttle_data, copy_server_retry_throttle_data};
148
149 } // namespace
150
151 //
152 // ServerRetryThrottleMap
153 //
154
155 static gpr_mu g_mu;
156 static grpc_avl g_avl;
157
Init()158 void ServerRetryThrottleMap::Init() {
159 gpr_mu_init(&g_mu);
160 g_avl = grpc_avl_create(&avl_vtable);
161 }
162
Shutdown()163 void ServerRetryThrottleMap::Shutdown() {
164 gpr_mu_destroy(&g_mu);
165 grpc_avl_unref(g_avl, nullptr);
166 }
167
GetDataForServer(const std::string & server_name,intptr_t max_milli_tokens,intptr_t milli_token_ratio)168 RefCountedPtr<ServerRetryThrottleData> ServerRetryThrottleMap::GetDataForServer(
169 const std::string& server_name, intptr_t max_milli_tokens,
170 intptr_t milli_token_ratio) {
171 RefCountedPtr<ServerRetryThrottleData> result;
172 gpr_mu_lock(&g_mu);
173 ServerRetryThrottleData* throttle_data =
174 static_cast<ServerRetryThrottleData*>(
175 grpc_avl_get(g_avl, const_cast<char*>(server_name.c_str()), nullptr));
176 if (throttle_data == nullptr ||
177 throttle_data->max_milli_tokens() != max_milli_tokens ||
178 throttle_data->milli_token_ratio() != milli_token_ratio) {
179 // Entry not found, or found with old parameters. Create a new one.
180 result = MakeRefCounted<ServerRetryThrottleData>(
181 max_milli_tokens, milli_token_ratio, throttle_data);
182 g_avl = grpc_avl_add(g_avl, gpr_strdup(server_name.c_str()),
183 result->Ref().release(), nullptr);
184 } else {
185 // Entry found. Return a new ref to it.
186 result = throttle_data->Ref();
187 }
188 gpr_mu_unlock(&g_mu);
189 return result;
190 }
191
192 } // namespace internal
193 } // namespace grpc_core
194