diff options
| author |  Yash Tibrewal <yashkt@google.com> | 2018-09-13 15:16:02 -0700 |
|---|---|---|
| committer | Yash Tibrewal <yashkt@google.com> | 2018-09-13 15:16:02 -0700 |
| commit | 1819935efbb2fb014c0c29bc47f5045fbcf5c6cf (patch) | |
| tree | b59127474ac1b007a44d4209caadc27e7355a52a | |
| parent | 30e7b02b5c8bc55c109ed84dfa30663ce90e134b (diff) | |
| parent | 07eecd8421314d62e8df33b14a33203864f91bee (diff) | |
Merge branch 'master' into statuscaution
128 files changed, 3055 insertions, 589 deletions
@@ -119,6 +119,7 @@ GRPCXX_SRCS = [ "src/cpp/client/credentials_cc.cc", "src/cpp/client/generic_stub.cc", "src/cpp/common/alarm.cc", + "src/cpp/common/callback_common.cc", "src/cpp/common/channel_arguments.cc", "src/cpp/common/channel_filter.cc", "src/cpp/common/completion_queue_cc.cc", @@ -243,6 +244,7 @@ GRPCXX_PUBLIC_HDRS = [ "include/grpcpp/support/async_unary_call.h", "include/grpcpp/support/byte_buffer.h", "include/grpcpp/support/channel_arguments.h", + "include/grpcpp/support/client_callback.h", "include/grpcpp/support/config.h", "include/grpcpp/support/proto_buffer_reader.h", "include/grpcpp/support/proto_buffer_writer.h", @@ -1979,7 +1981,9 @@ grpc_cc_library( "include/grpcpp/impl/codegen/byte_buffer.h", "include/grpcpp/impl/codegen/call.h", "include/grpcpp/impl/codegen/call_hook.h", + "include/grpcpp/impl/codegen/callback_common.h", "include/grpcpp/impl/codegen/channel_interface.h", + "include/grpcpp/impl/codegen/client_callback.h", "include/grpcpp/impl/codegen/client_context.h", "include/grpcpp/impl/codegen/client_unary_call.h", "include/grpcpp/impl/codegen/completion_queue.h", diff --git a/CMakeLists.txt b/CMakeLists.txt index a21bb8b5fa..c358e9bd43 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -361,7 +361,8 @@ endif() if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX) add_dependencies(buildtests_c resolve_address_posix_test) endif() -add_dependencies(buildtests_c resolve_address_test) +add_dependencies(buildtests_c resolve_address_using_ares_resolver_test) +add_dependencies(buildtests_c resolve_address_using_native_resolver_test) add_dependencies(buildtests_c resource_quota_test) add_dependencies(buildtests_c secure_channel_create_test) add_dependencies(buildtests_c secure_endpoint_test) @@ -564,6 +565,7 @@ add_dependencies(buildtests_cxx check_gcp_environment_linux_test) add_dependencies(buildtests_cxx check_gcp_environment_windows_test) add_dependencies(buildtests_cxx chttp2_settings_timeout_test) add_dependencies(buildtests_cxx cli_call_test) +add_dependencies(buildtests_cxx client_callback_end2end_test) add_dependencies(buildtests_cxx client_channel_stress_test) if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX) add_dependencies(buildtests_cxx client_crash_test) @@ -2771,6 +2773,7 @@ add_library(grpc++ src/cpp/client/credentials_cc.cc src/cpp/client/generic_stub.cc src/cpp/common/alarm.cc + src/cpp/common/callback_common.cc src/cpp/common/channel_arguments.cc src/cpp/common/channel_filter.cc src/cpp/common/completion_queue_cc.cc @@ -2917,6 +2920,7 @@ foreach(_hdr include/grpcpp/support/async_unary_call.h include/grpcpp/support/byte_buffer.h include/grpcpp/support/channel_arguments.h + include/grpcpp/support/client_callback.h include/grpcpp/support/config.h include/grpcpp/support/proto_buffer_reader.h include/grpcpp/support/proto_buffer_writer.h @@ -3014,7 +3018,9 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h + include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h + include/grpcpp/impl/codegen/client_callback.h include/grpcpp/impl/codegen/client_context.h include/grpcpp/impl/codegen/client_unary_call.h include/grpcpp/impl/codegen/completion_queue.h @@ -3129,6 +3135,7 @@ add_library(grpc++_cronet src/cpp/client/credentials_cc.cc src/cpp/client/generic_stub.cc src/cpp/common/alarm.cc + src/cpp/common/callback_common.cc src/cpp/common/channel_arguments.cc src/cpp/common/channel_filter.cc src/cpp/common/completion_queue_cc.cc @@ -3486,6 +3493,7 @@ foreach(_hdr include/grpcpp/support/async_unary_call.h include/grpcpp/support/byte_buffer.h include/grpcpp/support/channel_arguments.h + include/grpcpp/support/client_callback.h include/grpcpp/support/config.h include/grpcpp/support/proto_buffer_reader.h include/grpcpp/support/proto_buffer_writer.h @@ -3583,7 +3591,9 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h + include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h + include/grpcpp/impl/codegen/client_callback.h include/grpcpp/impl/codegen/client_context.h include/grpcpp/impl/codegen/client_unary_call.h include/grpcpp/impl/codegen/completion_queue.h @@ -3993,7 +4003,9 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h + include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h + include/grpcpp/impl/codegen/client_callback.h include/grpcpp/impl/codegen/client_context.h include/grpcpp/impl/codegen/client_unary_call.h include/grpcpp/impl/codegen/completion_queue.h @@ -4171,7 +4183,9 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h + include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h + include/grpcpp/impl/codegen/client_callback.h include/grpcpp/impl/codegen/client_context.h include/grpcpp/impl/codegen/client_unary_call.h include/grpcpp/impl/codegen/completion_queue.h @@ -4247,6 +4261,7 @@ add_library(grpc++_unsecure src/cpp/client/credentials_cc.cc src/cpp/client/generic_stub.cc src/cpp/common/alarm.cc + src/cpp/common/callback_common.cc src/cpp/common/channel_arguments.cc src/cpp/common/channel_filter.cc src/cpp/common/completion_queue_cc.cc @@ -4392,6 +4407,7 @@ foreach(_hdr include/grpcpp/support/async_unary_call.h include/grpcpp/support/byte_buffer.h include/grpcpp/support/channel_arguments.h + include/grpcpp/support/client_callback.h include/grpcpp/support/config.h include/grpcpp/support/proto_buffer_reader.h include/grpcpp/support/proto_buffer_writer.h @@ -4489,7 +4505,9 @@ foreach(_hdr include/grpcpp/impl/codegen/byte_buffer.h include/grpcpp/impl/codegen/call.h include/grpcpp/impl/codegen/call_hook.h + include/grpcpp/impl/codegen/callback_common.h include/grpcpp/impl/codegen/channel_interface.h + include/grpcpp/impl/codegen/client_callback.h include/grpcpp/impl/codegen/client_context.h include/grpcpp/impl/codegen/client_unary_call.h include/grpcpp/impl/codegen/completion_queue.h @@ -8553,12 +8571,12 @@ endif() endif (gRPC_BUILD_TESTS) if (gRPC_BUILD_TESTS) -add_executable(resolve_address_test +add_executable(resolve_address_using_ares_resolver_test test/core/iomgr/resolve_address_test.cc ) -target_include_directories(resolve_address_test +target_include_directories(resolve_address_using_ares_resolver_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include PRIVATE ${_gRPC_SSL_INCLUDE_DIR} @@ -8571,7 +8589,36 @@ target_include_directories(resolve_address_test PRIVATE ${_gRPC_NANOPB_INCLUDE_DIR} ) -target_link_libraries(resolve_address_test +target_link_libraries(resolve_address_using_ares_resolver_test + ${_gRPC_ALLTARGETS_LIBRARIES} + grpc_test_util + grpc + gpr_test_util + gpr +) + +endif (gRPC_BUILD_TESTS) +if (gRPC_BUILD_TESTS) + +add_executable(resolve_address_using_native_resolver_test + test/core/iomgr/resolve_address_test.cc +) + + +target_include_directories(resolve_address_using_native_resolver_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 ${_gRPC_NANOPB_INCLUDE_DIR} +) + +target_link_libraries(resolve_address_using_native_resolver_test ${_gRPC_ALLTARGETS_LIBRARIES} grpc_test_util grpc @@ -11297,6 +11344,46 @@ target_link_libraries(cli_call_test endif (gRPC_BUILD_TESTS) if (gRPC_BUILD_TESTS) +add_executable(client_callback_end2end_test + test/cpp/end2end/client_callback_end2end_test.cc + third_party/googletest/googletest/src/gtest-all.cc + third_party/googletest/googlemock/src/gmock-all.cc +) + + +target_include_directories(client_callback_end2end_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 ${_gRPC_NANOPB_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(client_callback_end2end_test + ${_gRPC_PROTOBUF_LIBRARIES} + ${_gRPC_ALLTARGETS_LIBRARIES} + grpc++_test_util + grpc_test_util + grpc++ + grpc + gpr_test_util + gpr + ${_gRPC_GFLAGS_LIBRARIES} +) + +endif (gRPC_BUILD_TESTS) +if (gRPC_BUILD_TESTS) + add_executable(client_channel_stress_test ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/lb/v1/load_balancer.pb.cc ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/lb/v1/load_balancer.grpc.pb.cc @@ -1081,7 +1081,8 @@ percent_encode_fuzzer: $(BINDIR)/$(CONFIG)/percent_encode_fuzzer percent_encoding_test: $(BINDIR)/$(CONFIG)/percent_encoding_test pollset_set_test: $(BINDIR)/$(CONFIG)/pollset_set_test resolve_address_posix_test: $(BINDIR)/$(CONFIG)/resolve_address_posix_test -resolve_address_test: $(BINDIR)/$(CONFIG)/resolve_address_test +resolve_address_using_ares_resolver_test: $(BINDIR)/$(CONFIG)/resolve_address_using_ares_resolver_test +resolve_address_using_native_resolver_test: $(BINDIR)/$(CONFIG)/resolve_address_using_native_resolver_test resource_quota_test: $(BINDIR)/$(CONFIG)/resource_quota_test secure_channel_create_test: $(BINDIR)/$(CONFIG)/secure_channel_create_test secure_endpoint_test: $(BINDIR)/$(CONFIG)/secure_endpoint_test @@ -1160,6 +1161,7 @@ check_gcp_environment_linux_test: $(BINDIR)/$(CONFIG)/check_gcp_environment_linu check_gcp_environment_windows_test: $(BINDIR)/$(CONFIG)/check_gcp_environment_windows_test chttp2_settings_timeout_test: $(BINDIR)/$(CONFIG)/chttp2_settings_timeout_test cli_call_test: $(BINDIR)/$(CONFIG)/cli_call_test +client_callback_end2end_test: $(BINDIR)/$(CONFIG)/client_callback_end2end_test client_channel_stress_test: $(BINDIR)/$(CONFIG)/client_channel_stress_test client_crash_test: $(BINDIR)/$(CONFIG)/client_crash_test client_crash_test_server: $(BINDIR)/$(CONFIG)/client_crash_test_server @@ -1524,7 +1526,8 @@ buildtests_c: privatelibs_c \ $(BINDIR)/$(CONFIG)/percent_encoding_test \ $(BINDIR)/$(CONFIG)/pollset_set_test \ $(BINDIR)/$(CONFIG)/resolve_address_posix_test \ - $(BINDIR)/$(CONFIG)/resolve_address_test \ + $(BINDIR)/$(CONFIG)/resolve_address_using_ares_resolver_test \ + $(BINDIR)/$(CONFIG)/resolve_address_using_native_resolver_test \ $(BINDIR)/$(CONFIG)/resource_quota_test \ $(BINDIR)/$(CONFIG)/secure_channel_create_test \ $(BINDIR)/$(CONFIG)/secure_endpoint_test \ @@ -1665,6 +1668,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/check_gcp_environment_windows_test \ $(BINDIR)/$(CONFIG)/chttp2_settings_timeout_test \ $(BINDIR)/$(CONFIG)/cli_call_test \ + $(BINDIR)/$(CONFIG)/client_callback_end2end_test \ $(BINDIR)/$(CONFIG)/client_channel_stress_test \ $(BINDIR)/$(CONFIG)/client_crash_test \ $(BINDIR)/$(CONFIG)/client_crash_test_server \ @@ -1845,6 +1849,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/check_gcp_environment_windows_test \ $(BINDIR)/$(CONFIG)/chttp2_settings_timeout_test \ $(BINDIR)/$(CONFIG)/cli_call_test \ + $(BINDIR)/$(CONFIG)/client_callback_end2end_test \ $(BINDIR)/$(CONFIG)/client_channel_stress_test \ $(BINDIR)/$(CONFIG)/client_crash_test \ $(BINDIR)/$(CONFIG)/client_crash_test_server \ @@ -2118,8 +2123,10 @@ test_c: buildtests_c $(Q) $(BINDIR)/$(CONFIG)/pollset_set_test || ( echo test pollset_set_test failed ; exit 1 ) $(E) "[RUN] Testing resolve_address_posix_test" $(Q) $(BINDIR)/$(CONFIG)/resolve_address_posix_test || ( echo test resolve_address_posix_test failed ; exit 1 ) - $(E) "[RUN] Testing resolve_address_test" - $(Q) $(BINDIR)/$(CONFIG)/resolve_address_test || ( echo test resolve_address_test failed ; exit 1 ) + $(E) "[RUN] Testing resolve_address_using_ares_resolver_test" + $(Q) $(BINDIR)/$(CONFIG)/resolve_address_using_ares_resolver_test || ( echo test resolve_address_using_ares_resolver_test failed ; exit 1 ) + $(E) "[RUN] Testing resolve_address_using_native_resolver_test" + $(Q) $(BINDIR)/$(CONFIG)/resolve_address_using_native_resolver_test || ( echo test resolve_address_using_native_resolver_test failed ; exit 1 ) $(E) "[RUN] Testing resource_quota_test" $(Q) $(BINDIR)/$(CONFIG)/resource_quota_test || ( echo test resource_quota_test failed ; exit 1 ) $(E) "[RUN] Testing secure_channel_create_test" @@ -2302,6 +2309,8 @@ test_cxx: buildtests_cxx $(Q) $(BINDIR)/$(CONFIG)/chttp2_settings_timeout_test || ( echo test chttp2_settings_timeout_test failed ; exit 1 ) $(E) "[RUN] Testing cli_call_test" $(Q) $(BINDIR)/$(CONFIG)/cli_call_test || ( echo test cli_call_test failed ; exit 1 ) + $(E) "[RUN] Testing client_callback_end2end_test" + $(Q) $(BINDIR)/$(CONFIG)/client_callback_end2end_test || ( echo test client_callback_end2end_test failed ; exit 1 ) $(E) "[RUN] Testing client_channel_stress_test" $(Q) $(BINDIR)/$(CONFIG)/client_channel_stress_test || ( echo test client_channel_stress_test failed ; exit 1 ) $(E) "[RUN] Testing client_crash_test" @@ -5219,6 +5228,7 @@ LIBGRPC++_SRC = \ src/cpp/client/credentials_cc.cc \ src/cpp/client/generic_stub.cc \ src/cpp/common/alarm.cc \ + src/cpp/common/callback_common.cc \ src/cpp/common/channel_arguments.cc \ src/cpp/common/channel_filter.cc \ src/cpp/common/completion_queue_cc.cc \ @@ -5329,6 +5339,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/support/async_unary_call.h \ include/grpcpp/support/byte_buffer.h \ include/grpcpp/support/channel_arguments.h \ + include/grpcpp/support/client_callback.h \ include/grpcpp/support/config.h \ include/grpcpp/support/proto_buffer_reader.h \ include/grpcpp/support/proto_buffer_writer.h \ @@ -5426,7 +5437,9 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ + include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ + include/grpcpp/impl/codegen/client_callback.h \ include/grpcpp/impl/codegen/client_context.h \ include/grpcpp/impl/codegen/client_unary_call.h \ include/grpcpp/impl/codegen/completion_queue.h \ @@ -5585,6 +5598,7 @@ LIBGRPC++_CRONET_SRC = \ src/cpp/client/credentials_cc.cc \ src/cpp/client/generic_stub.cc \ src/cpp/common/alarm.cc \ + src/cpp/common/callback_common.cc \ src/cpp/common/channel_arguments.cc \ src/cpp/common/channel_filter.cc \ src/cpp/common/completion_queue_cc.cc \ @@ -5905,6 +5919,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/support/async_unary_call.h \ include/grpcpp/support/byte_buffer.h \ include/grpcpp/support/channel_arguments.h \ + include/grpcpp/support/client_callback.h \ include/grpcpp/support/config.h \ include/grpcpp/support/proto_buffer_reader.h \ include/grpcpp/support/proto_buffer_writer.h \ @@ -6002,7 +6017,9 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ + include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ + include/grpcpp/impl/codegen/client_callback.h \ include/grpcpp/impl/codegen/client_context.h \ include/grpcpp/impl/codegen/client_unary_call.h \ include/grpcpp/impl/codegen/completion_queue.h \ @@ -6392,7 +6409,9 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ + include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ + include/grpcpp/impl/codegen/client_callback.h \ include/grpcpp/impl/codegen/client_context.h \ include/grpcpp/impl/codegen/client_unary_call.h \ include/grpcpp/impl/codegen/completion_queue.h \ @@ -6546,7 +6565,9 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ + include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ + include/grpcpp/impl/codegen/client_callback.h \ include/grpcpp/impl/codegen/client_context.h \ include/grpcpp/impl/codegen/client_unary_call.h \ include/grpcpp/impl/codegen/completion_queue.h \ @@ -6661,6 +6682,7 @@ LIBGRPC++_UNSECURE_SRC = \ src/cpp/client/credentials_cc.cc \ src/cpp/client/generic_stub.cc \ src/cpp/common/alarm.cc \ + src/cpp/common/callback_common.cc \ src/cpp/common/channel_arguments.cc \ src/cpp/common/channel_filter.cc \ src/cpp/common/completion_queue_cc.cc \ @@ -6771,6 +6793,7 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/support/async_unary_call.h \ include/grpcpp/support/byte_buffer.h \ include/grpcpp/support/channel_arguments.h \ + include/grpcpp/support/client_callback.h \ include/grpcpp/support/config.h \ include/grpcpp/support/proto_buffer_reader.h \ include/grpcpp/support/proto_buffer_writer.h \ @@ -6868,7 +6891,9 @@ PUBLIC_HEADERS_CXX += \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ + include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ + include/grpcpp/impl/codegen/client_callback.h \ include/grpcpp/impl/codegen/client_context.h \ include/grpcpp/impl/codegen/client_unary_call.h \ include/grpcpp/impl/codegen/completion_queue.h \ @@ -14031,34 +14056,66 @@ endif endif -RESOLVE_ADDRESS_TEST_SRC = \ +RESOLVE_ADDRESS_USING_ARES_RESOLVER_TEST_SRC = \ test/core/iomgr/resolve_address_test.cc \ -RESOLVE_ADDRESS_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(RESOLVE_ADDRESS_TEST_SRC)))) +RESOLVE_ADDRESS_USING_ARES_RESOLVER_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(RESOLVE_ADDRESS_USING_ARES_RESOLVER_TEST_SRC)))) ifeq ($(NO_SECURE),true) # You can't build secure targets if you don't have OpenSSL. -$(BINDIR)/$(CONFIG)/resolve_address_test: openssl_dep_error +$(BINDIR)/$(CONFIG)/resolve_address_using_ares_resolver_test: openssl_dep_error else -$(BINDIR)/$(CONFIG)/resolve_address_test: $(RESOLVE_ADDRESS_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a +$(BINDIR)/$(CONFIG)/resolve_address_using_ares_resolver_test: $(RESOLVE_ADDRESS_USING_ARES_RESOLVER_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) $(LD) $(LDFLAGS) $(RESOLVE_ADDRESS_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBS) $(LDLIBS_SECURE) -o $(BINDIR)/$(CONFIG)/resolve_address_test + $(Q) $(LD) $(LDFLAGS) $(RESOLVE_ADDRESS_USING_ARES_RESOLVER_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBS) $(LDLIBS_SECURE) -o $(BINDIR)/$(CONFIG)/resolve_address_using_ares_resolver_test endif $(OBJDIR)/$(CONFIG)/test/core/iomgr/resolve_address_test.o: $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a -deps_resolve_address_test: $(RESOLVE_ADDRESS_TEST_OBJS:.o=.dep) +deps_resolve_address_using_ares_resolver_test: $(RESOLVE_ADDRESS_USING_ARES_RESOLVER_TEST_OBJS:.o=.dep) ifneq ($(NO_SECURE),true) ifneq ($(NO_DEPS),true) --include $(RESOLVE_ADDRESS_TEST_OBJS:.o=.dep) +-include $(RESOLVE_ADDRESS_USING_ARES_RESOLVER_TEST_OBJS:.o=.dep) +endif +endif + + +RESOLVE_ADDRESS_USING_NATIVE_RESOLVER_TEST_SRC = \ + test/core/iomgr/resolve_address_test.cc \ + +RESOLVE_ADDRESS_USING_NATIVE_RESOLVER_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(RESOLVE_ADDRESS_USING_NATIVE_RESOLVER_TEST_SRC)))) +ifeq ($(NO_SECURE),true) + +# You can't build secure targets if you don't have OpenSSL. + +$(BINDIR)/$(CONFIG)/resolve_address_using_native_resolver_test: openssl_dep_error + +else + + + +$(BINDIR)/$(CONFIG)/resolve_address_using_native_resolver_test: $(RESOLVE_ADDRESS_USING_NATIVE_RESOLVER_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) $(LD) $(LDFLAGS) $(RESOLVE_ADDRESS_USING_NATIVE_RESOLVER_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBS) $(LDLIBS_SECURE) -o $(BINDIR)/$(CONFIG)/resolve_address_using_native_resolver_test + +endif + +$(OBJDIR)/$(CONFIG)/test/core/iomgr/resolve_address_test.o: $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a + +deps_resolve_address_using_native_resolver_test: $(RESOLVE_ADDRESS_USING_NATIVE_RESOLVER_TEST_OBJS:.o=.dep) + +ifneq ($(NO_SECURE),true) +ifneq ($(NO_DEPS),true) +-include $(RESOLVE_ADDRESS_USING_NATIVE_RESOLVER_TEST_OBJS:.o=.dep) endif endif @@ -17076,6 +17133,49 @@ endif endif +CLIENT_CALLBACK_END2END_TEST_SRC = \ + test/cpp/end2end/client_callback_end2end_test.cc \ + +CLIENT_CALLBACK_END2END_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(CLIENT_CALLBACK_END2END_TEST_SRC)))) +ifeq ($(NO_SECURE),true) + +# You can't build secure targets if you don't have OpenSSL. + +$(BINDIR)/$(CONFIG)/client_callback_end2end_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.5.0+. + +$(BINDIR)/$(CONFIG)/client_callback_end2end_test: protobuf_dep_error + +else + +$(BINDIR)/$(CONFIG)/client_callback_end2end_test: $(PROTOBUF_DEP) $(CLIENT_CALLBACK_END2END_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.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) $(CLIENT_CALLBACK_END2END_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.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)/client_callback_end2end_test + +endif + +endif + +$(OBJDIR)/$(CONFIG)/test/cpp/end2end/client_callback_end2end_test.o: $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a + +deps_client_callback_end2end_test: $(CLIENT_CALLBACK_END2END_TEST_OBJS:.o=.dep) + +ifneq ($(NO_SECURE),true) +ifneq ($(NO_DEPS),true) +-include $(CLIENT_CALLBACK_END2END_TEST_OBJS:.o=.dep) +endif +endif + + CLIENT_CHANNEL_STRESS_TEST_SRC = \ $(GENDIR)/src/proto/grpc/lb/v1/load_balancer.pb.cc $(GENDIR)/src/proto/grpc/lb/v1/load_balancer.grpc.pb.cc \ test/cpp/client/client_channel_stress_test.cc \ diff --git a/build.yaml b/build.yaml index a50a0a4ab6..d6e67aa7ee 100644 --- a/build.yaml +++ b/build.yaml @@ -1169,7 +1169,9 @@ filegroups: - include/grpcpp/impl/codegen/byte_buffer.h - include/grpcpp/impl/codegen/call.h - include/grpcpp/impl/codegen/call_hook.h + - include/grpcpp/impl/codegen/callback_common.h - include/grpcpp/impl/codegen/channel_interface.h + - include/grpcpp/impl/codegen/client_callback.h - include/grpcpp/impl/codegen/client_context.h - include/grpcpp/impl/codegen/client_unary_call.h - include/grpcpp/impl/codegen/completion_queue.h @@ -1297,6 +1299,7 @@ filegroups: - include/grpcpp/support/async_unary_call.h - include/grpcpp/support/byte_buffer.h - include/grpcpp/support/channel_arguments.h + - include/grpcpp/support/client_callback.h - include/grpcpp/support/config.h - include/grpcpp/support/proto_buffer_reader.h - include/grpcpp/support/proto_buffer_writer.h @@ -1324,6 +1327,7 @@ filegroups: - src/cpp/client/credentials_cc.cc - src/cpp/client/generic_stub.cc - src/cpp/common/alarm.cc + - src/cpp/common/callback_common.cc - src/cpp/common/channel_arguments.cc - src/cpp/common/channel_filter.cc - src/cpp/common/completion_queue_cc.cc @@ -3363,7 +3367,7 @@ targets: - mac - linux - posix -- name: resolve_address_test +- name: resolve_address_using_ares_resolver_test build: test language: c src: @@ -3373,6 +3377,20 @@ targets: - grpc - gpr_test_util - gpr + args: + - --resolver=ares +- name: resolve_address_using_native_resolver_test + build: test + language: c + src: + - test/core/iomgr/resolve_address_test.cc + deps: + - grpc_test_util + - grpc + - gpr_test_util + - gpr + args: + - --resolver=native - name: resource_quota_test cpu_cost: 30 build: test @@ -4465,6 +4483,20 @@ targets: - grpc - gpr_test_util - gpr +- name: client_callback_end2end_test + gtest: true + cpu_cost: 0.5 + build: test + language: c++ + src: + - test/cpp/end2end/client_callback_end2end_test.cc + deps: + - grpc++_test_util + - grpc_test_util + - grpc++ + - grpc + - gpr_test_util + - gpr - name: client_channel_stress_test gtest: false build: test diff --git a/doc/interop-test-descriptions.md b/doc/interop-test-descriptions.md index 3c33189196..1d6535d7ea 100644 --- a/doc/interop-test-descriptions.md +++ b/doc/interop-test-descriptions.md @@ -944,6 +944,11 @@ the experimental flag, `soak_iterations`. This tests puts stress on several gRPC components; the resolver, the load balancer, and the RPC hotpath. +#### long_lived_channel + +The client performs a number of large_unary RPCs over a single long-lived +channel with a fixed but configurable interval between each RPC. + ### TODO Tests #### High priority: diff --git a/etc/roots.pem b/etc/roots.pem index c22dfe69f8..3e6bbcd76e 100644 --- a/etc/roots.pem +++ b/etc/roots.pem @@ -4317,3 +4317,26 @@ JJUEeKgDu+6B5dpffItKoZB0JaezPkvILFa9x8jvOOJckvB595yEunQtYQEgfn7R 8k8HWV+LLUNS60YMlOH1Zkd5d9VUWx+tJDfLRVpOoERIyNiwmcUVhAn21klJwGW4 5hpxbqCo8YLoRT5s1gLXCmeDBVrJpBA= -----END CERTIFICATE----- + +# Issuer: CN=OISTE WISeKey Global Root GC CA O=WISeKey OU=OISTE Foundation Endorsed +# Subject: CN=OISTE WISeKey Global Root GC CA O=WISeKey OU=OISTE Foundation Endorsed +# Label: "OISTE WISeKey Global Root GC CA" +# Serial: 44084345621038548146064804565436152554 +# MD5 Fingerprint: a9:d6:b9:2d:2f:93:64:f8:a5:69:ca:91:e9:68:07:23 +# SHA1 Fingerprint: e0:11:84:5e:34:de:be:88:81:b9:9c:f6:16:26:d1:96:1f:c3:b9:31 +# SHA256 Fingerprint: 85:60:f9:1c:36:24:da:ba:95:70:b5:fe:a0:db:e3:6f:f1:1a:83:23:be:94:86:85:4f:b3:f3:4a:55:71:19:8d +-----BEGIN CERTIFICATE----- +MIICaTCCAe+gAwIBAgIQISpWDK7aDKtARb8roi066jAKBggqhkjOPQQDAzBtMQsw +CQYDVQQGEwJDSDEQMA4GA1UEChMHV0lTZUtleTEiMCAGA1UECxMZT0lTVEUgRm91 +bmRhdGlvbiBFbmRvcnNlZDEoMCYGA1UEAxMfT0lTVEUgV0lTZUtleSBHbG9iYWwg +Um9vdCBHQyBDQTAeFw0xNzA1MDkwOTQ4MzRaFw00MjA1MDkwOTU4MzNaMG0xCzAJ +BgNVBAYTAkNIMRAwDgYDVQQKEwdXSVNlS2V5MSIwIAYDVQQLExlPSVNURSBGb3Vu +ZGF0aW9uIEVuZG9yc2VkMSgwJgYDVQQDEx9PSVNURSBXSVNlS2V5IEdsb2JhbCBS +b290IEdDIENBMHYwEAYHKoZIzj0CAQYFK4EEACIDYgAETOlQwMYPchi82PG6s4ni +eUqjFqdrVCTbUf/q9Akkwwsin8tqJ4KBDdLArzHkdIJuyiXZjHWd8dvQmqJLIX4W +p2OQ0jnUsYd4XxiWD1AbNTcPasbc2RNNpI6QN+a9WzGRo1QwUjAOBgNVHQ8BAf8E +BAMCAQYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4EFgQUSIcUrOPDnpBgOtfKie7T +rYy0UGYwEAYJKwYBBAGCNxUBBAMCAQAwCgYIKoZIzj0EAwMDaAAwZQIwJsdpW9zV +57LnyAyMjMPdeYwbY9XJUpROTYJKcx6ygISpJcBMWm1JKWB4E+J+SOtkAjEA2zQg +Mgj/mkkCtojeFK9dbJlxjRo/i9fgojaGHAeCOnZT/cKi7e97sIBPWA9LUzm9 +-----END CERTIFICATE----- diff --git a/gRPC-C++.podspec b/gRPC-C++.podspec index 581b9246bc..03ec223279 100644 --- a/gRPC-C++.podspec +++ b/gRPC-C++.podspec @@ -111,6 +111,7 @@ Pod::Spec.new do |s| 'include/grpcpp/support/async_unary_call.h', 'include/grpcpp/support/byte_buffer.h', 'include/grpcpp/support/channel_arguments.h', + 'include/grpcpp/support/client_callback.h', 'include/grpcpp/support/config.h', 'include/grpcpp/support/proto_buffer_reader.h', 'include/grpcpp/support/proto_buffer_writer.h', @@ -127,7 +128,9 @@ Pod::Spec.new do |s| 'include/grpcpp/impl/codegen/byte_buffer.h', 'include/grpcpp/impl/codegen/call.h', 'include/grpcpp/impl/codegen/call_hook.h', + 'include/grpcpp/impl/codegen/callback_common.h', 'include/grpcpp/impl/codegen/channel_interface.h', + 'include/grpcpp/impl/codegen/client_callback.h', 'include/grpcpp/impl/codegen/client_context.h', 'include/grpcpp/impl/codegen/client_unary_call.h', 'include/grpcpp/impl/codegen/completion_queue.h', @@ -187,6 +190,7 @@ Pod::Spec.new do |s| 'src/cpp/client/credentials_cc.cc', 'src/cpp/client/generic_stub.cc', 'src/cpp/common/alarm.cc', + 'src/cpp/common/callback_common.cc', 'src/cpp/common/channel_arguments.cc', 'src/cpp/common/channel_filter.cc', 'src/cpp/common/completion_queue_cc.cc', @@ -74,7 +74,9 @@ EXPORTS grpc_resource_quota_set_max_threads grpc_resource_quota_arg_vtable grpc_channelz_get_top_channels + grpc_channelz_get_servers grpc_channelz_get_channel + grpc_channelz_get_subchannel grpc_insecure_channel_create_from_fd grpc_server_add_insecure_channel_from_fd grpc_use_signal @@ -1381,6 +1381,7 @@ 'src/cpp/client/credentials_cc.cc', 'src/cpp/client/generic_stub.cc', 'src/cpp/common/alarm.cc', + 'src/cpp/common/callback_common.cc', 'src/cpp/common/channel_arguments.cc', 'src/cpp/common/channel_filter.cc', 'src/cpp/common/completion_queue_cc.cc', @@ -1528,6 +1529,7 @@ 'src/cpp/client/credentials_cc.cc', 'src/cpp/client/generic_stub.cc', 'src/cpp/common/alarm.cc', + 'src/cpp/common/callback_common.cc', 'src/cpp/common/channel_arguments.cc', 'src/cpp/common/channel_filter.cc', 'src/cpp/common/completion_queue_cc.cc', diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index 4e50cd0bac..3ef95ff462 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -499,10 +499,17 @@ GRPCAPI const grpc_arg_pointer_vtable* grpc_resource_quota_arg_vtable(void); The returned string is allocated and must be freed by the application. */ GRPCAPI char* grpc_channelz_get_top_channels(intptr_t start_channel_id); +/* Gets all servers that exist in the process. */ +GRPCAPI char* grpc_channelz_get_servers(intptr_t start_server_id); + /* Returns a single Channel, or else a NOT_FOUND code. The returned string is allocated and must be freed by the application. */ GRPCAPI char* grpc_channelz_get_channel(intptr_t channel_id); +/* Returns a single Subchannel, or else a NOT_FOUND code. The returned string + is allocated and must be freed by the application. */ +GRPCAPI char* grpc_channelz_get_subchannel(intptr_t subchannel_id); + #ifdef __cplusplus } #endif diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index b5353c1dea..5f3b96f40b 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -342,6 +342,9 @@ typedef struct { "grpc.disable_client_authority_filter" /** If set to zero, disables use of http proxies. Enabled by default. */ #define GRPC_ARG_ENABLE_HTTP_PROXY "grpc.enable_http_proxy" +/** If set to non zero, surfaces the user agent string to the server. User + agent is surfaced by default. */ +#define GRPC_ARG_SURFACE_USER_AGENT "grpc.surface_user_agent" /** \} */ /** Result of a grpc call. If the caller satisfies the prerequisites of a diff --git a/include/grpc/impl/codegen/port_platform.h b/include/grpc/impl/codegen/port_platform.h index 8d9bd83285..b2028a6305 100644 --- a/include/grpc/impl/codegen/port_platform.h +++ b/include/grpc/impl/codegen/port_platform.h @@ -174,6 +174,7 @@ #ifdef __GLIBC__ #define GPR_POSIX_CRASH_HANDLER 1 #define GPR_LINUX_PTHREAD_NAME 1 +#include <linux/version.h> #else /* musl libc */ #define GPR_MUSL_LIBC_COMPAT 1 #endif diff --git a/include/grpcpp/channel.h b/include/grpcpp/channel.h index fed02bf7bc..f1dba5b8ad 100644 --- a/include/grpcpp/channel.h +++ b/include/grpcpp/channel.h @@ -78,8 +78,19 @@ class Channel final : public ChannelInterface, bool WaitForStateChangeImpl(grpc_connectivity_state last_observed, gpr_timespec deadline) override; + CompletionQueue* CallbackCQ() override; + const grpc::string host_; grpc_channel* const c_channel_; // owned + + // mu_ protects callback_cq_ (the per-channel callbackable completion queue) + std::mutex mu_; + + // callback_cq_ references the callbackable completion queue associated + // with this channel (if any). It is set on the first call to CallbackCQ(). + // It is _not owned_ by the channel; ownership belongs with its internal + // shutdown callback tag (invoked when the CQ is fully shutdown). + CompletionQueue* callback_cq_ = nullptr; }; } // namespace grpc diff --git a/include/grpcpp/generic/generic_stub.h b/include/grpcpp/generic/generic_stub.h index 92405a43fa..d509d9a520 100644 --- a/include/grpcpp/generic/generic_stub.h +++ b/include/grpcpp/generic/generic_stub.h @@ -19,9 +19,12 @@ #ifndef GRPCPP_GENERIC_GENERIC_STUB_H #define GRPCPP_GENERIC_GENERIC_STUB_H +#include <functional> + #include <grpcpp/support/async_stream.h> #include <grpcpp/support/async_unary_call.h> #include <grpcpp/support/byte_buffer.h> +#include <grpcpp/support/status.h> namespace grpc { @@ -62,6 +65,26 @@ class GenericStub final { ClientContext* context, const grpc::string& method, CompletionQueue* cq, void* tag); + /// NOTE: class experimental_type is not part of the public API of this class + /// TODO(vjpai): Move these contents to the public API of GenericStub when + /// they are no longer experimental + class experimental_type { + public: + explicit experimental_type(GenericStub* stub) : stub_(stub) {} + + void UnaryCall(ClientContext* context, const grpc::string& method, + const ByteBuffer* request, ByteBuffer* response, + std::function<void(Status)> on_completion); + + private: + GenericStub* stub_; + }; + + /// NOTE: The function experimental() is not stable public API. It is a view + /// to the experimental components of this class. It may be changed or removed + /// at any time. + experimental_type experimental() { return experimental_type(this); } + private: std::shared_ptr<ChannelInterface> channel_; }; diff --git a/include/grpcpp/impl/codegen/async_stream.h b/include/grpcpp/impl/codegen/async_stream.h index b2134590c3..b306cd3604 100644 --- a/include/grpcpp/impl/codegen/async_stream.h +++ b/include/grpcpp/impl/codegen/async_stream.h @@ -195,6 +195,13 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { assert(size == sizeof(ClientAsyncReader)); } + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + void StartCall(void* tag) override { assert(!started_); started_ = true; @@ -336,6 +343,13 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { assert(size == sizeof(ClientAsyncWriter)); } + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + void StartCall(void* tag) override { assert(!started_); started_ = true; @@ -496,6 +510,13 @@ class ClientAsyncReaderWriter final assert(size == sizeof(ClientAsyncReaderWriter)); } + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + void StartCall(void* tag) override { assert(!started_); started_ = true; diff --git a/include/grpcpp/impl/codegen/call.h b/include/grpcpp/impl/codegen/call.h index e94adada94..7cadea0055 100644 --- a/include/grpcpp/impl/codegen/call.h +++ b/include/grpcpp/impl/codegen/call.h @@ -599,6 +599,11 @@ class CallOpSetInterface : public CompletionQueueTag { /// Fills in grpc_op, starting from ops[*nops] and moving /// upwards. virtual void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) = 0; + + /// Get the tag to be used at the core completion queue. Generally, the + /// value of cq_tag will be "this". However, it can be overridden if we + /// want core to process the tag differently (e.g., as a core callback) + virtual void* cq_tag() = 0; }; /// Primary implementation of CallOpSetInterface. @@ -618,7 +623,7 @@ class CallOpSet : public CallOpSetInterface, public Op5, public Op6 { public: - CallOpSet() : return_tag_(this), call_(nullptr) {} + CallOpSet() : cq_tag_(this), return_tag_(this), call_(nullptr) {} void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) override { this->Op1::AddOp(ops, nops); this->Op2::AddOp(ops, nops); @@ -645,7 +650,16 @@ class CallOpSet : public CallOpSetInterface, void set_output_tag(void* return_tag) { return_tag_ = return_tag; } + void* cq_tag() override { return cq_tag_; } + + /// set_cq_tag is used to provide a different core CQ tag than "this". + /// This is used for callback-based tags, where the core tag is the core + /// callback function. It does not change the use or behavior of any other + /// function (such as FinalizeResult) + void set_cq_tag(void* cq_tag) { cq_tag_ = cq_tag; } + private: + void* cq_tag_; void* return_tag_; grpc_call* call_; }; diff --git a/include/grpcpp/impl/codegen/callback_common.h b/include/grpcpp/impl/codegen/callback_common.h new file mode 100644 index 0000000000..68c318d2b4 --- /dev/null +++ b/include/grpcpp/impl/codegen/callback_common.h @@ -0,0 +1,103 @@ +/* + * + * 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. + * + */ + +#ifndef GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H +#define GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H + +#include <functional> + +#include <grpcpp/impl/codegen/call.h> +#include <grpcpp/impl/codegen/channel_interface.h> +#include <grpcpp/impl/codegen/config.h> +#include <grpcpp/impl/codegen/core_codegen_interface.h> +#include <grpcpp/impl/codegen/status.h> + +// Forward declarations +namespace grpc_core { +class CQCallbackInterface; +}; + +namespace grpc { +namespace internal { + +class CallbackWithStatusTag { + public: + // always allocated against a call arena, no memory free required + static void operator delete(void* ptr, std::size_t size) { + assert(size == sizeof(CallbackWithStatusTag)); + } + + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + + CallbackWithStatusTag(grpc_call* call, std::function<void(Status)> f, + CompletionQueueTag* ops); + ~CallbackWithStatusTag() {} + void* tag() { return static_cast<void*>(impl_); } + Status* status_ptr() { return status_; } + CompletionQueueTag* ops() { return ops_; } + + // force_run can not be performed on a tag if operations using this tag + // have been sent to PerformOpsOnCall. It is intended for error conditions + // that are detected before the operations are internally processed. + void force_run(Status s); + + private: + grpc_core::CQCallbackInterface* impl_; + Status* status_; + CompletionQueueTag* ops_; +}; + +class CallbackWithSuccessTag { + public: + // always allocated against a call arena, no memory free required + static void operator delete(void* ptr, std::size_t size) { + assert(size == sizeof(CallbackWithSuccessTag)); + } + + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + + CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f, + CompletionQueueTag* ops); + + void* tag() { return static_cast<void*>(impl_); } + CompletionQueueTag* ops() { return ops_; } + + // force_run can not be performed on a tag if operations using this tag + // have been sent to PerformOpsOnCall. It is intended for error conditions + // that are detected before the operations are internally processed. + void force_run(bool ok); + + private: + grpc_core::CQCallbackInterface* impl_; + CompletionQueueTag* ops_; +}; + +} // namespace internal +} // namespace grpc + +#endif // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H diff --git a/include/grpcpp/impl/codegen/channel_interface.h b/include/grpcpp/impl/codegen/channel_interface.h index ec1c6c25d8..b257acc1ab 100644 --- a/include/grpcpp/impl/codegen/channel_interface.h +++ b/include/grpcpp/impl/codegen/channel_interface.h @@ -41,6 +41,8 @@ class CallOpSetInterface; class RpcMethod; template <class InputMessage, class OutputMessage> class BlockingUnaryCallImpl; +template <class InputMessage, class OutputMessage> +class CallbackUnaryCallImpl; template <class R> class ClientAsyncReaderFactory; template <class W> @@ -103,6 +105,8 @@ class ChannelInterface { friend class ::grpc::internal::ClientAsyncResponseReaderFactory; template <class InputMessage, class OutputMessage> friend class ::grpc::internal::BlockingUnaryCallImpl; + template <class InputMessage, class OutputMessage> + friend class ::grpc::internal::CallbackUnaryCallImpl; friend class ::grpc::internal::RpcMethod; virtual internal::Call CreateCall(const internal::RpcMethod& method, ClientContext* context, @@ -115,6 +119,16 @@ class ChannelInterface { CompletionQueue* cq, void* tag) = 0; virtual bool WaitForStateChangeImpl(grpc_connectivity_state last_observed, gpr_timespec deadline) = 0; + + // EXPERIMENTAL + // A method to get the callbackable completion queue associated with this + // channel. If the return value is nullptr, this channel doesn't support + // callback operations. + // TODO(vjpai): Consider a better default like using a global CQ + // Returns nullptr (rather than being pure) since this is a new method + // and adding a new pure method to an interface would be a breaking change + // (even though this is private and non-API) + virtual CompletionQueue* CallbackCQ() { return nullptr; } }; } // namespace grpc diff --git a/include/grpcpp/impl/codegen/client_callback.h b/include/grpcpp/impl/codegen/client_callback.h new file mode 100644 index 0000000000..fc81c8aa0a --- /dev/null +++ b/include/grpcpp/impl/codegen/client_callback.h @@ -0,0 +1,95 @@ +/* + * + * 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. + * + */ + +#ifndef GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H +#define GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H + +#include <functional> + +#include <grpcpp/impl/codegen/call.h> +#include <grpcpp/impl/codegen/callback_common.h> +#include <grpcpp/impl/codegen/channel_interface.h> +#include <grpcpp/impl/codegen/config.h> +#include <grpcpp/impl/codegen/core_codegen_interface.h> +#include <grpcpp/impl/codegen/status.h> + +namespace grpc { + +class Channel; +class ClientContext; +class CompletionQueue; + +namespace internal { +class RpcMethod; + +/// Perform a callback-based unary call +/// TODO(vjpai): Combine as much as possible with the blocking unary call code +template <class InputMessage, class OutputMessage> +void CallbackUnaryCall(ChannelInterface* channel, const RpcMethod& method, + ClientContext* context, const InputMessage* request, + OutputMessage* result, + std::function<void(Status)> on_completion) { + CallbackUnaryCallImpl<InputMessage, OutputMessage> x( + channel, method, context, request, result, on_completion); +} + +template <class InputMessage, class OutputMessage> +class CallbackUnaryCallImpl { + public: + CallbackUnaryCallImpl(ChannelInterface* channel, const RpcMethod& method, + ClientContext* context, const InputMessage* request, + OutputMessage* result, + std::function<void(Status)> on_completion) { + CompletionQueue* cq = channel->CallbackCQ(); + GPR_CODEGEN_ASSERT(cq != nullptr); + Call call(channel->CreateCall(method, context, cq)); + + using FullCallOpSet = + CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, + CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>, + CallOpClientSendClose, CallOpClientRecvStatus>; + + auto* ops = new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(FullCallOpSet))) FullCallOpSet; + + auto* tag = new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(CallbackWithStatusTag))) + CallbackWithStatusTag(call.call(), on_completion, ops); + + // TODO(vjpai): Unify code with sync API as much as possible + Status s = ops->SendMessage(*request); + if (!s.ok()) { + tag->force_run(s); + return; + } + ops->SendInitialMetadata(context->send_initial_metadata_, + context->initial_metadata_flags()); + ops->RecvInitialMetadata(context); + ops->RecvMessage(result); + ops->AllowNoMessage(); + ops->ClientSendClose(); + ops->ClientRecvStatus(context, tag->status_ptr()); + ops->set_cq_tag(tag->tag()); + call.PerformOps(ops); + } +}; + +} // namespace internal +} // namespace grpc + +#endif // GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H diff --git a/include/grpcpp/impl/codegen/client_context.h b/include/grpcpp/impl/codegen/client_context.h index c6c9540950..46635a541a 100644 --- a/include/grpcpp/impl/codegen/client_context.h +++ b/include/grpcpp/impl/codegen/client_context.h @@ -68,6 +68,8 @@ class CallOpClientRecvStatus; class CallOpRecvInitialMetadata; template <class InputMessage, class OutputMessage> class BlockingUnaryCallImpl; +template <class InputMessage, class OutputMessage> +class CallbackUnaryCallImpl; } // namespace internal template <class R> @@ -389,6 +391,8 @@ class ClientContext { friend class ::grpc::ClientAsyncResponseReader; template <class InputMessage, class OutputMessage> friend class ::grpc::internal::BlockingUnaryCallImpl; + template <class InputMessage, class OutputMessage> + friend class ::grpc::internal::CallbackUnaryCallImpl; // Used by friend class CallOpClientRecvStatus void set_debug_error_string(const grpc::string& debug_error_string) { diff --git a/include/grpcpp/impl/codegen/completion_queue.h b/include/grpcpp/impl/codegen/completion_queue.h index 3f7d4fb765..f52f9a53be 100644 --- a/include/grpcpp/impl/codegen/completion_queue.h +++ b/include/grpcpp/impl/codegen/completion_queue.h @@ -274,6 +274,9 @@ class CompletionQueue : private GrpcLibraryCodegen { template <class InputMessage, class OutputMessage> friend class ::grpc::internal::BlockingUnaryCallImpl; + // Friends that need access to constructor for callback CQ + friend class ::grpc::Channel; + /// EXPERIMENTAL /// Creates a Thread Local cache to store the first event /// On this completion queue queued from this thread. Once diff --git a/include/grpcpp/impl/codegen/completion_queue_tag.h b/include/grpcpp/impl/codegen/completion_queue_tag.h index ffb642c56b..304386a9ec 100644 --- a/include/grpcpp/impl/codegen/completion_queue_tag.h +++ b/include/grpcpp/impl/codegen/completion_queue_tag.h @@ -26,10 +26,25 @@ namespace internal { class CompletionQueueTag { public: virtual ~CompletionQueueTag() {} - /// Called prior to returning from Next(), return value is the status of the - /// operation (return status is the default thing to do). If this function - /// returns false, the tag is dropped and not returned from the completion - /// queue + + /// FinalizeResult must be called before informing user code that the + /// operation bound to the underlying core completion queue tag has + /// completed. In practice, this means: + /// + /// 1. For the sync API - before returning from Pluck + /// 2. For the CQ-based async API - before returning from Next + /// 3. For the callback-based API - before invoking the user callback + /// + /// This is the method that translates from core-side tag/status to + /// C++ API-observable tag/status. + /// + /// The return value is the status of the operation (returning status is the + /// general behavior of this function). If this function returns false, the + /// tag is dropped and not returned from the completion queue: this concept is + /// for events that are observed at core but not requested by the user + /// application (e.g., server shutdown, for server unimplemented method + /// responses, or for cases where a server-side RPC doesn't have a completion + /// notification registered using AsyncNotifyWhenDone) virtual bool FinalizeResult(void** tag, bool* status) = 0; }; } // namespace internal diff --git a/include/grpcpp/support/client_callback.h b/include/grpcpp/support/client_callback.h new file mode 100644 index 0000000000..063fdc4f85 --- /dev/null +++ b/include/grpcpp/support/client_callback.h @@ -0,0 +1,24 @@ +/* + * + * 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. + * + */ + +#ifndef GRPCPP_SUPPORT_CLIENT_CALLBACK_H +#define GRPCPP_SUPPORT_CLIENT_CALLBACK_H + +#include <grpcpp/impl/codegen/client_callback.h> + +#endif // GRPCPP_SUPPORT_CLIENT_CALLBACK_H diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index 8044b4e4cd..388736b60a 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -933,6 +933,11 @@ typedef struct client_channel_call_data { grpc_closure pick_closure; grpc_closure pick_cancel_closure; + // state needed to support channelz interception of recv trailing metadata. + grpc_closure recv_trailing_metadata_ready_channelz; + grpc_closure* original_recv_trailing_metadata; + grpc_metadata_batch* recv_trailing_metadata; + grpc_polling_entity* pollent; bool pollent_added_to_interested_parties; @@ -994,6 +999,8 @@ static void start_internal_recv_trailing_metadata(grpc_call_element* elem); static void on_complete(void* arg, grpc_error* error); static void start_retriable_subchannel_batches(void* arg, grpc_error* ignored); static void start_pick_locked(void* arg, grpc_error* ignored); +static void maybe_intercept_recv_trailing_metadata_for_channelz( + grpc_call_element* elem, grpc_transport_stream_op_batch* batch); // // send op data caching @@ -1292,6 +1299,7 @@ static void pending_batches_resume(grpc_call_element* elem) { pending_batch* pending = &calld->pending_batches[i]; grpc_transport_stream_op_batch* batch = pending->batch; if (batch != nullptr) { + maybe_intercept_recv_trailing_metadata_for_channelz(elem, batch); batch->handler_private.extra_arg = calld->subchannel_call; GRPC_CLOSURE_INIT(&batch->handler_private.closure, resume_pending_batch_in_call_combiner, batch, @@ -1777,23 +1785,22 @@ static void recv_message_ready(void* arg, grpc_error* error) { // recv_trailing_metadata handling // -// Sets *status and *server_pushback_md based on batch_data and error. -static void get_call_status(subchannel_batch_data* batch_data, - grpc_error* error, grpc_status_code* status, +// Sets *status and *server_pushback_md based on md_batch and error. +// Only sets *server_pushback_md if server_pushback_md != nullptr. +static void get_call_status(grpc_call_element* elem, + grpc_metadata_batch* md_batch, grpc_error* error, + grpc_status_code* status, grpc_mdelem** server_pushback_md) { - grpc_call_element* elem = batch_data->elem; call_data* calld = static_cast<call_data*>(elem->call_data); if (error != GRPC_ERROR_NONE) { grpc_error_get_status(error, calld->deadline, status, nullptr, nullptr, nullptr); } else { - grpc_metadata_batch* md_batch = - batch_data->batch.payload->recv_trailing_metadata - .recv_trailing_metadata; GPR_ASSERT(md_batch->idx.named.grpc_status != nullptr); *status = grpc_get_status_code_from_metadata(md_batch->idx.named.grpc_status->md); - if (md_batch->idx.named.grpc_retry_pushback_ms != nullptr) { + if (server_pushback_md != nullptr && + md_batch->idx.named.grpc_retry_pushback_ms != nullptr) { *server_pushback_md = &md_batch->idx.named.grpc_retry_pushback_ms->md; } } @@ -1966,8 +1973,19 @@ static void recv_trailing_metadata_ready(void* arg, grpc_error* error) { // Get the call's status and check for server pushback metadata. grpc_status_code status = GRPC_STATUS_OK; grpc_mdelem* server_pushback_md = nullptr; - get_call_status(batch_data, GRPC_ERROR_REF(error), &status, + grpc_metadata_batch* md_batch = + batch_data->batch.payload->recv_trailing_metadata.recv_trailing_metadata; + get_call_status(elem, md_batch, GRPC_ERROR_REF(error), &status, &server_pushback_md); + grpc_core::channelz::SubchannelNode* channelz_subchannel = + calld->pick.connected_subchannel->channelz_subchannel(); + if (channelz_subchannel != nullptr) { + if (status == GRPC_STATUS_OK) { + channelz_subchannel->RecordCallSucceeded(); + } else { + channelz_subchannel->RecordCallFailed(); + } + } if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_INFO, "chand=%p calld=%p: call finished, status=%s", chand, calld, grpc_status_code_to_string(status)); @@ -2572,6 +2590,69 @@ static void start_retriable_subchannel_batches(void* arg, grpc_error* ignored) { } // +// Channelz +// + +static void recv_trailing_metadata_ready_channelz(void* arg, + grpc_error* error) { + grpc_call_element* elem = static_cast<grpc_call_element*>(arg); + channel_data* chand = static_cast<channel_data*>(elem->channel_data); + call_data* calld = static_cast<call_data*>(elem->call_data); + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "chand=%p calld=%p: got recv_trailing_metadata_ready_channelz, " + "error=%s", + chand, calld, grpc_error_string(error)); + } + GPR_ASSERT(calld->recv_trailing_metadata != nullptr); + grpc_status_code status = GRPC_STATUS_OK; + grpc_metadata_batch* md_batch = calld->recv_trailing_metadata; + get_call_status(elem, md_batch, GRPC_ERROR_REF(error), &status, nullptr); + grpc_core::channelz::SubchannelNode* channelz_subchannel = + calld->pick.connected_subchannel->channelz_subchannel(); + GPR_ASSERT(channelz_subchannel != nullptr); + if (status == GRPC_STATUS_OK) { + channelz_subchannel->RecordCallSucceeded(); + } else { + channelz_subchannel->RecordCallFailed(); + } + calld->recv_trailing_metadata = nullptr; + GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata, error); +} + +// If channelz is enabled, intercept recv_trailing so that we may check the +// status and associate it to a subchannel. +// Returns true if callback was intercepted, false otherwise. +static void maybe_intercept_recv_trailing_metadata_for_channelz( + grpc_call_element* elem, grpc_transport_stream_op_batch* batch) { + call_data* calld = static_cast<call_data*>(elem->call_data); + // only intercept payloads with recv trailing. + if (!batch->recv_trailing_metadata) { + return; + } + // only add interceptor is channelz is enabled. + if (calld->pick.connected_subchannel->channelz_subchannel() == nullptr) { + return; + } + if (grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, + "calld=%p batch=%p: intercepting recv trailing for channelz", calld, + batch); + } + GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_ready_channelz, + recv_trailing_metadata_ready_channelz, elem, + grpc_schedule_on_exec_ctx); + // save some state needed for the interception callback. + GPR_ASSERT(calld->recv_trailing_metadata == nullptr); + calld->recv_trailing_metadata = + batch->payload->recv_trailing_metadata.recv_trailing_metadata; + calld->original_recv_trailing_metadata = + batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready; + batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready = + &calld->recv_trailing_metadata_ready_channelz; +} + +// // LB pick // @@ -2600,6 +2681,11 @@ static void create_subchannel_call(grpc_call_element* elem, grpc_error* error) { new_error = grpc_error_add_child(new_error, error); pending_batches_fail(elem, new_error, true /* yield_call_combiner */); } else { + grpc_core::channelz::SubchannelNode* channelz_subchannel = + calld->pick.connected_subchannel->channelz_subchannel(); + if (channelz_subchannel != nullptr) { + channelz_subchannel->RecordCallStarted(); + } if (parent_data_size > 0) { subchannel_call_retry_state* retry_state = static_cast<subchannel_call_retry_state*>( diff --git a/src/core/ext/filters/client_channel/client_channel_channelz.cc b/src/core/ext/filters/client_channel/client_channel_channelz.cc index 86c765df52..7e8f59bcd3 100644 --- a/src/core/ext/filters/client_channel/client_channel_channelz.cc +++ b/src/core/ext/filters/client_channel/client_channel_channelz.cc @@ -20,10 +20,13 @@ #include "src/core/ext/filters/client_channel/client_channel.h" #include "src/core/ext/filters/client_channel/client_channel_channelz.h" +#include "src/core/lib/channel/channelz_registry.h" #include "src/core/lib/gpr/useful.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/transport/connectivity_state.h" +#include <grpc/support/string_util.h> + namespace grpc_core { namespace channelz { namespace { @@ -109,5 +112,62 @@ RefCountedPtr<ChannelNode> ClientChannelNode::MakeClientChannelNode( is_top_level_channel); } +SubchannelNode::SubchannelNode(grpc_subchannel* subchannel, + size_t channel_tracer_max_nodes) + : BaseNode(EntityType::kSubchannel), + subchannel_(subchannel), + target_( + UniquePtr<char>(gpr_strdup(grpc_subchannel_get_target(subchannel_)))), + trace_(channel_tracer_max_nodes) {} + +SubchannelNode::~SubchannelNode() {} + +void SubchannelNode::PopulateConnectivityState(grpc_json* json) { + grpc_connectivity_state state; + if (subchannel_ == nullptr) { + state = GRPC_CHANNEL_SHUTDOWN; + } else { + state = grpc_subchannel_check_connectivity(subchannel_, nullptr); + } + json = grpc_json_create_child(nullptr, json, "state", nullptr, + GRPC_JSON_OBJECT, false); + grpc_json_create_child(nullptr, json, "state", + grpc_connectivity_state_name(state), GRPC_JSON_STRING, + false); +} + +grpc_json* SubchannelNode::RenderJson() { + grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); + grpc_json* json = top_level_json; + grpc_json* json_iterator = nullptr; + json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr, + GRPC_JSON_OBJECT, false); + json = json_iterator; + json_iterator = nullptr; + json_iterator = grpc_json_add_number_string_child(json, json_iterator, + "subchannelId", uuid()); + // reset json iterators to top level object + json = top_level_json; + json_iterator = nullptr; + // create and fill the data child. + grpc_json* data = grpc_json_create_child(json_iterator, json, "data", nullptr, + GRPC_JSON_OBJECT, false); + json = data; + json_iterator = nullptr; + PopulateConnectivityState(json); + GPR_ASSERT(target_.get() != nullptr); + grpc_json_create_child(nullptr, json, "target", target_.get(), + GRPC_JSON_STRING, false); + // fill in the channel trace if applicable + grpc_json* trace_json = trace_.RenderJson(); + if (trace_json != nullptr) { + trace_json->key = "trace"; // this object is named trace in channelz.proto + grpc_json_link_child(json, trace_json, nullptr); + } + // ask CallCountingHelper to populate trace and call count data. + call_counter_.PopulateCallCounts(json); + return top_level_json; +} + } // namespace channelz } // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/client_channel_channelz.h b/src/core/ext/filters/client_channel/client_channel_channelz.h index 6f27b5c8b7..8ce331e529 100644 --- a/src/core/ext/filters/client_channel/client_channel_channelz.h +++ b/src/core/ext/filters/client_channel/client_channel_channelz.h @@ -23,9 +23,12 @@ #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/channel/channel_trace.h" #include "src/core/lib/channel/channelz.h" #include "src/core/lib/gprpp/inlined_vector.h" +typedef struct grpc_subchannel grpc_subchannel; + namespace grpc_core { // TODO(ncteisen), this only contains the uuids of the children for now, @@ -43,28 +46,59 @@ class ClientChannelNode : public ChannelNode { grpc_channel* channel, size_t channel_tracer_max_nodes, bool is_top_level_channel); - // Override this functionality since client_channels have a notion of - // channel connectivity. - void PopulateConnectivityState(grpc_json* json) override; + ClientChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes, + bool is_top_level_channel); + virtual ~ClientChannelNode() {} - // Override this functionality since client_channels have subchannels + // Overriding template methods from ChannelNode to render information that + // only ClientChannelNode knows about. + void PopulateConnectivityState(grpc_json* json) override; void PopulateChildRefs(grpc_json* json) override; // Helper to create a channel arg to ensure this type of ChannelNode is // created. static grpc_arg CreateChannelArg(); - protected: - GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_DELETE - GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_NEW - ClientChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes, - bool is_top_level_channel); - virtual ~ClientChannelNode() {} - private: grpc_channel_element* client_channel_; }; +// Handles channelz bookkeeping for sockets +class SubchannelNode : public BaseNode { + public: + SubchannelNode(grpc_subchannel* subchannel, size_t channel_tracer_max_nodes); + ~SubchannelNode() override; + + void MarkSubchannelDestroyed() { + GPR_ASSERT(subchannel_ != nullptr); + subchannel_ = nullptr; + } + + grpc_json* RenderJson() override; + + // proxy methods to composed classes. + void AddTraceEvent(ChannelTrace::Severity severity, grpc_slice data) { + trace_.AddTraceEvent(severity, data); + } + void AddTraceEventWithReference(ChannelTrace::Severity severity, + grpc_slice data, + RefCountedPtr<BaseNode> referenced_channel) { + trace_.AddTraceEventWithReference(severity, data, + std::move(referenced_channel)); + } + void RecordCallStarted() { call_counter_.RecordCallStarted(); } + void RecordCallFailed() { call_counter_.RecordCallFailed(); } + void RecordCallSucceeded() { call_counter_.RecordCallSucceeded(); } + + private: + grpc_subchannel* subchannel_; + UniquePtr<char> target_; + CallCountingHelper call_counter_; + ChannelTrace trace_; + + void PopulateConnectivityState(grpc_json* json); +}; + } // namespace channelz } // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc index 25b0149393..1ee1925a25 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc @@ -1265,7 +1265,7 @@ void GrpcLb::FillChildRefsForChannelz(ChildRefsList* child_subchannels, grpc_core::channelz::ChannelNode* channel_node = grpc_channel_get_channelz_node(lb_channel_); if (channel_node != nullptr) { - child_channels->push_back(channel_node->channel_uuid()); + child_channels->push_back(channel_node->uuid()); } } } diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc index 602d6e92f9..ed8cc60ea1 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc @@ -71,11 +71,12 @@ class PickFirst : public LoadBalancingPolicy { : public SubchannelData<PickFirstSubchannelList, PickFirstSubchannelData> { public: - PickFirstSubchannelData(PickFirstSubchannelList* subchannel_list, - const grpc_lb_user_data_vtable* user_data_vtable, - const grpc_lb_address& address, - grpc_subchannel* subchannel, - grpc_combiner* combiner) + PickFirstSubchannelData( + SubchannelList<PickFirstSubchannelList, PickFirstSubchannelData>* + subchannel_list, + const grpc_lb_user_data_vtable* user_data_vtable, + const grpc_lb_address& address, grpc_subchannel* subchannel, + grpc_combiner* combiner) : SubchannelData(subchannel_list, user_data_vtable, address, subchannel, combiner) {} diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc index 4195c1e9d1..8dd5820bae 100644 --- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc +++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc @@ -89,11 +89,12 @@ class RoundRobin : public LoadBalancingPolicy { : public SubchannelData<RoundRobinSubchannelList, RoundRobinSubchannelData> { public: - RoundRobinSubchannelData(RoundRobinSubchannelList* subchannel_list, - const grpc_lb_user_data_vtable* user_data_vtable, - const grpc_lb_address& address, - grpc_subchannel* subchannel, - grpc_combiner* combiner) + RoundRobinSubchannelData( + SubchannelList<RoundRobinSubchannelList, RoundRobinSubchannelData>* + subchannel_list, + const grpc_lb_user_data_vtable* user_data_vtable, + const grpc_lb_address& address, grpc_subchannel* subchannel, + grpc_combiner* combiner) : SubchannelData(subchannel_list, user_data_vtable, address, subchannel, combiner), user_data_vtable_(user_data_vtable), diff --git a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h index 91ddaec8b8..5e8682e056 100644 --- a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h +++ b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h @@ -65,6 +65,10 @@ class MySubchannelList namespace grpc_core { +// Forward declaration. +template <typename SubchannelListType, typename SubchannelDataType> +class SubchannelList; + // Stores data for a particular subchannel in a subchannel list. // Callers must create a subclass that implements the // ProcessConnectivityChangeLocked() method. @@ -72,7 +76,9 @@ template <typename SubchannelListType, typename SubchannelDataType> class SubchannelData { public: // Returns a pointer to the subchannel list containing this object. - SubchannelListType* subchannel_list() const { return subchannel_list_; } + SubchannelListType* subchannel_list() const { + return static_cast<SubchannelListType*>(subchannel_list_); + } // Returns the index into the subchannel list of this object. size_t Index() const { @@ -133,10 +139,11 @@ class SubchannelData { GRPC_ABSTRACT_BASE_CLASS protected: - SubchannelData(SubchannelListType* subchannel_list, - const grpc_lb_user_data_vtable* user_data_vtable, - const grpc_lb_address& address, grpc_subchannel* subchannel, - grpc_combiner* combiner); + SubchannelData( + SubchannelList<SubchannelListType, SubchannelDataType>* subchannel_list, + const grpc_lb_user_data_vtable* user_data_vtable, + const grpc_lb_address& address, grpc_subchannel* subchannel, + grpc_combiner* combiner); virtual ~SubchannelData(); @@ -161,7 +168,7 @@ class SubchannelData { static void OnConnectivityChangedLocked(void* arg, grpc_error* error); // Backpointer to owning subchannel list. Not owned. - SubchannelListType* subchannel_list_; + SubchannelList<SubchannelListType, SubchannelDataType>* subchannel_list_; // The subchannel and connected subchannel. grpc_subchannel* subchannel_; @@ -200,7 +207,7 @@ class SubchannelList grpc_core::channelz::SubchannelNode* subchannel_node = grpc_subchannel_get_channelz_node(subchannels_[i].subchannel()); if (subchannel_node != nullptr) { - refs_list->push_back(subchannel_node->subchannel_uuid()); + refs_list->push_back(subchannel_node->uuid()); } } } @@ -268,7 +275,7 @@ class SubchannelList template <typename SubchannelListType, typename SubchannelDataType> SubchannelData<SubchannelListType, SubchannelDataType>::SubchannelData( - SubchannelListType* subchannel_list, + SubchannelList<SubchannelListType, SubchannelDataType>* subchannel_list, const grpc_lb_user_data_vtable* user_data_vtable, const grpc_lb_address& address, grpc_subchannel* subchannel, grpc_combiner* combiner) @@ -532,8 +539,7 @@ SubchannelList<SubchannelListType, SubchannelDataType>::SubchannelList( address_uri); gpr_free(address_uri); } - subchannels_.emplace_back(static_cast<SubchannelListType*>(this), - addresses->user_data_vtable, + subchannels_.emplace_back(this, addresses->user_data_vtable, addresses->addresses[i], subchannel, combiner); } } diff --git a/src/core/ext/filters/client_channel/parse_address.cc b/src/core/ext/filters/client_channel/parse_address.cc index b3900114ad..b94429e207 100644 --- a/src/core/ext/filters/client_channel/parse_address.cc +++ b/src/core/ext/filters/client_channel/parse_address.cc @@ -125,9 +125,16 @@ bool grpc_parse_ipv6_hostport(const char* hostport, grpc_resolved_address* addr, char* host_end = static_cast<char*>(gpr_memrchr(host, '%', strlen(host))); if (host_end != nullptr) { GPR_ASSERT(host_end >= host); - char host_without_scope[GRPC_INET6_ADDRSTRLEN]; + char host_without_scope[GRPC_INET6_ADDRSTRLEN + 1]; size_t host_without_scope_len = static_cast<size_t>(host_end - host); uint32_t sin6_scope_id = 0; + if (host_without_scope_len > GRPC_INET6_ADDRSTRLEN) { + gpr_log(GPR_ERROR, + "invalid ipv6 address length %zu. Length cannot be greater than " + "GRPC_INET6_ADDRSTRLEN i.e %d)", + host_without_scope_len, GRPC_INET6_ADDRSTRLEN); + goto done; + } strncpy(host_without_scope, host, host_without_scope_len); host_without_scope[host_without_scope_len] = '\0'; if (grpc_inet_pton(GRPC_AF_INET6, host_without_scope, &in6->sin6_addr) == @@ -190,3 +197,12 @@ bool grpc_parse_uri(const grpc_uri* uri, grpc_resolved_address* resolved_addr) { gpr_log(GPR_ERROR, "Can't parse scheme '%s'", uri->scheme); return false; } + +uint16_t grpc_strhtons(const char* port) { + if (strcmp(port, "http") == 0) { + return htons(80); + } else if (strcmp(port, "https") == 0) { + return htons(443); + } + return htons(static_cast<unsigned short>(atoi(port))); +} diff --git a/src/core/ext/filters/client_channel/parse_address.h b/src/core/ext/filters/client_channel/parse_address.h index 9a88b66edc..c2af0e6c49 100644 --- a/src/core/ext/filters/client_channel/parse_address.h +++ b/src/core/ext/filters/client_channel/parse_address.h @@ -47,4 +47,7 @@ bool grpc_parse_ipv4_hostport(const char* hostport, grpc_resolved_address* addr, bool grpc_parse_ipv6_hostport(const char* hostport, grpc_resolved_address* addr, bool log_errors); +/* Converts named or numeric port to a uint16 suitable for use in a sockaddr. */ +uint16_t grpc_strhtons(const char* port); + #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_PARSE_ADDRESS_H */ diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc index 485998f5e4..4c795c34c8 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc @@ -87,15 +87,6 @@ typedef struct grpc_ares_hostbyname_request { static void do_basic_init(void) { gpr_mu_init(&g_init_mu); } -static uint16_t strhtons(const char* port) { - if (strcmp(port, "http") == 0) { - return htons(80); - } else if (strcmp(port, "https") == 0) { - return htons(443); - } - return htons(static_cast<unsigned short>(atoi(port))); -} - static void log_address_sorting_list(grpc_lb_addresses* lb_addrs, const char* input_output_str) { for (size_t i = 0; i < lb_addrs->num_addresses; i++) { @@ -139,12 +130,6 @@ void grpc_cares_wrapper_address_sorting_sort(grpc_lb_addresses* lb_addrs) { } } -/* Allow tests to access grpc_ares_wrapper_address_sorting_sort */ -void grpc_cares_wrapper_test_only_address_sorting_sort( - grpc_lb_addresses* lb_addrs) { - grpc_cares_wrapper_address_sorting_sort(lb_addrs); -} - static void grpc_ares_request_ref_locked(grpc_ares_request* r) { r->pending_queries++; } @@ -371,7 +356,8 @@ done: grpc_ares_request_unref_locked(r); } -static grpc_ares_request* grpc_dns_lookup_ares_locked_impl( +static grpc_ares_request* +grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked( const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, grpc_closure* on_done, grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json, @@ -454,12 +440,12 @@ static grpc_ares_request* grpc_dns_lookup_ares_locked_impl( } r->pending_queries = 1; if (grpc_ares_query_ipv6()) { - hr = create_hostbyname_request_locked(r, host, strhtons(port), + hr = create_hostbyname_request_locked(r, host, grpc_strhtons(port), false /* is_balancer */); ares_gethostbyname(*channel, hr->host, AF_INET6, on_hostbyname_done_locked, hr); } - hr = create_hostbyname_request_locked(r, host, strhtons(port), + hr = create_hostbyname_request_locked(r, host, grpc_strhtons(port), false /* is_balancer */); ares_gethostbyname(*channel, hr->host, AF_INET, on_hostbyname_done_locked, hr); @@ -494,6 +480,79 @@ error_cleanup: return nullptr; } +static bool inner_resolve_as_ip_literal_locked(const char* name, + const char* default_port, + grpc_lb_addresses** addrs, + char** host, char** port, + char** hostport) { + gpr_split_host_port(name, host, port); + if (*host == nullptr) { + gpr_log(GPR_ERROR, + "Failed to parse %s to host:port while attempting to resolve as ip " + "literal.", + name); + return false; + } + if (*port == nullptr) { + if (default_port == nullptr) { + gpr_log(GPR_ERROR, + "No port or default port for %s while attempting to resolve as " + "ip literal.", + name); + return false; + } + *port = gpr_strdup(default_port); + } + grpc_resolved_address addr; + GPR_ASSERT(gpr_join_host_port(hostport, *host, atoi(*port))); + if (grpc_parse_ipv4_hostport(*hostport, &addr, false /* log errors */) || + grpc_parse_ipv6_hostport(*hostport, &addr, false /* log errors */)) { + GPR_ASSERT(*addrs == nullptr); + *addrs = grpc_lb_addresses_create(1, nullptr); + grpc_lb_addresses_set_address( + *addrs, 0, addr.addr, addr.len, false /* is_balancer */, + nullptr /* balancer_name */, nullptr /* user_data */); + return true; + } + return false; +} + +static bool resolve_as_ip_literal_locked(const char* name, + const char* default_port, + grpc_lb_addresses** addrs) { + char* host = nullptr; + char* port = nullptr; + char* hostport = nullptr; + bool out = inner_resolve_as_ip_literal_locked(name, default_port, addrs, + &host, &port, &hostport); + gpr_free(host); + gpr_free(port); + gpr_free(hostport); + return out; +} + +static grpc_ares_request* grpc_dns_lookup_ares_locked_impl( + const char* dns_server, const char* name, const char* default_port, + grpc_pollset_set* interested_parties, grpc_closure* on_done, + grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json, + grpc_combiner* combiner) { + // Early out if the target is an ipv4 or ipv6 literal. + if (resolve_as_ip_literal_locked(name, default_port, addrs)) { + GRPC_CLOSURE_SCHED(on_done, GRPC_ERROR_NONE); + return nullptr; + } + // Early out if the target is localhost and we're on Windows. + if (grpc_ares_maybe_resolve_localhost_manually_locked(name, default_port, + addrs)) { + GRPC_CLOSURE_SCHED(on_done, GRPC_ERROR_NONE); + return nullptr; + } + // Look up name using c-ares lib. + return grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked( + dns_server, name, default_port, interested_parties, on_done, addrs, + check_grpclb, service_config_json, combiner); +} + grpc_ares_request* (*grpc_dns_lookup_ares_locked)( const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, grpc_closure* on_done, @@ -502,7 +561,9 @@ grpc_ares_request* (*grpc_dns_lookup_ares_locked)( void grpc_cancel_ares_request(grpc_ares_request* r) { if (grpc_dns_lookup_ares_locked == grpc_dns_lookup_ares_locked_impl) { - grpc_ares_ev_driver_shutdown_locked(r->ev_driver); + if (r != nullptr) { + grpc_ares_ev_driver_shutdown_locked(r->ev_driver); + } } } diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h index ca5779e1d7..1bc457d4cf 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h @@ -81,9 +81,15 @@ void grpc_ares_complete_request_locked(grpc_ares_request* request); /* E.g., return false if ipv6 is known to not be available. */ bool grpc_ares_query_ipv6(); -/* Exposed only for testing */ -void grpc_cares_wrapper_test_only_address_sorting_sort( - grpc_lb_addresses* lb_addrs); +/* Maybe (depending on the current platform) checks if "name" matches + * "localhost" and if so fills in addrs with the correct sockaddr structures. + * Returns a bool indicating whether or not such an action was performed. + * See https://github.com/grpc/grpc/issues/15158. */ +bool grpc_ares_maybe_resolve_localhost_manually_locked( + const char* name, const char* default_port, grpc_lb_addresses** addrs); + +/* Sorts destinations in lb_addrs according to RFC 6724. */ +void grpc_cares_wrapper_address_sorting_sort(grpc_lb_addresses* lb_addrs); #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_DNS_C_ARES_GRPC_ARES_WRAPPER_H \ */ diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc index 23c0fec74f..639eec2323 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc @@ -26,4 +26,9 @@ bool grpc_ares_query_ipv6() { return grpc_ipv6_loopback_available(); } +bool grpc_ares_maybe_resolve_localhost_manually_locked( + const char* name, const char* default_port, grpc_lb_addresses** addrs) { + return false; +} + #endif /* GRPC_ARES == 1 && defined(GRPC_POSIX_SOCKET_ARES_EV_DRIVER) */ diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc index ee827e284e..7e34784691 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc @@ -21,9 +21,79 @@ #include "src/core/lib/iomgr/port.h" #if GRPC_ARES == 1 && defined(GPR_WINDOWS) +#include <grpc/support/string_util.h> + +#include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h" +#include "src/core/lib/gpr/host_port.h" +#include "src/core/lib/gpr/string.h" #include "src/core/lib/iomgr/socket_windows.h" bool grpc_ares_query_ipv6() { return grpc_ipv6_loopback_available(); } +static bool inner_maybe_resolve_localhost_manually_locked( + const char* name, const char* default_port, grpc_lb_addresses** addrs, + char** host, char** port) { + gpr_split_host_port(name, host, port); + if (*host == nullptr) { + gpr_log(GPR_ERROR, + "Failed to parse %s into host:port during Windows localhost " + "resolution check.", + name); + return false; + } + if (*port == nullptr) { + if (default_port == nullptr) { + gpr_log(GPR_ERROR, + "No port or default port for %s during Windows localhost " + "resolution check.", + name); + return false; + } + *port = gpr_strdup(default_port); + } + if (gpr_stricmp(*host, "localhost") == 0) { + GPR_ASSERT(*addrs == nullptr); + *addrs = grpc_lb_addresses_create(2, nullptr); + uint16_t numeric_port = grpc_strhtons(*port); + // Append the ipv6 loopback address. + struct sockaddr_in6 ipv6_loopback_addr; + memset(&ipv6_loopback_addr, 0, sizeof(ipv6_loopback_addr)); + ((char*)&ipv6_loopback_addr.sin6_addr)[15] = 1; + ipv6_loopback_addr.sin6_family = AF_INET6; + ipv6_loopback_addr.sin6_port = numeric_port; + grpc_lb_addresses_set_address( + *addrs, 0, &ipv6_loopback_addr, sizeof(ipv6_loopback_addr), + false /* is_balancer */, nullptr /* balancer_name */, + nullptr /* user_data */); + // Append the ipv4 loopback address. + struct sockaddr_in ipv4_loopback_addr; + memset(&ipv4_loopback_addr, 0, sizeof(ipv4_loopback_addr)); + ((char*)&ipv4_loopback_addr.sin_addr)[0] = 0x7f; + ((char*)&ipv4_loopback_addr.sin_addr)[3] = 0x01; + ipv4_loopback_addr.sin_family = AF_INET; + ipv4_loopback_addr.sin_port = numeric_port; + grpc_lb_addresses_set_address( + *addrs, 1, &ipv4_loopback_addr, sizeof(ipv4_loopback_addr), + false /* is_balancer */, nullptr /* balancer_name */, + nullptr /* user_data */); + // Let the address sorter figure out which one should be tried first. + grpc_cares_wrapper_address_sorting_sort(*addrs); + return true; + } + return false; +} + +bool grpc_ares_maybe_resolve_localhost_manually_locked( + const char* name, const char* default_port, grpc_lb_addresses** addrs) { + char* host = nullptr; + char* port = nullptr; + bool out = inner_maybe_resolve_localhost_manually_locked(name, default_port, + addrs, &host, &port); + gpr_free(host); + gpr_free(port); + return out; +} + #endif /* GRPC_ARES == 1 && defined(GPR_WINDOWS) */ diff --git a/src/core/ext/filters/client_channel/subchannel.cc b/src/core/ext/filters/client_channel/subchannel.cc index 0e40f42e18..57d0b3759f 100644 --- a/src/core/ext/filters/client_channel/subchannel.cc +++ b/src/core/ext/filters/client_channel/subchannel.cc @@ -183,7 +183,13 @@ static void connection_destroy(void* arg, grpc_error* error) { static void subchannel_destroy(void* arg, grpc_error* error) { grpc_subchannel* c = static_cast<grpc_subchannel*>(arg); - c->channelz_subchannel.reset(); + if (c->channelz_subchannel != nullptr) { + c->channelz_subchannel->AddTraceEvent( + grpc_core::channelz::ChannelTrace::Severity::Info, + grpc_slice_from_static_string("Subchannel destroyed")); + c->channelz_subchannel->MarkSubchannelDestroyed(); + c->channelz_subchannel.reset(); + } gpr_free((void*)c->filters); grpc_channel_args_destroy(c->args); grpc_connectivity_state_destroy(&c->state_tracker); @@ -383,9 +389,18 @@ grpc_subchannel* grpc_subchannel_create(grpc_connector* connector, const grpc_arg* arg = grpc_channel_args_find(c->args, GRPC_ARG_ENABLE_CHANNELZ); bool channelz_enabled = grpc_channel_arg_get_bool(arg, false); + arg = grpc_channel_args_find(c->args, + GRPC_ARG_MAX_CHANNEL_TRACE_EVENTS_PER_NODE); + const grpc_integer_options options = {0, 0, INT_MAX}; + size_t channel_tracer_max_nodes = + (size_t)grpc_channel_arg_get_integer(arg, options); if (channelz_enabled) { c->channelz_subchannel = - grpc_core::MakeRefCounted<grpc_core::channelz::SubchannelNode>(); + grpc_core::MakeRefCounted<grpc_core::channelz::SubchannelNode>( + c, channel_tracer_max_nodes); + c->channelz_subchannel->AddTraceEvent( + grpc_core::channelz::ChannelTrace::Severity::Info, + grpc_slice_from_static_string("Subchannel created")); } return grpc_subchannel_index_register(key, c); @@ -625,8 +640,8 @@ static bool publish_transport_locked(grpc_subchannel* c) { } /* publish */ - c->connected_subchannel.reset( - grpc_core::New<grpc_core::ConnectedSubchannel>(stk)); + c->connected_subchannel.reset(grpc_core::New<grpc_core::ConnectedSubchannel>( + stk, c->channelz_subchannel.get())); gpr_log(GPR_INFO, "New connected subchannel at %p for subchannel %p", c->connected_subchannel.get(), c); @@ -770,6 +785,14 @@ void grpc_get_subchannel_address_arg(const grpc_channel_args* args, } } +const char* grpc_subchannel_get_target(grpc_subchannel* subchannel) { + const grpc_arg* addr_arg = + grpc_channel_args_find(subchannel->args, GRPC_ARG_SUBCHANNEL_ADDRESS); + const char* addr_str = grpc_channel_arg_get_string(addr_arg); + GPR_ASSERT(addr_str != nullptr); // Should have been set by LB policy. + return addr_str; +} + const char* grpc_get_subchannel_address_uri_arg(const grpc_channel_args* args) { const grpc_arg* addr_arg = grpc_channel_args_find(args, GRPC_ARG_SUBCHANNEL_ADDRESS); @@ -786,9 +809,12 @@ grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address* addr) { namespace grpc_core { -ConnectedSubchannel::ConnectedSubchannel(grpc_channel_stack* channel_stack) +ConnectedSubchannel::ConnectedSubchannel( + grpc_channel_stack* channel_stack, + channelz::SubchannelNode* channelz_subchannel) : RefCountedWithTracing<ConnectedSubchannel>(&grpc_trace_stream_refcount), - channel_stack_(channel_stack) {} + channel_stack_(channel_stack), + channelz_subchannel_(channelz_subchannel) {} ConnectedSubchannel::~ConnectedSubchannel() { GRPC_CHANNEL_STACK_UNREF(channel_stack_, "connected_subchannel_dtor"); diff --git a/src/core/ext/filters/client_channel/subchannel.h b/src/core/ext/filters/client_channel/subchannel.h index a135035d62..84febb5204 100644 --- a/src/core/ext/filters/client_channel/subchannel.h +++ b/src/core/ext/filters/client_channel/subchannel.h @@ -85,7 +85,8 @@ class ConnectedSubchannel : public RefCountedWithTracing<ConnectedSubchannel> { size_t parent_data_size; }; - explicit ConnectedSubchannel(grpc_channel_stack* channel_stack); + explicit ConnectedSubchannel(grpc_channel_stack* channel_stack, + channelz::SubchannelNode* channelz_subchannel); ~ConnectedSubchannel(); grpc_channel_stack* channel_stack() { return channel_stack_; } @@ -94,9 +95,15 @@ class ConnectedSubchannel : public RefCountedWithTracing<ConnectedSubchannel> { grpc_closure* closure); void Ping(grpc_closure* on_initiate, grpc_closure* on_ack); grpc_error* CreateCall(const CallArgs& args, grpc_subchannel_call** call); + channelz::SubchannelNode* channelz_subchannel() { + return channelz_subchannel_; + } private: grpc_channel_stack* channel_stack_; + // backpointer to the channelz node in this connected subchannel's + // owning subchannel. + channelz::SubchannelNode* channelz_subchannel_; }; } // namespace grpc_core @@ -184,6 +191,8 @@ grpc_subchannel* grpc_subchannel_create(grpc_connector* connector, void grpc_get_subchannel_address_arg(const grpc_channel_args* args, grpc_resolved_address* addr); +const char* grpc_subchannel_get_target(grpc_subchannel* subchannel); + /// Returns the URI string for the address to connect to. const char* grpc_get_subchannel_address_uri_arg(const grpc_channel_args* args); diff --git a/src/core/ext/filters/client_channel/subchannel_index.cc b/src/core/ext/filters/client_channel/subchannel_index.cc index cb02b1a748..f2b6c24e8e 100644 --- a/src/core/ext/filters/client_channel/subchannel_index.cc +++ b/src/core/ext/filters/client_channel/subchannel_index.cc @@ -42,7 +42,7 @@ struct grpc_subchannel_key { grpc_subchannel_args args; }; -static bool g_force_creation = false; +static gpr_atm g_force_creation = false; static grpc_subchannel_key* create_key( const grpc_subchannel_args* args, @@ -73,7 +73,8 @@ static grpc_subchannel_key* subchannel_key_copy(grpc_subchannel_key* k) { int grpc_subchannel_key_compare(const grpc_subchannel_key* a, const grpc_subchannel_key* b) { - if (g_force_creation) return false; + // To pretend the keys are different, return a non-zero value. + if (GPR_UNLIKELY(gpr_atm_no_barrier_load(&g_force_creation))) return 1; int c = GPR_ICMP(a->args.filter_count, b->args.filter_count); if (c != 0) return c; if (a->args.filter_count > 0) { @@ -250,5 +251,5 @@ void grpc_subchannel_index_unregister(grpc_subchannel_key* key, } void grpc_subchannel_index_test_only_set_force_creation(bool force_creation) { - g_force_creation = force_creation; + gpr_atm_no_barrier_store(&g_force_creation, force_creation); } diff --git a/src/core/ext/filters/client_channel/subchannel_index.h b/src/core/ext/filters/client_channel/subchannel_index.h index a7dae9d47d..c135613d26 100644 --- a/src/core/ext/filters/client_channel/subchannel_index.h +++ b/src/core/ext/filters/client_channel/subchannel_index.h @@ -65,13 +65,10 @@ void grpc_subchannel_index_ref(void); void grpc_subchannel_index_unref(void); /** \em TEST ONLY. - * If \a force_creation is true, all key comparisons will be false, resulting in + * If \a force_creation is true, all keys are regarded different, resulting in * new subchannels always being created. Otherwise, the keys will be compared as * usual. * - * This function is *not* threadsafe on purpose: it should *only* be used in - * test code. - * * Tests using this function \em MUST run tests with and without \a * force_creation set. */ void grpc_subchannel_index_test_only_set_force_creation(bool force_creation); diff --git a/src/core/ext/filters/http/server/http_server_filter.cc b/src/core/ext/filters/http/server/http_server_filter.cc index 90336103cf..cd3cab134a 100644 --- a/src/core/ext/filters/http/server/http_server_filter.cc +++ b/src/core/ext/filters/http/server/http_server_filter.cc @@ -23,6 +23,7 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <string.h> +#include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gprpp/manual_constructor.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/b64.h" @@ -69,6 +70,10 @@ struct call_data { bool seen_recv_trailing_metadata_ready; }; +struct channel_data { + bool surface_user_agent; +}; + } // namespace static grpc_error* hs_filter_outgoing_metadata(grpc_call_element* elem, @@ -265,6 +270,11 @@ static grpc_error* hs_filter_incoming_metadata(grpc_call_element* elem, GRPC_ERROR_STR_KEY, grpc_slice_from_static_string(":authority"))); } + channel_data* chand = static_cast<channel_data*>(elem->channel_data); + if (!chand->surface_user_agent && b->idx.named.user_agent != nullptr) { + grpc_metadata_batch_remove(b, b->idx.named.user_agent); + } + return error; } @@ -451,7 +461,12 @@ static void hs_destroy_call_elem(grpc_call_element* elem, /* Constructor for channel_data */ static grpc_error* hs_init_channel_elem(grpc_channel_element* elem, grpc_channel_element_args* args) { + channel_data* chand = static_cast<channel_data*>(elem->channel_data); GPR_ASSERT(!args->is_last); + chand->surface_user_agent = grpc_channel_arg_get_bool( + grpc_channel_args_find(args->channel_args, + const_cast<char*>(GRPC_ARG_SURFACE_USER_AGENT)), + true); return GRPC_ERROR_NONE; } @@ -465,7 +480,7 @@ const grpc_channel_filter grpc_http_server_filter = { hs_init_call_elem, grpc_call_stack_ignore_set_pollset_or_pollset_set, hs_destroy_call_elem, - 0, + sizeof(channel_data), hs_init_channel_elem, hs_destroy_channel_elem, grpc_channel_next_get_info, diff --git a/src/core/lib/channel/channel_trace.cc b/src/core/lib/channel/channel_trace.cc index b3443310ac..cfb2faba51 100644 --- a/src/core/lib/channel/channel_trace.cc +++ b/src/core/lib/channel/channel_trace.cc @@ -41,16 +41,14 @@ namespace grpc_core { namespace channelz { -ChannelTrace::TraceEvent::TraceEvent( - Severity severity, grpc_slice data, - RefCountedPtr<ChannelNode> referenced_channel, ReferencedType type) +ChannelTrace::TraceEvent::TraceEvent(Severity severity, grpc_slice data, + RefCountedPtr<BaseNode> referenced_entity) : severity_(severity), data_(data), timestamp_(grpc_millis_to_timespec(grpc_core::ExecCtx::Get()->Now(), GPR_CLOCK_REALTIME)), next_(nullptr), - referenced_channel_(std::move(referenced_channel)), - referenced_type_(type) {} + referenced_entity_(std::move(referenced_entity)) {} ChannelTrace::TraceEvent::TraceEvent(Severity severity, grpc_slice data) : severity_(severity), @@ -110,23 +108,13 @@ void ChannelTrace::AddTraceEvent(Severity severity, grpc_slice data) { AddTraceEventHelper(New<TraceEvent>(severity, data)); } -void ChannelTrace::AddTraceEventReferencingChannel( - Severity severity, grpc_slice data, - RefCountedPtr<ChannelNode> referenced_channel) { - if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0 - // create and fill up the new event - AddTraceEventHelper(New<TraceEvent>( - severity, data, std::move(referenced_channel), ReferencedType::Channel)); -} - -void ChannelTrace::AddTraceEventReferencingSubchannel( +void ChannelTrace::AddTraceEventWithReference( Severity severity, grpc_slice data, - RefCountedPtr<ChannelNode> referenced_subchannel) { + RefCountedPtr<BaseNode> referenced_entity) { if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0 // create and fill up the new event - AddTraceEventHelper(New<TraceEvent>(severity, data, - std::move(referenced_subchannel), - ReferencedType::Subchannel)); + AddTraceEventHelper( + New<TraceEvent>(severity, data, std::move(referenced_entity))); } namespace { @@ -157,19 +145,18 @@ void ChannelTrace::TraceEvent::RenderTraceEvent(grpc_json* json) const { json_iterator = grpc_json_create_child(json_iterator, json, "timestamp", gpr_format_timespec(timestamp_), GRPC_JSON_STRING, true); - if (referenced_channel_ != nullptr) { + if (referenced_entity_ != nullptr) { + const bool is_channel = + (referenced_entity_->type() == BaseNode::EntityType::kTopLevelChannel || + referenced_entity_->type() == BaseNode::EntityType::kInternalChannel); char* uuid_str; - gpr_asprintf(&uuid_str, "%" PRIdPTR, referenced_channel_->channel_uuid()); + gpr_asprintf(&uuid_str, "%" PRIdPTR, referenced_entity_->uuid()); grpc_json* child_ref = grpc_json_create_child( - json_iterator, json, - (referenced_type_ == ReferencedType::Channel) ? "channelRef" - : "subchannelRef", + json_iterator, json, is_channel ? "channelRef" : "subchannelRef", nullptr, GRPC_JSON_OBJECT, false); json_iterator = grpc_json_create_child( - nullptr, child_ref, - (referenced_type_ == ReferencedType::Channel) ? "channelId" - : "subchannelId", - uuid_str, GRPC_JSON_STRING, true); + nullptr, child_ref, is_channel ? "channelId" : "subchannelId", uuid_str, + GRPC_JSON_STRING, true); json_iterator = child_ref; } } @@ -178,24 +165,26 @@ grpc_json* ChannelTrace::RenderJson() const { if (!max_list_size_) return nullptr; // tracing is disabled if max_events == 0 grpc_json* json = grpc_json_create(GRPC_JSON_OBJECT); - char* num_events_logged_str; - gpr_asprintf(&num_events_logged_str, "%" PRId64, num_events_logged_); grpc_json* json_iterator = nullptr; - json_iterator = - grpc_json_create_child(json_iterator, json, "numEventsLogged", - num_events_logged_str, GRPC_JSON_STRING, true); + if (num_events_logged_ > 0) { + json_iterator = grpc_json_add_number_string_child( + json, json_iterator, "numEventsLogged", num_events_logged_); + } json_iterator = grpc_json_create_child( json_iterator, json, "creationTimestamp", gpr_format_timespec(time_created_), GRPC_JSON_STRING, true); - grpc_json* events = grpc_json_create_child(json_iterator, json, "events", - nullptr, GRPC_JSON_ARRAY, false); - json_iterator = nullptr; - TraceEvent* it = head_trace_; - while (it != nullptr) { - json_iterator = grpc_json_create_child(json_iterator, events, nullptr, - nullptr, GRPC_JSON_OBJECT, false); - it->RenderTraceEvent(json_iterator); - it = it->next(); + // only add in the event list if it is non-empty. + if (num_events_logged_ > 0) { + grpc_json* events = grpc_json_create_child(json_iterator, json, "events", + nullptr, GRPC_JSON_ARRAY, false); + json_iterator = nullptr; + TraceEvent* it = head_trace_; + while (it != nullptr) { + json_iterator = grpc_json_create_child(json_iterator, events, nullptr, + nullptr, GRPC_JSON_OBJECT, false); + it->RenderTraceEvent(json_iterator); + it = it->next(); + } } return json; } diff --git a/src/core/lib/channel/channel_trace.h b/src/core/lib/channel/channel_trace.h index 596af7402f..94fea20b45 100644 --- a/src/core/lib/channel/channel_trace.h +++ b/src/core/lib/channel/channel_trace.h @@ -30,7 +30,7 @@ namespace grpc_core { namespace channelz { -class ChannelNode; +class BaseNode; // Object used to hold live data for a channel. This data is exposed via the // channelz service: @@ -55,35 +55,28 @@ class ChannelTrace { void AddTraceEvent(Severity severity, grpc_slice data); // Adds a new trace event to the tracing object. This trace event refers to a - // an event on a child of the channel. For example, if this channel has - // created a new subchannel, then it would record that with a TraceEvent - // referencing the new subchannel. + // an event that concerns a different channelz entity. For example, if this + // channel has created a new subchannel, then it would record that with + // a TraceEvent referencing the new subchannel. // // TODO(ncteisen): as this call is used more and more throughout the gRPC // stack, determine if it makes more sense to accept a char* instead of a // slice. - void AddTraceEventReferencingChannel( - Severity severity, grpc_slice data, - RefCountedPtr<ChannelNode> referenced_channel); - void AddTraceEventReferencingSubchannel( - Severity severity, grpc_slice data, - RefCountedPtr<ChannelNode> referenced_subchannel); + void AddTraceEventWithReference(Severity severity, grpc_slice data, + RefCountedPtr<BaseNode> referenced_entity); // Creates and returns the raw grpc_json object, so a parent channelz // object may incorporate the json before rendering. grpc_json* RenderJson() const; private: - // Types of objects that can be references by trace events. - enum class ReferencedType { Channel, Subchannel }; // Private class to encapsulate all the data and bookkeeping needed for a // a trace event. class TraceEvent { public: - // Constructor for a TraceEvent that references a different channel. + // Constructor for a TraceEvent that references a channel. TraceEvent(Severity severity, grpc_slice data, - RefCountedPtr<ChannelNode> referenced_channel, - ReferencedType type); + RefCountedPtr<BaseNode> referenced_entity_); // Constructor for a TraceEvent that does not reverence a different // channel. @@ -105,10 +98,7 @@ class ChannelTrace { gpr_timespec timestamp_; TraceEvent* next_; // the tracer object for the (sub)channel that this trace event refers to. - RefCountedPtr<ChannelNode> referenced_channel_; - // the type that the referenced tracer points to. Unused if this trace - // does not point to any channel or subchannel - ReferencedType referenced_type_; + RefCountedPtr<BaseNode> referenced_entity_; }; // TraceEvent // Internal helper to add and link in a trace event diff --git a/src/core/lib/channel/channelz.cc b/src/core/lib/channel/channelz.cc index 9d6002ed8a..375cf25cc6 100644 --- a/src/core/lib/channel/channelz.cc +++ b/src/core/lib/channel/channelz.cc @@ -41,33 +41,62 @@ namespace grpc_core { namespace channelz { -ChannelNode::ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes, - bool is_top_level_channel) - : channel_(channel), - target_(nullptr), - channel_uuid_(-1), - is_top_level_channel_(is_top_level_channel) { - trace_.Init(channel_tracer_max_nodes); - target_ = UniquePtr<char>(grpc_channel_get_target(channel_)); - channel_uuid_ = ChannelzRegistry::RegisterChannelNode(this); +BaseNode::BaseNode(EntityType type) + : type_(type), uuid_(ChannelzRegistry::Register(this)) {} + +BaseNode::~BaseNode() { ChannelzRegistry::Unregister(uuid_); } + +char* BaseNode::RenderJsonString() { + grpc_json* json = RenderJson(); + GPR_ASSERT(json != nullptr); + char* json_str = grpc_json_dump_to_string(json, 0); + grpc_json_destroy(json); + return json_str; +} + +CallCountingHelper::CallCountingHelper() { gpr_atm_no_barrier_store(&last_call_started_millis_, (gpr_atm)ExecCtx::Get()->Now()); } -ChannelNode::~ChannelNode() { - trace_.Destroy(); - ChannelzRegistry::UnregisterChannelNode(channel_uuid_); -} +CallCountingHelper::~CallCountingHelper() {} -void ChannelNode::RecordCallStarted() { +void CallCountingHelper::RecordCallStarted() { gpr_atm_no_barrier_fetch_add(&calls_started_, (gpr_atm)1); gpr_atm_no_barrier_store(&last_call_started_millis_, (gpr_atm)ExecCtx::Get()->Now()); } -void ChannelNode::PopulateConnectivityState(grpc_json* json) {} +void CallCountingHelper::PopulateCallCounts(grpc_json* json) { + grpc_json* json_iterator = nullptr; + if (calls_started_ != 0) { + json_iterator = grpc_json_add_number_string_child( + json, json_iterator, "callsStarted", calls_started_); + } + if (calls_succeeded_ != 0) { + json_iterator = grpc_json_add_number_string_child( + json, json_iterator, "callsSucceeded", calls_succeeded_); + } + if (calls_failed_) { + json_iterator = grpc_json_add_number_string_child( + json, json_iterator, "callsFailed", calls_failed_); + } + gpr_timespec ts = + grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME); + json_iterator = + grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp", + gpr_format_timespec(ts), GRPC_JSON_STRING, true); +} + +ChannelNode::ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes, + bool is_top_level_channel) + : BaseNode(is_top_level_channel ? EntityType::kTopLevelChannel + : EntityType::kInternalChannel), + channel_(channel), + target_(UniquePtr<char>(grpc_channel_get_target(channel_))), + trace_(channel_tracer_max_nodes) {} -void ChannelNode::PopulateChildRefs(grpc_json* json) {} +ChannelNode::~ChannelNode() {} grpc_json* ChannelNode::RenderJson() { // We need to track these three json objects to build our object @@ -80,7 +109,7 @@ grpc_json* ChannelNode::RenderJson() { json = json_iterator; json_iterator = nullptr; json_iterator = grpc_json_add_number_string_child(json, json_iterator, - "channelId", channel_uuid_); + "channelId", uuid()); // reset json iterators to top level object json = top_level_json; json_iterator = nullptr; @@ -89,51 +118,28 @@ grpc_json* ChannelNode::RenderJson() { GRPC_JSON_OBJECT, false); json = data; json_iterator = nullptr; + // template method. Child classes may override this to add their specific + // functionality. PopulateConnectivityState(json); + // populate the target. GPR_ASSERT(target_.get() != nullptr); - json_iterator = grpc_json_create_child( - json_iterator, json, "target", target_.get(), GRPC_JSON_STRING, false); + grpc_json_create_child(nullptr, json, "target", target_.get(), + GRPC_JSON_STRING, false); // fill in the channel trace if applicable - grpc_json* trace = trace_->RenderJson(); - if (trace != nullptr) { - // we manually link up and fill the child since it was created for us in - // ChannelTrace::RenderJson - trace->key = "trace"; // this object is named trace in channelz.proto - json_iterator = grpc_json_link_child(json, trace, json_iterator); - } - // reset the parent to be the data object. - json = data; - json_iterator = nullptr; - if (calls_started_ != 0) { - json_iterator = grpc_json_add_number_string_child( - json, json_iterator, "callsStarted", calls_started_); + grpc_json* trace_json = trace_.RenderJson(); + if (trace_json != nullptr) { + trace_json->key = "trace"; // this object is named trace in channelz.proto + grpc_json_link_child(json, trace_json, nullptr); } - if (calls_succeeded_ != 0) { - json_iterator = grpc_json_add_number_string_child( - json, json_iterator, "callsSucceeded", calls_succeeded_); - } - if (calls_failed_) { - json_iterator = grpc_json_add_number_string_child( - json, json_iterator, "callsFailed", calls_failed_); - } - gpr_timespec ts = - grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME); - json_iterator = - grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp", - gpr_format_timespec(ts), GRPC_JSON_STRING, true); + // ask CallCountingHelper to populate trace and call count data. + call_counter_.PopulateCallCounts(json); json = top_level_json; - json_iterator = nullptr; + // template method. Child classes may override this to add their specific + // functionality. PopulateChildRefs(json); return top_level_json; } -char* ChannelNode::RenderJsonString() { - grpc_json* json = RenderJson(); - char* json_str = grpc_json_dump_to_string(json, 0); - grpc_json_destroy(json); - return json_str; -} - RefCountedPtr<ChannelNode> ChannelNode::MakeChannelNode( grpc_channel* channel, size_t channel_tracer_max_nodes, bool is_top_level_channel) { @@ -141,12 +147,41 @@ RefCountedPtr<ChannelNode> ChannelNode::MakeChannelNode( channel, channel_tracer_max_nodes, is_top_level_channel); } -SubchannelNode::SubchannelNode() { - subchannel_uuid_ = ChannelzRegistry::RegisterSubchannelNode(this); -} +ServerNode::ServerNode(size_t channel_tracer_max_nodes) + : BaseNode(EntityType::kServer), trace_(channel_tracer_max_nodes) {} + +ServerNode::~ServerNode() {} -SubchannelNode::~SubchannelNode() { - ChannelzRegistry::UnregisterSubchannelNode(subchannel_uuid_); +grpc_json* ServerNode::RenderJson() { + // We need to track these three json objects to build our object + grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); + grpc_json* json = top_level_json; + grpc_json* json_iterator = nullptr; + // create and fill the ref child + json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr, + GRPC_JSON_OBJECT, false); + json = json_iterator; + json_iterator = nullptr; + json_iterator = grpc_json_add_number_string_child(json, json_iterator, + "serverId", uuid()); + // reset json iterators to top level object + json = top_level_json; + json_iterator = nullptr; + // create and fill the data child. + grpc_json* data = grpc_json_create_child(json_iterator, json, "data", nullptr, + GRPC_JSON_OBJECT, false); + json = data; + json_iterator = nullptr; + // fill in the channel trace if applicable + grpc_json* trace_json = trace_.RenderJson(); + if (trace_json != nullptr) { + trace_json->key = "trace"; // this object is named trace in channelz.proto + grpc_json_link_child(json, trace_json, nullptr); + } + // ask CallCountingHelper to populate trace and call count data. + call_counter_.PopulateCallCounts(json); + json = top_level_json; + return top_level_json; } } // namespace channelz diff --git a/src/core/lib/channel/channelz.h b/src/core/lib/channel/channelz.h index 07eb73d626..9be256147b 100644 --- a/src/core/lib/channel/channelz.h +++ b/src/core/lib/channel/channelz.h @@ -43,14 +43,52 @@ namespace grpc_core { namespace channelz { namespace testing { +class CallCountingHelperPeer; class ChannelNodePeer; -} +} // namespace testing -class ChannelNode : public RefCounted<ChannelNode> { +// base class for all channelz entities +class BaseNode : public RefCounted<BaseNode> { public: - static RefCountedPtr<ChannelNode> MakeChannelNode( - grpc_channel* channel, size_t channel_tracer_max_nodes, - bool is_top_level_channel); + // There are only four high level channelz entities. However, to support + // GetTopChannelsRequest, we split the Channel entity into two different + // types. All children of BaseNode must be one of these types. + enum class EntityType { + kTopLevelChannel, + kInternalChannel, + kSubchannel, + kServer, + kSocket, + }; + + explicit BaseNode(EntityType type); + virtual ~BaseNode(); + + // All children must implement this function. + virtual grpc_json* RenderJson() GRPC_ABSTRACT; + + // Renders the json and returns allocated string that must be freed by the + // caller. + char* RenderJsonString(); + + EntityType type() const { return type_; } + intptr_t uuid() const { return uuid_; } + + private: + const EntityType type_; + const intptr_t uuid_; +}; + +// This class is a helper class for channelz entities that deal with Channels, +// Subchannels, and Servers, since those have similar proto definitions. +// This class has the ability to: +// - track calls_{started,succeeded,failed} +// - track last_call_started_timestamp +// - perform rendering of the above items +class CallCountingHelper { + public: + CallCountingHelper(); + ~CallCountingHelper(); void RecordCallStarted(); void RecordCallFailed() { @@ -60,17 +98,46 @@ class ChannelNode : public RefCounted<ChannelNode> { gpr_atm_no_barrier_fetch_add(&calls_succeeded_, (gpr_atm(1))); } - grpc_json* RenderJson(); - char* RenderJsonString(); + // Common rendering of the call count data and last_call_started_timestamp. + void PopulateCallCounts(grpc_json* json); - // helper for getting and populating connectivity state. It is virtual - // because it allows the client_channel specific code to live in ext/ - // instead of lib/ - virtual void PopulateConnectivityState(grpc_json* json); + private: + // testing peer friend. + friend class testing::CallCountingHelperPeer; - virtual void PopulateChildRefs(grpc_json* json); + gpr_atm calls_started_ = 0; + gpr_atm calls_succeeded_ = 0; + gpr_atm calls_failed_ = 0; + gpr_atm last_call_started_millis_ = 0; +}; - ChannelTrace* trace() { return trace_.get(); } +// Handles channelz bookkeeping for channels +class ChannelNode : public BaseNode { + public: + static RefCountedPtr<ChannelNode> MakeChannelNode( + grpc_channel* channel, size_t channel_tracer_max_nodes, + bool is_top_level_channel); + + ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes, + bool is_top_level_channel); + ~ChannelNode() override; + + grpc_json* RenderJson() override; + + // template methods. RenderJSON uses these methods to render its JSON + // representation. These are virtual so that children classes may provide + // their specific mechanism for populating these parts of the channelz + // object. + // + // ChannelNode does not have a notion of connectivity state or child refs, + // so it leaves these implementations blank. + // + // This is utilizing the template method design pattern. + // + // TODO(ncteisen): remove these template methods in favor of manual traversal + // and mutation of the grpc_json object. + virtual void PopulateConnectivityState(grpc_json* json) {} + virtual void PopulateChildRefs(grpc_json* json) {} void MarkChannelDestroyed() { GPR_ASSERT(channel_ != nullptr); @@ -79,47 +146,62 @@ class ChannelNode : public RefCounted<ChannelNode> { bool ChannelIsDestroyed() { return channel_ == nullptr; } - intptr_t channel_uuid() { return channel_uuid_; } - bool is_top_level_channel() { return is_top_level_channel_; } - - protected: - GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_DELETE - GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_NEW - ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes, - bool is_top_level_channel); - virtual ~ChannelNode(); + // proxy methods to composed classes. + void AddTraceEvent(ChannelTrace::Severity severity, grpc_slice data) { + trace_.AddTraceEvent(severity, data); + } + void AddTraceEventWithReference(ChannelTrace::Severity severity, + grpc_slice data, + RefCountedPtr<BaseNode> referenced_channel) { + trace_.AddTraceEventWithReference(severity, data, + std::move(referenced_channel)); + } + void RecordCallStarted() { call_counter_.RecordCallStarted(); } + void RecordCallFailed() { call_counter_.RecordCallFailed(); } + void RecordCallSucceeded() { call_counter_.RecordCallSucceeded(); } private: - // testing peer friend. + // to allow the channel trace test to access trace_. friend class testing::ChannelNodePeer; - grpc_channel* channel_ = nullptr; UniquePtr<char> target_; - gpr_atm calls_started_ = 0; - gpr_atm calls_succeeded_ = 0; - gpr_atm calls_failed_ = 0; - gpr_atm last_call_started_millis_ = 0; - intptr_t channel_uuid_; - bool is_top_level_channel_ = true; - ManualConstructor<ChannelTrace> trace_; + CallCountingHelper call_counter_; + ChannelTrace trace_; }; -// Placeholds channelz class for subchannels. All this can do now is track its -// uuid (this information is needed by the parent channelz class). -// TODO(ncteisen): build this out to support the GetSubchannel channelz request. -class SubchannelNode : public RefCounted<SubchannelNode> { +// Handles channelz bookkeeping for servers +class ServerNode : public BaseNode { public: - SubchannelNode(); - virtual ~SubchannelNode(); + explicit ServerNode(size_t channel_tracer_max_nodes); + ~ServerNode() override; - intptr_t subchannel_uuid() { return subchannel_uuid_; } + grpc_json* RenderJson() override; - protected: - GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_DELETE - GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_NEW + // proxy methods to composed classes. + void AddTraceEvent(ChannelTrace::Severity severity, grpc_slice data) { + trace_.AddTraceEvent(severity, data); + } + void AddTraceEventWithReference(ChannelTrace::Severity severity, + grpc_slice data, + RefCountedPtr<BaseNode> referenced_channel) { + trace_.AddTraceEventWithReference(severity, data, + std::move(referenced_channel)); + } + void RecordCallStarted() { call_counter_.RecordCallStarted(); } + void RecordCallFailed() { call_counter_.RecordCallFailed(); } + void RecordCallSucceeded() { call_counter_.RecordCallSucceeded(); } private: - intptr_t subchannel_uuid_; + CallCountingHelper call_counter_; + ChannelTrace trace_; +}; + +// Handles channelz bookkeeping for sockets +// TODO(ncteisen): implement in subsequent PR. +class SocketNode : public BaseNode { + public: + SocketNode() : BaseNode(EntityType::kSocket) {} + ~SocketNode() override {} }; // Creation functions diff --git a/src/core/lib/channel/channelz_registry.cc b/src/core/lib/channel/channelz_registry.cc index f79d2f0c17..adc7b6ba44 100644 --- a/src/core/lib/channel/channelz_registry.cc +++ b/src/core/lib/channel/channelz_registry.cc @@ -53,54 +53,46 @@ ChannelzRegistry::ChannelzRegistry() { gpr_mu_init(&mu_); } ChannelzRegistry::~ChannelzRegistry() { gpr_mu_destroy(&mu_); } -intptr_t ChannelzRegistry::InternalRegisterEntry(const RegistryEntry& entry) { +intptr_t ChannelzRegistry::InternalRegister(BaseNode* node) { MutexLock lock(&mu_); - entities_.push_back(entry); + entities_.push_back(node); intptr_t uuid = entities_.size(); return uuid; } -void ChannelzRegistry::InternalUnregisterEntry(intptr_t uuid, EntityType type) { +void ChannelzRegistry::InternalUnregister(intptr_t uuid) { GPR_ASSERT(uuid >= 1); MutexLock lock(&mu_); GPR_ASSERT(static_cast<size_t>(uuid) <= entities_.size()); - GPR_ASSERT(entities_[uuid - 1].type == type); - entities_[uuid - 1].object = nullptr; - entities_[uuid - 1].type = EntityType::kUnset; + entities_[uuid - 1] = nullptr; } -void* ChannelzRegistry::InternalGetEntry(intptr_t uuid, EntityType type) { +BaseNode* ChannelzRegistry::InternalGet(intptr_t uuid) { MutexLock lock(&mu_); if (uuid < 1 || uuid > static_cast<intptr_t>(entities_.size())) { return nullptr; } - if (entities_[uuid - 1].type == type) { - return entities_[uuid - 1].object; - } else { - return nullptr; - } + return entities_[uuid - 1]; } char* ChannelzRegistry::InternalGetTopChannels(intptr_t start_channel_id) { grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); grpc_json* json = top_level_json; grpc_json* json_iterator = nullptr; - InlinedVector<ChannelNode*, 10> top_level_channels; + InlinedVector<BaseNode*, 10> top_level_channels; // uuids index into entities one-off (idx 0 is really uuid 1, since 0 is // reserved). However, we want to support requests coming in with // start_channel_id=0, which signifies "give me everything." Hence this // funky looking line below. size_t start_idx = start_channel_id == 0 ? 0 : start_channel_id - 1; for (size_t i = start_idx; i < entities_.size(); ++i) { - if (entities_[i].type == EntityType::kChannelNode) { - ChannelNode* channel_node = - static_cast<ChannelNode*>(entities_[i].object); - if (channel_node->is_top_level_channel()) { - top_level_channels.push_back(channel_node); - } + if (entities_[i] != nullptr && + entities_[i]->type() == + grpc_core::channelz::BaseNode::EntityType::kTopLevelChannel) { + top_level_channels.push_back(entities_[i]); } } - if (top_level_channels.size() > 0) { + if (!top_level_channels.empty()) { // create list of channels grpc_json* array_parent = grpc_json_create_child( nullptr, json, "channel", nullptr, GRPC_JSON_ARRAY, false); @@ -120,6 +112,42 @@ char* ChannelzRegistry::InternalGetTopChannels(intptr_t start_channel_id) { return json_str; } +char* ChannelzRegistry::InternalGetServers(intptr_t start_server_id) { + grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); + grpc_json* json = top_level_json; + grpc_json* json_iterator = nullptr; + InlinedVector<BaseNode*, 10> servers; + // uuids index into entities one-off (idx 0 is really uuid 1, since 0 is + // reserved). However, we want to support requests coming in with + // start_server_id=0, which signifies "give me everything." + size_t start_idx = start_server_id == 0 ? 0 : start_server_id - 1; + for (size_t i = start_idx; i < entities_.size(); ++i) { + if (entities_[i] != nullptr && + entities_[i]->type() == + grpc_core::channelz::BaseNode::EntityType::kServer) { + servers.push_back(entities_[i]); + } + } + if (!servers.empty()) { + // create list of servers + grpc_json* array_parent = grpc_json_create_child( + nullptr, json, "server", nullptr, GRPC_JSON_ARRAY, false); + for (size_t i = 0; i < servers.size(); ++i) { + grpc_json* server_json = servers[i]->RenderJson(); + json_iterator = + grpc_json_link_child(array_parent, server_json, json_iterator); + } + } + // For now we do not have any pagination rules. In the future we could + // pick a constant for max_channels_sent for a GetServers request. + // Tracking: https://github.com/grpc/grpc/issues/16019. + json_iterator = grpc_json_create_child(nullptr, json, "end", nullptr, + GRPC_JSON_TRUE, false); + char* json_str = grpc_json_dump_to_string(top_level_json, 0); + grpc_json_destroy(top_level_json); + return json_str; +} + } // namespace channelz } // namespace grpc_core @@ -128,10 +156,18 @@ char* grpc_channelz_get_top_channels(intptr_t start_channel_id) { start_channel_id); } +char* grpc_channelz_get_servers(intptr_t start_server_id) { + return grpc_core::channelz::ChannelzRegistry::GetServers(start_server_id); +} + char* grpc_channelz_get_channel(intptr_t channel_id) { - grpc_core::channelz::ChannelNode* channel_node = - grpc_core::channelz::ChannelzRegistry::GetChannelNode(channel_id); - if (channel_node == nullptr) { + grpc_core::channelz::BaseNode* channel_node = + grpc_core::channelz::ChannelzRegistry::Get(channel_id); + if (channel_node == nullptr || + (channel_node->type() != + grpc_core::channelz::BaseNode::EntityType::kTopLevelChannel && + channel_node->type() != + grpc_core::channelz::BaseNode::EntityType::kInternalChannel)) { return nullptr; } grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); @@ -143,3 +179,21 @@ char* grpc_channelz_get_channel(intptr_t channel_id) { grpc_json_destroy(top_level_json); return json_str; } + +char* grpc_channelz_get_subchannel(intptr_t subchannel_id) { + grpc_core::channelz::BaseNode* subchannel_node = + grpc_core::channelz::ChannelzRegistry::Get(subchannel_id); + if (subchannel_node == nullptr || + subchannel_node->type() != + grpc_core::channelz::BaseNode::EntityType::kSubchannel) { + return nullptr; + } + grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); + grpc_json* json = top_level_json; + grpc_json* subchannel_json = subchannel_node->RenderJson(); + subchannel_json->key = "subchannel"; + grpc_json_link_child(json, subchannel_json, nullptr); + char* json_str = grpc_json_dump_to_string(top_level_json, 0); + grpc_json_destroy(top_level_json); + return json_str; +} diff --git a/src/core/lib/channel/channelz_registry.h b/src/core/lib/channel/channelz_registry.h index 5d7c936726..d0d660600d 100644 --- a/src/core/lib/channel/channelz_registry.h +++ b/src/core/lib/channel/channelz_registry.h @@ -40,32 +40,11 @@ class ChannelzRegistry { // To be called in grpc_shutdown(); static void Shutdown(); - // Register/Unregister/Get for ChannelNode - static intptr_t RegisterChannelNode(ChannelNode* channel_node) { - RegistryEntry entry(channel_node, EntityType::kChannelNode); - return Default()->InternalRegisterEntry(entry); - } - static void UnregisterChannelNode(intptr_t uuid) { - Default()->InternalUnregisterEntry(uuid, EntityType::kChannelNode); - } - static ChannelNode* GetChannelNode(intptr_t uuid) { - void* gotten = Default()->InternalGetEntry(uuid, EntityType::kChannelNode); - return gotten == nullptr ? nullptr : static_cast<ChannelNode*>(gotten); - } - - // Register/Unregister/Get for SubchannelNode - static intptr_t RegisterSubchannelNode(SubchannelNode* channel_node) { - RegistryEntry entry(channel_node, EntityType::kSubchannelNode); - return Default()->InternalRegisterEntry(entry); - } - static void UnregisterSubchannelNode(intptr_t uuid) { - Default()->InternalUnregisterEntry(uuid, EntityType::kSubchannelNode); - } - static SubchannelNode* GetSubchannelNode(intptr_t uuid) { - void* gotten = - Default()->InternalGetEntry(uuid, EntityType::kSubchannelNode); - return gotten == nullptr ? nullptr : static_cast<SubchannelNode*>(gotten); + static intptr_t Register(BaseNode* node) { + return Default()->InternalRegister(node); } + static void Unregister(intptr_t uuid) { Default()->InternalUnregister(uuid); } + static BaseNode* Get(intptr_t uuid) { return Default()->InternalGet(uuid); } // Returns the allocated JSON string that represents the proto // GetTopChannelsResponse as per channelz.proto. @@ -73,20 +52,13 @@ class ChannelzRegistry { return Default()->InternalGetTopChannels(start_channel_id); } - private: - enum class EntityType { - kChannelNode, - kSubchannelNode, - kUnset, - }; - - struct RegistryEntry { - RegistryEntry(void* object_in, EntityType type_in) - : object(object_in), type(type_in) {} - void* object; - EntityType type; - }; + // Returns the allocated JSON string that represents the proto + // GetServersResponse as per channelz.proto. + static char* GetServers(intptr_t start_server_id) { + return Default()->InternalGetServers(start_server_id); + } + private: GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_NEW GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_DELETE @@ -97,21 +69,22 @@ class ChannelzRegistry { static ChannelzRegistry* Default(); // globally registers an Entry. Returns its unique uuid - intptr_t InternalRegisterEntry(const RegistryEntry& entry); + intptr_t InternalRegister(BaseNode* node); // globally unregisters the object that is associated to uuid. Also does // sanity check that an object doesn't try to unregister the wrong type. - void InternalUnregisterEntry(intptr_t uuid, EntityType type); + void InternalUnregister(intptr_t uuid); // if object with uuid has previously been registered as the correct type, // returns the void* associated with that uuid. Else returns nullptr. - void* InternalGetEntry(intptr_t uuid, EntityType type); + BaseNode* InternalGet(intptr_t uuid); char* InternalGetTopChannels(intptr_t start_channel_id); + char* InternalGetServers(intptr_t start_server_id); // protects entities_ and uuid_ gpr_mu mu_; - InlinedVector<RegistryEntry, 20> entities_; + InlinedVector<BaseNode*, 20> entities_; }; } // namespace channelz diff --git a/src/core/lib/iomgr/port.h b/src/core/lib/iomgr/port.h index abf96662f5..3d459059d7 100644 --- a/src/core/lib/iomgr/port.h +++ b/src/core/lib/iomgr/port.h @@ -82,6 +82,11 @@ #define GRPC_LINUX_SOCKETUTILS 1 #endif #endif +#ifdef LINUX_VERSION_CODE +#if LINUX_VERSION_CODE >= KERNEL_VERSION(2, 6, 37) +#define GRPC_HAVE_TCP_USER_TIMEOUT +#endif /* LINUX_VERSION_CODE >= KERNEL_VERSION(2, 6, 37) */ +#endif /* LINUX_VERSION_CODE */ #ifndef __GLIBC__ #define GRPC_LINUX_EPOLL 1 #define GRPC_LINUX_EPOLL_CREATE1 1 diff --git a/src/core/lib/iomgr/socket_utils_common_posix.cc b/src/core/lib/iomgr/socket_utils_common_posix.cc index c4b991c94d..50674b0845 100644 --- a/src/core/lib/iomgr/socket_utils_common_posix.cc +++ b/src/core/lib/iomgr/socket_utils_common_posix.cc @@ -41,6 +41,7 @@ #include <grpc/support/log.h> #include <grpc/support/sync.h> +#include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gpr/string.h" #include "src/core/lib/iomgr/sockaddr.h" @@ -222,6 +223,95 @@ grpc_error* grpc_set_socket_low_latency(int fd, int low_latency) { return GRPC_ERROR_NONE; } +/* The default values for TCP_USER_TIMEOUT are currently configured to be in + * line with the default values of KEEPALIVE_TIMEOUT as proposed in + * https://github.com/grpc/proposal/blob/master/A18-tcp-user-timeout.md */ +#define DEFAULT_CLIENT_TCP_USER_TIMEOUT_MS 20000 /* 20 seconds */ +#define DEFAULT_SERVER_TCP_USER_TIMEOUT_MS 20000 /* 20 seconds */ + +static int g_default_client_tcp_user_timeout_ms = + DEFAULT_CLIENT_TCP_USER_TIMEOUT_MS; +static int g_default_server_tcp_user_timeout_ms = + DEFAULT_SERVER_TCP_USER_TIMEOUT_MS; +static bool g_default_client_tcp_user_timeout_enabled = false; +static bool g_default_server_tcp_user_timeout_enabled = true; + +void config_default_tcp_user_timeout(bool enable, int timeout, bool is_client) { + if (is_client) { + g_default_client_tcp_user_timeout_enabled = enable; + if (timeout > 0) { + g_default_client_tcp_user_timeout_ms = timeout; + } + } else { + g_default_server_tcp_user_timeout_enabled = enable; + if (timeout > 0) { + g_default_server_tcp_user_timeout_ms = timeout; + } + } +} + +/* Set TCP_USER_TIMEOUT */ +grpc_error* grpc_set_socket_tcp_user_timeout( + int fd, const grpc_channel_args* channel_args, bool is_client) { +#ifdef GRPC_HAVE_TCP_USER_TIMEOUT + bool enable; + int timeout; + if (is_client) { + enable = g_default_client_tcp_user_timeout_enabled; + timeout = g_default_client_tcp_user_timeout_ms; + } else { + enable = g_default_server_tcp_user_timeout_enabled; + timeout = g_default_server_tcp_user_timeout_ms; + } + if (channel_args) { + for (unsigned int i = 0; i < channel_args->num_args; i++) { + if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_KEEPALIVE_TIME_MS)) { + const int value = grpc_channel_arg_get_integer( + &channel_args->args[i], grpc_integer_options{0, 1, INT_MAX}); + /* Continue using default if value is 0 */ + if (value == 0) { + continue; + } + /* Disable if value is INT_MAX */ + enable = value != INT_MAX; + } else if (0 == strcmp(channel_args->args[i].key, + GRPC_ARG_KEEPALIVE_TIMEOUT_MS)) { + const int value = grpc_channel_arg_get_integer( + &channel_args->args[i], grpc_integer_options{0, 1, INT_MAX}); + /* Continue using default if value is 0 */ + if (value == 0) { + continue; + } + timeout = value; + } + } + } + if (enable) { + extern grpc_core::TraceFlag grpc_tcp_trace; + if (grpc_tcp_trace.enabled()) { + gpr_log(GPR_INFO, "Enabling TCP_USER_TIMEOUT with a timeout of %d ms", + timeout); + } + int newval; + socklen_t len = sizeof(newval); + if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &timeout, + sizeof(timeout))) { + return GRPC_OS_ERROR(errno, "setsockopt(TCP_USER_TIMEOUT)"); + } + if (0 != getsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &newval, &len)) { + return GRPC_OS_ERROR(errno, "getsockopt(TCP_USER_TIMEOUT)"); + } + if (newval != timeout) { + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Failed to set TCP_USER_TIMEOUT"); + } + } +#else + gpr_log(GPR_INFO, "TCP_USER_TIMEOUT not supported for this platform"); +#endif /* GRPC_HAVE_TCP_USER_TIMEOUT */ + return GRPC_ERROR_NONE; +} + /* set a socket using a grpc_socket_mutator */ grpc_error* grpc_set_socket_with_mutator(int fd, grpc_socket_mutator* mutator) { GPR_ASSERT(mutator); diff --git a/src/core/lib/iomgr/socket_utils_posix.h b/src/core/lib/iomgr/socket_utils_posix.h index b3fd58a530..71a304dc4e 100644 --- a/src/core/lib/iomgr/socket_utils_posix.h +++ b/src/core/lib/iomgr/socket_utils_posix.h @@ -53,6 +53,13 @@ grpc_error* grpc_set_socket_low_latency(int fd, int low_latency); /* set SO_REUSEPORT */ grpc_error* grpc_set_socket_reuse_port(int fd, int reuse); +/* Configure the default values for TCP_USER_TIMEOUT */ +void config_default_tcp_user_timeout(bool enable, int timeout, bool is_client); + +/* Set TCP_USER_TIMEOUT */ +grpc_error* grpc_set_socket_tcp_user_timeout( + int fd, const grpc_channel_args* channel_args, bool is_client); + /* Returns true if this system can create AF_INET6 sockets bound to ::1. The value is probed once, and cached for the life of the process. diff --git a/src/core/lib/iomgr/tcp_client_posix.cc b/src/core/lib/iomgr/tcp_client_posix.cc index 9c989b7dfe..8553ed0db4 100644 --- a/src/core/lib/iomgr/tcp_client_posix.cc +++ b/src/core/lib/iomgr/tcp_client_posix.cc @@ -76,6 +76,9 @@ static grpc_error* prepare_socket(const grpc_resolved_address* addr, int fd, if (!grpc_is_unix_socket(addr)) { err = grpc_set_socket_low_latency(fd, 1); if (err != GRPC_ERROR_NONE) goto error; + err = grpc_set_socket_tcp_user_timeout(fd, channel_args, + true /* is_client */); + if (err != GRPC_ERROR_NONE) goto error; } err = grpc_set_socket_no_sigpipe_if_possible(fd); if (err != GRPC_ERROR_NONE) goto error; diff --git a/src/core/lib/iomgr/tcp_server_utils_posix_common.cc b/src/core/lib/iomgr/tcp_server_utils_posix_common.cc index 9595c028ce..8d8d3f4273 100644 --- a/src/core/lib/iomgr/tcp_server_utils_posix_common.cc +++ b/src/core/lib/iomgr/tcp_server_utils_posix_common.cc @@ -166,6 +166,9 @@ grpc_error* grpc_tcp_server_prepare_socket(grpc_tcp_server* s, int fd, if (err != GRPC_ERROR_NONE) goto error; err = grpc_set_socket_reuse_addr(fd, 1); if (err != GRPC_ERROR_NONE) goto error; + err = grpc_set_socket_tcp_user_timeout(fd, s->channel_args, + false /* is_client */); + if (err != GRPC_ERROR_NONE) goto error; } err = grpc_set_socket_no_sigpipe_if_possible(fd); if (err != GRPC_ERROR_NONE) goto error; diff --git a/src/core/lib/surface/call.cc b/src/core/lib/surface/call.cc index b07c4d6c10..11b438f5dc 100644 --- a/src/core/lib/surface/call.cc +++ b/src/core/lib/surface/call.cc @@ -48,6 +48,7 @@ #include "src/core/lib/surface/call_test_only.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/completion_queue.h" +#include "src/core/lib/surface/server.h" #include "src/core/lib/surface/validate_metadata.h" #include "src/core/lib/transport/error_utils.h" #include "src/core/lib/transport/metadata.h" @@ -95,7 +96,7 @@ typedef struct batch_control { grpc_closure start_batch; grpc_closure finish_batch; gpr_refcount steps_to_complete; - grpc_error* batch_error; + gpr_atm batch_error; grpc_transport_stream_op_batch op; } batch_control; @@ -202,6 +203,8 @@ struct grpc_call { } client; struct { int* cancelled; + // backpointer to owning server if this is a server side call. + grpc_server* server; } server; } final_op; grpc_error* status_error; @@ -311,14 +314,10 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, /* Always support no compression */ GPR_BITSET(&call->encodings_accepted_by_peer, GRPC_MESSAGE_COMPRESS_NONE); call->is_client = args->server_transport_data == nullptr; - if (call->is_client) { - GRPC_STATS_INC_CLIENT_CALLS_CREATED(); - } else { - GRPC_STATS_INC_SERVER_CALLS_CREATED(); - } call->stream_op_payload.context = call->context; grpc_slice path = grpc_empty_slice(); if (call->is_client) { + GRPC_STATS_INC_CLIENT_CALLS_CREATED(); GPR_ASSERT(args->add_initial_metadata_count < MAX_SEND_EXTRA_METADATA_COUNT); for (i = 0; i < args->add_initial_metadata_count; i++) { @@ -332,6 +331,8 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, call->send_extra_metadata_count = static_cast<int>(args->add_initial_metadata_count); } else { + GRPC_STATS_INC_SERVER_CALLS_CREATED(); + call->final_op.server.server = args->server; GPR_ASSERT(args->add_initial_metadata_count == 0); call->send_extra_metadata_count = 0; } @@ -435,10 +436,18 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, &call->pollent); } - grpc_core::channelz::ChannelNode* channelz_channel = - grpc_channel_get_channelz_node(call->channel); - if (channelz_channel != nullptr) { - channelz_channel->RecordCallStarted(); + if (call->is_client) { + grpc_core::channelz::ChannelNode* channelz_channel = + grpc_channel_get_channelz_node(call->channel); + if (channelz_channel != nullptr) { + channelz_channel->RecordCallStarted(); + } + } else { + grpc_core::channelz::ServerNode* channelz_server = + grpc_server_get_channelz_node(call->final_op.server.server); + if (channelz_server != nullptr) { + channelz_server->RecordCallStarted(); + } } grpc_slice_unref_internal(path); @@ -709,14 +718,15 @@ static void set_final_status(grpc_call* call, grpc_error* error) { } else { *call->final_op.server.cancelled = error != GRPC_ERROR_NONE || call->status_error != GRPC_ERROR_NONE; - /* TODO(ncteisen) : Update channelz handling for server - if (channelz_channel != nullptr) { + grpc_core::channelz::ServerNode* channelz_server = + grpc_server_get_channelz_node(call->final_op.server.server); + if (channelz_server != nullptr) { if (*call->final_op.server.cancelled) { - channelz_channel->RecordCallFailed(); + channelz_server->RecordCallFailed(); } else { - channelz_channel->RecordCallSucceeded(); + channelz_server->RecordCallSucceeded(); } - } */ + } GRPC_ERROR_UNREF(error); } } @@ -1106,14 +1116,17 @@ static void finish_batch_completion(void* user_data, } static void reset_batch_errors(batch_control* bctl) { - GRPC_ERROR_UNREF(bctl->batch_error); - bctl->batch_error = GRPC_ERROR_NONE; + GRPC_ERROR_UNREF( + reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error))); + gpr_atm_rel_store(&bctl->batch_error, + reinterpret_cast<gpr_atm>(GRPC_ERROR_NONE)); } static void post_batch_completion(batch_control* bctl) { grpc_call* next_child_call; grpc_call* call = bctl->call; - grpc_error* error = GRPC_ERROR_REF(bctl->batch_error); + grpc_error* error = GRPC_ERROR_REF( + reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error))); if (bctl->op.send_initial_metadata) { grpc_metadata_batch_destroy( @@ -1277,8 +1290,10 @@ static void receiving_stream_ready(void* bctlp, grpc_error* error) { grpc_call* call = bctl->call; if (error != GRPC_ERROR_NONE) { call->receiving_stream.reset(); - if (bctl->batch_error == GRPC_ERROR_NONE) { - bctl->batch_error = GRPC_ERROR_REF(error); + if (reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)) == + GRPC_ERROR_NONE) { + gpr_atm_rel_store(&bctl->batch_error, + reinterpret_cast<gpr_atm>(GRPC_ERROR_REF(error))); } cancel_with_error(call, GRPC_ERROR_REF(error)); } @@ -1384,8 +1399,10 @@ static void receiving_initial_metadata_ready(void* bctlp, grpc_error* error) { call->send_deadline = md->deadline; } } else { - if (bctl->batch_error == GRPC_ERROR_NONE) { - bctl->batch_error = GRPC_ERROR_REF(error); + if (reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)) == + GRPC_ERROR_NONE) { + gpr_atm_rel_store(&bctl->batch_error, + reinterpret_cast<gpr_atm>(GRPC_ERROR_REF(error))); } cancel_with_error(call, GRPC_ERROR_REF(error)); } @@ -1435,8 +1452,10 @@ static void finish_batch(void* bctlp, grpc_error* error) { batch_control* bctl = static_cast<batch_control*>(bctlp); grpc_call* call = bctl->call; GRPC_CALL_COMBINER_STOP(&call->call_combiner, "on_complete"); - if (bctl->batch_error == GRPC_ERROR_NONE) { - bctl->batch_error = GRPC_ERROR_REF(error); + if (reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&bctl->batch_error)) == + GRPC_ERROR_NONE) { + gpr_atm_rel_store(&bctl->batch_error, + reinterpret_cast<gpr_atm>(GRPC_ERROR_REF(error))); } if (error != GRPC_ERROR_NONE) { cancel_with_error(call, GRPC_ERROR_REF(error)); diff --git a/src/core/lib/surface/call.h b/src/core/lib/surface/call.h index b3b06059d4..b34260505a 100644 --- a/src/core/lib/surface/call.h +++ b/src/core/lib/surface/call.h @@ -33,6 +33,7 @@ typedef void (*grpc_ioreq_completion_func)(grpc_call* call, int success, typedef struct grpc_call_create_args { grpc_channel* channel; + grpc_server* server; grpc_call* parent; uint32_t propagation_mask; diff --git a/src/core/lib/surface/channel.cc b/src/core/lib/surface/channel.cc index e3dde51d69..054fe105c3 100644 --- a/src/core/lib/surface/channel.cc +++ b/src/core/lib/surface/channel.cc @@ -165,11 +165,12 @@ grpc_channel* grpc_channel_create_with_builder( } grpc_channel_args_destroy(args); - if (channelz_enabled) { - bool is_top_level_channel = channel->is_client && !internal_channel; + // we only need to do the channelz bookkeeping for clients here. The channelz + // bookkeeping for server channels occurs in src/core/lib/surface/server.cc + if (channelz_enabled && channel->is_client) { channel->channelz_channel = channel_node_create_func( - channel, channel_tracer_max_nodes, is_top_level_channel); - channel->channelz_channel->trace()->AddTraceEvent( + channel, channel_tracer_max_nodes, !internal_channel); + channel->channelz_channel->AddTraceEvent( grpc_core::channelz::ChannelTrace::Severity::Info, grpc_slice_from_static_string("Channel created")); } @@ -427,6 +428,9 @@ void grpc_channel_internal_unref(grpc_channel* c REF_ARG) { static void destroy_channel(void* arg, grpc_error* error) { grpc_channel* channel = static_cast<grpc_channel*>(arg); if (channel->channelz_channel != nullptr) { + channel->channelz_channel->AddTraceEvent( + grpc_core::channelz::ChannelTrace::Severity::Info, + grpc_slice_from_static_string("Channel destroyed")); channel->channelz_channel->MarkChannelDestroyed(); channel->channelz_channel.reset(); } diff --git a/src/core/lib/surface/completion_queue.cc b/src/core/lib/surface/completion_queue.cc index 0769d9e4f6..c2cf450e94 100644 --- a/src/core/lib/surface/completion_queue.cc +++ b/src/core/lib/surface/completion_queue.cc @@ -1364,9 +1364,11 @@ static void cq_shutdown_callback(grpc_completion_queue* cq) { } cqd->shutdown_called = true; if (gpr_atm_full_fetch_add(&cqd->pending_events, -1) == 1) { + gpr_mu_unlock(cq->mu); cq_finish_shutdown_callback(cq); + } else { + gpr_mu_unlock(cq->mu); } - gpr_mu_unlock(cq->mu); GRPC_CQ_INTERNAL_UNREF(cq, "shutting_down (callback cq)"); } diff --git a/src/core/lib/surface/server.cc b/src/core/lib/surface/server.cc index 521825ca69..5fa58ffdec 100644 --- a/src/core/lib/surface/server.cc +++ b/src/core/lib/surface/server.cc @@ -222,6 +222,8 @@ struct grpc_server { /** when did we print the last shutdown progress message */ gpr_timespec last_shutdown_message_time; + + grpc_core::RefCountedPtr<grpc_core::channelz::ServerNode> channelz_server; }; #define SERVER_FROM_CALL_ELEM(elem) \ @@ -367,6 +369,7 @@ static void server_ref(grpc_server* server) { static void server_delete(grpc_server* server) { registered_method* rm; size_t i; + server->channelz_server.reset(); grpc_channel_args_destroy(server->channel_args); gpr_mu_destroy(&server->mu_global); gpr_mu_destroy(&server->mu_call); @@ -796,6 +799,7 @@ static void accept_stream(void* cd, grpc_transport* transport, args.channel = chand->channel; args.server_transport_data = transport_server_data; args.send_deadline = GRPC_MILLIS_INF_FUTURE; + args.server = chand->server; grpc_call* call; grpc_error* error = grpc_call_create(&args, &call); grpc_call_element* elem = @@ -960,6 +964,7 @@ void grpc_server_register_completion_queue(grpc_server* server, } grpc_server* grpc_server_create(const grpc_channel_args* args, void* reserved) { + grpc_core::ExecCtx exec_ctx; GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved)); grpc_server* server = @@ -976,6 +981,20 @@ grpc_server* grpc_server_create(const grpc_channel_args* args, void* reserved) { server->channel_args = grpc_channel_args_copy(args); + const grpc_arg* arg = grpc_channel_args_find(args, GRPC_ARG_ENABLE_CHANNELZ); + if (grpc_channel_arg_get_bool(arg, false)) { + arg = grpc_channel_args_find(args, + GRPC_ARG_MAX_CHANNEL_TRACE_EVENTS_PER_NODE); + size_t trace_events_per_node = + grpc_channel_arg_get_integer(arg, {0, 0, INT_MAX}); + server->channelz_server = + grpc_core::MakeRefCounted<grpc_core::channelz::ServerNode>( + trace_events_per_node); + server->channelz_server->AddTraceEvent( + grpc_core::channelz::ChannelTrace::Severity::Info, + grpc_slice_from_static_string("Server created")); + } + return server; } @@ -1478,3 +1497,11 @@ int grpc_server_has_open_connections(grpc_server* server) { gpr_mu_unlock(&server->mu_global); return r; } + +grpc_core::channelz::ServerNode* grpc_server_get_channelz_node( + grpc_server* server) { + if (server == nullptr) { + return nullptr; + } + return server->channelz_server.get(); +} diff --git a/src/core/lib/surface/server.h b/src/core/lib/surface/server.h index c617cc223e..0196743ff9 100644 --- a/src/core/lib/surface/server.h +++ b/src/core/lib/surface/server.h @@ -23,6 +23,7 @@ #include <grpc/grpc.h> #include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/channel/channelz.h" #include "src/core/lib/debug/trace.h" #include "src/core/lib/transport/transport.h" @@ -46,6 +47,9 @@ void grpc_server_setup_transport(grpc_server* server, grpc_transport* transport, grpc_pollset* accepting_pollset, const grpc_channel_args* args); +grpc_core::channelz::ServerNode* grpc_server_get_channelz_node( + grpc_server* server); + const grpc_channel_args* grpc_server_get_channel_args(grpc_server* server); int grpc_server_has_open_connections(grpc_server* server); diff --git a/src/core/tsi/alts/handshaker/alts_handshaker_client.cc b/src/core/tsi/alts/handshaker/alts_handshaker_client.cc index b5268add0d..17e8026096 100644 --- a/src/core/tsi/alts/handshaker/alts_handshaker_client.cc +++ b/src/core/tsi/alts/handshaker/alts_handshaker_client.cc @@ -24,6 +24,7 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include "src/core/lib/slice/slice_internal.h" #include "src/core/tsi/alts/handshaker/alts_handshaker_service_api.h" const int kHandshakerClientOpNum = 4; @@ -109,7 +110,7 @@ static grpc_byte_buffer* get_serialized_start_client(alts_tsi_event* event) { if (ok) { buffer = grpc_raw_byte_buffer_create(&slice, 1 /* number of slices */); } - grpc_slice_unref(slice); + grpc_slice_unref_internal(slice); gpr_free(target_name); grpc_gcp_handshaker_req_destroy(req); return buffer; @@ -157,7 +158,7 @@ static grpc_byte_buffer* get_serialized_start_server( if (ok) { buffer = grpc_raw_byte_buffer_create(&req_slice, 1 /* number of slices */); } - grpc_slice_unref(req_slice); + grpc_slice_unref_internal(req_slice); grpc_gcp_handshaker_req_destroy(req); return buffer; } @@ -195,7 +196,7 @@ static grpc_byte_buffer* get_serialized_next(grpc_slice* bytes_received) { if (ok) { buffer = grpc_raw_byte_buffer_create(&req_slice, 1 /* number of slices */); } - grpc_slice_unref(req_slice); + grpc_slice_unref_internal(req_slice); grpc_gcp_handshaker_req_destroy(req); return buffer; } @@ -258,7 +259,7 @@ alts_handshaker_client* alts_grpc_handshaker_client_create( grpc_slice_from_static_string(ALTS_SERVICE_METHOD), &slice, gpr_inf_future(GPR_CLOCK_REALTIME), nullptr); client->base.vtable = &vtable; - grpc_slice_unref(slice); + grpc_slice_unref_internal(slice); return &client->base; } diff --git a/src/core/tsi/alts/handshaker/alts_handshaker_service_api_util.cc b/src/core/tsi/alts/handshaker/alts_handshaker_service_api_util.cc index e0e4184686..d63d3538c5 100644 --- a/src/core/tsi/alts/handshaker/alts_handshaker_service_api_util.cc +++ b/src/core/tsi/alts/handshaker/alts_handshaker_service_api_util.cc @@ -20,6 +20,8 @@ #include "src/core/tsi/alts/handshaker/alts_handshaker_service_api_util.h" +#include "src/core/lib/slice/slice_internal.h" + void add_repeated_field(repeated_field** head, const void* data) { repeated_field* field = static_cast<repeated_field*>(gpr_zalloc(sizeof(*field))); @@ -67,7 +69,7 @@ grpc_slice* create_slice(const char* data, size_t size) { void destroy_slice(grpc_slice* slice) { if (slice != nullptr) { - grpc_slice_unref(*slice); + grpc_slice_unref_internal(*slice); gpr_free(slice); } } diff --git a/src/core/tsi/alts/handshaker/alts_tsi_event.cc b/src/core/tsi/alts/handshaker/alts_tsi_event.cc index ec0bf12b95..cb36d5ebd1 100644 --- a/src/core/tsi/alts/handshaker/alts_tsi_event.cc +++ b/src/core/tsi/alts/handshaker/alts_tsi_event.cc @@ -24,6 +24,8 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include "src/core/lib/slice/slice_internal.h" + tsi_result alts_tsi_event_create(alts_tsi_handshaker* handshaker, tsi_handshaker_on_next_done_cb cb, void* user_data, @@ -66,8 +68,8 @@ void alts_tsi_event_destroy(alts_tsi_event* event) { grpc_byte_buffer_destroy(event->recv_buffer); grpc_metadata_array_destroy(&event->initial_metadata); grpc_metadata_array_destroy(&event->trailing_metadata); - grpc_slice_unref(event->details); - grpc_slice_unref(event->target_name); + grpc_slice_unref_internal(event->details); + grpc_slice_unref_internal(event->target_name); grpc_alts_credentials_options_destroy(event->options); gpr_free(event); } diff --git a/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc b/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc index 1df1021bb1..34608a3de1 100644 --- a/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc +++ b/src/core/tsi/alts/handshaker/alts_tsi_handshaker.cc @@ -31,6 +31,7 @@ #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gprpp/thd.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/tsi/alts/frame_protector/alts_frame_protector.h" #include "src/core/tsi/alts/handshaker/alts_handshaker_client.h" #include "src/core/tsi/alts/handshaker/alts_tsi_utils.h" @@ -182,7 +183,7 @@ static void handshaker_result_destroy(tsi_handshaker_result* self) { gpr_free(result->peer_identity); gpr_free(result->key_data); gpr_free(result->unused_bytes); - grpc_slice_unref(result->rpc_versions); + grpc_slice_unref_internal(result->rpc_versions); gpr_free(result); } @@ -269,12 +270,12 @@ static tsi_result handshaker_next( handshaker->has_sent_start_message = true; } else { if (!GRPC_SLICE_IS_EMPTY(handshaker->recv_bytes)) { - grpc_slice_unref(handshaker->recv_bytes); + grpc_slice_unref_internal(handshaker->recv_bytes); } handshaker->recv_bytes = grpc_slice_ref(slice); ok = alts_handshaker_client_next(handshaker->client, event, &slice); } - grpc_slice_unref(slice); + grpc_slice_unref_internal(slice); if (ok != TSI_OK) { gpr_log(GPR_ERROR, "Failed to schedule ALTS handshaker requests"); return ok; @@ -299,8 +300,8 @@ static void handshaker_destroy(tsi_handshaker* self) { alts_tsi_handshaker* handshaker = reinterpret_cast<alts_tsi_handshaker*>(self); alts_handshaker_client_destroy(handshaker->client); - grpc_slice_unref(handshaker->recv_bytes); - grpc_slice_unref(handshaker->target_name); + grpc_slice_unref_internal(handshaker->recv_bytes); + grpc_slice_unref_internal(handshaker->target_name); grpc_alts_credentials_options_destroy(handshaker->options); gpr_free(handshaker->buffer); gpr_free(handshaker); diff --git a/src/core/tsi/alts/handshaker/alts_tsi_utils.cc b/src/core/tsi/alts/handshaker/alts_tsi_utils.cc index d9b5e6c945..1747f1ad04 100644 --- a/src/core/tsi/alts/handshaker/alts_tsi_utils.cc +++ b/src/core/tsi/alts/handshaker/alts_tsi_utils.cc @@ -22,6 +22,8 @@ #include <grpc/byte_buffer_reader.h> +#include "src/core/lib/slice/slice_internal.h" + tsi_result alts_tsi_utils_convert_to_tsi_result(grpc_status_code code) { switch (code) { case GRPC_STATUS_OK: @@ -47,7 +49,7 @@ grpc_gcp_handshaker_resp* alts_tsi_utils_deserialize_response( grpc_slice slice = grpc_byte_buffer_reader_readall(&bbr); grpc_gcp_handshaker_resp* resp = grpc_gcp_handshaker_resp_create(); bool ok = grpc_gcp_handshaker_resp_decode(slice, resp); - grpc_slice_unref(slice); + grpc_slice_unref_internal(slice); grpc_byte_buffer_reader_destroy(&bbr); if (!ok) { grpc_gcp_handshaker_resp_destroy(resp); diff --git a/src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_privacy_integrity_record_protocol.cc b/src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_privacy_integrity_record_protocol.cc index d4fd88d1e2..e7890903d5 100644 --- a/src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_privacy_integrity_record_protocol.cc +++ b/src/core/tsi/alts/zero_copy_frame_protector/alts_grpc_privacy_integrity_record_protocol.cc @@ -61,7 +61,7 @@ static tsi_result alts_grpc_privacy_integrity_protect( if (status != GRPC_STATUS_OK) { gpr_log(GPR_ERROR, "Failed to protect, %s", error_details); gpr_free(error_details); - grpc_slice_unref(protected_slice); + grpc_slice_unref_internal(protected_slice); return TSI_INTERNAL_ERROR; } grpc_slice_buffer_add(protected_slices, protected_slice); @@ -106,7 +106,7 @@ static tsi_result alts_grpc_privacy_integrity_unprotect( if (status != GRPC_STATUS_OK) { gpr_log(GPR_ERROR, "Failed to unprotect, %s", error_details); gpr_free(error_details); - grpc_slice_unref(unprotected_slice); + grpc_slice_unref_internal(unprotected_slice); return TSI_INTERNAL_ERROR; } grpc_slice_buffer_reset_and_unref_internal(&rp->header_sb); diff --git a/src/core/tsi/ssl/session_cache/ssl_session_cache.cc b/src/core/tsi/ssl/session_cache/ssl_session_cache.cc index ce74fde343..f9184bcc34 100644 --- a/src/core/tsi/ssl/session_cache/ssl_session_cache.cc +++ b/src/core/tsi/ssl/session_cache/ssl_session_cache.cc @@ -19,6 +19,7 @@ #include <grpc/support/port_platform.h> #include "src/core/lib/gprpp/mutex_lock.h" +#include "src/core/lib/slice/slice_internal.h" #include "src/core/tsi/ssl/session_cache/ssl_session.h" #include "src/core/tsi/ssl/session_cache/ssl_session_cache.h" @@ -53,7 +54,7 @@ class SslSessionLRUCache::Node { SetSession(std::move(session)); } - ~Node() { grpc_slice_unref(key_); } + ~Node() { grpc_slice_unref_internal(key_); } // Not copyable nor movable. Node(const Node&) = delete; diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc index 39b891c2e1..ad71286e05 100644 --- a/src/cpp/client/channel_cc.cc +++ b/src/cpp/client/channel_cc.cc @@ -42,8 +42,10 @@ #include <grpcpp/support/time.h> #include "src/core/lib/gpr/env.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/gprpp/memory.h" #include "src/core/lib/gprpp/thd.h" #include "src/core/lib/profiling/timers.h" +#include "src/core/lib/surface/completion_queue.h" namespace grpc { @@ -53,7 +55,12 @@ Channel::Channel(const grpc::string& host, grpc_channel* channel) g_gli_initializer.summon(); } -Channel::~Channel() { grpc_channel_destroy(c_channel_); } +Channel::~Channel() { + grpc_channel_destroy(c_channel_); + if (callback_cq_ != nullptr) { + callback_cq_->Shutdown(); + } +} namespace { @@ -135,8 +142,8 @@ void Channel::PerformOpsOnCall(internal::CallOpSetInterface* ops, size_t nops = 0; grpc_op cops[MAX_OPS]; ops->FillOps(call->call(), cops, &nops); - GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_batch(call->call(), cops, nops, ops, nullptr)); + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(call->call(), cops, nops, + ops->cq_tag(), nullptr)); } void* Channel::RegisterMethod(const char* method) { @@ -185,4 +192,39 @@ bool Channel::WaitForStateChangeImpl(grpc_connectivity_state last_observed, return ok; } +namespace { +class ShutdownCallback : public grpc_core::CQCallbackInterface { + public: + // TakeCQ takes ownership of the cq into the shutdown callback + // so that the shutdown callback will be responsible for destroying it + void TakeCQ(CompletionQueue* cq) { cq_ = cq; } + + // The Run function will get invoked by the completion queue library + // when the shutdown is actually complete + void Run(bool) override { + delete cq_; + grpc_core::Delete(this); + } + + private: + CompletionQueue* cq_ = nullptr; +}; +} // namespace + +CompletionQueue* Channel::CallbackCQ() { + // TODO(vjpai): Consider using a single global CQ for the default CQ + // if there is no explicit per-channel CQ registered + std::lock_guard<std::mutex> l(mu_); + if (callback_cq_ == nullptr) { + auto* shutdown_callback = grpc_core::New<ShutdownCallback>(); + callback_cq_ = new CompletionQueue(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING, + shutdown_callback}); + + // Transfer ownership of the new cq to its own shutdown callback + shutdown_callback->TakeCQ(callback_cq_); + } + return callback_cq_; +} + } // namespace grpc diff --git a/src/cpp/client/generic_stub.cc b/src/cpp/client/generic_stub.cc index 67ef46bebe..87902b26f0 100644 --- a/src/cpp/client/generic_stub.cc +++ b/src/cpp/client/generic_stub.cc @@ -16,9 +16,11 @@ * */ -#include <grpcpp/generic/generic_stub.h> +#include <functional> +#include <grpcpp/generic/generic_stub.h> #include <grpcpp/impl/rpc_method.h> +#include <grpcpp/support/client_callback.h> namespace grpc { @@ -60,4 +62,14 @@ std::unique_ptr<GenericClientAsyncResponseReader> GenericStub::PrepareUnaryCall( context, request, false)); } +void GenericStub::experimental_type::UnaryCall( + ClientContext* context, const grpc::string& method, + const ByteBuffer* request, ByteBuffer* response, + std::function<void(Status)> on_completion) { + internal::CallbackUnaryCall( + stub_->channel_.get(), + internal::RpcMethod(method.c_str(), internal::RpcMethod::NORMAL_RPC), + context, request, response, std::move(on_completion)); +} + } // namespace grpc diff --git a/src/cpp/common/callback_common.cc b/src/cpp/common/callback_common.cc new file mode 100644 index 0000000000..ae47901f1b --- /dev/null +++ b/src/cpp/common/callback_common.cc @@ -0,0 +1,131 @@ +/* + * + * 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 <functional> + +#include <grpcpp/impl/codegen/callback_common.h> +#include <grpcpp/impl/codegen/status.h> + +#include "src/core/lib/gprpp/memory.h" +#include "src/core/lib/surface/completion_queue.h" + +namespace grpc { +namespace internal { + +namespace { +class CallbackWithSuccessImpl : public grpc_core::CQCallbackInterface { + public: + static void operator delete(void* ptr, std::size_t size) { + assert(size == sizeof(CallbackWithSuccessImpl)); + } + + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + + CallbackWithSuccessImpl(grpc_call* call, CallbackWithSuccessTag* parent, + std::function<void(bool)> f) + : call_(call), parent_(parent), func_(std::move(f)) { + grpc_call_ref(call); + } + + void Run(bool ok) override { + void* ignored = parent_->ops(); + bool new_ok = ok; + GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &new_ok)); + GPR_ASSERT(ignored == parent_->ops()); + func_(ok); + func_ = nullptr; // release the function + grpc_call_unref(call_); + } + + private: + grpc_call* call_; + CallbackWithSuccessTag* parent_; + std::function<void(bool)> func_; +}; + +class CallbackWithStatusImpl : public grpc_core::CQCallbackInterface { + public: + static void operator delete(void* ptr, std::size_t size) { + assert(size == sizeof(CallbackWithStatusImpl)); + } + + // This operator should never be called as the memory should be freed as part + // of the arena destruction. It only exists to provide a matching operator + // delete to the operator new so that some compilers will not complain (see + // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this + // there are no tests catching the compiler warning. + static void operator delete(void*, void*) { assert(0); } + + CallbackWithStatusImpl(grpc_call* call, CallbackWithStatusTag* parent, + std::function<void(Status)> f) + : call_(call), parent_(parent), func_(std::move(f)), status_() { + grpc_call_ref(call); + } + + void Run(bool ok) override { + void* ignored = parent_->ops(); + + GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &ok)); + GPR_ASSERT(ignored == parent_->ops()); + + func_(status_); + func_ = nullptr; // release the function + grpc_call_unref(call_); + } + Status* status_ptr() { return &status_; } + + private: + grpc_call* call_; + CallbackWithStatusTag* parent_; + std::function<void(Status)> func_; + Status status_; +}; + +} // namespace + +CallbackWithSuccessTag::CallbackWithSuccessTag(grpc_call* call, + std::function<void(bool)> f, + CompletionQueueTag* ops) + : impl_(new (grpc_call_arena_alloc(call, sizeof(CallbackWithSuccessImpl))) + CallbackWithSuccessImpl(call, this, std::move(f))), + ops_(ops) {} + +void CallbackWithSuccessTag::force_run(bool ok) { impl_->Run(ok); } + +CallbackWithStatusTag::CallbackWithStatusTag(grpc_call* call, + std::function<void(Status)> f, + CompletionQueueTag* ops) + : ops_(ops) { + auto* impl = new (grpc_call_arena_alloc(call, sizeof(CallbackWithStatusImpl))) + CallbackWithStatusImpl(call, this, std::move(f)); + impl_ = impl; + status_ = impl->status_ptr(); +} + +void CallbackWithStatusTag::force_run(Status s) { + *status_ = std::move(s); + impl_->Run(true); +} + +} // namespace internal +} // namespace grpc diff --git a/src/cpp/server/channelz/channelz_service.cc b/src/cpp/server/channelz/channelz_service.cc index 77c175e5b8..e096c1f421 100644 --- a/src/cpp/server/channelz/channelz_service.cc +++ b/src/cpp/server/channelz/channelz_service.cc @@ -32,6 +32,25 @@ Status ChannelzService::GetTopChannels( ServerContext* unused, const channelz::v1::GetTopChannelsRequest* request, channelz::v1::GetTopChannelsResponse* response) { char* json_str = grpc_channelz_get_top_channels(request->start_channel_id()); + if (json_str == nullptr) { + return Status(INTERNAL, "grpc_channelz_get_top_channels returned null"); + } + google::protobuf::util::Status s = + google::protobuf::util::JsonStringToMessage(json_str, response); + gpr_free(json_str); + if (s != google::protobuf::util::Status::OK) { + return Status(INTERNAL, s.ToString()); + } + return Status::OK; +} + +Status ChannelzService::GetServers( + ServerContext* unused, const channelz::v1::GetServersRequest* request, + channelz::v1::GetServersResponse* response) { + char* json_str = grpc_channelz_get_servers(request->start_server_id()); + if (json_str == nullptr) { + return Status(INTERNAL, "grpc_channelz_get_servers returned null"); + } google::protobuf::util::Status s = google::protobuf::util::JsonStringToMessage(json_str, response); gpr_free(json_str); @@ -45,6 +64,25 @@ Status ChannelzService::GetChannel( ServerContext* unused, const channelz::v1::GetChannelRequest* request, channelz::v1::GetChannelResponse* response) { char* json_str = grpc_channelz_get_channel(request->channel_id()); + if (json_str == nullptr) { + return Status(NOT_FOUND, "No object found for that ChannelId"); + } + google::protobuf::util::Status s = + google::protobuf::util::JsonStringToMessage(json_str, response); + gpr_free(json_str); + if (s != google::protobuf::util::Status::OK) { + return Status(INTERNAL, s.ToString()); + } + return Status::OK; +} + +Status ChannelzService::GetSubchannel( + ServerContext* unused, const channelz::v1::GetSubchannelRequest* request, + channelz::v1::GetSubchannelResponse* response) { + char* json_str = grpc_channelz_get_subchannel(request->subchannel_id()); + if (json_str == nullptr) { + return Status(NOT_FOUND, "No object found for that SubchannelId"); + } google::protobuf::util::Status s = google::protobuf::util::JsonStringToMessage(json_str, response); gpr_free(json_str); diff --git a/src/cpp/server/channelz/channelz_service.h b/src/cpp/server/channelz/channelz_service.h index f619ea49e0..9e0b5b6ead 100644 --- a/src/cpp/server/channelz/channelz_service.h +++ b/src/cpp/server/channelz/channelz_service.h @@ -32,10 +32,18 @@ class ChannelzService final : public channelz::v1::Channelz::Service { Status GetTopChannels( ServerContext* unused, const channelz::v1::GetTopChannelsRequest* request, channelz::v1::GetTopChannelsResponse* response) override; + // implementation of GetServers rpc + Status GetServers(ServerContext* unused, + const channelz::v1::GetServersRequest* request, + channelz::v1::GetServersResponse* response) override; // implementation of GetChannel rpc Status GetChannel(ServerContext* unused, const channelz::v1::GetChannelRequest* request, channelz::v1::GetChannelResponse* response) override; + // implementation of GetSubchannel rpc + Status GetSubchannel(ServerContext* unused, + const channelz::v1::GetSubchannelRequest* request, + channelz::v1::GetSubchannelResponse* response) override; }; } // namespace grpc diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc index b8ba7042d9..36f7eb81f9 100644 --- a/src/cpp/server/server_cc.cc +++ b/src/cpp/server/server_cc.cc @@ -657,6 +657,7 @@ void Server::PerformOpsOnCall(internal::CallOpSetInterface* ops, size_t nops = 0; grpc_op cops[MAX_OPS]; ops->FillOps(call->call(), cops, &nops); + // TODO(vjpai): Use ops->cq_tag once this case supports callbacks auto result = grpc_call_start_batch(call->call(), cops, nops, ops, nullptr); if (result != GRPC_CALL_OK) { gpr_log(GPR_ERROR, "Fatal: grpc_call_start_batch returned %d", result); diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index bf0c027cda..b7254b6bb9 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -61,6 +61,9 @@ class ServerContext::CompletionOp final : public internal::CallOpSetInterface { tag_ = tag; } + /// TODO(vjpai): Allow override of cq_tag if appropriate for callback API + void* cq_tag() override { return this; } + void Unref(); private: diff --git a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs index 9aab54d2d0..775849d89b 100644 --- a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs +++ b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallTest.cs @@ -107,6 +107,42 @@ namespace Grpc.Core.Internal.Tests } [Test] + public void AsyncUnary_RequestSerializationExceptionDoesntLeakResources() + { + string nullRequest = null; // will throw when serializing + Assert.Throws(typeof(ArgumentNullException), () => asyncCall.UnaryCallAsync(nullRequest)); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + } + + [Test] + public void AsyncUnary_StartCallFailureDoesntLeakResources() + { + fakeCall.MakeStartCallFail(); + Assert.Throws(typeof(InvalidOperationException), () => asyncCall.UnaryCallAsync("request1")); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + } + + [Test] + public void SyncUnary_RequestSerializationExceptionDoesntLeakResources() + { + string nullRequest = null; // will throw when serializing + Assert.Throws(typeof(ArgumentNullException), () => asyncCall.UnaryCall(nullRequest)); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + } + + [Test] + public void SyncUnary_StartCallFailureDoesntLeakResources() + { + fakeCall.MakeStartCallFail(); + Assert.Throws(typeof(InvalidOperationException), () => asyncCall.UnaryCall("request1")); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + } + + [Test] public void ClientStreaming_StreamingReadNotAllowed() { asyncCall.ClientStreamingCallAsync(); @@ -328,6 +364,15 @@ namespace Grpc.Core.Internal.Tests } [Test] + public void ClientStreaming_StartCallFailureDoesntLeakResources() + { + fakeCall.MakeStartCallFail(); + Assert.Throws(typeof(InvalidOperationException), () => asyncCall.ClientStreamingCallAsync()); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + } + + [Test] public void ServerStreaming_StreamingSendNotAllowed() { asyncCall.StartServerStreamingCall("request1"); @@ -402,6 +447,27 @@ namespace Grpc.Core.Internal.Tests } [Test] + public void ServerStreaming_RequestSerializationExceptionDoesntLeakResources() + { + string nullRequest = null; // will throw when serializing + Assert.Throws(typeof(ArgumentNullException), () => asyncCall.StartServerStreamingCall(nullRequest)); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + + var responseStream = new ClientResponseStream<string, string>(asyncCall); + var readTask = responseStream.MoveNext(); + } + + [Test] + public void ServerStreaming_StartCallFailureDoesntLeakResources() + { + fakeCall.MakeStartCallFail(); + Assert.Throws(typeof(InvalidOperationException), () => asyncCall.StartServerStreamingCall("request1")); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + } + + [Test] public void DuplexStreaming_NoRequestNoResponse_Success() { asyncCall.StartDuplexStreamingCall(); @@ -558,6 +624,15 @@ namespace Grpc.Core.Internal.Tests AssertStreamingResponseError(asyncCall, fakeCall, readTask2, StatusCode.Cancelled); } + [Test] + public void DuplexStreaming_StartCallFailureDoesntLeakResources() + { + fakeCall.MakeStartCallFail(); + Assert.Throws(typeof(InvalidOperationException), () => asyncCall.StartDuplexStreamingCall()); + Assert.AreEqual(0, channel.GetCallReferenceCount()); + Assert.IsTrue(fakeCall.IsDisposed); + } + ClientSideStatus CreateClientSideStatus(StatusCode statusCode) { return new ClientSideStatus(new Status(statusCode, ""), new Metadata()); diff --git a/src/csharp/Grpc.Core.Tests/Internal/FakeNativeCall.cs b/src/csharp/Grpc.Core.Tests/Internal/FakeNativeCall.cs index 581ac3384b..ef67918dab 100644 --- a/src/csharp/Grpc.Core.Tests/Internal/FakeNativeCall.cs +++ b/src/csharp/Grpc.Core.Tests/Internal/FakeNativeCall.cs @@ -31,6 +31,7 @@ namespace Grpc.Core.Internal.Tests /// </summary> internal class FakeNativeCall : INativeCall { + private bool shouldStartCallFail; public IUnaryResponseClientCallback UnaryResponseClientCallback { get; @@ -102,26 +103,31 @@ namespace Grpc.Core.Internal.Tests public void StartUnary(IUnaryResponseClientCallback callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags) { + StartCallMaybeFail(); UnaryResponseClientCallback = callback; } public void StartUnary(BatchContextSafeHandle ctx, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags) { + StartCallMaybeFail(); throw new NotImplementedException(); } public void StartClientStreaming(IUnaryResponseClientCallback callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags) { + StartCallMaybeFail(); UnaryResponseClientCallback = callback; } public void StartServerStreaming(IReceivedStatusOnClientCallback callback, byte[] payload, WriteFlags writeFlags, MetadataArraySafeHandle metadataArray, CallFlags callFlags) { + StartCallMaybeFail(); ReceivedStatusOnClientCallback = callback; } public void StartDuplexStreaming(IReceivedStatusOnClientCallback callback, MetadataArraySafeHandle metadataArray, CallFlags callFlags) { + StartCallMaybeFail(); ReceivedStatusOnClientCallback = callback; } @@ -165,5 +171,22 @@ namespace Grpc.Core.Internal.Tests { IsDisposed = true; } + + /// <summary> + /// Emulate CallSafeHandle.CheckOk() failure for all future attempts + /// to start a call. + /// </summary> + public void MakeStartCallFail() + { + shouldStartCallFail = true; + } + + private void StartCallMaybeFail() + { + if (shouldStartCallFail) + { + throw new InvalidOperationException("Start call has failed."); + } + } } } diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs index 4c89ed7393..7ce929dfa3 100644 --- a/src/csharp/Grpc.Core/Channel.cs +++ b/src/csharp/Grpc.Core/Channel.cs @@ -297,6 +297,12 @@ namespace Grpc.Core activeCallCounter.Decrement(); } + // for testing only + internal long GetCallReferenceCount() + { + return activeCallCounter.Count; + } + private ChannelState GetConnectivityState(bool tryToConnect) { try diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index 66902f3caa..4cdf0ee6a7 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -17,6 +17,7 @@ #endregion using System; +using System.Threading; using System.Threading.Tasks; using Grpc.Core.Logging; using Grpc.Core.Profiling; @@ -34,6 +35,8 @@ namespace Grpc.Core.Internal readonly CallInvocationDetails<TRequest, TResponse> details; readonly INativeCall injectedNativeCall; // for testing + bool registeredWithChannel; + // Dispose of to de-register cancellation token registration IDisposable cancellationTokenRegistration; @@ -77,43 +80,59 @@ namespace Grpc.Core.Internal using (profiler.NewScope("AsyncCall.UnaryCall")) using (CompletionQueueSafeHandle cq = CompletionQueueSafeHandle.CreateSync()) { - byte[] payload = UnsafeSerialize(msg); + bool callStartedOk = false; + try + { + unaryResponseTcs = new TaskCompletionSource<TResponse>(); - unaryResponseTcs = new TaskCompletionSource<TResponse>(); + lock (myLock) + { + GrpcPreconditions.CheckState(!started); + started = true; + Initialize(cq); - lock (myLock) - { - GrpcPreconditions.CheckState(!started); - started = true; - Initialize(cq); + halfcloseRequested = true; + readingDone = true; + } - halfcloseRequested = true; - readingDone = true; - } + byte[] payload = UnsafeSerialize(msg); - using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) - { - var ctx = details.Channel.Environment.BatchContextPool.Lease(); - try + using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) { - call.StartUnary(ctx, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); - var ev = cq.Pluck(ctx.Handle); - bool success = (ev.success != 0); + var ctx = details.Channel.Environment.BatchContextPool.Lease(); try { - using (profiler.NewScope("AsyncCall.UnaryCall.HandleBatch")) + call.StartUnary(ctx, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); + callStartedOk = true; + + var ev = cq.Pluck(ctx.Handle); + bool success = (ev.success != 0); + try + { + using (profiler.NewScope("AsyncCall.UnaryCall.HandleBatch")) + { + HandleUnaryResponse(success, ctx.GetReceivedStatusOnClient(), ctx.GetReceivedMessage(), ctx.GetReceivedInitialMetadata()); + } + } + catch (Exception e) { - HandleUnaryResponse(success, ctx.GetReceivedStatusOnClient(), ctx.GetReceivedMessage(), ctx.GetReceivedInitialMetadata()); + Logger.Error(e, "Exception occurred while invoking completion delegate."); } } - catch (Exception e) + finally { - Logger.Error(e, "Exception occurred while invoking completion delegate."); + ctx.Recycle(); } } - finally + } + finally + { + if (!callStartedOk) { - ctx.Recycle(); + lock (myLock) + { + OnFailedToStartCallLocked(); + } } } @@ -130,22 +149,35 @@ namespace Grpc.Core.Internal { lock (myLock) { - GrpcPreconditions.CheckState(!started); - started = true; + bool callStartedOk = false; + try + { + GrpcPreconditions.CheckState(!started); + started = true; - Initialize(details.Channel.CompletionQueue); + Initialize(details.Channel.CompletionQueue); - halfcloseRequested = true; - readingDone = true; + halfcloseRequested = true; + readingDone = true; + + byte[] payload = UnsafeSerialize(msg); - byte[] payload = UnsafeSerialize(msg); + unaryResponseTcs = new TaskCompletionSource<TResponse>(); + using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + { + call.StartUnary(UnaryResponseClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); + callStartedOk = true; + } - unaryResponseTcs = new TaskCompletionSource<TResponse>(); - using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + return unaryResponseTcs.Task; + } + finally { - call.StartUnary(UnaryResponseClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); + if (!callStartedOk) + { + OnFailedToStartCallLocked(); + } } - return unaryResponseTcs.Task; } } @@ -157,20 +189,32 @@ namespace Grpc.Core.Internal { lock (myLock) { - GrpcPreconditions.CheckState(!started); - started = true; + bool callStartedOk = false; + try + { + GrpcPreconditions.CheckState(!started); + started = true; - Initialize(details.Channel.CompletionQueue); + Initialize(details.Channel.CompletionQueue); - readingDone = true; + readingDone = true; + + unaryResponseTcs = new TaskCompletionSource<TResponse>(); + using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + { + call.StartClientStreaming(UnaryResponseClientCallback, metadataArray, details.Options.Flags); + callStartedOk = true; + } - unaryResponseTcs = new TaskCompletionSource<TResponse>(); - using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + return unaryResponseTcs.Task; + } + finally { - call.StartClientStreaming(UnaryResponseClientCallback, metadataArray, details.Options.Flags); + if (!callStartedOk) + { + OnFailedToStartCallLocked(); + } } - - return unaryResponseTcs.Task; } } @@ -181,21 +225,33 @@ namespace Grpc.Core.Internal { lock (myLock) { - GrpcPreconditions.CheckState(!started); - started = true; + bool callStartedOk = false; + try + { + GrpcPreconditions.CheckState(!started); + started = true; - Initialize(details.Channel.CompletionQueue); + Initialize(details.Channel.CompletionQueue); - halfcloseRequested = true; + halfcloseRequested = true; - byte[] payload = UnsafeSerialize(msg); + byte[] payload = UnsafeSerialize(msg); - streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); - using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); + using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + { + call.StartServerStreaming(ReceivedStatusOnClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); + callStartedOk = true; + } + call.StartReceiveInitialMetadata(ReceivedResponseHeadersCallback); + } + finally { - call.StartServerStreaming(ReceivedStatusOnClientCallback, payload, GetWriteFlagsForCall(), metadataArray, details.Options.Flags); + if (!callStartedOk) + { + OnFailedToStartCallLocked(); + } } - call.StartReceiveInitialMetadata(ReceivedResponseHeadersCallback); } } @@ -207,17 +263,29 @@ namespace Grpc.Core.Internal { lock (myLock) { - GrpcPreconditions.CheckState(!started); - started = true; + bool callStartedOk = false; + try + { + GrpcPreconditions.CheckState(!started); + started = true; - Initialize(details.Channel.CompletionQueue); + Initialize(details.Channel.CompletionQueue); - streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); - using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + streamingResponseCallFinishedTcs = new TaskCompletionSource<object>(); + using (var metadataArray = MetadataArraySafeHandle.Create(details.Options.Headers)) + { + call.StartDuplexStreaming(ReceivedStatusOnClientCallback, metadataArray, details.Options.Flags); + callStartedOk = true; + } + call.StartReceiveInitialMetadata(ReceivedResponseHeadersCallback); + } + finally { - call.StartDuplexStreaming(ReceivedStatusOnClientCallback, metadataArray, details.Options.Flags); + if (!callStartedOk) + { + OnFailedToStartCallLocked(); + } } - call.StartReceiveInitialMetadata(ReceivedResponseHeadersCallback); } } @@ -327,7 +395,11 @@ namespace Grpc.Core.Internal protected override void OnAfterReleaseResourcesLocked() { - details.Channel.RemoveCallReference(this); + if (registeredWithChannel) + { + details.Channel.RemoveCallReference(this); + registeredWithChannel = false; + } } protected override void OnAfterReleaseResourcesUnlocked() @@ -394,10 +466,27 @@ namespace Grpc.Core.Internal var call = CreateNativeCall(cq); details.Channel.AddCallReference(this); + registeredWithChannel = true; InitializeInternal(call); + RegisterCancellationCallback(); } + private void OnFailedToStartCallLocked() + { + ReleaseResources(); + + // We need to execute the hook that disposes the cancellation token + // registration, but it cannot be done from under a lock. + // To make things simple, we just schedule the unregistering + // on a threadpool. + // - Once the native call is disposed, the Cancel() calls are ignored anyway + // - We don't care about the overhead as OnFailedToStartCallLocked() only happens + // when something goes very bad when initializing a call and that should + // never happen when gRPC is used correctly. + ThreadPool.QueueUserWorkItem((state) => OnAfterReleaseResourcesUnlocked()); + } + private INativeCall CreateNativeCall(CompletionQueueSafeHandle cq) { if (injectedNativeCall != null) diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index 5a53049e4b..a93dc34620 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs @@ -189,7 +189,7 @@ namespace Grpc.Core.Internal /// </summary> protected abstract Exception GetRpcExceptionClientOnly(); - private void ReleaseResources() + protected void ReleaseResources() { if (call != null) { diff --git a/src/csharp/Grpc.Core/RpcException.cs b/src/csharp/Grpc.Core/RpcException.cs index 94429d74ce..ff89897565 100644 --- a/src/csharp/Grpc.Core/RpcException.cs +++ b/src/csharp/Grpc.Core/RpcException.cs @@ -33,10 +33,8 @@ namespace Grpc.Core /// Creates a new <c>RpcException</c> associated with given status. /// </summary> /// <param name="status">Resulting status of a call.</param> - public RpcException(Status status) : base(status.ToString()) + public RpcException(Status status) : this(status, Metadata.Empty, status.ToString()) { - this.status = status; - this.trailers = Metadata.Empty; } /// <summary> @@ -44,10 +42,8 @@ namespace Grpc.Core /// </summary> /// <param name="status">Resulting status of a call.</param> /// <param name="message">The exception message.</param> - public RpcException(Status status, string message) : base(message) + public RpcException(Status status, string message) : this(status, Metadata.Empty, message) { - this.status = status; - this.trailers = Metadata.Empty; } /// <summary> @@ -55,7 +51,17 @@ namespace Grpc.Core /// </summary> /// <param name="status">Resulting status of a call.</param> /// <param name="trailers">Response trailing metadata.</param> - public RpcException(Status status, Metadata trailers) : base(status.ToString()) + public RpcException(Status status, Metadata trailers) : this(status, trailers, status.ToString()) + { + } + + /// <summary> + /// Creates a new <c>RpcException</c> associated with given status, message and trailing response metadata. + /// </summary> + /// <param name="status">Resulting status of a call.</param> + /// <param name="trailers">Response trailing metadata.</param> + /// <param name="message">The exception message.</param> + public RpcException(Status status, Metadata trailers, string message) : base(message) { this.status = status; this.trailers = GrpcPreconditions.CheckNotNull(trailers); diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py index 6876601785..3494c9b15a 100644 --- a/src/python/grpcio/grpc/_channel.py +++ b/src/python/grpcio/grpc/_channel.py @@ -24,6 +24,7 @@ from grpc import _grpcio_metadata from grpc._cython import cygrpc from grpc.framework.foundation import callable_util +logging.basicConfig() _LOGGER = logging.getLogger(__name__) _USER_AGENT = 'grpc-python/{}'.format(_grpcio_metadata.__version__) diff --git a/src/python/grpcio/grpc/_common.py b/src/python/grpcio/grpc/_common.py index 8358cbec5b..3805c7e82a 100644 --- a/src/python/grpcio/grpc/_common.py +++ b/src/python/grpcio/grpc/_common.py @@ -20,6 +20,7 @@ import six import grpc from grpc._cython import cygrpc +logging.basicConfig() _LOGGER = logging.getLogger(__name__) CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi index 00a1b23a67..334e561baa 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc_string.pyx.pxi @@ -14,6 +14,7 @@ import logging +logging.basicConfig() _LOGGER = logging.getLogger(__name__) # This function will ascii encode unicode string inputs if neccesary. diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi index ce701724fd..5779437b92 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi @@ -18,6 +18,7 @@ import logging import time import grpc +logging.basicConfig() _LOGGER = logging.getLogger(__name__) cdef class Server: diff --git a/src/python/grpcio/grpc/_plugin_wrapping.py b/src/python/grpcio/grpc/_plugin_wrapping.py index 916ee080b6..88ab4d8371 100644 --- a/src/python/grpcio/grpc/_plugin_wrapping.py +++ b/src/python/grpcio/grpc/_plugin_wrapping.py @@ -20,6 +20,7 @@ import grpc from grpc import _common from grpc._cython import cygrpc +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio/grpc/_server.py b/src/python/grpcio/grpc/_server.py index 7276a7fd90..daa000a6e1 100644 --- a/src/python/grpcio/grpc/_server.py +++ b/src/python/grpcio/grpc/_server.py @@ -27,6 +27,7 @@ from grpc import _interceptor from grpc._cython import cygrpc from grpc.framework.foundation import callable_util +logging.basicConfig() _LOGGER = logging.getLogger(__name__) _SHUTDOWN_TAG = 'shutdown' diff --git a/src/python/grpcio/grpc/framework/foundation/callable_util.py b/src/python/grpcio/grpc/framework/foundation/callable_util.py index 24daf3406f..fb8d5f7c1e 100644 --- a/src/python/grpcio/grpc/framework/foundation/callable_util.py +++ b/src/python/grpcio/grpc/framework/foundation/callable_util.py @@ -21,6 +21,7 @@ import logging import six +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio/grpc/framework/foundation/logging_pool.py b/src/python/grpcio/grpc/framework/foundation/logging_pool.py index 216e3990db..7702d1785f 100644 --- a/src/python/grpcio/grpc/framework/foundation/logging_pool.py +++ b/src/python/grpcio/grpc/framework/foundation/logging_pool.py @@ -17,6 +17,7 @@ import logging from concurrent import futures +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio/grpc/framework/foundation/stream_util.py b/src/python/grpcio/grpc/framework/foundation/stream_util.py index 1faaf29bd7..9184f95873 100644 --- a/src/python/grpcio/grpc/framework/foundation/stream_util.py +++ b/src/python/grpcio/grpc/framework/foundation/stream_util.py @@ -19,6 +19,7 @@ import threading from grpc.framework.foundation import stream _NO_VALUE = object() +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py b/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py index 191b1c1726..d7205ca579 100644 --- a/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py +++ b/src/python/grpcio_testing/grpc_testing/_channel/_invocation.py @@ -18,6 +18,7 @@ import threading import grpc _NOT_YET_OBSERVED = object() +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio_testing/grpc_testing/_server/_rpc.py b/src/python/grpcio_testing/grpc_testing/_server/_rpc.py index b856da100f..736b714dc6 100644 --- a/src/python/grpcio_testing/grpc_testing/_server/_rpc.py +++ b/src/python/grpcio_testing/grpc_testing/_server/_rpc.py @@ -18,6 +18,7 @@ import threading import grpc from grpc_testing import _common +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio_testing/grpc_testing/_time.py b/src/python/grpcio_testing/grpc_testing/_time.py index 75e6db3458..9692c34e6f 100644 --- a/src/python/grpcio_testing/grpc_testing/_time.py +++ b/src/python/grpcio_testing/grpc_testing/_time.py @@ -21,6 +21,7 @@ import time as _time import grpc import grpc_testing +logging.basicConfig() _LOGGER = logging.getLogger(__name__) diff --git a/src/python/grpcio_tests/tests/interop/server.py b/src/python/grpcio_tests/tests/interop/server.py index fd28d498a1..768cdaf468 100644 --- a/src/python/grpcio_tests/tests/interop/server.py +++ b/src/python/grpcio_tests/tests/interop/server.py @@ -25,6 +25,7 @@ from tests.interop import methods from tests.interop import resources from tests.unit import test_common +logging.basicConfig() _ONE_DAY_IN_SECONDS = 60 * 60 * 24 _LOGGER = logging.getLogger(__name__) diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.c b/src/ruby/ext/grpc/rb_grpc_imports.generated.c index f44f89dfa0..0c46f6c85a 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.c +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.c @@ -97,7 +97,9 @@ grpc_resource_quota_resize_type grpc_resource_quota_resize_import; grpc_resource_quota_set_max_threads_type grpc_resource_quota_set_max_threads_import; grpc_resource_quota_arg_vtable_type grpc_resource_quota_arg_vtable_import; grpc_channelz_get_top_channels_type grpc_channelz_get_top_channels_import; +grpc_channelz_get_servers_type grpc_channelz_get_servers_import; grpc_channelz_get_channel_type grpc_channelz_get_channel_import; +grpc_channelz_get_subchannel_type grpc_channelz_get_subchannel_import; grpc_insecure_channel_create_from_fd_type grpc_insecure_channel_create_from_fd_import; grpc_server_add_insecure_channel_from_fd_type grpc_server_add_insecure_channel_from_fd_import; grpc_use_signal_type grpc_use_signal_import; @@ -351,7 +353,9 @@ void grpc_rb_load_imports(HMODULE library) { grpc_resource_quota_set_max_threads_import = (grpc_resource_quota_set_max_threads_type) GetProcAddress(library, "grpc_resource_quota_set_max_threads"); grpc_resource_quota_arg_vtable_import = (grpc_resource_quota_arg_vtable_type) GetProcAddress(library, "grpc_resource_quota_arg_vtable"); grpc_channelz_get_top_channels_import = (grpc_channelz_get_top_channels_type) GetProcAddress(library, "grpc_channelz_get_top_channels"); + grpc_channelz_get_servers_import = (grpc_channelz_get_servers_type) GetProcAddress(library, "grpc_channelz_get_servers"); grpc_channelz_get_channel_import = (grpc_channelz_get_channel_type) GetProcAddress(library, "grpc_channelz_get_channel"); + grpc_channelz_get_subchannel_import = (grpc_channelz_get_subchannel_type) GetProcAddress(library, "grpc_channelz_get_subchannel"); grpc_insecure_channel_create_from_fd_import = (grpc_insecure_channel_create_from_fd_type) GetProcAddress(library, "grpc_insecure_channel_create_from_fd"); grpc_server_add_insecure_channel_from_fd_import = (grpc_server_add_insecure_channel_from_fd_type) GetProcAddress(library, "grpc_server_add_insecure_channel_from_fd"); grpc_use_signal_import = (grpc_use_signal_type) GetProcAddress(library, "grpc_use_signal"); diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h index 99433f0e40..d00e75c326 100644 --- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h +++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h @@ -266,9 +266,15 @@ extern grpc_resource_quota_arg_vtable_type grpc_resource_quota_arg_vtable_import typedef char*(*grpc_channelz_get_top_channels_type)(intptr_t start_channel_id); extern grpc_channelz_get_top_channels_type grpc_channelz_get_top_channels_import; #define grpc_channelz_get_top_channels grpc_channelz_get_top_channels_import +typedef char*(*grpc_channelz_get_servers_type)(intptr_t start_server_id); +extern grpc_channelz_get_servers_type grpc_channelz_get_servers_import; +#define grpc_channelz_get_servers grpc_channelz_get_servers_import typedef char*(*grpc_channelz_get_channel_type)(intptr_t channel_id); extern grpc_channelz_get_channel_type grpc_channelz_get_channel_import; #define grpc_channelz_get_channel grpc_channelz_get_channel_import +typedef char*(*grpc_channelz_get_subchannel_type)(intptr_t subchannel_id); +extern grpc_channelz_get_subchannel_type grpc_channelz_get_subchannel_import; +#define grpc_channelz_get_subchannel grpc_channelz_get_subchannel_import typedef grpc_channel*(*grpc_insecure_channel_create_from_fd_type)(const char* target, int fd, const grpc_channel_args* args); extern grpc_insecure_channel_create_from_fd_type grpc_insecure_channel_create_from_fd_import; #define grpc_insecure_channel_create_from_fd grpc_insecure_channel_create_from_fd_import diff --git a/templates/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile.template b/templates/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile.template index 4fb7b4642d..34ea645cfa 100644 --- a/templates/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile.template +++ b/templates/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile.template @@ -14,7 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. - FROM google/dart:latest + FROM google/dart:2.0 # Upgrade Dart to version 2. RUN apt-get update && apt-get upgrade -y dart diff --git a/test/core/channel/channel_stack_test.cc b/test/core/channel/channel_stack_test.cc index 2f5329a96d..6f0bfa06d2 100644 --- a/test/core/channel/channel_stack_test.cc +++ b/test/core/channel/channel_stack_test.cc @@ -124,7 +124,7 @@ static void test_create_channel_stack(void) { gpr_now(GPR_CLOCK_MONOTONIC), /* start_time */ GRPC_MILLIS_INF_FUTURE, /* deadline */ nullptr, /* arena */ - nullptr /* call_combiner */ + nullptr, /* call_combiner */ }; grpc_error* error = grpc_call_stack_init(channel_stack, 1, free_call, call_stack, &args); diff --git a/test/core/channel/channel_trace_test.cc b/test/core/channel/channel_trace_test.cc index f224457a55..d594445bfb 100644 --- a/test/core/channel/channel_trace_test.cc +++ b/test/core/channel/channel_trace_test.cc @@ -40,6 +40,17 @@ namespace grpc_core { namespace channelz { namespace testing { + +// testing peer to access channel internals +class ChannelNodePeer { + public: + explicit ChannelNodePeer(ChannelNode* node) : node_(node) {} + ChannelTrace* trace() const { return &node_->trace_; } + + private: + ChannelNode* node_; +}; + namespace { grpc_json* GetJsonChild(grpc_json* parent, const char* key) { @@ -156,28 +167,29 @@ TEST_P(ChannelTracerTest, ComplexTest) { ChannelFixture channel1(GetParam()); RefCountedPtr<ChannelNode> sc1 = MakeRefCounted<ChannelNode>(channel1.channel(), GetParam(), true); - tracer.AddTraceEventReferencingSubchannel( + ChannelNodePeer sc1_peer(sc1.get()); + tracer.AddTraceEventWithReference( ChannelTrace::Severity::Info, grpc_slice_from_static_string("subchannel one created"), sc1); ValidateChannelTrace(&tracer, 3, GetParam()); - AddSimpleTrace(sc1->trace()); - AddSimpleTrace(sc1->trace()); - AddSimpleTrace(sc1->trace()); - ValidateChannelTrace(sc1->trace(), 3, GetParam()); - AddSimpleTrace(sc1->trace()); - AddSimpleTrace(sc1->trace()); - AddSimpleTrace(sc1->trace()); - ValidateChannelTrace(sc1->trace(), 6, GetParam()); + AddSimpleTrace(sc1_peer.trace()); + AddSimpleTrace(sc1_peer.trace()); + AddSimpleTrace(sc1_peer.trace()); + ValidateChannelTrace(sc1_peer.trace(), 3, GetParam()); + AddSimpleTrace(sc1_peer.trace()); + AddSimpleTrace(sc1_peer.trace()); + AddSimpleTrace(sc1_peer.trace()); + ValidateChannelTrace(sc1_peer.trace(), 6, GetParam()); AddSimpleTrace(&tracer); AddSimpleTrace(&tracer); ValidateChannelTrace(&tracer, 5, GetParam()); ChannelFixture channel2(GetParam()); RefCountedPtr<ChannelNode> sc2 = MakeRefCounted<ChannelNode>(channel2.channel(), GetParam(), true); - tracer.AddTraceEventReferencingChannel( + tracer.AddTraceEventWithReference( ChannelTrace::Severity::Info, grpc_slice_from_static_string("LB channel two created"), sc2); - tracer.AddTraceEventReferencingSubchannel( + tracer.AddTraceEventWithReference( ChannelTrace::Severity::Warning, grpc_slice_from_static_string("subchannel one inactive"), sc1); ValidateChannelTrace(&tracer, 7, GetParam()); @@ -203,33 +215,35 @@ TEST_P(ChannelTracerTest, TestNesting) { ChannelFixture channel1(GetParam()); RefCountedPtr<ChannelNode> sc1 = MakeRefCounted<ChannelNode>(channel1.channel(), GetParam(), true); - tracer.AddTraceEventReferencingChannel( + ChannelNodePeer sc1_peer(sc1.get()); + tracer.AddTraceEventWithReference( ChannelTrace::Severity::Info, grpc_slice_from_static_string("subchannel one created"), sc1); ValidateChannelTrace(&tracer, 3, GetParam()); - AddSimpleTrace(sc1->trace()); + AddSimpleTrace(sc1_peer.trace()); ChannelFixture channel2(GetParam()); RefCountedPtr<ChannelNode> conn1 = MakeRefCounted<ChannelNode>(channel2.channel(), GetParam(), true); + ChannelNodePeer conn1_peer(conn1.get()); // nesting one level deeper. - sc1->trace()->AddTraceEventReferencingSubchannel( + sc1_peer.trace()->AddTraceEventWithReference( ChannelTrace::Severity::Info, grpc_slice_from_static_string("connection one created"), conn1); ValidateChannelTrace(&tracer, 3, GetParam()); - AddSimpleTrace(conn1->trace()); + AddSimpleTrace(conn1_peer.trace()); AddSimpleTrace(&tracer); AddSimpleTrace(&tracer); ValidateChannelTrace(&tracer, 5, GetParam()); - ValidateChannelTrace(conn1->trace(), 1, GetParam()); + ValidateChannelTrace(conn1_peer.trace(), 1, GetParam()); ChannelFixture channel3(GetParam()); RefCountedPtr<ChannelNode> sc2 = MakeRefCounted<ChannelNode>(channel3.channel(), GetParam(), true); - tracer.AddTraceEventReferencingSubchannel( + tracer.AddTraceEventWithReference( ChannelTrace::Severity::Info, grpc_slice_from_static_string("subchannel two created"), sc2); // this trace should not get added to the parents children since it is already // present in the tracer. - tracer.AddTraceEventReferencingChannel( + tracer.AddTraceEventWithReference( ChannelTrace::Severity::Warning, grpc_slice_from_static_string("subchannel one inactive"), sc1); AddSimpleTrace(&tracer); diff --git a/test/core/channel/channelz_registry_test.cc b/test/core/channel/channelz_registry_test.cc index 581e867584..c02d525c81 100644 --- a/test/core/channel/channelz_registry_test.cc +++ b/test/core/channel/channelz_registry_test.cc @@ -44,22 +44,22 @@ namespace channelz { namespace testing { TEST(ChannelzRegistryTest, UuidStartsAboveZeroTest) { - ChannelNode* channelz_channel = nullptr; - intptr_t uuid = ChannelzRegistry::RegisterChannelNode(channelz_channel); + BaseNode* channelz_channel = nullptr; + intptr_t uuid = ChannelzRegistry::Register(channelz_channel); EXPECT_GT(uuid, 0) << "First uuid chose must be greater than zero. Zero if " "reserved according to " "https://github.com/grpc/proposal/blob/master/" "A14-channelz.md"; - ChannelzRegistry::UnregisterChannelNode(uuid); + ChannelzRegistry::Unregister(uuid); } TEST(ChannelzRegistryTest, UuidsAreIncreasing) { - ChannelNode* channelz_channel = nullptr; + BaseNode* channelz_channel = nullptr; std::vector<intptr_t> uuids; uuids.reserve(10); for (int i = 0; i < 10; ++i) { // reregister the same object. It's ok since we are just testing uuids - uuids.push_back(ChannelzRegistry::RegisterChannelNode(channelz_channel)); + uuids.push_back(ChannelzRegistry::Register(channelz_channel)); } for (size_t i = 1; i < uuids.size(); ++i) { EXPECT_LT(uuids[i - 1], uuids[i]) << "Uuids must always be increasing"; @@ -68,30 +68,30 @@ TEST(ChannelzRegistryTest, UuidsAreIncreasing) { TEST(ChannelzRegistryTest, RegisterGetTest) { // we hackily jam an intptr_t into this pointer to check for equality later - ChannelNode* channelz_channel = (ChannelNode*)42; - intptr_t uuid = ChannelzRegistry::RegisterChannelNode(channelz_channel); - ChannelNode* retrieved = ChannelzRegistry::GetChannelNode(uuid); + BaseNode* channelz_channel = (BaseNode*)42; + intptr_t uuid = ChannelzRegistry::Register(channelz_channel); + BaseNode* retrieved = ChannelzRegistry::Get(uuid); EXPECT_EQ(channelz_channel, retrieved); } TEST(ChannelzRegistryTest, RegisterManyItems) { // we hackily jam an intptr_t into this pointer to check for equality later - ChannelNode* channelz_channel = (ChannelNode*)42; + BaseNode* channelz_channel = (BaseNode*)42; for (int i = 0; i < 100; i++) { - intptr_t uuid = ChannelzRegistry::RegisterChannelNode(channelz_channel); - ChannelNode* retrieved = ChannelzRegistry::GetChannelNode(uuid); + intptr_t uuid = ChannelzRegistry::Register(channelz_channel); + BaseNode* retrieved = ChannelzRegistry::Get(uuid); EXPECT_EQ(channelz_channel, retrieved); } } TEST(ChannelzRegistryTest, NullIfNotPresentTest) { // we hackily jam an intptr_t into this pointer to check for equality later - ChannelNode* channelz_channel = (ChannelNode*)42; - intptr_t uuid = ChannelzRegistry::RegisterChannelNode(channelz_channel); + BaseNode* channelz_channel = (BaseNode*)42; + intptr_t uuid = ChannelzRegistry::Register(channelz_channel); // try to pull out a uuid that does not exist. - ChannelNode* nonexistant = ChannelzRegistry::GetChannelNode(uuid + 1); + BaseNode* nonexistant = ChannelzRegistry::Get(uuid + 1); EXPECT_EQ(nonexistant, nullptr); - ChannelNode* retrieved = ChannelzRegistry::GetChannelNode(uuid); + BaseNode* retrieved = ChannelzRegistry::Get(uuid); EXPECT_EQ(channelz_channel, retrieved); } diff --git a/test/core/channel/channelz_test.cc b/test/core/channel/channelz_test.cc index ad5f86d934..bcda30d9f2 100644 --- a/test/core/channel/channelz_test.cc +++ b/test/core/channel/channelz_test.cc @@ -31,6 +31,7 @@ #include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/json/json.h" #include "src/core/lib/surface/channel.h" +#include "src/core/lib/surface/server.h" #include "test/core/util/test_config.h" #include "test/cpp/util/channel_trace_proto_helper.h" @@ -44,16 +45,16 @@ namespace channelz { namespace testing { // testing peer to access channel internals -class ChannelNodePeer { +class CallCountingHelperPeer { public: - ChannelNodePeer(ChannelNode* channel) : channel_(channel) {} - grpc_millis last_call_started_millis() { + explicit CallCountingHelperPeer(CallCountingHelper* node) : node_(node) {} + grpc_millis last_call_started_millis() const { return (grpc_millis)gpr_atm_no_barrier_load( - &channel_->last_call_started_millis_); + &node_->last_call_started_millis_); } private: - ChannelNode* channel_; + CallCountingHelper* node_; }; namespace { @@ -102,6 +103,25 @@ void ValidateGetTopChannels(size_t expected_channels) { gpr_free(core_api_json_str); } +void ValidateGetServers(size_t expected_servers) { + char* json_str = ChannelzRegistry::GetServers(0); + grpc::testing::ValidateGetServersResponseProtoJsonTranslation(json_str); + grpc_json* parsed_json = grpc_json_parse_string(json_str); + // This check will naturally have to change when we support pagination. + // tracked: https://github.com/grpc/grpc/issues/16019. + ValidateJsonArraySize(parsed_json, "server", expected_servers); + grpc_json* end = GetJsonChild(parsed_json, "end"); + ASSERT_NE(end, nullptr); + EXPECT_EQ(end->type, GRPC_JSON_TRUE); + grpc_json_destroy(parsed_json); + gpr_free(json_str); + // also check that the core API formats this correctly + char* core_api_json_str = grpc_channelz_get_servers(0); + grpc::testing::ValidateGetServersResponseProtoJsonTranslation( + core_api_json_str); + gpr_free(core_api_json_str); +} + class ChannelFixture { public: ChannelFixture(int max_trace_nodes = 0) { @@ -124,6 +144,28 @@ class ChannelFixture { grpc_channel* channel_; }; +class ServerFixture { + public: + explicit ServerFixture(int max_trace_nodes = 0) { + grpc_arg server_a[] = { + grpc_channel_arg_integer_create( + const_cast<char*>(GRPC_ARG_MAX_CHANNEL_TRACE_EVENTS_PER_NODE), + max_trace_nodes), + grpc_channel_arg_integer_create( + const_cast<char*>(GRPC_ARG_ENABLE_CHANNELZ), true), + }; + grpc_channel_args server_args = {GPR_ARRAY_SIZE(server_a), server_a}; + server_ = grpc_server_create(&server_args, nullptr); + } + + ~ServerFixture() { grpc_server_destroy(server_); } + + grpc_server* server() const { return server_; } + + private: + grpc_server* server_; +}; + struct validate_channel_data_args { int64_t calls_started; int64_t calls_failed; @@ -157,14 +199,21 @@ void ValidateChannel(ChannelNode* channel, validate_channel_data_args args) { ValidateCounters(json_str, args); gpr_free(json_str); // also check that the core API formats this the correct way - char* core_api_json_str = grpc_channelz_get_channel(channel->channel_uuid()); + char* core_api_json_str = grpc_channelz_get_channel(channel->uuid()); grpc::testing::ValidateGetChannelResponseProtoJsonTranslation( core_api_json_str); gpr_free(core_api_json_str); } -grpc_millis GetLastCallStartedMillis(ChannelNode* channel) { - ChannelNodePeer peer(channel); +void ValidateServer(ServerNode* server, validate_channel_data_args args) { + char* json_str = server->RenderJsonString(); + grpc::testing::ValidateServerProtoJsonTranslation(json_str); + ValidateCounters(json_str, args); + gpr_free(json_str); +} + +grpc_millis GetLastCallStartedMillis(CallCountingHelper* channel) { + CallCountingHelperPeer peer(channel); return peer.last_call_started_millis(); } @@ -215,31 +264,29 @@ TEST_P(ChannelzChannelTest, BasicChannelAPIFunctionality) { TEST_P(ChannelzChannelTest, LastCallStartedMillis) { grpc_core::ExecCtx exec_ctx; - ChannelFixture channel(GetParam()); - ChannelNode* channelz_channel = - grpc_channel_get_channelz_node(channel.channel()); + CallCountingHelper counter; // start a call to set the last call started timestamp - channelz_channel->RecordCallStarted(); - grpc_millis millis1 = GetLastCallStartedMillis(channelz_channel); + counter.RecordCallStarted(); + grpc_millis millis1 = GetLastCallStartedMillis(&counter); // time gone by should not affect the timestamp ChannelzSleep(100); - grpc_millis millis2 = GetLastCallStartedMillis(channelz_channel); + grpc_millis millis2 = GetLastCallStartedMillis(&counter); EXPECT_EQ(millis1, millis2); // calls succeeded or failed should not affect the timestamp ChannelzSleep(100); - channelz_channel->RecordCallFailed(); - channelz_channel->RecordCallSucceeded(); - grpc_millis millis3 = GetLastCallStartedMillis(channelz_channel); + counter.RecordCallFailed(); + counter.RecordCallSucceeded(); + grpc_millis millis3 = GetLastCallStartedMillis(&counter); EXPECT_EQ(millis1, millis3); // another call started should affect the timestamp // sleep for extra long to avoid flakes (since we cache Now()) ChannelzSleep(5000); - channelz_channel->RecordCallStarted(); - grpc_millis millis4 = GetLastCallStartedMillis(channelz_channel); + counter.RecordCallStarted(); + grpc_millis millis4 = GetLastCallStartedMillis(&counter); EXPECT_NE(millis1, millis4); } -TEST(ChannelzGetTopChannelsTest, BasicTest) { +TEST(ChannelzGetTopChannelsTest, BasicGetTopChannelsTest) { grpc_core::ExecCtx exec_ctx; ChannelFixture channel; ValidateGetTopChannels(1); @@ -275,9 +322,49 @@ TEST(ChannelzGetTopChannelsTest, InternalChannelTest) { grpc_channel_destroy(internal_channel); } +class ChannelzServerTest : public ::testing::TestWithParam<size_t> {}; + +TEST_P(ChannelzServerTest, BasicServerAPIFunctionality) { + grpc_core::ExecCtx exec_ctx; + ServerFixture server(10); + ServerNode* channelz_server = grpc_server_get_channelz_node(server.server()); + channelz_server->RecordCallStarted(); + channelz_server->RecordCallFailed(); + channelz_server->RecordCallSucceeded(); + ValidateServer(channelz_server, {1, 1, 1}); + channelz_server->RecordCallStarted(); + channelz_server->RecordCallFailed(); + channelz_server->RecordCallSucceeded(); + channelz_server->RecordCallStarted(); + channelz_server->RecordCallFailed(); + channelz_server->RecordCallSucceeded(); + ValidateServer(channelz_server, {3, 3, 3}); +} + +TEST(ChannelzGetServersTest, BasicGetServersTest) { + grpc_core::ExecCtx exec_ctx; + ServerFixture server; + ValidateGetServers(1); +} + +TEST(ChannelzGetServersTest, NoServersTest) { + grpc_core::ExecCtx exec_ctx; + ValidateGetServers(0); +} + +TEST(ChannelzGetServersTest, ManyServersTest) { + grpc_core::ExecCtx exec_ctx; + ServerFixture servers[10]; + (void)servers; // suppress unused variable error + ValidateGetServers(10); +} + INSTANTIATE_TEST_CASE_P(ChannelzChannelTestSweep, ChannelzChannelTest, ::testing::Values(0, 1, 2, 6, 10, 15)); +INSTANTIATE_TEST_CASE_P(ChannelzServerTestSweep, ChannelzServerTest, + ::testing::Values(0, 1, 2, 6, 10, 15)); + } // namespace testing } // namespace channelz } // namespace grpc_core diff --git a/test/core/client_channel/parse_address_test.cc b/test/core/client_channel/parse_address_test.cc index ae157fbb8b..004549fa62 100644 --- a/test/core/client_channel/parse_address_test.cc +++ b/test/core/client_channel/parse_address_test.cc @@ -91,6 +91,15 @@ static void test_grpc_parse_ipv6(const char* uri_text, const char* host, grpc_uri_destroy(uri); } +/* Test parsing invalid ipv6 addresses (valid uri_text but invalid ipv6 addr) */ +static void test_grpc_parse_ipv6_invalid(const char* uri_text) { + grpc_core::ExecCtx exec_ctx; + grpc_uri* uri = grpc_uri_parse(uri_text, 0); + grpc_resolved_address addr; + GPR_ASSERT(!grpc_parse_ipv6(uri, &addr)); + grpc_uri_destroy(uri); +} + int main(int argc, char** argv) { grpc_test_init(argc, argv); grpc_init(); @@ -100,5 +109,10 @@ int main(int argc, char** argv) { test_grpc_parse_ipv6("ipv6:[2001:db8::1]:12345", "2001:db8::1", 12345, 0); test_grpc_parse_ipv6("ipv6:[2001:db8::1%252]:12345", "2001:db8::1", 12345, 2); + /* Address length greater than GRPC_INET6_ADDRSTRLEN */ + test_grpc_parse_ipv6_invalid( + "ipv6:WWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWW45%" + "v6:45%x$1*"); + grpc_shutdown(); } diff --git a/test/core/end2end/tests/channelz.cc b/test/core/end2end/tests/channelz.cc index 533703a2be..40a0370f0e 100644 --- a/test/core/end2end/tests/channelz.cc +++ b/test/core/end2end/tests/channelz.cc @@ -22,6 +22,7 @@ #include <string.h> #include "src/core/lib/surface/channel.h" +#include "src/core/lib/surface/server.h" #include <grpc/byte_buffer.h> #include <grpc/grpc.h> @@ -198,17 +199,21 @@ static void run_one_request(grpc_end2end_test_config config, static void test_channelz(grpc_end2end_test_config config) { grpc_end2end_test_fixture f; - grpc_arg client_a; - client_a.type = GRPC_ARG_INTEGER; - client_a.key = const_cast<char*>(GRPC_ARG_ENABLE_CHANNELZ); - client_a.value.integer = true; - grpc_channel_args client_args = {1, &client_a}; + grpc_arg arg; + arg.type = GRPC_ARG_INTEGER; + arg.key = const_cast<char*>(GRPC_ARG_ENABLE_CHANNELZ); + arg.value.integer = true; + grpc_channel_args args = {1, &arg}; - f = begin_test(config, "test_channelz", &client_args, nullptr); + f = begin_test(config, "test_channelz", &args, &args); grpc_core::channelz::ChannelNode* channelz_channel = grpc_channel_get_channelz_node(f.client); - GPR_ASSERT(channelz_channel != nullptr); + + grpc_core::channelz::ServerNode* channelz_server = + grpc_server_get_channelz_node(f.server); + GPR_ASSERT(channelz_server != nullptr); + char* json = channelz_channel->RenderJsonString(); GPR_ASSERT(json != nullptr); // nothing is present yet @@ -235,7 +240,19 @@ static void test_channelz(grpc_end2end_test_config config) { GPR_ASSERT(nullptr != strstr(json, "\"callsStarted\":\"2\"")); GPR_ASSERT(nullptr != strstr(json, "\"callsFailed\":\"1\"")); GPR_ASSERT(nullptr != strstr(json, "\"callsSucceeded\":\"1\"")); - // channel tracing is not enables, so these should not be preset. + // channel tracing is not enabled, so these should not be preset. + GPR_ASSERT(nullptr == strstr(json, "\"trace\"")); + GPR_ASSERT(nullptr == strstr(json, "\"description\":\"Channel created\"")); + GPR_ASSERT(nullptr == strstr(json, "\"severity\":\"CT_INFO\"")); + gpr_free(json); + + json = channelz_server->RenderJsonString(); + GPR_ASSERT(json != nullptr); + gpr_log(GPR_INFO, "%s", json); + GPR_ASSERT(nullptr != strstr(json, "\"callsStarted\":\"2\"")); + GPR_ASSERT(nullptr != strstr(json, "\"callsFailed\":\"1\"")); + GPR_ASSERT(nullptr != strstr(json, "\"callsSucceeded\":\"1\"")); + // channel tracing is not enabled, so these should not be preset. GPR_ASSERT(nullptr == strstr(json, "\"trace\"")); GPR_ASSERT(nullptr == strstr(json, "\"description\":\"Channel created\"")); GPR_ASSERT(nullptr == strstr(json, "\"severity\":\"CT_INFO\"")); @@ -248,22 +265,24 @@ static void test_channelz(grpc_end2end_test_config config) { static void test_channelz_with_channel_trace(grpc_end2end_test_config config) { grpc_end2end_test_fixture f; - grpc_arg client_a[2]; - client_a[0].type = GRPC_ARG_INTEGER; - client_a[0].key = - const_cast<char*>(GRPC_ARG_MAX_CHANNEL_TRACE_EVENTS_PER_NODE); - client_a[0].value.integer = 5; - client_a[1].type = GRPC_ARG_INTEGER; - client_a[1].key = const_cast<char*>(GRPC_ARG_ENABLE_CHANNELZ); - client_a[1].value.integer = true; - grpc_channel_args client_args = {GPR_ARRAY_SIZE(client_a), client_a}; - - f = begin_test(config, "test_channelz_with_channel_trace", &client_args, - nullptr); + grpc_arg arg[2]; + arg[0].type = GRPC_ARG_INTEGER; + arg[0].key = const_cast<char*>(GRPC_ARG_MAX_CHANNEL_TRACE_EVENTS_PER_NODE); + arg[0].value.integer = 5; + arg[1].type = GRPC_ARG_INTEGER; + arg[1].key = const_cast<char*>(GRPC_ARG_ENABLE_CHANNELZ); + arg[1].value.integer = true; + grpc_channel_args args = {GPR_ARRAY_SIZE(arg), arg}; + + f = begin_test(config, "test_channelz_with_channel_trace", &args, &args); grpc_core::channelz::ChannelNode* channelz_channel = grpc_channel_get_channelz_node(f.client); - GPR_ASSERT(channelz_channel != nullptr); + + grpc_core::channelz::ServerNode* channelz_server = + grpc_server_get_channelz_node(f.server); + GPR_ASSERT(channelz_server != nullptr); + char* json = channelz_channel->RenderJsonString(); GPR_ASSERT(json != nullptr); gpr_log(GPR_INFO, "%s", json); @@ -272,6 +291,14 @@ static void test_channelz_with_channel_trace(grpc_end2end_test_config config) { GPR_ASSERT(nullptr != strstr(json, "\"severity\":\"CT_INFO\"")); gpr_free(json); + json = channelz_server->RenderJsonString(); + GPR_ASSERT(json != nullptr); + gpr_log(GPR_INFO, "%s", json); + GPR_ASSERT(nullptr != strstr(json, "\"trace\"")); + GPR_ASSERT(nullptr != strstr(json, "\"description\":\"Server created\"")); + GPR_ASSERT(nullptr != strstr(json, "\"severity\":\"CT_INFO\"")); + gpr_free(json); + end_test(&f); config.tear_down_data(&f); } diff --git a/test/core/iomgr/BUILD b/test/core/iomgr/BUILD index 675d9e6278..7754bc4970 100644 --- a/test/core/iomgr/BUILD +++ b/test/core/iomgr/BUILD @@ -174,9 +174,27 @@ grpc_cc_test( ) grpc_cc_test( - name = "resolve_address_test", + name = "resolve_address_using_ares_resolver_test", srcs = ["resolve_address_test.cc"], language = "C++", + args = [ + "--resolver=ares", + ], + deps = [ + "//:gpr", + "//:grpc", + "//test/core/util:gpr_test_util", + "//test/core/util:grpc_test_util", + ], +) + +grpc_cc_test( + name = "resolve_address_using_native_resolver_test", + srcs = ["resolve_address_test.cc"], + language = "C++", + args = [ + "--resolver=native", + ], deps = [ "//:gpr", "//:grpc", diff --git a/test/core/iomgr/resolve_address_test.cc b/test/core/iomgr/resolve_address_test.cc index 2fb831a6a4..52e4840c7c 100644 --- a/test/core/iomgr/resolve_address_test.cc +++ b/test/core/iomgr/resolve_address_test.cc @@ -22,8 +22,14 @@ #include <grpc/support/log.h> #include <grpc/support/sync.h> #include <grpc/support/time.h> + +#include <string.h> + +#include "src/core/lib/gpr/env.h" +#include "src/core/lib/gpr/string.h" #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/iomgr.h" +#include "test/core/util/cmdline.h" #include "test/core/util/test_config.h" static gpr_timespec test_deadline(void) { @@ -240,6 +246,28 @@ static void test_unparseable_hostports(void) { } int main(int argc, char** argv) { + // First set the resolver type based off of --resolver + const char* resolver_type = nullptr; + gpr_cmdline* cl = gpr_cmdline_create("resolve address test"); + gpr_cmdline_add_string(cl, "resolver", "Resolver type (ares or native)", + &resolver_type); + gpr_cmdline_parse(cl, argc, argv); + const char* cur_resolver = gpr_getenv("GRPC_DNS_RESOLVER"); + if (cur_resolver != nullptr && strlen(cur_resolver) != 0) { + gpr_log(GPR_INFO, "Warning: overriding resolver setting of %s", + cur_resolver); + } + if (gpr_stricmp(resolver_type, "native") == 0) { + gpr_setenv("GRPC_DNS_RESOLVER", "native"); + } else if (gpr_stricmp(resolver_type, "ares") == 0) { +#ifndef GRPC_UV + gpr_setenv("GRPC_DNS_RESOLVER", "ares"); +#endif + } else { + gpr_log(GPR_ERROR, "--resolver_type was not set to ares or native"); + abort(); + } + // Run the test. grpc_test_init(argc, argv); grpc_init(); { @@ -250,10 +278,18 @@ int main(int argc, char** argv) { test_missing_default_port(); test_ipv6_with_port(); test_ipv6_without_port(); - test_invalid_ip_addresses(); - test_unparseable_hostports(); + if (gpr_stricmp(resolver_type, "ares") != 0) { + // These tests can trigger DNS queries to the nearby nameserver + // that need to come back in order for the test to succeed. + // c-ares is prone to not using the local system caches that the + // native getaddrinfo implementations take advantage of, so running + // these unit tests under c-ares risks flakiness. + test_invalid_ip_addresses(); + test_unparseable_hostports(); + } grpc_executor_shutdown(); } + gpr_cmdline_destroy(cl); grpc_shutdown(); return 0; diff --git a/test/core/surface/public_headers_must_be_c89.c b/test/core/surface/public_headers_must_be_c89.c index b832a1661b..b0af788796 100644 --- a/test/core/surface/public_headers_must_be_c89.c +++ b/test/core/surface/public_headers_must_be_c89.c @@ -136,7 +136,9 @@ int main(int argc, char **argv) { printf("%lx", (unsigned long) grpc_resource_quota_set_max_threads); printf("%lx", (unsigned long) grpc_resource_quota_arg_vtable); printf("%lx", (unsigned long) grpc_channelz_get_top_channels); + printf("%lx", (unsigned long) grpc_channelz_get_servers); printf("%lx", (unsigned long) grpc_channelz_get_channel); + printf("%lx", (unsigned long) grpc_channelz_get_subchannel); printf("%lx", (unsigned long) grpc_auth_property_iterator_next); printf("%lx", (unsigned long) grpc_auth_context_property_iterator); printf("%lx", (unsigned long) grpc_auth_context_peer_identity); diff --git a/test/core/util/port_isolated_runtime_environment.cc b/test/core/util/port_isolated_runtime_environment.cc index c1ca185a67..1f678c08e5 100644 --- a/test/core/util/port_isolated_runtime_environment.cc +++ b/test/core/util/port_isolated_runtime_environment.cc @@ -30,8 +30,8 @@ #include "test/core/util/port.h" -#define MIN_PORT 49152 -#define MAX_PORT 65535 +#define MIN_PORT 1025 +#define MAX_PORT 32766 static int get_random_port_offset() { srand(gpr_now(GPR_CLOCK_REALTIME).tv_nsec); diff --git a/test/cpp/end2end/BUILD b/test/cpp/end2end/BUILD index 75dec56a60..0415efc1ef 100644 --- a/test/cpp/end2end/BUILD +++ b/test/cpp/end2end/BUILD @@ -98,6 +98,25 @@ grpc_cc_binary( ], ) +grpc_cc_test( + name = "client_callback_end2end_test", + srcs = ["client_callback_end2end_test.cc"], + external_deps = [ + "gtest", + ], + deps = [ + ":test_service_impl", + "//:gpr", + "//:grpc", + "//:grpc++", + "//src/proto/grpc/testing:echo_messages_proto", + "//src/proto/grpc/testing:echo_proto", + "//test/core/util:gpr_test_util", + "//test/core/util:grpc_test_util", + "//test/cpp/util:test_util", + ], +) + grpc_cc_library( name = "end2end_test_lib", testonly = True, diff --git a/test/cpp/end2end/channelz_service_test.cc b/test/cpp/end2end/channelz_service_test.cc index 933e4a1ff6..e96d68f93b 100644 --- a/test/cpp/end2end/channelz_service_test.cc +++ b/test/cpp/end2end/channelz_service_test.cc @@ -35,10 +35,16 @@ #include "test/core/util/test_config.h" #include "test/cpp/end2end/test_service_impl.h" +#include <google/protobuf/text_format.h> + #include <gtest/gtest.h> using grpc::channelz::v1::GetChannelRequest; using grpc::channelz::v1::GetChannelResponse; +using grpc::channelz::v1::GetServersRequest; +using grpc::channelz::v1::GetServersResponse; +using grpc::channelz::v1::GetSubchannelRequest; +using grpc::channelz::v1::GetSubchannelResponse; using grpc::channelz::v1::GetTopChannelsRequest; using grpc::channelz::v1::GetTopChannelsResponse; @@ -140,7 +146,7 @@ class ChannelzServerTest : public ::testing::Test { ClientContext context; Status s = echo_stub_->Echo(&context, request, &response); EXPECT_EQ(response.message(), request.message()); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); } void SendFailedEcho(int channel_idx) { @@ -156,6 +162,19 @@ class ChannelzServerTest : public ::testing::Test { EXPECT_FALSE(s.ok()); } + // Uses GetTopChannels to return the channel_id of a particular channel, + // so that the unit tests may test GetChannel call. + intptr_t GetChannelId(int channel_idx) { + GetTopChannelsRequest request; + GetTopChannelsResponse response; + request.set_start_channel_id(0); + ClientContext context; + Status s = channelz_stub_->GetTopChannels(&context, request, &response); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); + EXPECT_GT(response.channel_size(), channel_idx); + return response.channel(channel_idx).ref().channel_id(); + } + static string to_string(const int number) { std::stringstream strs; strs << number; @@ -190,7 +209,7 @@ TEST_F(ChannelzServerTest, BasicTest) { request.set_start_channel_id(0); ClientContext context; Status s = channelz_stub_->GetTopChannels(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel_size(), 1); } @@ -202,7 +221,7 @@ TEST_F(ChannelzServerTest, HighStartId) { request.set_start_channel_id(10000); ClientContext context; Status s = channelz_stub_->GetTopChannels(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel_size(), 0); } @@ -212,10 +231,10 @@ TEST_F(ChannelzServerTest, SuccessfulRequestTest) { SendSuccessfulEcho(0); GetChannelRequest request; GetChannelResponse response; - request.set_channel_id(1); + request.set_channel_id(GetChannelId(0)); ClientContext context; Status s = channelz_stub_->GetChannel(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel().data().calls_started(), 1); EXPECT_EQ(response.channel().data().calls_succeeded(), 1); EXPECT_EQ(response.channel().data().calls_failed(), 0); @@ -227,10 +246,10 @@ TEST_F(ChannelzServerTest, FailedRequestTest) { SendFailedEcho(0); GetChannelRequest request; GetChannelResponse response; - request.set_channel_id(1); + request.set_channel_id(GetChannelId(0)); ClientContext context; Status s = channelz_stub_->GetChannel(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel().data().calls_started(), 1); EXPECT_EQ(response.channel().data().calls_succeeded(), 0); EXPECT_EQ(response.channel().data().calls_failed(), 1); @@ -250,10 +269,10 @@ TEST_F(ChannelzServerTest, ManyRequestsTest) { } GetChannelRequest request; GetChannelResponse response; - request.set_channel_id(1); + request.set_channel_id(GetChannelId(0)); ClientContext context; Status s = channelz_stub_->GetChannel(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel().data().calls_started(), kNumSuccess + kNumFailed); EXPECT_EQ(response.channel().data().calls_succeeded(), kNumSuccess); @@ -269,7 +288,7 @@ TEST_F(ChannelzServerTest, ManyChannels) { request.set_start_channel_id(0); ClientContext context; Status s = channelz_stub_->GetTopChannels(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel_size(), kNumChannels); } @@ -292,10 +311,10 @@ TEST_F(ChannelzServerTest, ManyRequestsManyChannels) { { GetChannelRequest request; GetChannelResponse response; - request.set_channel_id(1); + request.set_channel_id(GetChannelId(0)); ClientContext context; Status s = channelz_stub_->GetChannel(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel().data().calls_started(), kNumSuccess); EXPECT_EQ(response.channel().data().calls_succeeded(), kNumSuccess); EXPECT_EQ(response.channel().data().calls_failed(), 0); @@ -305,10 +324,10 @@ TEST_F(ChannelzServerTest, ManyRequestsManyChannels) { { GetChannelRequest request; GetChannelResponse response; - request.set_channel_id(2); + request.set_channel_id(GetChannelId(1)); ClientContext context; Status s = channelz_stub_->GetChannel(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel().data().calls_started(), kNumFailed); EXPECT_EQ(response.channel().data().calls_succeeded(), 0); EXPECT_EQ(response.channel().data().calls_failed(), kNumFailed); @@ -318,10 +337,10 @@ TEST_F(ChannelzServerTest, ManyRequestsManyChannels) { { GetChannelRequest request; GetChannelResponse response; - request.set_channel_id(3); + request.set_channel_id(GetChannelId(2)); ClientContext context; Status s = channelz_stub_->GetChannel(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel().data().calls_started(), kNumSuccess + kNumFailed); EXPECT_EQ(response.channel().data().calls_succeeded(), kNumSuccess); @@ -332,16 +351,103 @@ TEST_F(ChannelzServerTest, ManyRequestsManyChannels) { { GetChannelRequest request; GetChannelResponse response; - request.set_channel_id(4); + request.set_channel_id(GetChannelId(3)); ClientContext context; Status s = channelz_stub_->GetChannel(&context, request, &response); - EXPECT_TRUE(s.ok()); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); EXPECT_EQ(response.channel().data().calls_started(), 0); EXPECT_EQ(response.channel().data().calls_succeeded(), 0); EXPECT_EQ(response.channel().data().calls_failed(), 0); } } +TEST_F(ChannelzServerTest, ManySubchannels) { + ResetStubs(); + const int kNumChannels = 4; + ConfigureProxy(kNumChannels); + const int kNumSuccess = 10; + const int kNumFailed = 11; + for (int i = 0; i < kNumSuccess; ++i) { + SendSuccessfulEcho(0); + SendSuccessfulEcho(2); + } + for (int i = 0; i < kNumFailed; ++i) { + SendFailedEcho(1); + SendFailedEcho(2); + } + GetTopChannelsRequest gtc_request; + GetTopChannelsResponse gtc_response; + gtc_request.set_start_channel_id(0); + ClientContext context; + Status s = + channelz_stub_->GetTopChannels(&context, gtc_request, >c_response); + EXPECT_TRUE(s.ok()) << s.error_message(); + EXPECT_EQ(gtc_response.channel_size(), kNumChannels); + for (int i = 0; i < gtc_response.channel_size(); ++i) { + // if the channel sent no RPCs, then expect no subchannels to have been + // created. + if (gtc_response.channel(i).data().calls_started() == 0) { + EXPECT_EQ(gtc_response.channel(i).subchannel_ref_size(), 0); + continue; + } + // The resolver must return at least one address. + ASSERT_GT(gtc_response.channel(i).subchannel_ref_size(), 0); + GetSubchannelRequest gsc_request; + GetSubchannelResponse gsc_response; + gsc_request.set_subchannel_id( + gtc_response.channel(i).subchannel_ref(0).subchannel_id()); + ClientContext context; + Status s = + channelz_stub_->GetSubchannel(&context, gsc_request, &gsc_response); + EXPECT_TRUE(s.ok()) << s.error_message(); + EXPECT_EQ(gtc_response.channel(i).data().calls_started(), + gsc_response.subchannel().data().calls_started()); + EXPECT_EQ(gtc_response.channel(i).data().calls_succeeded(), + gsc_response.subchannel().data().calls_succeeded()); + EXPECT_EQ(gtc_response.channel(i).data().calls_failed(), + gsc_response.subchannel().data().calls_failed()); + } +} + +TEST_F(ChannelzServerTest, BasicServerTest) { + ResetStubs(); + ConfigureProxy(1); + GetServersRequest request; + GetServersResponse response; + request.set_start_server_id(0); + ClientContext context; + Status s = channelz_stub_->GetServers(&context, request, &response); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); + EXPECT_EQ(response.server_size(), 1); +} + +TEST_F(ChannelzServerTest, ServerCallTest) { + ResetStubs(); + ConfigureProxy(1); + const int kNumSuccess = 10; + const int kNumFailed = 11; + for (int i = 0; i < kNumSuccess; ++i) { + SendSuccessfulEcho(0); + } + for (int i = 0; i < kNumFailed; ++i) { + SendFailedEcho(0); + } + GetServersRequest request; + GetServersResponse response; + request.set_start_server_id(0); + ClientContext context; + Status s = channelz_stub_->GetServers(&context, request, &response); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); + EXPECT_EQ(response.server_size(), 1); + EXPECT_EQ(response.server(0).data().calls_succeeded(), kNumSuccess); + EXPECT_EQ(response.server(0).data().calls_failed(), kNumFailed); + // This is success+failure+1 because the call that retrieved this information + // will be counted as started. It will not track success/failure until after + // it has returned, so that is not included in the response. + EXPECT_EQ(response.server(0).data().calls_started(), + kNumSuccess + kNumFailed + 1); +} + } // namespace testing } // namespace grpc diff --git a/test/cpp/end2end/client_callback_end2end_test.cc b/test/cpp/end2end/client_callback_end2end_test.cc new file mode 100644 index 0000000000..75b896b33d --- /dev/null +++ b/test/cpp/end2end/client_callback_end2end_test.cc @@ -0,0 +1,126 @@ +/* + * + * 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 <functional> +#include <mutex> + +#include <grpcpp/channel.h> +#include <grpcpp/client_context.h> +#include <grpcpp/create_channel.h> +#include <grpcpp/generic/generic_stub.h> +#include <grpcpp/impl/codegen/proto_utils.h> +#include <grpcpp/server.h> +#include <grpcpp/server_builder.h> +#include <grpcpp/server_context.h> +#include <grpcpp/support/client_callback.h> + +#include "src/proto/grpc/testing/echo.grpc.pb.h" +#include "test/core/util/test_config.h" +#include "test/cpp/end2end/test_service_impl.h" +#include "test/cpp/util/byte_buffer_proto_helper.h" + +#include <gtest/gtest.h> + +namespace grpc { +namespace testing { +namespace { + +class ClientCallbackEnd2endTest : public ::testing::Test { + protected: + ClientCallbackEnd2endTest() {} + + void SetUp() override { + ServerBuilder builder; + builder.RegisterService(&service_); + + server_ = builder.BuildAndStart(); + is_server_started_ = true; + } + + void ResetStub() { + ChannelArguments args; + channel_ = server_->InProcessChannel(args); + stub_.reset(new GenericStub(channel_)); + } + + void TearDown() override { + if (is_server_started_) { + server_->Shutdown(); + } + } + + void SendRpcs(int num_rpcs) { + const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo"); + grpc::string test_string(""); + for (int i = 0; i < num_rpcs; i++) { + EchoRequest request; + std::unique_ptr<ByteBuffer> send_buf; + ByteBuffer recv_buf; + ClientContext cli_ctx; + + test_string += "Hello world. "; + request.set_message(test_string); + send_buf = SerializeToByteBuffer(&request); + + std::mutex mu; + std::condition_variable cv; + bool done = false; + stub_->experimental().UnaryCall( + &cli_ctx, kMethodName, send_buf.get(), &recv_buf, + [&request, &recv_buf, &done, &mu, &cv](Status s) { + GPR_ASSERT(s.ok()); + + EchoResponse response; + EXPECT_TRUE(ParseFromByteBuffer(&recv_buf, &response)); + EXPECT_EQ(request.message(), response.message()); + std::lock_guard<std::mutex> l(mu); + done = true; + cv.notify_one(); + }); + std::unique_lock<std::mutex> l(mu); + while (!done) { + cv.wait(l); + } + } + } + bool is_server_started_; + std::shared_ptr<Channel> channel_; + std::unique_ptr<grpc::GenericStub> stub_; + TestServiceImpl service_; + std::unique_ptr<Server> server_; +}; + +TEST_F(ClientCallbackEnd2endTest, SimpleRpc) { + ResetStub(); + SendRpcs(1); +} + +TEST_F(ClientCallbackEnd2endTest, SequentialRpcs) { + ResetStub(); + SendRpcs(10); +} + +} // namespace +} // namespace testing +} // namespace grpc + +int main(int argc, char** argv) { + grpc_test_init(argc, argv); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc index 7bcf23c0eb..a4b1a85f85 100644 --- a/test/cpp/interop/client.cc +++ b/test/cpp/interop/client.cc @@ -57,6 +57,7 @@ DEFINE_string( "half_duplex : half-duplex streaming;\n" "jwt_token_creds: large_unary with JWT token auth;\n" "large_unary : single request and (large) response;\n" + "long_lived_channel: sends large_unary rpcs over a long-lived channel;\n" "oauth2_auth_token: raw oauth2 access token auth;\n" "per_rpc_creds: raw oauth2 access token on a single rpc;\n" "ping_pong : full-duplex streaming;\n" @@ -84,10 +85,12 @@ DEFINE_bool(do_not_abort_on_transient_failures, false, "whether abort() is called or not. It does not control whether the " "test is retried in case of transient failures (and currently the " "interop tests are not retried even if this flag is set to true)"); - DEFINE_int32(soak_iterations, 1000, "number of iterations to use for the two soak tests; rpc_soak and " "channel_soak"); +DEFINE_int32(iteration_interval, 10, + "The interval in seconds between rpcs. This is used by " + "long_connection test"); using grpc::testing::CreateChannelForTestCase; using grpc::testing::GetServiceAccountJsonKey; @@ -163,6 +166,9 @@ int main(int argc, char** argv) { FLAGS_soak_iterations); actions["rpc_soak"] = std::bind(&grpc::testing::InteropClient::DoRpcSoakTest, &client, FLAGS_soak_iterations); + actions["long_lived_channel"] = + std::bind(&grpc::testing::InteropClient::DoLongLivedChannelTest, &client, + FLAGS_soak_iterations, FLAGS_iteration_interval); UpdateActions(&actions); diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc index b7ce90803b..a99cf8122f 100644 --- a/test/cpp/interop/interop_client.cc +++ b/test/cpp/interop/interop_client.cc @@ -1052,6 +1052,34 @@ bool InteropClient::DoChannelSoakTest(int32_t soak_iterations) { return true; } +bool InteropClient::DoLongLivedChannelTest(int32_t soak_iterations, + int32_t iteration_interval) { + gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations); + GPR_ASSERT(soak_iterations > 0); + GPR_ASSERT(iteration_interval > 0); + SimpleRequest request; + SimpleResponse response; + int num_failures = 0; + for (int i = 0; i < soak_iterations; ++i) { + gpr_log(GPR_DEBUG, "Sending RPC number %d...", i); + if (!PerformLargeUnary(&request, &response)) { + gpr_log(GPR_ERROR, "Iteration %d failed.", i); + num_failures++; + } + gpr_sleep_until( + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_seconds(iteration_interval, GPR_TIMESPAN))); + } + if (num_failures == 0) { + gpr_log(GPR_DEBUG, "long_lived_channel test done."); + return true; + } else { + gpr_log(GPR_DEBUG, "long_lived_channel test failed with %d rpc failures.", + num_failures); + return false; + } +} + bool InteropClient::DoUnimplementedService() { gpr_log(GPR_DEBUG, "Sending a request for an unimplemented service..."); diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h index e5be44d1d4..0ceff55c5c 100644 --- a/test/cpp/interop/interop_client.h +++ b/test/cpp/interop/interop_client.h @@ -76,6 +76,8 @@ class InteropClient { // languages bool DoChannelSoakTest(int32_t soak_iterations); bool DoRpcSoakTest(int32_t soak_iterations); + bool DoLongLivedChannelTest(int32_t soak_iterations, + int32_t iteration_interval); // Auth tests. // username is a string containing the user email diff --git a/test/cpp/naming/address_sorting_test.cc b/test/cpp/naming/address_sorting_test.cc index 04c300876c..fc6721d0ba 100644 --- a/test/cpp/naming/address_sorting_test.cc +++ b/test/cpp/naming/address_sorting_test.cc @@ -216,7 +216,7 @@ TEST_F(AddressSortingTest, TestDepriotizesUnreachableAddresses) { {"1.2.3.4:443", AF_INET}, {"5.6.7.8:443", AF_INET}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "1.2.3.4:443", "5.6.7.8:443", @@ -235,7 +235,7 @@ TEST_F(AddressSortingTest, TestDepriotizesUnsupportedDomainIpv6) { {"[2607:f8b0:400a:801::1002]:443", AF_INET6}, {"1.2.3.4:443", AF_INET}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "1.2.3.4:443", "[2607:f8b0:400a:801::1002]:443", @@ -255,7 +255,7 @@ TEST_F(AddressSortingTest, TestDepriotizesUnsupportedDomainIpv4) { {"[2607:f8b0:400a:801::1002]:443", AF_INET6}, {"1.2.3.4:443", AF_INET}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[2607:f8b0:400a:801::1002]:443", "1.2.3.4:443", @@ -279,7 +279,7 @@ TEST_F(AddressSortingTest, TestDepriotizesNonMatchingScope) { {"[2000:f8b0:400a:801::1002]:443", AF_INET6}, {"[fec0::5000]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[fec0::5000]:443", "[2000:f8b0:400a:801::1002]:443", @@ -302,7 +302,7 @@ TEST_F(AddressSortingTest, TestUsesLabelFromDefaultTable) { {"[2002::5001]:443", AF_INET6}, {"[2001::5001]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[2001::5001]:443", "[2002::5001]:443", @@ -325,7 +325,7 @@ TEST_F(AddressSortingTest, TestUsesLabelFromDefaultTableInputFlipped) { {"[2001::5001]:443", AF_INET6}, {"[2002::5001]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[2001::5001]:443", "[2002::5001]:443", @@ -348,7 +348,7 @@ TEST_F(AddressSortingTest, {"[3ffe::5001]:443", AF_INET6}, {"1.2.3.4:443", AF_INET}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs( lb_addrs, { // The AF_INET address should be IPv4-mapped by the sort, @@ -381,7 +381,7 @@ TEST_F(AddressSortingTest, {v4_compat_dest, AF_INET6}, {"[::1]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[::1]:443", v4_compat_dest, @@ -404,7 +404,7 @@ TEST_F(AddressSortingTest, {"[1234::2]:443", AF_INET6}, {"[::1]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs( lb_addrs, { @@ -428,7 +428,7 @@ TEST_F(AddressSortingTest, {"[2001::1234]:443", AF_INET6}, {"[2000::5001]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs( lb_addrs, { // The 2000::/16 address should match the ::/0 prefix rule @@ -452,7 +452,7 @@ TEST_F( {"[2001::1231]:443", AF_INET6}, {"[2000::5001]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[2000::5001]:443", "[2001::1231]:443", @@ -473,7 +473,7 @@ TEST_F(AddressSortingTest, {"[fec0::1234]:443", AF_INET6}, {"[fc00::5001]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[fc00::5001]:443", "[fec0::1234]:443", @@ -498,7 +498,7 @@ TEST_F( {"[::ffff:1.1.1.2]:443", AF_INET6}, {"[1234::2]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { // ::ffff:0:2 should match the v4-mapped // precedence entry and be deprioritized. @@ -525,7 +525,7 @@ TEST_F(AddressSortingTest, TestPrefersSmallerScope) { {"[3ffe::5001]:443", AF_INET6}, {"[fec0::1234]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[fec0::1234]:443", "[3ffe::5001]:443", @@ -550,7 +550,7 @@ TEST_F(AddressSortingTest, TestPrefersLongestMatchingSrcDstPrefix) { {"[3ffe:5001::]:443", AF_INET6}, {"[3ffe:1234::]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe:1234::]:443", "[3ffe:5001::]:443", @@ -571,7 +571,7 @@ TEST_F(AddressSortingTest, {"[3ffe::5001]:443", AF_INET6}, {"[3ffe::1234]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe::1234]:443", "[3ffe::5001]:443", @@ -591,7 +591,7 @@ TEST_F(AddressSortingTest, TestPrefersLongestPrefixStressInnerBytePrefix) { {"[3ffe:8000::]:443", AF_INET6}, {"[3ffe:2000::]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe:2000::]:443", "[3ffe:8000::]:443", @@ -611,7 +611,7 @@ TEST_F(AddressSortingTest, TestPrefersLongestPrefixDiffersOnHighestBitOfByte) { {"[3ffe:6::]:443", AF_INET6}, {"[3ffe:c::]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe:c::]:443", "[3ffe:6::]:443", @@ -633,7 +633,7 @@ TEST_F(AddressSortingTest, TestPrefersLongestPrefixDiffersByLastBit) { {"[3ffe:1111:1111:1110::]:443", AF_INET6}, {"[3ffe:1111:1111:1111::]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe:1111:1111:1111::]:443", "[3ffe:1111:1111:1110::]:443", @@ -655,7 +655,7 @@ TEST_F(AddressSortingTest, TestStableSort) { {"[3ffe::1234]:443", AF_INET6}, {"[3ffe::1235]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe::1234]:443", "[3ffe::1235]:443", @@ -681,7 +681,7 @@ TEST_F(AddressSortingTest, TestStableSortFiveElements) { {"[3ffe::1234]:443", AF_INET6}, {"[3ffe::1235]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe::1231]:443", "[3ffe::1232]:443", @@ -702,7 +702,7 @@ TEST_F(AddressSortingTest, TestStableSortNoSrcAddrsExist) { {"[3ffe::1234]:443", AF_INET6}, {"[3ffe::1235]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[3ffe::1231]:443", "[3ffe::1232]:443", @@ -720,7 +720,7 @@ TEST_F(AddressSortingTest, TestStableSortNoSrcAddrsExistWithIpv4) { {"[::ffff:5.6.7.8]:443", AF_INET6}, {"1.2.3.4:443", AF_INET}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[::ffff:5.6.7.8]:443", "1.2.3.4:443", @@ -748,7 +748,7 @@ TEST_F(AddressSortingTest, TestStableSortV4CompatAndSiteLocalAddresses) { {"[fec0::2000]:443", AF_INET6}, {v4_compat_dest, AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { // The sort should be stable since @@ -769,7 +769,7 @@ TEST_F(AddressSortingTest, TestPrefersIpv6Loopback) { {"[::1]:443", AF_INET6}, {"127.0.0.1:443", AF_INET}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[::1]:443", "127.0.0.1:443", @@ -783,7 +783,7 @@ TEST_F(AddressSortingTest, TestPrefersIpv6LoopbackInputsFlipped) { {"127.0.0.1:443", AF_INET}, {"[::1]:443", AF_INET6}, }); - grpc_cares_wrapper_test_only_address_sorting_sort(lb_addrs); + grpc_cares_wrapper_address_sorting_sort(lb_addrs); VerifyLbAddrOutputs(lb_addrs, { "[::1]:443", "127.0.0.1:443", diff --git a/test/cpp/qps/qps_benchmark_script.bzl b/test/cpp/qps/qps_benchmark_script.bzl index 0904ceb036..b2b67d988c 100644 --- a/test/cpp/qps/qps_benchmark_script.bzl +++ b/test/cpp/qps/qps_benchmark_script.bzl @@ -49,6 +49,9 @@ def qps_json_driver_batch(): "//test/cpp/util:test_config", "//test/cpp/util:test_util", ], + tags = [ + "qps_json_driver", + ], ) def json_run_localhost_batch(): @@ -71,4 +74,7 @@ def json_run_localhost_batch(): "//test/cpp/util:test_config", "//test/cpp/util:test_util", ], + tags = [ + "json_run_localhost", + ], ) diff --git a/test/cpp/util/channel_trace_proto_helper.cc b/test/cpp/util/channel_trace_proto_helper.cc index b4704bfe6a..42a436d49b 100644 --- a/test/cpp/util/channel_trace_proto_helper.cc +++ b/test/cpp/util/channel_trace_proto_helper.cc @@ -82,5 +82,18 @@ void ValidateGetChannelResponseProtoJsonTranslation(char* json_c_str) { json_c_str); } +void ValidateSubchannelProtoJsonTranslation(char* json_c_str) { + VaidateProtoJsonTranslation<grpc::channelz::v1::Subchannel>(json_c_str); +} + +void ValidateServerProtoJsonTranslation(char* json_c_str) { + VaidateProtoJsonTranslation<grpc::channelz::v1::Server>(json_c_str); +} + +void ValidateGetServersResponseProtoJsonTranslation(char* json_c_str) { + VaidateProtoJsonTranslation<grpc::channelz::v1::GetServersResponse>( + json_c_str); +} + } // namespace testing } // namespace grpc diff --git a/test/cpp/util/channel_trace_proto_helper.h b/test/cpp/util/channel_trace_proto_helper.h index 18e3d54b6b..67c363e89b 100644 --- a/test/cpp/util/channel_trace_proto_helper.h +++ b/test/cpp/util/channel_trace_proto_helper.h @@ -26,6 +26,9 @@ void ValidateChannelTraceProtoJsonTranslation(char* json_c_str); void ValidateChannelProtoJsonTranslation(char* json_c_str); void ValidateGetTopChannelsResponseProtoJsonTranslation(char* json_c_str); void ValidateGetChannelResponseProtoJsonTranslation(char* json_c_str); +void ValidateSubchannelProtoJsonTranslation(char* json_c_str); +void ValidateServerProtoJsonTranslation(char* json_c_str); +void ValidateGetServersResponseProtoJsonTranslation(char* json_c_str); } // namespace testing } // namespace grpc diff --git a/test/cpp/util/cli_credentials.cc b/test/cpp/util/cli_credentials.cc index acf4ef8ef1..1125b2d945 100644 --- a/test/cpp/util/cli_credentials.cc +++ b/test/cpp/util/cli_credentials.cc @@ -19,6 +19,11 @@ #include "test/cpp/util/cli_credentials.h" #include <gflags/gflags.h> +#include <grpc/slice.h> +#include <grpc/support/log.h> +#include <grpcpp/impl/codegen/slice.h> + +#include "src/core/lib/iomgr/load_file.h" DEFINE_bool( enable_ssl, false, @@ -28,19 +33,52 @@ DEFINE_bool(use_auth, false, "--channel_creds_type=gdc."); DEFINE_string( access_token, "", - "The access token that will be sent to the server to authenticate RPCs."); + "The access token that will be sent to the server to authenticate RPCs. " + "Deprecated. Use --call_creds=access_token=<token>."); DEFINE_string( ssl_target, "", "If not empty, treat the server host name as this for ssl/tls certificate " "validation."); DEFINE_string( + ssl_client_cert, "", + "If not empty, load this PEM formated client certificate file. Requires " + "use of --ssl_client_key."); +DEFINE_string( + ssl_client_key, "", + "If not empty, load this PEM formated private key. Requires use of " + "--ssl_client_cert"); +DEFINE_string( channel_creds_type, "", "The channel creds type: insecure, ssl, gdc (Google Default Credentials) " "or alts."); +DEFINE_string( + call_creds, "", + "Call credentials to use: none (default), or access_token=<token>. If " + "provided, the call creds are composited on top of channel creds."); namespace grpc { namespace testing { +namespace { + +const char ACCESS_TOKEN_PREFIX[] = "access_token="; +constexpr int ACCESS_TOKEN_PREFIX_LEN = + sizeof(ACCESS_TOKEN_PREFIX) / sizeof(*ACCESS_TOKEN_PREFIX) - 1; + +bool IsAccessToken(const grpc::string& auth) { + return auth.length() > ACCESS_TOKEN_PREFIX_LEN && + auth.compare(0, ACCESS_TOKEN_PREFIX_LEN, ACCESS_TOKEN_PREFIX) == 0; +} + +grpc::string AccessToken(const grpc::string& auth) { + if (!IsAccessToken(auth)) { + return ""; + } + return grpc::string(auth, ACCESS_TOKEN_PREFIX_LEN); +} + +} // namespace + grpc::string CliCredentials::GetDefaultChannelCredsType() const { // Compatibility logic for --enable_ssl. if (FLAGS_enable_ssl) { @@ -59,12 +97,42 @@ grpc::string CliCredentials::GetDefaultChannelCredsType() const { return "insecure"; } +grpc::string CliCredentials::GetDefaultCallCreds() const { + if (!FLAGS_access_token.empty()) { + fprintf(stderr, + "warning: --access_token is deprecated. Use " + "--call_creds=access_token=<token>.\n"); + return grpc::string("access_token=") + FLAGS_access_token; + } + return "none"; +} + std::shared_ptr<grpc::ChannelCredentials> CliCredentials::GetChannelCredentials() const { if (FLAGS_channel_creds_type.compare("insecure") == 0) { return grpc::InsecureChannelCredentials(); } else if (FLAGS_channel_creds_type.compare("ssl") == 0) { - return grpc::SslCredentials(grpc::SslCredentialsOptions()); + grpc::SslCredentialsOptions ssl_creds_options; + // TODO(@Capstan): This won't affect Google Default Credentials using SSL. + if (!FLAGS_ssl_client_cert.empty()) { + grpc_slice cert_slice = grpc_empty_slice(); + GRPC_LOG_IF_ERROR( + "load_file", + grpc_load_file(FLAGS_ssl_client_cert.c_str(), 1, &cert_slice)); + ssl_creds_options.pem_cert_chain = + grpc::StringFromCopiedSlice(cert_slice); + grpc_slice_unref(cert_slice); + } + if (!FLAGS_ssl_client_key.empty()) { + grpc_slice key_slice = grpc_empty_slice(); + GRPC_LOG_IF_ERROR( + "load_file", + grpc_load_file(FLAGS_ssl_client_key.c_str(), 1, &key_slice)); + ssl_creds_options.pem_private_key = + grpc::StringFromCopiedSlice(key_slice); + grpc_slice_unref(key_slice); + } + return grpc::SslCredentials(ssl_creds_options); } else if (FLAGS_channel_creds_type.compare("gdc") == 0) { return grpc::GoogleDefaultCredentials(); } else if (FLAGS_channel_creds_type.compare("alts") == 0) { @@ -80,18 +148,30 @@ CliCredentials::GetChannelCredentials() const { std::shared_ptr<grpc::CallCredentials> CliCredentials::GetCallCredentials() const { - if (!FLAGS_access_token.empty()) { - if (FLAGS_use_auth) { - fprintf(stderr, - "warning: use_auth is ignored when access_token is provided."); - } - return grpc::AccessTokenCredentials(FLAGS_access_token); + if (IsAccessToken(FLAGS_call_creds)) { + return grpc::AccessTokenCredentials(AccessToken(FLAGS_call_creds)); + } + if (FLAGS_call_creds.compare("none") != 0) { + // Nothing to do; creds, if any, are baked into the channel. + return std::shared_ptr<grpc::CallCredentials>(); } + fprintf(stderr, + "--call_creds=%s invalid; must be none " + "or access_token=<token>.\n", + FLAGS_call_creds.c_str()); return std::shared_ptr<grpc::CallCredentials>(); } std::shared_ptr<grpc::ChannelCredentials> CliCredentials::GetCredentials() const { + if (FLAGS_call_creds.empty()) { + FLAGS_call_creds = GetDefaultCallCreds(); + } else if (!FLAGS_access_token.empty() && !IsAccessToken(FLAGS_call_creds)) { + fprintf(stderr, + "warning: ignoring --access_token because --call_creds " + "already set to %s.\n", + FLAGS_call_creds.c_str()); + } if (FLAGS_channel_creds_type.empty()) { FLAGS_channel_creds_type = GetDefaultChannelCredsType(); } else if (FLAGS_enable_ssl && FLAGS_channel_creds_type.compare("ssl") != 0) { @@ -106,7 +186,7 @@ std::shared_ptr<grpc::ChannelCredentials> CliCredentials::GetCredentials() FLAGS_channel_creds_type.c_str()); } // Legacy transport upgrade logic for insecure requests. - if (!FLAGS_access_token.empty() && + if (IsAccessToken(FLAGS_call_creds) && FLAGS_channel_creds_type.compare("insecure") == 0) { fprintf(stderr, "warning: --channel_creds_type=insecure upgraded to ssl because " @@ -126,10 +206,16 @@ const grpc::string CliCredentials::GetCredentialUsage() const { return " --enable_ssl ; Set whether to use ssl (deprecated)\n" " --use_auth ; Set whether to create default google" " credentials\n" + " ; (deprecated)\n" " --access_token ; Set the access token in metadata," " overrides --use_auth\n" + " ; (deprecated)\n" " --ssl_target ; Set server host for ssl validation\n" - " --channel_creds_type ; Set to insecure, ssl, gdc, or alts\n"; + " --ssl_client_cert ; Client cert for ssl\n" + " --ssl_client_key ; Client private key for ssl\n" + " --channel_creds_type ; Set to insecure, ssl, gdc, or alts\n" + " --call_creds ; Set to none, or" + " access_token=<token>\n"; } const grpc::string CliCredentials::GetSslTargetNameOverride() const { diff --git a/test/cpp/util/cli_credentials.h b/test/cpp/util/cli_credentials.h index 4636d3ca14..472c7abef2 100644 --- a/test/cpp/util/cli_credentials.h +++ b/test/cpp/util/cli_credentials.h @@ -36,6 +36,9 @@ class CliCredentials { // Returns the appropriate channel_creds_type value for the set of legacy // flag arguments. virtual grpc::string GetDefaultChannelCredsType() const; + // Returns the appropriate call_creds value for the set of legacy flag + // arguments. + virtual grpc::string GetDefaultCallCreds() const; // Returns the base transport channel credentials. Child classes can override // to support additional channel_creds_types unknown to this base class. virtual std::shared_ptr<grpc::ChannelCredentials> GetChannelCredentials() diff --git a/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile b/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile index bff20b5d06..897354891c 100644 --- a/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile +++ b/tools/dockerfile/interoptest/grpc_interop_dart/Dockerfile @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -FROM google/dart:latest +FROM google/dart:2.0 # Upgrade Dart to version 2. RUN apt-get update && apt-get upgrade -y dart diff --git a/tools/doxygen/Doxyfile.c++ b/tools/doxygen/Doxyfile.c++ index 9a97ee84f2..3b7fd1fa8e 100644 --- a/tools/doxygen/Doxyfile.c++ +++ b/tools/doxygen/Doxyfile.c++ @@ -945,7 +945,9 @@ include/grpcpp/impl/codegen/async_unary_call.h \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ +include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ +include/grpcpp/impl/codegen/client_callback.h \ include/grpcpp/impl/codegen/client_context.h \ include/grpcpp/impl/codegen/client_unary_call.h \ include/grpcpp/impl/codegen/completion_queue.h \ @@ -997,6 +999,7 @@ include/grpcpp/support/async_stream.h \ include/grpcpp/support/async_unary_call.h \ include/grpcpp/support/byte_buffer.h \ include/grpcpp/support/channel_arguments.h \ +include/grpcpp/support/client_callback.h \ include/grpcpp/support/config.h \ include/grpcpp/support/proto_buffer_reader.h \ include/grpcpp/support/proto_buffer_writer.h \ diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index 0cd4cfd647..a72390d9f8 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -946,7 +946,9 @@ include/grpcpp/impl/codegen/async_unary_call.h \ include/grpcpp/impl/codegen/byte_buffer.h \ include/grpcpp/impl/codegen/call.h \ include/grpcpp/impl/codegen/call_hook.h \ +include/grpcpp/impl/codegen/callback_common.h \ include/grpcpp/impl/codegen/channel_interface.h \ +include/grpcpp/impl/codegen/client_callback.h \ include/grpcpp/impl/codegen/client_context.h \ include/grpcpp/impl/codegen/client_unary_call.h \ include/grpcpp/impl/codegen/completion_queue.h \ @@ -999,6 +1001,7 @@ include/grpcpp/support/async_stream.h \ include/grpcpp/support/async_unary_call.h \ include/grpcpp/support/byte_buffer.h \ include/grpcpp/support/channel_arguments.h \ +include/grpcpp/support/client_callback.h \ include/grpcpp/support/config.h \ include/grpcpp/support/proto_buffer_reader.h \ include/grpcpp/support/proto_buffer_writer.h \ @@ -1187,6 +1190,7 @@ src/cpp/client/secure_credentials.h \ src/cpp/codegen/codegen_init.cc \ src/cpp/common/alarm.cc \ src/cpp/common/auth_property_iterator.cc \ +src/cpp/common/callback_common.cc \ src/cpp/common/channel_arguments.cc \ src/cpp/common/channel_filter.cc \ src/cpp/common/channel_filter.h \ diff --git a/tools/internal_ci/linux/grpc_asan_on_foundry.sh b/tools/internal_ci/linux/grpc_asan_on_foundry.sh index a6367ad962..dfef004a60 100644..100755 --- a/tools/internal_ci/linux/grpc_asan_on_foundry.sh +++ b/tools/internal_ci/linux/grpc_asan_on_foundry.sh @@ -15,5 +15,6 @@ export UPLOAD_TEST_RESULTS=true EXTRA_FLAGS="--copt=-gmlt --strip=never --copt=-fsanitize=address --linkopt=-fsanitize=address --test_timeout=3600 --cache_test_results=no" -github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" +EXCLUDE_TESTS="--test_tag_filters=-qps_json_driver,-json_run_localhost" +github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" "${EXCLUDE_TESTS}" diff --git a/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh b/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh index 8b42779366..bb2a85138c 100755 --- a/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh +++ b/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh @@ -60,7 +60,7 @@ export KOKORO_FOUNDRY_PROJECT_ID="projects/grpc-testing/instances/default_instan --platforms=//third_party/toolchains:rbe_ubuntu1604 \ --test_env=GRPC_VERBOSITY=debug \ --remote_instance_name=projects/grpc-testing/instances/default_instance \ - $1 \ + $@ \ -- //test/... || FAILED="true" if [ "$UPLOAD_TEST_RESULTS" != "" ] diff --git a/tools/internal_ci/linux/grpc_tsan_on_foundry.sh b/tools/internal_ci/linux/grpc_tsan_on_foundry.sh index 2ba7d469ec..366b5cbe34 100644 --- a/tools/internal_ci/linux/grpc_tsan_on_foundry.sh +++ b/tools/internal_ci/linux/grpc_tsan_on_foundry.sh @@ -15,4 +15,5 @@ export UPLOAD_TEST_RESULTS=true EXTRA_FLAGS="--copt=-gmlt --strip=never --copt=-fsanitize=thread --linkopt=-fsanitize=thread --test_timeout=3600 --action_env=TSAN_OPTIONS=suppressions=test/core/util/tsan_suppressions.txt:halt_on_error=1:second_deadlock_stack=1 --cache_test_results=no" -github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" +EXCLUDE_TESTS="--test_tag_filters=-qps_json_driver,-json_run_localhost" +github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" "${EXCLUDE_TESTS}" diff --git a/tools/internal_ci/linux/pull_request/grpc_asan_on_foundry.sh b/tools/internal_ci/linux/pull_request/grpc_asan_on_foundry.sh index 2aebb65552..39c991f291 100644 --- a/tools/internal_ci/linux/pull_request/grpc_asan_on_foundry.sh +++ b/tools/internal_ci/linux/pull_request/grpc_asan_on_foundry.sh @@ -14,5 +14,6 @@ # limitations under the License. EXTRA_FLAGS="--copt=-gmlt --strip=never --copt=-fsanitize=address --linkopt=-fsanitize=address --test_timeout=3600" -github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" +EXCLUDE_TESTS="--test_tag_filters=-qps_json_driver,-json_run_localhost" +github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" "${EXCLUDE_TESTS}" diff --git a/tools/internal_ci/linux/pull_request/grpc_tsan_on_foundry.sh b/tools/internal_ci/linux/pull_request/grpc_tsan_on_foundry.sh index edd8f92975..3dee115300 100644 --- a/tools/internal_ci/linux/pull_request/grpc_tsan_on_foundry.sh +++ b/tools/internal_ci/linux/pull_request/grpc_tsan_on_foundry.sh @@ -14,4 +14,5 @@ # limitations under the License. EXTRA_FLAGS="--copt=-gmlt --strip=never --copt=-fsanitize=thread --linkopt=-fsanitize=thread --test_timeout=3600 --action_env=TSAN_OPTIONS=suppressions=test/core/util/tsan_suppressions.txt:halt_on_error=1:second_deadlock_stack=1" -github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" +EXCLUDE_TESTS="--test_tag_filters=-qps_json_driver,-json_run_localhost" +github/grpc/tools/internal_ci/linux/grpc_bazel_on_foundry_base.sh "${EXTRA_FLAGS}" "${EXCLUDE_TESTS}" diff --git a/tools/interop_matrix/client_matrix.py b/tools/interop_matrix/client_matrix.py index bb9222d953..c55789f27e 100644 --- a/tools/interop_matrix/client_matrix.py +++ b/tools/interop_matrix/client_matrix.py @@ -184,6 +184,9 @@ LANG_RELEASE_MATRIX = { { 'v1.14.0': None }, + { + 'v1.15.0': None + }, ], 'python': [ { diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index 8ea5126fde..f3e93a0874 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -1897,7 +1897,24 @@ "headers": [], "is_filegroup": false, "language": "c", - "name": "resolve_address_test", + "name": "resolve_address_using_ares_resolver_test", + "src": [ + "test/core/iomgr/resolve_address_test.cc" + ], + "third_party": false, + "type": "target" + }, + { + "deps": [ + "gpr", + "gpr_test_util", + "grpc", + "grpc_test_util" + ], + "headers": [], + "is_filegroup": false, + "language": "c", + "name": "resolve_address_using_native_resolver_test", "src": [ "test/core/iomgr/resolve_address_test.cc" ], @@ -3331,6 +3348,25 @@ "grpc++_test_util", "grpc_test_util" ], + "headers": [], + "is_filegroup": false, + "language": "c++", + "name": "client_callback_end2end_test", + "src": [ + "test/cpp/end2end/client_callback_end2end_test.cc" + ], + "third_party": false, + "type": "target" + }, + { + "deps": [ + "gpr", + "gpr_test_util", + "grpc", + "grpc++", + "grpc++_test_util", + "grpc_test_util" + ], "headers": [ "src/proto/grpc/lb/v1/load_balancer.grpc.pb.h", "src/proto/grpc/lb/v1/load_balancer.pb.h", @@ -11081,7 +11117,9 @@ "include/grpcpp/impl/codegen/byte_buffer.h", "include/grpcpp/impl/codegen/call.h", "include/grpcpp/impl/codegen/call_hook.h", + "include/grpcpp/impl/codegen/callback_common.h", "include/grpcpp/impl/codegen/channel_interface.h", + "include/grpcpp/impl/codegen/client_callback.h", "include/grpcpp/impl/codegen/client_context.h", "include/grpcpp/impl/codegen/client_unary_call.h", "include/grpcpp/impl/codegen/completion_queue.h", @@ -11147,7 +11185,9 @@ "include/grpcpp/impl/codegen/byte_buffer.h", "include/grpcpp/impl/codegen/call.h", "include/grpcpp/impl/codegen/call_hook.h", + "include/grpcpp/impl/codegen/callback_common.h", "include/grpcpp/impl/codegen/channel_interface.h", + "include/grpcpp/impl/codegen/client_callback.h", "include/grpcpp/impl/codegen/client_context.h", "include/grpcpp/impl/codegen/client_unary_call.h", "include/grpcpp/impl/codegen/completion_queue.h", @@ -11305,6 +11345,7 @@ "include/grpcpp/support/async_unary_call.h", "include/grpcpp/support/byte_buffer.h", "include/grpcpp/support/channel_arguments.h", + "include/grpcpp/support/client_callback.h", "include/grpcpp/support/config.h", "include/grpcpp/support/proto_buffer_reader.h", "include/grpcpp/support/proto_buffer_writer.h", @@ -11409,6 +11450,7 @@ "include/grpcpp/support/async_unary_call.h", "include/grpcpp/support/byte_buffer.h", "include/grpcpp/support/channel_arguments.h", + "include/grpcpp/support/client_callback.h", "include/grpcpp/support/config.h", "include/grpcpp/support/proto_buffer_reader.h", "include/grpcpp/support/proto_buffer_writer.h", @@ -11428,6 +11470,7 @@ "src/cpp/client/credentials_cc.cc", "src/cpp/client/generic_stub.cc", "src/cpp/common/alarm.cc", + "src/cpp/common/callback_common.cc", "src/cpp/common/channel_arguments.cc", "src/cpp/common/channel_filter.cc", "src/cpp/common/channel_filter.h", diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index fba76d69d1..b3c07d9215 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -2146,7 +2146,35 @@ "uses_polling": true }, { - "args": [], + "args": [ + "--resolver=ares" + ], + "benchmark": false, + "ci_platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "cpu_cost": 1.0, + "exclude_configs": [], + "exclude_iomgrs": [], + "flaky": false, + "gtest": false, + "language": "c", + "name": "resolve_address_using_ares_resolver_test", + "platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "uses_polling": true + }, + { + "args": [ + "--resolver=native" + ], "benchmark": false, "ci_platforms": [ "linux", @@ -2160,7 +2188,7 @@ "flaky": false, "gtest": false, "language": "c", - "name": "resolve_address_test", + "name": "resolve_address_using_native_resolver_test", "platforms": [ "linux", "mac", @@ -3978,6 +4006,30 @@ "posix", "windows" ], + "cpu_cost": 0.5, + "exclude_configs": [], + "exclude_iomgrs": [], + "flaky": false, + "gtest": true, + "language": "c++", + "name": "client_callback_end2end_test", + "platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "uses_polling": true + }, + { + "args": [], + "benchmark": false, + "ci_platforms": [ + "linux", + "mac", + "posix", + "windows" + ], "cpu_cost": 1.0, "exclude_configs": [], "exclude_iomgrs": [], diff --git a/tools/run_tests/python_utils/upload_rbe_results.py b/tools/run_tests/python_utils/upload_rbe_results.py index d29ebc6219..74f329048f 100644 --- a/tools/run_tests/python_utils/upload_rbe_results.py +++ b/tools/run_tests/python_utils/upload_rbe_results.py @@ -40,13 +40,14 @@ _RESULTS_SCHEMA = [ ('test_case', 'STRING', 'Name of test case'), ('result', 'STRING', 'Test or build result'), ('timestamp', 'TIMESTAMP', 'Timestamp of test run'), + ('duration', 'FLOAT', 'Duration of the test run'), ] _TABLE_ID = 'rbe_test_results' def _get_api_key(): """Returns string with API key to access ResultStore. - Intended to be used in Kokoro envrionment.""" + Intended to be used in Kokoro environment.""" api_key_directory = os.getenv('KOKORO_GFILE_DIR') api_key_file = os.path.join(api_key_directory, 'resultstore_api_key') assert os.path.isfile(api_key_file), 'Must add --api_key arg if not on ' \ @@ -57,7 +58,7 @@ def _get_api_key(): def _get_invocation_id(): """Returns String of Bazel invocation ID. Intended to be used in - Kokoro envirionment.""" + Kokoro environment.""" bazel_id_directory = os.getenv('KOKORO_ARTIFACTS_DIR') bazel_id_file = os.path.join(bazel_id_directory, 'bazel_invocation_ids') assert os.path.isfile(bazel_id_file), 'bazel_invocation_ids file, written ' \ @@ -66,6 +67,16 @@ def _get_invocation_id(): return f.read().replace('\n', '') +def _parse_test_duration(duration_str): + """Parse test duration string in '123.567s' format""" + try: + if duration_str.endswith('s'): + duration_str = duration_str[:-1] + return float(duration_str) + except: + return None + + def _upload_results_to_bq(rows): """Upload test results to a BQ table. @@ -205,6 +216,8 @@ if __name__ == "__main__": result, 'timestamp': action['timing']['startTime'], + 'duration': + _parse_test_duration(action['timing']['duration']), } }) except Exception as e: |