Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

[party] Make party a concrete type (#37015)

Browse Source 
Move event engine context to be an arena based context, and have party host arena directly.

Now `Party` can directly inject all context necessary without needing to call out to the derived type; this makes `Party` instantiable by itself (no need to inherit from it - though we'll still want to for `CallSpine` to simplify some arena lifetime management), meaning we can start to consider eliminating the base class `Activity` and merging these types.

Closes #37015

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/37015 from ctiller:chaotic-party 7f44b37b5d
PiperOrigin-RevId: 646963003
pull/37036/head^2
Craig Tiller 5 months ago committed by Copybara-Service
parent c901884627
commit 678ee0a45d
44 changed files with 343 additions and 1227 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 262
      CMakeLists.txt
  2. 651
      build_autogenerated.yaml
  3. 16
      src/core/BUILD
  4. 17
      src/core/client_channel/client_channel.cc
  5. 7
      src/core/client_channel/direct_channel.cc
  6. 11
      src/core/ext/filters/channel_idle/legacy_channel_idle_filter.cc
  7. 8
      src/core/ext/transport/chaotic_good/client/chaotic_good_connector.cc
  8. 4
      src/core/ext/transport/chaotic_good/server/chaotic_good_server.cc
  9. 7
      src/core/ext/transport/chaotic_good/server_transport.cc
  10. 7
      src/core/ext/transport/inproc/inproc_transport.cc
  11. 6
      src/core/lib/channel/promise_based_filter.cc
  12. 16
      src/core/lib/channel/promise_based_filter.h
  13. 6
      src/core/lib/event_engine/event_engine_context.h
  14. 1
      src/core/lib/promise/activity.h
  15. 40
      src/core/lib/promise/party.cc
  16. 38
      src/core/lib/promise/party.h
  17. 26
      src/core/lib/surface/call.cc
  18. 11
      src/core/lib/surface/call.h
  19. 39
      src/core/lib/surface/client_call.cc
  20. 16
      src/core/lib/surface/client_call.h
  21. 5
      src/core/lib/surface/filter_stack_call.cc
  22. 2
      src/core/lib/surface/server_call.h
  23. 13
      src/core/lib/transport/call_spine.cc
  24. 57
      src/core/lib/transport/call_spine.h
  25. 3
      src/core/lib/transport/interception_chain.cc
  26. 7
      test/core/call/bm_client_call.cc
  27. 12
      test/core/call/client_call_test.cc
  28. 6
      test/core/call/server_call_test.cc
  29. 8
      test/core/call/yodel/yodel_test.h
  30. 10
      test/core/client_channel/client_channel_test.cc
  31. 7
      test/core/client_channel/connected_subchannel_test.cc
  32. 6
      test/core/client_channel/load_balanced_call_destination_test.cc
  33. 45
      test/core/promise/bm_party.cc
  34. 93
      test/core/promise/party_test.cc
  35. 20
      test/core/promise/sleep_test.cc
  36. 8
      test/core/surface/channel_init_test.cc
  37. 2
      test/core/transport/BUILD
  38. 21
      test/core/transport/bm_call_spine.cc
  39. 16
      test/core/transport/call_spine_benchmarks.h
  40. 7
      test/core/transport/call_spine_test.cc
  41. 7
      test/core/transport/chaotic_good/client_transport_error_test.cc
  42. 9
      test/core/transport/chaotic_good/transport_test.h
  43. 16
      test/core/transport/interception_chain_test.cc
  44. 1
      tools/distrib/fix_build_deps.py

262
CMakeLists.txt generated
Unescape View File

@ -18659,256 +18659,14 @@ endif()
if(gRPC_BUILD_TESTS)
add_executable(interception_chain_test
src/core/channelz/channel_trace.cc
src/core/channelz/channelz.cc
src/core/channelz/channelz_registry.cc
src/core/ext/upb-gen/google/protobuf/any.upb_minitable.c
src/core/ext/upb-gen/google/rpc/status.upb_minitable.c
src/core/ext/upb-gen/src/proto/grpc/gcp/altscontext.upb_minitable.c
src/core/ext/upb-gen/src/proto/grpc/gcp/handshaker.upb_minitable.c
src/core/ext/upb-gen/src/proto/grpc/gcp/transport_security_common.upb_minitable.c
src/core/handshaker/handshaker_registry.cc
src/core/handshaker/proxy_mapper_registry.cc
src/core/lib/address_utils/parse_address.cc
src/core/lib/address_utils/sockaddr_utils.cc
src/core/lib/backoff/backoff.cc
src/core/lib/channel/channel_args.cc
src/core/lib/channel/channel_args_preconditioning.cc
src/core/lib/channel/channel_stack.cc
src/core/lib/channel/channel_stack_builder.cc
src/core/lib/channel/channel_stack_builder_impl.cc
src/core/lib/channel/connected_channel.cc
src/core/lib/channel/promise_based_filter.cc
src/core/lib/channel/status_util.cc
src/core/lib/compression/compression.cc
src/core/lib/compression/compression_internal.cc
src/core/lib/compression/message_compress.cc
src/core/lib/config/core_configuration.cc
src/core/lib/debug/event_log.cc
src/core/lib/debug/trace.cc
src/core/lib/debug/trace_flags.cc
src/core/lib/event_engine/ares_resolver.cc
src/core/lib/event_engine/cf_engine/cf_engine.cc
src/core/lib/event_engine/cf_engine/cfstream_endpoint.cc
src/core/lib/event_engine/cf_engine/dns_service_resolver.cc
src/core/lib/event_engine/channel_args_endpoint_config.cc
src/core/lib/event_engine/default_event_engine.cc
src/core/lib/event_engine/default_event_engine_factory.cc
src/core/lib/event_engine/event_engine.cc
src/core/lib/event_engine/forkable.cc
src/core/lib/event_engine/posix_engine/ev_epoll1_linux.cc
src/core/lib/event_engine/posix_engine/ev_poll_posix.cc
src/core/lib/event_engine/posix_engine/event_poller_posix_default.cc
src/core/lib/event_engine/posix_engine/internal_errqueue.cc
src/core/lib/event_engine/posix_engine/lockfree_event.cc
src/core/lib/event_engine/posix_engine/native_posix_dns_resolver.cc
src/core/lib/event_engine/posix_engine/posix_endpoint.cc
src/core/lib/event_engine/posix_engine/posix_engine.cc
src/core/lib/event_engine/posix_engine/posix_engine_listener.cc
src/core/lib/event_engine/posix_engine/posix_engine_listener_utils.cc
src/core/lib/event_engine/posix_engine/tcp_socket_utils.cc
src/core/lib/event_engine/posix_engine/timer.cc
src/core/lib/event_engine/posix_engine/timer_heap.cc
src/core/lib/event_engine/posix_engine/timer_manager.cc
src/core/lib/event_engine/posix_engine/traced_buffer_list.cc
src/core/lib/event_engine/posix_engine/wakeup_fd_eventfd.cc
src/core/lib/event_engine/posix_engine/wakeup_fd_pipe.cc
src/core/lib/event_engine/posix_engine/wakeup_fd_posix_default.cc
src/core/lib/event_engine/resolved_address.cc
src/core/lib/event_engine/shim.cc
src/core/lib/event_engine/slice.cc
src/core/lib/event_engine/slice_buffer.cc
src/core/lib/event_engine/tcp_socket_utils.cc
src/core/lib/event_engine/thread_pool/thread_count.cc
src/core/lib/event_engine/thread_pool/thread_pool_factory.cc
src/core/lib/event_engine/thread_pool/work_stealing_thread_pool.cc
src/core/lib/event_engine/thready_event_engine/thready_event_engine.cc
src/core/lib/event_engine/time_util.cc
src/core/lib/event_engine/utils.cc
src/core/lib/event_engine/windows/grpc_polled_fd_windows.cc
src/core/lib/event_engine/windows/iocp.cc
src/core/lib/event_engine/windows/native_windows_dns_resolver.cc
src/core/lib/event_engine/windows/win_socket.cc
src/core/lib/event_engine/windows/windows_endpoint.cc
src/core/lib/event_engine/windows/windows_engine.cc
src/core/lib/event_engine/windows/windows_listener.cc
src/core/lib/event_engine/work_queue/basic_work_queue.cc
src/core/lib/experiments/config.cc
src/core/lib/experiments/experiments.cc
src/core/lib/gprpp/dump_args.cc
src/core/lib/gprpp/glob.cc
src/core/lib/gprpp/load_file.cc
src/core/lib/gprpp/per_cpu.cc
src/core/lib/gprpp/ref_counted_string.cc
src/core/lib/gprpp/status_helper.cc
src/core/lib/gprpp/time.cc
src/core/lib/gprpp/time_averaged_stats.cc
src/core/lib/gprpp/validation_errors.cc
src/core/lib/gprpp/work_serializer.cc
src/core/lib/iomgr/buffer_list.cc
src/core/lib/iomgr/call_combiner.cc
src/core/lib/iomgr/cfstream_handle.cc
src/core/lib/iomgr/closure.cc
src/core/lib/iomgr/combiner.cc
src/core/lib/iomgr/dualstack_socket_posix.cc
src/core/lib/iomgr/endpoint.cc
src/core/lib/iomgr/endpoint_cfstream.cc
src/core/lib/iomgr/endpoint_pair_posix.cc
src/core/lib/iomgr/endpoint_pair_windows.cc
src/core/lib/iomgr/error.cc
src/core/lib/iomgr/error_cfstream.cc
src/core/lib/iomgr/ev_apple.cc
src/core/lib/iomgr/ev_epoll1_linux.cc
src/core/lib/iomgr/ev_poll_posix.cc
src/core/lib/iomgr/ev_posix.cc
src/core/lib/iomgr/event_engine_shims/closure.cc
src/core/lib/iomgr/event_engine_shims/endpoint.cc
src/core/lib/iomgr/event_engine_shims/tcp_client.cc
src/core/lib/iomgr/exec_ctx.cc
src/core/lib/iomgr/executor.cc
src/core/lib/iomgr/fork_posix.cc
src/core/lib/iomgr/fork_windows.cc
src/core/lib/iomgr/gethostname_fallback.cc
src/core/lib/iomgr/gethostname_host_name_max.cc
src/core/lib/iomgr/gethostname_sysconf.cc
src/core/lib/iomgr/grpc_if_nametoindex_posix.cc
src/core/lib/iomgr/grpc_if_nametoindex_unsupported.cc
src/core/lib/iomgr/internal_errqueue.cc
src/core/lib/iomgr/iocp_windows.cc
src/core/lib/iomgr/iomgr.cc
src/core/lib/iomgr/iomgr_internal.cc
src/core/lib/iomgr/iomgr_posix.cc
src/core/lib/iomgr/iomgr_posix_cfstream.cc
src/core/lib/iomgr/iomgr_windows.cc
src/core/lib/iomgr/lockfree_event.cc
src/core/lib/iomgr/polling_entity.cc
src/core/lib/iomgr/pollset.cc
src/core/lib/iomgr/pollset_set.cc
src/core/lib/iomgr/pollset_set_windows.cc
src/core/lib/iomgr/pollset_windows.cc
src/core/lib/iomgr/resolve_address.cc
src/core/lib/iomgr/resolve_address_posix.cc
src/core/lib/iomgr/resolve_address_windows.cc
src/core/lib/iomgr/sockaddr_utils_posix.cc
src/core/lib/iomgr/socket_factory_posix.cc
src/core/lib/iomgr/socket_mutator.cc
src/core/lib/iomgr/socket_utils_common_posix.cc
src/core/lib/iomgr/socket_utils_linux.cc
src/core/lib/iomgr/socket_utils_posix.cc
src/core/lib/iomgr/socket_utils_windows.cc
src/core/lib/iomgr/socket_windows.cc
src/core/lib/iomgr/systemd_utils.cc
src/core/lib/iomgr/tcp_client.cc
src/core/lib/iomgr/tcp_client_cfstream.cc
src/core/lib/iomgr/tcp_client_posix.cc
src/core/lib/iomgr/tcp_client_windows.cc
src/core/lib/iomgr/tcp_posix.cc
src/core/lib/iomgr/tcp_server.cc
src/core/lib/iomgr/tcp_server_posix.cc
src/core/lib/iomgr/tcp_server_utils_posix_common.cc
src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.cc
src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.cc
src/core/lib/iomgr/tcp_server_windows.cc
src/core/lib/iomgr/tcp_windows.cc
src/core/lib/iomgr/timer.cc
src/core/lib/iomgr/timer_generic.cc
src/core/lib/iomgr/timer_heap.cc
src/core/lib/iomgr/timer_manager.cc
src/core/lib/iomgr/unix_sockets_posix.cc
src/core/lib/iomgr/unix_sockets_posix_noop.cc
src/core/lib/iomgr/vsock.cc
src/core/lib/iomgr/wakeup_fd_eventfd.cc
src/core/lib/iomgr/wakeup_fd_nospecial.cc
src/core/lib/iomgr/wakeup_fd_pipe.cc
src/core/lib/iomgr/wakeup_fd_posix.cc
src/core/lib/promise/activity.cc
src/core/lib/promise/party.cc
src/core/lib/resource_quota/api.cc
src/core/lib/resource_quota/arena.cc
src/core/lib/resource_quota/connection_quota.cc
src/core/lib/resource_quota/memory_quota.cc
src/core/lib/resource_quota/periodic_update.cc
src/core/lib/resource_quota/resource_quota.cc
src/core/lib/resource_quota/thread_quota.cc
src/core/lib/security/certificate_provider/certificate_provider_registry.cc
src/core/lib/security/credentials/alts/check_gcp_environment.cc
src/core/lib/security/credentials/alts/check_gcp_environment_linux.cc
src/core/lib/security/credentials/alts/check_gcp_environment_no_op.cc
src/core/lib/security/credentials/alts/check_gcp_environment_windows.cc
src/core/lib/security/credentials/alts/grpc_alts_credentials_client_options.cc
src/core/lib/security/credentials/alts/grpc_alts_credentials_options.cc
src/core/lib/security/credentials/alts/grpc_alts_credentials_server_options.cc
src/core/lib/slice/percent_encoding.cc
src/core/lib/slice/slice.cc
src/core/lib/slice/slice_buffer.cc
src/core/lib/slice/slice_string_helpers.cc
src/core/lib/surface/byte_buffer.cc
src/core/lib/surface/byte_buffer_reader.cc
src/core/lib/surface/call.cc
src/core/lib/surface/call_details.cc
src/core/lib/surface/call_log_batch.cc
src/core/lib/surface/call_utils.cc
src/core/lib/surface/channel.cc
src/core/lib/surface/channel_init.cc
src/core/lib/surface/channel_stack_type.cc
src/core/lib/surface/client_call.cc
src/core/lib/surface/completion_queue.cc
src/core/lib/surface/completion_queue_factory.cc
src/core/lib/surface/event_string.cc
src/core/lib/surface/filter_stack_call.cc
src/core/lib/surface/init_internally.cc
src/core/lib/surface/lame_client.cc
src/core/lib/surface/metadata_array.cc
src/core/lib/surface/server_call.cc
src/core/lib/surface/validate_metadata.cc
src/core/lib/surface/version.cc
src/core/lib/transport/call_arena_allocator.cc
src/core/lib/transport/call_filters.cc
src/core/lib/transport/call_final_info.cc
src/core/lib/transport/call_spine.cc
src/core/lib/transport/call_state.cc
src/core/lib/transport/connectivity_state.cc
src/core/lib/transport/error_utils.cc
src/core/lib/transport/interception_chain.cc
src/core/lib/transport/message.cc
src/core/lib/transport/metadata.cc
src/core/lib/transport/metadata_batch.cc
src/core/lib/transport/parsed_metadata.cc
src/core/lib/transport/status_conversion.cc
src/core/lib/transport/timeout_encoding.cc
src/core/lib/transport/transport.cc
src/core/lib/transport/transport_op_string.cc
src/core/lib/uri/uri_parser.cc
src/core/load_balancing/lb_policy.cc
src/core/load_balancing/lb_policy_registry.cc
src/core/resolver/endpoint_addresses.cc
src/core/resolver/resolver.cc
src/core/resolver/resolver_registry.cc
src/core/service_config/service_config_parser.cc
src/core/telemetry/call_tracer.cc
src/core/telemetry/histogram_view.cc
src/core/telemetry/metrics.cc
src/core/telemetry/stats.cc
src/core/telemetry/stats_data.cc
src/core/tsi/alts/handshaker/transport_security_common_api.cc
src/core/util/json/json_writer.cc
test/core/transport/interception_chain_test.cc
third_party/upb/upb/mini_descriptor/build_enum.c
third_party/upb/upb/mini_descriptor/decode.c
third_party/upb/upb/mini_descriptor/internal/base92.c
third_party/upb/upb/mini_descriptor/internal/encode.c
third_party/upb/upb/mini_descriptor/link.c
third_party/upb/upb/wire/decode.c
third_party/upb/upb/wire/encode.c
third_party/upb/upb/wire/eps_copy_input_stream.c
third_party/upb/upb/wire/internal/decode_fast.c
third_party/upb/upb/wire/reader.c
)
if(WIN32 AND MSVC)
if(BUILD_SHARED_LIBS)
target_compile_definitions(interception_chain_test
PRIVATE
"GPR_DLL_IMPORTS"
"GRPC_DLL_IMPORTS"
)
endif()
endif()
@ -18935,23 +18693,7 @@ target_include_directories(interception_chain_test
target_link_libraries(interception_chain_test
${_gRPC_ALLTARGETS_LIBRARIES}
gtest
utf8_range_lib
upb_message_lib
${_gRPC_ZLIB_LIBRARIES}
absl::config
absl::no_destructor
absl::cleanup
absl::flat_hash_map
absl::inlined_vector
absl::function_ref
absl::hash
absl::type_traits
absl::statusor
absl::span
absl::utility
${_gRPC_CARES_LIBRARIES}
gpr
${_gRPC_ADDRESS_SORTING_LIBRARIES}
grpc
)

651
build_autogenerated.yaml generated
Unescape View File

@ -5353,7 +5353,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/bitset.h
- src/core/lib/gprpp/down_cast.h
@ -6527,7 +6526,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -6798,7 +6796,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/down_cast.h
- src/core/lib/gprpp/dump_args.h
@ -7789,7 +7786,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -8609,7 +8605,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -10196,7 +10191,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -10756,7 +10750,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -10890,7 +10883,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -12535,637 +12527,51 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/down_cast.h
- src/core/lib/gprpp/dump_args.h
- src/core/lib/gprpp/glob.h
- src/core/lib/gprpp/orphanable.h
- src/core/lib/gprpp/ref_counted.h
- src/core/lib/gprpp/ref_counted_ptr.h
- src/core/lib/promise/activity.h
- src/core/lib/promise/context.h
- src/core/lib/promise/detail/basic_seq.h
- src/core/lib/promise/detail/promise_factory.h
- src/core/lib/promise/detail/promise_like.h
- src/core/lib/promise/detail/seq_state.h
- src/core/lib/promise/detail/status.h
- src/core/lib/promise/inter_activity_pipe.h
- src/core/lib/promise/poll.h
- src/core/lib/promise/seq.h
- test/core/promise/test_wakeup_schedulers.h
src:
- src/core/lib/debug/trace.cc
- src/core/lib/debug/trace_flags.cc
- src/core/lib/gprpp/dump_args.cc
- src/core/lib/gprpp/glob.cc
- src/core/lib/promise/activity.cc
- test/core/promise/inter_activity_pipe_test.cc
deps:
- gtest
- absl/base:config
- absl/container:flat_hash_map
- absl/hash:hash
- absl/meta:type_traits
- absl/status:statusor
- gpr
uses_polling: false
- name: interception_chain_test
gtest: true
build: test
language: c++
headers:
- src/core/channelz/channel_trace.h
- src/core/channelz/channelz.h
- src/core/channelz/channelz_registry.h
- src/core/ext/upb-gen/google/protobuf/any.upb.h
- src/core/ext/upb-gen/google/protobuf/any.upb_minitable.h
- src/core/ext/upb-gen/google/rpc/status.upb.h
- src/core/ext/upb-gen/google/rpc/status.upb_minitable.h
- src/core/ext/upb-gen/src/proto/grpc/gcp/altscontext.upb.h
- src/core/ext/upb-gen/src/proto/grpc/gcp/altscontext.upb_minitable.h
- src/core/ext/upb-gen/src/proto/grpc/gcp/handshaker.upb.h
- src/core/ext/upb-gen/src/proto/grpc/gcp/handshaker.upb_minitable.h
- src/core/ext/upb-gen/src/proto/grpc/gcp/transport_security_common.upb.h
- src/core/ext/upb-gen/src/proto/grpc/gcp/transport_security_common.upb_minitable.h
- src/core/handshaker/handshaker_factory.h
- src/core/handshaker/handshaker_registry.h
- src/core/handshaker/proxy_mapper.h
- src/core/handshaker/proxy_mapper_registry.h
- src/core/lib/address_utils/parse_address.h
- src/core/lib/address_utils/sockaddr_utils.h
- src/core/lib/avl/avl.h
- src/core/lib/backoff/backoff.h
- src/core/lib/channel/call_finalization.h
- src/core/lib/channel/channel_args.h
- src/core/lib/channel/channel_args_preconditioning.h
- src/core/lib/channel/channel_fwd.h
- src/core/lib/channel/channel_stack.h
- src/core/lib/channel/channel_stack_builder.h
- src/core/lib/channel/channel_stack_builder_impl.h
- src/core/lib/channel/connected_channel.h
- src/core/lib/channel/promise_based_filter.h
- src/core/lib/channel/status_util.h
- src/core/lib/compression/compression_internal.h
- src/core/lib/compression/message_compress.h
- src/core/lib/config/core_configuration.h
- src/core/lib/debug/event_log.h
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/ares_resolver.h
- src/core/lib/event_engine/cf_engine/cf_engine.h
- src/core/lib/event_engine/cf_engine/cfstream_endpoint.h
- src/core/lib/event_engine/cf_engine/cftype_unique_ref.h
- src/core/lib/event_engine/cf_engine/dns_service_resolver.h
- src/core/lib/event_engine/channel_args_endpoint_config.h
- src/core/lib/event_engine/common_closures.h
- src/core/lib/event_engine/default_event_engine.h
- src/core/lib/event_engine/default_event_engine_factory.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/event_engine/extensions/can_track_errors.h
- src/core/lib/event_engine/extensions/chaotic_good_extension.h
- src/core/lib/event_engine/extensions/supports_fd.h
- src/core/lib/event_engine/extensions/tcp_trace.h
- src/core/lib/event_engine/forkable.h
- src/core/lib/event_engine/grpc_polled_fd.h
- src/core/lib/event_engine/handle_containers.h
- src/core/lib/event_engine/memory_allocator_factory.h
- src/core/lib/event_engine/nameser.h
- src/core/lib/event_engine/poller.h
- src/core/lib/event_engine/posix.h
- src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h
- src/core/lib/event_engine/posix_engine/ev_poll_posix.h
- src/core/lib/event_engine/posix_engine/event_poller.h
- src/core/lib/event_engine/posix_engine/event_poller_posix_default.h
- src/core/lib/event_engine/posix_engine/grpc_polled_fd_posix.h
- src/core/lib/event_engine/posix_engine/internal_errqueue.h
- src/core/lib/event_engine/posix_engine/lockfree_event.h
- src/core/lib/event_engine/posix_engine/native_posix_dns_resolver.h
- src/core/lib/event_engine/posix_engine/posix_endpoint.h
- src/core/lib/event_engine/posix_engine/posix_engine.h
- src/core/lib/event_engine/posix_engine/posix_engine_closure.h
- src/core/lib/event_engine/posix_engine/posix_engine_listener.h
- src/core/lib/event_engine/posix_engine/posix_engine_listener_utils.h
- src/core/lib/event_engine/posix_engine/tcp_socket_utils.h
- src/core/lib/event_engine/posix_engine/timer.h
- src/core/lib/event_engine/posix_engine/timer_heap.h
- src/core/lib/event_engine/posix_engine/timer_manager.h
- src/core/lib/event_engine/posix_engine/traced_buffer_list.h
- src/core/lib/event_engine/posix_engine/wakeup_fd_eventfd.h
- src/core/lib/event_engine/posix_engine/wakeup_fd_pipe.h
- src/core/lib/event_engine/posix_engine/wakeup_fd_posix.h
- src/core/lib/event_engine/posix_engine/wakeup_fd_posix_default.h
- src/core/lib/event_engine/query_extensions.h
- src/core/lib/event_engine/ref_counted_dns_resolver_interface.h
- src/core/lib/event_engine/resolved_address_internal.h
- src/core/lib/event_engine/shim.h
- src/core/lib/event_engine/tcp_socket_utils.h
- src/core/lib/event_engine/thread_pool/thread_count.h
- src/core/lib/event_engine/thread_pool/thread_pool.h
- src/core/lib/event_engine/thread_pool/work_stealing_thread_pool.h
- src/core/lib/event_engine/thready_event_engine/thready_event_engine.h
- src/core/lib/event_engine/time_util.h
- src/core/lib/event_engine/utils.h
- src/core/lib/event_engine/windows/grpc_polled_fd_windows.h
- src/core/lib/event_engine/windows/iocp.h
- src/core/lib/event_engine/windows/native_windows_dns_resolver.h
- src/core/lib/event_engine/windows/win_socket.h
- src/core/lib/event_engine/windows/windows_endpoint.h
- src/core/lib/event_engine/windows/windows_engine.h
- src/core/lib/event_engine/windows/windows_listener.h
- src/core/lib/event_engine/work_queue/basic_work_queue.h
- src/core/lib/event_engine/work_queue/work_queue.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/bitset.h
- src/core/lib/gprpp/chunked_vector.h
- src/core/lib/gprpp/cpp_impl_of.h
- src/core/lib/gprpp/down_cast.h
- src/core/lib/gprpp/dual_ref_counted.h
- src/core/lib/gprpp/dump_args.h
- src/core/lib/gprpp/glob.h
- src/core/lib/gprpp/if_list.h
- src/core/lib/gprpp/load_file.h
- src/core/lib/gprpp/manual_constructor.h
- src/core/lib/gprpp/match.h
- src/core/lib/gprpp/notification.h
- src/core/lib/gprpp/orphanable.h
- src/core/lib/gprpp/overload.h
- src/core/lib/gprpp/packed_table.h
- src/core/lib/gprpp/per_cpu.h
- src/core/lib/gprpp/ref_counted.h
- src/core/lib/gprpp/ref_counted_ptr.h
- src/core/lib/gprpp/ref_counted_string.h
- src/core/lib/gprpp/single_set_ptr.h
- src/core/lib/gprpp/sorted_pack.h
- src/core/lib/gprpp/status_helper.h
- src/core/lib/gprpp/table.h
- src/core/lib/gprpp/time.h
- src/core/lib/gprpp/time_averaged_stats.h
- src/core/lib/gprpp/type_list.h
- src/core/lib/gprpp/unique_type_name.h
- src/core/lib/gprpp/validation_errors.h
- src/core/lib/gprpp/work_serializer.h
- src/core/lib/iomgr/block_annotate.h
- src/core/lib/iomgr/buffer_list.h
- src/core/lib/iomgr/call_combiner.h
- src/core/lib/iomgr/cfstream_handle.h
- src/core/lib/iomgr/closure.h
- src/core/lib/iomgr/combiner.h
- src/core/lib/iomgr/dynamic_annotations.h
- src/core/lib/iomgr/endpoint.h
- src/core/lib/iomgr/endpoint_cfstream.h
- src/core/lib/iomgr/endpoint_pair.h
- src/core/lib/iomgr/error.h
- src/core/lib/iomgr/error_cfstream.h
- src/core/lib/iomgr/ev_apple.h
- src/core/lib/iomgr/ev_epoll1_linux.h
- src/core/lib/iomgr/ev_poll_posix.h
- src/core/lib/iomgr/ev_posix.h
- src/core/lib/iomgr/event_engine_shims/closure.h
- src/core/lib/iomgr/event_engine_shims/endpoint.h
- src/core/lib/iomgr/event_engine_shims/tcp_client.h
- src/core/lib/iomgr/exec_ctx.h
- src/core/lib/iomgr/executor.h
- src/core/lib/iomgr/gethostname.h
- src/core/lib/iomgr/grpc_if_nametoindex.h
- src/core/lib/iomgr/internal_errqueue.h
- src/core/lib/iomgr/iocp_windows.h
- src/core/lib/iomgr/iomgr.h
- src/core/lib/iomgr/iomgr_fwd.h
- src/core/lib/iomgr/iomgr_internal.h
- src/core/lib/iomgr/lockfree_event.h
- src/core/lib/iomgr/nameser.h
- src/core/lib/iomgr/polling_entity.h
- src/core/lib/iomgr/pollset.h
- src/core/lib/iomgr/pollset_set.h
- src/core/lib/iomgr/pollset_set_windows.h
- src/core/lib/iomgr/pollset_windows.h
- src/core/lib/iomgr/port.h
- src/core/lib/iomgr/python_util.h
- src/core/lib/iomgr/resolve_address.h
- src/core/lib/iomgr/resolve_address_impl.h
- src/core/lib/iomgr/resolve_address_posix.h
- src/core/lib/iomgr/resolve_address_windows.h
- src/core/lib/iomgr/resolved_address.h
- src/core/lib/iomgr/sockaddr.h
- src/core/lib/iomgr/sockaddr_posix.h
- src/core/lib/iomgr/sockaddr_windows.h
- src/core/lib/iomgr/socket_factory_posix.h
- src/core/lib/iomgr/socket_mutator.h
- src/core/lib/iomgr/socket_utils.h
- src/core/lib/iomgr/socket_utils_posix.h
- src/core/lib/iomgr/socket_windows.h
- src/core/lib/iomgr/systemd_utils.h
- src/core/lib/iomgr/tcp_client.h
- src/core/lib/iomgr/tcp_client_posix.h
- src/core/lib/iomgr/tcp_posix.h
- src/core/lib/iomgr/tcp_server.h
- src/core/lib/iomgr/tcp_server_utils_posix.h
- src/core/lib/iomgr/tcp_windows.h
- src/core/lib/iomgr/timer.h
- src/core/lib/iomgr/timer_generic.h
- src/core/lib/iomgr/timer_heap.h
- src/core/lib/iomgr/timer_manager.h
- src/core/lib/iomgr/unix_sockets_posix.h
- src/core/lib/iomgr/vsock.h
- src/core/lib/iomgr/wakeup_fd_pipe.h
- src/core/lib/iomgr/wakeup_fd_posix.h
- src/core/lib/promise/activity.h
- src/core/lib/promise/all_ok.h
- src/core/lib/promise/arena_promise.h
- src/core/lib/promise/cancel_callback.h
- src/core/lib/promise/context.h
- src/core/lib/promise/detail/basic_seq.h
- src/core/lib/promise/detail/join_state.h
- src/core/lib/promise/detail/promise_factory.h
- src/core/lib/promise/detail/promise_like.h
- src/core/lib/promise/detail/seq_state.h
- src/core/lib/promise/detail/status.h
- src/core/lib/promise/exec_ctx_wakeup_scheduler.h
- src/core/lib/promise/for_each.h
- src/core/lib/promise/if.h
- src/core/lib/promise/interceptor_list.h
- src/core/lib/promise/latch.h
- src/core/lib/promise/loop.h
- src/core/lib/promise/map.h
- src/core/lib/promise/party.h
- src/core/lib/promise/pipe.h
- src/core/lib/promise/poll.h
- src/core/lib/promise/prioritized_race.h
- src/core/lib/promise/promise.h
- src/core/lib/promise/race.h
- src/core/lib/promise/seq.h
- src/core/lib/promise/status_flag.h
- src/core/lib/promise/try_seq.h
- src/core/lib/resource_quota/api.h
- src/core/lib/resource_quota/arena.h
- src/core/lib/resource_quota/connection_quota.h
- src/core/lib/resource_quota/memory_quota.h
- src/core/lib/resource_quota/periodic_update.h
- src/core/lib/resource_quota/resource_quota.h
- src/core/lib/resource_quota/thread_quota.h
- src/core/lib/security/certificate_provider/certificate_provider_factory.h
- src/core/lib/security/certificate_provider/certificate_provider_registry.h
- src/core/lib/security/credentials/alts/check_gcp_environment.h
- src/core/lib/security/credentials/alts/grpc_alts_credentials_options.h
- src/core/lib/security/credentials/channel_creds_registry.h
- src/core/lib/slice/percent_encoding.h
- src/core/lib/slice/slice.h
- src/core/lib/slice/slice_buffer.h
- src/core/lib/slice/slice_internal.h
- src/core/lib/slice/slice_refcount.h
- src/core/lib/slice/slice_string_helpers.h
- src/core/lib/surface/api_trace.h
- src/core/lib/surface/call.h
- src/core/lib/surface/call_test_only.h
- src/core/lib/surface/call_utils.h
- src/core/lib/surface/channel.h
- src/core/lib/surface/channel_init.h
- src/core/lib/surface/channel_stack_type.h
- src/core/lib/surface/client_call.h
- src/core/lib/surface/completion_queue.h
- src/core/lib/surface/completion_queue_factory.h
- src/core/lib/surface/event_string.h
- src/core/lib/surface/filter_stack_call.h
- src/core/lib/surface/init.h
- src/core/lib/surface/init_internally.h
- src/core/lib/surface/lame_client.h
- src/core/lib/surface/server_call.h
- src/core/lib/surface/validate_metadata.h
- src/core/lib/transport/call_arena_allocator.h
- src/core/lib/transport/call_destination.h
- src/core/lib/transport/call_filters.h
- src/core/lib/transport/call_final_info.h
- src/core/lib/transport/call_spine.h
- src/core/lib/transport/call_state.h
- src/core/lib/transport/connectivity_state.h
- src/core/lib/transport/custom_metadata.h
- src/core/lib/transport/error_utils.h
- src/core/lib/transport/http2_errors.h
- src/core/lib/transport/interception_chain.h
- src/core/lib/transport/message.h
- src/core/lib/transport/metadata.h
- src/core/lib/transport/metadata_batch.h
- src/core/lib/transport/metadata_compression_traits.h
- src/core/lib/transport/parsed_metadata.h
- src/core/lib/transport/simple_slice_based_metadata.h
- src/core/lib/transport/status_conversion.h
- src/core/lib/transport/timeout_encoding.h
- src/core/lib/transport/transport.h
- src/core/lib/transport/transport_fwd.h
- src/core/lib/uri/uri_parser.h
- src/core/load_balancing/backend_metric_data.h
- src/core/load_balancing/lb_policy.h
- src/core/load_balancing/lb_policy_factory.h
- src/core/load_balancing/lb_policy_registry.h
- src/core/load_balancing/subchannel_interface.h
- src/core/resolver/endpoint_addresses.h
- src/core/resolver/resolver.h
- src/core/resolver/resolver_factory.h
- src/core/resolver/resolver_registry.h
- src/core/resolver/server_address.h
- src/core/server/server_interface.h
- src/core/service_config/service_config.h
- src/core/service_config/service_config_call_data.h
- src/core/service_config/service_config_parser.h
- src/core/telemetry/call_tracer.h
- src/core/telemetry/histogram_view.h
- src/core/telemetry/metrics.h
- src/core/telemetry/stats.h
- src/core/telemetry/stats_data.h
- src/core/telemetry/tcp_tracer.h
- src/core/tsi/alts/handshaker/transport_security_common_api.h
- src/core/util/json/json.h
- src/core/util/json/json_args.h
- src/core/util/json/json_writer.h
- src/core/util/spinlock.h
- test/core/promise/poll_matcher.h
- third_party/upb/upb/generated_code_support.h
- third_party/upb/upb/mini_descriptor/build_enum.h
- third_party/upb/upb/mini_descriptor/decode.h
- third_party/upb/upb/mini_descriptor/internal/base92.h
- third_party/upb/upb/mini_descriptor/internal/decoder.h
- third_party/upb/upb/mini_descriptor/internal/encode.h
- third_party/upb/upb/mini_descriptor/internal/encode.hpp
- third_party/upb/upb/mini_descriptor/internal/modifiers.h
- third_party/upb/upb/mini_descriptor/internal/wire_constants.h
- third_party/upb/upb/mini_descriptor/link.h
- third_party/upb/upb/wire/decode.h
- third_party/upb/upb/wire/encode.h
- third_party/upb/upb/wire/eps_copy_input_stream.h
- third_party/upb/upb/wire/internal/constants.h
- third_party/upb/upb/wire/internal/decode_fast.h
- third_party/upb/upb/wire/internal/decoder.h
- third_party/upb/upb/wire/internal/reader.h
- third_party/upb/upb/wire/reader.h
- third_party/upb/upb/wire/types.h
src:
- src/core/channelz/channel_trace.cc
- src/core/channelz/channelz.cc
- src/core/channelz/channelz_registry.cc
- src/core/ext/upb-gen/google/protobuf/any.upb_minitable.c
- src/core/ext/upb-gen/google/rpc/status.upb_minitable.c
- src/core/ext/upb-gen/src/proto/grpc/gcp/altscontext.upb_minitable.c
- src/core/ext/upb-gen/src/proto/grpc/gcp/handshaker.upb_minitable.c
- src/core/ext/upb-gen/src/proto/grpc/gcp/transport_security_common.upb_minitable.c
- src/core/handshaker/handshaker_registry.cc
- src/core/handshaker/proxy_mapper_registry.cc
- src/core/lib/address_utils/parse_address.cc
- src/core/lib/address_utils/sockaddr_utils.cc
- src/core/lib/backoff/backoff.cc
- src/core/lib/channel/channel_args.cc
- src/core/lib/channel/channel_args_preconditioning.cc
- src/core/lib/channel/channel_stack.cc
- src/core/lib/channel/channel_stack_builder.cc
- src/core/lib/channel/channel_stack_builder_impl.cc
- src/core/lib/channel/connected_channel.cc
- src/core/lib/channel/promise_based_filter.cc
- src/core/lib/channel/status_util.cc
- src/core/lib/compression/compression.cc
- src/core/lib/compression/compression_internal.cc
- src/core/lib/compression/message_compress.cc
- src/core/lib/config/core_configuration.cc
- src/core/lib/debug/event_log.cc
- src/core/lib/promise/activity.h
- src/core/lib/promise/context.h
- src/core/lib/promise/detail/basic_seq.h
- src/core/lib/promise/detail/promise_factory.h
- src/core/lib/promise/detail/promise_like.h
- src/core/lib/promise/detail/seq_state.h
- src/core/lib/promise/detail/status.h
- src/core/lib/promise/inter_activity_pipe.h
- src/core/lib/promise/poll.h
- src/core/lib/promise/seq.h
- test/core/promise/test_wakeup_schedulers.h
src:
- src/core/lib/debug/trace.cc
- src/core/lib/debug/trace_flags.cc
- src/core/lib/event_engine/ares_resolver.cc
- src/core/lib/event_engine/cf_engine/cf_engine.cc
- src/core/lib/event_engine/cf_engine/cfstream_endpoint.cc
- src/core/lib/event_engine/cf_engine/dns_service_resolver.cc
- src/core/lib/event_engine/channel_args_endpoint_config.cc
- src/core/lib/event_engine/default_event_engine.cc
- src/core/lib/event_engine/default_event_engine_factory.cc
- src/core/lib/event_engine/event_engine.cc
- src/core/lib/event_engine/forkable.cc
- src/core/lib/event_engine/posix_engine/ev_epoll1_linux.cc
- src/core/lib/event_engine/posix_engine/ev_poll_posix.cc
- src/core/lib/event_engine/posix_engine/event_poller_posix_default.cc
- src/core/lib/event_engine/posix_engine/internal_errqueue.cc
- src/core/lib/event_engine/posix_engine/lockfree_event.cc
- src/core/lib/event_engine/posix_engine/native_posix_dns_resolver.cc
- src/core/lib/event_engine/posix_engine/posix_endpoint.cc
- src/core/lib/event_engine/posix_engine/posix_engine.cc
- src/core/lib/event_engine/posix_engine/posix_engine_listener.cc
- src/core/lib/event_engine/posix_engine/posix_engine_listener_utils.cc
- src/core/lib/event_engine/posix_engine/tcp_socket_utils.cc
- src/core/lib/event_engine/posix_engine/timer.cc
- src/core/lib/event_engine/posix_engine/timer_heap.cc
- src/core/lib/event_engine/posix_engine/timer_manager.cc
- src/core/lib/event_engine/posix_engine/traced_buffer_list.cc
- src/core/lib/event_engine/posix_engine/wakeup_fd_eventfd.cc
- src/core/lib/event_engine/posix_engine/wakeup_fd_pipe.cc
- src/core/lib/event_engine/posix_engine/wakeup_fd_posix_default.cc
- src/core/lib/event_engine/resolved_address.cc
- src/core/lib/event_engine/shim.cc
- src/core/lib/event_engine/slice.cc
- src/core/lib/event_engine/slice_buffer.cc
- src/core/lib/event_engine/tcp_socket_utils.cc
- src/core/lib/event_engine/thread_pool/thread_count.cc
- src/core/lib/event_engine/thread_pool/thread_pool_factory.cc
- src/core/lib/event_engine/thread_pool/work_stealing_thread_pool.cc
- src/core/lib/event_engine/thready_event_engine/thready_event_engine.cc
- src/core/lib/event_engine/time_util.cc
- src/core/lib/event_engine/utils.cc
- src/core/lib/event_engine/windows/grpc_polled_fd_windows.cc
- src/core/lib/event_engine/windows/iocp.cc
- src/core/lib/event_engine/windows/native_windows_dns_resolver.cc
- src/core/lib/event_engine/windows/win_socket.cc
- src/core/lib/event_engine/windows/windows_endpoint.cc
- src/core/lib/event_engine/windows/windows_engine.cc
- src/core/lib/event_engine/windows/windows_listener.cc
- src/core/lib/event_engine/work_queue/basic_work_queue.cc
- src/core/lib/experiments/config.cc
- src/core/lib/experiments/experiments.cc
- src/core/lib/gprpp/dump_args.cc
- src/core/lib/gprpp/glob.cc
- src/core/lib/gprpp/load_file.cc
- src/core/lib/gprpp/per_cpu.cc
- src/core/lib/gprpp/ref_counted_string.cc
- src/core/lib/gprpp/status_helper.cc
- src/core/lib/gprpp/time.cc
- src/core/lib/gprpp/time_averaged_stats.cc
- src/core/lib/gprpp/validation_errors.cc
- src/core/lib/gprpp/work_serializer.cc
- src/core/lib/iomgr/buffer_list.cc
- src/core/lib/iomgr/call_combiner.cc
- src/core/lib/iomgr/cfstream_handle.cc
- src/core/lib/iomgr/closure.cc
- src/core/lib/iomgr/combiner.cc
- src/core/lib/iomgr/dualstack_socket_posix.cc
- src/core/lib/iomgr/endpoint.cc
- src/core/lib/iomgr/endpoint_cfstream.cc
- src/core/lib/iomgr/endpoint_pair_posix.cc
- src/core/lib/iomgr/endpoint_pair_windows.cc
- src/core/lib/iomgr/error.cc
- src/core/lib/iomgr/error_cfstream.cc
- src/core/lib/iomgr/ev_apple.cc
- src/core/lib/iomgr/ev_epoll1_linux.cc
- src/core/lib/iomgr/ev_poll_posix.cc
- src/core/lib/iomgr/ev_posix.cc
- src/core/lib/iomgr/event_engine_shims/closure.cc
- src/core/lib/iomgr/event_engine_shims/endpoint.cc
- src/core/lib/iomgr/event_engine_shims/tcp_client.cc
- src/core/lib/iomgr/exec_ctx.cc
- src/core/lib/iomgr/executor.cc
- src/core/lib/iomgr/fork_posix.cc
- src/core/lib/iomgr/fork_windows.cc
- src/core/lib/iomgr/gethostname_fallback.cc
- src/core/lib/iomgr/gethostname_host_name_max.cc
- src/core/lib/iomgr/gethostname_sysconf.cc
- src/core/lib/iomgr/grpc_if_nametoindex_posix.cc
- src/core/lib/iomgr/grpc_if_nametoindex_unsupported.cc
- src/core/lib/iomgr/internal_errqueue.cc
- src/core/lib/iomgr/iocp_windows.cc
- src/core/lib/iomgr/iomgr.cc
- src/core/lib/iomgr/iomgr_internal.cc
- src/core/lib/iomgr/iomgr_posix.cc
- src/core/lib/iomgr/iomgr_posix_cfstream.cc
- src/core/lib/iomgr/iomgr_windows.cc
- src/core/lib/iomgr/lockfree_event.cc
- src/core/lib/iomgr/polling_entity.cc
- src/core/lib/iomgr/pollset.cc
- src/core/lib/iomgr/pollset_set.cc
- src/core/lib/iomgr/pollset_set_windows.cc
- src/core/lib/iomgr/pollset_windows.cc
- src/core/lib/iomgr/resolve_address.cc
- src/core/lib/iomgr/resolve_address_posix.cc
- src/core/lib/iomgr/resolve_address_windows.cc
- src/core/lib/iomgr/sockaddr_utils_posix.cc
- src/core/lib/iomgr/socket_factory_posix.cc
- src/core/lib/iomgr/socket_mutator.cc
- src/core/lib/iomgr/socket_utils_common_posix.cc
- src/core/lib/iomgr/socket_utils_linux.cc
- src/core/lib/iomgr/socket_utils_posix.cc
- src/core/lib/iomgr/socket_utils_windows.cc
- src/core/lib/iomgr/socket_windows.cc
- src/core/lib/iomgr/systemd_utils.cc
- src/core/lib/iomgr/tcp_client.cc
- src/core/lib/iomgr/tcp_client_cfstream.cc
- src/core/lib/iomgr/tcp_client_posix.cc
- src/core/lib/iomgr/tcp_client_windows.cc
- src/core/lib/iomgr/tcp_posix.cc
- src/core/lib/iomgr/tcp_server.cc
- src/core/lib/iomgr/tcp_server_posix.cc
- src/core/lib/iomgr/tcp_server_utils_posix_common.cc
- src/core/lib/iomgr/tcp_server_utils_posix_ifaddrs.cc
- src/core/lib/iomgr/tcp_server_utils_posix_noifaddrs.cc
- src/core/lib/iomgr/tcp_server_windows.cc
- src/core/lib/iomgr/tcp_windows.cc
- src/core/lib/iomgr/timer.cc
- src/core/lib/iomgr/timer_generic.cc
- src/core/lib/iomgr/timer_heap.cc
- src/core/lib/iomgr/timer_manager.cc
- src/core/lib/iomgr/unix_sockets_posix.cc
- src/core/lib/iomgr/unix_sockets_posix_noop.cc
- src/core/lib/iomgr/vsock.cc
- src/core/lib/iomgr/wakeup_fd_eventfd.cc
- src/core/lib/iomgr/wakeup_fd_nospecial.cc
- src/core/lib/iomgr/wakeup_fd_pipe.cc
- src/core/lib/iomgr/wakeup_fd_posix.cc
- src/core/lib/promise/activity.cc
- src/core/lib/promise/party.cc
- src/core/lib/resource_quota/api.cc
- src/core/lib/resource_quota/arena.cc
- src/core/lib/resource_quota/connection_quota.cc
- src/core/lib/resource_quota/memory_quota.cc
- src/core/lib/resource_quota/periodic_update.cc
- src/core/lib/resource_quota/resource_quota.cc
- src/core/lib/resource_quota/thread_quota.cc
- src/core/lib/security/certificate_provider/certificate_provider_registry.cc
- src/core/lib/security/credentials/alts/check_gcp_environment.cc
- src/core/lib/security/credentials/alts/check_gcp_environment_linux.cc
- src/core/lib/security/credentials/alts/check_gcp_environment_no_op.cc
- src/core/lib/security/credentials/alts/check_gcp_environment_windows.cc
- src/core/lib/security/credentials/alts/grpc_alts_credentials_client_options.cc
- src/core/lib/security/credentials/alts/grpc_alts_credentials_options.cc
- src/core/lib/security/credentials/alts/grpc_alts_credentials_server_options.cc
- src/core/lib/slice/percent_encoding.cc
- src/core/lib/slice/slice.cc
- src/core/lib/slice/slice_buffer.cc
- src/core/lib/slice/slice_string_helpers.cc
- src/core/lib/surface/byte_buffer.cc
- src/core/lib/surface/byte_buffer_reader.cc
- src/core/lib/surface/call.cc
- src/core/lib/surface/call_details.cc
- src/core/lib/surface/call_log_batch.cc
- src/core/lib/surface/call_utils.cc
- src/core/lib/surface/channel.cc
- src/core/lib/surface/channel_init.cc
- src/core/lib/surface/channel_stack_type.cc
- src/core/lib/surface/client_call.cc
- src/core/lib/surface/completion_queue.cc
- src/core/lib/surface/completion_queue_factory.cc
- src/core/lib/surface/event_string.cc
- src/core/lib/surface/filter_stack_call.cc
- src/core/lib/surface/init_internally.cc
- src/core/lib/surface/lame_client.cc
- src/core/lib/surface/metadata_array.cc
- src/core/lib/surface/server_call.cc
- src/core/lib/surface/validate_metadata.cc
- src/core/lib/surface/version.cc
- src/core/lib/transport/call_arena_allocator.cc
- src/core/lib/transport/call_filters.cc
- src/core/lib/transport/call_final_info.cc
- src/core/lib/transport/call_spine.cc
- src/core/lib/transport/call_state.cc
- src/core/lib/transport/connectivity_state.cc
- src/core/lib/transport/error_utils.cc
- src/core/lib/transport/interception_chain.cc
- src/core/lib/transport/message.cc
- src/core/lib/transport/metadata.cc
- src/core/lib/transport/metadata_batch.cc
- src/core/lib/transport/parsed_metadata.cc
- src/core/lib/transport/status_conversion.cc
- src/core/lib/transport/timeout_encoding.cc
- src/core/lib/transport/transport.cc
- src/core/lib/transport/transport_op_string.cc
- src/core/lib/uri/uri_parser.cc
- src/core/load_balancing/lb_policy.cc
- src/core/load_balancing/lb_policy_registry.cc
- src/core/resolver/endpoint_addresses.cc
- src/core/resolver/resolver.cc
- src/core/resolver/resolver_registry.cc
- src/core/service_config/service_config_parser.cc
- src/core/telemetry/call_tracer.cc
- src/core/telemetry/histogram_view.cc
- src/core/telemetry/metrics.cc
- src/core/telemetry/stats.cc
- src/core/telemetry/stats_data.cc
- src/core/tsi/alts/handshaker/transport_security_common_api.cc
- src/core/util/json/json_writer.cc
- test/core/transport/interception_chain_test.cc
- third_party/upb/upb/mini_descriptor/build_enum.c
- third_party/upb/upb/mini_descriptor/decode.c
- third_party/upb/upb/mini_descriptor/internal/base92.c
- third_party/upb/upb/mini_descriptor/internal/encode.c
- third_party/upb/upb/mini_descriptor/link.c
- third_party/upb/upb/wire/decode.c
- third_party/upb/upb/wire/encode.c
- third_party/upb/upb/wire/eps_copy_input_stream.c
- third_party/upb/upb/wire/internal/decode_fast.c
- third_party/upb/upb/wire/reader.c
- test/core/promise/inter_activity_pipe_test.cc
deps:
- gtest
- utf8_range_lib
- upb_message_lib
- z
- absl/base:config
- absl/base:no_destructor
- absl/cleanup:cleanup
- absl/container:flat_hash_map
- absl/container:inlined_vector
- absl/functional:function_ref
- absl/hash:hash
- absl/meta:type_traits
- absl/status:statusor
- absl/types:span
- absl/utility:utility
- cares
- gpr
- address_sorting
uses_polling: false
- name: interception_chain_test
gtest: true
build: test
language: c++
headers:
- test/core/promise/poll_matcher.h
src:
- test/core/transport/interception_chain_test.cc
deps:
- gtest
- grpc
uses_polling: false
- name: interceptor_list_test
gtest: true
@ -13179,7 +12585,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -13700,7 +13105,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/bitset.h
- src/core/lib/gprpp/down_cast.h
@ -13887,7 +13291,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/experiments/config.h
- src/core/lib/experiments/experiments.h
- src/core/lib/gprpp/atomic_utils.h
@ -14503,7 +13906,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/down_cast.h
- src/core/lib/gprpp/dump_args.h
@ -14812,7 +14214,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/down_cast.h
- src/core/lib/gprpp/dump_args.h
@ -15747,7 +15148,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/bitset.h
- src/core/lib/gprpp/down_cast.h
@ -21533,7 +20933,6 @@ targets:
- src/core/lib/debug/trace.h
- src/core/lib/debug/trace_flags.h
- src/core/lib/debug/trace_impl.h
- src/core/lib/event_engine/event_engine_context.h
- src/core/lib/gprpp/atomic_utils.h
- src/core/lib/gprpp/down_cast.h
- src/core/lib/gprpp/dump_args.h

16
src/core/BUILD
Unescape View File

@ -570,7 +570,6 @@ grpc_cc_library(
external_deps = [
"absl/base:core_headers",
"absl/log:check",
"absl/log:log",
"absl/strings",
"absl/strings:str_format",
],
@ -580,6 +579,7 @@ grpc_cc_library(
"arena",
"construct_destruct",
"context",
"event_engine_context",
"poll",
"promise_factory",
"ref_counted",
@ -959,7 +959,6 @@ grpc_cc_library(
"construct_destruct",
"context",
"dump_args",
"event_engine_context",
"no_destruct",
"poll",
"promise_factory",
@ -2753,7 +2752,7 @@ grpc_cc_library(
"@grpc:alt_grpc_base_legacy",
],
deps = [
"context",
"arena",
"//:event_engine_base_hdrs",
"//:gpr",
],
@ -4606,8 +4605,6 @@ grpc_cc_library(
],
external_deps = [
"absl/base:core_headers",
"absl/log:check",
"absl/log:log",
"absl/meta:type_traits",
"absl/random",
"absl/status",
@ -4616,6 +4613,7 @@ grpc_cc_library(
],
deps = [
"activity",
"arena",
"arena_promise",
"channel_args",
"channel_fwd",
@ -7158,6 +7156,7 @@ grpc_cc_library(
"closure",
"connectivity_state",
"error",
"event_engine_context",
"experiments",
"iomgr_fwd",
"metadata",
@ -7491,7 +7490,6 @@ grpc_cc_library(
"absl/container:flat_hash_map",
"absl/functional:any_invocable",
"absl/log:check",
"absl/log:log",
"absl/random",
"absl/random:bit_gen_ref",
"absl/status",
@ -7509,6 +7507,7 @@ grpc_cc_library(
"chaotic_good_transport",
"context",
"default_event_engine",
"event_engine_context",
"event_engine_wakeup_scheduler",
"for_each",
"grpc_promise_endpoint",
@ -7715,6 +7714,7 @@ grpc_cc_library(
"call_arena_allocator",
"call_filters",
"dual_ref_counted",
"event_engine_context",
"for_each",
"if",
"latch",
@ -7740,6 +7740,7 @@ grpc_cc_library(
],
deps = [
"channel_stack_type",
"event_engine_context",
"interception_chain",
"//:channel",
"//:config",
@ -7931,7 +7932,6 @@ grpc_cc_library(
],
external_deps = [
"absl/log:check",
"absl/log:log",
"absl/random",
"absl/random:bit_gen_ref",
"absl/status",
@ -7949,9 +7949,9 @@ grpc_cc_library(
"chaotic_good_settings_metadata",
"closure",
"context",
"default_event_engine",
"error",
"error_utils",
"event_engine_context",
"event_engine_extensions",
"event_engine_query_extensions",
"event_engine_tcp_socket_utils",

17
src/core/client_channel/client_channel.cc
Unescape View File

@ -73,6 +73,7 @@
#include "src/core/lib/promise/poll.h"
#include "src/core/lib/promise/sleep.h"
#include "src/core/lib/promise/try_seq.h"
#include "src/core/lib/resource_quota/arena.h"
#include "src/core/lib/security/credentials/credentials.h"
#include "src/core/lib/slice/slice.h"
#include "src/core/lib/slice/slice_internal.h"
@ -769,10 +770,12 @@ grpc_call* ClientChannel::CreateCall(
grpc_call* parent_call, uint32_t propagation_mask,
grpc_completion_queue* cq, grpc_pollset_set* /*pollset_set_alternative*/,
Slice path, absl::optional<Slice> authority, Timestamp deadline, bool) {
auto arena = call_arena_allocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine());
return MakeClientCall(parent_call, propagation_mask, cq, std::move(path),
std::move(authority), false, deadline,
compression_options(), event_engine_.get(),
call_arena_allocator()->MakeArena(), Ref());
compression_options(), std::move(arena), Ref());
}
void ClientChannel::StartCall(UnstartedCallHandler unstarted_handler) {
@ -1256,6 +1259,9 @@ void ClientChannel::StartIdleTimer() {
}
});
});
auto arena = SimpleArenaAllocator(0)->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine());
idle_activity_.Set(MakeActivity(
std::move(promise), ExecCtxWakeupScheduler{},
[self = std::move(self)](absl::Status status) mutable {
@ -1267,13 +1273,14 @@ void ClientChannel::StartIdleTimer() {
GRPC_CHANNEL_IDLE, absl::OkStatus(),
"channel entering IDLE", nullptr);
// TODO(roth): In case there's a race condition, we
// might need to check for any calls that are queued
// waiting for a resolver result or an LB pick.
// might need to check for any calls that are
// queued waiting for a resolver result or an LB
// pick.
},
DEBUG_LOCATION);
}
},
GetContext<EventEngine>()));
std::move(arena)));
}
absl::Status ClientChannel::ApplyServiceConfigToCall(

7
src/core/client_channel/direct_channel.cc
Unescape View File

@ -15,6 +15,7 @@
#include "src/core/client_channel/direct_channel.h"
#include "src/core/lib/config/core_configuration.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/orphanable.h"
#include "src/core/lib/surface/channel_stack_type.h"
#include "src/core/lib/surface/client_call.h"
@ -71,10 +72,12 @@ grpc_call* DirectChannel::CreateCall(
grpc_completion_queue* cq, grpc_pollset_set* /*pollset_set_alternative*/,
Slice path, absl::optional<Slice> authority, Timestamp deadline,
bool /*registered_method*/) {
auto arena = call_arena_allocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
return MakeClientCall(parent_call, propagation_mask, cq, std::move(path),
std::move(authority), false, deadline,
compression_options(), event_engine_.get(),
call_arena_allocator()->MakeArena(), Ref());
compression_options(), std::move(arena), Ref());
}
} // namespace grpc_core

11
src/core/ext/filters/channel_idle/legacy_channel_idle_filter.cc
Unescape View File

@ -51,6 +51,7 @@
#include "src/core/lib/promise/promise.h"
#include "src/core/lib/promise/sleep.h"
#include "src/core/lib/promise/try_seq.h"
#include "src/core/lib/resource_quota/arena.h"
#include "src/core/lib/surface/channel_stack_type.h"
#include "src/core/lib/transport/http2_errors.h"
#include "src/core/lib/transport/metadata_batch.h"
@ -176,6 +177,9 @@ void LegacyMaxAgeFilter::PostInit() {
// Start the max age timer
if (max_connection_age_ != Duration::Infinity()) {
auto arena = SimpleArenaAllocator(0)->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
channel_stack->EventEngine());
max_age_activity_.Set(MakeActivity(
TrySeq(
// First sleep until the max connection age
@ -213,7 +217,7 @@ void LegacyMaxAgeFilter::PostInit() {
// (if it did not, it was cancelled)
if (status.ok()) CloseChannel();
},
channel_stack->EventEngine()));
std::move(arena)));
}
}
@ -270,12 +274,15 @@ void LegacyChannelIdleFilter::StartIdleTimer() {
}
});
});
auto arena = SimpleArenaAllocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
channel_stack_->EventEngine());
activity_.Set(MakeActivity(
std::move(promise), ExecCtxWakeupScheduler{},
[channel_stack, this](absl::Status status) {
if (status.ok()) CloseChannel();
},
channel_stack->EventEngine()));
std::move(arena)));
}
void LegacyChannelIdleFilter::CloseChannel() {

8
src/core/ext/transport/chaotic_good/client/chaotic_good_connector.cc
Unescape View File

@ -36,6 +36,7 @@
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/config/core_configuration.h"
#include "src/core/lib/event_engine/channel_args_endpoint_config.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/event_engine/extensions/chaotic_good_extension.h"
#include "src/core/lib/event_engine/query_extensions.h"
#include "src/core/lib/event_engine/tcp_socket_utils.h"
@ -76,7 +77,10 @@ const int32_t kTimeoutSecs = 120;
ChaoticGoodConnector::ChaoticGoodConnector(
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine)
: event_engine_(std::move(event_engine)),
handshake_mgr_(MakeRefCounted<HandshakeManager>()) {}
handshake_mgr_(MakeRefCounted<HandshakeManager>()) {
arena_->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
}
ChaoticGoodConnector::~ChaoticGoodConnector() {
CHECK_EQ(notify_, nullptr);
@ -333,7 +337,7 @@ void ChaoticGoodConnector::OnHandshakeDone(
status);
}
},
arena_, event_engine_.get());
arena_);
MutexLock lock(&mu_);
if (!is_shutdown_) {
connect_activity_ = std::move(activity);

4
src/core/ext/transport/chaotic_good/server/chaotic_good_server.cc
Unescape View File

@ -150,6 +150,8 @@ ChaoticGoodServerListener::ActiveConnection::ActiveConnection(
RefCountedPtr<ChaoticGoodServerListener> listener,
std::unique_ptr<EventEngine::Endpoint> endpoint)
: listener_(std::move(listener)) {
arena_->SetContext<grpc_event_engine::experimental::EventEngine>(
listener_->event_engine_.get());
handshaking_state_ = MakeRefCounted<HandshakingState>(Ref());
handshaking_state_->Start(std::move(endpoint));
}
@ -438,7 +440,7 @@ void ChaoticGoodServerListener::ActiveConnection::HandshakingState::
self->connection_->Done();
}
},
connection_->arena_.get(), connection_->listener_->event_engine_.get());
connection_->arena_.get());
MutexLock lock(&connection_->mu_);
if (connection_->orphaned_) return;
connection_->receive_settings_activity_ = std::move(activity);

