diff --git a/BUILD b/BUILD index e87e2e81c85..58206feeb03 100644 --- a/BUILD +++ b/BUILD @@ -1170,6 +1170,7 @@ grpc_cc_library( "//src/core:lib/compression/compression.cc", "//src/core:lib/compression/compression_internal.cc", "//src/core:lib/compression/message_compress.cc", + "//src/core:lib/event_engine/channel_args_endpoint_config.cc", "//src/core:lib/iomgr/buffer_list.cc", "//src/core:lib/iomgr/call_combiner.cc", "//src/core:lib/iomgr/cfstream_handle.cc", @@ -1273,6 +1274,7 @@ grpc_cc_library( "//src/core:lib/channel/status_util.h", "//src/core:lib/compression/compression_internal.h", "//src/core:lib/compression/message_compress.h", + "//src/core:lib/event_engine/channel_args_endpoint_config.h", "//src/core:lib/iomgr/block_annotate.h", "//src/core:lib/iomgr/buffer_list.h", "//src/core:lib/iomgr/call_combiner.h", @@ -1386,7 +1388,6 @@ grpc_cc_library( "//src/core:atomic_utils", "//src/core:bitset", "//src/core:channel_args", - "//src/core:channel_args_endpoint_config", "//src/core:channel_args_preconditioning", "//src/core:channel_fwd", "//src/core:channel_init", diff --git a/CMakeLists.txt b/CMakeLists.txt index 995ed3af96e..46b88c26e80 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -2153,7 +2153,6 @@ add_library(grpc src/core/lib/event_engine/utils.cc src/core/lib/event_engine/windows/iocp.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/experiments/config.cc src/core/lib/experiments/experiments.cc @@ -2823,7 +2822,6 @@ add_library(grpc_unsecure src/core/lib/event_engine/utils.cc src/core/lib/event_engine/windows/iocp.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/experiments/config.cc src/core/lib/experiments/experiments.cc @@ -4310,7 +4308,6 @@ add_library(grpc_authorization_provider src/core/lib/event_engine/utils.cc src/core/lib/event_engine/windows/iocp.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/experiments/config.cc src/core/lib/experiments/experiments.cc @@ -9929,10 +9926,6 @@ endif() if(gRPC_BUILD_TESTS) add_executable(endpoint_config_test - src/core/lib/channel/channel_args.cc - src/core/lib/event_engine/channel_args_endpoint_config.cc - src/core/lib/gprpp/time.cc - src/core/lib/surface/channel_stack_type.cc test/core/event_engine/endpoint_config_test.cc third_party/googletest/googletest/src/gtest-all.cc third_party/googletest/googlemock/src/gmock-all.cc @@ -9962,10 +9955,7 @@ target_link_libraries(endpoint_config_test ${_gRPC_PROTOBUF_LIBRARIES} ${_gRPC_ZLIB_LIBRARIES} ${_gRPC_ALLTARGETS_LIBRARIES} - absl::any_invocable - absl::type_traits - absl::statusor - gpr + grpc ) @@ -11280,7 +11270,6 @@ add_executable(frame_test src/core/lib/event_engine/utils.cc src/core/lib/event_engine/windows/iocp.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/experiments/config.cc src/core/lib/experiments/experiments.cc @@ -21803,6 +21792,7 @@ if(gRPC_BUILD_TESTS) if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_POSIX OR _gRPC_PLATFORM_WINDOWS) add_executable(windows_endpoint_test + src/core/lib/event_engine/windows/windows_endpoint.cc test/core/event_engine/windows/create_sockpair.cc test/core/event_engine/windows/windows_endpoint_test.cc third_party/googletest/googletest/src/gtest-all.cc diff --git a/Makefile b/Makefile index 95b48a4ed49..446c37a5fa7 100644 --- a/Makefile +++ b/Makefile @@ -1428,7 +1428,6 @@ LIBGRPC_SRC = \ src/core/lib/event_engine/utils.cc \ src/core/lib/event_engine/windows/iocp.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/experiments/config.cc \ src/core/lib/experiments/experiments.cc \ @@ -1957,7 +1956,6 @@ LIBGRPC_UNSECURE_SRC = \ src/core/lib/event_engine/utils.cc \ src/core/lib/event_engine/windows/iocp.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/experiments/config.cc \ src/core/lib/experiments/experiments.cc \ diff --git a/build_autogenerated.yaml b/build_autogenerated.yaml index 54c907cec99..8625192d49e 100644 --- a/build_autogenerated.yaml +++ b/build_autogenerated.yaml @@ -786,7 +786,6 @@ libs: - src/core/lib/event_engine/utils.h - src/core/lib/event_engine/windows/iocp.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/experiments/config.h - src/core/lib/experiments/experiments.h @@ -1530,7 +1529,6 @@ libs: - src/core/lib/event_engine/utils.cc - src/core/lib/event_engine/windows/iocp.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/experiments/config.cc - src/core/lib/experiments/experiments.cc @@ -2079,7 +2077,6 @@ libs: - src/core/lib/event_engine/utils.h - src/core/lib/event_engine/windows/iocp.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/experiments/config.h - src/core/lib/experiments/experiments.h @@ -2459,7 +2456,6 @@ libs: - src/core/lib/event_engine/utils.cc - src/core/lib/event_engine/windows/iocp.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/experiments/config.cc - src/core/lib/experiments/experiments.cc @@ -3516,7 +3512,6 @@ libs: - src/core/lib/event_engine/utils.h - src/core/lib/event_engine/windows/iocp.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/experiments/config.h - src/core/lib/experiments/experiments.h @@ -3776,7 +3771,6 @@ libs: - src/core/lib/event_engine/utils.cc - src/core/lib/event_engine/windows/iocp.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/experiments/config.cc - src/core/lib/experiments/experiments.cc @@ -6379,31 +6373,11 @@ targets: gtest: true build: test language: c++ - headers: - - src/core/lib/avl/avl.h - - src/core/lib/channel/channel_args.h - - src/core/lib/event_engine/channel_args_endpoint_config.h - - src/core/lib/gprpp/atomic_utils.h - - src/core/lib/gprpp/debug_location.h - - src/core/lib/gprpp/dual_ref_counted.h - - src/core/lib/gprpp/match.h - - src/core/lib/gprpp/orphanable.h - - src/core/lib/gprpp/overload.h - - src/core/lib/gprpp/ref_counted.h - - src/core/lib/gprpp/ref_counted_ptr.h - - src/core/lib/gprpp/time.h - - src/core/lib/surface/channel_stack_type.h + headers: [] src: - - src/core/lib/channel/channel_args.cc - - src/core/lib/event_engine/channel_args_endpoint_config.cc - - src/core/lib/gprpp/time.cc - - src/core/lib/surface/channel_stack_type.cc - test/core/event_engine/endpoint_config_test.cc deps: - - absl/functional:any_invocable - - absl/meta:type_traits - - absl/status:statusor - - gpr + - grpc uses_polling: false - name: endpoint_pair_test gtest: true @@ -7252,7 +7226,6 @@ targets: - src/core/lib/event_engine/utils.h - src/core/lib/event_engine/windows/iocp.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/experiments/config.h - src/core/lib/experiments/experiments.h @@ -7494,7 +7467,6 @@ targets: - src/core/lib/event_engine/utils.cc - src/core/lib/event_engine/windows/iocp.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/experiments/config.cc - src/core/lib/experiments/experiments.cc @@ -12216,8 +12188,10 @@ targets: build: test language: c++ headers: + - src/core/lib/event_engine/windows/windows_endpoint.h - test/core/event_engine/windows/create_sockpair.h src: + - src/core/lib/event_engine/windows/windows_endpoint.cc - test/core/event_engine/windows/create_sockpair.cc - test/core/event_engine/windows/windows_endpoint_test.cc deps: diff --git a/config.m4 b/config.m4 index 267eb0bf676..db476c3bb39 100644 --- a/config.m4 +++ b/config.m4 @@ -511,7 +511,6 @@ if test "$PHP_GRPC" != "no"; then src/core/lib/event_engine/utils.cc \ src/core/lib/event_engine/windows/iocp.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/experiments/config.cc \ src/core/lib/experiments/experiments.cc \ diff --git a/config.w32 b/config.w32 index 8dc7b220d7a..3edbcb8466d 100644 --- a/config.w32 +++ b/config.w32 @@ -477,7 +477,6 @@ if (PHP_GRPC != "no") { "src\\core\\lib\\event_engine\\utils.cc " + "src\\core\\lib\\event_engine\\windows\\iocp.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\\experiments\\config.cc " + "src\\core\\lib\\experiments\\experiments.cc " + diff --git a/gRPC-C++.podspec b/gRPC-C++.podspec index 8a7b315eaf5..997576ae10d 100644 --- a/gRPC-C++.podspec +++ b/gRPC-C++.podspec @@ -739,7 +739,6 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/utils.h', 'src/core/lib/event_engine/windows/iocp.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/experiments/config.h', 'src/core/lib/experiments/experiments.h', @@ -1633,7 +1632,6 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/utils.h', 'src/core/lib/event_engine/windows/iocp.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/experiments/config.h', 'src/core/lib/experiments/experiments.h', diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 81adb310660..a9f45eb5afa 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -1140,8 +1140,6 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/windows/iocp.h', 'src/core/lib/event_engine/windows/win_socket.cc', 'src/core/lib/event_engine/windows/win_socket.h', - 'src/core/lib/event_engine/windows/windows_endpoint.cc', - 'src/core/lib/event_engine/windows/windows_endpoint.h', 'src/core/lib/event_engine/windows/windows_engine.cc', 'src/core/lib/event_engine/windows/windows_engine.h', 'src/core/lib/experiments/config.cc', @@ -2285,7 +2283,6 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/utils.h', 'src/core/lib/event_engine/windows/iocp.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/experiments/config.h', 'src/core/lib/experiments/experiments.h', diff --git a/grpc.gemspec b/grpc.gemspec index a82c45bfa96..bd46521e276 100644 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -1051,8 +1051,6 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/event_engine/windows/iocp.h ) s.files += %w( src/core/lib/event_engine/windows/win_socket.cc ) s.files += %w( src/core/lib/event_engine/windows/win_socket.h ) - s.files += %w( src/core/lib/event_engine/windows/windows_endpoint.cc ) - s.files += %w( src/core/lib/event_engine/windows/windows_endpoint.h ) s.files += %w( src/core/lib/event_engine/windows/windows_engine.cc ) s.files += %w( src/core/lib/event_engine/windows/windows_engine.h ) s.files += %w( src/core/lib/experiments/config.cc ) diff --git a/grpc.gyp b/grpc.gyp index 23d5df2bb3c..2ba5cda5dbf 100644 --- a/grpc.gyp +++ b/grpc.gyp @@ -841,7 +841,6 @@ 'src/core/lib/event_engine/utils.cc', 'src/core/lib/event_engine/windows/iocp.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/experiments/config.cc', 'src/core/lib/experiments/experiments.cc', @@ -1312,7 +1311,6 @@ 'src/core/lib/event_engine/utils.cc', 'src/core/lib/event_engine/windows/iocp.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/experiments/config.cc', 'src/core/lib/experiments/experiments.cc', @@ -1810,7 +1808,6 @@ 'src/core/lib/event_engine/utils.cc', 'src/core/lib/event_engine/windows/iocp.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/experiments/config.cc', 'src/core/lib/experiments/experiments.cc', diff --git a/package.xml b/package.xml index 52b63520658..87c80c7da69 100644 --- a/package.xml +++ b/package.xml @@ -1033,8 +1033,6 @@ - - diff --git a/src/core/BUILD b/src/core/BUILD index 6abd0840b44..1b6d6bc3a0b 100644 --- a/src/core/BUILD +++ b/src/core/BUILD @@ -908,7 +908,6 @@ grpc_cc_library( ], deps = [ "channel_args", - "channel_args_endpoint_config", "closure", "error", "handshaker_factory", @@ -1870,19 +1869,13 @@ grpc_cc_library( "absl/strings", ], deps = [ - "channel_args_endpoint_config", - "common_event_engine_closures", - "error", "event_engine_common", - "event_engine_executor", - "event_engine_tcp_socket_utils", "event_engine_thread_pool", "event_engine_trace", "event_engine_utils", "init_internally", "posix_event_engine_timer_manager", "time", - "windows_endpoint", "windows_iocp", "//:event_engine_base_hdrs", "//:gpr", @@ -2005,26 +1998,6 @@ grpc_cc_library( ], ) -grpc_cc_library( - name = "channel_args_endpoint_config", - srcs = [ - "//src/core:lib/event_engine/channel_args_endpoint_config.cc", - ], - hdrs = [ - "//src/core:lib/event_engine/channel_args_endpoint_config.h", - ], - external_deps = [ - "absl/strings", - "absl/types:optional", - ], - visibility = ["@grpc:alt_grpc_base_legacy"], - deps = [ - "channel_args", - "//:event_engine_base_hdrs", - "//:gpr_platform", - ], -) - grpc_cc_library( name = "default_event_engine", srcs = [ @@ -4893,7 +4866,6 @@ grpc_cc_library( language = "c++", deps = [ "channel_args", - "channel_args_endpoint_config", "channel_args_preconditioning", "channel_stack_type", "closure", @@ -4943,7 +4915,6 @@ grpc_cc_library( language = "c++", deps = [ "channel_args", - "channel_args_endpoint_config", "closure", "error", "grpc_insecure_credentials", diff --git a/src/core/lib/event_engine/handle_containers.h b/src/core/lib/event_engine/handle_containers.h index e966ca47457..a1739ef3c08 100644 --- a/src/core/lib/event_engine/handle_containers.h +++ b/src/core/lib/event_engine/handle_containers.h @@ -54,13 +54,6 @@ using TaskHandleSet = absl::flat_hash_set< TaskHandleComparator< grpc_event_engine::experimental::EventEngine::TaskHandle>::Eq>; -using ConnectionHandleSet = absl::flat_hash_set< - grpc_event_engine::experimental::EventEngine::ConnectionHandle, - TaskHandleComparator< - grpc_event_engine::experimental::EventEngine::ConnectionHandle>::Hash, - TaskHandleComparator< - grpc_event_engine::experimental::EventEngine::ConnectionHandle>::Eq>; - using LookupTaskHandleSet = absl::flat_hash_set< grpc_event_engine::experimental::EventEngine::DNSResolver::LookupTaskHandle, TaskHandleComparatorsin6_addr.s6_addr[12], &addr4->sin_addr, 4); addr6_out->sin6_port = addr4->sin_port; *resolved_addr6_out = EventEngine::ResolvedAddress( - reinterpret_cast(addr6_out), sizeof(sockaddr_in6)); + reinterpret_cast(addr6_out), + static_cast(sizeof(sockaddr_in6))); return true; } return false; diff --git a/src/core/lib/event_engine/utils.cc b/src/core/lib/event_engine/utils.cc index 0c9dddf97df..9769faeeb8b 100644 --- a/src/core/lib/event_engine/utils.cc +++ b/src/core/lib/event_engine/utils.cc @@ -28,8 +28,8 @@ namespace grpc_event_engine { namespace experimental { -std::string HandleToStringInternal(uintptr_t a, uintptr_t b) { - return absl::StrCat("{", a, ",", b, "}"); +std::string HandleToString(EventEngine::TaskHandle handle) { + return absl::StrCat("{", handle.keys[0], ",", handle.keys[1], "}"); } grpc_core::Timestamp ToTimestamp(grpc_core::Timestamp now, diff --git a/src/core/lib/event_engine/utils.h b/src/core/lib/event_engine/utils.h index 04068d1d215..768d16ceea9 100644 --- a/src/core/lib/event_engine/utils.h +++ b/src/core/lib/event_engine/utils.h @@ -16,8 +16,6 @@ #include -#include - #include #include @@ -27,13 +25,7 @@ namespace grpc_event_engine { namespace experimental { -std::string HandleToStringInternal(uintptr_t a, uintptr_t b); - -// Returns a string representation of the EventEngine::*Handle types -template -std::string HandleToString(const Handle& handle) { - return HandleToStringInternal(handle.keys[0], handle.keys[1]); -} +std::string HandleToString(EventEngine::TaskHandle handle); grpc_core::Timestamp ToTimestamp(grpc_core::Timestamp now, EventEngine::Duration delta); @@ -41,4 +33,4 @@ grpc_core::Timestamp ToTimestamp(grpc_core::Timestamp now, } // namespace experimental } // namespace grpc_event_engine -#endif // GRPC_CORE_LIB_EVENT_ENGINE_UTILS_H +#endif // GRPC_CORE_LIB_EVENT_ENGINE_UTILS_H \ No newline at end of file diff --git a/src/core/lib/event_engine/windows/iocp.cc b/src/core/lib/event_engine/windows/iocp.cc index dbfbbc56b2c..23fe4969eb0 100644 --- a/src/core/lib/event_engine/windows/iocp.cc +++ b/src/core/lib/event_engine/windows/iocp.cc @@ -32,8 +32,8 @@ namespace experimental { IOCP::IOCP(Executor* executor) noexcept : executor_(executor), - iocp_handle_(CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, - (ULONG_PTR) nullptr, 0)) { + iocp_handle_(CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, + (ULONG_PTR)NULL, 0)) { GPR_ASSERT(iocp_handle_); WSASocketFlagsInit(); } @@ -58,10 +58,6 @@ std::unique_ptr IOCP::Watch(SOCKET socket) { } void IOCP::Shutdown() { - if (GRPC_TRACE_FLAG_ENABLED(grpc_event_engine_trace)) { - gpr_log(GPR_DEBUG, "IOCP::%p shutting down. Outstanding kicks: %d", this, - outstanding_kicks_.load()); - } while (outstanding_kicks_.load() > 0) { Work(std::chrono::hours(42), []() {}); } @@ -70,6 +66,10 @@ void IOCP::Shutdown() { Poller::WorkResult IOCP::Work(EventEngine::Duration timeout, absl::FunctionRef schedule_poll_again) { + static const absl::Status kDeadlineExceeded = absl::DeadlineExceededError( + absl::StrFormat("IOCP::%p: Received no completions", this)); + static const absl::Status kKicked = + absl::AbortedError(absl::StrFormat("IOCP::%p: Awoken from a kick", this)); DWORD bytes = 0; ULONG_PTR completion_key; LPOVERLAPPED overlapped; @@ -79,7 +79,7 @@ Poller::WorkResult IOCP::Work(EventEngine::Duration timeout, BOOL success = GetQueuedCompletionStatus( iocp_handle_, &bytes, &completion_key, &overlapped, static_cast(Milliseconds(timeout))); - if (success == 0 && overlapped == nullptr) { + if (success == 0 && overlapped == NULL) { if (GRPC_TRACE_FLAG_ENABLED(grpc_event_engine_trace)) { gpr_log(GPR_DEBUG, "IOCP::%p deadline exceeded", this); } @@ -139,7 +139,7 @@ DWORD IOCP::WSASocketFlagsInit() { // versions, see // https://msdn.microsoft.com/en-us/library/windows/desktop/ms742212(v=vs.85).aspx // for details. - SOCKET sock = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, + SOCKET sock = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, NULL, 0, wsa_socket_flags | WSA_FLAG_NO_HANDLE_INHERIT); if (sock != INVALID_SOCKET) { // Windows 7, Windows 2008 R2 with SP1 or later diff --git a/src/core/lib/event_engine/windows/win_socket.cc b/src/core/lib/event_engine/windows/win_socket.cc index c6a9f3c7423..4ae64aaf41a 100644 --- a/src/core/lib/event_engine/windows/win_socket.cc +++ b/src/core/lib/event_engine/windows/win_socket.cc @@ -98,9 +98,7 @@ void WinSocket::NotifyOnWrite(EventEngine::Closure* on_write) { } WinSocket::OpState::OpState(WinSocket* win_socket) noexcept - : win_socket_(win_socket), closure_(nullptr) { - memset(&overlapped_, 0, sizeof(OVERLAPPED)); -} + : win_socket_(win_socket), closure_(nullptr) {} void WinSocket::OpState::SetReady() { GPR_ASSERT(!has_pending_iocp_); diff --git a/src/core/lib/event_engine/windows/windows_endpoint.cc b/src/core/lib/event_engine/windows/windows_endpoint.cc index 9afb25e4d0e..c77b925c96f 100644 --- a/src/core/lib/event_engine/windows/windows_endpoint.cc +++ b/src/core/lib/event_engine/windows/windows_endpoint.cc @@ -60,17 +60,13 @@ WindowsEndpoint::WindowsEndpoint( handle_read_event_(this), handle_write_event_(this), executor_(executor) { - char addr[EventEngine::ResolvedAddress::MAX_SIZE_BYTES]; + sockaddr addr; int addr_len = sizeof(addr); - if (getsockname(socket_->socket(), reinterpret_cast(addr), - &addr_len) < 0) { - gpr_log( - GPR_ERROR, "Unrecoverable error: Failed to get local socket name. %s", - GRPC_WSA_ERROR(WSAGetLastError(), "getsockname").ToString().c_str()); + if (getsockname(socket_->socket(), &addr, &addr_len) < 0) { + gpr_log(GPR_ERROR, "Unrecoverable error: Failed to get local socket name."); abort(); } - local_address_ = - EventEngine::ResolvedAddress(reinterpret_cast(addr), addr_len); + local_address_ = EventEngine::ResolvedAddress(&addr, addr_len); local_address_string_ = *ResolvedAddressToURI(local_address_); peer_address_string_ = *ResolvedAddressToURI(peer_address_); } diff --git a/src/core/lib/event_engine/windows/windows_engine.cc b/src/core/lib/event_engine/windows/windows_engine.cc index 6772fe486f8..28b648848c3 100644 --- a/src/core/lib/event_engine/windows/windows_engine.cc +++ b/src/core/lib/event_engine/windows/windows_engine.cc @@ -26,67 +26,33 @@ #include #include -#include "src/core/lib/event_engine/channel_args_endpoint_config.h" -#include "src/core/lib/event_engine/common_closures.h" -#include "src/core/lib/event_engine/executor/executor.h" #include "src/core/lib/event_engine/handle_containers.h" #include "src/core/lib/event_engine/posix_engine/timer_manager.h" -#include "src/core/lib/event_engine/tcp_socket_utils.h" #include "src/core/lib/event_engine/trace.h" #include "src/core/lib/event_engine/utils.h" #include "src/core/lib/event_engine/windows/iocp.h" -#include "src/core/lib/event_engine/windows/windows_endpoint.h" #include "src/core/lib/event_engine/windows/windows_engine.h" #include "src/core/lib/gprpp/sync.h" #include "src/core/lib/gprpp/time.h" -#include "src/core/lib/iomgr/error.h" namespace grpc_event_engine { namespace experimental { -// ---- IOCPWorkClosure ---- - -WindowsEventEngine::IOCPWorkClosure::IOCPWorkClosure(Executor* executor, - IOCP* iocp) - : executor_(executor), iocp_(iocp) { - executor_->Run(this); -} - -void WindowsEventEngine::IOCPWorkClosure::Run() { - auto result = iocp_->Work(std::chrono::seconds(60), [this] { - workers_.fetch_add(1); - executor_->Run(this); - }); - if (result == Poller::WorkResult::kDeadlineExceeded) { - // iocp received no messages. restart the worker - workers_.fetch_add(1); - executor_->Run(this); - } - if (workers_.fetch_sub(1) == 1) done_signal_.Notify(); -} - -void WindowsEventEngine::IOCPWorkClosure::WaitForShutdown() { - done_signal_.WaitForNotification(); -} - -// ---- WindowsEventEngine ---- - // TODO(hork): The iomgr timer and execution engine can be reused. It should // be separated out from the posix_engine and instantiated as components. It is // effectively copied below. -struct WindowsEventEngine::TimerClosure final : public EventEngine::Closure { +struct WindowsEventEngine::Closure final : public EventEngine::Closure { absl::AnyInvocable cb; Timer timer; WindowsEventEngine* engine; EventEngine::TaskHandle handle; void Run() override { - GRPC_EVENT_ENGINE_TRACE( - "WindowsEventEngine:%p executing callback:%s", engine, - HandleToString(handle).c_str()); + GRPC_EVENT_ENGINE_TRACE("WindowsEventEngine:%p executing callback:%s", + engine, HandleToString(handle).c_str()); { - grpc_core::MutexLock lock(&engine->task_mu_); + grpc_core::MutexLock lock(&engine->mu_); engine->known_handles_.erase(handle); } cb(); @@ -97,8 +63,7 @@ struct WindowsEventEngine::TimerClosure final : public EventEngine::Closure { WindowsEventEngine::WindowsEventEngine() : executor_(std::make_shared()), iocp_(executor_.get()), - timer_manager_(executor_), - iocp_worker_(executor_.get(), &iocp_) { + timer_manager_(executor_) { WSADATA wsaData; int status = WSAStartup(MAKEWORD(2, 0), &wsaData); GPR_ASSERT(status == 0); @@ -106,28 +71,25 @@ WindowsEventEngine::WindowsEventEngine() WindowsEventEngine::~WindowsEventEngine() { { - grpc_core::MutexLock lock(&task_mu_); + grpc_core::MutexLock lock(&mu_); if (GRPC_TRACE_FLAG_ENABLED(grpc_event_engine_trace)) { for (auto handle : known_handles_) { gpr_log(GPR_ERROR, "WindowsEventEngine:%p uncleared TaskHandle at shutdown:%s", - this, HandleToString(handle).c_str()); + this, HandleToString(handle).c_str()); } } GPR_ASSERT(GPR_LIKELY(known_handles_.empty())); + GPR_ASSERT(WSACleanup() == 0); + timer_manager_.Shutdown(); } - iocp_.Kick(); - iocp_worker_.WaitForShutdown(); - iocp_.Shutdown(); - GPR_ASSERT(WSACleanup() == 0); - timer_manager_.Shutdown(); executor_->Quiesce(); } bool WindowsEventEngine::Cancel(EventEngine::TaskHandle handle) { - grpc_core::MutexLock lock(&task_mu_); + grpc_core::MutexLock lock(&mu_); if (!known_handles_.contains(handle)) return false; - auto* cd = reinterpret_cast(handle.keys[0]); + auto* cd = reinterpret_cast(handle.keys[0]); bool r = timer_manager_.TimerCancel(&cd->timer); known_handles_.erase(handle); if (r) delete cd; @@ -155,17 +117,16 @@ void WindowsEventEngine::Run(EventEngine::Closure* closure) { EventEngine::TaskHandle WindowsEventEngine::RunAfterInternal( Duration when, absl::AnyInvocable cb) { auto when_ts = ToTimestamp(timer_manager_.Now(), when); - auto* cd = new TimerClosure; + auto* cd = new Closure; cd->cb = std::move(cb); cd->engine = this; EventEngine::TaskHandle handle{reinterpret_cast(cd), aba_token_.fetch_add(1)}; - grpc_core::MutexLock lock(&task_mu_); + grpc_core::MutexLock lock(&mu_); known_handles_.insert(handle); cd->handle = handle; - GRPC_EVENT_ENGINE_TRACE( - "WindowsEventEngine:%p scheduling callback:%s", this, - HandleToString(handle).c_str()); + GRPC_EVENT_ENGINE_TRACE("WindowsEventEngine:%p scheduling callback:%s", this, + HandleToString(handle).c_str()); timer_manager_.TimerInit(&cd->timer, when_ts, cd); return handle; } @@ -179,194 +140,15 @@ bool WindowsEventEngine::IsWorkerThread() { GPR_ASSERT(false && "unimplemented"); } -void WindowsEventEngine::OnConnectCompleted( - std::shared_ptr state) { - // Connection attempt complete! - grpc_core::MutexLock lock(&state->mu); - state->on_connected = nullptr; - { - grpc_core::MutexLock handle_lock(&connection_mu_); - known_connection_handles_.erase(state->connection_handle); - } - // return early if we cannot cancel the connection timeout timer. - if (!Cancel(state->timer_handle)) return; - auto write_info = state->socket->write_info(); - if (write_info->wsa_error() != 0) { - auto error = GRPC_WSA_ERROR(write_info->wsa_error(), "ConnectEx"); - state->socket->MaybeShutdown(error); - state->on_connected_user_callback(error); - return; - } - // This code should be running in an executor thread already, so the callback - // can be run directly. - ChannelArgsEndpointConfig cfg; - state->on_connected_user_callback(std::make_unique( - state->address, std::move(state->socket), std::move(state->allocator), - cfg, executor_.get())); +bool WindowsEventEngine::CancelConnect(EventEngine::ConnectionHandle handle) { + GPR_ASSERT(false && "unimplemented"); } EventEngine::ConnectionHandle WindowsEventEngine::Connect( OnConnectCallback on_connect, const ResolvedAddress& addr, - const EndpointConfig& /* args */, MemoryAllocator memory_allocator, - Duration timeout) { - // TODO(hork): utilize the endpoint config - absl::Status status; - int istatus; - auto uri = ResolvedAddressToURI(addr); - if (!uri.ok()) { - Run([on_connect = std::move(on_connect), status = uri.status()]() mutable { - on_connect(status); - }); - return invalid_connection_handle; - } - GRPC_EVENT_ENGINE_TRACE("EventEngine::%p connecting to %s", this, - uri->c_str()); - // Use dualstack sockets where available. - ResolvedAddress address = addr; - ResolvedAddress addr6_v4mapped; - if (ResolvedAddressToV4Mapped(addr, &addr6_v4mapped)) { - address = addr6_v4mapped; - } - SOCKET sock = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, - IOCP::GetDefaultSocketFlags()); - if (sock == INVALID_SOCKET) { - Run([on_connect = std::move(on_connect), - status = GRPC_WSA_ERROR(WSAGetLastError(), "WSASocket")]() mutable { - on_connect(status); - }); - return invalid_connection_handle; - } - status = PrepareSocket(sock); - if (!status.ok()) { - Run([on_connect = std::move(on_connect), status]() mutable { - on_connect(status); - }); - return invalid_connection_handle; - } - // Grab the function pointer for ConnectEx for that specific socket It may - // change depending on the interface. - LPFN_CONNECTEX ConnectEx; - GUID guid = WSAID_CONNECTEX; - DWORD ioctl_num_bytes; - istatus = WSAIoctl(sock, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, - sizeof(guid), &ConnectEx, sizeof(ConnectEx), - &ioctl_num_bytes, nullptr, nullptr); - if (istatus != 0) { - Run([on_connect = std::move(on_connect), - status = GRPC_WSA_ERROR( - WSAGetLastError(), - "WSAIoctl(SIO_GET_EXTENSION_FUNCTION_POINTER)")]() mutable { - on_connect(status); - }); - return invalid_connection_handle; - } - // bind the local address - auto local_address = ResolvedAddressMakeWild6(0); - istatus = bind(sock, local_address.address(), local_address.size()); - if (istatus != 0) { - Run([on_connect = std::move(on_connect), - status = GRPC_WSA_ERROR(WSAGetLastError(), "bind")]() mutable { - on_connect(status); - }); - return invalid_connection_handle; - } - // Connect - auto watched_socket = iocp_.Watch(sock); - auto* info = watched_socket->write_info(); - bool success = - ConnectEx(watched_socket->socket(), address.address(), address.size(), - nullptr, 0, nullptr, info->overlapped()); - // It wouldn't be unusual to get a success immediately. But we'll still get an - // IOCP notification, so let's ignore it. - if (!success) { - int last_error = WSAGetLastError(); - if (last_error != ERROR_IO_PENDING) { - auto status = GRPC_WSA_ERROR(WSAGetLastError(), "ConnectEx"); - Run([on_connect = std::move(on_connect), status]() mutable { - on_connect(status); - }); - watched_socket->MaybeShutdown(status); - return invalid_connection_handle; - } - } - GPR_ASSERT(watched_socket != nullptr); - auto connection_state = std::make_shared(); - grpc_core::MutexLock lock(&connection_state->mu); - connection_state->address = address; - connection_state->socket = std::move(watched_socket); - connection_state->on_connected_user_callback = std::move(on_connect); - connection_state->allocator = std::move(memory_allocator); - connection_state->on_connected = - SelfDeletingClosure::Create([this, connection_state]() mutable { - OnConnectCompleted(std::move(connection_state)); - }); - { - grpc_core::MutexLock conn_lock(&connection_mu_); - connection_state->connection_handle = - ConnectionHandle{reinterpret_cast(connection_state.get()), - aba_token_.fetch_add(1)}; - known_connection_handles_.insert(connection_state->connection_handle); - } - connection_state->timer_handle = - RunAfter(timeout, [this, connection_state]() { - grpc_core::MutexLock lock(&connection_state->mu); - if (CancelConnectFromDeadlineTimer(connection_state.get())) { - connection_state->on_connected_user_callback( - absl::DeadlineExceededError("Connection timed out")); - } - // else: The connection attempt could not be canceled. We can assume the - // connection callback will be called. - }); - connection_state->socket->NotifyOnWrite(connection_state->on_connected); - return connection_state->connection_handle; -} - -bool WindowsEventEngine::CancelConnect(EventEngine::ConnectionHandle handle) { - if (TaskHandleComparator::Eq()(handle, - invalid_connection_handle)) { - GRPC_EVENT_ENGINE_TRACE("%s", - "Attempted to cancel an invalid connection handle"); - return false; - } - // Erase the connection handle, which may be unknown - { - grpc_core::MutexLock lock(&connection_mu_); - if (!known_connection_handles_.contains(handle)) { - GRPC_EVENT_ENGINE_TRACE( - "Unknown connection handle: %s", - HandleToString(handle).c_str()); - return false; - } - known_connection_handles_.erase(handle); - } - auto* connection_state = reinterpret_cast(handle.keys[0]); - grpc_core::MutexLock state_lock(&connection_state->mu); - if (!Cancel(connection_state->timer_handle)) return false; - return CancelConnectInternalStateLocked(connection_state); -} - -bool WindowsEventEngine::CancelConnectFromDeadlineTimer( - ConnectionState* connection_state) { - // Erase the connection handle, which is guaranteed to exist. - { - grpc_core::MutexLock lock(&connection_mu_); - GPR_ASSERT(known_connection_handles_.erase( - connection_state->connection_handle) == 1); - } - return CancelConnectInternalStateLocked(connection_state); -} - -bool WindowsEventEngine::CancelConnectInternalStateLocked( - ConnectionState* connection_state) { - connection_state->socket->MaybeShutdown( - absl::CancelledError("CancelConnect")); - // Release the connection_state shared_ptr. connection_state is now invalid. - delete connection_state->on_connected; - GRPC_EVENT_ENGINE_TRACE("Successfully cancelled connection %s", - HandleToString( - connection_state->connection_handle) - .c_str()); - return true; + const EndpointConfig& args, MemoryAllocator memory_allocator, + Duration deadline) { + GPR_ASSERT(false && "unimplemented"); } absl::StatusOr> diff --git a/src/core/lib/event_engine/windows/windows_engine.h b/src/core/lib/event_engine/windows/windows_engine.h index 23ba1d036f5..59a2dd0f0a4 100644 --- a/src/core/lib/event_engine/windows/windows_engine.h +++ b/src/core/lib/event_engine/windows/windows_engine.h @@ -32,7 +32,6 @@ #include "src/core/lib/event_engine/posix_engine/timer_manager.h" #include "src/core/lib/event_engine/thread_pool.h" #include "src/core/lib/event_engine/windows/iocp.h" -#include "src/core/lib/event_engine/windows/windows_endpoint.h" #include "src/core/lib/gprpp/sync.h" #include "src/core/lib/gprpp/time.h" #include "src/core/lib/surface/init_internally.h" @@ -45,10 +44,16 @@ namespace experimental { class WindowsEventEngine : public EventEngine, public grpc_core::KeepsGrpcInitialized { public: - constexpr static TaskHandle invalid_handle{-1, -1}; - constexpr static EventEngine::ConnectionHandle invalid_connection_handle{-1, - -1}; - + class WindowsEndpoint : public EventEngine::Endpoint { + public: + ~WindowsEndpoint() override; + void Read(absl::AnyInvocable on_read, + SliceBuffer* buffer, const ReadArgs* args) override; + void Write(absl::AnyInvocable on_writable, + SliceBuffer* data, const WriteArgs* args) override; + const ResolvedAddress& GetPeerAddress() const override; + const ResolvedAddress& GetLocalAddress() const override; + }; class WindowsListener : public EventEngine::Listener { public: ~WindowsListener() override; @@ -99,64 +104,16 @@ class WindowsEventEngine : public EventEngine, bool Cancel(TaskHandle handle) override; private: - // State of an active connection. - // Managed by a shared_ptr, owned exclusively by the timeout callback and the - // OnConnectCompleted callback herein. - struct ConnectionState { - // everything is guarded by mu; - grpc_core::Mutex mu - ABSL_ACQUIRED_BEFORE(WindowsEventEngine::connection_mu_); - EventEngine::ConnectionHandle connection_handle ABSL_GUARDED_BY(mu); - EventEngine::TaskHandle timer_handle ABSL_GUARDED_BY(mu); - EventEngine::OnConnectCallback on_connected_user_callback - ABSL_GUARDED_BY(mu); - EventEngine::Closure* on_connected ABSL_GUARDED_BY(mu); - std::unique_ptr socket ABSL_GUARDED_BY(mu); - EventEngine::ResolvedAddress address ABSL_GUARDED_BY(mu); - MemoryAllocator allocator ABSL_GUARDED_BY(mu); - }; - - // A poll worker which schedules itself unless kicked - class IOCPWorkClosure : public EventEngine::Closure { - public: - explicit IOCPWorkClosure(Executor* executor, IOCP* iocp); - void Run() override; - void WaitForShutdown(); - - private: - std::atomic workers_{1}; - grpc_core::Notification done_signal_; - Executor* executor_; - IOCP* iocp_; - }; - - void OnConnectCompleted(std::shared_ptr state); - - // CancelConnect called from within the deadline timer. - // In this case, the connection_state->mu is already locked, and timer - // cancellation is not possible. - bool CancelConnectFromDeadlineTimer(ConnectionState* connection_state) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(ConnectionState::mu); - - // Completes the connection cancellation logic after checking handle validity - // and optionally cancelling deadline timers. - bool CancelConnectInternalStateLocked(ConnectionState* connection_state) - ABSL_EXCLUSIVE_LOCKS_REQUIRED(ConnectionState::mu); - - class TimerClosure; + struct Closure; EventEngine::TaskHandle RunAfterInternal(Duration when, absl::AnyInvocable cb); - grpc_core::Mutex task_mu_; - TaskHandleSet known_handles_ ABSL_GUARDED_BY(task_mu_); - grpc_core::Mutex connection_mu_ ABSL_ACQUIRED_AFTER(ConnectionState::mu); - grpc_core::CondVar connection_cv_; - ConnectionHandleSet known_connection_handles_ ABSL_GUARDED_BY(connection_mu_); + grpc_core::Mutex mu_; + TaskHandleSet known_handles_ ABSL_GUARDED_BY(mu_); std::atomic aba_token_{0}; std::shared_ptr executor_; IOCP iocp_; TimerManager timer_manager_; - IOCPWorkClosure iocp_worker_; }; } // namespace experimental diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index 814460b689c..29291b9d1d0 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -486,7 +486,6 @@ CORE_SOURCE_FILES = [ 'src/core/lib/event_engine/utils.cc', 'src/core/lib/event_engine/windows/iocp.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/experiments/config.cc', 'src/core/lib/experiments/experiments.cc', diff --git a/test/core/end2end/BUILD b/test/core/end2end/BUILD index 460791fab1a..d4babbf29f7 100644 --- a/test/core/end2end/BUILD +++ b/test/core/end2end/BUILD @@ -71,7 +71,6 @@ grpc_cc_library( "//:httpcli", "//:sockaddr_utils", "//src/core:channel_args", - "//src/core:channel_args_endpoint_config", "//src/core:channel_args_preconditioning", "//src/core:closure", "//src/core:error", diff --git a/test/core/event_engine/BUILD b/test/core/event_engine/BUILD index 80c3219099f..1e2e7d0fd8d 100644 --- a/test/core/event_engine/BUILD +++ b/test/core/event_engine/BUILD @@ -72,8 +72,8 @@ grpc_cc_test( uses_polling = False, deps = [ "//:gpr_platform", + "//:grpc", "//src/core:channel_args", - "//src/core:channel_args_endpoint_config", ], ) diff --git a/test/core/util/BUILD b/test/core/util/BUILD index dfd9a925b2c..5048965cae9 100644 --- a/test/core/util/BUILD +++ b/test/core/util/BUILD @@ -136,7 +136,6 @@ grpc_cc_library( "//:ref_counted_ptr", "//:tsi_ssl_credentials", "//:uri_parser", - "//src/core:channel_args_endpoint_config", "//src/core:channel_args_preconditioning", "//src/core:closure", "//src/core:error", @@ -177,7 +176,6 @@ grpc_cc_library( "//:orphanable", "//:ref_counted_ptr", "//:uri_parser", - "//src/core:channel_args_endpoint_config", "//src/core:channel_args_preconditioning", "//src/core:closure", "//src/core:error", diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index 7734c1880c1..941dba65016 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -2045,8 +2045,6 @@ src/core/lib/event_engine/windows/iocp.cc \ src/core/lib/event_engine/windows/iocp.h \ src/core/lib/event_engine/windows/win_socket.cc \ src/core/lib/event_engine/windows/win_socket.h \ -src/core/lib/event_engine/windows/windows_endpoint.cc \ -src/core/lib/event_engine/windows/windows_endpoint.h \ src/core/lib/event_engine/windows/windows_engine.cc \ src/core/lib/event_engine/windows/windows_engine.h \ src/core/lib/experiments/config.cc \ diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 7dcb54749a0..9d9b1c7eae8 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -1824,8 +1824,6 @@ src/core/lib/event_engine/windows/iocp.cc \ src/core/lib/event_engine/windows/iocp.h \ src/core/lib/event_engine/windows/win_socket.cc \ src/core/lib/event_engine/windows/win_socket.h \ -src/core/lib/event_engine/windows/windows_endpoint.cc \ -src/core/lib/event_engine/windows/windows_endpoint.h \ src/core/lib/event_engine/windows/windows_engine.cc \ src/core/lib/event_engine/windows/windows_engine.h \ src/core/lib/experiments/config.cc \