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 "src/core/client_channel/retry_throttle.h"
20
21 #include <grpc/support/port_platform.h>
22
23 #include <atomic>
24 #include <cstdint>
25 #include <limits>
26 #include <map>
27 #include <string>
28 #include <utility>
29
30 #include "src/core/util/useful.h"
31
32 namespace grpc_core {
33 namespace internal {
34
35 namespace {
36 template <typename T>
ClampedAdd(std::atomic<T> & value,T delta,T min,T max)37 T ClampedAdd(std::atomic<T>& value, T delta, T min, T max) {
38 T prev_value = value.load(std::memory_order_relaxed);
39 T new_value;
40 do {
41 new_value = Clamp(SaturatingAdd(prev_value, delta), min, max);
42 } while (!value.compare_exchange_weak(prev_value, new_value,
43 std::memory_order_relaxed));
44 return new_value;
45 }
46 } // namespace
47
48 //
49 // ServerRetryThrottleData
50 //
51
ServerRetryThrottleData(uintptr_t max_milli_tokens,uintptr_t milli_token_ratio,ServerRetryThrottleData * old_throttle_data)52 ServerRetryThrottleData::ServerRetryThrottleData(
53 uintptr_t max_milli_tokens, uintptr_t milli_token_ratio,
54 ServerRetryThrottleData* old_throttle_data)
55 : max_milli_tokens_(max_milli_tokens),
56 milli_token_ratio_(milli_token_ratio) {
57 uintptr_t initial_milli_tokens = max_milli_tokens;
58 // If there was a pre-existing entry for this server name, initialize
59 // the token count by scaling proportionately to the old data. This
60 // ensures that if we're already throttling retries on the old scale,
61 // we will start out doing the same thing on the new one.
62 if (old_throttle_data != nullptr) {
63 double token_fraction =
64 static_cast<double>(
65 old_throttle_data->milli_tokens_.load(std::memory_order_relaxed)) /
66 static_cast<double>(old_throttle_data->max_milli_tokens_);
67 initial_milli_tokens =
68 static_cast<uintptr_t>(token_fraction * max_milli_tokens);
69 }
70 milli_tokens_.store(initial_milli_tokens, std::memory_order_relaxed);
71 // If there was a pre-existing entry, mark it as stale and give it a
72 // pointer to the new entry, which is its replacement.
73 if (old_throttle_data != nullptr) {
74 Ref().release(); // Ref held by pre-existing entry.
75 old_throttle_data->replacement_.store(this, std::memory_order_release);
76 }
77 }
78
~ServerRetryThrottleData()79 ServerRetryThrottleData::~ServerRetryThrottleData() {
80 ServerRetryThrottleData* replacement =
81 replacement_.load(std::memory_order_acquire);
82 if (replacement != nullptr) {
83 replacement->Unref();
84 }
85 }
86
GetReplacementThrottleDataIfNeeded(ServerRetryThrottleData ** throttle_data)87 void ServerRetryThrottleData::GetReplacementThrottleDataIfNeeded(
88 ServerRetryThrottleData** throttle_data) {
89 while (true) {
90 ServerRetryThrottleData* new_throttle_data =
91 (*throttle_data)->replacement_.load(std::memory_order_acquire);
92 if (new_throttle_data == nullptr) return;
93 *throttle_data = new_throttle_data;
94 }
95 }
96
RecordFailure()97 bool ServerRetryThrottleData::RecordFailure() {
98 // First, check if we are stale and need to be replaced.
99 ServerRetryThrottleData* throttle_data = this;
100 GetReplacementThrottleDataIfNeeded(&throttle_data);
101 // We decrement milli_tokens by 1000 (1 token) for each failure.
102 const uintptr_t new_value = ClampedAdd<intptr_t>(
103 throttle_data->milli_tokens_, -1000, 0,
104 std::min<uintptr_t>(throttle_data->max_milli_tokens_,
105 std::numeric_limits<intptr_t>::max()));
106 // Retries are allowed as long as the new value is above the threshold
107 // (max_milli_tokens / 2).
108 return new_value > throttle_data->max_milli_tokens_ / 2;
109 }
110
RecordSuccess()111 void ServerRetryThrottleData::RecordSuccess() {
112 // First, check if we are stale and need to be replaced.
113 ServerRetryThrottleData* throttle_data = this;
114 GetReplacementThrottleDataIfNeeded(&throttle_data);
115 // We increment milli_tokens by milli_token_ratio for each success.
116 ClampedAdd<intptr_t>(
117 throttle_data->milli_tokens_, throttle_data->milli_token_ratio_, 0,
118 std::max<intptr_t>(
119 0, std::min<uintptr_t>(throttle_data->max_milli_tokens_,
120 std::numeric_limits<intptr_t>::max())));
121 }
122
123 //
124 // ServerRetryThrottleMap
125 //
126
Get()127 ServerRetryThrottleMap* ServerRetryThrottleMap::Get() {
128 static ServerRetryThrottleMap* m = new ServerRetryThrottleMap();
129 return m;
130 }
131
GetDataForServer(const std::string & server_name,uintptr_t max_milli_tokens,uintptr_t milli_token_ratio)132 RefCountedPtr<ServerRetryThrottleData> ServerRetryThrottleMap::GetDataForServer(
133 const std::string& server_name, uintptr_t max_milli_tokens,
134 uintptr_t milli_token_ratio) {
135 MutexLock lock(&mu_);
136 auto it = map_.find(server_name);
137 ServerRetryThrottleData* throttle_data =
138 it == map_.end() ? nullptr : it->second.get();
139 if (throttle_data == nullptr ||
140 throttle_data->max_milli_tokens() != max_milli_tokens ||
141 throttle_data->milli_token_ratio() != milli_token_ratio) {
142 // Entry not found, or found with old parameters. Create a new one.
143 it = map_.emplace(server_name,
144 MakeRefCounted<ServerRetryThrottleData>(
145 max_milli_tokens, milli_token_ratio, throttle_data))
146 .first;
147 throttle_data = it->second.get();
148 }
149 return throttle_data->Ref();
150 }
151
152 } // namespace internal
153 } // namespace grpc_core
154