7
src/core/ext/transport/chaotic_good/server_transport.cc
Unescape View File

@ -34,6 +34,7 @@
#include "src/core/ext/transport/chaotic_good/frame.h"
#include "src/core/ext/transport/chaotic_good/frame_header.h"
#include "src/core/ext/transport/chttp2/transport/hpack_encoder.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/ref_counted_ptr.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/promise/activity.h"
@ -239,13 +240,15 @@ auto ChaoticGoodServerTransport::DeserializeAndPushFragmentToNewCall(
ChaoticGoodTransport& transport) {
ClientFragmentFrame fragment_frame;
RefCountedPtr<Arena> arena(call_arena_allocator_->MakeArena());
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
absl::Status status = transport.DeserializeFrame(
frame_header, std::move(buffers), arena.get(), fragment_frame,
FrameLimits{1024 * 1024 * 1024, aligned_bytes_ - 1});
absl::optional<CallInitiator> call_initiator;
if (status.ok()) {
auto call = MakeCallPair(std::move(fragment_frame.headers),
event_engine_.get(), std::move(arena));
auto call =
MakeCallPair(std::move(fragment_frame.headers), std::move(arena));
call_initiator.emplace(std::move(call.initiator));
auto add_result = NewStream(frame_header.stream_id, *call_initiator);
if (add_result.ok()) {

7
src/core/ext/transport/inproc/inproc_transport.cc
Unescape View File

@ -26,6 +26,7 @@
#include "src/core/ext/transport/inproc/legacy_inproc_transport.h"
#include "src/core/lib/config/core_configuration.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/experiments/experiments.h"
#include "src/core/lib/gprpp/crash.h"
#include "src/core/lib/gprpp/debug_location.h"
@ -112,8 +113,10 @@ class InprocServerTransport final : public ServerTransport {
case ConnectionState::kReady:
break;
}
auto server_call = MakeCallPair(std::move(md), event_engine_.get(),
call_arena_allocator_->MakeArena());
auto arena = call_arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
auto server_call = MakeCallPair(std::move(md), std::move(arena));
unstarted_call_handler_->StartCall(std::move(server_call.handler));
return std::move(server_call.initiator);
}

6
src/core/lib/channel/promise_based_filter.cc
Unescape View File

@ -100,11 +100,7 @@ BaseCallData::BaseCallData(
receive_message_(
flags & kFilterExaminesInboundMessages
? arena_->New<ReceiveMessage>(this, make_recv_interceptor())
: nullptr),
event_engine_(
ChannelFilterFromElem(elem)
->hack_until_per_channel_stack_event_engines_land_get_event_engine()) {
}
: nullptr) {}
BaseCallData::~BaseCallData() {
FakeActivity(this).Run([this] {

16
src/core/lib/channel/promise_based_filter.h
Unescape View File

@ -150,16 +150,6 @@ class ChannelFilter {
virtual bool GetChannelInfo(const grpc_channel_info*) { return false; }
virtual ~ChannelFilter() = default;
grpc_event_engine::experimental::EventEngine*
hack_until_per_channel_stack_event_engines_land_get_event_engine() {
return event_engine_.get();
}
private:
// TODO(ctiller): remove once per-channel-stack EventEngines land
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_ =
grpc_event_engine::experimental::GetDefaultEventEngine();
};
namespace promise_filter_detail {
@ -953,9 +943,8 @@ class BaseCallData : public Activity, private Wakeable {
: promise_detail::Context<Arena>(call_data->arena_),
promise_detail::Context<grpc_polling_entity>(
call_data->pollent_.load(std::memory_order_acquire)),
promise_detail::Context<CallFinalization>(&call_data->finalization_),
promise_detail::Context<grpc_event_engine::experimental::EventEngine>(
call_data->event_engine_) {}
promise_detail::Context<CallFinalization>(&call_data->finalization_) {
}
};
class Flusher {
@ -1305,7 +1294,6 @@ class BaseCallData : public Activity, private Wakeable {
Pipe<ServerMetadataHandle>* const server_initial_metadata_pipe_;
SendMessage* const send_message_;
ReceiveMessage* const receive_message_;
grpc_event_engine::experimental::EventEngine* event_engine_;
};
class ClientCallData : public BaseCallData {

6
src/core/lib/event_engine/event_engine_context.h
Unescape View File

@ -18,12 +18,14 @@
#include <grpc/event_engine/event_engine.h>
#include <grpc/support/port_platform.h>
#include "src/core/lib/promise/context.h"
#include "src/core/lib/resource_quota/arena.h"
namespace grpc_core {
template <>
struct ContextType<grpc_event_engine::experimental::EventEngine> {};
struct ArenaContextType<grpc_event_engine::experimental::EventEngine> {
static void Destroy(grpc_event_engine::experimental::EventEngine*) {}
};
} // namespace grpc_core

1
src/core/lib/promise/activity.h
Unescape View File

@ -33,7 +33,6 @@
#include <grpc/support/port_platform.h>
#include "src/core/lib/debug/trace.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/construct_destruct.h"
#include "src/core/lib/gprpp/dump_args.h"
#include "src/core/lib/gprpp/no_destruct.h"

40
src/core/lib/promise/party.cc
Unescape View File

@ -23,6 +23,7 @@
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/sync.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/promise/activity.h"
@ -180,6 +181,7 @@ Party::~Party() {}
void Party::CancelRemainingParticipants() {
if (!sync_.has_participants()) return;
ScopedActivity activity(this);
promise_detail::Context<Arena> arena_ctx(arena_.get());
for (size_t i = 0; i < party_detail::kMaxParticipants; i++) {
if (auto* p =
participants_[i].exchange(nullptr, std::memory_order_acquire)) {
@ -222,11 +224,12 @@ void Party::RunLocked() {
} else {
// But if there's already a party queued, we're better off asking event
// engine to run it so we can spread load.
event_engine()->Run([this]() {
ApplicationCallbackExecCtx app_exec_ctx;
ExecCtx exec_ctx;
RunLocked();
});
arena_->GetContext<grpc_event_engine::experimental::EventEngine>()->Run(
[this]() {
ApplicationCallbackExecCtx app_exec_ctx;
ExecCtx exec_ctx;
RunLocked();
});
}
return;
}
@ -238,8 +241,7 @@ void Party::RunLocked() {
DCHECK(g_current_party_run_next == &run_next);
g_current_party_run_next = nullptr;
if (done) {
ScopedActivity activity(this);
PartyOver();
PartyIsOver();
}
if (run_next != nullptr) {
run_next->RunLocked();
@ -262,6 +264,7 @@ void Party::RunLocked() {
bool Party::RunParty() {
ScopedActivity activity(this);
promise_detail::Context<Arena> arena_ctx(arena_.get());
return sync_.RunParty([this](int i) { return RunOneParticipant(i); });
}
@ -325,12 +328,13 @@ void Party::Wakeup(WakeupMask wakeup_mask) {
void Party::WakeupAsync(WakeupMask wakeup_mask) {
if (sync_.ScheduleWakeup(wakeup_mask)) {
event_engine()->Run([this]() {
ApplicationCallbackExecCtx app_exec_ctx;
ExecCtx exec_ctx;
RunLocked();
Unref();
});
arena_->GetContext<grpc_event_engine::experimental::EventEngine>()->Run(
[this]() {
ApplicationCallbackExecCtx app_exec_ctx;
ExecCtx exec_ctx;
RunLocked();
Unref();
});
} else {
Unref();
}
@ -339,8 +343,14 @@ void Party::WakeupAsync(WakeupMask wakeup_mask) {
void Party::Drop(WakeupMask) { Unref(); }
void Party::PartyIsOver() {
ScopedActivity activity(this);
PartyOver();
auto arena = arena_;
{
ScopedActivity activity(this);
promise_detail::Context<Arena> arena_ctx(arena_.get());
CancelRemainingParticipants();
arena->DestroyManagedNewObjects();
}
this->~Party();
}
} // namespace grpc_core

38
src/core/lib/promise/party.h
Unescape View File

@ -41,6 +41,7 @@
#include "src/core/lib/promise/context.h"
#include "src/core/lib/promise/detail/promise_factory.h"
#include "src/core/lib/promise/poll.h"
#include "src/core/lib/resource_quota/arena.h"
// Two implementations of party synchronization are provided: one using a single
// atomic, the other using a mutex and a set of state variables.
@ -83,7 +84,7 @@ class PartySyncUsingAtomics {
}
GRPC_MUST_USE_RESULT bool RefIfNonZero();
// Returns true if the ref count is now zero and the caller should call
// PartyOver
// PartyIsOver
GRPC_MUST_USE_RESULT bool Unref() {
const uint64_t prev_state =
state_.fetch_sub(kOneRef, std::memory_order_acq_rel);
@ -375,6 +376,11 @@ class Party : public Activity, private Wakeable {
Party(const Party&) = delete;
Party& operator=(const Party&) = delete;
static RefCountedPtr<Party> Make(RefCountedPtr<Arena> arena) {
auto* arena_ptr = arena.get();
return RefCountedPtr<Party>(arena_ptr->New<Party>(std::move(arena)));
}
// Spawn one promise into the party.
// The promise will be polled until it is resolved, or until the party is shut
// down.
@ -414,6 +420,8 @@ class Party : public Activity, private Wakeable {
return RefCountedPtr<Party>(this);
}
Arena* arena() { return arena_.get(); }
// Return a promise that resolves to Empty{} when the current party poll is
// complete.
// This is useful for implementing batching and the like: we can hold some
@ -446,26 +454,21 @@ class Party : public Activity, private Wakeable {
};
protected:
explicit Party(size_t initial_refs) : sync_(initial_refs) {}
friend class Arena;
// Derived types should be constructed upon `arena`.
explicit Party(RefCountedPtr<Arena> arena)
: sync_(1), arena_(std::move(arena)) {}
~Party() override;
// Main run loop. Must be locked.
// Polls participants and drains the add queue until there is no work left to
// be done.
// Derived types will likely want to override this to set up their
// contexts before polling.
// Should not be called by derived types except as a tail call to the base
// class RunParty when overriding this method to add custom context.
// Returns true if the party is over.
GRPC_MUST_USE_RESULT virtual bool RunParty();
GRPC_MUST_USE_RESULT bool RunParty();
bool RefIfNonZero() { return sync_.RefIfNonZero(); }
// Destroy any remaining participants.
// Should be called by derived types in response to PartyOver.
// Needs to have normal context setup before calling.
void CancelRemainingParticipants();
private:
// Concrete implementation of a participant for some promise & oncomplete
// type.
@ -598,10 +601,9 @@ class Party : public Activity, private Wakeable {
std::atomic<State> state_{State::kFactory};
};
// Notification that the party has finished and this instance can be deleted.
// Derived types should arrange to call CancelRemainingParticipants during
// this sequence.
virtual void PartyOver() = 0;
// Destroy any remaining participants.
// Needs to have normal context setup before calling.
void CancelRemainingParticipants();
// Run the locked part of the party until it is unlocked.
void RunLocked();
@ -619,9 +621,6 @@ class Party : public Activity, private Wakeable {
void AddParticipants(Participant** participant, size_t count);
bool RunOneParticipant(int i);
virtual grpc_event_engine::experimental::EventEngine* event_engine()
const = 0;
// Sentinal value for currently_polling_ when no participant is being polled.
static constexpr uint8_t kNotPolling = 255;
@ -638,6 +637,7 @@ class Party : public Activity, private Wakeable {
// If the lower bit is unset, then this is a Participant*.
// If the lower bit is set, then this is a ParticipantFactory*.
std::atomic<Participant*> participants_[party_detail::kMaxParticipants] = {};
RefCountedPtr<Arena> arena_;
};
template <>

26
src/core/lib/surface/call.cc
Unescape View File

@ -63,6 +63,7 @@
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/channel/status_util.h"
#include "src/core/lib/compression/compression_internal.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/experiments/experiments.h"
#include "src/core/lib/gprpp/bitset.h"
#include "src/core/lib/gprpp/cpp_impl_of.h"
@ -118,12 +119,13 @@ using GrpcClosure = Closure;
///////////////////////////////////////////////////////////////////////////////
// Call
Call::Call(bool is_client, Timestamp send_deadline, RefCountedPtr<Arena> arena,
grpc_event_engine::experimental::EventEngine* event_engine)
Call::Call(bool is_client, Timestamp send_deadline, RefCountedPtr<Arena> arena)
: arena_(std::move(arena)),
send_deadline_(send_deadline),
is_client_(is_client),
event_engine_(event_engine) {
is_client_(is_client) {
DCHECK_NE(arena_.get(), nullptr);
DCHECK_NE(arena_->GetContext<grpc_event_engine::experimental::EventEngine>(),
nullptr);
arena_->SetContext<Call>(this);
}
@ -350,20 +352,25 @@ void Call::UpdateDeadline(Timestamp deadline) {
StatusIntProperty::kRpcStatus, GRPC_STATUS_DEADLINE_EXCEEDED));
return;
}
auto* event_engine =
arena_->GetContext<grpc_event_engine::experimental::EventEngine>();
if (deadline_ != Timestamp::InfFuture()) {
if (!event_engine_->Cancel(deadline_task_)) return;
if (!event_engine->Cancel(deadline_task_)) return;
} else {
InternalRef("deadline");
}
deadline_ = deadline;
deadline_task_ = event_engine_->RunAfter(deadline - Timestamp::Now(), this);
deadline_task_ = event_engine->RunAfter(deadline - Timestamp::Now(), this);
}
void Call::ResetDeadline() {
{
MutexLock lock(&deadline_mu_);
if (deadline_ == Timestamp::InfFuture()) return;
if (!event_engine_->Cancel(deadline_task_)) return;
if (!arena_->GetContext<grpc_event_engine::experimental::EventEngine>()
->Cancel(deadline_task_)) {
return;
}
deadline_ = Timestamp::InfFuture();
}
InternalUnref("deadline[reset]");
@ -563,5 +570,8 @@ const char* grpc_call_error_to_string(grpc_call_error error) {
void grpc_call_run_in_event_engine(const grpc_call* call,
absl::AnyInvocable<void()> cb) {
grpc_core::Call::FromC(call)->event_engine()->Run(std::move(cb));
grpc_core::Call::FromC(call)
->arena()
->GetContext<grpc_event_engine::experimental::EventEngine>()
->Run(std::move(cb));
}

11
src/core/lib/surface/call.h
Unescape View File

@ -85,7 +85,7 @@ class Call : public CppImplOf<Call, grpc_call>,
public grpc_event_engine::experimental::EventEngine::
Closure /* for deadlines */ {
public:
Arena* arena() { return arena_.get(); }
Arena* arena() const { return arena_.get(); }
bool is_client() const { return is_client_; }
virtual bool Completed() = 0;
@ -120,11 +120,6 @@ class Call : public CppImplOf<Call, grpc_call>,
// for that functionality be invented)
virtual grpc_call_stack* call_stack() = 0;
// Return the EventEngine used for this call's async execution.
grpc_event_engine::experimental::EventEngine* event_engine() const {
return event_engine_;
}
// Implementation of EventEngine::Closure, called when deadline expires
void Run() final;
@ -162,8 +157,7 @@ class Call : public CppImplOf<Call, grpc_call>,
Call* sibling_prev = nullptr;
};
Call(bool is_client, Timestamp send_deadline, RefCountedPtr<Arena> arena,
grpc_event_engine::experimental::EventEngine* event_engine);
Call(bool is_client, Timestamp send_deadline, RefCountedPtr<Arena> arena);
~Call() override = default;
ParentCall* GetOrCreateParentCall();
@ -233,7 +227,6 @@ class Call : public CppImplOf<Call, grpc_call>,
Timestamp deadline_ ABSL_GUARDED_BY(deadline_mu_) = Timestamp::InfFuture();
grpc_event_engine::experimental::EventEngine::TaskHandle ABSL_GUARDED_BY(
deadline_mu_) deadline_task_;
grpc_event_engine::experimental::EventEngine* const event_engine_;
gpr_cycle_counter start_time_ = gpr_get_cycle_counter();
};

39
src/core/lib/surface/client_call.cc
Unescape View File

@ -45,6 +45,7 @@
#include <grpc/support/port_platform.h>
#include <grpc/support/string_util.h>
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/bitset.h"
#include "src/core/lib/gprpp/crash.h"
#include "src/core/lib/gprpp/ref_counted.h"
@ -100,14 +101,13 @@ grpc_call_error ValidateClientBatch(const grpc_op* ops, size_t nops) {
} // namespace
ClientCall::ClientCall(
grpc_call*, uint32_t, grpc_completion_queue* cq, Slice path,
absl::optional<Slice> authority, bool registered_method, Timestamp deadline,
grpc_compression_options compression_options,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena,
RefCountedPtr<UnstartedCallDestination> destination)
: Call(false, deadline, std::move(arena), event_engine),
ClientCall::ClientCall(grpc_call*, uint32_t, grpc_completion_queue* cq,
Slice path, absl::optional<Slice> authority,
bool registered_method, Timestamp deadline,
grpc_compression_options compression_options,
RefCountedPtr<Arena> arena,
RefCountedPtr<UnstartedCallDestination> destination)
: Call(false, deadline, std::move(arena)),
DualRefCounted("ClientCall"),
cq_(cq),
call_destination_(std::move(destination)),
@ -228,8 +228,7 @@ void ClientCall::StartCall(const grpc_op& send_initial_metadata_op) {
send_initial_metadata_.get());
PrepareOutgoingInitialMetadata(send_initial_metadata_op,
*send_initial_metadata_);
auto call = MakeCallPair(std::move(send_initial_metadata_), event_engine(),
arena()->Ref());
auto call = MakeCallPair(std::move(send_initial_metadata_), arena()->Ref());
started_call_initiator_ = std::move(call.initiator);
while (true) {
GRPC_TRACE_LOG(call, INFO)
@ -409,18 +408,20 @@ char* ClientCall::GetPeer() {
return gpr_strdup("unknown");
}
grpc_call* MakeClientCall(
grpc_call* parent_call, uint32_t propagation_mask,
grpc_completion_queue* cq, Slice path, absl::optional<Slice> authority,
bool registered_method, Timestamp deadline,
grpc_compression_options compression_options,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena,
RefCountedPtr<UnstartedCallDestination> destination) {
grpc_call* MakeClientCall(grpc_call* parent_call, uint32_t propagation_mask,
grpc_completion_queue* cq, Slice path,
absl::optional<Slice> authority,
bool registered_method, Timestamp deadline,
grpc_compression_options compression_options,
RefCountedPtr<Arena> arena,
RefCountedPtr<UnstartedCallDestination> destination) {
DCHECK_NE(arena.get(), nullptr);
DCHECK_NE(arena->GetContext<grpc_event_engine::experimental::EventEngine>(),
nullptr);
return arena
->New<ClientCall>(parent_call, propagation_mask, cq, std::move(path),
std::move(authority), registered_method, deadline,
compression_options, event_engine, arena, destination)
compression_options, arena, destination)
->c_ptr();
}

16
src/core/lib/surface/client_call.h
Unescape View File

@ -64,7 +64,6 @@ class ClientCall final
grpc_completion_queue* cq, Slice path,
absl::optional<Slice> authority, bool registered_method,
Timestamp deadline, grpc_compression_options compression_options,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena,
RefCountedPtr<UnstartedCallDestination> destination);
@ -168,14 +167,13 @@ class ClientCall final
std::atomic<bool> saw_trailing_metadata_{false};
};
grpc_call* MakeClientCall(
grpc_call* parent_call, uint32_t propagation_mask,
grpc_completion_queue* cq, Slice path, absl::optional<Slice> authority,
bool registered_method, Timestamp deadline,
grpc_compression_options compression_options,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena,
RefCountedPtr<UnstartedCallDestination> destination);
grpc_call* MakeClientCall(grpc_call* parent_call, uint32_t propagation_mask,
grpc_completion_queue* cq, Slice path,
absl::optional<Slice> authority,
bool registered_method, Timestamp deadline,
grpc_compression_options compression_options,
RefCountedPtr<Arena> arena,
RefCountedPtr<UnstartedCallDestination> destination);
} // namespace grpc_core

5
src/core/lib/surface/filter_stack_call.cc
Unescape View File

@ -46,6 +46,7 @@
#include "src/core/channelz/channelz.h"
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/experiments/experiments.h"
#include "src/core/lib/gprpp/crash.h"
#include "src/core/lib/gprpp/debug_location.h"
@ -82,7 +83,7 @@ using GrpcClosure = Closure;
FilterStackCall::FilterStackCall(RefCountedPtr<Arena> arena,
const grpc_call_create_args& args)
: Call(args.server_transport_data == nullptr, args.send_deadline,
std::move(arena), args.channel->event_engine()),
std::move(arena)),
channel_(args.channel->RefAsSubclass<Channel>()),
cq_(args.cq),
stream_op_payload_{} {}
@ -108,6 +109,8 @@ grpc_error_handle FilterStackCall::Create(grpc_call_create_args* args,
channel_stack->call_stack_size;
RefCountedPtr<Arena> arena = channel->call_arena_allocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
args->channel->event_engine());
call = new (arena->Alloc(call_alloc_size)) FilterStackCall(arena, *args);
DCHECK(FromC(call->c_ptr()) == call);
DCHECK(FromCallStack(call->call_stack()) == call);

2
src/core/lib/surface/server_call.h
Unescape View File

@ -69,7 +69,7 @@ class ServerCall final : public Call, public DualRefCounted<ServerCall> {
: Call(false,
client_initial_metadata->get(GrpcTimeoutMetadata())
.value_or(Timestamp::InfFuture()),
call_handler.arena()->Ref(), call_handler.event_engine()),
call_handler.arena()->Ref()),
call_handler_(std::move(call_handler)),
client_initial_metadata_stored_(std::move(client_initial_metadata)),
cq_(cq),

13
src/core/lib/transport/call_spine.cc
Unescape View File

@ -18,6 +18,7 @@
#include <grpc/support/port_platform.h>
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/promise/for_each.h"
#include "src/core/lib/promise/try_seq.h"
@ -119,12 +120,12 @@ void ForwardCall(CallHandler call_handler, CallInitiator call_initiator,
}
CallInitiatorAndHandler MakeCallPair(
ClientMetadataHandle client_initial_metadata,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena) {
CHECK_NE(arena.get(), nullptr);
auto spine = CallSpine::Create(std::move(client_initial_metadata),
event_engine, std::move(arena));
ClientMetadataHandle client_initial_metadata, RefCountedPtr<Arena> arena) {
DCHECK_NE(arena.get(), nullptr);
DCHECK_NE(arena->GetContext<grpc_event_engine::experimental::EventEngine>(),
nullptr);
auto spine =
CallSpine::Create(std::move(client_initial_metadata), std::move(arena));
return {CallInitiator(spine), UnstartedCallHandler(spine)};
}

57
src/core/lib/transport/call_spine.h
Unescape View File

@ -45,17 +45,15 @@ class CallSpine final : public Party {
public:
static RefCountedPtr<CallSpine> Create(
ClientMetadataHandle client_initial_metadata,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena) {
Arena* arena_ptr = arena.get();
return RefCountedPtr<CallSpine>(arena_ptr->New<CallSpine>(
std::move(client_initial_metadata), event_engine, std::move(arena)));
std::move(client_initial_metadata), std::move(arena)));
}
~CallSpine() override {}
CallFilters& call_filters() { return call_filters_; }
Arena* arena() { return arena_.get(); }
// Add a callback to be called when server trailing metadata is received.
void OnDone(absl::AnyInvocable<void()> fn) {
@ -116,10 +114,6 @@ class CallSpine final : public Party {
return *call_filters().unprocessed_client_initial_metadata();
}
grpc_event_engine::experimental::EventEngine* event_engine() const override {
return event_engine_;
}
// Wrap a promise so that if it returns failure it automatically cancels
// the rest of the call.
// The resulting (returned) promise will resolve to Empty.
@ -190,45 +184,12 @@ class CallSpine final : public Party {
private:
friend class Arena;
CallSpine(ClientMetadataHandle client_initial_metadata,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena)
: Party(1),
arena_(std::move(arena)),
call_filters_(std::move(client_initial_metadata)),
event_engine_(event_engine) {}
class ScopedContext : public ScopedActivity,
public promise_detail::Context<Arena>,
public promise_detail::Context<
grpc_event_engine::experimental::EventEngine> {
public:
explicit ScopedContext(CallSpine* spine)
: ScopedActivity(spine),
Context<Arena>(spine->arena_.get()),
Context<grpc_event_engine::experimental::EventEngine>(
spine->event_engine()) {}
};
bool RunParty() override {
ScopedContext context(this);
return Party::RunParty();
}
void PartyOver() override {
auto arena = arena_;
{
ScopedContext context(this);
CancelRemainingParticipants();
arena->DestroyManagedNewObjects();
}
this->~CallSpine();
}
: Party(std::move(arena)),
call_filters_(std::move(client_initial_metadata)) {}
const RefCountedPtr<Arena> arena_;
// Call filters/pipes part of the spine
CallFilters call_filters_;
// Event engine associated with this call
grpc_event_engine::experimental::EventEngine* const event_engine_;
absl::AnyInvocable<void()> on_done_{nullptr};
};
@ -292,10 +253,6 @@ class CallInitiator {
Arena* arena() { return spine_->arena(); }
Party* party() { return spine_.get(); }
grpc_event_engine::experimental::EventEngine* event_engine() const {
return spine_->event_engine();
}
private:
RefCountedPtr<CallSpine> spine_;
};
@ -357,10 +314,6 @@ class CallHandler {
Arena* arena() { return spine_->arena(); }
Party* party() { return spine_.get(); }
grpc_event_engine::experimental::EventEngine* event_engine() const {
return spine_->event_engine();
}
private:
RefCountedPtr<CallSpine> spine_;
};
@ -428,9 +381,7 @@ struct CallInitiatorAndHandler {
};
CallInitiatorAndHandler MakeCallPair(
ClientMetadataHandle client_initial_metadata,
grpc_event_engine::experimental::EventEngine* event_engine,
RefCountedPtr<Arena> arena);
ClientMetadataHandle client_initial_metadata, RefCountedPtr<Arena> arena);
template <typename CallHalf>
auto OutgoingMessages(CallHalf h) {

3
src/core/lib/transport/interception_chain.cc
Unescape View File

@ -39,8 +39,7 @@ CallInitiator HijackedCall::MakeCall() {
CallInitiator HijackedCall::MakeCallWithMetadata(
ClientMetadataHandle metadata) {
auto call = MakeCallPair(std::move(metadata), call_handler_.event_engine(),
call_handler_.arena()->Ref());
auto call = MakeCallPair(std::move(metadata), call_handler_.arena()->Ref());
destination_->StartCall(std::move(call.handler));
return std::move(call.initiator);
}

7
test/core/call/bm_client_call.cc
Unescape View File

@ -17,6 +17,7 @@
#include <grpc/grpc.h>
#include "src/core/lib/event_engine/default_event_engine.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/promise/all_ok.h"
#include "src/core/lib/promise/map.h"
#include "src/core/lib/resource_quota/arena.h"
@ -58,11 +59,13 @@ class Helper {
}
auto MakeCall() {
auto arena = arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
return std::unique_ptr<grpc_call, void (*)(grpc_call*)>(
MakeClientCall(nullptr, 0, cq_, path_.Copy(), absl::nullopt, true,
Timestamp::InfFuture(), compression_options_,
event_engine_.get(), arena_allocator_->MakeArena(),
destination_),
std::move(arena), destination_),
grpc_call_unref);
}

12
test/core/call/client_call_test.cc
Unescape View File

@ -67,11 +67,13 @@ class ClientCallTest : public YodelTest {
grpc_call* InitCall(const CallOptions& options) {
CHECK_EQ(call_, nullptr);
call_ = MakeClientCall(nullptr, 0, cq_, options.path(), options.authority(),
options.registered_method(),
options.timeout() + Timestamp::Now(),
options.compression_options(), event_engine().get(),
SimpleArenaAllocator()->MakeArena(), destination_);
auto arena = SimpleArenaAllocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine().get());
call_ = MakeClientCall(
nullptr, 0, cq_, options.path(), options.authority(),
options.registered_method(), options.timeout() + Timestamp::Now(),
options.compression_options(), std::move(arena), destination_);
return call_;
}

6
test/core/call/server_call_test.cc
Unescape View File

@ -44,9 +44,11 @@ class ServerCallTest : public YodelTest {
grpc_call* InitCall(ClientMetadataHandle client_initial_metadata) {
CHECK_EQ(call_, nullptr);
auto arena = SimpleArenaAllocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine().get());
auto call =
MakeCallPair(std::move(client_initial_metadata), event_engine().get(),
SimpleArenaAllocator()->MakeArena());
MakeCallPair(std::move(client_initial_metadata), std::move(arena));
call.initiator.SpawnGuarded(
"initial_metadata",
[this, handler = call.handler.StartCall()]() mutable {

8
test/core/call/yodel/yodel_test.h
Unescape View File

@ -23,6 +23,7 @@
#include <grpc/event_engine/event_engine.h>
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/debug_location.h"
#include "src/core/lib/promise/cancel_callback.h"
#include "src/core/lib/promise/detail/promise_factory.h"
@ -363,9 +364,10 @@ class YodelTest : public ::testing::Test {
}
auto MakeCall(ClientMetadataHandle client_initial_metadata) {
return MakeCallPair(std::move(client_initial_metadata),
state_->event_engine.get(),
state_->call_arena_allocator->MakeArena());
auto arena = state_->call_arena_allocator->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
state_->event_engine.get());
return MakeCallPair(std::move(client_initial_metadata), std::move(arena));
}
void WaitForAllPendingWork();

10
test/core/client_channel/client_channel_test.cc
Unescape View File

@ -267,8 +267,9 @@ CLIENT_CHANNEL_TEST(NoOp) { InitChannel(ChannelArgs()); }
CLIENT_CHANNEL_TEST(StartCall) {
auto& channel = InitChannel(ChannelArgs());
auto call = MakeCallPair(MakeClientInitialMetadata(), channel.event_engine(),
channel.call_arena_allocator()->MakeArena());
auto arena = channel.call_arena_allocator()->MakeArena();
arena->SetContext<EventEngine>(channel.event_engine());
auto call = MakeCallPair(MakeClientInitialMetadata(), std::move(arena));
channel.StartCall(std::move(call.handler));
QueueNameResolutionResult(
MakeSuccessfulResolutionResult("ipv4:127.0.0.1:1234"));
@ -337,8 +338,9 @@ class TestConfigSelector : public ConfigSelector {
CLIENT_CHANNEL_TEST(ConfigSelectorWithDynamicFilters) {
auto& channel = InitChannel(ChannelArgs());
auto call = MakeCallPair(MakeClientInitialMetadata(), channel.event_engine(),
channel.call_arena_allocator()->MakeArena());
auto arena = channel.call_arena_allocator()->MakeArena();
arena->SetContext<EventEngine>(channel.event_engine());
auto call = MakeCallPair(MakeClientInitialMetadata(), std::move(arena));
channel.StartCall(std::move(call.handler));
auto service_config = ServiceConfigImpl::Create(ChannelArgs(), "{}");
ASSERT_TRUE(service_config.ok());

7
test/core/client_channel/connected_subchannel_test.cc
Unescape View File

@ -67,9 +67,10 @@ class ConnectedSubchannelTest : public YodelTest {
CallInitiatorAndHandler MakeCall(
ClientMetadataHandle client_initial_metadata) {
return MakeCallPair(std::move(client_initial_metadata),
event_engine().get(),
SimpleArenaAllocator()->MakeArena());
auto arena = SimpleArenaAllocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine().get());
return MakeCallPair(std::move(client_initial_metadata), std::move(arena));
}
CallHandler TickUntilCallStarted() {

6
test/core/client_channel/load_balanced_call_destination_test.cc
Unescape View File

@ -49,9 +49,9 @@ class LoadBalancedCallDestinationTest : public YodelTest {
CallInitiatorAndHandler MakeCall(
ClientMetadataHandle client_initial_metadata) {
return MakeCallPair(std::move(client_initial_metadata),
event_engine().get(),
call_arena_allocator_->MakeArena());
auto arena = call_arena_allocator_->MakeArena();
arena->SetContext<EventEngine>(event_engine().get());
return MakeCallPair(std::move(client_initial_metadata), std::move(arena));
}
CallHandler TickUntilCallStarted() {

45
test/core/promise/bm_party.cc
Unescape View File

@ -18,63 +18,30 @@
#include "src/core/lib/event_engine/default_event_engine.h"
#include "src/core/lib/promise/party.h"
#include "src/core/lib/resource_quota/arena.h"
namespace grpc_core {
namespace {
class TestParty final : public Party {
public:
TestParty() : Party(1) {}
~TestParty() override {}
std::string DebugTag() const override { return "TestParty"; }
using Party::IncrementRefCount;
using Party::Unref;
bool RunParty() override {
promise_detail::Context<grpc_event_engine::experimental::EventEngine>
ee_ctx(ee_.get());
return Party::RunParty();
}
void PartyOver() override {
{
promise_detail::Context<grpc_event_engine::experimental::EventEngine>
ee_ctx(ee_.get());
CancelRemainingParticipants();
}
delete this;
}
private:
grpc_event_engine::experimental::EventEngine* event_engine() const final {
return ee_.get();
}
std::shared_ptr<grpc_event_engine::experimental::EventEngine> ee_ =
grpc_event_engine::experimental::GetDefaultEventEngine();
};
void BM_PartyCreate(benchmark::State& state) {
auto arena = SimpleArenaAllocator()->MakeArena();
for (auto _ : state) {
auto* party = new TestParty();
party->Unref();
Party::Make(arena);
}
}
BENCHMARK(BM_PartyCreate);
void BM_AddParticipant(benchmark::State& state) {
auto* party = new TestParty();
auto party = Party::Make(SimpleArenaAllocator()->MakeArena());
for (auto _ : state) {
party->Spawn(
"participant", []() { return Success{}; }, [](StatusFlag) {});
}
party->Unref();
}
BENCHMARK(BM_AddParticipant);
void BM_WakeupParticipant(benchmark::State& state) {
auto* party = new TestParty();
auto party = Party::Make(SimpleArenaAllocator()->MakeArena());
party->Spawn(
"driver",
[&state, w = IntraActivityWaiter()]() mutable -> Poll<StatusFlag> {
@ -85,7 +52,7 @@ void BM_WakeupParticipant(benchmark::State& state) {
}
return Success{};
},
[party](StatusFlag) { party->Unref(); });
[party](StatusFlag) {});
}
BENCHMARK(BM_WakeupParticipant);

93
test/core/promise/party_test.cc
Unescape View File

@ -31,6 +31,7 @@
#include <grpc/grpc.h>
#include "src/core/lib/event_engine/default_event_engine.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/notification.h"
#include "src/core/lib/gprpp/ref_counted_ptr.h"
#include "src/core/lib/gprpp/sync.h"
@ -41,6 +42,7 @@
#include "src/core/lib/promise/poll.h"
#include "src/core/lib/promise/seq.h"
#include "src/core/lib/promise/sleep.h"
#include "src/core/lib/resource_quota/arena.h"
#include "src/core/lib/resource_quota/memory_quota.h"
#include "src/core/lib/resource_quota/resource_quota.h"
@ -230,52 +232,28 @@ TYPED_TEST(PartySyncTest, UnrefWhileRunning) {
///////////////////////////////////////////////////////////////////////////////
// PartyTest
class TestParty final : public Party {
public:
TestParty() : Party(1) {}
~TestParty() override {}
std::string DebugTag() const override { return "TestParty"; }
using Party::IncrementRefCount;
using Party::Unref;
bool RunParty() override {
promise_detail::Context<grpc_event_engine::experimental::EventEngine>
ee_ctx(ee_.get());
return Party::RunParty();
}
void PartyOver() override {
{
promise_detail::Context<grpc_event_engine::experimental::EventEngine>
ee_ctx(ee_.get());
CancelRemainingParticipants();
}
delete this;
class PartyTest : public ::testing::Test {
protected:
RefCountedPtr<Party> MakeParty() {
auto arena = SimpleArenaAllocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
return Party::Make(std::move(arena));
}
private:
grpc_event_engine::experimental::EventEngine* event_engine() const final {
return ee_.get();
}
std::shared_ptr<grpc_event_engine::experimental::EventEngine> ee_ =
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_ =
grpc_event_engine::experimental::GetDefaultEventEngine();
};
class PartyTest : public ::testing::Test {
protected:
};
TEST_F(PartyTest, Noop) { auto party = MakeRefCounted<TestParty>(); }
TEST_F(PartyTest, Noop) { auto party = MakeParty(); }
TEST_F(PartyTest, CanSpawnAndRun) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification n;
party->Spawn(
"TestSpawn",
[i = 10]() mutable -> Poll<int> {
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
EXPECT_GT(i, 0);
GetContext<Activity>()->ForceImmediateRepoll();
--i;
@ -290,8 +268,8 @@ TEST_F(PartyTest, CanSpawnAndRun) {
}
TEST_F(PartyTest, CanSpawnWaitableAndRun) {
auto party1 = MakeRefCounted<TestParty>();
auto party2 = MakeRefCounted<TestParty>();
auto party1 = MakeParty();
auto party2 = MakeParty();
Notification n;
InterActivityLatch<void> done;
// Spawn a task on party1 that will wait for a task on party2.
@ -315,17 +293,15 @@ TEST_F(PartyTest, CanSpawnWaitableAndRun) {
}
TEST_F(PartyTest, CanSpawnFromSpawn) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification n1;
Notification n2;
party->Spawn(
"TestSpawn",
[party, &n2]() -> Poll<int> {
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
party->Spawn(
"TestSpawnInner",
[i = 10]() mutable -> Poll<int> {
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
GetContext<Activity>()->ForceImmediateRepoll();
--i;
if (i == 0) return 42;
@ -346,14 +322,13 @@ TEST_F(PartyTest, CanSpawnFromSpawn) {
}
TEST_F(PartyTest, CanWakeupWithOwningWaker) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification n[10];
Notification complete;
Waker waker;
party->Spawn(
"TestSpawn",
[i = 0, &waker, &n]() mutable -> Poll<int> {
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
waker = GetContext<Activity>()->MakeOwningWaker();
n[i].Notify();
i++;
@ -372,14 +347,13 @@ TEST_F(PartyTest, CanWakeupWithOwningWaker) {
}
TEST_F(PartyTest, CanWakeupWithNonOwningWaker) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification n[10];
Notification complete;
Waker waker;
party->Spawn(
"TestSpawn",
[i = 10, &waker, &n]() mutable -> Poll<int> {
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
waker = GetContext<Activity>()->MakeNonOwningWaker();
--i;
n[9 - i].Notify();
@ -399,14 +373,13 @@ TEST_F(PartyTest, CanWakeupWithNonOwningWaker) {
}
TEST_F(PartyTest, CanWakeupWithNonOwningWakerAfterOrphaning) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification set_waker;
Waker waker;
party->Spawn(
"TestSpawn",
[&waker, &set_waker]() mutable -> Poll<int> {
EXPECT_FALSE(set_waker.HasBeenNotified());
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
waker = GetContext<Activity>()->MakeNonOwningWaker();
set_waker.Notify();
return Pending{};
@ -420,14 +393,13 @@ TEST_F(PartyTest, CanWakeupWithNonOwningWakerAfterOrphaning) {
}
TEST_F(PartyTest, CanDropNonOwningWakeAfterOrphaning) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification set_waker;
std::unique_ptr<Waker> waker;
party->Spawn(
"TestSpawn",
[&waker, &set_waker]() mutable -> Poll<int> {
EXPECT_FALSE(set_waker.HasBeenNotified());
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
waker = std::make_unique<Waker>(
GetContext<Activity>()->MakeNonOwningWaker());
set_waker.Notify();
@ -441,14 +413,13 @@ TEST_F(PartyTest, CanDropNonOwningWakeAfterOrphaning) {
}
TEST_F(PartyTest, CanWakeupNonOwningOrphanedWakerWithNoEffect) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification set_waker;
Waker waker;
party->Spawn(
"TestSpawn",
[&waker, &set_waker]() mutable -> Poll<int> {
EXPECT_FALSE(set_waker.HasBeenNotified());
EXPECT_EQ(GetContext<Activity>()->DebugTag(), "TestParty");
waker = GetContext<Activity>()->MakeNonOwningWaker();
set_waker.Notify();
return Pending{};
@ -462,7 +433,7 @@ TEST_F(PartyTest, CanWakeupNonOwningOrphanedWakerWithNoEffect) {
}
TEST_F(PartyTest, CanBulkSpawn) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification n1;
Notification n2;
{
@ -481,7 +452,7 @@ TEST_F(PartyTest, CanBulkSpawn) {
}
TEST_F(PartyTest, AfterCurrentPollWorks) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
Notification n;
int state = 0;
{
@ -529,7 +500,7 @@ TEST_F(PartyTest, AfterCurrentPollWorks) {
}
TEST_F(PartyTest, ThreadStressTest) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
std::vector<std::thread> threads;
threads.reserve(8);
for (int i = 0; i < 8; i++) {
@ -599,7 +570,7 @@ class PromiseNotification {
};
TEST_F(PartyTest, ThreadStressTestWithOwningWaker) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
std::vector<std::thread> threads;
threads.reserve(8);
for (int i = 0; i < 8; i++) {
@ -627,7 +598,7 @@ TEST_F(PartyTest, ThreadStressTestWithOwningWaker) {
}
TEST_F(PartyTest, ThreadStressTestWithOwningWakerHoldingLock) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
std::vector<std::thread> threads;
threads.reserve(8);
for (int i = 0; i < 8; i++) {
@ -655,7 +626,7 @@ TEST_F(PartyTest, ThreadStressTestWithOwningWakerHoldingLock) {
}
TEST_F(PartyTest, ThreadStressTestWithNonOwningWaker) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
std::vector<std::thread> threads;
threads.reserve(8);
for (int i = 0; i < 8; i++) {
@ -683,7 +654,7 @@ TEST_F(PartyTest, ThreadStressTestWithNonOwningWaker) {
}
TEST_F(PartyTest, ThreadStressTestWithOwningWakerNoSleep) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
std::vector<std::thread> threads;
threads.reserve(8);
for (int i = 0; i < 8; i++) {
@ -709,7 +680,7 @@ TEST_F(PartyTest, ThreadStressTestWithOwningWakerNoSleep) {
}
TEST_F(PartyTest, ThreadStressTestWithNonOwningWakerNoSleep) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
std::vector<std::thread> threads;
threads.reserve(8);
for (int i = 0; i < 8; i++) {
@ -735,7 +706,7 @@ TEST_F(PartyTest, ThreadStressTestWithNonOwningWakerNoSleep) {
}
TEST_F(PartyTest, ThreadStressTestWithInnerSpawn) {
auto party = MakeRefCounted<TestParty>();
auto party = MakeParty();
std::vector<std::thread> threads;
threads.reserve(8);
for (int i = 0; i < 8; i++) {
@ -777,9 +748,9 @@ TEST_F(PartyTest, ThreadStressTestWithInnerSpawn) {
}
TEST_F(PartyTest, NestedWakeup) {
auto party1 = MakeRefCounted<TestParty>();
auto party2 = MakeRefCounted<TestParty>();
auto party3 = MakeRefCounted<TestParty>();
auto party1 = MakeParty();
auto party2 = MakeParty();
auto party3 = MakeParty();
int whats_going_on = 0;
Notification started2;
Notification done2;

20
test/core/promise/sleep_test.cc
Unescape View File

@ -27,11 +27,13 @@
#include <grpc/grpc.h>
#include "src/core/lib/event_engine/default_event_engine.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/notification.h"
#include "src/core/lib/gprpp/orphanable.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/promise/exec_ctx_wakeup_scheduler.h"
#include "src/core/lib/promise/race.h"
#include "src/core/lib/resource_quota/arena.h"
#include "test/core/event_engine/mock_event_engine.h"
#include "test/core/promise/test_wakeup_schedulers.h"
@ -49,6 +51,12 @@ using testing::StrictMock;
namespace grpc_core {
namespace {
RefCountedPtr<Arena> ArenaWithEventEngine(EventEngine* ee) {
auto arena = SimpleArenaAllocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(ee);
return arena;
}
TEST(Sleep, Zzzz) {
ExecCtx exec_ctx;
Notification done;
@ -61,7 +69,7 @@ TEST(Sleep, Zzzz) {
EXPECT_EQ(r, absl::OkStatus());
done.Notify();
},
engine.get());
ArenaWithEventEngine(engine.get()));
done.WaitForNotification();
exec_ctx.InvalidateNow();
EXPECT_GE(Timestamp::Now(), done_time);
@ -84,7 +92,7 @@ TEST(Sleep, OverlyEagerEventEngine) {
EXPECT_EQ(r, absl::OkStatus());
done = true;
},
static_cast<EventEngine*>(&mock_event_engine));
ArenaWithEventEngine(static_cast<EventEngine*>(&mock_event_engine)));
Mock::VerifyAndClearExpectations(&mock_event_engine);
EXPECT_NE(wakeup, nullptr);
EXPECT_FALSE(done);
@ -106,7 +114,7 @@ TEST(Sleep, AlreadyDone) {
EXPECT_EQ(r, absl::OkStatus());
done.Notify();
},
engine.get());
ArenaWithEventEngine(engine.get()));
done.WaitForNotification();
}
@ -123,7 +131,7 @@ TEST(Sleep, Cancel) {
EXPECT_EQ(r, absl::CancelledError());
done.Notify();
},
engine.get());
ArenaWithEventEngine(engine.get()));
done.WaitForNotification();
exec_ctx.InvalidateNow();
EXPECT_LT(Timestamp::Now(), done_time);
@ -143,7 +151,7 @@ TEST(Sleep, MoveSemantics) {
EXPECT_EQ(r, absl::OkStatus());
done.Notify();
},
engine.get());
ArenaWithEventEngine(engine.get()));
done.WaitForNotification();
exec_ctx.InvalidateNow();
EXPECT_GE(Timestamp::Now(), done_time);
@ -163,7 +171,7 @@ TEST(Sleep, StressTest) {
Sleep(Timestamp::Now() + Duration::Seconds(1)),
ExecCtxWakeupScheduler(),
[notification](absl::Status /*r*/) { notification->Notify(); },
engine.get());
ArenaWithEventEngine(engine.get()));
notifications.push_back(std::move(notification));
activities.push_back(std::move(activity));
}

8
test/core/surface/channel_init_test.cc
Unescape View File

@ -334,8 +334,12 @@ TEST(ChannelInitTest, CanCreateFilterWithCall) {
ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator(
"test"),
1024);
auto call = MakeCallPair(Arena::MakePooled<ClientMetadata>(), nullptr,
allocator->MakeArena());
auto event_engine = grpc_event_engine::experimental::GetDefaultEventEngine();
auto arena = allocator->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine.get());
auto call =
MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena));
(*stack)->StartCall(std::move(call.handler));
EXPECT_EQ(p, 1);
EXPECT_EQ(handled, 1);

2
test/core/transport/BUILD
Unescape View File

@ -65,7 +65,7 @@ grpc_cc_test(
uses_event_engine = False,
uses_polling = False,
deps = [
"//:grpc_base",
"//:grpc",
"//src/core:interception_chain",
"//src/core:resource_quota",
"//test/core/promise:poll_matcher",

21
test/core/transport/bm_call_spine.cc
Unescape View File

@ -25,8 +25,11 @@ namespace grpc_core {
class CallSpineFixture {
public:
BenchmarkCall MakeCall() {
auto p = MakeCallPair(Arena::MakePooled<ClientMetadata>(),
event_engine_.get(), arena_allocator_->MakeArena());
auto arena = arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
auto p =
MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena));
return {std::move(p.initiator), p.handler.StartCall()};
}
@ -54,10 +57,16 @@ GRPC_CALL_SPINE_BENCHMARK(CallSpineFixture);
class ForwardCallFixture {
public:
BenchmarkCall MakeCall() {
auto p1 = MakeCallPair(Arena::MakePooled<ClientMetadata>(),
event_engine_.get(), arena_allocator_->MakeArena());
auto p2 = MakeCallPair(Arena::MakePooled<ClientMetadata>(),
event_engine_.get(), arena_allocator_->MakeArena());
auto arena1 = arena_allocator_->MakeArena();
auto arena2 = arena_allocator_->MakeArena();
arena1->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
arena2->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
auto p1 =
MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena1));
auto p2 =
MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena2));
p1.handler.SpawnInfallible("initial_metadata", [&]() {
auto p1_handler = p1.handler.StartCall();
return Map(

16
test/core/transport/call_spine_benchmarks.h
Unescape View File

@ -20,6 +20,7 @@
#include "benchmark/benchmark.h"
#include "src/core/lib/event_engine/default_event_engine.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/notification.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/promise/all_ok.h"
@ -237,9 +238,11 @@ template <class Traits>
class FilterFixture {
public:
BenchmarkCall MakeCall() {
auto p = MakeCallPair(traits_.MakeClientInitialMetadata(),
event_engine_.get(), arena_allocator_->MakeArena());
p.handler.AddCallStack(stack_);
auto arena = arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
auto p =
MakeCallPair(traits_.MakeClientInitialMetadata(), std::move(arena));
return {std::move(p.initiator), p.handler.StartCall()};
}
@ -278,8 +281,11 @@ template <class Traits>
class UnstartedCallDestinationFixture {
public:
BenchmarkCall MakeCall() {
auto p = MakeCallPair(traits_->MakeClientInitialMetadata(),
event_engine_.get(), arena_allocator_->MakeArena());
auto arena = arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
auto p =
MakeCallPair(traits_->MakeClientInitialMetadata(), std::move(arena));
top_destination_->StartCall(std::move(p.handler));
auto handler = bottom_destination_->TakeHandler();
absl::optional<CallHandler> started_handler;

7
test/core/transport/call_spine_test.cc
Unescape View File

@ -49,9 +49,10 @@ class CallSpineTest : public YodelTest {
CallInitiatorAndHandler MakeCall(
ClientMetadataHandle client_initial_metadata) {
return MakeCallPair(std::move(client_initial_metadata),
event_engine().get(),
SimpleArenaAllocator()->MakeArena());
auto arena = SimpleArenaAllocator()->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine().get());
return MakeCallPair(std::move(client_initial_metadata), std::move(arena));
}
void UnaryRequest(CallInitiator initiator, CallHandler handler);

7
test/core/transport/chaotic_good/client_transport_error_test.cc
Unescape View File

@ -38,6 +38,7 @@
#include "src/core/ext/transport/chaotic_good/client_transport.h"
#include "src/core/lib/config/core_configuration.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/gprpp/ref_counted_ptr.h"
#include "src/core/lib/iomgr/timer_manager.h"
#include "src/core/lib/promise/activity.h"
@ -141,8 +142,10 @@ class ClientTransportTest : public ::testing::Test {
}
auto MakeCall(ClientMetadataHandle client_initial_metadata) {
return MakeCallPair(std::move(client_initial_metadata), event_engine_.get(),
call_arena_allocator_->MakeArena());
auto arena = call_arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
return MakeCallPair(std::move(client_initial_metadata), std::move(arena));
}
private:

9
test/core/transport/chaotic_good/transport_test.h
Unescape View File

@ -19,6 +19,7 @@
#include "gtest/gtest.h"
#include "src/core/ext/transport/chaotic_good/frame.h"
#include "src/core/lib/event_engine/event_engine_context.h"
#include "src/core/lib/iomgr/timer_manager.h"
#include "src/core/lib/resource_quota/memory_quota.h"
#include "src/core/lib/resource_quota/resource_quota.h"
@ -37,7 +38,10 @@ class TransportTest : public ::testing::Test {
}
RefCountedPtr<Arena> MakeArena() {
return call_arena_allocator_->MakeArena();
auto arena = call_arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
return arena;
}
RefCountedPtr<CallArenaAllocator> call_arena_allocator() {
@ -45,8 +49,7 @@ class TransportTest : public ::testing::Test {
}
auto MakeCall(ClientMetadataHandle client_initial_metadata) {
return MakeCallPair(std::move(client_initial_metadata), event_engine_.get(),
MakeArena());
return MakeCallPair(std::move(client_initial_metadata), MakeArena());
}
private:

16
test/core/transport/interception_chain_test.cc
Unescape View File

@ -20,6 +20,8 @@
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include <grpc/grpc.h>
#include "src/core/lib/channel/promise_based_filter.h"
#include "src/core/lib/resource_quota/resource_quota.h"
#include "test/core/promise/poll_matcher.h"
@ -247,8 +249,11 @@ class InterceptionChainTest : public ::testing::Test {
// Run a call through a UnstartedCallDestination until it's complete.
FinishedCall RunCall(UnstartedCallDestination* destination) {
auto call = MakeCallPair(Arena::MakePooled<ClientMetadata>(), nullptr,
call_arena_allocator_->MakeArena());
auto arena = call_arena_allocator_->MakeArena();
arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine_.get());
auto call =
MakeCallPair(Arena::MakePooled<ClientMetadata>(), std::move(arena));
Poll<ServerMetadataHandle> trailing_md;
call.initiator.SpawnInfallible(
"run_call", [destination, &call, &trailing_md]() mutable {
@ -287,6 +292,8 @@ class InterceptionChainTest : public ::testing::Test {
private:
ClientMetadataHandle metadata_;
};
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_ =
grpc_event_engine::experimental::GetDefaultEventEngine();
RefCountedPtr<Destination> destination_ = MakeRefCounted<Destination>();
RefCountedPtr<CallArenaAllocator> call_arena_allocator_ =
MakeRefCounted<CallArenaAllocator>(
@ -437,5 +444,8 @@ int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
grpc_tracer_init();
gpr_log_verbosity_init();
return RUN_ALL_TESTS();
grpc_init();
auto r = RUN_ALL_TESTS();
grpc_shutdown();
return r;
}

1
tools/distrib/fix_build_deps.py
Unescape View File

@ -433,6 +433,7 @@ for dirname in [
"grpc_cc_binary": grpc_cc_library,
"grpc_cc_library": grpc_cc_library,
"grpc_cc_test": grpc_cc_library,
"grpc_cc_benchmark": grpc_cc_library,
"grpc_core_end2end_test": lambda **kwargs: None,
"grpc_filegroup": lambda **kwargs: None,
"grpc_transport_test": lambda **kwargs: None,

Write Preview
Loading…
Cancel
Save