diff options
50 files changed, 1124 insertions, 679 deletions
@@ -1006,7 +1006,6 @@ grpc_cc_library( "include/grpc/impl/codegen/byte_buffer_reader.h", "include/grpc/impl/codegen/compression_types.h", "include/grpc/impl/codegen/connectivity_state.h", - "include/grpc/impl/codegen/exec_ctx_fwd.h", "include/grpc/impl/codegen/grpc_types.h", "include/grpc/impl/codegen/propagation_bits.h", "include/grpc/impl/codegen/status.h", diff --git a/CMakeLists.txt b/CMakeLists.txt index eed1205268..78ccfb2132 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -594,6 +594,7 @@ add_dependencies(buildtests_cxx stress_test) add_dependencies(buildtests_cxx thread_manager_test) add_dependencies(buildtests_cxx thread_stress_test) add_dependencies(buildtests_cxx transport_pid_controller_test) +add_dependencies(buildtests_cxx vector_test) if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX) add_dependencies(buildtests_cxx writes_per_rpc_test) endif() @@ -1080,7 +1081,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -1394,7 +1394,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -1680,7 +1679,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -1950,7 +1948,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -2239,7 +2236,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -2552,7 +2548,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -3038,7 +3033,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -3438,7 +3432,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -3579,7 +3572,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -3784,7 +3776,6 @@ foreach(_hdr include/grpc/impl/codegen/byte_buffer_reader.h include/grpc/impl/codegen/compression_types.h include/grpc/impl/codegen/connectivity_state.h - include/grpc/impl/codegen/exec_ctx_fwd.h include/grpc/impl/codegen/grpc_types.h include/grpc/impl/codegen/propagation_bits.h include/grpc/impl/codegen/slice.h @@ -12497,6 +12488,45 @@ target_link_libraries(transport_pid_controller_test endif (gRPC_BUILD_TESTS) if (gRPC_BUILD_TESTS) + +add_executable(vector_test + test/core/support/vector_test.cc + third_party/googletest/googletest/src/gtest-all.cc + third_party/googletest/googlemock/src/gmock-all.cc +) + + +target_include_directories(vector_test + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include + PRIVATE ${_gRPC_SSL_INCLUDE_DIR} + PRIVATE ${PROTOBUF_ROOT_DIR}/src + PRIVATE ${BENCHMARK_ROOT_DIR}/include + PRIVATE ${ZLIB_ROOT_DIR} + PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/zlib + PRIVATE ${CARES_INCLUDE_DIR} + PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/cares/cares + PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/gflags/include + 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(vector_test + ${_gRPC_PROTOBUF_LIBRARIES} + ${_gRPC_ALLTARGETS_LIBRARIES} + grpc_test_util + grpc++ + grpc + gpr_test_util + gpr + ${_gRPC_GFLAGS_LIBRARIES} +) + +endif (gRPC_BUILD_TESTS) +if (gRPC_BUILD_TESTS) if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX) add_executable(writes_per_rpc_test diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fc02f16f3b..c2dc4eba57 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -2,53 +2,92 @@ We definitely welcome your patches and contributions to gRPC! -If you are new to github, please start by reading [Pull Request howto](https://help.github.com/articles/about-pull-requests/) +If you are new to github, please start by reading [Pull Request +howto](https://help.github.com/articles/about-pull-requests/) ## Legal requirements In order to protect both you and ourselves, you will need to sign the -[Contributor License Agreement](https://identity.linuxfoundation.org/projects/cncf). +[Contributor License +Agreement](https://identity.linuxfoundation.org/projects/cncf). ## Running tests -Use `tools/run_tests/run_tests.py` script to run the unit tests. -See [tools/run_tests](tools/run_tests) for how to run tests for a given language. +Use `tools/run_tests/run_tests.py` script to run the unit tests. See +[tools/run_tests](tools/run_tests) for how to run tests for a given language. -Prerequisites for building and running tests are listed in [INSTALL.md](INSTALL.md) -and in `src/YOUR-LANGUAGE` (e.g. `src/csharp`) +Prerequisites for building and running tests are listed in +[INSTALL.md](INSTALL.md) and in `src/YOUR-LANGUAGE` (e.g. `src/csharp`) ## Generated project files -To ease maintenance of language- and platform- specific build systems, -many projects files are generated using templates and should not be edited -by hand. -Run `tools/buildgen/generate_projects.sh` to regenerate. -See [templates](templates) for details. +To ease maintenance of language- and platform- specific build systems, many +projects files are generated using templates and should not be edited by hand. +Run `tools/buildgen/generate_projects.sh` to regenerate. See +[templates](templates) for details. -As a rule of thumb, if you see the "sanity tests" failing you've most likely edited generated files or you didn't regenerate the projects properly (or your code formatting doesn't match our code style). +As a rule of thumb, if you see the "sanity tests" failing you've most likely +edited generated files or you didn't regenerate the projects properly (or your +code formatting doesn't match our code style). ## Guidelines for Pull Requests How to get your contributions merged smoothly and quickly. -- Create **small PRs** that are narrowly focused on **addressing a single concern**. We often times receive PRs that are trying to fix several things at a time, but only one fix is considered acceptable, nothing gets merged and both author's & review's time is wasted. Create more PRs to address different concerns and everyone will be happy. +- Create **small PRs** that are narrowly focused on **addressing a single + concern**. We often times receive PRs that are trying to fix several things + at a time, but only one fix is considered acceptable, nothing gets merged and + both author's & review's time is wasted. Create more PRs to address different + concerns and everyone will be happy. -- For speculative changes, consider opening an issue and discussing it first. If you are suggesting a behavioral or API change, consider starting with a [gRFC proposal](https://github.com/grpc/proposal). +- For speculative changes, consider opening an issue and discussing it first. + If you are suggesting a behavioral or API change, consider starting with a + [gRFC proposal](https://github.com/grpc/proposal). -- Provide a good **PR description** as a record of **what** change is being made and **why** it was made. Link to a github issue if it exists. +- Provide a good **PR description** as a record of **what** change is being made + and **why** it was made. Link to a GitHub issue if it exists. -- Don't fix code style and formatting unless you are already changing that line to address an issue. PRs with irrelevant changes won't be merged. If you do want to fix formatting or style, do that in a separate PR. +- Don't fix code style and formatting unless you are already changing that line + to address an issue. PRs with irrelevant changes won't be merged. If you do + want to fix formatting or style, do that in a separate PR. -- Unless your PR is trivial, you should expect there will be reviewer comments that you'll need to address before merging. We expect you to be reasonably responsive to those comments, otherwise the PR will be closed after 2-3 weeks of inactivity. +- Unless your PR is trivial, you should expect there will be reviewer comments + that you'll need to address before merging. We expect you to be reasonably + responsive to those comments, otherwise the PR will be closed after 2-3 weeks + of inactivity. + +- If you have non-trivial contributions, please consider adding an entry to [the + AUTHORS file](https://github.com/grpc/grpc/blob/master/AUTHORS) listing the + copyright holder for the contribution (yourself, if you are signing the + individual CLA, or your company, for corporate CLAs) in the same PR as your + contribution. This needs to be done only once, for each company, or + individual. -- Maintain **clean commit history** and use **meaningful commit messages**. PRs with messy commit history are difficult to review and won't be merged. Use `rebase -i upstream/master` to curate your commit history and/or to bring in latest changes from master (but avoid rebasing in the middle of a code review). +- Maintain **clean commit history** and use **meaningful commit messages**. + PRs with messy commit history are difficult to review and won't be merged. + Use `rebase -i upstream/master` to curate your commit history and/or to + bring in latest changes from master (but avoid rebasing in the middle of + a code review). -- Keep your PR up to date with upstream/master (if there are merge conflicts, we can't really merge your change). +- Keep your PR up to date with upstream/master (if there are merge conflicts, + we can't really merge your change). -- if you are regenerating the projects using `tools/buildgen/generate_projects.sh`, make changes to generated files a separate commit with commit message `regenerate projects`. Mixing changes to generated and hand-written files make your PR difficult to review. Note that running this script requires the installation of Python packages `pyyaml` and `mako` (typically installed using `pip`) as well as a recent version of [`go`](https://golang.org/doc/install#install). +- If you are regenerating the projects using + `tools/buildgen/generate_projects.sh`, make changes to generated files a + separate commit with commit message `regenerate projects`. Mixing changes + to generated and hand-written files make your PR difficult to review. + Note that running this script requires the installation of Python packages + `pyyaml` and `mako` (typically installed using `pip`) as well as a recent + version of [`go`](https://golang.org/doc/install#install). -- **All tests need to be passing** before your change can be merged. We recommend you **run tests locally** before creating your PR to catch breakages early on (see [tools/run_tests](tools/run_tests). Ultimately, the green signal will be provided by our testing infrastructure. The reviewer will help you if there are test failures that seem not related to the change you are making. +- **All tests need to be passing** before your change can be merged. + We recommend you **run tests locally** before creating your PR to catch + breakages early on (see [tools/run_tests](tools/run_tests). Ultimately, the + green signal will be provided by our testing infrastructure. The reviewer + will help you if there are test failures that seem not related to the change + you are making. -- Exceptions to the rules can be made if there's a compelling reason for doing so. +- Exceptions to the rules can be made if there's a compelling reason for doing + so. @@ -1180,6 +1180,7 @@ stress_test: $(BINDIR)/$(CONFIG)/stress_test thread_manager_test: $(BINDIR)/$(CONFIG)/thread_manager_test thread_stress_test: $(BINDIR)/$(CONFIG)/thread_stress_test transport_pid_controller_test: $(BINDIR)/$(CONFIG)/transport_pid_controller_test +vector_test: $(BINDIR)/$(CONFIG)/vector_test writes_per_rpc_test: $(BINDIR)/$(CONFIG)/writes_per_rpc_test public_headers_must_be_c89: $(BINDIR)/$(CONFIG)/public_headers_must_be_c89 gen_hpack_tables: $(BINDIR)/$(CONFIG)/gen_hpack_tables @@ -1620,6 +1621,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/thread_manager_test \ $(BINDIR)/$(CONFIG)/thread_stress_test \ $(BINDIR)/$(CONFIG)/transport_pid_controller_test \ + $(BINDIR)/$(CONFIG)/vector_test \ $(BINDIR)/$(CONFIG)/writes_per_rpc_test \ $(BINDIR)/$(CONFIG)/boringssl_aes_test \ $(BINDIR)/$(CONFIG)/boringssl_asn1_test \ @@ -1749,6 +1751,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/thread_manager_test \ $(BINDIR)/$(CONFIG)/thread_stress_test \ $(BINDIR)/$(CONFIG)/transport_pid_controller_test \ + $(BINDIR)/$(CONFIG)/vector_test \ $(BINDIR)/$(CONFIG)/writes_per_rpc_test \ $(BINDIR)/$(CONFIG)/resolver_component_test_unsecure \ $(BINDIR)/$(CONFIG)/resolver_component_test \ @@ -2167,6 +2170,8 @@ test_cxx: buildtests_cxx $(Q) $(BINDIR)/$(CONFIG)/thread_stress_test || ( echo test thread_stress_test failed ; exit 1 ) $(E) "[RUN] Testing transport_pid_controller_test" $(Q) $(BINDIR)/$(CONFIG)/transport_pid_controller_test || ( echo test transport_pid_controller_test failed ; exit 1 ) + $(E) "[RUN] Testing vector_test" + $(Q) $(BINDIR)/$(CONFIG)/vector_test || ( echo test vector_test failed ; exit 1 ) $(E) "[RUN] Testing writes_per_rpc_test" $(Q) $(BINDIR)/$(CONFIG)/writes_per_rpc_test || ( echo test writes_per_rpc_test failed ; exit 1 ) $(E) "[RUN] Testing resolver_component_tests_runner_invoker_unsecure" @@ -3226,7 +3231,6 @@ PUBLIC_HEADERS_C += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -3540,7 +3544,6 @@ PUBLIC_HEADERS_C += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -3827,7 +3830,6 @@ PUBLIC_HEADERS_C += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -4088,7 +4090,6 @@ PUBLIC_HEADERS_C += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -4354,7 +4355,6 @@ PUBLIC_HEADERS_C += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -4646,7 +4646,6 @@ PUBLIC_HEADERS_CXX += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -5133,7 +5132,6 @@ PUBLIC_HEADERS_CXX += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -5526,7 +5524,6 @@ PUBLIC_HEADERS_CXX += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -5644,7 +5641,6 @@ PUBLIC_HEADERS_CXX += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -5854,7 +5850,6 @@ PUBLIC_HEADERS_CXX += \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ - include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/grpc_types.h \ include/grpc/impl/codegen/propagation_bits.h \ include/grpc/impl/codegen/slice.h \ @@ -17270,6 +17265,49 @@ endif endif +VECTOR_TEST_SRC = \ + test/core/support/vector_test.cc \ + +VECTOR_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(VECTOR_TEST_SRC)))) +ifeq ($(NO_SECURE),true) + +# You can't build secure targets if you don't have OpenSSL. + +$(BINDIR)/$(CONFIG)/vector_test: openssl_dep_error + +else + + + + +ifeq ($(NO_PROTOBUF),true) + +# You can't build the protoc plugins or protobuf-enabled targets if you don't have protobuf 3.0.0+. + +$(BINDIR)/$(CONFIG)/vector_test: protobuf_dep_error + +else + +$(BINDIR)/$(CONFIG)/vector_test: $(PROTOBUF_DEP) $(VECTOR_TEST_OBJS) $(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) $(VECTOR_TEST_OBJS) $(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)/vector_test + +endif + +endif + +$(OBJDIR)/$(CONFIG)/test/core/support/vector_test.o: $(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_vector_test: $(VECTOR_TEST_OBJS:.o=.dep) + +ifneq ($(NO_SECURE),true) +ifneq ($(NO_DEPS),true) +-include $(VECTOR_TEST_OBJS:.o=.dep) +endif +endif + + WRITES_PER_RPC_TEST_SRC = \ test/cpp/performance/writes_per_rpc_test.cc \ diff --git a/build.yaml b/build.yaml index fef7d6189f..db2ff8828b 100644 --- a/build.yaml +++ b/build.yaml @@ -399,6 +399,7 @@ filegroups: - src/core/lib/support/debug_location.h - src/core/lib/support/ref_counted.h - src/core/lib/support/ref_counted_ptr.h + - src/core/lib/support/vector.h - src/core/lib/surface/alarm_internal.h - src/core/lib/surface/api_trace.h - src/core/lib/surface/call.h @@ -485,7 +486,6 @@ filegroups: - include/grpc/impl/codegen/byte_buffer_reader.h - include/grpc/impl/codegen/compression_types.h - include/grpc/impl/codegen/connectivity_state.h - - include/grpc/impl/codegen/exec_ctx_fwd.h - include/grpc/impl/codegen/grpc_types.h - include/grpc/impl/codegen/propagation_bits.h - include/grpc/impl/codegen/slice.h @@ -4798,6 +4798,20 @@ targets: - grpc - gpr_test_util - gpr +- name: vector_test + gtest: true + build: test + language: c++ + src: + - test/core/support/vector_test.cc + deps: + - grpc_test_util + - grpc++ + - grpc + - gpr_test_util + - gpr + uses: + - grpc++_test - name: writes_per_rpc_test gtest: true cpu_cost: 0.5 diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index c127660dd5..358fad3d98 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -152,7 +152,6 @@ Pod::Spec.new do |s| 'include/grpc/impl/codegen/byte_buffer_reader.h', 'include/grpc/impl/codegen/compression_types.h', 'include/grpc/impl/codegen/connectivity_state.h', - 'include/grpc/impl/codegen/exec_ctx_fwd.h', 'include/grpc/impl/codegen/grpc_types.h', 'include/grpc/impl/codegen/propagation_bits.h', 'include/grpc/impl/codegen/slice.h', @@ -423,6 +422,7 @@ Pod::Spec.new do |s| 'src/core/lib/support/debug_location.h', 'src/core/lib/support/ref_counted.h', 'src/core/lib/support/ref_counted_ptr.h', + 'src/core/lib/support/vector.h', 'src/core/lib/surface/alarm_internal.h', 'src/core/lib/surface/api_trace.h', 'src/core/lib/surface/call.h', @@ -903,6 +903,7 @@ Pod::Spec.new do |s| 'src/core/lib/support/debug_location.h', 'src/core/lib/support/ref_counted.h', 'src/core/lib/support/ref_counted_ptr.h', + 'src/core/lib/support/vector.h', 'src/core/lib/surface/alarm_internal.h', 'src/core/lib/surface/api_trace.h', 'src/core/lib/surface/call.h', diff --git a/grpc.gemspec b/grpc.gemspec index d185995261..7547bc85de 100644 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -149,7 +149,6 @@ Gem::Specification.new do |s| s.files += %w( include/grpc/impl/codegen/byte_buffer_reader.h ) s.files += %w( include/grpc/impl/codegen/compression_types.h ) s.files += %w( include/grpc/impl/codegen/connectivity_state.h ) - s.files += %w( include/grpc/impl/codegen/exec_ctx_fwd.h ) s.files += %w( include/grpc/impl/codegen/grpc_types.h ) s.files += %w( include/grpc/impl/codegen/propagation_bits.h ) s.files += %w( include/grpc/impl/codegen/slice.h ) @@ -349,6 +348,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/support/debug_location.h ) s.files += %w( src/core/lib/support/ref_counted.h ) s.files += %w( src/core/lib/support/ref_counted_ptr.h ) + s.files += %w( src/core/lib/support/vector.h ) s.files += %w( src/core/lib/surface/alarm_internal.h ) s.files += %w( src/core/lib/surface/api_trace.h ) s.files += %w( src/core/lib/surface/call.h ) diff --git a/include/grpc/impl/codegen/exec_ctx_fwd.h b/include/grpc/impl/codegen/exec_ctx_fwd.h deleted file mode 100644 index 005ff14e7e..0000000000 --- a/include/grpc/impl/codegen/exec_ctx_fwd.h +++ /dev/null @@ -1,26 +0,0 @@ -/* - * - * Copyright 2016 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 GRPC_IMPL_CODEGEN_EXEC_CTX_FWD_H -#define GRPC_IMPL_CODEGEN_EXEC_CTX_FWD_H - -/* forward declaration for exec_ctx.h */ -struct grpc_exec_ctx; -typedef struct grpc_exec_ctx grpc_exec_ctx; - -#endif /* GRPC_IMPL_CODEGEN_EXEC_CTX_FWD_H */ diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index fcbc8ac5a1..d481a70ab9 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -22,7 +22,6 @@ #include <grpc/impl/codegen/port_platform.h> #include <grpc/impl/codegen/compression_types.h> -#include <grpc/impl/codegen/exec_ctx_fwd.h> #include <grpc/impl/codegen/gpr_types.h> #include <grpc/impl/codegen/slice.h> #include <grpc/impl/codegen/status.h> diff --git a/include/grpc/impl/codegen/slice.h b/include/grpc/impl/codegen/slice.h index ad026b685e..a3cd1f1bbe 100644 --- a/include/grpc/impl/codegen/slice.h +++ b/include/grpc/impl/codegen/slice.h @@ -23,7 +23,6 @@ #include <stddef.h> -#include <grpc/impl/codegen/exec_ctx_fwd.h> #include <grpc/impl/codegen/gpr_slice.h> typedef struct grpc_slice grpc_slice; diff --git a/include/grpc/module.modulemap b/include/grpc/module.modulemap index 67136cba8a..da95515d8e 100644 --- a/include/grpc/module.modulemap +++ b/include/grpc/module.modulemap @@ -30,7 +30,6 @@ framework module grpc { header "impl/codegen/byte_buffer_reader.h" header "impl/codegen/compression_types.h" header "impl/codegen/connectivity_state.h" - header "impl/codegen/exec_ctx_fwd.h" header "impl/codegen/grpc_types.h" header "impl/codegen/propagation_bits.h" header "impl/codegen/slice.h" diff --git a/package.xml b/package.xml index b4d8c88693..ff3d0797ab 100644 --- a/package.xml +++ b/package.xml @@ -161,7 +161,6 @@ <file baseinstalldir="/" name="include/grpc/impl/codegen/byte_buffer_reader.h" role="src" /> <file baseinstalldir="/" name="include/grpc/impl/codegen/compression_types.h" role="src" /> <file baseinstalldir="/" name="include/grpc/impl/codegen/connectivity_state.h" role="src" /> - <file baseinstalldir="/" name="include/grpc/impl/codegen/exec_ctx_fwd.h" role="src" /> <file baseinstalldir="/" name="include/grpc/impl/codegen/grpc_types.h" role="src" /> <file baseinstalldir="/" name="include/grpc/impl/codegen/propagation_bits.h" role="src" /> <file baseinstalldir="/" name="include/grpc/impl/codegen/slice.h" role="src" /> @@ -361,6 +360,7 @@ <file baseinstalldir="/" name="src/core/lib/support/debug_location.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/support/ref_counted.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/support/ref_counted_ptr.h" role="src" /> + <file baseinstalldir="/" name="src/core/lib/support/vector.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/surface/alarm_internal.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/surface/api_trace.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/surface/call.h" role="src" /> diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index 3f3334d44a..e99022a91b 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -553,7 +553,6 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { } grpc_pollset_set_del_pollset_set(chand->lb_policy->interested_parties, chand->interested_parties); - grpc_lb_policy_shutdown_locked(chand->lb_policy, new_lb_policy); GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel"); } chand->lb_policy = new_lb_policy; @@ -659,7 +658,6 @@ static void start_transport_op_locked(void* arg, grpc_error* error_ignored) { if (chand->lb_policy != nullptr) { grpc_pollset_set_del_pollset_set(chand->lb_policy->interested_parties, chand->interested_parties); - grpc_lb_policy_shutdown_locked(chand->lb_policy, nullptr); GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel"); chand->lb_policy = nullptr; } @@ -794,7 +792,6 @@ static void cc_destroy_channel_elem(grpc_channel_element* elem) { if (chand->lb_policy != nullptr) { grpc_pollset_set_del_pollset_set(chand->lb_policy->interested_parties, chand->interested_parties); - grpc_lb_policy_shutdown_locked(chand->lb_policy, nullptr); GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel"); } gpr_free(chand->info_lb_policy_name); @@ -855,10 +852,12 @@ typedef struct client_channel_call_data { grpc_subchannel_call* subchannel_call; grpc_error* error; - grpc_lb_policy_pick_state pick; + grpc_lb_policy* lb_policy; // Holds ref while LB pick is pending. grpc_closure lb_pick_closure; grpc_closure lb_pick_cancel_closure; + grpc_connected_subchannel* connected_subchannel; + grpc_call_context_element subchannel_call_context[GRPC_CONTEXT_COUNT]; grpc_polling_entity* pollent; grpc_transport_stream_op_batch* waiting_for_pick_batches[MAX_WAITING_BATCHES]; @@ -867,6 +866,8 @@ typedef struct client_channel_call_data { grpc_transport_stream_op_batch* initial_metadata_batch; + grpc_linked_mdelem lb_token_mdelem; + grpc_closure on_complete; grpc_closure* original_on_complete; } call_data; @@ -1004,16 +1005,16 @@ static void create_subchannel_call_locked(grpc_call_element* elem, channel_data* chand = (channel_data*)elem->channel_data; call_data* calld = (call_data*)elem->call_data; const grpc_connected_subchannel_call_args call_args = { - calld->pollent, // pollent - calld->path, // path - calld->call_start_time, // start_time - calld->deadline, // deadline - calld->arena, // arena - calld->pick.subchannel_call_context, // context - calld->call_combiner // call_combiner + calld->pollent, // pollent + calld->path, // path + calld->call_start_time, // start_time + calld->deadline, // deadline + calld->arena, // arena + calld->subchannel_call_context, // context + calld->call_combiner // call_combiner }; grpc_error* new_error = grpc_connected_subchannel_create_call( - calld->pick.connected_subchannel, &call_args, &calld->subchannel_call); + calld->connected_subchannel, &call_args, &calld->subchannel_call); if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: create subchannel_call=%p: error=%s", chand, calld, calld->subchannel_call, grpc_error_string(new_error)); @@ -1031,7 +1032,7 @@ static void create_subchannel_call_locked(grpc_call_element* elem, static void pick_done_locked(grpc_call_element* elem, grpc_error* error) { call_data* calld = (call_data*)elem->call_data; channel_data* chand = (channel_data*)elem->channel_data; - if (calld->pick.connected_subchannel == nullptr) { + if (calld->connected_subchannel == nullptr) { // Failed to create subchannel. GRPC_ERROR_UNREF(calld->error); calld->error = error == GRPC_ERROR_NONE @@ -1070,16 +1071,13 @@ static void pick_callback_cancel_locked(void* arg, grpc_error* error) { grpc_call_element* elem = (grpc_call_element*)arg; channel_data* chand = (channel_data*)elem->channel_data; call_data* calld = (call_data*)elem->call_data; - // Note: chand->lb_policy may have changed since we started our pick, - // in which case we will be cancelling the pick on a policy other than - // the one we started it on. However, this will just be a no-op. - if (error != GRPC_ERROR_NONE && chand->lb_policy != nullptr) { + if (calld->lb_policy != nullptr) { if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: cancelling pick from LB policy %p", - chand, calld, chand->lb_policy); + chand, calld, calld->lb_policy); } - grpc_lb_policy_cancel_pick_locked(chand->lb_policy, &calld->pick, - GRPC_ERROR_REF(error)); + grpc_lb_policy_cancel_pick_locked( + calld->lb_policy, &calld->connected_subchannel, GRPC_ERROR_REF(error)); } GRPC_CALL_STACK_UNREF(calld->owning_call, "pick_callback_cancel"); } @@ -1094,6 +1092,9 @@ static void pick_callback_done_locked(void* arg, grpc_error* error) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed asynchronously", chand, calld); } + GPR_ASSERT(calld->lb_policy != nullptr); + GRPC_LB_POLICY_UNREF(calld->lb_policy, "pick_subchannel"); + calld->lb_policy = nullptr; async_pick_done_locked(elem, GRPC_ERROR_REF(error)); } @@ -1127,21 +1128,26 @@ static bool pick_callback_start_locked(grpc_call_element* elem) { initial_metadata_flags &= ~GRPC_INITIAL_METADATA_WAIT_FOR_READY; } } - calld->pick.initial_metadata = + const grpc_lb_policy_pick_args inputs = { calld->initial_metadata_batch->payload->send_initial_metadata - .send_initial_metadata; - calld->pick.initial_metadata_flags = initial_metadata_flags; + .send_initial_metadata, + initial_metadata_flags, &calld->lb_token_mdelem}; + // Keep a ref to the LB policy in calld while the pick is pending. + GRPC_LB_POLICY_REF(chand->lb_policy, "pick_subchannel"); + calld->lb_policy = chand->lb_policy; GRPC_CLOSURE_INIT(&calld->lb_pick_closure, pick_callback_done_locked, elem, grpc_combiner_scheduler(chand->combiner)); - calld->pick.on_complete = &calld->lb_pick_closure; - const bool pick_done = - grpc_lb_policy_pick_locked(chand->lb_policy, &calld->pick); + const bool pick_done = grpc_lb_policy_pick_locked( + chand->lb_policy, &inputs, &calld->connected_subchannel, + calld->subchannel_call_context, nullptr, &calld->lb_pick_closure); if (pick_done) { /* synchronous grpc_lb_policy_pick call. Unref the LB policy. */ if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed synchronously", chand, calld); } + GRPC_LB_POLICY_UNREF(calld->lb_policy, "pick_subchannel"); + calld->lb_policy = nullptr; } else { GRPC_CALL_STACK_REF(calld->owning_call, "pick_callback_cancel"); grpc_call_combiner_set_notify_on_cancel( @@ -1283,7 +1289,7 @@ static void start_pick_locked(void* arg, grpc_error* ignored) { grpc_call_element* elem = (grpc_call_element*)arg; call_data* calld = (call_data*)elem->call_data; channel_data* chand = (channel_data*)elem->channel_data; - GPR_ASSERT(calld->pick.connected_subchannel == nullptr); + GPR_ASSERT(calld->connected_subchannel == nullptr); if (chand->lb_policy != nullptr) { // We already have an LB policy, so ask it for a pick. if (pick_callback_start_locked(elem)) { @@ -1461,14 +1467,15 @@ static void cc_destroy_call_elem(grpc_call_element* elem, GRPC_SUBCHANNEL_CALL_UNREF(calld->subchannel_call, "client_channel_destroy_call"); } + GPR_ASSERT(calld->lb_policy == nullptr); GPR_ASSERT(calld->waiting_for_pick_batches_count == 0); - if (calld->pick.connected_subchannel != nullptr) { - GRPC_CONNECTED_SUBCHANNEL_UNREF(calld->pick.connected_subchannel, "picked"); + if (calld->connected_subchannel != nullptr) { + GRPC_CONNECTED_SUBCHANNEL_UNREF(calld->connected_subchannel, "picked"); } for (size_t i = 0; i < GRPC_CONTEXT_COUNT; ++i) { - if (calld->pick.subchannel_call_context[i].value != nullptr) { - calld->pick.subchannel_call_context[i].destroy( - calld->pick.subchannel_call_context[i].value); + if (calld->subchannel_call_context[i].value != nullptr) { + calld->subchannel_call_context[i].destroy( + calld->subchannel_call_context[i].value); } } GRPC_CLOSURE_SCHED(then_schedule_closure, GRPC_ERROR_NONE); diff --git a/src/core/ext/filters/client_channel/lb_policy.cc b/src/core/ext/filters/client_channel/lb_policy.cc index cc4fe7ec62..7a5a8dec34 100644 --- a/src/core/ext/filters/client_channel/lb_policy.cc +++ b/src/core/ext/filters/client_channel/lb_policy.cc @@ -19,6 +19,8 @@ #include "src/core/ext/filters/client_channel/lb_policy.h" #include "src/core/lib/iomgr/combiner.h" +#define WEAK_REF_BITS 16 + grpc_core::DebugOnlyTraceFlag grpc_trace_lb_policy_refcount( false, "lb_policy_refcount"); @@ -26,60 +28,91 @@ void grpc_lb_policy_init(grpc_lb_policy* policy, const grpc_lb_policy_vtable* vtable, grpc_combiner* combiner) { policy->vtable = vtable; - gpr_ref_init(&policy->refs, 1); + gpr_atm_no_barrier_store(&policy->ref_pair, 1 << WEAK_REF_BITS); policy->interested_parties = grpc_pollset_set_create(); policy->combiner = GRPC_COMBINER_REF(combiner, "lb_policy"); } #ifndef NDEBUG -void grpc_lb_policy_ref(grpc_lb_policy* lb_policy, const char* file, int line, - const char* reason) { - if (grpc_trace_lb_policy_refcount.enabled()) { - gpr_atm old_refs = gpr_atm_no_barrier_load(&lb_policy->refs.count); - gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "LB_POLICY:%p ref %" PRIdPTR " -> %" PRIdPTR " %s", lb_policy, - old_refs, old_refs + 1, reason); - } +#define REF_FUNC_EXTRA_ARGS , const char *file, int line, const char *reason +#define REF_MUTATE_EXTRA_ARGS REF_FUNC_EXTRA_ARGS, const char* purpose +#define REF_FUNC_PASS_ARGS(new_reason) , file, line, new_reason +#define REF_MUTATE_PASS_ARGS(purpose) , file, line, reason, purpose #else -void grpc_lb_policy_ref(grpc_lb_policy* lb_policy) { +#define REF_FUNC_EXTRA_ARGS +#define REF_MUTATE_EXTRA_ARGS +#define REF_FUNC_PASS_ARGS(new_reason) +#define REF_MUTATE_PASS_ARGS(x) #endif - gpr_ref(&lb_policy->refs); -} +static gpr_atm ref_mutate(grpc_lb_policy* c, gpr_atm delta, + int barrier REF_MUTATE_EXTRA_ARGS) { + gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta) + : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta); #ifndef NDEBUG -void grpc_lb_policy_unref(grpc_lb_policy* lb_policy, const char* file, int line, - const char* reason) { if (grpc_trace_lb_policy_refcount.enabled()) { - gpr_atm old_refs = gpr_atm_no_barrier_load(&lb_policy->refs.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "LB_POLICY:%p unref %" PRIdPTR " -> %" PRIdPTR " %s", lb_policy, - old_refs, old_refs - 1, reason); + "LB_POLICY: %p %12s 0x%" PRIxPTR " -> 0x%" PRIxPTR " [%s]", c, + purpose, old_val, old_val + delta, reason); } -#else -void grpc_lb_policy_unref(grpc_lb_policy* lb_policy) { #endif - if (gpr_unref(&lb_policy->refs)) { - grpc_pollset_set_destroy(lb_policy->interested_parties); - grpc_combiner* combiner = lb_policy->combiner; - lb_policy->vtable->destroy(lb_policy); - GRPC_COMBINER_UNREF(combiner, "lb_policy"); + return old_val; +} + +void grpc_lb_policy_ref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) { + ref_mutate(policy, 1 << WEAK_REF_BITS, 0 REF_MUTATE_PASS_ARGS("STRONG_REF")); +} + +static void shutdown_locked(void* arg, grpc_error* error) { + grpc_lb_policy* policy = (grpc_lb_policy*)arg; + policy->vtable->shutdown_locked(policy); + GRPC_LB_POLICY_WEAK_UNREF(policy, "strong-unref"); +} + +void grpc_lb_policy_unref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) { + gpr_atm old_val = + ref_mutate(policy, (gpr_atm)1 - (gpr_atm)(1 << WEAK_REF_BITS), + 1 REF_MUTATE_PASS_ARGS("STRONG_UNREF")); + gpr_atm mask = ~(gpr_atm)((1 << WEAK_REF_BITS) - 1); + gpr_atm check = 1 << WEAK_REF_BITS; + if ((old_val & mask) == check) { + GRPC_CLOSURE_SCHED( + GRPC_CLOSURE_CREATE(shutdown_locked, policy, + grpc_combiner_scheduler(policy->combiner)), + GRPC_ERROR_NONE); + } else { + grpc_lb_policy_weak_unref(policy REF_FUNC_PASS_ARGS("strong-unref")); } } -void grpc_lb_policy_shutdown_locked(grpc_lb_policy* policy, - grpc_lb_policy* new_policy) { - policy->vtable->shutdown_locked(policy, new_policy); +void grpc_lb_policy_weak_ref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) { + ref_mutate(policy, 1, 0 REF_MUTATE_PASS_ARGS("WEAK_REF")); +} + +void grpc_lb_policy_weak_unref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) { + gpr_atm old_val = + ref_mutate(policy, -(gpr_atm)1, 1 REF_MUTATE_PASS_ARGS("WEAK_UNREF")); + if (old_val == 1) { + grpc_pollset_set_destroy(policy->interested_parties); + grpc_combiner* combiner = policy->combiner; + policy->vtable->destroy(policy); + GRPC_COMBINER_UNREF(combiner, "lb_policy"); + } } int grpc_lb_policy_pick_locked(grpc_lb_policy* policy, - grpc_lb_policy_pick_state* pick) { - return policy->vtable->pick_locked(policy, pick); + const grpc_lb_policy_pick_args* pick_args, + grpc_connected_subchannel** target, + grpc_call_context_element* context, + void** user_data, grpc_closure* on_complete) { + return policy->vtable->pick_locked(policy, pick_args, target, context, + user_data, on_complete); } void grpc_lb_policy_cancel_pick_locked(grpc_lb_policy* policy, - grpc_lb_policy_pick_state* pick, + grpc_connected_subchannel** target, grpc_error* error) { - policy->vtable->cancel_pick_locked(policy, pick, error); + policy->vtable->cancel_pick_locked(policy, target, error); } void grpc_lb_policy_cancel_picks_locked(grpc_lb_policy* policy, diff --git a/src/core/ext/filters/client_channel/lb_policy.h b/src/core/ext/filters/client_channel/lb_policy.h index 1176a05b78..3572c97ed1 100644 --- a/src/core/ext/filters/client_channel/lb_policy.h +++ b/src/core/ext/filters/client_channel/lb_policy.h @@ -33,7 +33,7 @@ extern grpc_core::DebugOnlyTraceFlag grpc_trace_lb_policy_refcount; struct grpc_lb_policy { const grpc_lb_policy_vtable* vtable; - gpr_refcount refs; + gpr_atm ref_pair; /* owned pointer to interested parties in load balancing decisions */ grpc_pollset_set* interested_parties; /* combiner under which lb_policy actions take place */ @@ -42,42 +42,32 @@ struct grpc_lb_policy { grpc_closure* request_reresolution; }; -/// State used for an LB pick. -typedef struct grpc_lb_policy_pick_state { - /// Initial metadata associated with the picking call. +/** Extra arguments for an LB pick */ +typedef struct grpc_lb_policy_pick_args { + /** Initial metadata associated with the picking call. */ grpc_metadata_batch* initial_metadata; - /// Bitmask used for selective cancelling. See \a - /// grpc_lb_policy_cancel_picks() and \a GRPC_INITIAL_METADATA_* in - /// grpc_types.h. + /** Bitmask used for selective cancelling. See \a + * grpc_lb_policy_cancel_picks() and \a GRPC_INITIAL_METADATA_* in + * grpc_types.h */ uint32_t initial_metadata_flags; - /// Storage for LB token in \a initial_metadata, or NULL if not used. - grpc_linked_mdelem lb_token_mdelem_storage; - /// Closure to run when pick is complete, if not completed synchronously. - grpc_closure* on_complete; - /// Will be set to the selected subchannel, or NULL on failure or when - /// the LB policy decides to drop the call. - grpc_connected_subchannel* connected_subchannel; - /// Will be populated with context to pass to the subchannel call, if needed. - grpc_call_context_element subchannel_call_context[GRPC_CONTEXT_COUNT]; - /// Upon success, \a *user_data will be set to whatever opaque information - /// may need to be propagated from the LB policy, or NULL if not needed. - void** user_data; - /// Next pointer. For internal use by LB policy. - struct grpc_lb_policy_pick_state* next; -} grpc_lb_policy_pick_state; + /** Storage for LB token in \a initial_metadata, or NULL if not used */ + grpc_linked_mdelem* lb_token_mdelem_storage; +} grpc_lb_policy_pick_args; struct grpc_lb_policy_vtable { void (*destroy)(grpc_lb_policy* policy); - - /// \see grpc_lb_policy_shutdown_locked(). - void (*shutdown_locked)(grpc_lb_policy* policy, grpc_lb_policy* new_policy); + void (*shutdown_locked)(grpc_lb_policy* policy); /** \see grpc_lb_policy_pick */ - int (*pick_locked)(grpc_lb_policy* policy, grpc_lb_policy_pick_state* pick); + int (*pick_locked)(grpc_lb_policy* policy, + const grpc_lb_policy_pick_args* pick_args, + grpc_connected_subchannel** target, + grpc_call_context_element* context, void** user_data, + grpc_closure* on_complete); /** \see grpc_lb_policy_cancel_pick */ void (*cancel_pick_locked)(grpc_lb_policy* policy, - grpc_lb_policy_pick_state* pick, + grpc_connected_subchannel** target, grpc_error* error); /** \see grpc_lb_policy_cancel_picks */ @@ -113,19 +103,37 @@ struct grpc_lb_policy_vtable { }; #ifndef NDEBUG + +/* Strong references: the policy will shutdown when they reach zero */ #define GRPC_LB_POLICY_REF(p, r) \ grpc_lb_policy_ref((p), __FILE__, __LINE__, (r)) #define GRPC_LB_POLICY_UNREF(p, r) \ grpc_lb_policy_unref((p), __FILE__, __LINE__, (r)) + +/* Weak references: they don't prevent the shutdown of the LB policy. When no + * strong references are left but there are still weak ones, shutdown is called. + * Once the weak reference also reaches zero, the LB policy is destroyed. */ +#define GRPC_LB_POLICY_WEAK_REF(p, r) \ + grpc_lb_policy_weak_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_LB_POLICY_WEAK_UNREF(p, r) \ + grpc_lb_policy_weak_unref((p), __FILE__, __LINE__, (r)) void grpc_lb_policy_ref(grpc_lb_policy* policy, const char* file, int line, const char* reason); void grpc_lb_policy_unref(grpc_lb_policy* policy, const char* file, int line, const char* reason); -#else // !NDEBUG +void grpc_lb_policy_weak_ref(grpc_lb_policy* policy, const char* file, int line, + const char* reason); +void grpc_lb_policy_weak_unref(grpc_lb_policy* policy, const char* file, + int line, const char* reason); +#else #define GRPC_LB_POLICY_REF(p, r) grpc_lb_policy_ref((p)) #define GRPC_LB_POLICY_UNREF(p, r) grpc_lb_policy_unref((p)) +#define GRPC_LB_POLICY_WEAK_REF(p, r) grpc_lb_policy_weak_ref((p)) +#define GRPC_LB_POLICY_WEAK_UNREF(p, r) grpc_lb_policy_weak_unref((p)) void grpc_lb_policy_ref(grpc_lb_policy* policy); void grpc_lb_policy_unref(grpc_lb_policy* policy); +void grpc_lb_policy_weak_ref(grpc_lb_policy* policy); +void grpc_lb_policy_weak_unref(grpc_lb_policy* policy); #endif /** called by concrete implementations to initialize the base struct */ @@ -133,24 +141,28 @@ void grpc_lb_policy_init(grpc_lb_policy* policy, const grpc_lb_policy_vtable* vtable, grpc_combiner* combiner); -/// Shuts down \a policy. -/// If \a new_policy is non-null, any pending picks will be restarted -/// on that policy; otherwise, they will be failed. -void grpc_lb_policy_shutdown_locked(grpc_lb_policy* policy, - grpc_lb_policy* new_policy); +/** Finds an appropriate subchannel for a call, based on \a pick_args. + + \a target will be set to the selected subchannel, or NULL on failure + or when the LB policy decides to drop the call. -/** Finds an appropriate subchannel for a call, based on data in \a pick. - \a pick must remain alive until the pick is complete. + Upon success, \a user_data will be set to whatever opaque information + may need to be propagated from the LB policy, or NULL if not needed. + \a context will be populated with context to pass to the subchannel + call, if needed. If the pick succeeds and a result is known immediately, a non-zero - value will be returned. Otherwise, \a pick->on_complete will be invoked + value will be returned. Otherwise, \a on_complete will be invoked once the pick is complete with its error argument set to indicate success or failure. Any IO should be done under the \a interested_parties \a grpc_pollset_set in the \a grpc_lb_policy struct. */ int grpc_lb_policy_pick_locked(grpc_lb_policy* policy, - grpc_lb_policy_pick_state* pick); + const grpc_lb_policy_pick_args* pick_args, + grpc_connected_subchannel** target, + grpc_call_context_element* context, + void** user_data, grpc_closure* on_complete); /** Perform a connected subchannel ping (see \a grpc_connected_subchannel_ping) against one of the connected subchannels managed by \a policy. */ @@ -158,11 +170,11 @@ void grpc_lb_policy_ping_one_locked(grpc_lb_policy* policy, grpc_closure* on_initiate, grpc_closure* on_ack); -/** Cancel picks for \a pick. +/** Cancel picks for \a target. The \a on_complete callback of the pending picks will be invoked with \a *target set to NULL. */ void grpc_lb_policy_cancel_pick_locked(grpc_lb_policy* policy, - grpc_lb_policy_pick_state* pick, + grpc_connected_subchannel** target, grpc_error* error); /** Cancel all pending picks for which their \a initial_metadata_flags (as given 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 5849ac9d2d..1317cdcf75 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 @@ -54,7 +54,7 @@ * operations in progress over the old RR instance. This is done by * decreasing the reference count on the old policy. The moment no more * references are held on the old RR policy, it'll be destroyed and \a - * on_rr_connectivity_changed notified with a \a GRPC_CHANNEL_SHUTDOWN + * glb_rr_connectivity_changed notified with a \a GRPC_CHANNEL_SHUTDOWN * state. At this point we can transition to a new RR instance safely, which * is done once again via \a rr_handover_locked(). * @@ -128,48 +128,187 @@ grpc_core::TraceFlag grpc_lb_glb_trace(false, "glb"); -struct glb_lb_policy; +/* add lb_token of selected subchannel (address) to the call's initial + * metadata */ +static grpc_error* initial_metadata_add_lb_token( + grpc_metadata_batch* initial_metadata, + grpc_linked_mdelem* lb_token_mdelem_storage, grpc_mdelem lb_token) { + GPR_ASSERT(lb_token_mdelem_storage != nullptr); + GPR_ASSERT(!GRPC_MDISNULL(lb_token)); + return grpc_metadata_batch_add_tail(initial_metadata, lb_token_mdelem_storage, + lb_token); +} -namespace { +static void destroy_client_stats(void* arg) { + grpc_grpclb_client_stats_unref((grpc_grpclb_client_stats*)arg); +} -/// Linked list of pending pick requests. It stores all information needed to -/// eventually call (Round Robin's) pick() on them. They mainly stay pending -/// waiting for the RR policy to be created. -/// -/// Note that when a pick is sent to the RR policy, we inject our own -/// on_complete callback, so that we can intercept the result before -/// invoking the original on_complete callback. This allows us to set the -/// LB token metadata and add client_stats to the call context. -/// See \a pending_pick_complete() for details. -struct pending_pick { - // Our on_complete closure and the original one. - grpc_closure on_complete; - grpc_closure* original_on_complete; - // The original pick. - grpc_lb_policy_pick_state* pick; - // Stats for client-side load reporting. Note that this holds a - // reference, which must be either passed on via context or unreffed. +typedef struct wrapped_rr_closure_arg { + /* the closure instance using this struct as argument */ + grpc_closure wrapper_closure; + + /* the original closure. Usually a on_complete/notify cb for pick() and ping() + * calls against the internal RR instance, respectively. */ + grpc_closure* wrapped_closure; + + /* the pick's initial metadata, kept in order to append the LB token for the + * pick */ + grpc_metadata_batch* initial_metadata; + + /* the picked target, used to determine which LB token to add to the pick's + * initial metadata */ + grpc_connected_subchannel** target; + + /* the context to be populated for the subchannel call */ + grpc_call_context_element* context; + + /* Stats for client-side load reporting. Note that this holds a + * reference, which must be either passed on via context or unreffed. */ grpc_grpclb_client_stats* client_stats; - // The LB token associated with the pick. This is set via user_data in - // the pick. + + /* the LB token associated with the pick */ grpc_mdelem lb_token; - // The grpclb instance that created the wrapping. This instance is not owned, - // reference counts are untouched. It's used only for logging purposes. - glb_lb_policy* glb_policy; - // Next pending pick. + + /* storage for the lb token initial metadata mdelem */ + grpc_linked_mdelem* lb_token_mdelem_storage; + + /* The RR instance related to the closure */ + grpc_lb_policy* rr_policy; + + /* The grpclb instance that created the wrapping. This instance is not owned, + * reference counts are untouched. It's used only for logging purposes. */ + grpc_lb_policy* glb_policy; + + /* heap memory to be freed upon closure execution. */ + void* free_when_done; +} wrapped_rr_closure_arg; + +/* The \a on_complete closure passed as part of the pick requires keeping a + * reference to its associated round robin instance. We wrap this closure in + * order to unref the round robin instance upon its invocation */ +static void wrapped_rr_closure(void* arg, grpc_error* error) { + wrapped_rr_closure_arg* wc_arg = (wrapped_rr_closure_arg*)arg; + + GPR_ASSERT(wc_arg->wrapped_closure != nullptr); + GRPC_CLOSURE_SCHED(wc_arg->wrapped_closure, GRPC_ERROR_REF(error)); + + if (wc_arg->rr_policy != nullptr) { + /* if *target is nullptr, no pick has been made by the RR policy (eg, all + * addresses failed to connect). There won't be any user_data/token + * available */ + if (*wc_arg->target != nullptr) { + if (!GRPC_MDISNULL(wc_arg->lb_token)) { + initial_metadata_add_lb_token(wc_arg->initial_metadata, + wc_arg->lb_token_mdelem_storage, + GRPC_MDELEM_REF(wc_arg->lb_token)); + } else { + gpr_log( + GPR_ERROR, + "[grpclb %p] No LB token for connected subchannel pick %p (from RR " + "instance %p).", + wc_arg->glb_policy, *wc_arg->target, wc_arg->rr_policy); + abort(); + } + // Pass on client stats via context. Passes ownership of the reference. + GPR_ASSERT(wc_arg->client_stats != nullptr); + wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].value = wc_arg->client_stats; + wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].destroy = destroy_client_stats; + } else { + grpc_grpclb_client_stats_unref(wc_arg->client_stats); + } + if (grpc_lb_glb_trace.enabled()) { + gpr_log(GPR_INFO, "[grpclb %p] Unreffing RR %p", wc_arg->glb_policy, + wc_arg->rr_policy); + } + GRPC_LB_POLICY_UNREF(wc_arg->rr_policy, "wrapped_rr_closure"); + } + GPR_ASSERT(wc_arg->free_when_done != nullptr); + gpr_free(wc_arg->free_when_done); +} + +namespace { +/* Linked list of pending pick requests. It stores all information needed to + * eventually call (Round Robin's) pick() on them. They mainly stay pending + * waiting for the RR policy to be created/updated. + * + * One particularity is the wrapping of the user-provided \a on_complete closure + * (in \a wrapped_on_complete and \a wrapped_on_complete_arg). This is needed in + * order to correctly unref the RR policy instance upon completion of the pick. + * See \a wrapped_rr_closure for details. */ +struct pending_pick { struct pending_pick* next; + + /* original pick()'s arguments */ + grpc_lb_policy_pick_args pick_args; + + /* output argument where to store the pick()ed connected subchannel, or + * nullptr upon error. */ + grpc_connected_subchannel** target; + + /* args for wrapped_on_complete */ + wrapped_rr_closure_arg wrapped_on_complete_arg; }; +} // namespace + +static void add_pending_pick(pending_pick** root, + const grpc_lb_policy_pick_args* pick_args, + grpc_connected_subchannel** target, + grpc_call_context_element* context, + grpc_closure* on_complete) { + pending_pick* pp = (pending_pick*)gpr_zalloc(sizeof(*pp)); + pp->next = *root; + pp->pick_args = *pick_args; + pp->target = target; + pp->wrapped_on_complete_arg.wrapped_closure = on_complete; + pp->wrapped_on_complete_arg.target = target; + pp->wrapped_on_complete_arg.context = context; + pp->wrapped_on_complete_arg.initial_metadata = pick_args->initial_metadata; + pp->wrapped_on_complete_arg.lb_token_mdelem_storage = + pick_args->lb_token_mdelem_storage; + pp->wrapped_on_complete_arg.free_when_done = pp; + GRPC_CLOSURE_INIT(&pp->wrapped_on_complete_arg.wrapper_closure, + wrapped_rr_closure, &pp->wrapped_on_complete_arg, + grpc_schedule_on_exec_ctx); + *root = pp; +} -/// A linked list of pending pings waiting for the RR policy to be created. -struct pending_ping { - grpc_closure* on_initiate; - grpc_closure* on_ack; +/* Same as the \a pending_pick struct but for ping operations */ +typedef struct pending_ping { struct pending_ping* next; -}; -} // namespace + /* args for sending the ping */ + wrapped_rr_closure_arg* on_initiate; + wrapped_rr_closure_arg* on_ack; +} pending_ping; + +static void add_pending_ping(pending_ping** root, grpc_closure* on_initiate, + grpc_closure* on_ack) { + pending_ping* pping = (pending_ping*)gpr_zalloc(sizeof(*pping)); + if (on_initiate != nullptr) { + pping->on_initiate = + (wrapped_rr_closure_arg*)gpr_zalloc(sizeof(*pping->on_initiate)); + pping->on_initiate->wrapped_closure = on_initiate; + pping->on_initiate->free_when_done = pping->on_initiate; + GRPC_CLOSURE_INIT(&pping->on_initiate->wrapper_closure, wrapped_rr_closure, + &pping->on_initiate, grpc_schedule_on_exec_ctx); + } + if (on_ack != nullptr) { + pping->on_ack = (wrapped_rr_closure_arg*)gpr_zalloc(sizeof(*pping->on_ack)); + pping->on_ack->wrapped_closure = on_ack; + pping->on_ack->free_when_done = pping->on_ack; + GRPC_CLOSURE_INIT(&pping->on_ack->wrapper_closure, wrapped_rr_closure, + &pping->on_ack, grpc_schedule_on_exec_ctx); + } + pping->next = *root; + *root = pping; +} -struct glb_lb_policy { +/* + * glb_lb_policy + */ +typedef struct rr_connectivity_data rr_connectivity_data; + +typedef struct glb_lb_policy { /** base policy: must be first */ grpc_lb_policy base; @@ -194,9 +333,6 @@ struct glb_lb_policy { /** the RR policy to use of the backend servers returned by the LB server */ grpc_lb_policy* rr_policy; - grpc_closure on_rr_connectivity_changed; - grpc_connectivity_state rr_connectivity_state; - bool started_picking; /** our connectivity state tracker */ @@ -301,84 +437,14 @@ struct glb_lb_policy { grpc_closure client_load_report_closure; /* Client load report message payload. */ grpc_byte_buffer* client_load_report_payload; -}; - -/* add lb_token of selected subchannel (address) to the call's initial - * metadata */ -static grpc_error* initial_metadata_add_lb_token( - grpc_metadata_batch* initial_metadata, - grpc_linked_mdelem* lb_token_mdelem_storage, grpc_mdelem lb_token) { - GPR_ASSERT(lb_token_mdelem_storage != nullptr); - GPR_ASSERT(!GRPC_MDISNULL(lb_token)); - return grpc_metadata_batch_add_tail(initial_metadata, lb_token_mdelem_storage, - lb_token); -} - -static void destroy_client_stats(void* arg) { - grpc_grpclb_client_stats_unref((grpc_grpclb_client_stats*)arg); -} +} glb_lb_policy; -static void pending_pick_set_metadata_and_context(pending_pick* pp) { - /* if connected_subchannel is nullptr, no pick has been made by the RR - * policy (e.g., all addresses failed to connect). There won't be any - * user_data/token available */ - if (pp->pick->connected_subchannel != nullptr) { - if (!GRPC_MDISNULL(pp->lb_token)) { - initial_metadata_add_lb_token(pp->pick->initial_metadata, - &pp->pick->lb_token_mdelem_storage, - GRPC_MDELEM_REF(pp->lb_token)); - } else { - gpr_log(GPR_ERROR, - "[grpclb %p] No LB token for connected subchannel pick %p", - pp->glb_policy, pp->pick); - abort(); - } - // Pass on client stats via context. Passes ownership of the reference. - GPR_ASSERT(pp->client_stats != nullptr); - pp->pick->subchannel_call_context[GRPC_GRPCLB_CLIENT_STATS].value = - pp->client_stats; - pp->pick->subchannel_call_context[GRPC_GRPCLB_CLIENT_STATS].destroy = - destroy_client_stats; - } else { - grpc_grpclb_client_stats_unref(pp->client_stats); - } -} - -/* The \a on_complete closure passed as part of the pick requires keeping a - * reference to its associated round robin instance. We wrap this closure in - * order to unref the round robin instance upon its invocation */ -static void pending_pick_complete(void* arg, grpc_error* error) { - pending_pick* pp = (pending_pick*)arg; - pending_pick_set_metadata_and_context(pp); - GRPC_CLOSURE_SCHED(pp->original_on_complete, GRPC_ERROR_REF(error)); - gpr_free(pp); -} - -static pending_pick* pending_pick_create(glb_lb_policy* glb_policy, - grpc_lb_policy_pick_state* pick) { - pending_pick* pp = (pending_pick*)gpr_zalloc(sizeof(*pp)); - pp->pick = pick; - pp->glb_policy = glb_policy; - GRPC_CLOSURE_INIT(&pp->on_complete, pending_pick_complete, pp, - grpc_schedule_on_exec_ctx); - pp->original_on_complete = pick->on_complete; - pp->pick->on_complete = &pp->on_complete; - return pp; -} - -static void pending_pick_add(pending_pick** root, pending_pick* new_pp) { - new_pp->next = *root; - *root = new_pp; -} - -static void pending_ping_add(pending_ping** root, grpc_closure* on_initiate, - grpc_closure* on_ack) { - pending_ping* pping = (pending_ping*)gpr_zalloc(sizeof(*pping)); - pping->on_initiate = on_initiate; - pping->on_ack = on_ack; - pping->next = *root; - *root = pping; -} +/* Keeps track and reacts to changes in connectivity of the RR instance */ +struct rr_connectivity_data { + grpc_closure on_change; + grpc_connectivity_state state; + glb_lb_policy* glb_policy; +}; static bool is_server_valid(const grpc_grpclb_server* server, size_t idx, bool log) { @@ -491,6 +557,7 @@ static grpc_lb_addresses* process_serverlist_locked( gpr_free(uri); user_data = (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload; } + grpc_lb_addresses_set_address(lb_addresses, addr_idx, &addr.addr, addr.len, false /* is_balancer */, nullptr /* balancer_name */, user_data); @@ -531,6 +598,7 @@ static void update_lb_connectivity_status_locked( grpc_error* rr_state_error) { const grpc_connectivity_state curr_glb_state = grpc_connectivity_state_check(&glb_policy->state_tracker); + /* The new connectivity status is a function of the previous one and the new * input coming from the status of the RR policy. * @@ -560,6 +628,7 @@ static void update_lb_connectivity_status_locked( * * (*) This function mustn't be called during shutting down. */ GPR_ASSERT(curr_glb_state != GRPC_CHANNEL_SHUTDOWN); + switch (rr_state) { case GRPC_CHANNEL_TRANSIENT_FAILURE: case GRPC_CHANNEL_SHUTDOWN: @@ -570,6 +639,7 @@ static void update_lb_connectivity_status_locked( case GRPC_CHANNEL_READY: GPR_ASSERT(rr_state_error == GRPC_ERROR_NONE); } + if (grpc_lb_glb_trace.enabled()) { gpr_log( GPR_INFO, @@ -587,8 +657,10 @@ static void update_lb_connectivity_status_locked( * cleanups this callback would otherwise be responsible for. * If \a force_async is true, then we will manually schedule the * completion callback even if the pick is available immediately. */ -static bool pick_from_internal_rr_locked(glb_lb_policy* glb_policy, - bool force_async, pending_pick* pp) { +static bool pick_from_internal_rr_locked( + glb_lb_policy* glb_policy, const grpc_lb_policy_pick_args* pick_args, + bool force_async, grpc_connected_subchannel** target, + wrapped_rr_closure_arg* wc_arg) { // Check for drops if we are not using fallback backend addresses. if (glb_policy->serverlist != nullptr) { // Look at the index into the serverlist to see if we should drop this call. @@ -598,36 +670,57 @@ static bool pick_from_internal_rr_locked(glb_lb_policy* glb_policy, glb_policy->serverlist_index = 0; // Wrap-around. } if (server->drop) { + // Not using the RR policy, so unref it. + if (grpc_lb_glb_trace.enabled()) { + gpr_log(GPR_INFO, "[grpclb %p] Unreffing RR %p for drop", glb_policy, + wc_arg->rr_policy); + } + GRPC_LB_POLICY_UNREF(wc_arg->rr_policy, "glb_pick_sync"); // Update client load reporting stats to indicate the number of // dropped calls. Note that we have to do this here instead of in // the client_load_reporting filter, because we do not create a // subchannel call (and therefore no client_load_reporting filter) // for dropped calls. - GPR_ASSERT(glb_policy->client_stats != nullptr); + GPR_ASSERT(wc_arg->client_stats != nullptr); grpc_grpclb_client_stats_add_call_dropped_locked( - server->load_balance_token, glb_policy->client_stats); + server->load_balance_token, wc_arg->client_stats); + grpc_grpclb_client_stats_unref(wc_arg->client_stats); if (force_async) { - GRPC_CLOSURE_SCHED(pp->original_on_complete, GRPC_ERROR_NONE); - gpr_free(pp); + GPR_ASSERT(wc_arg->wrapped_closure != nullptr); + GRPC_CLOSURE_SCHED(wc_arg->wrapped_closure, GRPC_ERROR_NONE); + gpr_free(wc_arg->free_when_done); return false; } - gpr_free(pp); + gpr_free(wc_arg->free_when_done); return true; } } - // Set client_stats and user_data. - pp->client_stats = grpc_grpclb_client_stats_ref(glb_policy->client_stats); - GPR_ASSERT(pp->pick->user_data == nullptr); - pp->pick->user_data = (void**)&pp->lb_token; // Pick via the RR policy. - bool pick_done = grpc_lb_policy_pick_locked(glb_policy->rr_policy, pp->pick); + const bool pick_done = grpc_lb_policy_pick_locked( + wc_arg->rr_policy, pick_args, target, wc_arg->context, + (void**)&wc_arg->lb_token, &wc_arg->wrapper_closure); if (pick_done) { - pending_pick_set_metadata_and_context(pp); + /* synchronous grpc_lb_policy_pick call. Unref the RR policy. */ + if (grpc_lb_glb_trace.enabled()) { + gpr_log(GPR_INFO, "[grpclb %p] Unreffing RR %p", glb_policy, + wc_arg->rr_policy); + } + GRPC_LB_POLICY_UNREF(wc_arg->rr_policy, "glb_pick_sync"); + /* add the load reporting initial metadata */ + initial_metadata_add_lb_token(pick_args->initial_metadata, + pick_args->lb_token_mdelem_storage, + GRPC_MDELEM_REF(wc_arg->lb_token)); + // Pass on client stats via context. Passes ownership of the reference. + GPR_ASSERT(wc_arg->client_stats != nullptr); + wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].value = wc_arg->client_stats; + wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].destroy = destroy_client_stats; if (force_async) { - GRPC_CLOSURE_SCHED(pp->original_on_complete, GRPC_ERROR_NONE); - pick_done = false; + GPR_ASSERT(wc_arg->wrapped_closure != nullptr); + GRPC_CLOSURE_SCHED(wc_arg->wrapped_closure, GRPC_ERROR_NONE); + gpr_free(wc_arg->free_when_done); + return false; } - gpr_free(pp); + gpr_free(wc_arg->free_when_done); } /* else, the pending pick will be registered and taken care of by the * pending pick list inside the RR policy (glb_policy->rr_policy). @@ -669,7 +762,7 @@ static void lb_policy_args_destroy(grpc_lb_policy_args* args) { gpr_free(args); } -static void on_rr_connectivity_changed_locked(void* arg, grpc_error* error); +static void glb_rr_connectivity_changed_locked(void* arg, grpc_error* error); static void create_rr_locked(glb_lb_policy* glb_policy, grpc_lb_policy_args* args) { GPR_ASSERT(glb_policy->rr_policy == nullptr); @@ -691,46 +784,72 @@ static void create_rr_locked(glb_lb_policy* glb_policy, glb_policy->base.request_reresolution = nullptr; glb_policy->rr_policy = new_rr_policy; grpc_error* rr_state_error = nullptr; - glb_policy->rr_connectivity_state = grpc_lb_policy_check_connectivity_locked( - glb_policy->rr_policy, &rr_state_error); + const grpc_connectivity_state rr_state = + grpc_lb_policy_check_connectivity_locked(glb_policy->rr_policy, + &rr_state_error); /* Connectivity state is a function of the RR policy updated/created */ - update_lb_connectivity_status_locked( - glb_policy, glb_policy->rr_connectivity_state, rr_state_error); + update_lb_connectivity_status_locked(glb_policy, rr_state, rr_state_error); /* Add the gRPC LB's interested_parties pollset_set to that of the newly * created RR policy. This will make the RR policy progress upon activity on * gRPC LB, which in turn is tied to the application's call */ grpc_pollset_set_add_pollset_set(glb_policy->rr_policy->interested_parties, glb_policy->base.interested_parties); - GRPC_CLOSURE_INIT(&glb_policy->on_rr_connectivity_changed, - on_rr_connectivity_changed_locked, glb_policy, + + /* Allocate the data for the tracking of the new RR policy's connectivity. + * It'll be deallocated in glb_rr_connectivity_changed() */ + rr_connectivity_data* rr_connectivity = + (rr_connectivity_data*)gpr_zalloc(sizeof(rr_connectivity_data)); + GRPC_CLOSURE_INIT(&rr_connectivity->on_change, + glb_rr_connectivity_changed_locked, rr_connectivity, grpc_combiner_scheduler(glb_policy->base.combiner)); + rr_connectivity->glb_policy = glb_policy; + rr_connectivity->state = rr_state; + /* Subscribe to changes to the connectivity of the new RR */ - GRPC_LB_POLICY_REF(&glb_policy->base, "glb_rr_connectivity_cb"); - grpc_lb_policy_notify_on_state_change_locked( - glb_policy->rr_policy, &glb_policy->rr_connectivity_state, - &glb_policy->on_rr_connectivity_changed); + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "glb_rr_connectivity_cb"); + grpc_lb_policy_notify_on_state_change_locked(glb_policy->rr_policy, + &rr_connectivity->state, + &rr_connectivity->on_change); grpc_lb_policy_exit_idle_locked(glb_policy->rr_policy); - // Send pending picks to RR policy. + + /* Update picks and pings in wait */ pending_pick* pp; while ((pp = glb_policy->pending_picks)) { glb_policy->pending_picks = pp->next; + GRPC_LB_POLICY_REF(glb_policy->rr_policy, "rr_handover_pending_pick"); + pp->wrapped_on_complete_arg.rr_policy = glb_policy->rr_policy; + pp->wrapped_on_complete_arg.client_stats = + grpc_grpclb_client_stats_ref(glb_policy->client_stats); if (grpc_lb_glb_trace.enabled()) { gpr_log(GPR_INFO, "[grpclb %p] Pending pick about to (async) PICK from RR %p", glb_policy, glb_policy->rr_policy); } - pick_from_internal_rr_locked(glb_policy, true /* force_async */, pp); + pick_from_internal_rr_locked(glb_policy, &pp->pick_args, + true /* force_async */, pp->target, + &pp->wrapped_on_complete_arg); } - // Send pending pings to RR policy. + pending_ping* pping; while ((pping = glb_policy->pending_pings)) { glb_policy->pending_pings = pping->next; + grpc_closure* on_initiate = nullptr; + grpc_closure* on_ack = nullptr; + if (pping->on_initiate != nullptr) { + GRPC_LB_POLICY_REF(glb_policy->rr_policy, "rr_handover_pending_ping"); + pping->on_initiate->rr_policy = glb_policy->rr_policy; + on_initiate = &pping->on_initiate->wrapper_closure; + } + if (pping->on_ack != nullptr) { + GRPC_LB_POLICY_REF(glb_policy->rr_policy, "rr_handover_pending_ping"); + pping->on_ack->rr_policy = glb_policy->rr_policy; + on_ack = &pping->on_ack->wrapper_closure; + } if (grpc_lb_glb_trace.enabled()) { gpr_log(GPR_INFO, "[grpclb %p] Pending ping about to PING from RR %p", glb_policy, glb_policy->rr_policy); } - grpc_lb_policy_ping_one_locked(glb_policy->rr_policy, pping->on_initiate, - pping->on_ack); + grpc_lb_policy_ping_one_locked(glb_policy->rr_policy, on_initiate, on_ack); gpr_free(pping); } } @@ -756,28 +875,31 @@ static void rr_handover_locked(glb_lb_policy* glb_policy) { lb_policy_args_destroy(args); } -static void on_rr_connectivity_changed_locked(void* arg, grpc_error* error) { - glb_lb_policy* glb_policy = (glb_lb_policy*)arg; +static void glb_rr_connectivity_changed_locked(void* arg, grpc_error* error) { + rr_connectivity_data* rr_connectivity = (rr_connectivity_data*)arg; + glb_lb_policy* glb_policy = rr_connectivity->glb_policy; if (glb_policy->shutting_down) { - GRPC_LB_POLICY_UNREF(&glb_policy->base, "glb_rr_connectivity_cb"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "glb_rr_connectivity_cb"); + gpr_free(rr_connectivity); return; } - if (glb_policy->rr_connectivity_state == GRPC_CHANNEL_SHUTDOWN) { + if (rr_connectivity->state == GRPC_CHANNEL_SHUTDOWN) { /* An RR policy that has transitioned into the SHUTDOWN connectivity state * should not be considered for picks or updates: the SHUTDOWN state is a * sink, policies can't transition back from it. .*/ GRPC_LB_POLICY_UNREF(glb_policy->rr_policy, "rr_connectivity_shutdown"); glb_policy->rr_policy = nullptr; - GRPC_LB_POLICY_UNREF(&glb_policy->base, "glb_rr_connectivity_cb"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "glb_rr_connectivity_cb"); + gpr_free(rr_connectivity); return; } /* rr state != SHUTDOWN && !glb_policy->shutting down: biz as usual */ - update_lb_connectivity_status_locked( - glb_policy, glb_policy->rr_connectivity_state, GRPC_ERROR_REF(error)); - /* Resubscribe. Reuse the "glb_rr_connectivity_cb" ref. */ - grpc_lb_policy_notify_on_state_change_locked( - glb_policy->rr_policy, &glb_policy->rr_connectivity_state, - &glb_policy->on_rr_connectivity_changed); + update_lb_connectivity_status_locked(glb_policy, rr_connectivity->state, + GRPC_ERROR_REF(error)); + /* Resubscribe. Reuse the "glb_rr_connectivity_cb" weak ref. */ + grpc_lb_policy_notify_on_state_change_locked(glb_policy->rr_policy, + &rr_connectivity->state, + &rr_connectivity->on_change); } static void destroy_balancer_name(void* balancer_name) { @@ -885,17 +1007,22 @@ static void glb_destroy(grpc_lb_policy* pol) { gpr_free(glb_policy); } -static void glb_shutdown_locked(grpc_lb_policy* pol, - grpc_lb_policy* new_policy) { +static void glb_shutdown_locked(grpc_lb_policy* pol) { glb_lb_policy* glb_policy = (glb_lb_policy*)pol; grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown"); glb_policy->shutting_down = true; + + /* We need a copy of the lb_call pointer because we can't cancell the call + * while holding glb_policy->mu: lb_on_server_status_received, invoked due to + * the cancel, needs to acquire that same lock */ + grpc_call* lb_call = glb_policy->lb_call; + /* glb_policy->lb_call and this local lb_call must be consistent at this point * because glb_policy->lb_call is only assigned in lb_call_init_locked as part * of query_for_backends_locked, which can only be invoked while * glb_policy->shutting_down is false. */ - if (glb_policy->lb_call != nullptr) { - grpc_call_cancel(glb_policy->lb_call, nullptr); + if (lb_call != nullptr) { + grpc_call_cancel(lb_call, nullptr); /* lb_on_server_status_received will pick up the cancel and clean up */ } if (glb_policy->retry_timer_callback_pending) { @@ -904,8 +1031,12 @@ static void glb_shutdown_locked(grpc_lb_policy* pol, if (glb_policy->fallback_timer_callback_pending) { grpc_timer_cancel(&glb_policy->lb_fallback_timer); } + + pending_pick* pp = glb_policy->pending_picks; + glb_policy->pending_picks = nullptr; + pending_ping* pping = glb_policy->pending_pings; + glb_policy->pending_pings = nullptr; if (glb_policy->rr_policy != nullptr) { - grpc_lb_policy_shutdown_locked(glb_policy->rr_policy, nullptr); GRPC_LB_POLICY_UNREF(glb_policy->rr_policy, "glb_shutdown"); } else { grpc_lb_policy_try_reresolve(pol, &grpc_lb_glb_trace, GRPC_ERROR_CANCELLED); @@ -920,33 +1051,28 @@ static void glb_shutdown_locked(grpc_lb_policy* pol, } grpc_connectivity_state_set(&glb_policy->state_tracker, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "glb_shutdown"); - // Clear pending picks. - pending_pick* pp = glb_policy->pending_picks; - glb_policy->pending_picks = nullptr; + while (pp != nullptr) { pending_pick* next = pp->next; - if (new_policy != nullptr) { - // Hand pick over to new policy. - grpc_grpclb_client_stats_unref(pp->client_stats); - pp->pick->on_complete = pp->original_on_complete; - if (grpc_lb_policy_pick_locked(new_policy, pp->pick)) { - // Synchronous return; schedule callback. - GRPC_CLOSURE_SCHED(pp->pick->on_complete, GRPC_ERROR_NONE); - } - gpr_free(pp); - } else { - pp->pick->connected_subchannel = nullptr; - GRPC_CLOSURE_SCHED(&pp->on_complete, GRPC_ERROR_REF(error)); - } + *pp->target = nullptr; + GRPC_CLOSURE_SCHED(&pp->wrapped_on_complete_arg.wrapper_closure, + GRPC_ERROR_REF(error)); + gpr_free(pp); pp = next; } - // Clear pending pings. - pending_ping* pping = glb_policy->pending_pings; - glb_policy->pending_pings = nullptr; + while (pping != nullptr) { pending_ping* next = pping->next; - GRPC_CLOSURE_SCHED(pping->on_initiate, GRPC_ERROR_REF(error)); - GRPC_CLOSURE_SCHED(pping->on_ack, GRPC_ERROR_REF(error)); + if (pping->on_initiate != nullptr) { + GRPC_CLOSURE_SCHED(&pping->on_initiate->wrapper_closure, + GRPC_ERROR_REF(error)); + gpr_free(pping->on_initiate); + } + if (pping->on_ack != nullptr) { + GRPC_CLOSURE_SCHED(&pping->on_ack->wrapper_closure, + GRPC_ERROR_REF(error)); + gpr_free(pping->on_ack); + } gpr_free(pping); pping = next; } @@ -964,16 +1090,16 @@ static void glb_shutdown_locked(grpc_lb_policy* pol, // level (grpclb), inside the glb_policy->pending_picks list. To cancel these, // we invoke the completion closure and set *target to nullptr right here. static void glb_cancel_pick_locked(grpc_lb_policy* pol, - grpc_lb_policy_pick_state* pick, + grpc_connected_subchannel** target, grpc_error* error) { glb_lb_policy* glb_policy = (glb_lb_policy*)pol; pending_pick* pp = glb_policy->pending_picks; glb_policy->pending_picks = nullptr; while (pp != nullptr) { pending_pick* next = pp->next; - if (pp->pick == pick) { - pick->connected_subchannel = nullptr; - GRPC_CLOSURE_SCHED(&pp->on_complete, + if (pp->target == target) { + *target = nullptr; + GRPC_CLOSURE_SCHED(&pp->wrapped_on_complete_arg.wrapper_closure, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick Cancelled", &error, 1)); } else { @@ -983,7 +1109,7 @@ static void glb_cancel_pick_locked(grpc_lb_policy* pol, pp = next; } if (glb_policy->rr_policy != nullptr) { - grpc_lb_policy_cancel_pick_locked(glb_policy->rr_policy, pick, + grpc_lb_policy_cancel_pick_locked(glb_policy->rr_policy, target, GRPC_ERROR_REF(error)); } GRPC_ERROR_UNREF(error); @@ -1008,9 +1134,9 @@ static void glb_cancel_picks_locked(grpc_lb_policy* pol, glb_policy->pending_picks = nullptr; while (pp != nullptr) { pending_pick* next = pp->next; - if ((pp->pick->initial_metadata_flags & initial_metadata_flags_mask) == + if ((pp->pick_args.initial_metadata_flags & initial_metadata_flags_mask) == initial_metadata_flags_eq) { - GRPC_CLOSURE_SCHED(&pp->on_complete, + GRPC_CLOSURE_SCHED(&pp->wrapped_on_complete_arg.wrapper_closure, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick Cancelled", &error, 1)); } else { @@ -1036,7 +1162,7 @@ static void start_picking_locked(glb_lb_policy* glb_policy) { !glb_policy->fallback_timer_callback_pending) { grpc_millis deadline = grpc_core::ExecCtx::Get()->Now() + glb_policy->lb_fallback_timeout_ms; - GRPC_LB_POLICY_REF(&glb_policy->base, "grpclb_fallback_timer"); + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "grpclb_fallback_timer"); GRPC_CLOSURE_INIT(&glb_policy->lb_on_fallback, lb_on_fallback_timer_locked, glb_policy, grpc_combiner_scheduler(glb_policy->base.combiner)); @@ -1058,9 +1184,19 @@ static void glb_exit_idle_locked(grpc_lb_policy* pol) { } static int glb_pick_locked(grpc_lb_policy* pol, - grpc_lb_policy_pick_state* pick) { + const grpc_lb_policy_pick_args* pick_args, + grpc_connected_subchannel** target, + grpc_call_context_element* context, void** user_data, + grpc_closure* on_complete) { + if (pick_args->lb_token_mdelem_storage == nullptr) { + *target = nullptr; + GRPC_CLOSURE_SCHED(on_complete, + GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "No mdelem storage for the LB token. Load reporting " + "won't work without it. Failing")); + return 0; + } glb_lb_policy* glb_policy = (glb_lb_policy*)pol; - pending_pick* pp = pending_pick_create(glb_policy, pick); bool pick_done = false; if (glb_policy->rr_policy != nullptr) { const grpc_connectivity_state rr_connectivity_state = @@ -1068,7 +1204,7 @@ static int glb_pick_locked(grpc_lb_policy* pol, nullptr); // The glb_policy->rr_policy may have transitioned to SHUTDOWN but the // callback registered to capture this event - // (on_rr_connectivity_changed_locked) may not have been invoked yet. We + // (glb_rr_connectivity_changed_locked) may not have been invoked yet. We // need to make sure we aren't trying to pick from a RR policy instance // that's in shutdown. if (rr_connectivity_state == GRPC_CHANNEL_SHUTDOWN) { @@ -1078,16 +1214,32 @@ static int glb_pick_locked(grpc_lb_policy* pol, glb_policy, glb_policy->rr_policy, grpc_connectivity_state_name(rr_connectivity_state)); } - pending_pick_add(&glb_policy->pending_picks, pp); + add_pending_pick(&glb_policy->pending_picks, pick_args, target, context, + on_complete); pick_done = false; } else { // RR not in shutdown if (grpc_lb_glb_trace.enabled()) { gpr_log(GPR_INFO, "[grpclb %p] about to PICK from RR %p", glb_policy, glb_policy->rr_policy); } + GRPC_LB_POLICY_REF(glb_policy->rr_policy, "glb_pick"); + wrapped_rr_closure_arg* wc_arg = + (wrapped_rr_closure_arg*)gpr_zalloc(sizeof(wrapped_rr_closure_arg)); + GRPC_CLOSURE_INIT(&wc_arg->wrapper_closure, wrapped_rr_closure, wc_arg, + grpc_schedule_on_exec_ctx); + wc_arg->rr_policy = glb_policy->rr_policy; + wc_arg->target = target; + wc_arg->context = context; GPR_ASSERT(glb_policy->client_stats != nullptr); - pick_done = - pick_from_internal_rr_locked(glb_policy, false /* force_async */, pp); + wc_arg->client_stats = + grpc_grpclb_client_stats_ref(glb_policy->client_stats); + wc_arg->wrapped_closure = on_complete; + wc_arg->lb_token_mdelem_storage = pick_args->lb_token_mdelem_storage; + wc_arg->initial_metadata = pick_args->initial_metadata; + wc_arg->free_when_done = wc_arg; + wc_arg->glb_policy = pol; + pick_done = pick_from_internal_rr_locked( + glb_policy, pick_args, false /* force_async */, target, wc_arg); } } else { // glb_policy->rr_policy == NULL if (grpc_lb_glb_trace.enabled()) { @@ -1095,7 +1247,8 @@ static int glb_pick_locked(grpc_lb_policy* pol, "[grpclb %p] No RR policy. Adding to grpclb's pending picks", glb_policy); } - pending_pick_add(&glb_policy->pending_picks, pp); + add_pending_pick(&glb_policy->pending_picks, pick_args, target, context, + on_complete); if (!glb_policy->started_picking) { start_picking_locked(glb_policy); } @@ -1117,7 +1270,7 @@ static void glb_ping_one_locked(grpc_lb_policy* pol, grpc_closure* on_initiate, if (glb_policy->rr_policy) { grpc_lb_policy_ping_one_locked(glb_policy->rr_policy, on_initiate, on_ack); } else { - pending_ping_add(&glb_policy->pending_pings, on_initiate, on_ack); + add_pending_ping(&glb_policy->pending_pings, on_initiate, on_ack); if (!glb_policy->started_picking) { start_picking_locked(glb_policy); } @@ -1142,7 +1295,7 @@ static void lb_call_on_retry_timer_locked(void* arg, grpc_error* error) { } query_for_backends_locked(glb_policy); } - GRPC_LB_POLICY_UNREF(&glb_policy->base, "grpclb_retry_timer"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "grpclb_retry_timer"); } static void maybe_restart_lb_call(glb_lb_policy* glb_policy) { @@ -1168,7 +1321,7 @@ static void maybe_restart_lb_call(glb_lb_policy* glb_policy) { glb_policy); } } - GRPC_LB_POLICY_REF(&glb_policy->base, "grpclb_retry_timer"); + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "grpclb_retry_timer"); GRPC_CLOSURE_INIT(&glb_policy->lb_on_call_retry, lb_call_on_retry_timer_locked, glb_policy, grpc_combiner_scheduler(glb_policy->base.combiner)); @@ -1176,8 +1329,8 @@ static void maybe_restart_lb_call(glb_lb_policy* glb_policy) { grpc_timer_init(&glb_policy->lb_call_retry_timer, next_try, &glb_policy->lb_on_call_retry); } - GRPC_LB_POLICY_UNREF(&glb_policy->base, - "lb_on_server_status_received_locked"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, + "lb_on_server_status_received_locked"); } static void send_client_load_report_locked(void* arg, grpc_error* error); @@ -1200,7 +1353,7 @@ static void client_load_report_done_locked(void* arg, grpc_error* error) { glb_policy->client_load_report_payload = nullptr; if (error != GRPC_ERROR_NONE || glb_policy->lb_call == nullptr) { glb_policy->client_load_report_timer_callback_pending = false; - GRPC_LB_POLICY_UNREF(&glb_policy->base, "client_load_report"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "client_load_report"); if (glb_policy->lb_call == nullptr) { maybe_restart_lb_call(glb_policy); } @@ -1241,7 +1394,7 @@ static void send_client_load_report_locked(void* arg, grpc_error* error) { glb_lb_policy* glb_policy = (glb_lb_policy*)arg; if (error == GRPC_ERROR_CANCELLED || glb_policy->lb_call == nullptr) { glb_policy->client_load_report_timer_callback_pending = false; - GRPC_LB_POLICY_UNREF(&glb_policy->base, "client_load_report"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "client_load_report"); if (glb_policy->lb_call == nullptr) { maybe_restart_lb_call(glb_policy); } @@ -1394,8 +1547,10 @@ static void query_for_backends_locked(glb_lb_policy* glb_policy) { op->flags = 0; op->reserved = nullptr; op++; - /* take a ref to be released in lb_on_sent_initial_request_locked() */ - GRPC_LB_POLICY_REF(&glb_policy->base, "lb_on_sent_initial_request_locked"); + /* take a weak ref (won't prevent calling of \a glb_shutdown if the strong ref + * count goes to zero) to be unref'd in lb_on_sent_initial_request_locked() */ + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, + "lb_on_sent_initial_request_locked"); call_error = grpc_call_start_batch_and_execute( glb_policy->lb_call, ops, (size_t)(op - ops), &glb_policy->lb_on_sent_initial_request); @@ -1411,8 +1566,10 @@ static void query_for_backends_locked(glb_lb_policy* glb_policy) { op->flags = 0; op->reserved = nullptr; op++; - /* take a ref to be released in lb_on_server_status_received_locked() */ - GRPC_LB_POLICY_REF(&glb_policy->base, "lb_on_server_status_received_locked"); + /* take a weak ref (won't prevent calling of \a glb_shutdown if the strong ref + * count goes to zero) to be unref'd in lb_on_server_status_received_locked */ + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, + "lb_on_server_status_received_locked"); call_error = grpc_call_start_batch_and_execute( glb_policy->lb_call, ops, (size_t)(op - ops), &glb_policy->lb_on_server_status_received); @@ -1424,8 +1581,9 @@ static void query_for_backends_locked(glb_lb_policy* glb_policy) { op->flags = 0; op->reserved = nullptr; op++; - /* take a ref to be unref'd/reused in lb_on_response_received_locked() */ - GRPC_LB_POLICY_REF(&glb_policy->base, "lb_on_response_received_locked"); + /* take another weak ref to be unref'd/reused in + * lb_on_response_received_locked */ + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "lb_on_response_received_locked"); call_error = grpc_call_start_batch_and_execute( glb_policy->lb_call, ops, (size_t)(op - ops), &glb_policy->lb_on_response_received); @@ -1440,7 +1598,8 @@ static void lb_on_sent_initial_request_locked(void* arg, grpc_error* error) { if (glb_policy->client_load_report_payload != nullptr) { do_send_client_load_report_locked(glb_policy); } - GRPC_LB_POLICY_UNREF(&glb_policy->base, "lb_on_sent_initial_request_locked"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, + "lb_on_sent_initial_request_locked"); } static void lb_on_response_received_locked(void* arg, grpc_error* error) { @@ -1472,9 +1631,11 @@ static void lb_on_response_received_locked(void* arg, grpc_error* error) { "client load reporting interval = %" PRIdPTR " milliseconds", glb_policy, glb_policy->client_stats_report_interval); } - /* take a ref to be unref'd in send_client_load_report_locked() */ + /* take a weak ref (won't prevent calling of \a glb_shutdown() if the + * strong ref count goes to zero) to be unref'd in + * send_client_load_report_locked() */ glb_policy->client_load_report_timer_callback_pending = true; - GRPC_LB_POLICY_REF(&glb_policy->base, "client_load_report"); + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "client_load_report"); schedule_next_client_load_report(glb_policy); } else if (grpc_lb_glb_trace.enabled()) { gpr_log(GPR_INFO, @@ -1556,21 +1717,21 @@ static void lb_on_response_received_locked(void* arg, grpc_error* error) { op->flags = 0; op->reserved = nullptr; op++; - /* reuse the "lb_on_response_received_locked" ref taken in + /* reuse the "lb_on_response_received_locked" weak ref taken in * query_for_backends_locked() */ const grpc_call_error call_error = grpc_call_start_batch_and_execute( glb_policy->lb_call, ops, (size_t)(op - ops), &glb_policy->lb_on_response_received); /* loop */ GPR_ASSERT(GRPC_CALL_OK == call_error); } else { - GRPC_LB_POLICY_UNREF(&glb_policy->base, - "lb_on_response_received_locked_shutdown"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, + "lb_on_response_received_locked_shutdown"); } } else { /* empty payload: call cancelled. */ - /* dispose of the "lb_on_response_received_locked" ref taken in + /* dispose of the "lb_on_response_received_locked" weak ref taken in * query_for_backends_locked() and reused in every reception loop */ - GRPC_LB_POLICY_UNREF(&glb_policy->base, - "lb_on_response_received_locked_empty_payload"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, + "lb_on_response_received_locked_empty_payload"); } } @@ -1590,7 +1751,7 @@ static void lb_on_fallback_timer_locked(void* arg, grpc_error* error) { rr_handover_locked(glb_policy); } } - GRPC_LB_POLICY_UNREF(&glb_policy->base, "grpclb_fallback_timer"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "grpclb_fallback_timer"); } static void lb_on_server_status_received_locked(void* arg, grpc_error* error) { @@ -1674,7 +1835,7 @@ static void glb_update_locked(grpc_lb_policy* policy, grpc_channel_get_channel_stack(glb_policy->lb_channel)); GPR_ASSERT(client_channel_elem->filter == &grpc_client_channel_filter); glb_policy->watching_lb_channel = true; - GRPC_LB_POLICY_REF(&glb_policy->base, "watch_lb_channel_connectivity"); + GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "watch_lb_channel_connectivity"); grpc_client_channel_watch_connectivity_state( client_channel_elem, grpc_polling_entity_create_from_pollset_set( @@ -1730,8 +1891,8 @@ static void glb_lb_channel_on_connectivity_changed_cb(void* arg, case GRPC_CHANNEL_SHUTDOWN: done: glb_policy->watching_lb_channel = false; - GRPC_LB_POLICY_UNREF(&glb_policy->base, - "watch_lb_channel_connectivity_cb_shutdown"); + GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, + "watch_lb_channel_connectivity_cb_shutdown"); break; } } 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 60385272cf..9ff40aa53c 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 @@ -31,6 +31,15 @@ grpc_core::TraceFlag grpc_lb_pick_first_trace(false, "pick_first"); +namespace { +struct pending_pick { + struct pending_pick* next; + uint32_t initial_metadata_flags; + grpc_connected_subchannel** target; + grpc_closure* on_complete; +}; +} // namespace + typedef struct { /** base policy: must be first */ grpc_lb_policy base; @@ -45,7 +54,7 @@ typedef struct { /** are we shut down? */ bool shutdown; /** list of picks that are waiting on connectivity */ - grpc_lb_policy_pick_state* pending_picks; + pending_pick* pending_picks; /** our connectivity state tracker */ grpc_connectivity_state_tracker state_tracker; } pick_first_lb_policy; @@ -63,27 +72,19 @@ static void pf_destroy(grpc_lb_policy* pol) { } } -static void pf_shutdown_locked(grpc_lb_policy* pol, - grpc_lb_policy* new_policy) { +static void pf_shutdown_locked(grpc_lb_policy* pol) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown"); if (grpc_lb_pick_first_trace.enabled()) { gpr_log(GPR_DEBUG, "Pick First %p Shutting down", p); } p->shutdown = true; - grpc_lb_policy_pick_state* pick; - while ((pick = p->pending_picks) != nullptr) { - p->pending_picks = pick->next; - if (new_policy != nullptr) { - // Hand off to new LB policy. - if (grpc_lb_policy_pick_locked(new_policy, pick)) { - // Synchronous return, schedule closure. - GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE); - } - } else { - pick->connected_subchannel = nullptr; - GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_REF(error)); - } + pending_pick* pp; + while ((pp = p->pending_picks) != nullptr) { + p->pending_picks = pp->next; + *pp->target = nullptr; + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_REF(error)); + gpr_free(pp); } grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "shutdown"); @@ -103,18 +104,19 @@ static void pf_shutdown_locked(grpc_lb_policy* pol, } static void pf_cancel_pick_locked(grpc_lb_policy* pol, - grpc_lb_policy_pick_state* pick, + grpc_connected_subchannel** target, grpc_error* error) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; - grpc_lb_policy_pick_state* pp = p->pending_picks; + pending_pick* pp = p->pending_picks; p->pending_picks = nullptr; while (pp != nullptr) { - grpc_lb_policy_pick_state* next = pp->next; - if (pp == pick) { - pick->connected_subchannel = nullptr; - GRPC_CLOSURE_SCHED(pick->on_complete, + pending_pick* next = pp->next; + if (pp->target == target) { + *target = nullptr; + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick Cancelled", &error, 1)); + gpr_free(pp); } else { pp->next = p->pending_picks; p->pending_picks = pp; @@ -129,20 +131,21 @@ static void pf_cancel_picks_locked(grpc_lb_policy* pol, uint32_t initial_metadata_flags_eq, grpc_error* error) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; - grpc_lb_policy_pick_state* pick = p->pending_picks; + pending_pick* pp = p->pending_picks; p->pending_picks = nullptr; - while (pick != nullptr) { - grpc_lb_policy_pick_state* next = pick->next; - if ((pick->initial_metadata_flags & initial_metadata_flags_mask) == + while (pp != nullptr) { + pending_pick* next = pp->next; + if ((pp->initial_metadata_flags & initial_metadata_flags_mask) == initial_metadata_flags_eq) { - GRPC_CLOSURE_SCHED(pick->on_complete, + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick Cancelled", &error, 1)); + gpr_free(pp); } else { - pick->next = p->pending_picks; - p->pending_picks = pick; + pp->next = p->pending_picks; + p->pending_picks = pp; } - pick = next; + pp = next; } GRPC_ERROR_UNREF(error); } @@ -172,20 +175,27 @@ static void pf_exit_idle_locked(grpc_lb_policy* pol) { } static int pf_pick_locked(grpc_lb_policy* pol, - grpc_lb_policy_pick_state* pick) { + const grpc_lb_policy_pick_args* pick_args, + grpc_connected_subchannel** target, + grpc_call_context_element* context, void** user_data, + grpc_closure* on_complete) { pick_first_lb_policy* p = (pick_first_lb_policy*)pol; // If we have a selected subchannel already, return synchronously. if (p->selected != nullptr) { - pick->connected_subchannel = GRPC_CONNECTED_SUBCHANNEL_REF( - p->selected->connected_subchannel, "picked"); + *target = GRPC_CONNECTED_SUBCHANNEL_REF(p->selected->connected_subchannel, + "picked"); return 1; } // No subchannel selected yet, so handle asynchronously. if (!p->started_picking) { start_picking_locked(p); } - pick->next = p->pending_picks; - p->pending_picks = pick; + pending_pick* pp = (pending_pick*)gpr_malloc(sizeof(*pp)); + pp->next = p->pending_picks; + pp->target = target; + pp->initial_metadata_flags = pick_args->initial_metadata_flags; + pp->on_complete = on_complete; + p->pending_picks = pp; return 0; } @@ -471,17 +481,18 @@ static void pf_connectivity_changed_locked(void* arg, grpc_error* error) { // Drop all other subchannels, since we are now connected. destroy_unselected_subchannels_locked(p); // Update any calls that were waiting for a pick. - grpc_lb_policy_pick_state* pick; - while ((pick = p->pending_picks)) { - p->pending_picks = pick->next; - pick->connected_subchannel = GRPC_CONNECTED_SUBCHANNEL_REF( + pending_pick* pp; + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = GRPC_CONNECTED_SUBCHANNEL_REF( p->selected->connected_subchannel, "picked"); if (grpc_lb_pick_first_trace.enabled()) { gpr_log(GPR_INFO, "Servicing pending pick with selected subchannel %p", (void*)p->selected); } - GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE); + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_NONE); + gpr_free(pp); } // Renew notification. grpc_lb_subchannel_data_start_connectivity_watch(sd); 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 92c7d5bd5d..a964af0627 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 @@ -41,6 +41,31 @@ grpc_core::TraceFlag grpc_lb_round_robin_trace(false, "round_robin"); +namespace { +/** List of entities waiting for a pick. + * + * Once a pick is available, \a target is updated and \a on_complete called. */ +struct pending_pick { + pending_pick* next; + + /* output argument where to store the pick()ed user_data. It'll be NULL if no + * such data is present or there's an error (the definite test for errors is + * \a target being NULL). */ + void** user_data; + + /* bitmask passed to pick() and used for selective cancelling. See + * grpc_lb_policy_cancel_picks() */ + uint32_t initial_metadata_flags; + + /* output argument where to store the pick()ed connected subchannel, or NULL + * upon error. */ + grpc_connected_subchannel** target; + + /* to be invoked once the pick() has completed (regardless of success) */ + grpc_closure* on_complete; +}; +} // namespace + typedef struct round_robin_lb_policy { /** base policy: must be first */ grpc_lb_policy base; @@ -52,7 +77,7 @@ typedef struct round_robin_lb_policy { /** are we shutting down? */ bool shutdown; /** List of picks that are waiting on connectivity */ - grpc_lb_policy_pick_state* pending_picks; + pending_pick* pending_picks; /** our connectivity state tracker */ grpc_connectivity_state_tracker state_tracker; @@ -144,27 +169,19 @@ static void rr_destroy(grpc_lb_policy* pol) { gpr_free(p); } -static void rr_shutdown_locked(grpc_lb_policy* pol, - grpc_lb_policy* new_policy) { +static void rr_shutdown_locked(grpc_lb_policy* pol) { round_robin_lb_policy* p = (round_robin_lb_policy*)pol; grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown"); if (grpc_lb_round_robin_trace.enabled()) { gpr_log(GPR_DEBUG, "[RR %p] Shutting down", p); } p->shutdown = true; - grpc_lb_policy_pick_state* pick; - while ((pick = p->pending_picks) != nullptr) { - p->pending_picks = pick->next; - if (new_policy != nullptr) { - // Hand off to new LB policy. - if (grpc_lb_policy_pick_locked(new_policy, pick)) { - // Synchronous return; schedule callback. - GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE); - } - } else { - pick->connected_subchannel = nullptr; - GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_REF(error)); - } + pending_pick* pp; + while ((pp = p->pending_picks) != nullptr) { + p->pending_picks = pp->next; + *pp->target = nullptr; + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_REF(error)); + gpr_free(pp); } grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "rr_shutdown"); @@ -184,18 +201,19 @@ static void rr_shutdown_locked(grpc_lb_policy* pol, } static void rr_cancel_pick_locked(grpc_lb_policy* pol, - grpc_lb_policy_pick_state* pick, + grpc_connected_subchannel** target, grpc_error* error) { round_robin_lb_policy* p = (round_robin_lb_policy*)pol; - grpc_lb_policy_pick_state* pp = p->pending_picks; + pending_pick* pp = p->pending_picks; p->pending_picks = nullptr; while (pp != nullptr) { - grpc_lb_policy_pick_state* next = pp->next; - if (pp == pick) { - pick->connected_subchannel = nullptr; - GRPC_CLOSURE_SCHED(pick->on_complete, + pending_pick* next = pp->next; + if (pp->target == target) { + *target = nullptr; + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick cancelled", &error, 1)); + gpr_free(pp); } else { pp->next = p->pending_picks; p->pending_picks = pp; @@ -210,21 +228,22 @@ static void rr_cancel_picks_locked(grpc_lb_policy* pol, uint32_t initial_metadata_flags_eq, grpc_error* error) { round_robin_lb_policy* p = (round_robin_lb_policy*)pol; - grpc_lb_policy_pick_state* pick = p->pending_picks; + pending_pick* pp = p->pending_picks; p->pending_picks = nullptr; - while (pick != nullptr) { - grpc_lb_policy_pick_state* next = pick->next; - if ((pick->initial_metadata_flags & initial_metadata_flags_mask) == + while (pp != nullptr) { + pending_pick* next = pp->next; + if ((pp->initial_metadata_flags & initial_metadata_flags_mask) == initial_metadata_flags_eq) { - pick->connected_subchannel = nullptr; - GRPC_CLOSURE_SCHED(pick->on_complete, + *pp->target = nullptr; + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( "Pick cancelled", &error, 1)); + gpr_free(pp); } else { - pick->next = p->pending_picks; - p->pending_picks = pick; + pp->next = p->pending_picks; + p->pending_picks = pp; } - pick = next; + pp = next; } GRPC_ERROR_UNREF(error); } @@ -249,10 +268,13 @@ static void rr_exit_idle_locked(grpc_lb_policy* pol) { } static int rr_pick_locked(grpc_lb_policy* pol, - grpc_lb_policy_pick_state* pick) { + const grpc_lb_policy_pick_args* pick_args, + grpc_connected_subchannel** target, + grpc_call_context_element* context, void** user_data, + grpc_closure* on_complete) { round_robin_lb_policy* p = (round_robin_lb_policy*)pol; if (grpc_lb_round_robin_trace.enabled()) { - gpr_log(GPR_INFO, "[RR %p] Trying to pick (shutdown: %d)", pol, + gpr_log(GPR_INFO, "[RR %p] Trying to pick (shutdown: %d)", (void*)pol, p->shutdown); } GPR_ASSERT(!p->shutdown); @@ -262,18 +284,18 @@ static int rr_pick_locked(grpc_lb_policy* pol, /* readily available, report right away */ grpc_lb_subchannel_data* sd = &p->subchannel_list->subchannels[next_ready_index]; - pick->connected_subchannel = + *target = GRPC_CONNECTED_SUBCHANNEL_REF(sd->connected_subchannel, "rr_picked"); - if (pick->user_data != nullptr) { - *pick->user_data = sd->user_data; + if (user_data != nullptr) { + *user_data = sd->user_data; } if (grpc_lb_round_robin_trace.enabled()) { gpr_log( GPR_DEBUG, "[RR %p] Picked target <-- Subchannel %p (connected %p) (sl %p, " - "index %" PRIuPTR ")", - p, sd->subchannel, pick->connected_subchannel, sd->subchannel_list, - next_ready_index); + "index %lu)", + (void*)p, (void*)sd->subchannel, (void*)*target, + (void*)sd->subchannel_list, (unsigned long)next_ready_index); } /* only advance the last picked pointer if the selection was used */ update_last_ready_subchannel_index_locked(p, next_ready_index); @@ -284,8 +306,13 @@ static int rr_pick_locked(grpc_lb_policy* pol, if (!p->started_picking) { start_picking_locked(p); } - pick->next = p->pending_picks; - p->pending_picks = pick; + pending_pick* pp = (pending_pick*)gpr_malloc(sizeof(*pp)); + pp->next = p->pending_picks; + pp->target = target; + pp->on_complete = on_complete; + pp->initial_metadata_flags = pick_args->initial_metadata_flags; + pp->user_data = user_data; + p->pending_picks = pp; return 0; } @@ -468,13 +495,13 @@ static void rr_connectivity_changed_locked(void* arg, grpc_error* error) { // picks, update the last picked pointer update_last_ready_subchannel_index_locked(p, next_ready_index); } - grpc_lb_policy_pick_state* pick; - while ((pick = p->pending_picks)) { - p->pending_picks = pick->next; - pick->connected_subchannel = GRPC_CONNECTED_SUBCHANNEL_REF( + pending_pick* pp; + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = GRPC_CONNECTED_SUBCHANNEL_REF( selected->connected_subchannel, "rr_picked"); - if (pick->user_data != nullptr) { - *pick->user_data = selected->user_data; + if (pp->user_data != nullptr) { + *pp->user_data = selected->user_data; } if (grpc_lb_round_robin_trace.enabled()) { gpr_log(GPR_DEBUG, @@ -483,7 +510,8 @@ static void rr_connectivity_changed_locked(void* arg, grpc_error* error) { (void*)p, (void*)selected->subchannel, (void*)p->subchannel_list, (unsigned long)next_ready_index); } - GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE); + GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_NONE); + gpr_free(pp); } } // Renew notification. diff --git a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc index 5ce1298afc..a3b4c8e524 100644 --- a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc +++ b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc @@ -213,13 +213,13 @@ void grpc_lb_subchannel_list_unref(grpc_lb_subchannel_list* subchannel_list, void grpc_lb_subchannel_list_ref_for_connectivity_watch( grpc_lb_subchannel_list* subchannel_list, const char* reason) { - GRPC_LB_POLICY_REF(subchannel_list->policy, reason); + GRPC_LB_POLICY_WEAK_REF(subchannel_list->policy, reason); grpc_lb_subchannel_list_ref(subchannel_list, reason); } void grpc_lb_subchannel_list_unref_for_connectivity_watch( grpc_lb_subchannel_list* subchannel_list, const char* reason) { - GRPC_LB_POLICY_UNREF(subchannel_list->policy, reason); + GRPC_LB_POLICY_WEAK_UNREF(subchannel_list->policy, reason); grpc_lb_subchannel_list_unref(subchannel_list, reason); } diff --git a/src/core/lib/iomgr/closure.h b/src/core/lib/iomgr/closure.h index 88af76006a..4c58c0e4bf 100644 --- a/src/core/lib/iomgr/closure.h +++ b/src/core/lib/iomgr/closure.h @@ -22,7 +22,6 @@ #include <grpc/support/port_platform.h> #include <assert.h> -#include <grpc/impl/codegen/exec_ctx_fwd.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <stdbool.h> diff --git a/src/core/lib/iomgr/iomgr.h b/src/core/lib/iomgr/iomgr.h index 3f238c660a..c7cde7ea59 100644 --- a/src/core/lib/iomgr/iomgr.h +++ b/src/core/lib/iomgr/iomgr.h @@ -19,7 +19,6 @@ #ifndef GRPC_CORE_LIB_IOMGR_IOMGR_H #define GRPC_CORE_LIB_IOMGR_IOMGR_H -#include <grpc/impl/codegen/exec_ctx_fwd.h> #include "src/core/lib/iomgr/port.h" /** Initializes the iomgr. */ diff --git a/src/core/lib/support/vector.h b/src/core/lib/support/vector.h index 4a7db80676..2f249a5b9e 100644 --- a/src/core/lib/support/vector.h +++ b/src/core/lib/support/vector.h @@ -19,13 +19,93 @@ #ifndef GRPC_CORE_LIB_SUPPORT_VECTOR_H #define GRPC_CORE_LIB_SUPPORT_VECTOR_H -#include "absl/container/inlined_vector.h" +#include <cassert> + #include "src/core/lib/support/memory.h" namespace grpc_core { +// NOTE: We eventually want to use absl::InlinedVector here. However, +// there are currently build problems that prevent us from using absl. +// In the interim, we define a custom implementation as a place-holder, +// with the intent to eventually replace this with the absl +// implementation. +// +// This place-holder implementation does not implement the full set of +// functionality from the absl version; it has just the methods that we +// currently happen to need in gRPC. If additional functionality is +// needed before this gets replaced with the absl version, it can be +// added, with the following proviso: +// +// ANY METHOD ADDED HERE MUST COMPLY WITH THE INTERFACE IN THE absl +// IMPLEMENTATION! +// +// TODO(ctiller, nnoble, roth): Replace this with absl::InlinedVector +// once we integrate absl into the gRPC build system in a usable way. template <typename T, size_t N> -using InlinedVector = absl::InlinedVector<T, N, Allocator<T>>; +class InlinedVector { + public: + InlinedVector() {} + ~InlinedVector() { + for (size_t i = 0; i < size_ && i < N; ++i) { + T& value = *reinterpret_cast<T*>(inline_ + i); + value.~T(); + } + if (size_ > N) { // Avoid subtracting two signed values. + for (size_t i = 0; i < size_ - N; ++i) { + dynamic_[i].~T(); + } + } + gpr_free(dynamic_); + } + + // For now, we do not support copying. + InlinedVector(const InlinedVector&) = delete; + InlinedVector& operator=(const InlinedVector&) = delete; + + T& operator[](size_t offset) { + assert(offset < size_); + if (offset < N) { + return *reinterpret_cast<T*>(inline_ + offset); + } else { + return dynamic_[offset - N]; + } + } + + template <typename... Args> + void emplace_back(Args&&... args) { + if (size_ < N) { + new (&inline_[size_]) T(std::forward<Args>(args)...); + } else { + if (size_ - N == dynamic_capacity_) { + size_t new_capacity = + dynamic_capacity_ == 0 ? 2 : dynamic_capacity_ * 2; + T* new_dynamic = static_cast<T*>(gpr_malloc(sizeof(T) * new_capacity)); + for (size_t i = 0; i < dynamic_capacity_; ++i) { + new (&new_dynamic[i]) T(std::move(dynamic_[i])); + dynamic_[i].~T(); + } + gpr_free(dynamic_); + dynamic_ = new_dynamic; + dynamic_capacity_ = new_capacity; + } + new (&dynamic_[size_ - N]) T(std::forward<Args>(args)...); + } + ++size_; + } + + void push_back(const T& value) { emplace_back(value); } + + void push_back(T&& value) { emplace_back(std::move(value)); } + + size_t size() const { return size_; } + + private: + typename std::aligned_storage<sizeof(T)>::type inline_[N]; + T* dynamic_ = nullptr; + size_t size_ = 0; + size_t dynamic_capacity_ = 0; +}; } // namespace grpc_core diff --git a/src/python/grpcio/grpc/_channel.py b/src/python/grpcio/grpc/_channel.py index 24be042f61..bfc7208310 100644 --- a/src/python/grpcio/grpc/_channel.py +++ b/src/python/grpcio/grpc/_channel.py @@ -27,7 +27,6 @@ from grpc.framework.foundation import callable_util _USER_AGENT = 'grpc-python/{}'.format(_grpcio_metadata.__version__) _EMPTY_FLAGS = 0 -_INFINITE_FUTURE = cygrpc.Timespec(float('+inf')) _UNARY_UNARY_INITIAL_DUE = ( cygrpc.OperationType.send_initial_metadata, @@ -61,11 +60,7 @@ _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE = ( def _deadline(timeout): - if timeout is None: - return None, _INFINITE_FUTURE - else: - deadline = time.time() + timeout - return deadline, cygrpc.Timespec(deadline) + return None if timeout is None else time.time() + timeout def _unknown_code_details(unknown_cygrpc_code, details): @@ -420,15 +415,15 @@ class _Rendezvous(grpc.RpcError, grpc.Future, grpc.Call): def _start_unary_request(request, timeout, request_serializer): - deadline, deadline_timespec = _deadline(timeout) + deadline = _deadline(timeout) serialized_request = _common.serialize(request, request_serializer) if serialized_request is None: state = _RPCState((), (), (), grpc.StatusCode.INTERNAL, 'Exception serializing request!') rendezvous = _Rendezvous(state, None, None, deadline) - return deadline, deadline_timespec, None, rendezvous + return deadline, None, rendezvous else: - return deadline, deadline_timespec, serialized_request, None + return deadline, serialized_request, None def _end_unary_response_blocking(state, call, with_call, deadline): @@ -453,10 +448,10 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): self._response_deserializer = response_deserializer def _prepare(self, request, timeout, metadata): - deadline, deadline_timespec, serialized_request, rendezvous = ( - _start_unary_request(request, timeout, self._request_serializer)) + deadline, serialized_request, rendezvous = (_start_unary_request( + request, timeout, self._request_serializer)) if serialized_request is None: - return None, None, None, None, rendezvous + return None, None, None, rendezvous else: state = _RPCState(_UNARY_UNARY_INITIAL_DUE, None, None, None, None) operations = ( @@ -467,18 +462,17 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS), cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS), ) - return state, operations, deadline, deadline_timespec, None + return state, operations, deadline, None def _blocking(self, request, timeout, metadata, credentials): - state, operations, deadline, deadline_timespec, rendezvous = self._prepare( + state, operations, deadline, rendezvous = self._prepare( request, timeout, metadata) if rendezvous: raise rendezvous else: completion_queue = cygrpc.CompletionQueue() call = self._channel.create_call(None, 0, completion_queue, - self._method, None, - deadline_timespec) + self._method, None, deadline) if credentials is not None: call.set_credentials(credentials._credentials) call_error = call.start_client_batch(operations, None) @@ -498,13 +492,13 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): return _end_unary_response_blocking(state, call, True, deadline) def future(self, request, timeout=None, metadata=None, credentials=None): - state, operations, deadline, deadline_timespec, rendezvous = self._prepare( + state, operations, deadline, rendezvous = self._prepare( request, timeout, metadata) if rendezvous: return rendezvous else: call, drive_call = self._managed_call(None, 0, self._method, None, - deadline_timespec) + deadline) if credentials is not None: call.set_credentials(credentials._credentials) event_handler = _event_handler(state, call, @@ -530,14 +524,14 @@ class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): self._response_deserializer = response_deserializer def __call__(self, request, timeout=None, metadata=None, credentials=None): - deadline, deadline_timespec, serialized_request, rendezvous = ( - _start_unary_request(request, timeout, self._request_serializer)) + deadline, serialized_request, rendezvous = (_start_unary_request( + request, timeout, self._request_serializer)) if serialized_request is None: raise rendezvous else: state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None) call, drive_call = self._managed_call(None, 0, self._method, None, - deadline_timespec) + deadline) if credentials is not None: call.set_credentials(credentials._credentials) event_handler = _event_handler(state, call, @@ -573,11 +567,11 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): self._response_deserializer = response_deserializer def _blocking(self, request_iterator, timeout, metadata, credentials): - deadline, deadline_timespec = _deadline(timeout) + deadline = _deadline(timeout) state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) completion_queue = cygrpc.CompletionQueue() call = self._channel.create_call(None, 0, completion_queue, - self._method, None, deadline_timespec) + self._method, None, deadline) if credentials is not None: call.set_credentials(credentials._credentials) with state.condition: @@ -624,10 +618,10 @@ class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): timeout=None, metadata=None, credentials=None): - deadline, deadline_timespec = _deadline(timeout) + deadline = _deadline(timeout) state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) call, drive_call = self._managed_call(None, 0, self._method, None, - deadline_timespec) + deadline) if credentials is not None: call.set_credentials(credentials._credentials) event_handler = _event_handler(state, call, self._response_deserializer) @@ -665,10 +659,10 @@ class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): timeout=None, metadata=None, credentials=None): - deadline, deadline_timespec = _deadline(timeout) + deadline = _deadline(timeout) state = _RPCState(_STREAM_STREAM_INITIAL_DUE, None, None, None, None) call, drive_call = self._managed_call(None, 0, self._method, None, - deadline_timespec) + deadline) if credentials is not None: call.set_credentials(credentials._credentials) event_handler = _event_handler(state, call, self._response_deserializer) @@ -737,7 +731,8 @@ def _channel_managed_call_management(state): flags: An integer bitfield of call flags. method: The RPC method. host: A host string for the created call. - deadline: A cygrpc.Timespec to be the deadline of the created call. + deadline: A float to be the deadline of the created call or None if the + call is to have an infinite deadline. Returns: A cygrpc.Call with which to conduct an RPC and a function to call if @@ -827,8 +822,8 @@ def _poll_connectivity(state, channel, initial_try_to_connect): completion_queue = cygrpc.CompletionQueue() while True: channel.watch_connectivity_state(connectivity, - cygrpc.Timespec(time.time() + 0.2), - completion_queue, None) + time.time() + 0.2, completion_queue, + None) event = completion_queue.poll() with state.lock: if not state.callbacks_and_connectivities and not state.try_to_connect: diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi index 443d534d7e..efe5f2e0db 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi @@ -42,7 +42,7 @@ cdef class Channel: def create_call(self, Call parent, int flags, CompletionQueue queue not None, - method, host, Timespec deadline not None): + method, host, object deadline): if queue.is_shutting_down: raise ValueError("queue must not be shutting down or shutdown") cdef grpc_slice method_slice = _slice_from_bytes(method) @@ -56,14 +56,13 @@ cdef class Channel: cdef grpc_call *parent_call = NULL if parent is not None: parent_call = parent.c_call - with nogil: - operation_call.c_call = grpc_channel_create_call( - self.c_channel, parent_call, flags, - queue.c_completion_queue, method_slice, host_slice_ptr, - deadline.c_time, NULL) - grpc_slice_unref(method_slice) - if host_slice_ptr: - grpc_slice_unref(host_slice) + operation_call.c_call = grpc_channel_create_call( + self.c_channel, parent_call, flags, + queue.c_completion_queue, method_slice, host_slice_ptr, + _timespec_from_time(deadline), NULL) + grpc_slice_unref(method_slice) + if host_slice_ptr: + grpc_slice_unref(host_slice) return operation_call def check_connectivity_state(self, bint try_to_connect): @@ -75,13 +74,12 @@ cdef class Channel: def watch_connectivity_state( self, grpc_connectivity_state last_observed_state, - Timespec deadline not None, CompletionQueue queue not None, tag): + object deadline, CompletionQueue queue not None, tag): cdef _ConnectivityTag connectivity_tag = _ConnectivityTag(tag) cpython.Py_INCREF(connectivity_tag) - with nogil: - grpc_channel_watch_connectivity_state( - self.c_channel, last_observed_state, deadline.c_time, - queue.c_completion_queue, <cpython.PyObject *>connectivity_tag) + grpc_channel_watch_connectivity_state( + self.c_channel, last_observed_state, _timespec_from_time(deadline), + queue.c_completion_queue, <cpython.PyObject *>connectivity_tag) def target(self): cdef char *target = NULL diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi index e259789b35..40496d1124 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi @@ -52,17 +52,18 @@ cdef class CompletionQueue: cpython.Py_DECREF(tag) return tag.event(event) - def poll(self, Timespec deadline=None): + def poll(self, deadline=None): # We name this 'poll' to avoid problems with CPython's expectations for # 'special' methods (like next and __next__). cdef gpr_timespec c_increment cdef gpr_timespec c_timeout cdef gpr_timespec c_deadline + if deadline is None: + c_deadline = gpr_inf_future(GPR_CLOCK_REALTIME) + else: + c_deadline = _timespec_from_time(deadline) with nogil: c_increment = gpr_time_from_millis(_INTERRUPT_CHECK_PERIOD_MS, GPR_TIMESPAN) - c_deadline = gpr_inf_future(GPR_CLOCK_REALTIME) - if deadline is not None: - c_deadline = deadline.c_time while True: c_timeout = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), c_increment) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi index 7b2482d947..297bbadfe0 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi @@ -18,11 +18,6 @@ cdef grpc_slice _copy_slice(grpc_slice slice) nogil cdef grpc_slice _slice_from_bytes(bytes value) nogil -cdef class Timespec: - - cdef gpr_timespec c_time - - cdef class CallDetails: cdef grpc_call_details c_details diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi index bc2cd0338e..b2343b53d6 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi @@ -123,74 +123,6 @@ class CompressionLevel: high = GRPC_COMPRESS_LEVEL_HIGH -cdef class Timespec: - - def __cinit__(self, time): - if time is None: - with nogil: - self.c_time = gpr_now(GPR_CLOCK_REALTIME) - return - if isinstance(time, int): - time = float(time) - if isinstance(time, float): - if time == float("+inf"): - with nogil: - self.c_time = gpr_inf_future(GPR_CLOCK_REALTIME) - elif time == float("-inf"): - with nogil: - self.c_time = gpr_inf_past(GPR_CLOCK_REALTIME) - else: - self.c_time.seconds = time - self.c_time.nanoseconds = (time - float(self.c_time.seconds)) * 1e9 - self.c_time.clock_type = GPR_CLOCK_REALTIME - elif isinstance(time, Timespec): - self.c_time = (<Timespec>time).c_time - else: - raise TypeError("expected time to be float, int, or Timespec, not {}" - .format(type(time))) - - @property - def seconds(self): - # TODO(atash) ensure that everywhere a Timespec is created that it's - # converted to GPR_CLOCK_REALTIME then and not every time someone wants to - # read values off in Python. - cdef gpr_timespec real_time - with nogil: - real_time = ( - gpr_convert_clock_type(self.c_time, GPR_CLOCK_REALTIME)) - return real_time.seconds - - @property - def nanoseconds(self): - cdef gpr_timespec real_time = ( - gpr_convert_clock_type(self.c_time, GPR_CLOCK_REALTIME)) - return real_time.nanoseconds - - def __float__(self): - cdef gpr_timespec real_time = ( - gpr_convert_clock_type(self.c_time, GPR_CLOCK_REALTIME)) - return <double>real_time.seconds + <double>real_time.nanoseconds / 1e9 - - def __richcmp__(Timespec self not None, Timespec other not None, int op): - cdef gpr_timespec self_c_time = self.c_time - cdef gpr_timespec other_c_time = other.c_time - cdef int result = gpr_time_cmp(self_c_time, other_c_time) - if op == 0: # < - return result < 0 - elif op == 2: # == - return result == 0 - elif op == 4: # > - return result > 0 - elif op == 1: # <= - return result <= 0 - elif op == 3: # != - return result != 0 - elif op == 5: # >= - return result >= 0 - else: - raise ValueError('__richcmp__ `op` contract violated') - - cdef class CallDetails: def __cinit__(self): @@ -213,9 +145,7 @@ cdef class CallDetails: @property def deadline(self): - timespec = Timespec(float("-inf")) - timespec.c_time = self.c_details.deadline - return timespec + return _time_from_timespec(self.c_details.deadline) cdef class SslPemKeyCertPair: diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi index c19beccde6..e5d28a85d5 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi @@ -106,7 +106,7 @@ cdef class Server: with nogil: grpc_server_start(self.c_server) # Ensure the core has gotten a chance to do the start-up work - self.backup_shutdown_queue.poll(Timespec(None)) + self.backup_shutdown_queue.poll(deadline=time.time()) def add_http2_port(self, bytes address, ServerCredentials server_credentials=None): diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi new file mode 100644 index 0000000000..ce67c61eaf --- /dev/null +++ b/src/python/grpcio/grpc/_cython/_cygrpc/time.pxd.pxi @@ -0,0 +1,19 @@ +# 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. + + +cdef gpr_timespec _timespec_from_time(object time) + + +cdef double _time_from_timespec(gpr_timespec timespec) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi new file mode 100644 index 0000000000..7a668680b8 --- /dev/null +++ b/src/python/grpcio/grpc/_cython/_cygrpc/time.pyx.pxi @@ -0,0 +1,30 @@ +# 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. + + +cdef gpr_timespec _timespec_from_time(object time): + cdef gpr_timespec timespec + if time is None: + return gpr_inf_future(GPR_CLOCK_REALTIME) + else: + timespec.seconds = time + timespec.nanoseconds = (time - float(timespec.seconds)) * 1e9 + timespec.clock_type = GPR_CLOCK_REALTIME + return timespec + + +cdef double _time_from_timespec(gpr_timespec timespec): + cdef gpr_timespec real_timespec = gpr_convert_clock_type( + timespec, GPR_CLOCK_REALTIME) + return <double>real_timespec.seconds + <double>real_timespec.nanoseconds / 1e9 diff --git a/src/python/grpcio/grpc/_cython/cygrpc.pxd b/src/python/grpcio/grpc/_cython/cygrpc.pxd index b32fa518fc..01e2da6d54 100644 --- a/src/python/grpcio/grpc/_cython/cygrpc.pxd +++ b/src/python/grpcio/grpc/_cython/cygrpc.pxd @@ -25,3 +25,4 @@ include "_cygrpc/records.pxd.pxi" include "_cygrpc/security.pxd.pxi" include "_cygrpc/server.pxd.pxi" include "_cygrpc/tag.pxd.pxi" +include "_cygrpc/time.pxd.pxi" diff --git a/src/python/grpcio/grpc/_cython/cygrpc.pyx b/src/python/grpcio/grpc/_cython/cygrpc.pyx index 5106394708..d8ac84a317 100644 --- a/src/python/grpcio/grpc/_cython/cygrpc.pyx +++ b/src/python/grpcio/grpc/_cython/cygrpc.pyx @@ -32,6 +32,7 @@ include "_cygrpc/records.pyx.pxi" include "_cygrpc/security.pyx.pxi" include "_cygrpc/server.pyx.pxi" include "_cygrpc/tag.pyx.pxi" +include "_cygrpc/time.pyx.pxi" # # initialize gRPC diff --git a/src/python/grpcio/grpc/_server.py b/src/python/grpcio/grpc/_server.py index 1cdb2d45b6..9402941bab 100644 --- a/src/python/grpcio/grpc/_server.py +++ b/src/python/grpcio/grpc/_server.py @@ -220,8 +220,7 @@ class _Context(grpc.ServicerContext): return self._state.client is not _CANCELLED and not self._state.statused def time_remaining(self): - return max( - float(self._rpc_event.call_details.deadline) - time.time(), 0) + return max(self._rpc_event.call_details.deadline - time.time(), 0) def cancel(self): self._rpc_event.call.cancel() diff --git a/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py b/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py index b81d6fbc61..2ca1fa82f4 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py @@ -20,7 +20,6 @@ from grpc._cython import cygrpc from grpc.framework.foundation import logging_pool from tests.unit.framework.common import test_constants -_INFINITE_FUTURE = cygrpc.Timespec(float('+inf')) _EMPTY_FLAGS = 0 _EMPTY_METADATA = () @@ -171,9 +170,9 @@ class CancelManyCallsTest(unittest.TestCase): with client_condition: client_calls = [] for index in range(test_constants.RPC_CONCURRENCY): - client_call = channel.create_call( - None, _EMPTY_FLAGS, client_completion_queue, b'/twinkies', - None, _INFINITE_FUTURE) + client_call = channel.create_call(None, _EMPTY_FLAGS, + client_completion_queue, + b'/twinkies', None, None) operations = ( cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA, _EMPTY_FLAGS), diff --git a/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py b/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py index 4eeb34b92e..c22c77ddbd 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_channel_test.py @@ -31,9 +31,9 @@ def _connectivity_loop(channel, completion_queue): for _ in range(100): connectivity = channel.check_connectivity_state(True) channel.watch_connectivity_state(connectivity, - cygrpc.Timespec(time.time() + 0.2), - completion_queue, None) - completion_queue.poll(deadline=cygrpc.Timespec(float('+inf'))) + time.time() + 0.2, completion_queue, + None) + completion_queue.poll() def _create_loop_destroy(): diff --git a/src/python/grpcio_tests/tests/unit/_cython/_common.py b/src/python/grpcio_tests/tests/unit/_cython/_common.py index ffd226fa95..d4b01ca38b 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_common.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_common.py @@ -20,7 +20,6 @@ from grpc._cython import cygrpc RPC_COUNT = 4000 -INFINITE_FUTURE = cygrpc.Timespec(float('+inf')) EMPTY_FLAGS = 0 INVOCATION_METADATA = ( diff --git a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py index 4ef4ad33e5..7caa98f72d 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py @@ -41,9 +41,9 @@ class Test(_common.RpcTest, unittest.TestCase): server_request_call_tag, }) - client_call = self.channel.create_call( - None, _common.EMPTY_FLAGS, self.client_completion_queue, - b'/twinkies', None, _common.INFINITE_FUTURE) + client_call = self.channel.create_call(None, _common.EMPTY_FLAGS, + self.client_completion_queue, + b'/twinkies', None, None) client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag' client_complete_rpc_tag = 'client_complete_rpc_tag' with self.client_condition: diff --git a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py index 85395c9680..8582a39c01 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py @@ -36,9 +36,9 @@ class Test(_common.RpcTest, unittest.TestCase): server_request_call_tag, }) - client_call = self.channel.create_call( - None, _common.EMPTY_FLAGS, self.client_completion_queue, - b'/twinkies', None, _common.INFINITE_FUTURE) + client_call = self.channel.create_call(None, _common.EMPTY_FLAGS, + self.client_completion_queue, + b'/twinkies', None, None) client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag' client_complete_rpc_tag = 'client_complete_rpc_tag' with self.client_condition: diff --git a/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py b/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py index 82ef25b2a7..ecd23afda7 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py @@ -18,7 +18,6 @@ import unittest from grpc._cython import cygrpc -_INFINITE_FUTURE = cygrpc.Timespec(float('+inf')) _EMPTY_FLAGS = 0 _EMPTY_METADATA = () @@ -156,7 +155,7 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase): client_call = channel.create_call(None, _EMPTY_FLAGS, client_completion_queue, b'/twinkies', - None, _INFINITE_FUTURE) + None, None) client_receive_initial_metadata_tag = 'client_receive_initial_metadata_tag' client_complete_rpc_tag = 'client_complete_rpc_tag' with client_condition: diff --git a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py index 5f9b74ba98..561adf7dff 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py @@ -37,21 +37,6 @@ def _metadata_plugin(context, callback): class TypeSmokeTest(unittest.TestCase): - def testTimespec(self): - now = time.time() - now_timespec_a = cygrpc.Timespec(now) - now_timespec_b = cygrpc.Timespec(now) - self.assertAlmostEqual(now, float(now_timespec_a), places=8) - self.assertEqual(now_timespec_a, now_timespec_b) - self.assertLess(cygrpc.Timespec(now - 1), cygrpc.Timespec(now)) - self.assertGreater(cygrpc.Timespec(now + 1), cygrpc.Timespec(now)) - self.assertGreaterEqual(cygrpc.Timespec(now + 1), cygrpc.Timespec(now)) - self.assertGreaterEqual(cygrpc.Timespec(now), cygrpc.Timespec(now)) - self.assertLessEqual(cygrpc.Timespec(now - 1), cygrpc.Timespec(now)) - self.assertLessEqual(cygrpc.Timespec(now), cygrpc.Timespec(now)) - self.assertNotEqual(cygrpc.Timespec(now - 1), cygrpc.Timespec(now)) - self.assertNotEqual(cygrpc.Timespec(now + 1), cygrpc.Timespec(now)) - def testCompletionQueueUpDown(self): completion_queue = cygrpc.CompletionQueue() del completion_queue @@ -147,7 +132,7 @@ class ServerClientMixin(object): try: call_result = call.start_client_batch(operations, tag) self.assertEqual(cygrpc.CallError.ok, call_result) - event = queue.poll(deadline) + event = queue.poll(deadline=deadline) self.assertEqual(cygrpc.CompletionType.operation_complete, event.completion_type) self.assertTrue(event.success) @@ -176,8 +161,6 @@ class ServerClientMixin(object): RESPONSE = b'his name is robert paulson' METHOD = b'twinkies' - cygrpc_deadline = cygrpc.Timespec(DEADLINE) - server_request_tag = object() request_call_result = self.server.request_call( self.server_completion_queue, self.server_completion_queue, @@ -188,7 +171,7 @@ class ServerClientMixin(object): client_call_tag = object() client_call = self.client_channel.create_call( None, 0, self.client_completion_queue, METHOD, self.host_argument, - cygrpc_deadline) + DEADLINE) client_initial_metadata = ( ( CLIENT_METADATA_ASCII_KEY, @@ -210,9 +193,9 @@ class ServerClientMixin(object): ], client_call_tag) self.assertEqual(cygrpc.CallError.ok, client_start_batch_result) client_event_future = test_utilities.CompletionQueuePollFuture( - self.client_completion_queue, cygrpc_deadline) + self.client_completion_queue, DEADLINE) - request_event = self.server_completion_queue.poll(cygrpc_deadline) + request_event = self.server_completion_queue.poll(deadline=DEADLINE) self.assertEqual(cygrpc.CompletionType.operation_complete, request_event.completion_type) self.assertIsInstance(request_event.call, cygrpc.Call) @@ -223,7 +206,7 @@ class ServerClientMixin(object): self.assertEqual(METHOD, request_event.call_details.method) self.assertEqual(self.expected_host, request_event.call_details.host) self.assertLess( - abs(DEADLINE - float(request_event.call_details.deadline)), + abs(DEADLINE - request_event.call_details.deadline), DEADLINE_TOLERANCE) server_call_tag = object() @@ -248,7 +231,7 @@ class ServerClientMixin(object): ], server_call_tag) self.assertEqual(cygrpc.CallError.ok, server_start_batch_result) - server_event = self.server_completion_queue.poll(cygrpc_deadline) + server_event = self.server_completion_queue.poll(deadline=DEADLINE) client_event = client_event_future.result() self.assertEqual(6, len(client_event.batch_operations)) @@ -310,7 +293,6 @@ class ServerClientMixin(object): DEADLINE_TOLERANCE = 0.25 METHOD = b'twinkies' - cygrpc_deadline = cygrpc.Timespec(DEADLINE) empty_metadata = () server_request_tag = object() @@ -319,26 +301,26 @@ class ServerClientMixin(object): server_request_tag) client_call = self.client_channel.create_call( None, 0, self.client_completion_queue, METHOD, self.host_argument, - cygrpc_deadline) + DEADLINE) # Prologue def perform_client_operations(operations, description): return self._perform_operations(operations, client_call, self.client_completion_queue, - cygrpc_deadline, description) + DEADLINE, description) client_event_future = perform_client_operations([ cygrpc.SendInitialMetadataOperation(empty_metadata, _EMPTY_FLAGS), cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS), ], "Client prologue") - request_event = self.server_completion_queue.poll(cygrpc_deadline) + request_event = self.server_completion_queue.poll(deadline=DEADLINE) server_call = request_event.call def perform_server_operations(operations, description): return self._perform_operations(operations, server_call, self.server_completion_queue, - cygrpc_deadline, description) + DEADLINE, description) server_event_future = perform_server_operations([ cygrpc.SendInitialMetadataOperation(empty_metadata, _EMPTY_FLAGS), diff --git a/src/python/grpcio_tests/tests/unit/_cython/test_utilities.py b/src/python/grpcio_tests/tests/unit/_cython/test_utilities.py index 8e91161f80..4a00b9ef2f 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/test_utilities.py +++ b/src/python/grpcio_tests/tests/unit/_cython/test_utilities.py @@ -49,4 +49,4 @@ class CompletionQueuePollFuture(SimpleFuture): def __init__(self, completion_queue, deadline): super(CompletionQueuePollFuture, - self).__init__(lambda: completion_queue.poll(deadline)) + self).__init__(lambda: completion_queue.poll(deadline=deadline)) diff --git a/test/core/support/vector_test.cc b/test/core/support/vector_test.cc index aad9f3be90..82607a1b26 100644 --- a/test/core/support/vector_test.cc +++ b/test/core/support/vector_test.cc @@ -18,18 +18,50 @@ #include "src/core/lib/support/vector.h" #include <gtest/gtest.h> +#include "src/core/lib/support/memory.h" #include "test/core/util/test_config.h" namespace grpc_core { namespace testing { TEST(InlinedVectorTest, CreateAndIterate) { - InlinedVector<int, 1> v{1, 2, 3}; - int sum = 0; - for (auto i : v) { - sum += i; + const int kNumElements = 9; + InlinedVector<int, 2> v; + for (int i = 0; i < kNumElements; ++i) { + v.push_back(i); } - EXPECT_EQ(6, sum); + EXPECT_EQ(static_cast<size_t>(kNumElements), v.size()); + for (int i = 0; i < kNumElements; ++i) { + EXPECT_EQ(i, v[i]); + } +} + +TEST(InlinedVectorTest, ValuesAreInlined) { + const int kNumElements = 5; + InlinedVector<int, 10> v; + for (int i = 0; i < kNumElements; ++i) { + v.push_back(i); + } + EXPECT_EQ(static_cast<size_t>(kNumElements), v.size()); + for (int i = 0; i < kNumElements; ++i) { + EXPECT_EQ(i, v[i]); + } +} + +TEST(InlinedVectorTest, PushBackWithMove) { + InlinedVector<UniquePtr<int>, 1> v; + UniquePtr<int> i = MakeUnique<int>(3); + v.push_back(std::move(i)); + EXPECT_EQ(nullptr, i.get()); + EXPECT_EQ(1UL, v.size()); + EXPECT_EQ(3, *v[0]); +} + +TEST(InlinedVectorTest, EmplaceBack) { + InlinedVector<UniquePtr<int>, 1> v; + v.emplace_back(New<int>(3)); + EXPECT_EQ(1UL, v.size()); + EXPECT_EQ(3, *v[0]); } } // namespace testing diff --git a/test/core/surface/public_headers_must_be_c89.c b/test/core/surface/public_headers_must_be_c89.c index 8d2384ba61..7fd36a241a 100644 --- a/test/core/surface/public_headers_must_be_c89.c +++ b/test/core/surface/public_headers_must_be_c89.c @@ -29,7 +29,6 @@ #include <grpc/impl/codegen/byte_buffer_reader.h> #include <grpc/impl/codegen/compression_types.h> #include <grpc/impl/codegen/connectivity_state.h> -#include <grpc/impl/codegen/exec_ctx_fwd.h> #include <grpc/impl/codegen/fork.h> #include <grpc/impl/codegen/gpr_slice.h> #include <grpc/impl/codegen/gpr_types.h> diff --git a/tools/doxygen/Doxyfile.c++ b/tools/doxygen/Doxyfile.c++ index e62278cb9f..5bdbcd71f5 100644 --- a/tools/doxygen/Doxyfile.c++ +++ b/tools/doxygen/Doxyfile.c++ @@ -887,7 +887,6 @@ include/grpc/impl/codegen/byte_buffer.h \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ -include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/fork.h \ include/grpc/impl/codegen/gpr_slice.h \ include/grpc/impl/codegen/gpr_types.h \ diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index d09b325c97..d9184f49a2 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -888,7 +888,6 @@ include/grpc/impl/codegen/byte_buffer.h \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ -include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/fork.h \ include/grpc/impl/codegen/gpr_slice.h \ include/grpc/impl/codegen/gpr_types.h \ @@ -1047,6 +1046,7 @@ src/core/lib/support/string_windows.h \ src/core/lib/support/thd_internal.h \ src/core/lib/support/time_precise.h \ src/core/lib/support/tmpfile.h \ +src/core/lib/support/vector.h \ src/core/lib/surface/alarm_internal.h \ src/core/lib/surface/api_trace.h \ src/core/lib/surface/call.h \ diff --git a/tools/doxygen/Doxyfile.core b/tools/doxygen/Doxyfile.core index 6ce9041747..916d3b1e49 100644 --- a/tools/doxygen/Doxyfile.core +++ b/tools/doxygen/Doxyfile.core @@ -818,7 +818,6 @@ include/grpc/impl/codegen/byte_buffer.h \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ -include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/fork.h \ include/grpc/impl/codegen/fork.h \ include/grpc/impl/codegen/gpr_slice.h \ diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 1aff0075a6..3d3c6711d0 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -818,7 +818,6 @@ include/grpc/impl/codegen/byte_buffer.h \ include/grpc/impl/codegen/byte_buffer_reader.h \ include/grpc/impl/codegen/compression_types.h \ include/grpc/impl/codegen/connectivity_state.h \ -include/grpc/impl/codegen/exec_ctx_fwd.h \ include/grpc/impl/codegen/fork.h \ include/grpc/impl/codegen/fork.h \ include/grpc/impl/codegen/gpr_slice.h \ @@ -1333,6 +1332,7 @@ src/core/lib/support/tmpfile.h \ src/core/lib/support/tmpfile_msys.cc \ src/core/lib/support/tmpfile_posix.cc \ src/core/lib/support/tmpfile_windows.cc \ +src/core/lib/support/vector.h \ src/core/lib/support/wrap_memcpy.cc \ src/core/lib/surface/README.md \ src/core/lib/surface/alarm.cc \ diff --git a/tools/interop_matrix/client_matrix.py b/tools/interop_matrix/client_matrix.py index 71d3a79023..59231096b9 100644 --- a/tools/interop_matrix/client_matrix.py +++ b/tools/interop_matrix/client_matrix.py @@ -103,6 +103,9 @@ LANG_RELEASE_MATRIX = { { 'v1.8.2': None }, + { + 'v1.9.1': None + }, ], 'java': [ { diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index d432bd0e53..51f0ac7ca6 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -4268,6 +4268,25 @@ "gpr_test_util", "grpc", "grpc++", + "grpc++_test", + "grpc_test_util" + ], + "headers": [], + "is_filegroup": false, + "language": "c++", + "name": "vector_test", + "src": [ + "test/core/support/vector_test.cc" + ], + "third_party": false, + "type": "target" + }, + { + "deps": [ + "gpr", + "gpr_test_util", + "grpc", + "grpc++", "grpc++_test_util", "grpc_test_util" ], @@ -8250,6 +8269,7 @@ "src/core/lib/support/debug_location.h", "src/core/lib/support/ref_counted.h", "src/core/lib/support/ref_counted_ptr.h", + "src/core/lib/support/vector.h", "src/core/lib/surface/alarm_internal.h", "src/core/lib/surface/api_trace.h", "src/core/lib/surface/call.h", @@ -8389,6 +8409,7 @@ "src/core/lib/support/debug_location.h", "src/core/lib/support/ref_counted.h", "src/core/lib/support/ref_counted_ptr.h", + "src/core/lib/support/vector.h", "src/core/lib/surface/alarm_internal.h", "src/core/lib/surface/api_trace.h", "src/core/lib/surface/call.h", @@ -8505,7 +8526,6 @@ "include/grpc/impl/codegen/byte_buffer_reader.h", "include/grpc/impl/codegen/compression_types.h", "include/grpc/impl/codegen/connectivity_state.h", - "include/grpc/impl/codegen/exec_ctx_fwd.h", "include/grpc/impl/codegen/grpc_types.h", "include/grpc/impl/codegen/propagation_bits.h", "include/grpc/impl/codegen/slice.h", @@ -8519,7 +8539,6 @@ "include/grpc/impl/codegen/byte_buffer_reader.h", "include/grpc/impl/codegen/compression_types.h", "include/grpc/impl/codegen/connectivity_state.h", - "include/grpc/impl/codegen/exec_ctx_fwd.h", "include/grpc/impl/codegen/grpc_types.h", "include/grpc/impl/codegen/propagation_bits.h", "include/grpc/impl/codegen/slice.h", diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index 98517cba2e..6b83cecd41 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -4510,6 +4510,30 @@ "ci_platforms": [ "linux", "mac", + "posix", + "windows" + ], + "cpu_cost": 1.0, + "exclude_configs": [], + "exclude_iomgrs": [], + "flaky": false, + "gtest": true, + "language": "c++", + "name": "vector_test", + "platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "uses_polling": true + }, + { + "args": [], + "benchmark": false, + "ci_platforms": [ + "linux", + "mac", "posix" ], "cpu_cost": 0.5, |