diff --git a/BUILD b/BUILD index 7ada5282efb..7cdea65f6d6 100644 --- a/BUILD +++ b/BUILD @@ -1478,6 +1478,7 @@ grpc_cc_library( "//src/core:dual_ref_counted", "//src/core:error", "//src/core:event_engine_common", + "//src/core:event_engine_memory_allocator_factory", "//src/core:event_engine_shim", "//src/core:event_engine_tcp_socket_utils", "//src/core:event_engine_trace", @@ -1521,6 +1522,8 @@ grpc_cc_library( "//src/core:try_join", "//src/core:try_seq", "//src/core:useful", + "//src/core:windows_event_engine", + "//src/core:windows_event_engine_listener", ], ) diff --git a/build_autogenerated.yaml b/build_autogenerated.yaml index 577fbbac247..502fb5888ff 100644 --- a/build_autogenerated.yaml +++ b/build_autogenerated.yaml @@ -797,6 +797,7 @@ libs: - src/core/lib/event_engine/executor/executor.h - src/core/lib/event_engine/forkable.h - src/core/lib/event_engine/handle_containers.h + - src/core/lib/event_engine/memory_allocator_factory.h - src/core/lib/event_engine/poller.h - src/core/lib/event_engine/posix.h - src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h @@ -2140,6 +2141,7 @@ libs: - src/core/lib/event_engine/executor/executor.h - src/core/lib/event_engine/forkable.h - src/core/lib/event_engine/handle_containers.h + - src/core/lib/event_engine/memory_allocator_factory.h - src/core/lib/event_engine/poller.h - src/core/lib/event_engine/posix.h - src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h @@ -3604,6 +3606,7 @@ libs: - src/core/lib/event_engine/executor/executor.h - src/core/lib/event_engine/forkable.h - src/core/lib/event_engine/handle_containers.h + - src/core/lib/event_engine/memory_allocator_factory.h - src/core/lib/event_engine/poller.h - src/core/lib/event_engine/posix.h - src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h @@ -7414,6 +7417,7 @@ targets: - src/core/lib/event_engine/executor/executor.h - src/core/lib/event_engine/forkable.h - src/core/lib/event_engine/handle_containers.h + - src/core/lib/event_engine/memory_allocator_factory.h - src/core/lib/event_engine/poller.h - src/core/lib/event_engine/posix.h - src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h diff --git a/gRPC-C++.podspec b/gRPC-C++.podspec index 5656c0bcc96..d09f057781b 100644 --- a/gRPC-C++.podspec +++ b/gRPC-C++.podspec @@ -741,6 +741,7 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/executor/executor.h', 'src/core/lib/event_engine/forkable.h', 'src/core/lib/event_engine/handle_containers.h', + 'src/core/lib/event_engine/memory_allocator_factory.h', 'src/core/lib/event_engine/poller.h', 'src/core/lib/event_engine/posix.h', 'src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h', @@ -1676,6 +1677,7 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/executor/executor.h', 'src/core/lib/event_engine/forkable.h', 'src/core/lib/event_engine/handle_containers.h', + 'src/core/lib/event_engine/memory_allocator_factory.h', 'src/core/lib/event_engine/poller.h', 'src/core/lib/event_engine/posix.h', 'src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h', diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 5f1b1963a26..4ec4870a7e0 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -1143,6 +1143,7 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/forkable.h', 'src/core/lib/event_engine/handle_containers.h', 'src/core/lib/event_engine/memory_allocator.cc', + 'src/core/lib/event_engine/memory_allocator_factory.h', 'src/core/lib/event_engine/poller.h', 'src/core/lib/event_engine/posix.h', 'src/core/lib/event_engine/posix_engine/ev_epoll1_linux.cc', @@ -2365,6 +2366,7 @@ Pod::Spec.new do |s| 'src/core/lib/event_engine/executor/executor.h', 'src/core/lib/event_engine/forkable.h', 'src/core/lib/event_engine/handle_containers.h', + 'src/core/lib/event_engine/memory_allocator_factory.h', 'src/core/lib/event_engine/poller.h', 'src/core/lib/event_engine/posix.h', 'src/core/lib/event_engine/posix_engine/ev_epoll1_linux.h', diff --git a/grpc.gemspec b/grpc.gemspec index fdbd9757882..9426def7f7c 100644 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -1052,6 +1052,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/event_engine/forkable.h ) s.files += %w( src/core/lib/event_engine/handle_containers.h ) s.files += %w( src/core/lib/event_engine/memory_allocator.cc ) + s.files += %w( src/core/lib/event_engine/memory_allocator_factory.h ) s.files += %w( src/core/lib/event_engine/poller.h ) s.files += %w( src/core/lib/event_engine/posix.h ) s.files += %w( src/core/lib/event_engine/posix_engine/ev_epoll1_linux.cc ) diff --git a/package.xml b/package.xml index bf49b7242cc..abbbad4d4d1 100644 --- a/package.xml +++ b/package.xml @@ -1034,6 +1034,7 @@ + diff --git a/src/core/BUILD b/src/core/BUILD index dd2c2f50f17..88ddc0d17cf 100644 --- a/src/core/BUILD +++ b/src/core/BUILD @@ -1041,6 +1041,20 @@ grpc_cc_library( ], ) +grpc_cc_library( + name = "event_engine_memory_allocator_factory", + hdrs = [ + "lib/event_engine/memory_allocator_factory.h", + ], + external_deps = ["absl/strings"], + language = "c++", + deps = [ + "event_engine_memory_allocator", + "memory_quota", + "//:gpr_platform", + ], +) + grpc_cc_library( name = "memory_quota", srcs = [ diff --git a/src/core/lib/event_engine/memory_allocator_factory.h b/src/core/lib/event_engine/memory_allocator_factory.h new file mode 100644 index 00000000000..0af1b713970 --- /dev/null +++ b/src/core/lib/event_engine/memory_allocator_factory.h @@ -0,0 +1,47 @@ +// Copyright 2023 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_SRC_CORE_LIB_EVENT_ENGINE_MEMORY_ALLOCATOR_FACTORY_H +#define GRPC_SRC_CORE_LIB_EVENT_ENGINE_MEMORY_ALLOCATOR_FACTORY_H +#include + +#include +#include + +#include "absl/strings/string_view.h" + +#include + +#include "src/core/lib/resource_quota/memory_quota.h" + +namespace grpc_event_engine { +namespace experimental { + +class MemoryQuotaBasedMemoryAllocatorFactory : public MemoryAllocatorFactory { + public: + explicit MemoryQuotaBasedMemoryAllocatorFactory( + grpc_core::MemoryQuotaRefPtr memory_quota) + : memory_quota_(std::move(memory_quota)) {} + + MemoryAllocator CreateMemoryAllocator(absl::string_view name) override { + return memory_quota_->CreateMemoryAllocator(name); + } + + private: + grpc_core::MemoryQuotaRefPtr memory_quota_; +}; + +} // namespace experimental +} // namespace grpc_event_engine + +#endif // GRPC_SRC_CORE_LIB_EVENT_ENGINE_MEMORY_ALLOCATOR_FACTORY_H diff --git a/src/core/lib/event_engine/windows/win_socket.cc b/src/core/lib/event_engine/windows/win_socket.cc index 82a22e6cdd7..790b38b4803 100644 --- a/src/core/lib/event_engine/windows/win_socket.cc +++ b/src/core/lib/event_engine/windows/win_socket.cc @@ -58,7 +58,6 @@ void WinSocket::Shutdown() { this); return; } - GRPC_EVENT_ENGINE_ENDPOINT_TRACE("WinSocket::%p shutting down now.", this); // Grab the function pointer for DisconnectEx for that specific socket. // It may change depending on the interface. GUID guid = WSAID_DISCONNECTEX; diff --git a/src/core/lib/event_engine/windows/windows_endpoint.cc b/src/core/lib/event_engine/windows/windows_endpoint.cc index 15b2dbd3302..bd2d5c5498d 100644 --- a/src/core/lib/event_engine/windows/windows_endpoint.cc +++ b/src/core/lib/event_engine/windows/windows_endpoint.cc @@ -55,8 +55,8 @@ WindowsEndpoint::WindowsEndpoint( : peer_address_(peer_address), allocator_(std::move(allocator)), executor_(executor), - io_state_(std::make_shared(this, std::move(socket))), - engine_(engine) { + io_state_(std::make_shared(this, std::move(socket), + std::move(engine))) { char addr[EventEngine::ResolvedAddress::MAX_SIZE_BYTES]; int addr_len = sizeof(addr); if (getsockname(io_state_->socket->raw_socket(), @@ -73,7 +73,7 @@ WindowsEndpoint::WindowsEndpoint( WindowsEndpoint::~WindowsEndpoint() { io_state_->socket->Shutdown(DEBUG_LOCATION, "~WindowsEndpoint"); - GRPC_EVENT_ENGINE_ENDPOINT_TRACE("WindowsEndpoint::%p destroyed", this); + GRPC_EVENT_ENGINE_ENDPOINT_TRACE("~WindowsEndpoint::%p", this); } absl::Status WindowsEndpoint::DoTcpRead(SliceBuffer* buffer) { @@ -125,7 +125,6 @@ absl::Status WindowsEndpoint::DoTcpRead(SliceBuffer* buffer) { bool WindowsEndpoint::Read(absl::AnyInvocable on_read, SliceBuffer* buffer, const ReadArgs* /* args */) { - GRPC_EVENT_ENGINE_ENDPOINT_TRACE("WindowsEndpoint::%p reading", this); if (io_state_->socket->IsShutdown()) { executor_->Run([on_read = std::move(on_read)]() mutable { on_read(absl::UnavailableError("Socket is shutting down.")); @@ -247,11 +246,13 @@ void AbortOnEvent(absl::Status) { } // namespace void WindowsEndpoint::HandleReadClosure::Reset() { + io_state_.reset(); cb_ = &AbortOnEvent; buffer_ = nullptr; } void WindowsEndpoint::HandleWriteClosure::Reset() { + io_state_.reset(); cb_ = &AbortOnEvent; buffer_ = nullptr; } @@ -310,6 +311,7 @@ void WindowsEndpoint::HandleReadClosure::Run() { io_state_ = std::move(io_state); status = io_state_->endpoint->DoTcpRead(buffer_); if (!status.ok()) { + io_state_.reset(); ExecuteCallbackAndReset(status); } } @@ -360,8 +362,11 @@ void WindowsEndpoint::HandleWriteClosure::Run() { // ---- AsyncIOState ---- WindowsEndpoint::AsyncIOState::AsyncIOState(WindowsEndpoint* endpoint, - std::unique_ptr socket) - : endpoint(endpoint), socket(std::move(socket)) {} + std::unique_ptr socket, + std::shared_ptr engine) + : endpoint(endpoint), + socket(std::move(socket)), + engine(std::move(engine)) {} WindowsEndpoint::AsyncIOState::~AsyncIOState() { socket->Shutdown(DEBUG_LOCATION, "~AsyncIOState"); diff --git a/src/core/lib/event_engine/windows/windows_endpoint.h b/src/core/lib/event_engine/windows/windows_endpoint.h index bf8171c486b..723c874932e 100644 --- a/src/core/lib/event_engine/windows/windows_endpoint.h +++ b/src/core/lib/event_engine/windows/windows_endpoint.h @@ -88,12 +88,14 @@ class WindowsEndpoint : public EventEngine::Endpoint { // Endpoint, and be destroyed asynchronously when all pending overlapped // events are complete. struct AsyncIOState { - AsyncIOState(WindowsEndpoint* endpoint, std::unique_ptr socket); + AsyncIOState(WindowsEndpoint* endpoint, std::unique_ptr socket, + std::shared_ptr engine); ~AsyncIOState(); WindowsEndpoint* const endpoint; std::unique_ptr socket; HandleReadClosure handle_read_event; HandleWriteClosure handle_write_event; + std::shared_ptr engine; }; // Perform the low-level calls and execute the HandleReadClosure @@ -107,7 +109,6 @@ class WindowsEndpoint : public EventEngine::Endpoint { MemoryAllocator allocator_; Executor* executor_; std::shared_ptr io_state_; - std::shared_ptr engine_; }; } // namespace experimental diff --git a/src/core/lib/event_engine/windows/windows_engine.h b/src/core/lib/event_engine/windows/windows_engine.h index 4e81cca6d15..d9bd119aa14 100644 --- a/src/core/lib/event_engine/windows/windows_engine.h +++ b/src/core/lib/event_engine/windows/windows_engine.h @@ -93,6 +93,7 @@ class WindowsEventEngine : public EventEngine, // WindowsEventEngine type are effectively friends. // Not intended for external use. Executor* executor() { return executor_.get(); } + IOCP* poller() { return &iocp_; } private: // State of an active connection. diff --git a/src/core/lib/event_engine/windows/windows_listener.cc b/src/core/lib/event_engine/windows/windows_listener.cc index e176412123a..3c1b92386f2 100644 --- a/src/core/lib/event_engine/windows/windows_listener.cc +++ b/src/core/lib/event_engine/windows/windows_listener.cc @@ -69,6 +69,7 @@ void WindowsEventEngineListener::SinglePortSocketListener:: WindowsEventEngineListener::SinglePortSocketListener:: ~SinglePortSocketListener() { + grpc_core::MutexLock lock(&io_state_->mu); io_state_->listener_socket->Shutdown(DEBUG_LOCATION, "~SinglePortSocketListener"); GRPC_EVENT_ENGINE_TRACE("~SinglePortSocketListener::%p", this); @@ -276,13 +277,8 @@ WindowsEventEngineListener::WindowsEventEngineListener( WindowsEventEngineListener::~WindowsEventEngineListener() { GRPC_EVENT_ENGINE_TRACE( - "%s", - absl::StrFormat("WindowsEventEngineListener::%p shutting down", this) - .c_str()); - // Shut down each port listener before destroying this EventEngine::Listener - for (auto& port_listener : port_listeners_) { - port_listener.reset(); - } + "%s", absl::StrFormat("~WindowsEventEngineListener::%p", this).c_str()); + ShutdownListeners(); on_shutdown_(absl::OkStatus()); } @@ -300,7 +296,7 @@ absl::StatusOr WindowsEventEngineListener::Bind( // Check if this is a wildcard port, and if so, try to keep the port the same // as some previously created listener. if (out_port == 0) { - grpc_core::MutexLock lock(&socket_listeners_mu_); + grpc_core::MutexLock lock(&port_listeners_mu_); for (const auto& port_listener : port_listeners_) { tmp_addr = port_listener->listener_sockname(); out_port = ResolvedAddressGetPort(tmp_addr); @@ -332,13 +328,22 @@ absl::StatusOr WindowsEventEngineListener::Bind( absl::Status WindowsEventEngineListener::Start() { GPR_ASSERT(!started_.exchange(true)); - grpc_core::MutexLock lock(&socket_listeners_mu_); + grpc_core::MutexLock lock(&port_listeners_mu_); for (auto& port_listener : port_listeners_) { GRPC_RETURN_IF_ERROR(port_listener->Start()); } return absl::OkStatus(); } +void WindowsEventEngineListener::ShutdownListeners() { + grpc_core::MutexLock lock(&port_listeners_mu_); + if (std::exchange(listeners_shutdown_, true)) return; + // Shut down each port listener before destroying this EventEngine::Listener + for (auto& port_listener : port_listeners_) { + port_listener.reset(); + } +} + absl::StatusOr WindowsEventEngineListener::AddSinglePortSocketListener( SOCKET sock, EventEngine::ResolvedAddress addr) { @@ -346,7 +351,7 @@ WindowsEventEngineListener::AddSinglePortSocketListener( SinglePortSocketListener::Create(this, sock, addr); GRPC_RETURN_IF_ERROR(single_port_listener.status()); auto* single_port_listener_ptr = single_port_listener->get(); - grpc_core::MutexLock lock(&socket_listeners_mu_); + grpc_core::MutexLock lock(&port_listeners_mu_); port_listeners_.emplace_back(std::move(*single_port_listener)); if (started_.load()) { gpr_log(GPR_ERROR, diff --git a/src/core/lib/event_engine/windows/windows_listener.h b/src/core/lib/event_engine/windows/windows_listener.h index 9047180cbdd..4183de1646f 100644 --- a/src/core/lib/event_engine/windows/windows_listener.h +++ b/src/core/lib/event_engine/windows/windows_listener.h @@ -44,6 +44,9 @@ class WindowsEventEngineListener : public EventEngine::Listener { ~WindowsEventEngineListener() override; absl::StatusOr Bind(const EventEngine::ResolvedAddress& addr) override; absl::Status Start() override; + // TODO(hork): this may only be needed for the iomgr shim, to accommodate + // calls to grpc_tcp_server_shutdown_listeners + void ShutdownListeners(); private: /// Responsible for listening on a single port. @@ -138,9 +141,10 @@ class WindowsEventEngineListener : public EventEngine::Listener { AcceptCallback accept_cb_; absl::AnyInvocable on_shutdown_; std::atomic started_{false}; - grpc_core::Mutex socket_listeners_mu_; + grpc_core::Mutex port_listeners_mu_; std::list> port_listeners_ - ABSL_GUARDED_BY(socket_listeners_mu_); + ABSL_GUARDED_BY(port_listeners_mu_); + bool listeners_shutdown_ ABSL_GUARDED_BY(port_listeners_mu_) = false; }; } // namespace experimental diff --git a/src/core/lib/iomgr/iomgr_windows.cc b/src/core/lib/iomgr/iomgr_windows.cc index 11441f203a4..29bb9ea4711 100644 --- a/src/core/lib/iomgr/iomgr_windows.cc +++ b/src/core/lib/iomgr/iomgr_windows.cc @@ -24,6 +24,7 @@ #include +#include "src/core/lib/experiments/experiments.h" #include "src/core/lib/gprpp/crash.h" #include "src/core/lib/iomgr/iocp_windows.h" #include "src/core/lib/iomgr/iomgr.h" @@ -37,6 +38,7 @@ #include "src/core/lib/iomgr/timer.h" extern grpc_tcp_server_vtable grpc_windows_tcp_server_vtable; +extern grpc_tcp_server_vtable grpc_windows_event_engine_tcp_server_vtable; extern grpc_tcp_client_vtable grpc_windows_tcp_client_vtable; extern grpc_timer_vtable grpc_generic_timer_vtable; extern grpc_pollset_vtable grpc_windows_pollset_vtable; @@ -95,7 +97,11 @@ static grpc_iomgr_platform_vtable vtable = { void grpc_set_default_iomgr_platform() { grpc_set_tcp_client_impl(&grpc_windows_tcp_client_vtable); - grpc_set_tcp_server_impl(&grpc_windows_tcp_server_vtable); + if (grpc_core::IsEventEngineListenerEnabled()) { + grpc_set_tcp_server_impl(&grpc_windows_event_engine_tcp_server_vtable); + } else { + grpc_set_tcp_server_impl(&grpc_windows_tcp_server_vtable); + } grpc_set_timer_impl(&grpc_generic_timer_vtable); grpc_set_pollset_vtable(&grpc_windows_pollset_vtable); grpc_set_pollset_set_vtable(&grpc_windows_pollset_set_vtable); diff --git a/src/core/lib/iomgr/tcp_server_posix.cc b/src/core/lib/iomgr/tcp_server_posix.cc index a1db16d916d..dbe7bec3383 100644 --- a/src/core/lib/iomgr/tcp_server_posix.cc +++ b/src/core/lib/iomgr/tcp_server_posix.cc @@ -52,6 +52,7 @@ #include "src/core/lib/address_utils/sockaddr_utils.h" #include "src/core/lib/event_engine/default_event_engine.h" +#include "src/core/lib/event_engine/memory_allocator_factory.h" #include "src/core/lib/event_engine/resolved_address_internal.h" #include "src/core/lib/event_engine/shim.h" #include "src/core/lib/event_engine/tcp_socket_utils.h" @@ -78,24 +79,10 @@ static std::atomic num_dropped_connections{0}; using ::grpc_event_engine::experimental::EndpointConfig; using ::grpc_event_engine::experimental::EventEngine; using ::grpc_event_engine::experimental::MemoryAllocator; +using ::grpc_event_engine::experimental::MemoryQuotaBasedMemoryAllocatorFactory; using ::grpc_event_engine::experimental::PosixEventEngineWithFdSupport; using ::grpc_event_engine::experimental::SliceBuffer; -class MemoryAllocatorFactoryWrapper - : public grpc_event_engine::experimental::MemoryAllocatorFactory { - public: - explicit MemoryAllocatorFactoryWrapper( - grpc_core::MemoryQuotaRefPtr memory_quota) - : memory_quota_(std::move(memory_quota)) {} - - MemoryAllocator CreateMemoryAllocator(absl::string_view name) override { - return memory_quota_->CreateMemoryAllocator(name); - } - - private: - grpc_core::MemoryQuotaRefPtr memory_quota_; -}; - static grpc_error_handle CreateEventEngineListener( grpc_tcp_server* s, grpc_closure* shutdown_complete, const EndpointConfig& config, grpc_tcp_server** server) { @@ -189,7 +176,9 @@ static grpc_error_handle CreateEventEngineListener( delete s->fd_handler; delete s; }, - config, std::make_unique(s->memory_quota)); + config, + std::make_unique( + s->memory_quota)); if (!listener.ok()) { delete s; *server = nullptr; diff --git a/src/core/lib/iomgr/tcp_server_windows.cc b/src/core/lib/iomgr/tcp_server_windows.cc index d0e5f6d6fc3..a2508cf4d2e 100644 --- a/src/core/lib/iomgr/tcp_server_windows.cc +++ b/src/core/lib/iomgr/tcp_server_windows.cc @@ -30,6 +30,8 @@ #include "absl/strings/str_cat.h" #include +#include +#include #include #include #include @@ -38,7 +40,15 @@ #include #include "src/core/lib/address_utils/sockaddr_utils.h" +#include "src/core/lib/event_engine/memory_allocator_factory.h" +#include "src/core/lib/event_engine/resolved_address_internal.h" +#include "src/core/lib/event_engine/tcp_socket_utils.h" +#include "src/core/lib/event_engine/windows/windows_engine.h" +#include "src/core/lib/event_engine/windows/windows_listener.h" #include "src/core/lib/gprpp/crash.h" +#include "src/core/lib/iomgr/closure.h" +#include "src/core/lib/iomgr/event_engine_shims/closure.h" +#include "src/core/lib/iomgr/event_engine_shims/endpoint.h" #include "src/core/lib/iomgr/iocp_windows.h" #include "src/core/lib/iomgr/pollset_windows.h" #include "src/core/lib/iomgr/resolve_address.h" @@ -47,11 +57,23 @@ #include "src/core/lib/iomgr/tcp_server.h" #include "src/core/lib/iomgr/tcp_windows.h" #include "src/core/lib/resource_quota/api.h" +#include "src/core/lib/resource_quota/resource_quota.h" #include "src/core/lib/slice/slice_internal.h" #define MIN_SAFE_ACCEPT_QUEUE_SIZE 100 +namespace { +using ::grpc_event_engine::experimental::CreateResolvedAddress; using ::grpc_event_engine::experimental::EndpointConfig; +using ::grpc_event_engine::experimental::EventEngine; +using ::grpc_event_engine::experimental::grpc_event_engine_endpoint_create; +using ::grpc_event_engine::experimental::MemoryAllocator; +using ::grpc_event_engine::experimental::MemoryQuotaBasedMemoryAllocatorFactory; +using ::grpc_event_engine::experimental::ResolvedAddressSetPort; +using ::grpc_event_engine::experimental::RunEventEngineClosure; +using ::grpc_event_engine::experimental::WindowsEventEngine; +using ::grpc_event_engine::experimental::WindowsEventEngineListener; +} // namespace // one listening port typedef struct grpc_tcp_listener grpc_tcp_listener; @@ -98,6 +120,9 @@ struct grpc_tcp_server { // shutdown callback grpc_closure* shutdown_complete; + + // used for the EventEngine shim + WindowsEventEngineListener* ee_listener; }; // Public function. Allocates the proper data structures to hold a @@ -562,4 +587,140 @@ grpc_tcp_server_vtable grpc_windows_tcp_server_vtable = { tcp_server_ref, tcp_server_shutdown_starting_add, tcp_server_unref, tcp_server_shutdown_listeners, tcp_pre_allocated_fd, tcp_set_pre_allocated_fd}; + +// ---- EventEngine shim ------------------------------------------------------ + +namespace { + +static grpc_error_handle event_engine_create(grpc_closure* shutdown_complete, + const EndpointConfig& config, + grpc_tcp_server_cb on_accept_cb, + void* on_accept_cb_arg, + grpc_tcp_server** server) { + // On Windows, the event_engine_listener experiment only supports the + // default engine + WindowsEventEngine* engine_ptr = reinterpret_cast( + config.GetVoidPointer(GRPC_INTERNAL_ARG_EVENT_ENGINE)); + grpc_tcp_server* s = (grpc_tcp_server*)gpr_malloc(sizeof(grpc_tcp_server)); + GPR_ASSERT(on_accept_cb != nullptr); + auto accept_cb = [s, on_accept_cb, on_accept_cb_arg]( + std::unique_ptr endpoint, + MemoryAllocator memory_allocator) { + grpc_core::ApplicationCallbackExecCtx app_ctx; + grpc_core::ExecCtx exec_ctx; + grpc_tcp_server_acceptor* acceptor = + static_cast(gpr_malloc(sizeof(*acceptor))); + acceptor->from_server = s; + acceptor->port_index = -1; + acceptor->fd_index = -1; + acceptor->external_connection = false; + on_accept_cb(on_accept_cb_arg, + grpc_event_engine_endpoint_create(std::move(endpoint)), + nullptr, acceptor); + }; + auto on_shutdown = [shutdown_complete](absl::Status status) { + RunEventEngineClosure(shutdown_complete, status); + }; + grpc_core::RefCountedPtr resource_quota; + { + void* tmp_quota = config.GetVoidPointer(GRPC_ARG_RESOURCE_QUOTA); + GPR_ASSERT(tmp_quota != nullptr); + resource_quota = + reinterpret_cast(tmp_quota)->Ref(); + } + gpr_ref_init(&s->refs, 1); + gpr_mu_init(&s->mu); + s->ee_listener = new WindowsEventEngineListener( + engine_ptr->poller(), std::move(accept_cb), std::move(on_shutdown), + std::make_unique( + resource_quota->memory_quota()), + engine_ptr->shared_from_this(), engine_ptr->executor(), config); + s->active_ports = -1; + s->on_accept_cb = [](void* /* arg */, grpc_endpoint* /* ep */, + grpc_pollset* /* accepting_pollset */, + grpc_tcp_server_acceptor* /* acceptor */) { + grpc_core::Crash("iomgr on_accept_cb callback should be unused"); + }; + s->on_accept_cb_arg = nullptr; + s->head = nullptr; + s->tail = nullptr; + s->shutdown_starting.head = nullptr; + s->shutdown_starting.tail = nullptr; + s->shutdown_complete = grpc_core::NewClosure([](absl::Status) { + grpc_core::Crash("iomgr shutdown_complete callback should be unused"); + }); + *server = s; + return absl::OkStatus(); +} + +static void event_engine_start(grpc_tcp_server* s, + const std::vector* /*pollsets*/) { + GPR_ASSERT(s->ee_listener->Start().ok()); +} + +static grpc_error_handle event_engine_add_port( + grpc_tcp_server* s, const grpc_resolved_address* addr, int* port) { + GPR_ASSERT(addr != nullptr); + GPR_ASSERT(port != nullptr); + auto ee_addr = CreateResolvedAddress(*addr); + auto out_port = s->ee_listener->Bind(ee_addr); + *port = out_port.ok() ? *out_port : -1; + return out_port.status(); +} + +static grpc_core::TcpServerFdHandler* event_engine_create_fd_handler( + grpc_tcp_server* /* s */) { + return nullptr; +} + +static unsigned event_engine_port_fd_count(grpc_tcp_server* /* s */, + unsigned /* port_index */) { + return 0; +} + +static int event_engine_port_fd(grpc_tcp_server* /* s */, + unsigned /* port_index */, + unsigned /* fd_index */) { + return -1; +} + +static grpc_tcp_server* event_engine_ref(grpc_tcp_server* s) { + gpr_ref_non_zero(&s->refs); + return s; +} + +static void event_engine_shutdown_listeners(grpc_tcp_server* s) { + s->ee_listener->ShutdownListeners(); +} + +static void event_engine_unref(grpc_tcp_server* s) { + if (gpr_unref(&s->refs)) { + event_engine_shutdown_listeners(s); + gpr_mu_lock(&s->mu); + grpc_core::ExecCtx::RunList(DEBUG_LOCATION, &s->shutdown_starting); + gpr_mu_unlock(&s->mu); + gpr_mu_destroy(&s->mu); + delete s->ee_listener; + gpr_free(s); + } +} + +static void event_engine_shutdown_starting_add( + grpc_tcp_server* s, grpc_closure* shutdown_starting) { + gpr_mu_lock(&s->mu); + grpc_closure_list_append(&s->shutdown_starting, shutdown_starting, + absl::OkStatus()); + gpr_mu_unlock(&s->mu); +} + +} // namespace + +grpc_tcp_server_vtable grpc_windows_event_engine_tcp_server_vtable = { + event_engine_create, event_engine_start, + event_engine_add_port, event_engine_create_fd_handler, + event_engine_port_fd_count, event_engine_port_fd, + event_engine_ref, event_engine_shutdown_starting_add, + event_engine_unref, event_engine_shutdown_listeners, + tcp_pre_allocated_fd, tcp_set_pre_allocated_fd}; + #endif // GRPC_WINSOCK_SOCKET diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index 4a41380df9c..5d5d4cef2fa 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -2047,6 +2047,7 @@ src/core/lib/event_engine/forkable.cc \ src/core/lib/event_engine/forkable.h \ src/core/lib/event_engine/handle_containers.h \ src/core/lib/event_engine/memory_allocator.cc \ +src/core/lib/event_engine/memory_allocator_factory.h \ src/core/lib/event_engine/poller.h \ src/core/lib/event_engine/posix.h \ src/core/lib/event_engine/posix_engine/ev_epoll1_linux.cc \ diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 22bc220dc85..66ee1646416 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -1825,6 +1825,7 @@ src/core/lib/event_engine/forkable.cc \ src/core/lib/event_engine/forkable.h \ src/core/lib/event_engine/handle_containers.h \ src/core/lib/event_engine/memory_allocator.cc \ +src/core/lib/event_engine/memory_allocator_factory.h \ src/core/lib/event_engine/poller.h \ src/core/lib/event_engine/posix.h \ src/core/lib/event_engine/posix_engine/ev_epoll1_linux.cc \