Convert retry throttle code to C++ and add tests.

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

Makefile

@@ -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 \
 

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

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) {

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();
 }