From 250b8d2dae59d8fdbc051eb9296dcc6dc918a676 Mon Sep 17 00:00:00 2001 From: AJ Heller Date: Mon, 16 May 2022 10:32:07 -0700 Subject: [PATCH] IomgrEventEngine Redux (#29693) * Revert "Revert "IomgrEventEngine (#29616)" (#29692)" This reverts commit 246d13e3928efb86dcba620a7a9fdff70e76e083. * temporarily disable EE usage to coordinate landing * spelling --- BUILD | 121 +++++++++- CMakeLists.txt | 58 ++++- Makefile | 4 + build_autogenerated.yaml | 28 ++- config.m4 | 2 + config.w32 | 2 + gRPC-C++.podspec | 8 + gRPC-Core.podspec | 10 + grpc.gemspec | 8 + grpc.gyp | 8 + include/grpc/event_engine/event_engine.h | 2 +- package.xml | 8 + .../default_event_engine_factory.cc | 6 +- src/core/lib/event_engine/event_engine.cc | 9 + .../lib/event_engine/event_engine_factory.h | 3 + src/core/lib/event_engine/handle_containers.h | 67 ++++++ src/core/lib/event_engine/iomgr_engine.cc | 206 ++++++++++++++++++ src/core/lib/event_engine/iomgr_engine.h | 118 ++++++++++ src/core/lib/event_engine/trace.cc | 18 ++ src/core/lib/event_engine/trace.h | 30 +++ src/python/grpcio/grpc_core_dependencies.py | 2 + test/core/event_engine/test_init.cc | 9 +- test/core/event_engine/test_suite/BUILD | 11 +- test/core/event_engine/test_suite/README.md | 2 +- .../test_suite/iomgr_event_engine_test.cc | 31 +++ .../event_engine/test_suite/timer_test.cc | 6 +- test/core/transport/bdp_estimator_test.cc | 7 +- tools/doxygen/Doxyfile.c++.internal | 5 + tools/doxygen/Doxyfile.core.internal | 5 + tools/run_tests/generated/tests.json | 24 ++ 30 files changed, 786 insertions(+), 32 deletions(-) create mode 100644 src/core/lib/event_engine/handle_containers.h create mode 100644 src/core/lib/event_engine/iomgr_engine.cc create mode 100644 src/core/lib/event_engine/iomgr_engine.h create mode 100644 src/core/lib/event_engine/trace.cc create mode 100644 src/core/lib/event_engine/trace.h create mode 100644 test/core/event_engine/test_suite/iomgr_event_engine_test.cc diff --git a/BUILD b/BUILD index 604dad10bfd..4cab5759cc0 100644 --- a/BUILD +++ b/BUILD @@ -399,6 +399,7 @@ grpc_cc_library( "channel_init", "channel_stack_type", "config", + "default_event_engine_factory_hdrs", "gpr_base", "grpc_authorization_base", "grpc_base", @@ -407,6 +408,7 @@ grpc_cc_library( "grpc_security_base", "grpc_trace", "http_connect_handshaker", + "iomgr_timer", "slice", "tcp_connect_handshaker", ], @@ -455,6 +457,7 @@ grpc_cc_library( "channel_init", "channel_stack_type", "config", + "default_event_engine_factory_hdrs", "gpr_base", "grpc_authorization_base", "grpc_base", @@ -464,6 +467,7 @@ grpc_cc_library( "grpc_security_base", "grpc_trace", "http_connect_handshaker", + "iomgr_timer", "slice", "tcp_connect_handshaker", ], @@ -1514,6 +1518,7 @@ grpc_cc_library( "grpc_base", "grpc_codegen", "grpc_trace", + "iomgr_timer", "slice", ], ) @@ -1885,6 +1890,41 @@ grpc_cc_library( ], ) +grpc_cc_library( + name = "iomgr_timer", + srcs = [ + "src/core/lib/iomgr/time_averaged_stats.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", + ], + hdrs = [ + "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/time_averaged_stats.h", + ] + [ + # TODO(hork): deduplicate + "src/core/lib/iomgr/iomgr.h", + ], + external_deps = [ + "absl/strings", + ], + deps = [ + "event_engine_base_hdrs", + "exec_ctx", + "gpr_base", + "gpr_platform", + "gpr_tls", + "grpc_trace", + "iomgr_port", + "time", + "useful", + ], +) + grpc_cc_library( name = "iomgr_fwd", hdrs = [ @@ -1954,13 +1994,39 @@ grpc_cc_library( srcs = [ "src/core/lib/event_engine/default_event_engine_factory.cc", ], + external_deps = ["absl/memory"], + deps = [ + "default_event_engine_factory_hdrs", + "event_engine_base_hdrs", + "gpr_base", + "iomgr_event_engine", + "iomgr_port", + ], +) + +grpc_cc_library( + name = "iomgr_event_engine", + srcs = ["src/core/lib/event_engine/iomgr_engine.cc"], + hdrs = ["src/core/lib/event_engine/iomgr_engine.h"], external_deps = [ - # TODO(hork): uv, in a subsequent PR + "absl/cleanup", + "absl/container:flat_hash_set", + "absl/time", + "absl/strings", ], deps = [ - "default_event_engine_factory_hdrs", + "closure", + "error", "event_engine_base_hdrs", + "event_engine_common", + "event_engine_trace", + "exec_ctx", "gpr_base", + "gpr_platform", + "grpc_trace", + "iomgr_timer", + "match", + "time", ], ) @@ -1971,8 +2037,15 @@ grpc_cc_library( "src/core/lib/event_engine/slice.cc", "src/core/lib/event_engine/slice_buffer.cc", ], + hdrs = [ + "src/core/lib/event_engine/handle_containers.h", + ], + external_deps = [ + "absl/container:flat_hash_set", + ], deps = [ "event_engine_base_hdrs", + "event_engine_trace", "gpr_base", "gpr_platform", "ref_counted", @@ -1981,6 +2054,20 @@ grpc_cc_library( ], ) +grpc_cc_library( + name = "event_engine_trace", + srcs = [ + "src/core/lib/event_engine/trace.cc", + ], + hdrs = [ + "src/core/lib/event_engine/trace.h", + ], + deps = [ + "gpr_platform", + "grpc_trace", + ], +) + grpc_cc_library( name = "event_engine_base", srcs = [ @@ -1990,7 +2077,9 @@ grpc_cc_library( "default_event_engine_factory", "default_event_engine_factory_hdrs", "event_engine_base_hdrs", + "event_engine_trace", "gpr_base", + "grpc_trace", ], ) @@ -2120,11 +2209,6 @@ grpc_cc_library( "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/time_averaged_stats.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/wakeup_fd_eventfd.cc", @@ -2247,11 +2331,6 @@ grpc_cc_library( "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/time_averaged_stats.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/wakeup_fd_pipe.h", "src/core/lib/iomgr/wakeup_fd_posix.h", @@ -2358,6 +2437,7 @@ grpc_cc_library( "grpc_trace", "handshaker_registry", "iomgr_port", + "iomgr_timer", "json", "latch", "memory_quota", @@ -2733,6 +2813,7 @@ grpc_cc_library( "http_connect_handshaker", "httpcli", "iomgr_fwd", + "iomgr_timer", "json", "json_util", "orphanable", @@ -2817,6 +2898,7 @@ grpc_cc_library( "gpr_base", "grpc_base", "idle_filter_state", + "iomgr_timer", "loop", "promise", "single_set_ptr", @@ -2838,6 +2920,7 @@ grpc_cc_library( "config", "gpr_base", "grpc_base", + "iomgr_timer", "slice", ], ) @@ -3044,6 +3127,7 @@ grpc_cc_library( "grpc_sockaddr", "grpc_trace", "grpc_transport_chttp2_client_connector", + "iomgr_timer", "json", "orphanable", "protobuf_duration_upb", @@ -3093,6 +3177,7 @@ grpc_cc_library( "grpc_security_base", "grpc_service_config_impl", "grpc_trace", + "iomgr_timer", "json", "json_util", "orphanable", @@ -3223,6 +3308,7 @@ grpc_cc_library( "grpc_sockaddr", "grpc_tls_credentials", "grpc_transport_chttp2_client_connector", + "iomgr_timer", "json", "json_util", "orphanable", @@ -3466,6 +3552,7 @@ grpc_cc_library( "grpc_codegen", "grpc_resolver_xds_header", "grpc_trace", + "iomgr_timer", "json", "orphanable", "ref_counted", @@ -3624,6 +3711,7 @@ grpc_cc_library( "grpc_codegen", "grpc_lb_address_filtering", "grpc_trace", + "iomgr_timer", "json", "orphanable", "ref_counted", @@ -3654,6 +3742,7 @@ grpc_cc_library( "grpc_codegen", "grpc_lb_address_filtering", "grpc_trace", + "iomgr_timer", "json", "orphanable", "ref_counted", @@ -3826,6 +3915,7 @@ grpc_cc_library( "gpr_base", "grpc_base", "grpc_resolver", + "iomgr_timer", "orphanable", "ref_counted_ptr", "uri_parser", @@ -3871,6 +3961,7 @@ grpc_cc_library( "grpc_resolver", "grpc_resolver_dns_selection", "grpc_trace", + "iomgr_timer", "orphanable", "polling_resolver", "ref_counted_ptr", @@ -3926,6 +4017,7 @@ grpc_cc_library( "grpc_trace", "iomgr_fwd", "iomgr_port", + "iomgr_timer", "json", "orphanable", "polling_resolver", @@ -5057,6 +5149,7 @@ grpc_cc_library( "hpack_encoder_table", "httpcli", "iomgr_fwd", + "iomgr_timer", "memory_quota", "orphanable", "pid_controller", @@ -5158,6 +5251,7 @@ grpc_cc_library( "handshaker", "handshaker_registry", "iomgr_fwd", + "iomgr_timer", "memory_quota", "orphanable", "ref_counted", @@ -5287,6 +5381,7 @@ grpc_cc_library( "grpc_service_config_impl", "grpc_trace", "grpc_transport_inproc", + "iomgr_timer", "ref_counted", "useful", ], @@ -5321,6 +5416,7 @@ grpc_cc_library( "grpc_trace", "grpc_transport_inproc", "grpc_unsecure", + "iomgr_timer", "ref_counted", "useful", ], @@ -5495,6 +5591,7 @@ grpc_cc_library( "grpc++", "grpc++_codegen_base", "grpc_base", + "iomgr_timer", "protobuf_duration_upb", "ref_counted", "time", diff --git a/CMakeLists.txt b/CMakeLists.txt index 95b711256d8..71e4524774b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -110,6 +110,8 @@ set(gRPC_ABSL_USED_TARGETS absl_bits absl_city absl_civil_time + absl_cleanup + absl_cleanup_internal absl_compressed_tuple absl_config absl_container_common @@ -132,6 +134,7 @@ set(gRPC_ABSL_USED_TARGETS absl_fast_type_id absl_fixed_array absl_flat_hash_map + absl_flat_hash_set absl_function_ref absl_graphcycles_internal absl_hash @@ -1044,6 +1047,7 @@ if(gRPC_BUILD_TESTS) add_dependencies(buildtests_cxx insecure_security_connector_test) add_dependencies(buildtests_cxx interop_client) add_dependencies(buildtests_cxx interop_server) + add_dependencies(buildtests_cxx iomgr_event_engine_test) add_dependencies(buildtests_cxx join_test) add_dependencies(buildtests_cxx json_test) add_dependencies(buildtests_cxx large_metadata_bad_client_test) @@ -2024,11 +2028,13 @@ add_library(grpc src/core/lib/event_engine/channel_args_endpoint_config.cc src/core/lib/event_engine/default_event_engine_factory.cc src/core/lib/event_engine/event_engine.cc + src/core/lib/event_engine/iomgr_engine.cc src/core/lib/event_engine/memory_allocator.cc src/core/lib/event_engine/resolved_address.cc src/core/lib/event_engine/slice.cc src/core/lib/event_engine/slice_buffer.cc src/core/lib/event_engine/sockaddr.cc + src/core/lib/event_engine/trace.cc src/core/lib/gprpp/time.cc src/core/lib/http/format_request.cc src/core/lib/http/httpcli.cc @@ -2310,7 +2316,9 @@ target_link_libraries(grpc ${_gRPC_RE2_LIBRARIES} ${_gRPC_UPB_LIBRARIES} ${_gRPC_ALLTARGETS_LIBRARIES} + absl::cleanup absl::flat_hash_map + absl::flat_hash_set absl::inlined_vector absl::bind_front absl::hash @@ -2639,11 +2647,13 @@ add_library(grpc_unsecure src/core/lib/event_engine/channel_args_endpoint_config.cc src/core/lib/event_engine/default_event_engine_factory.cc src/core/lib/event_engine/event_engine.cc + src/core/lib/event_engine/iomgr_engine.cc src/core/lib/event_engine/memory_allocator.cc src/core/lib/event_engine/resolved_address.cc src/core/lib/event_engine/slice.cc src/core/lib/event_engine/slice_buffer.cc src/core/lib/event_engine/sockaddr.cc + src/core/lib/event_engine/trace.cc src/core/lib/gprpp/time.cc src/core/lib/http/format_request.cc src/core/lib/http/httpcli.cc @@ -2861,7 +2871,9 @@ target_link_libraries(grpc_unsecure ${_gRPC_RE2_LIBRARIES} ${_gRPC_UPB_LIBRARIES} ${_gRPC_ALLTARGETS_LIBRARIES} + absl::cleanup absl::flat_hash_map + absl::flat_hash_set absl::inlined_vector absl::bind_front absl::hash @@ -8445,7 +8457,6 @@ target_include_directories(cel_authorization_engine_test target_link_libraries(cel_authorization_engine_test ${_gRPC_PROTOBUF_LIBRARIES} ${_gRPC_ALLTARGETS_LIBRARIES} - absl::flat_hash_set grpc_test_util ) @@ -12483,6 +12494,43 @@ target_link_libraries(interop_server ) +endif() +if(gRPC_BUILD_TESTS) + +add_executable(iomgr_event_engine_test + test/core/event_engine/test_suite/event_engine_test.cc + test/core/event_engine/test_suite/iomgr_event_engine_test.cc + test/core/event_engine/test_suite/timer_test.cc + third_party/googletest/googletest/src/gtest-all.cc + third_party/googletest/googlemock/src/gmock-all.cc +) + +target_include_directories(iomgr_event_engine_test + PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ${CMAKE_CURRENT_SOURCE_DIR}/include + ${_gRPC_ADDRESS_SORTING_INCLUDE_DIR} + ${_gRPC_RE2_INCLUDE_DIR} + ${_gRPC_SSL_INCLUDE_DIR} + ${_gRPC_UPB_GENERATED_DIR} + ${_gRPC_UPB_GRPC_GENERATED_DIR} + ${_gRPC_UPB_INCLUDE_DIR} + ${_gRPC_XXHASH_INCLUDE_DIR} + ${_gRPC_ZLIB_INCLUDE_DIR} + third_party/googletest/googletest/include + third_party/googletest/googletest + third_party/googletest/googlemock/include + third_party/googletest/googlemock + ${_gRPC_PROTO_GENS_DIR} +) + +target_link_libraries(iomgr_event_engine_test + ${_gRPC_PROTOBUF_LIBRARIES} + ${_gRPC_ALLTARGETS_LIBRARIES} + grpc_test_util +) + + endif() if(gRPC_BUILD_TESTS) @@ -19231,7 +19279,7 @@ generate_pkgconfig( "gRPC" "high performance general RPC framework" "${gRPC_CORE_VERSION}" - "gpr openssl absl_base absl_bind_front absl_cord absl_core_headers absl_flat_hash_map absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" + "gpr openssl absl_base absl_bind_front absl_cleanup absl_cord absl_core_headers absl_flat_hash_map absl_flat_hash_set absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" "-lgrpc -laddress_sorting -lre2 -lupb -lcares -lz" "" "grpc.pc") @@ -19241,7 +19289,7 @@ generate_pkgconfig( "gRPC unsecure" "high performance general RPC framework without SSL" "${gRPC_CORE_VERSION}" - "gpr absl_base absl_bind_front absl_cord absl_core_headers absl_flat_hash_map absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" + "gpr absl_base absl_bind_front absl_cleanup absl_cord absl_core_headers absl_flat_hash_map absl_flat_hash_set absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" "-lgrpc_unsecure" "" "grpc_unsecure.pc") @@ -19251,7 +19299,7 @@ generate_pkgconfig( "gRPC++" "C++ wrapper for gRPC" "${gRPC_CPP_VERSION}" - "grpc absl_base absl_bind_front absl_cord absl_core_headers absl_flat_hash_map absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" + "grpc absl_base absl_bind_front absl_cleanup absl_cord absl_core_headers absl_flat_hash_map absl_flat_hash_set absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" "-lgrpc++" "" "grpc++.pc") @@ -19261,7 +19309,7 @@ generate_pkgconfig( "gRPC++ unsecure" "C++ wrapper for gRPC without SSL" "${gRPC_CPP_VERSION}" - "grpc_unsecure absl_base absl_bind_front absl_cord absl_core_headers absl_flat_hash_map absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" + "grpc_unsecure absl_base absl_bind_front absl_cleanup absl_cord absl_core_headers absl_flat_hash_map absl_flat_hash_set absl_hash absl_inlined_vector absl_memory absl_optional absl_random_random absl_span absl_status absl_statusor absl_str_format absl_strings absl_synchronization absl_time absl_type_traits absl_utility absl_variant" "-lgrpc++_unsecure" "" "grpc++_unsecure.pc") diff --git a/Makefile b/Makefile index bae84705a77..f1af688661b 100644 --- a/Makefile +++ b/Makefile @@ -1441,11 +1441,13 @@ LIBGRPC_SRC = \ src/core/lib/event_engine/channel_args_endpoint_config.cc \ src/core/lib/event_engine/default_event_engine_factory.cc \ src/core/lib/event_engine/event_engine.cc \ + src/core/lib/event_engine/iomgr_engine.cc \ src/core/lib/event_engine/memory_allocator.cc \ src/core/lib/event_engine/resolved_address.cc \ src/core/lib/event_engine/slice.cc \ src/core/lib/event_engine/slice_buffer.cc \ src/core/lib/event_engine/sockaddr.cc \ + src/core/lib/event_engine/trace.cc \ src/core/lib/gprpp/time.cc \ src/core/lib/http/format_request.cc \ src/core/lib/http/httpcli.cc \ @@ -1899,11 +1901,13 @@ LIBGRPC_UNSECURE_SRC = \ src/core/lib/event_engine/channel_args_endpoint_config.cc \ src/core/lib/event_engine/default_event_engine_factory.cc \ src/core/lib/event_engine/event_engine.cc \ + src/core/lib/event_engine/iomgr_engine.cc \ src/core/lib/event_engine/memory_allocator.cc \ src/core/lib/event_engine/resolved_address.cc \ src/core/lib/event_engine/slice.cc \ src/core/lib/event_engine/slice_buffer.cc \ src/core/lib/event_engine/sockaddr.cc \ + src/core/lib/event_engine/trace.cc \ src/core/lib/gprpp/time.cc \ src/core/lib/http/format_request.cc \ src/core/lib/http/httpcli.cc \ diff --git a/build_autogenerated.yaml b/build_autogenerated.yaml index 50ccbae91df..b2877ffd13c 100644 --- a/build_autogenerated.yaml +++ b/build_autogenerated.yaml @@ -726,7 +726,10 @@ libs: - src/core/lib/debug/trace.h - src/core/lib/event_engine/channel_args_endpoint_config.h - src/core/lib/event_engine/event_engine_factory.h + - src/core/lib/event_engine/handle_containers.h + - src/core/lib/event_engine/iomgr_engine.h - src/core/lib/event_engine/sockaddr.h + - src/core/lib/event_engine/trace.h - src/core/lib/gprpp/atomic_utils.h - src/core/lib/gprpp/bitset.h - src/core/lib/gprpp/capture.h @@ -1402,11 +1405,13 @@ libs: - src/core/lib/event_engine/channel_args_endpoint_config.cc - src/core/lib/event_engine/default_event_engine_factory.cc - src/core/lib/event_engine/event_engine.cc + - src/core/lib/event_engine/iomgr_engine.cc - src/core/lib/event_engine/memory_allocator.cc - src/core/lib/event_engine/resolved_address.cc - src/core/lib/event_engine/slice.cc - src/core/lib/event_engine/slice_buffer.cc - src/core/lib/event_engine/sockaddr.cc + - src/core/lib/event_engine/trace.cc - src/core/lib/gprpp/time.cc - src/core/lib/http/format_request.cc - src/core/lib/http/httpcli.cc @@ -1650,7 +1655,9 @@ libs: - src/core/tsi/transport_security.cc - src/core/tsi/transport_security_grpc.cc deps: + - absl/cleanup:cleanup - absl/container:flat_hash_map + - absl/container:flat_hash_set - absl/container:inlined_vector - absl/functional:bind_front - absl/hash:hash @@ -1919,7 +1926,10 @@ libs: - src/core/lib/debug/trace.h - src/core/lib/event_engine/channel_args_endpoint_config.h - src/core/lib/event_engine/event_engine_factory.h + - src/core/lib/event_engine/handle_containers.h + - src/core/lib/event_engine/iomgr_engine.h - src/core/lib/event_engine/sockaddr.h + - src/core/lib/event_engine/trace.h - src/core/lib/gprpp/atomic_utils.h - src/core/lib/gprpp/bitset.h - src/core/lib/gprpp/capture.h @@ -2248,11 +2258,13 @@ libs: - src/core/lib/event_engine/channel_args_endpoint_config.cc - src/core/lib/event_engine/default_event_engine_factory.cc - src/core/lib/event_engine/event_engine.cc + - src/core/lib/event_engine/iomgr_engine.cc - src/core/lib/event_engine/memory_allocator.cc - src/core/lib/event_engine/resolved_address.cc - src/core/lib/event_engine/slice.cc - src/core/lib/event_engine/slice_buffer.cc - src/core/lib/event_engine/sockaddr.cc + - src/core/lib/event_engine/trace.cc - src/core/lib/gprpp/time.cc - src/core/lib/http/format_request.cc - src/core/lib/http/httpcli.cc @@ -2432,7 +2444,9 @@ libs: - src/core/tsi/transport_security.cc - src/core/tsi/transport_security_grpc.cc deps: + - absl/cleanup:cleanup - absl/container:flat_hash_map + - absl/container:flat_hash_set - absl/container:inlined_vector - absl/functional:bind_front - absl/hash:hash @@ -4846,7 +4860,6 @@ targets: - src/core/lib/security/authorization/cel_authorization_engine.cc - test/core/security/cel_authorization_engine_test.cc deps: - - absl/container:flat_hash_set - grpc_test_util - name: certificate_provider_registry_test gtest: true @@ -6372,6 +6385,19 @@ targets: deps: - grpc++_test_config - grpc++_test_util +- name: iomgr_event_engine_test + gtest: true + build: test + language: c++ + headers: + - test/core/event_engine/test_suite/event_engine_test.h + src: + - test/core/event_engine/test_suite/event_engine_test.cc + - test/core/event_engine/test_suite/iomgr_event_engine_test.cc + - test/core/event_engine/test_suite/timer_test.cc + deps: + - grpc_test_util + uses_polling: false - name: join_test gtest: true build: test diff --git a/config.m4 b/config.m4 index df036077eb8..82477ddd57c 100644 --- a/config.m4 +++ b/config.m4 @@ -463,11 +463,13 @@ if test "$PHP_GRPC" != "no"; then src/core/lib/event_engine/channel_args_endpoint_config.cc \ src/core/lib/event_engine/default_event_engine_factory.cc \ src/core/lib/event_engine/event_engine.cc \ + src/core/lib/event_engine/iomgr_engine.cc \ src/core/lib/event_engine/memory_allocator.cc \ src/core/lib/event_engine/resolved_address.cc \ src/core/lib/event_engine/slice.cc \ src/core/lib/event_engine/slice_buffer.cc \ src/core/lib/event_engine/sockaddr.cc \ + src/core/lib/event_engine/trace.cc \ src/core/lib/gpr/alloc.cc \ src/core/lib/gpr/atm.cc \ src/core/lib/gpr/cpu_iphone.cc \ diff --git a/config.w32 b/config.w32 index 5a7e0474935..2d26f0f7f45 100644 --- a/config.w32 +++ b/config.w32 @@ -429,11 +429,13 @@ if (PHP_GRPC != "no") { "src\\core\\lib\\event_engine\\channel_args_endpoint_config.cc " + "src\\core\\lib\\event_engine\\default_event_engine_factory.cc " + "src\\core\\lib\\event_engine\\event_engine.cc " + + "src\\core\\lib\\event_engine\\iomgr_engine.cc " + "src\\core\\lib\\event_engine\\memory_allocator.cc " + "src\\core\\lib\\event_engine\\resolved_address.cc " + "src\\core\\lib\\event_engine\\slice.cc " + "src\\core\\lib\\event_engine\\slice_buffer.cc " + "src\\core\\lib\\event_engine\\sockaddr.cc " + + "src\\core\\lib\\event_engine\\trace.cc " + "src\\core\\lib\\gpr\\alloc.cc " + "src\\core\\lib\\gpr\\atm.cc " + "src\\core\\lib\\gpr\\cpu_iphone.cc " + diff --git a/gRPC-C++.podspec b/gRPC-C++.podspec index 435754224db..5fec0547f6e 100644 --- a/gRPC-C++.podspec +++ b/gRPC-C++.podspec @@ -197,7 +197,9 @@ Pod::Spec.new do |s| abseil_version = '1.20211102.0' ss.dependency 'abseil/base/base', abseil_version ss.dependency 'abseil/base/core_headers', abseil_version + ss.dependency 'abseil/cleanup/cleanup', abseil_version ss.dependency 'abseil/container/flat_hash_map', abseil_version + ss.dependency 'abseil/container/flat_hash_set', abseil_version ss.dependency 'abseil/container/inlined_vector', abseil_version ss.dependency 'abseil/functional/bind_front', abseil_version ss.dependency 'abseil/hash/hash', abseil_version @@ -665,7 +667,10 @@ Pod::Spec.new do |s| 'src/core/lib/debug/trace.h', 'src/core/lib/event_engine/channel_args_endpoint_config.h', 'src/core/lib/event_engine/event_engine_factory.h', + 'src/core/lib/event_engine/handle_containers.h', + 'src/core/lib/event_engine/iomgr_engine.h', 'src/core/lib/event_engine/sockaddr.h', + 'src/core/lib/event_engine/trace.h', 'src/core/lib/gpr/alloc.h', 'src/core/lib/gpr/env.h', 'src/core/lib/gpr/murmur_hash.h', @@ -1482,7 +1487,10 @@ Pod::Spec.new do |s| 'src/core/lib/debug/trace.h', 'src/core/lib/event_engine/channel_args_endpoint_config.h', 'src/core/lib/event_engine/event_engine_factory.h', + 'src/core/lib/event_engine/handle_containers.h', + 'src/core/lib/event_engine/iomgr_engine.h', 'src/core/lib/event_engine/sockaddr.h', + 'src/core/lib/event_engine/trace.h', 'src/core/lib/gpr/alloc.h', 'src/core/lib/gpr/env.h', 'src/core/lib/gpr/murmur_hash.h', diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index a2dd24358ad..c5bad19e34c 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -175,7 +175,9 @@ Pod::Spec.new do |s| ss.dependency 'Libuv-gRPC', '0.0.10' ss.dependency 'abseil/base/base', abseil_version ss.dependency 'abseil/base/core_headers', abseil_version + ss.dependency 'abseil/cleanup/cleanup', abseil_version ss.dependency 'abseil/container/flat_hash_map', abseil_version + ss.dependency 'abseil/container/flat_hash_set', abseil_version ss.dependency 'abseil/container/inlined_vector', abseil_version ss.dependency 'abseil/functional/bind_front', abseil_version ss.dependency 'abseil/hash/hash', abseil_version @@ -1029,12 +1031,17 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/default_event_engine_factory.cc', 'src/core/lib/event_engine/event_engine.cc', 'src/core/lib/event_engine/event_engine_factory.h', + 'src/core/lib/event_engine/handle_containers.h', + 'src/core/lib/event_engine/iomgr_engine.cc', + 'src/core/lib/event_engine/iomgr_engine.h', 'src/core/lib/event_engine/memory_allocator.cc', 'src/core/lib/event_engine/resolved_address.cc', 'src/core/lib/event_engine/slice.cc', 'src/core/lib/event_engine/slice_buffer.cc', 'src/core/lib/event_engine/sockaddr.cc', 'src/core/lib/event_engine/sockaddr.h', + 'src/core/lib/event_engine/trace.cc', + 'src/core/lib/event_engine/trace.h', 'src/core/lib/gpr/alloc.cc', 'src/core/lib/gpr/alloc.h', 'src/core/lib/gpr/atm.cc', @@ -2092,7 +2099,10 @@ Pod::Spec.new do |s| 'src/core/lib/debug/trace.h', 'src/core/lib/event_engine/channel_args_endpoint_config.h', 'src/core/lib/event_engine/event_engine_factory.h', + 'src/core/lib/event_engine/handle_containers.h', + 'src/core/lib/event_engine/iomgr_engine.h', 'src/core/lib/event_engine/sockaddr.h', + 'src/core/lib/event_engine/trace.h', 'src/core/lib/gpr/alloc.h', 'src/core/lib/gpr/env.h', 'src/core/lib/gpr/murmur_hash.h', diff --git a/grpc.gemspec b/grpc.gemspec index 565d7cb0390..f023e972495 100644 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -944,12 +944,17 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/event_engine/default_event_engine_factory.cc ) s.files += %w( src/core/lib/event_engine/event_engine.cc ) s.files += %w( src/core/lib/event_engine/event_engine_factory.h ) + s.files += %w( src/core/lib/event_engine/handle_containers.h ) + s.files += %w( src/core/lib/event_engine/iomgr_engine.cc ) + s.files += %w( src/core/lib/event_engine/iomgr_engine.h ) s.files += %w( src/core/lib/event_engine/memory_allocator.cc ) s.files += %w( src/core/lib/event_engine/resolved_address.cc ) s.files += %w( src/core/lib/event_engine/slice.cc ) s.files += %w( src/core/lib/event_engine/slice_buffer.cc ) s.files += %w( src/core/lib/event_engine/sockaddr.cc ) s.files += %w( src/core/lib/event_engine/sockaddr.h ) + s.files += %w( src/core/lib/event_engine/trace.cc ) + s.files += %w( src/core/lib/event_engine/trace.h ) s.files += %w( src/core/lib/gpr/alloc.cc ) s.files += %w( src/core/lib/gpr/alloc.h ) s.files += %w( src/core/lib/gpr/atm.cc ) @@ -1568,8 +1573,11 @@ Gem::Specification.new do |s| s.files += %w( third_party/abseil-cpp/absl/base/policy_checks.h ) s.files += %w( third_party/abseil-cpp/absl/base/port.h ) s.files += %w( third_party/abseil-cpp/absl/base/thread_annotations.h ) + s.files += %w( third_party/abseil-cpp/absl/cleanup/cleanup.h ) + s.files += %w( third_party/abseil-cpp/absl/cleanup/internal/cleanup.h ) s.files += %w( third_party/abseil-cpp/absl/container/fixed_array.h ) s.files += %w( third_party/abseil-cpp/absl/container/flat_hash_map.h ) + s.files += %w( third_party/abseil-cpp/absl/container/flat_hash_set.h ) s.files += %w( third_party/abseil-cpp/absl/container/inlined_vector.h ) s.files += %w( third_party/abseil-cpp/absl/container/internal/common.h ) s.files += %w( third_party/abseil-cpp/absl/container/internal/compressed_tuple.h ) diff --git a/grpc.gyp b/grpc.gyp index f0c0f2527b7..cc049e7a91b 100644 --- a/grpc.gyp +++ b/grpc.gyp @@ -360,7 +360,9 @@ 'target_name': 'grpc', 'type': 'static_library', 'dependencies': [ + 'absl/cleanup:cleanup', 'absl/container:flat_hash_map', + 'absl/container:flat_hash_set', 'absl/container:inlined_vector', 'absl/functional:bind_front', 'absl/hash:hash', @@ -796,11 +798,13 @@ 'src/core/lib/event_engine/channel_args_endpoint_config.cc', 'src/core/lib/event_engine/default_event_engine_factory.cc', 'src/core/lib/event_engine/event_engine.cc', + 'src/core/lib/event_engine/iomgr_engine.cc', 'src/core/lib/event_engine/memory_allocator.cc', 'src/core/lib/event_engine/resolved_address.cc', 'src/core/lib/event_engine/slice.cc', 'src/core/lib/event_engine/slice_buffer.cc', 'src/core/lib/event_engine/sockaddr.cc', + 'src/core/lib/event_engine/trace.cc', 'src/core/lib/gprpp/time.cc', 'src/core/lib/http/format_request.cc', 'src/core/lib/http/httpcli.cc', @@ -1116,7 +1120,9 @@ 'target_name': 'grpc_unsecure', 'type': 'static_library', 'dependencies': [ + 'absl/cleanup:cleanup', 'absl/container:flat_hash_map', + 'absl/container:flat_hash_set', 'absl/container:inlined_vector', 'absl/functional:bind_front', 'absl/hash:hash', @@ -1261,11 +1267,13 @@ 'src/core/lib/event_engine/channel_args_endpoint_config.cc', 'src/core/lib/event_engine/default_event_engine_factory.cc', 'src/core/lib/event_engine/event_engine.cc', + 'src/core/lib/event_engine/iomgr_engine.cc', 'src/core/lib/event_engine/memory_allocator.cc', 'src/core/lib/event_engine/resolved_address.cc', 'src/core/lib/event_engine/slice.cc', 'src/core/lib/event_engine/slice_buffer.cc', 'src/core/lib/event_engine/sockaddr.cc', + 'src/core/lib/event_engine/trace.cc', 'src/core/lib/gprpp/time.cc', 'src/core/lib/http/format_request.cc', 'src/core/lib/http/httpcli.cc', diff --git a/include/grpc/event_engine/event_engine.h b/include/grpc/event_engine/event_engine.h index 8a0d882c465..0454ce921c1 100644 --- a/include/grpc/event_engine/event_engine.h +++ b/include/grpc/event_engine/event_engine.h @@ -28,7 +28,7 @@ #include #include -// TODO(hork): Define the Endpoint::Write metrics collection system +// TODO(vigneshbabu): Define the Endpoint::Write metrics collection system namespace grpc_event_engine { namespace experimental { diff --git a/package.xml b/package.xml index ac3419a56de..ce7b0d76c17 100644 --- a/package.xml +++ b/package.xml @@ -926,12 +926,17 @@ + + + + + @@ -1572,8 +1577,11 @@ + + + diff --git a/src/core/lib/event_engine/default_event_engine_factory.cc b/src/core/lib/event_engine/default_event_engine_factory.cc index 7d1b4ea69c1..214651781b4 100644 --- a/src/core/lib/event_engine/default_event_engine_factory.cc +++ b/src/core/lib/event_engine/default_event_engine_factory.cc @@ -15,16 +15,18 @@ #include +#include "absl/memory/memory.h" + #include #include "src/core/lib/event_engine/event_engine_factory.h" +#include "src/core/lib/event_engine/iomgr_engine.h" namespace grpc_event_engine { namespace experimental { std::unique_ptr DefaultEventEngineFactory() { - // TODO(hork): call LibuvEventEngineFactory - return nullptr; + return absl::make_unique(); } } // namespace experimental diff --git a/src/core/lib/event_engine/event_engine.cc b/src/core/lib/event_engine/event_engine.cc index 79490acb96b..a84cb90d2a6 100644 --- a/src/core/lib/event_engine/event_engine.cc +++ b/src/core/lib/event_engine/event_engine.cc @@ -18,7 +18,9 @@ #include +#include "src/core/lib/debug/trace.h" #include "src/core/lib/event_engine/event_engine_factory.h" +#include "src/core/lib/event_engine/trace.h" #include "src/core/lib/gprpp/sync.h" namespace grpc_event_engine { @@ -49,5 +51,12 @@ EventEngine* GetDefaultEventEngine() { return default_event_engine; } +void InitializeEventEngine() { + GetDefaultEventEngine()->Run([]() { + GRPC_EVENT_ENGINE_TRACE("EventEngine:%p initialized", + GetDefaultEventEngine()); + }); +} + } // namespace experimental } // namespace grpc_event_engine diff --git a/src/core/lib/event_engine/event_engine_factory.h b/src/core/lib/event_engine/event_engine_factory.h index d73b3b2785b..b99dca6badc 100644 --- a/src/core/lib/event_engine/event_engine_factory.h +++ b/src/core/lib/event_engine/event_engine_factory.h @@ -32,6 +32,9 @@ EventEngine* GetDefaultEventEngine(); /// Create an EventEngine using the default factory provided at link time. std::unique_ptr DefaultEventEngineFactory(); +// TODO(hork): remove this when any other EE usage is landed +void InitializeEventEngine(); + } // namespace experimental } // namespace grpc_event_engine diff --git a/src/core/lib/event_engine/handle_containers.h b/src/core/lib/event_engine/handle_containers.h new file mode 100644 index 00000000000..a1739ef3c08 --- /dev/null +++ b/src/core/lib/event_engine/handle_containers.h @@ -0,0 +1,67 @@ +// Copyright 2022 The gRPC Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#ifndef GRPC_CORE_LIB_EVENT_ENGINE_HANDLE_CONTAINERS_H +#define GRPC_CORE_LIB_EVENT_ENGINE_HANDLE_CONTAINERS_H + +#include + +#include + +#include +#include + +#include "absl/container/flat_hash_set.h" +#include "absl/hash/hash.h" + +#include + +namespace grpc_event_engine { +namespace experimental { + +// Used for heterogeneous lookup of TaskHandles in abseil containers. +template +struct TaskHandleComparator { + struct Hash { + using HashType = std::pair; + using is_transparent = void; + size_t operator()(const TaskHandle& handle) const { + return absl::Hash()({handle.keys[0], handle.keys[1]}); + } + }; + struct Eq { + using is_transparent = void; + bool operator()(const TaskHandle& lhs, const TaskHandle& rhs) const { + return lhs.keys[0] == rhs.keys[0] && lhs.keys[1] == rhs.keys[1]; + } + }; +}; + +using TaskHandleSet = absl::flat_hash_set< + grpc_event_engine::experimental::EventEngine::TaskHandle, + TaskHandleComparator< + grpc_event_engine::experimental::EventEngine::TaskHandle>::Hash, + TaskHandleComparator< + grpc_event_engine::experimental::EventEngine::TaskHandle>::Eq>; + +using LookupTaskHandleSet = absl::flat_hash_set< + grpc_event_engine::experimental::EventEngine::DNSResolver::LookupTaskHandle, + TaskHandleComparator::Hash, + TaskHandleComparator::Eq>; + +} // namespace experimental +} // namespace grpc_event_engine + +#endif // GRPC_CORE_LIB_EVENT_ENGINE_HANDLE_CONTAINERS_H diff --git a/src/core/lib/event_engine/iomgr_engine.cc b/src/core/lib/event_engine/iomgr_engine.cc new file mode 100644 index 00000000000..f532f611965 --- /dev/null +++ b/src/core/lib/event_engine/iomgr_engine.cc @@ -0,0 +1,206 @@ +// Copyright 2022 The gRPC Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#include + +#include "src/core/lib/event_engine/iomgr_engine.h" + +#include +#include +#include + +#include "absl/cleanup/cleanup.h" +#include "absl/container/flat_hash_set.h" +#include "absl/strings/str_cat.h" +#include "absl/time/clock.h" + +#include +#include + +#include "src/core/lib/debug/trace.h" +#include "src/core/lib/event_engine/trace.h" +#include "src/core/lib/gprpp/match.h" +#include "src/core/lib/gprpp/time.h" +#include "src/core/lib/iomgr/closure.h" +#include "src/core/lib/iomgr/error.h" +#include "src/core/lib/iomgr/exec_ctx.h" +#include "src/core/lib/iomgr/executor.h" +#include "src/core/lib/iomgr/timer.h" + +namespace grpc_event_engine { +namespace experimental { + +namespace { + +struct ClosureData { + grpc_timer timer; + grpc_closure closure; + absl::variant, EventEngine::Closure*> cb; + IomgrEventEngine* engine; + EventEngine::TaskHandle handle; +}; + +// Timer limits due to quirks in the iomgr implementation. +// If deadline <= Now, the callback will be run inline, which can result in lock +// issues. And absl::InfiniteFuture yields UB. +absl::Time Clamp(absl::Time when) { + absl::Time max = absl::Now() + absl::Hours(8766); + absl::Time min = absl::Now() + absl::Milliseconds(2); + if (when > max) return max; + if (when < min) return min; + return when; +} + +std::string HandleToString(EventEngine::TaskHandle handle) { + return absl::StrCat("{", handle.keys[0], ",", handle.keys[1], "}"); +} + +} // namespace + +IomgrEventEngine::IomgrEventEngine() {} + +IomgrEventEngine::~IomgrEventEngine() { + grpc_core::MutexLock lock(&mu_); + if (GRPC_TRACE_FLAG_ENABLED(grpc_event_engine_trace)) { + for (auto handle : known_handles_) { + gpr_log(GPR_ERROR, + "(event_engine) IomgrEventEngine:%p uncleared TaskHandle at " + "shutdown:%s", + this, HandleToString(handle).c_str()); + } + } + GPR_ASSERT(GPR_LIKELY(known_handles_.empty())); +} + +bool IomgrEventEngine::Cancel(EventEngine::TaskHandle handle) { + grpc_core::ExecCtx ctx; + grpc_core::MutexLock lock(&mu_); + if (!known_handles_.contains(handle)) return false; + auto* cd = reinterpret_cast(handle.keys[0]); + grpc_timer_cancel(&cd->timer); + known_handles_.erase(handle); + return true; +} + +EventEngine::TaskHandle IomgrEventEngine::RunAt(absl::Time when, + std::function closure) { + return RunAtInternal(when, std::move(closure)); +} + +EventEngine::TaskHandle IomgrEventEngine::RunAt(absl::Time when, + EventEngine::Closure* closure) { + return RunAtInternal(when, closure); +} + +void IomgrEventEngine::Run(std::function closure) { + RunInternal(closure); +} + +void IomgrEventEngine::Run(EventEngine::Closure* closure) { + RunInternal(closure); +} + +EventEngine::TaskHandle IomgrEventEngine::RunAtInternal( + absl::Time when, + absl::variant, EventEngine::Closure*> cb) { + when = Clamp(when); + grpc_core::ExecCtx ctx; + auto* cd = new ClosureData; + cd->cb = std::move(cb); + cd->engine = this; + GRPC_CLOSURE_INIT( + &cd->closure, + [](void* arg, grpc_error_handle error) { + auto* cd = static_cast(arg); + GRPC_EVENT_ENGINE_TRACE("IomgrEventEngine:%p executing callback:%s", + cd->engine, HandleToString(cd->handle).c_str()); + { + grpc_core::MutexLock lock(&cd->engine->mu_); + cd->engine->known_handles_.erase(cd->handle); + } + auto cleaner = absl::MakeCleanup([cd] { delete cd; }); + if (error == GRPC_ERROR_CANCELLED) return; + grpc_core::Match( + cd->cb, [](EventEngine::Closure* cb) { cb->Run(); }, + [](std::function fn) { fn(); }); + }, + cd, nullptr); + // kludge to deal with realtime/monotonic clock conversion + absl::Time absl_now = absl::Now(); + grpc_core::Duration duration = grpc_core::Duration::Milliseconds( + absl::ToInt64Milliseconds(when - absl_now) + 1); + grpc_core::ExecCtx::Get()->InvalidateNow(); + grpc_core::Timestamp when_internal = grpc_core::ExecCtx::Get()->Now() + + duration + + grpc_core::Duration::Milliseconds(1); + EventEngine::TaskHandle handle{reinterpret_cast(cd), + aba_token_.fetch_add(1)}; + grpc_core::MutexLock lock(&mu_); + known_handles_.insert(handle); + cd->handle = handle; + GRPC_EVENT_ENGINE_TRACE("IomgrEventEngine:%p scheduling callback:%s", this, + HandleToString(handle).c_str()); + grpc_timer_init(&cd->timer, when_internal, &cd->closure); + return handle; +} + +void IomgrEventEngine::RunInternal( + absl::variant, EventEngine::Closure*> cb) { + auto* cd = new ClosureData; + cd->cb = std::move(cb); + cd->engine = this; + GRPC_CLOSURE_INIT( + &cd->closure, + [](void* arg, grpc_error_handle /*error*/) { + auto* cd = static_cast(arg); + auto cleaner = absl::MakeCleanup([cd] { delete cd; }); + grpc_core::Match( + cd->cb, [](EventEngine::Closure* cb) { cb->Run(); }, + [](std::function fn) { fn(); }); + }, + cd, nullptr); + // TODO(hork): have the EE spawn dedicated closure thread(s) + grpc_core::Executor::Run(&cd->closure, GRPC_ERROR_NONE); +} + +std::unique_ptr IomgrEventEngine::GetDNSResolver( + EventEngine::DNSResolver::ResolverOptions const& /*options*/) { + GPR_ASSERT(false && "unimplemented"); +} + +bool IomgrEventEngine::IsWorkerThread() { + GPR_ASSERT(false && "unimplemented"); +} + +bool IomgrEventEngine::CancelConnect(EventEngine::ConnectionHandle /*handle*/) { + GPR_ASSERT(false && "unimplemented"); +} + +EventEngine::ConnectionHandle IomgrEventEngine::Connect( + OnConnectCallback /*on_connect*/, const ResolvedAddress& /*addr*/, + const EndpointConfig& /*args*/, MemoryAllocator /*memory_allocator*/, + absl::Time /*deadline*/) { + GPR_ASSERT(false && "unimplemented"); +} + +absl::StatusOr> +IomgrEventEngine::CreateListener( + Listener::AcceptCallback /*on_accept*/, + std::function /*on_shutdown*/, + const EndpointConfig& /*config*/, + std::unique_ptr /*memory_allocator_factory*/) { + GPR_ASSERT(false && "unimplemented"); +} + +} // namespace experimental +} // namespace grpc_event_engine diff --git a/src/core/lib/event_engine/iomgr_engine.h b/src/core/lib/event_engine/iomgr_engine.h new file mode 100644 index 00000000000..ea618e3c96a --- /dev/null +++ b/src/core/lib/event_engine/iomgr_engine.h @@ -0,0 +1,118 @@ +// Copyright 2022 The gRPC Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#ifndef GRPC_CORE_LIB_EVENT_ENGINE_IOMGR_ENGINE_H +#define GRPC_CORE_LIB_EVENT_ENGINE_IOMGR_ENGINE_H +#include + +#include + +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/string_view.h" +#include "absl/time/time.h" +#include "absl/types/variant.h" + +#include +#include +#include +#include + +#include "src/core/lib/event_engine/handle_containers.h" +#include "src/core/lib/gprpp/sync.h" + +namespace grpc_event_engine { +namespace experimental { + +class IomgrEventEngine final : public EventEngine { + public: + class IomgrEndpoint : public EventEngine::Endpoint { + public: + ~IomgrEndpoint() override; + void Read(std::function on_read, SliceBuffer* buffer, + const ReadArgs* args) override; + void Write(std::function on_writable, SliceBuffer* data, + const WriteArgs* args) override; + const ResolvedAddress& GetPeerAddress() const override; + const ResolvedAddress& GetLocalAddress() const override; + }; + class IomgrListener : public EventEngine::Listener { + public: + ~IomgrListener() override; + absl::StatusOr Bind(const ResolvedAddress& addr) override; + absl::Status Start() override; + }; + class IomgrDNSResolver : public EventEngine::DNSResolver { + public: + ~IomgrDNSResolver() override; + LookupTaskHandle LookupHostname(LookupHostnameCallback on_resolve, + absl::string_view name, + absl::string_view default_port, + absl::Time deadline) override; + LookupTaskHandle LookupSRV(LookupSRVCallback on_resolve, + absl::string_view name, + absl::Time deadline) override; + LookupTaskHandle LookupTXT(LookupTXTCallback on_resolve, + absl::string_view name, + absl::Time deadline) override; + bool CancelLookup(LookupTaskHandle handle) override; + }; + + IomgrEventEngine(); + ~IomgrEventEngine() override; + + absl::StatusOr> CreateListener( + Listener::AcceptCallback on_accept, + std::function on_shutdown, + const EndpointConfig& config, + std::unique_ptr memory_allocator_factory) + override; + + ConnectionHandle Connect(OnConnectCallback on_connect, + const ResolvedAddress& addr, + const EndpointConfig& args, + MemoryAllocator memory_allocator, + absl::Time deadline) override; + + bool CancelConnect(ConnectionHandle handle) override; + bool IsWorkerThread() override; + std::unique_ptr GetDNSResolver( + const DNSResolver::ResolverOptions& options) override; + void Run(Closure* closure) override; + void Run(std::function closure) override; + TaskHandle RunAt(absl::Time when, Closure* closure) override; + TaskHandle RunAt(absl::Time when, std::function closure) override; + bool Cancel(TaskHandle handle) override; + + private: + EventEngine::TaskHandle RunAtInternal( + absl::Time when, + absl::variant, EventEngine::Closure*> cb); + + void RunInternal( + absl::variant, EventEngine::Closure*> cb); + + grpc_core::Mutex mu_; + TaskHandleSet known_handles_ ABSL_GUARDED_BY(mu_); + std::atomic aba_token_{0}; +}; + +} // namespace experimental +} // namespace grpc_event_engine + +#endif // GRPC_CORE_LIB_EVENT_ENGINE_IOMGR_ENGINE_H diff --git a/src/core/lib/event_engine/trace.cc b/src/core/lib/event_engine/trace.cc new file mode 100644 index 00000000000..52cbd35f4f5 --- /dev/null +++ b/src/core/lib/event_engine/trace.cc @@ -0,0 +1,18 @@ +// Copyright 2022 The gRPC Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#include + +#include "src/core/lib/debug/trace.h" + +grpc_core::TraceFlag grpc_event_engine_trace(false, "event_engine"); diff --git a/src/core/lib/event_engine/trace.h b/src/core/lib/event_engine/trace.h new file mode 100644 index 00000000000..d26dee3628f --- /dev/null +++ b/src/core/lib/event_engine/trace.h @@ -0,0 +1,30 @@ +// Copyright 2022 The gRPC Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#ifndef GRPC_CORE_LIB_EVENT_ENGINE_TRACE_H +#define GRPC_CORE_LIB_EVENT_ENGINE_TRACE_H + +#include + +#include + +#include "src/core/lib/debug/trace.h" + +extern grpc_core::TraceFlag grpc_event_engine_trace; + +#define GRPC_EVENT_ENGINE_TRACE(format, ...) \ + if (GRPC_TRACE_FLAG_ENABLED(grpc_event_engine_trace)) { \ + gpr_log(GPR_DEBUG, "(event_engine) " format, __VA_ARGS__); \ + } + +#endif // GRPC_CORE_LIB_EVENT_ENGINE_TRACE_H diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index e867f8f3576..dacad2006a9 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -438,11 +438,13 @@ CORE_SOURCE_FILES = [ 'src/core/lib/event_engine/channel_args_endpoint_config.cc', 'src/core/lib/event_engine/default_event_engine_factory.cc', 'src/core/lib/event_engine/event_engine.cc', + 'src/core/lib/event_engine/iomgr_engine.cc', 'src/core/lib/event_engine/memory_allocator.cc', 'src/core/lib/event_engine/resolved_address.cc', 'src/core/lib/event_engine/slice.cc', 'src/core/lib/event_engine/slice_buffer.cc', 'src/core/lib/event_engine/sockaddr.cc', + 'src/core/lib/event_engine/trace.cc', 'src/core/lib/gpr/alloc.cc', 'src/core/lib/gpr/atm.cc', 'src/core/lib/gpr/cpu_iphone.cc', diff --git a/test/core/event_engine/test_init.cc b/test/core/event_engine/test_init.cc index f4c4e102799..95cea74c928 100644 --- a/test/core/event_engine/test_init.cc +++ b/test/core/event_engine/test_init.cc @@ -23,12 +23,11 @@ namespace grpc_event_engine { namespace experimental { /// Sets the default EventEngine factory, used for testing. -/// Valid engines are: -/// * 'default' or 'libuv': the LibuvEventEngine +/// Currently the only valid engine is 'default' or ''. +/// When more engines are added, this should be updated accordingly. absl::Status InitializeTestingEventEngineFactory(absl::string_view engine) { - if (engine == "default" || engine == "libuv") { - // TODO(hork): SetDefaultEventEngineFactory(LibuvEventEngineFactory) - gpr_log(GPR_DEBUG, "Libuv EventEngine initialized."); + if (engine == "default" || engine.empty()) { + // No-op, the default will be used return absl::OkStatus(); } return absl::InvalidArgumentError( diff --git a/test/core/event_engine/test_suite/BUILD b/test/core/event_engine/test_suite/BUILD index eff948f7d82..06a5783b2e4 100644 --- a/test/core/event_engine/test_suite/BUILD +++ b/test/core/event_engine/test_suite/BUILD @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_package") +load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_cc_test", "grpc_package") licenses(["notice"]) @@ -69,6 +69,15 @@ grpc_cc_library( alwayslink = 1, ) +# -- EventEngine implementations -- + +grpc_cc_test( + name = "iomgr_event_engine_test", + srcs = ["iomgr_event_engine_test.cc"], + uses_polling = False, + deps = ["//test/core/event_engine/test_suite:timer"], +) + # -- Internal targets -- grpc_cc_library( diff --git a/test/core/event_engine/test_suite/README.md b/test/core/event_engine/test_suite/README.md index cb756c44fbe..fd428aee290 100644 --- a/test/core/event_engine/test_suite/README.md +++ b/test/core/event_engine/test_suite/README.md @@ -19,7 +19,7 @@ And the main function will be similar to: ``` #include "path/to/my_custom_event_engine.h" -#include "src/core/event_engine/test_suite/event_engine_test.h" +#include "test/core/event_engine/test_suite/event_engine_test.h" int main(int argc, char** argv) { testing::InitGoogleTest(&argc, argv); diff --git a/test/core/event_engine/test_suite/iomgr_event_engine_test.cc b/test/core/event_engine/test_suite/iomgr_event_engine_test.cc new file mode 100644 index 00000000000..387ac5b5036 --- /dev/null +++ b/test/core/event_engine/test_suite/iomgr_event_engine_test.cc @@ -0,0 +1,31 @@ +// Copyright 2022 The gRPC Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +#include + +#include "src/core/lib/event_engine/iomgr_engine.h" +#include "test/core/event_engine/test_suite/event_engine_test.h" +#include "test/core/util/test_config.h" + +int main(int argc, char** argv) { + testing::InitGoogleTest(&argc, argv); + grpc::testing::TestEnvironment env(&argc, argv); + SetEventEngineFactory([]() { + return absl::make_unique< + grpc_event_engine::experimental::IomgrEventEngine>(); + }); + grpc_init(); + auto result = RUN_ALL_TESTS(); + grpc_shutdown(); + return result; +} diff --git a/test/core/event_engine/test_suite/timer_test.cc b/test/core/event_engine/test_suite/timer_test.cc index 088dc2ce366..fa924dcbe3a 100644 --- a/test/core/event_engine/test_suite/timer_test.cc +++ b/test/core/event_engine/test_suite/timer_test.cc @@ -80,7 +80,7 @@ TEST_F(EventEngineTimerTest, TimersRespectScheduleOrdering) { grpc_core::MutexLock lock(&mu_); { auto engine = this->NewEventEngine(); - engine->RunAt(absl::Now() + absl::Seconds(1), [&]() { + engine->RunAt(absl::Now() + absl::Milliseconds(100), [&]() { grpc_core::MutexLock lock(&mu_); ordered.push_back(2); ++count; @@ -92,9 +92,9 @@ TEST_F(EventEngineTimerTest, TimersRespectScheduleOrdering) { ++count; cv_.Signal(); }); - // Ensure both callbacks have run. Simpler than a mutex. + // Ensure both callbacks have run. while (count != 2) { - cv_.WaitWithTimeout(&mu_, absl::Microseconds(100)); + cv_.WaitWithTimeout(&mu_, absl::Milliseconds(8)); } } // The engine is deleted, and all closures should have been flushed beforehand diff --git a/test/core/transport/bdp_estimator_test.cc b/test/core/transport/bdp_estimator_test.cc index f839723c09c..81ba915dff6 100644 --- a/test/core/transport/bdp_estimator_test.cc +++ b/test/core/transport/bdp_estimator_test.cc @@ -39,8 +39,10 @@ namespace grpc_core { namespace testing { namespace { int g_clock = 123; +Mutex mu_; gpr_timespec fake_gpr_now(gpr_clock_type clock_type) { + MutexLock lock(&mu_); gpr_timespec ts; ts.tv_sec = g_clock; ts.tv_nsec = 0; @@ -48,7 +50,10 @@ gpr_timespec fake_gpr_now(gpr_clock_type clock_type) { return ts; } -void inc_time(void) { g_clock += 30; } +void inc_time(void) { + MutexLock lock(&mu_); + g_clock += 30; +} } // namespace TEST(BdpEstimatorTest, NoOp) { BdpEstimator est("test"); } diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index e666a366900..9a0045512a7 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -1926,12 +1926,17 @@ src/core/lib/event_engine/channel_args_endpoint_config.h \ src/core/lib/event_engine/default_event_engine_factory.cc \ src/core/lib/event_engine/event_engine.cc \ src/core/lib/event_engine/event_engine_factory.h \ +src/core/lib/event_engine/handle_containers.h \ +src/core/lib/event_engine/iomgr_engine.cc \ +src/core/lib/event_engine/iomgr_engine.h \ src/core/lib/event_engine/memory_allocator.cc \ src/core/lib/event_engine/resolved_address.cc \ src/core/lib/event_engine/slice.cc \ src/core/lib/event_engine/slice_buffer.cc \ src/core/lib/event_engine/sockaddr.cc \ src/core/lib/event_engine/sockaddr.h \ +src/core/lib/event_engine/trace.cc \ +src/core/lib/event_engine/trace.h \ src/core/lib/gpr/alloc.cc \ src/core/lib/gpr/alloc.h \ src/core/lib/gpr/atm.cc \ diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 0eacf41d229..7c0ed1e1d08 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -1718,12 +1718,17 @@ src/core/lib/event_engine/channel_args_endpoint_config.h \ src/core/lib/event_engine/default_event_engine_factory.cc \ src/core/lib/event_engine/event_engine.cc \ src/core/lib/event_engine/event_engine_factory.h \ +src/core/lib/event_engine/handle_containers.h \ +src/core/lib/event_engine/iomgr_engine.cc \ +src/core/lib/event_engine/iomgr_engine.h \ src/core/lib/event_engine/memory_allocator.cc \ src/core/lib/event_engine/resolved_address.cc \ src/core/lib/event_engine/slice.cc \ src/core/lib/event_engine/slice_buffer.cc \ src/core/lib/event_engine/sockaddr.cc \ src/core/lib/event_engine/sockaddr.h \ +src/core/lib/event_engine/trace.cc \ +src/core/lib/event_engine/trace.h \ src/core/lib/gpr/README.md \ src/core/lib/gpr/alloc.cc \ src/core/lib/gpr/alloc.h \ diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index 0e5164ac880..0df3dae4f77 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -5043,6 +5043,30 @@ ], "uses_polling": true }, + { + "args": [], + "benchmark": false, + "ci_platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "cpu_cost": 1.0, + "exclude_configs": [], + "exclude_iomgrs": [], + "flaky": false, + "gtest": true, + "language": "c++", + "name": "iomgr_event_engine_test", + "platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "uses_polling": false + }, { "args": [], "benchmark": false,