• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 <grpc/support/alloc.h>
27 #include <grpc/support/atm.h>
28 #include <grpc/support/string_util.h>
29 #include <grpc/support/sync.h>
30 
31 #include "src/core/lib/avl/avl.h"
32 
33 namespace grpc_core {
34 namespace internal {
35 
36 //
37 // ServerRetryThrottleData
38 //
39 
ServerRetryThrottleData(intptr_t max_milli_tokens,intptr_t milli_token_ratio,ServerRetryThrottleData * old_throttle_data)40 ServerRetryThrottleData::ServerRetryThrottleData(
41     intptr_t max_milli_tokens, intptr_t milli_token_ratio,
42     ServerRetryThrottleData* old_throttle_data)
43     : max_milli_tokens_(max_milli_tokens),
44       milli_token_ratio_(milli_token_ratio) {
45   intptr_t initial_milli_tokens = max_milli_tokens;
46   // If there was a pre-existing entry for this server name, initialize
47   // the token count by scaling proportionately to the old data.  This
48   // ensures that if we're already throttling retries on the old scale,
49   // we will start out doing the same thing on the new one.
50   if (old_throttle_data != nullptr) {
51     double token_fraction =
52         static_cast<intptr_t>(
53             gpr_atm_acq_load(&old_throttle_data->milli_tokens_)) /
54         static_cast<double>(old_throttle_data->max_milli_tokens_);
55     initial_milli_tokens =
56         static_cast<intptr_t>(token_fraction * max_milli_tokens);
57   }
58   gpr_atm_rel_store(&milli_tokens_, static_cast<gpr_atm>(initial_milli_tokens));
59   // If there was a pre-existing entry, mark it as stale and give it a
60   // pointer to the new entry, which is its replacement.
61   if (old_throttle_data != nullptr) {
62     Ref().release();  // Ref held by pre-existing entry.
63     gpr_atm_rel_store(&old_throttle_data->replacement_,
64                       reinterpret_cast<gpr_atm>(this));
65   }
66 }
67 
~ServerRetryThrottleData()68 ServerRetryThrottleData::~ServerRetryThrottleData() {
69   ServerRetryThrottleData* replacement =
70       reinterpret_cast<ServerRetryThrottleData*>(
71           gpr_atm_acq_load(&replacement_));
72   if (replacement != nullptr) {
73     replacement->Unref();
74   }
75 }
76 
GetReplacementThrottleDataIfNeeded(ServerRetryThrottleData ** throttle_data)77 void ServerRetryThrottleData::GetReplacementThrottleDataIfNeeded(
78     ServerRetryThrottleData** throttle_data) {
79   while (true) {
80     ServerRetryThrottleData* new_throttle_data =
81         reinterpret_cast<ServerRetryThrottleData*>(
82             gpr_atm_acq_load(&(*throttle_data)->replacement_));
83     if (new_throttle_data == nullptr) return;
84     *throttle_data = new_throttle_data;
85   }
86 }
87 
RecordFailure()88 bool ServerRetryThrottleData::RecordFailure() {
89   // First, check if we are stale and need to be replaced.
90   ServerRetryThrottleData* throttle_data = this;
91   GetReplacementThrottleDataIfNeeded(&throttle_data);
92   // We decrement milli_tokens by 1000 (1 token) for each failure.
93   const intptr_t new_value =
94       static_cast<intptr_t>(gpr_atm_no_barrier_clamped_add(
95           &throttle_data->milli_tokens_, static_cast<gpr_atm>(-1000),
96           static_cast<gpr_atm>(0),
97           static_cast<gpr_atm>(throttle_data->max_milli_tokens_)));
98   // Retries are allowed as long as the new value is above the threshold
99   // (max_milli_tokens / 2).
100   return new_value > throttle_data->max_milli_tokens_ / 2;
101 }
102 
RecordSuccess()103 void ServerRetryThrottleData::RecordSuccess() {
104   // First, check if we are stale and need to be replaced.
105   ServerRetryThrottleData* throttle_data = this;
106   GetReplacementThrottleDataIfNeeded(&throttle_data);
107   // We increment milli_tokens by milli_token_ratio for each success.
108   gpr_atm_no_barrier_clamped_add(
109       &throttle_data->milli_tokens_,
110       static_cast<gpr_atm>(throttle_data->milli_token_ratio_),
111       static_cast<gpr_atm>(0),
112       static_cast<gpr_atm>(throttle_data->max_milli_tokens_));
113 }
114 
115 //
116 // avl vtable for string -> server_retry_throttle_data map
117 //
118 
119 namespace {
120 
copy_server_name(void * key,void * unused)121 void* copy_server_name(void* key, void* unused) {
122   return gpr_strdup(static_cast<const char*>(key));
123 }
124 
compare_server_name(void * key1,void * key2,void * unused)125 long compare_server_name(void* key1, void* key2, void* unused) {
126   return strcmp(static_cast<const char*>(key1), static_cast<const char*>(key2));
127 }
128 
destroy_server_retry_throttle_data(void * value,void * unused)129 void destroy_server_retry_throttle_data(void* value, void* unused) {
130   ServerRetryThrottleData* throttle_data =
131       static_cast<ServerRetryThrottleData*>(value);
132   throttle_data->Unref();
133 }
134 
copy_server_retry_throttle_data(void * value,void * unused)135 void* copy_server_retry_throttle_data(void* value, void* unused) {
136   ServerRetryThrottleData* throttle_data =
137       static_cast<ServerRetryThrottleData*>(value);
138   return throttle_data->Ref().release();
139 }
140 
destroy_server_name(void * key,void * unused)141 void destroy_server_name(void* key, void* unused) { gpr_free(key); }
142 
143 const grpc_avl_vtable avl_vtable = {
144     destroy_server_name, copy_server_name, compare_server_name,
145     destroy_server_retry_throttle_data, copy_server_retry_throttle_data};
146 
147 }  // namespace
148 
149 //
150 // ServerRetryThrottleMap
151 //
152 
153 static gpr_mu g_mu;
154 static grpc_avl g_avl;
155 
Init()156 void ServerRetryThrottleMap::Init() {
157   gpr_mu_init(&g_mu);
158   g_avl = grpc_avl_create(&avl_vtable);
159 }
160 
Shutdown()161 void ServerRetryThrottleMap::Shutdown() {
162   gpr_mu_destroy(&g_mu);
163   grpc_avl_unref(g_avl, nullptr);
164 }
165 
GetDataForServer(const char * server_name,intptr_t max_milli_tokens,intptr_t milli_token_ratio)166 RefCountedPtr<ServerRetryThrottleData> ServerRetryThrottleMap::GetDataForServer(
167     const char* server_name, intptr_t max_milli_tokens,
168     intptr_t milli_token_ratio) {
169   RefCountedPtr<ServerRetryThrottleData> result;
170   gpr_mu_lock(&g_mu);
171   ServerRetryThrottleData* throttle_data =
172       static_cast<ServerRetryThrottleData*>(
173           grpc_avl_get(g_avl, const_cast<char*>(server_name), nullptr));
174   if (throttle_data == nullptr ||
175       throttle_data->max_milli_tokens() != max_milli_tokens ||
176       throttle_data->milli_token_ratio() != milli_token_ratio) {
177     // Entry not found, or found with old parameters.  Create a new one.
178     result = MakeRefCounted<ServerRetryThrottleData>(
179         max_milli_tokens, milli_token_ratio, throttle_data);
180     g_avl = grpc_avl_add(g_avl, gpr_strdup(server_name),
181                          result->Ref().release(), nullptr);
182   } else {
183     // Entry found.  Return a new ref to it.
184     result = throttle_data->Ref();
185   }
186   gpr_mu_unlock(&g_mu);
187   return result;
188 }
189 
190 }  // namespace internal
191 }  // namespace grpc_core
192