diff options
| -rw-r--r-- | CMakeLists.txt | 38 | ||||
| -rw-r--r-- | Makefile | 48 | ||||
| -rw-r--r-- | build.yaml | 12 | ||||
| -rw-r--r-- | src/core/ext/filters/client_channel/client_channel.cc | 38 | ||||
| -rw-r--r-- | src/core/ext/filters/client_channel/client_channel_plugin.cc | 4 | ||||
| -rw-r--r-- | src/core/ext/filters/client_channel/retry_throttle.cc | 222 | ||||
| -rw-r--r-- | src/core/ext/filters/client_channel/retry_throttle.h | 75 | ||||
| -rw-r--r-- | test/core/client_channel/BUILD | 15 | ||||
| -rw-r--r-- | test/core/client_channel/retry_throttle_test.cc | 142 | ||||
| -rw-r--r-- | tools/run_tests/generated/sources_and_headers.json | 17 | ||||
| -rw-r--r-- | tools/run_tests/generated/tests.json | 24 |
11 files changed, 465 insertions, 170 deletions
diff --git a/CMakeLists.txt b/CMakeLists.txt index 56dd61e8cc..6bb12af59c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -605,6 +605,7 @@ add_dependencies(buildtests_cxx reconnect_interop_client) add_dependencies(buildtests_cxx reconnect_interop_server) add_dependencies(buildtests_cxx ref_counted_ptr_test) add_dependencies(buildtests_cxx ref_counted_test) +add_dependencies(buildtests_cxx retry_throttle_test) add_dependencies(buildtests_cxx secure_auth_context_test) if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX) add_dependencies(buildtests_cxx secure_sync_unary_ping_pong_test) @@ -12959,6 +12960,43 @@ target_link_libraries(ref_counted_test endif (gRPC_BUILD_TESTS) if (gRPC_BUILD_TESTS) +add_executable(retry_throttle_test + test/core/client_channel/retry_throttle_test.cc + third_party/googletest/googletest/src/gtest-all.cc + third_party/googletest/googlemock/src/gmock-all.cc +) + + +target_include_directories(retry_throttle_test + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include + PRIVATE ${_gRPC_SSL_INCLUDE_DIR} + PRIVATE ${_gRPC_PROTOBUF_INCLUDE_DIR} + PRIVATE ${_gRPC_ZLIB_INCLUDE_DIR} + PRIVATE ${_gRPC_BENCHMARK_INCLUDE_DIR} + PRIVATE ${_gRPC_CARES_INCLUDE_DIR} + PRIVATE ${_gRPC_GFLAGS_INCLUDE_DIR} + PRIVATE ${_gRPC_ADDRESS_SORTING_INCLUDE_DIR} + PRIVATE third_party/googletest/googletest/include + PRIVATE third_party/googletest/googletest + PRIVATE third_party/googletest/googlemock/include + PRIVATE third_party/googletest/googlemock + PRIVATE ${_gRPC_PROTO_GENS_DIR} +) + +target_link_libraries(retry_throttle_test + ${_gRPC_PROTOBUF_LIBRARIES} + ${_gRPC_ALLTARGETS_LIBRARIES} + grpc_test_util + grpc + gpr_test_util + gpr + ${_gRPC_GFLAGS_LIBRARIES} +) + +endif (gRPC_BUILD_TESTS) +if (gRPC_BUILD_TESTS) + add_executable(secure_auth_context_test test/cpp/common/secure_auth_context_test.cc third_party/googletest/googletest/src/gtest-all.cc @@ -1195,6 +1195,7 @@ reconnect_interop_client: $(BINDIR)/$(CONFIG)/reconnect_interop_client reconnect_interop_server: $(BINDIR)/$(CONFIG)/reconnect_interop_server ref_counted_ptr_test: $(BINDIR)/$(CONFIG)/ref_counted_ptr_test ref_counted_test: $(BINDIR)/$(CONFIG)/ref_counted_test +retry_throttle_test: $(BINDIR)/$(CONFIG)/retry_throttle_test secure_auth_context_test: $(BINDIR)/$(CONFIG)/secure_auth_context_test secure_sync_unary_ping_pong_test: $(BINDIR)/$(CONFIG)/secure_sync_unary_ping_pong_test server_builder_plugin_test: $(BINDIR)/$(CONFIG)/server_builder_plugin_test @@ -1677,6 +1678,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/reconnect_interop_server \ $(BINDIR)/$(CONFIG)/ref_counted_ptr_test \ $(BINDIR)/$(CONFIG)/ref_counted_test \ + $(BINDIR)/$(CONFIG)/retry_throttle_test \ $(BINDIR)/$(CONFIG)/secure_auth_context_test \ $(BINDIR)/$(CONFIG)/secure_sync_unary_ping_pong_test \ $(BINDIR)/$(CONFIG)/server_builder_plugin_test \ @@ -1847,6 +1849,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/reconnect_interop_server \ $(BINDIR)/$(CONFIG)/ref_counted_ptr_test \ $(BINDIR)/$(CONFIG)/ref_counted_test \ + $(BINDIR)/$(CONFIG)/retry_throttle_test \ $(BINDIR)/$(CONFIG)/secure_auth_context_test \ $(BINDIR)/$(CONFIG)/secure_sync_unary_ping_pong_test \ $(BINDIR)/$(CONFIG)/server_builder_plugin_test \ @@ -2304,6 +2307,8 @@ test_cxx: buildtests_cxx $(Q) $(BINDIR)/$(CONFIG)/ref_counted_ptr_test || ( echo test ref_counted_ptr_test failed ; exit 1 ) $(E) "[RUN] Testing ref_counted_test" $(Q) $(BINDIR)/$(CONFIG)/ref_counted_test || ( echo test ref_counted_test failed ; exit 1 ) + $(E) "[RUN] Testing retry_throttle_test" + $(Q) $(BINDIR)/$(CONFIG)/retry_throttle_test || ( echo test retry_throttle_test failed ; exit 1 ) $(E) "[RUN] Testing secure_auth_context_test" $(Q) $(BINDIR)/$(CONFIG)/secure_auth_context_test || ( echo test secure_auth_context_test failed ; exit 1 ) $(E) "[RUN] Testing secure_sync_unary_ping_pong_test" @@ -18725,6 +18730,49 @@ endif endif +RETRY_THROTTLE_TEST_SRC = \ + test/core/client_channel/retry_throttle_test.cc \ + +RETRY_THROTTLE_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(RETRY_THROTTLE_TEST_SRC)))) +ifeq ($(NO_SECURE),true) + +# You can't build secure targets if you don't have OpenSSL. + +$(BINDIR)/$(CONFIG)/retry_throttle_test: openssl_dep_error + +else + + + + +ifeq ($(NO_PROTOBUF),true) + +# You can't build the protoc plugins or protobuf-enabled targets if you don't have protobuf 3.0.0+. + +$(BINDIR)/$(CONFIG)/retry_throttle_test: protobuf_dep_error + +else + +$(BINDIR)/$(CONFIG)/retry_throttle_test: $(PROTOBUF_DEP) $(RETRY_THROTTLE_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a + $(E) "[LD] Linking $@" + $(Q) mkdir -p `dirname $@` + $(Q) $(LDXX) $(LDFLAGS) $(RETRY_THROTTLE_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBSXX) $(LDLIBS_PROTOBUF) $(LDLIBS) $(LDLIBS_SECURE) $(GTEST_LIB) -o $(BINDIR)/$(CONFIG)/retry_throttle_test + +endif + +endif + +$(OBJDIR)/$(CONFIG)/test/core/client_channel/retry_throttle_test.o: $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a + +deps_retry_throttle_test: $(RETRY_THROTTLE_TEST_OBJS:.o=.dep) + +ifneq ($(NO_SECURE),true) +ifneq ($(NO_DEPS),true) +-include $(RETRY_THROTTLE_TEST_OBJS:.o=.dep) +endif +endif + + SECURE_AUTH_CONTEXT_TEST_SRC = \ test/cpp/common/secure_auth_context_test.cc \ diff --git a/build.yaml b/build.yaml index 88f2283dad..a63de093c3 100644 --- a/build.yaml +++ b/build.yaml @@ -4996,6 +4996,18 @@ targets: - gpr uses: - grpc++_test +- name: retry_throttle_test + gtest: true + build: test + language: c++ + src: + - test/core/client_channel/retry_throttle_test.cc + deps: + - grpc_test_util + - grpc + - gpr_test_util + - gpr + uses_polling: false - name: secure_auth_context_test gtest: true build: test diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index 67dd3a1fa7..51f9ae000a 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -63,6 +63,7 @@ #include "src/core/lib/transport/status_metadata.h" using grpc_core::internal::ClientChannelMethodParams; +using grpc_core::internal::ServerRetryThrottleData; /* Client channel implementation */ @@ -99,7 +100,7 @@ typedef struct client_channel_channel_data { /** currently active load balancer */ grpc_core::OrphanablePtr<grpc_core::LoadBalancingPolicy> lb_policy; /** retry throttle data */ - grpc_server_retry_throttle_data* retry_throttle_data; + grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data; /** maps method names to method_parameters structs */ grpc_core::RefCountedPtr<MethodParamsTable> method_params_table; /** incoming resolver result - set by resolver.next() */ @@ -225,7 +226,7 @@ static void start_resolving_locked(channel_data* chand) { typedef struct { char* server_name; - grpc_server_retry_throttle_data* retry_throttle_data; + grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data; } service_config_parsing_state; static void parse_retry_throttle_params( @@ -278,7 +279,7 @@ static void parse_retry_throttle_params( } } parsing_state->retry_throttle_data = - grpc_retry_throttle_map_get_data_for_server( + grpc_core::internal::ServerRetryThrottleMap::GetDataForServer( parsing_state->server_name, max_milli_tokens, milli_token_ratio); } } @@ -321,7 +322,7 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { bool lb_policy_name_changed = false; grpc_core::OrphanablePtr<grpc_core::LoadBalancingPolicy> new_lb_policy; char* service_config_json = nullptr; - grpc_server_retry_throttle_data* retry_throttle_data = nullptr; + grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data; grpc_core::RefCountedPtr<MethodParamsTable> method_params_table; if (chand->resolver_result != nullptr) { if (chand->resolver != nullptr) { @@ -421,7 +422,7 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { service_config->ParseGlobalParams(parse_retry_throttle_params, &parsing_state); grpc_uri_destroy(uri); - retry_throttle_data = parsing_state.retry_throttle_data; + retry_throttle_data = std::move(parsing_state.retry_throttle_data); } method_params_table = service_config->CreateMethodConfigTable( ClientChannelMethodParams::CreateFromJson); @@ -452,10 +453,7 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { } gpr_mu_unlock(&chand->info_mu); // Swap out the retry throttle data. - if (chand->retry_throttle_data != nullptr) { - grpc_server_retry_throttle_data_unref(chand->retry_throttle_data); - } - chand->retry_throttle_data = retry_throttle_data; + chand->retry_throttle_data = std::move(retry_throttle_data); // Swap out the method params table. chand->method_params_table = std::move(method_params_table); // If we have a new LB policy or are shutting down (in which case @@ -725,12 +723,8 @@ static void cc_destroy_channel_elem(grpc_channel_element* elem) { } gpr_free(chand->info_lb_policy_name); gpr_free(chand->info_service_config_json); - if (chand->retry_throttle_data != nullptr) { - grpc_server_retry_throttle_data_unref(chand->retry_throttle_data); - } - if (chand->method_params_table != nullptr) { - chand->method_params_table.reset(); - } + chand->retry_throttle_data.reset(); + chand->method_params_table.reset(); grpc_client_channel_stop_backup_polling(chand->interested_parties); grpc_connectivity_state_destroy(&chand->state_tracker); grpc_pollset_set_destroy(chand->interested_parties); @@ -883,7 +877,7 @@ typedef struct client_channel_call_data { grpc_call_stack* owning_call; grpc_call_combiner* call_combiner; - grpc_server_retry_throttle_data* retry_throttle_data; + grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data; grpc_core::RefCountedPtr<ClientChannelMethodParams> method_params; grpc_subchannel_call* subchannel_call; @@ -1443,7 +1437,9 @@ static bool maybe_retry(grpc_call_element* elem, } // Check status. if (status == GRPC_STATUS_OK) { - grpc_server_retry_throttle_data_record_success(calld->retry_throttle_data); + if (calld->retry_throttle_data != nullptr) { + calld->retry_throttle_data->RecordSuccess(); + } if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: call succeeded", chand, calld); } @@ -1465,8 +1461,8 @@ static bool maybe_retry(grpc_call_element* elem, // things like failures due to malformed requests (INVALID_ARGUMENT). // Conversely, it's important for this to come before the remaining // checks, so that we don't fail to record failures due to other factors. - if (!grpc_server_retry_throttle_data_record_failure( - calld->retry_throttle_data)) { + if (calld->retry_throttle_data != nullptr && + !calld->retry_throttle_data->RecordFailure()) { if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: retries throttled", chand, calld); } @@ -2601,8 +2597,7 @@ static void apply_service_config_to_call_locked(grpc_call_element* elem) { chand, calld); } if (chand->retry_throttle_data != nullptr) { - calld->retry_throttle_data = - grpc_server_retry_throttle_data_ref(chand->retry_throttle_data); + calld->retry_throttle_data = chand->retry_throttle_data->Ref(); } if (chand->method_params_table != nullptr) { calld->method_params = grpc_core::ServiceConfig::MethodConfigTableLookup( @@ -2994,6 +2989,7 @@ static void cc_destroy_call_elem(grpc_call_element* elem, grpc_deadline_state_destroy(elem); } grpc_slice_unref_internal(calld->path); + calld->retry_throttle_data.reset(); calld->method_params.reset(); GRPC_ERROR_UNREF(calld->cancel_error); if (calld->subchannel_call != nullptr) { diff --git a/src/core/ext/filters/client_channel/client_channel_plugin.cc b/src/core/ext/filters/client_channel/client_channel_plugin.cc index 3c3a97532f..b4ef534d2e 100644 --- a/src/core/ext/filters/client_channel/client_channel_plugin.cc +++ b/src/core/ext/filters/client_channel/client_channel_plugin.cc @@ -65,7 +65,7 @@ static bool set_default_host_if_unset(grpc_channel_stack_builder* builder, void grpc_client_channel_init(void) { grpc_core::LoadBalancingPolicyRegistry::Builder::InitRegistry(); grpc_core::ResolverRegistry::Builder::InitRegistry(); - grpc_retry_throttle_map_init(); + grpc_core::internal::ServerRetryThrottleMap::Init(); grpc_proxy_mapper_registry_init(); grpc_register_http_proxy_mapper(); grpc_subchannel_index_init(); @@ -81,7 +81,7 @@ void grpc_client_channel_shutdown(void) { grpc_subchannel_index_shutdown(); grpc_channel_init_shutdown(); grpc_proxy_mapper_registry_shutdown(); - grpc_retry_throttle_map_shutdown(); + grpc_core::internal::ServerRetryThrottleMap::Shutdown(); grpc_core::ResolverRegistry::Builder::ShutdownRegistry(); grpc_core::LoadBalancingPolicyRegistry::Builder::ShutdownRegistry(); } 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 diff --git a/src/core/ext/filters/client_channel/retry_throttle.h b/src/core/ext/filters/client_channel/retry_throttle.h index 0505fc27f2..2b6fa0a70b 100644 --- a/src/core/ext/filters/client_channel/retry_throttle.h +++ b/src/core/ext/filters/client_channel/retry_throttle.h @@ -21,32 +21,57 @@ #include <grpc/support/port_platform.h> -#include <stdbool.h> +#include "src/core/lib/gprpp/ref_counted.h" + +namespace grpc_core { +namespace internal { /// Tracks retry throttling data for an individual server name. -typedef struct grpc_server_retry_throttle_data grpc_server_retry_throttle_data; - -/// Records a failure. Returns true if it's okay to send a retry. -bool grpc_server_retry_throttle_data_record_failure( - grpc_server_retry_throttle_data* throttle_data); -/// Records a success. -void grpc_server_retry_throttle_data_record_success( - grpc_server_retry_throttle_data* throttle_data); - -grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_ref( - grpc_server_retry_throttle_data* throttle_data); -void grpc_server_retry_throttle_data_unref( - grpc_server_retry_throttle_data* throttle_data); - -/// Initializes global map of failure data for each server name. -void grpc_retry_throttle_map_init(); -/// Shuts down global map of failure data for each server name. -void grpc_retry_throttle_map_shutdown(); - -/// Returns a reference to the failure data for \a server_name, creating -/// a new entry if needed. -/// Caller must eventually unref via \a grpc_server_retry_throttle_data_unref(). -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); +class ServerRetryThrottleData : public RefCounted<ServerRetryThrottleData> { + public: + ServerRetryThrottleData(intptr_t max_milli_tokens, intptr_t milli_token_ratio, + ServerRetryThrottleData* old_throttle_data); + + /// Records a failure. Returns true if it's okay to send a retry. + bool RecordFailure(); + + /// Records a success. + void RecordSuccess(); + + intptr_t max_milli_tokens() const { return max_milli_tokens_; } + intptr_t milli_token_ratio() const { return milli_token_ratio_; } + + private: + ~ServerRetryThrottleData(); + + void GetReplacementThrottleDataIfNeeded( + ServerRetryThrottleData** throttle_data); + + const intptr_t max_milli_tokens_; + const intptr_t milli_token_ratio_; + gpr_atm milli_tokens_; + // A pointer to the replacement for this ServerRetryThrottleData entry. + // If non-nullptr, then this entry is stale and must not be used. + // We hold a reference to the replacement. + gpr_atm replacement_ = 0; +}; + +/// Global map of server name to retry throttle data. +class ServerRetryThrottleMap { + public: + /// Initializes global map of failure data for each server name. + static void Init(); + /// Shuts down global map of failure data for each server name. + static void Shutdown(); + + /// Returns the failure data for \a server_name, creating a new entry if + /// needed. + static RefCountedPtr<ServerRetryThrottleData> GetDataForServer( + const char* server_name, intptr_t max_milli_tokens, + intptr_t milli_token_ratio); +}; + +} // namespace internal +} // namespace grpc_core #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RETRY_THROTTLE_H */ diff --git a/test/core/client_channel/BUILD b/test/core/client_channel/BUILD index 5148dc5f74..db98ffab77 100644 --- a/test/core/client_channel/BUILD +++ b/test/core/client_channel/BUILD @@ -53,3 +53,18 @@ grpc_cc_test( "//test/core/util:grpc_test_util", ], ) + +grpc_cc_test( + name = "retry_throttle_test", + srcs = ["retry_throttle_test.cc"], + external_deps = [ + "gtest", + ], + language = "C++", + deps = [ + "//:gpr", + "//:grpc", + "//test/core/util:gpr_test_util", + "//test/core/util:grpc_test_util", + ], +) diff --git a/test/core/client_channel/retry_throttle_test.cc b/test/core/client_channel/retry_throttle_test.cc new file mode 100644 index 0000000000..c6d5d3ebbb --- /dev/null +++ b/test/core/client_channel/retry_throttle_test.cc @@ -0,0 +1,142 @@ +/* + * + * Copyright 2018 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#include "src/core/ext/filters/client_channel/retry_throttle.h" + +#include <gtest/gtest.h> + +#include "test/core/util/test_config.h" + +namespace grpc_core { +namespace internal { +namespace { + +TEST(ServerRetryThrottleData, Basic) { + // Max token count is 4, so threshold for retrying is 2. + // Token count starts at 4. + // Each failure decrements by 1. Each success increments by 1.6. + auto throttle_data = + MakeRefCounted<ServerRetryThrottleData>(4000, 1600, nullptr); + // Failure: token_count=3. Above threshold. + EXPECT_TRUE(throttle_data->RecordFailure()); + // Success: token_count=4. Not incremented beyond max. + throttle_data->RecordSuccess(); + // Failure: token_count=3. Above threshold. + EXPECT_TRUE(throttle_data->RecordFailure()); + // Failure: token_count=2. At threshold, so no retries. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Failure: token_count=1. Below threshold, so no retries. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Failure: token_count=0. Below threshold, so no retries. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Failure: token_count=0. Below threshold, so no retries. Not + // decremented below min. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Success: token_count=1.6. + throttle_data->RecordSuccess(); + // Success: token_count=3.2. + throttle_data->RecordSuccess(); + // Failure: token_count=2.2. Above threshold. + EXPECT_TRUE(throttle_data->RecordFailure()); + // Failure: token_count=1.2. Below threshold, so no retries. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Success: token_count=2.8. + throttle_data->RecordSuccess(); + // Failure: token_count=1.8. Below threshold, so no retries. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Success: token_count=3.4. + throttle_data->RecordSuccess(); + // Failure: token_count=2.4. Above threshold. + EXPECT_TRUE(throttle_data->RecordFailure()); +} + +TEST(ServerRetryThrottleData, Replacement) { + // Create old throttle data. + // Max token count is 4, so threshold for retrying is 2. + // Token count starts at 4. + // Each failure decrements by 1. Each success increments by 1. + auto old_throttle_data = + MakeRefCounted<ServerRetryThrottleData>(4000, 1000, nullptr); + // Failure: token_count=3. Above threshold. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Create new throttle data. + // Max token count is 10, so threshold for retrying is 5. + // Token count starts at 7.5 (ratio inherited from old_throttle_data). + // Each failure decrements by 1. Each success increments by 3. + auto throttle_data = MakeRefCounted<ServerRetryThrottleData>( + 10000, 3000, old_throttle_data.get()); + // Failure via old_throttle_data: token_count=6.5. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Failure: token_count=5.5. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Failure via old_throttle_data: token_count=4.5. Below threshold. + EXPECT_FALSE(old_throttle_data->RecordFailure()); + // Failure: token_count=3.5. Below threshold. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Success: token_count=6.5. + throttle_data->RecordSuccess(); + // Failure via old_throttle_data: token_count=5.5. Above threshold. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Failure: token_count=4.5. Below threshold. + EXPECT_FALSE(throttle_data->RecordFailure()); +} + +TEST(ServerRetryThrottleMap, Replacement) { + ServerRetryThrottleMap::Init(); + const char kServerName[] = "server_name"; + // Create old throttle data. + // Max token count is 4, so threshold for retrying is 2. + // Token count starts at 4. + // Each failure decrements by 1. Each success increments by 1. + auto old_throttle_data = + ServerRetryThrottleMap::GetDataForServer(kServerName, 4000, 1000); + // Failure: token_count=3. Above threshold. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Create new throttle data. + // Max token count is 10, so threshold for retrying is 5. + // Token count starts at 7.5 (ratio inherited from old_throttle_data). + // Each failure decrements by 1. Each success increments by 3. + auto throttle_data = + ServerRetryThrottleMap::GetDataForServer(kServerName, 10000, 3000); + // Failure via old_throttle_data: token_count=6.5. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Failure: token_count=5.5. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Failure via old_throttle_data: token_count=4.5. Below threshold. + EXPECT_FALSE(old_throttle_data->RecordFailure()); + // Failure: token_count=3.5. Below threshold. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Success: token_count=6.5. + throttle_data->RecordSuccess(); + // Failure via old_throttle_data: token_count=5.5. Above threshold. + EXPECT_TRUE(old_throttle_data->RecordFailure()); + // Failure: token_count=4.5. Below threshold. + EXPECT_FALSE(throttle_data->RecordFailure()); + // Clean up. + ServerRetryThrottleMap::Shutdown(); +} + +} // namespace +} // namespace internal +} // namespace grpc_core + +int main(int argc, char** argv) { + grpc_test_init(argc, argv); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index fd6c526ea5..3c32ccc328 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -4248,6 +4248,23 @@ "gpr", "gpr_test_util", "grpc", + "grpc_test_util" + ], + "headers": [], + "is_filegroup": false, + "language": "c++", + "name": "retry_throttle_test", + "src": [ + "test/core/client_channel/retry_throttle_test.cc" + ], + "third_party": false, + "type": "target" + }, + { + "deps": [ + "gpr", + "gpr_test_util", + "grpc", "grpc++", "grpc++_test_util", "grpc_test_util" diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index c2bbf8c88a..15efff67e0 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -4666,6 +4666,30 @@ "flaky": false, "gtest": true, "language": "c++", + "name": "retry_throttle_test", + "platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "uses_polling": false + }, + { + "args": [], + "benchmark": false, + "ci_platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "cpu_cost": 1.0, + "exclude_configs": [], + "exclude_iomgrs": [], + "flaky": false, + "gtest": true, + "language": "c++", "name": "secure_auth_context_test", "platforms": [ "linux", |