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