diff options
author | 2018-03-28 07:42:20 -0700 | |
---|---|---|
committer | 2018-03-28 07:42:20 -0700 | |
commit | 9db86fcc6366a472abd50699719e4631c2df5393 (patch) | |
tree | 5efe7b7e1977b116f911197e10b4468d22b2400e /src/core/ext/filters/client_channel/retry_throttle.cc | |
parent | 31bdbbeacf84d3a1e867e50b312c9cf46afc20db (diff) |
Convert retry throttle code to C++ and add tests.
Diffstat (limited to 'src/core/ext/filters/client_channel/retry_throttle.cc')
-rw-r--r-- | src/core/ext/filters/client_channel/retry_throttle.cc | 222 |
1 files changed, 100 insertions, 122 deletions
diff --git a/src/core/ext/filters/client_channel/retry_throttle.cc b/src/core/ext/filters/client_channel/retry_throttle.cc index 45de6667c8..bdeb7e4cac 100644 --- a/src/core/ext/filters/client_channel/retry_throttle.cc +++ b/src/core/ext/filters/client_channel/retry_throttle.cc @@ -30,184 +30,162 @@ #include "src/core/lib/avl/avl.h" +namespace grpc_core { +namespace internal { + // -// server_retry_throttle_data +// ServerRetryThrottleData // -struct grpc_server_retry_throttle_data { - gpr_refcount refs; - int max_milli_tokens; - int milli_token_ratio; - gpr_atm milli_tokens; - // A pointer to the replacement for this grpc_server_retry_throttle_data - // entry. If non-nullptr, then this entry is stale and must not be used. - // We hold a reference to the replacement. - gpr_atm replacement; -}; - -static void get_replacement_throttle_data_if_needed( - grpc_server_retry_throttle_data** throttle_data) { +ServerRetryThrottleData::ServerRetryThrottleData( + intptr_t max_milli_tokens, intptr_t milli_token_ratio, + ServerRetryThrottleData* old_throttle_data) + : max_milli_tokens_(max_milli_tokens), + milli_token_ratio_(milli_token_ratio) { + intptr_t initial_milli_tokens = max_milli_tokens; + // If there was a pre-existing entry for this server name, initialize + // the token count by scaling proportionately to the old data. This + // ensures that if we're already throttling retries on the old scale, + // we will start out doing the same thing on the new one. + if (old_throttle_data != nullptr) { + double token_fraction = + static_cast<intptr_t>( + gpr_atm_acq_load(&old_throttle_data->milli_tokens_)) / + static_cast<double>(old_throttle_data->max_milli_tokens_); + initial_milli_tokens = + static_cast<intptr_t>(token_fraction * max_milli_tokens); + } + gpr_atm_rel_store(&milli_tokens_, static_cast<gpr_atm>(initial_milli_tokens)); + // If there was a pre-existing entry, mark it as stale and give it a + // pointer to the new entry, which is its replacement. + if (old_throttle_data != nullptr) { + Ref().release(); // Ref held by pre-existing entry. + gpr_atm_rel_store(&old_throttle_data->replacement_, + reinterpret_cast<gpr_atm>(this)); + } +} + +ServerRetryThrottleData::~ServerRetryThrottleData() { + ServerRetryThrottleData* replacement = + reinterpret_cast<ServerRetryThrottleData*>( + gpr_atm_acq_load(&replacement_)); + if (replacement != nullptr) { + replacement->Unref(); + } +} + +void ServerRetryThrottleData::GetReplacementThrottleDataIfNeeded( + ServerRetryThrottleData** throttle_data) { while (true) { - grpc_server_retry_throttle_data* new_throttle_data = - (grpc_server_retry_throttle_data*)gpr_atm_acq_load( - &(*throttle_data)->replacement); + ServerRetryThrottleData* new_throttle_data = + reinterpret_cast<ServerRetryThrottleData*>( + gpr_atm_acq_load(&(*throttle_data)->replacement_)); if (new_throttle_data == nullptr) return; *throttle_data = new_throttle_data; } } -bool grpc_server_retry_throttle_data_record_failure( - grpc_server_retry_throttle_data* throttle_data) { - if (throttle_data == nullptr) return true; +bool ServerRetryThrottleData::RecordFailure() { // First, check if we are stale and need to be replaced. - get_replacement_throttle_data_if_needed(&throttle_data); + ServerRetryThrottleData* throttle_data = this; + GetReplacementThrottleDataIfNeeded(&throttle_data); // We decrement milli_tokens by 1000 (1 token) for each failure. - const int new_value = static_cast<int>(gpr_atm_no_barrier_clamped_add( - &throttle_data->milli_tokens, static_cast<gpr_atm>(-1000), - static_cast<gpr_atm>(0), - static_cast<gpr_atm>(throttle_data->max_milli_tokens))); + const intptr_t new_value = + static_cast<intptr_t>(gpr_atm_no_barrier_clamped_add( + &throttle_data->milli_tokens_, static_cast<gpr_atm>(-1000), + static_cast<gpr_atm>(0), + static_cast<gpr_atm>(throttle_data->max_milli_tokens_))); // Retries are allowed as long as the new value is above the threshold // (max_milli_tokens / 2). - return new_value > throttle_data->max_milli_tokens / 2; + return new_value > throttle_data->max_milli_tokens_ / 2; } -void grpc_server_retry_throttle_data_record_success( - grpc_server_retry_throttle_data* throttle_data) { - if (throttle_data == nullptr) return; +void ServerRetryThrottleData::RecordSuccess() { // First, check if we are stale and need to be replaced. - get_replacement_throttle_data_if_needed(&throttle_data); + ServerRetryThrottleData* throttle_data = this; + GetReplacementThrottleDataIfNeeded(&throttle_data); // We increment milli_tokens by milli_token_ratio for each success. gpr_atm_no_barrier_clamped_add( - &throttle_data->milli_tokens, - static_cast<gpr_atm>(throttle_data->milli_token_ratio), + &throttle_data->milli_tokens_, + static_cast<gpr_atm>(throttle_data->milli_token_ratio_), static_cast<gpr_atm>(0), - static_cast<gpr_atm>(throttle_data->max_milli_tokens)); -} - -grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_ref( - grpc_server_retry_throttle_data* throttle_data) { - gpr_ref(&throttle_data->refs); - return throttle_data; -} - -void grpc_server_retry_throttle_data_unref( - grpc_server_retry_throttle_data* throttle_data) { - if (gpr_unref(&throttle_data->refs)) { - grpc_server_retry_throttle_data* replacement = - (grpc_server_retry_throttle_data*)gpr_atm_acq_load( - &throttle_data->replacement); - if (replacement != nullptr) { - grpc_server_retry_throttle_data_unref(replacement); - } - gpr_free(throttle_data); - } -} - -static grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_create( - int max_milli_tokens, int milli_token_ratio, - grpc_server_retry_throttle_data* old_throttle_data) { - grpc_server_retry_throttle_data* throttle_data = - static_cast<grpc_server_retry_throttle_data*>( - gpr_malloc(sizeof(*throttle_data))); - memset(throttle_data, 0, sizeof(*throttle_data)); - gpr_ref_init(&throttle_data->refs, 1); - throttle_data->max_milli_tokens = max_milli_tokens; - throttle_data->milli_token_ratio = milli_token_ratio; - int initial_milli_tokens = max_milli_tokens; - // If there was a pre-existing entry for this server name, initialize - // the token count by scaling proportionately to the old data. This - // ensures that if we're already throttling retries on the old scale, - // we will start out doing the same thing on the new one. - if (old_throttle_data != nullptr) { - double token_fraction = - static_cast<int>(gpr_atm_acq_load(&old_throttle_data->milli_tokens)) / - static_cast<double>(old_throttle_data->max_milli_tokens); - initial_milli_tokens = static_cast<int>(token_fraction * max_milli_tokens); - } - gpr_atm_rel_store(&throttle_data->milli_tokens, - (gpr_atm)initial_milli_tokens); - // If there was a pre-existing entry, mark it as stale and give it a - // pointer to the new entry, which is its replacement. - if (old_throttle_data != nullptr) { - grpc_server_retry_throttle_data_ref(throttle_data); - gpr_atm_rel_store(&old_throttle_data->replacement, (gpr_atm)throttle_data); - } - return throttle_data; + static_cast<gpr_atm>(throttle_data->max_milli_tokens_)); } // // avl vtable for string -> server_retry_throttle_data map // -static void* copy_server_name(void* key, void* unused) { +namespace { + +void* copy_server_name(void* key, void* unused) { return gpr_strdup(static_cast<const char*>(key)); } -static long compare_server_name(void* key1, void* key2, void* unused) { +long compare_server_name(void* key1, void* key2, void* unused) { return strcmp(static_cast<const char*>(key1), static_cast<const char*>(key2)); } -static void destroy_server_retry_throttle_data(void* value, void* unused) { - grpc_server_retry_throttle_data* throttle_data = - static_cast<grpc_server_retry_throttle_data*>(value); - grpc_server_retry_throttle_data_unref(throttle_data); +void destroy_server_retry_throttle_data(void* value, void* unused) { + ServerRetryThrottleData* throttle_data = + static_cast<ServerRetryThrottleData*>(value); + throttle_data->Unref(); } -static void* copy_server_retry_throttle_data(void* value, void* unused) { - grpc_server_retry_throttle_data* throttle_data = - static_cast<grpc_server_retry_throttle_data*>(value); - return grpc_server_retry_throttle_data_ref(throttle_data); +void* copy_server_retry_throttle_data(void* value, void* unused) { + ServerRetryThrottleData* throttle_data = + static_cast<ServerRetryThrottleData*>(value); + return throttle_data->Ref().release(); } -static void destroy_server_name(void* key, void* unused) { gpr_free(key); } +void destroy_server_name(void* key, void* unused) { gpr_free(key); } -static const grpc_avl_vtable avl_vtable = { +const grpc_avl_vtable avl_vtable = { destroy_server_name, copy_server_name, compare_server_name, destroy_server_retry_throttle_data, copy_server_retry_throttle_data}; +} // namespace + // -// server_retry_throttle_map +// ServerRetryThrottleMap // static gpr_mu g_mu; static grpc_avl g_avl; -void grpc_retry_throttle_map_init() { +void ServerRetryThrottleMap::Init() { gpr_mu_init(&g_mu); g_avl = grpc_avl_create(&avl_vtable); } -void grpc_retry_throttle_map_shutdown() { +void ServerRetryThrottleMap::Shutdown() { gpr_mu_destroy(&g_mu); grpc_avl_unref(g_avl, nullptr); } -grpc_server_retry_throttle_data* grpc_retry_throttle_map_get_data_for_server( - const char* server_name, int max_milli_tokens, int milli_token_ratio) { +RefCountedPtr<ServerRetryThrottleData> ServerRetryThrottleMap::GetDataForServer( + const char* server_name, intptr_t max_milli_tokens, + intptr_t milli_token_ratio) { + RefCountedPtr<ServerRetryThrottleData> result; gpr_mu_lock(&g_mu); - grpc_server_retry_throttle_data* throttle_data = - static_cast<grpc_server_retry_throttle_data*>( + ServerRetryThrottleData* throttle_data = + static_cast<ServerRetryThrottleData*>( grpc_avl_get(g_avl, const_cast<char*>(server_name), nullptr)); - if (throttle_data == nullptr) { - // Entry not found. Create a new one. - throttle_data = grpc_server_retry_throttle_data_create( - max_milli_tokens, milli_token_ratio, nullptr); - g_avl = grpc_avl_add(g_avl, const_cast<char*>(server_name), throttle_data, - nullptr); + if (throttle_data == nullptr || + throttle_data->max_milli_tokens() != max_milli_tokens || + throttle_data->milli_token_ratio() != milli_token_ratio) { + // Entry not found, or found with old parameters. Create a new one. + result = MakeRefCounted<ServerRetryThrottleData>( + max_milli_tokens, milli_token_ratio, throttle_data); + g_avl = grpc_avl_add(g_avl, gpr_strdup(server_name), + result->Ref().release(), nullptr); } else { - if (throttle_data->max_milli_tokens != max_milli_tokens || - throttle_data->milli_token_ratio != milli_token_ratio) { - // Entry found but with old parameters. Create a new one based on - // the original one. - throttle_data = grpc_server_retry_throttle_data_create( - max_milli_tokens, milli_token_ratio, throttle_data); - g_avl = grpc_avl_add(g_avl, const_cast<char*>(server_name), throttle_data, - nullptr); - } else { - // Entry found. Increase refcount. - grpc_server_retry_throttle_data_ref(throttle_data); - } + // Entry found. Return a new ref to it. + result = throttle_data->Ref(); } gpr_mu_unlock(&g_mu); - return throttle_data; + return result; } + +} // namespace internal +} // namespace grpc_core |