[EventEngine] Windows Listener iomgr shims (#32541)

Exercised with the `//test/core/end2end:h2_full_test@*` set of tests,
using the `event_engine_listener` experiment.

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",
     ],
 )
 

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

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',

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',

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 )

package.xml

@@ -1034,6 +1034,7 @@
     <file baseinstalldir="/" name="src/core/lib/event_engine/forkable.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/event_engine/handle_containers.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/event_engine/memory_allocator.cc" role="src" />
+    <file baseinstalldir="/" name="src/core/lib/event_engine/memory_allocator_factory.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/event_engine/poller.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/event_engine/posix.h" role="src" />
     <file baseinstalldir="/" name="src/core/lib/event_engine/posix_engine/ev_epoll1_linux.cc" role="src" />

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 = [

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 <grpc/support/port_platform.h>
+
+#include <memory>
+#include <utility>
+
+#include "absl/strings/string_view.h"
+
+#include <grpc/event_engine/memory_allocator.h>
+
+#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

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;

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<AsyncIOState>(this, std::move(socket))),
+      io_state_(std::make_shared<AsyncIOState>(this, std::move(socket),
-      engine_(engine) {
+                                               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<void(absl::Status)> 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<WinSocket> socket)
+                                            std::unique_ptr<WinSocket> socket,
-    : endpoint(endpoint), socket(std::move(socket)) {}
+                                            std::shared_ptr<EventEngine> engine)
+    : endpoint(endpoint),
+      socket(std::move(socket)),
+      engine(std::move(engine)) {}
 
 WindowsEndpoint::AsyncIOState::~AsyncIOState() {
   socket->Shutdown(DEBUG_LOCATION, "~AsyncIOState");

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<WinSocket> socket);
+    AsyncIOState(WindowsEndpoint* endpoint, std::unique_ptr<WinSocket> socket,
+                 std::shared_ptr<EventEngine> engine);
     ~AsyncIOState();
     WindowsEndpoint* const endpoint;
     std::unique_ptr<WinSocket> socket;
     HandleReadClosure handle_read_event;
     HandleWriteClosure handle_write_event;
+    std::shared_ptr<EventEngine> 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<AsyncIOState> io_state_;
-  std::shared_ptr<EventEngine> engine_;
 };
 
 }  // namespace experimental

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.

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",
+      "%s", absl::StrFormat("~WindowsEventEngineListener::%p", this).c_str());
-      absl::StrFormat("WindowsEventEngineListener::%p shutting down", this)
+  ShutdownListeners();
-          .c_str());
-  // Shut down each port listener before destroying this EventEngine::Listener
-  for (auto& port_listener : port_listeners_) {
-    port_listener.reset();
-  }
   on_shutdown_(absl::OkStatus());
 }
 
@@ -300,7 +296,7 @@ absl::StatusOr<int> 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<int> 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::SinglePortSocketListener*>
 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,

src/core/lib/event_engine/windows/windows_listener.h

@@ -44,6 +44,9 @@ class WindowsEventEngineListener : public EventEngine::Listener {
   ~WindowsEventEngineListener() override;
   absl::StatusOr<int> 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<void(absl::Status)> on_shutdown_;
   std::atomic<bool> started_{false};
-  grpc_core::Mutex socket_listeners_mu_;
+  grpc_core::Mutex port_listeners_mu_;
   std::list<std::unique_ptr<SinglePortSocketListener>> 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

src/core/lib/iomgr/iomgr_windows.cc

@@ -24,6 +24,7 @@
 
 #include <grpc/support/log.h>
 
+#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);

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<int64_t> 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<MemoryAllocatorFactoryWrapper>(s->memory_quota));
+      config,
+      std::make_unique<MemoryQuotaBasedMemoryAllocatorFactory>(
+          s->memory_quota));
   if (!listener.ok()) {
     delete s;
     *server = nullptr;

src/core/lib/iomgr/tcp_server_windows.cc

@@ -30,6 +30,8 @@
 #include "absl/strings/str_cat.h"
 
 #include <grpc/event_engine/endpoint_config.h>
+#include <grpc/event_engine/event_engine.h>
+#include <grpc/event_engine/memory_allocator.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/log_windows.h>
@@ -38,7 +40,15 @@
 #include <grpc/support/time.h>
 
 #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<WindowsEventEngine*>(
+      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<EventEngine::Endpoint> endpoint,
+                       MemoryAllocator memory_allocator) {
+    grpc_core::ApplicationCallbackExecCtx app_ctx;
+    grpc_core::ExecCtx exec_ctx;
+    grpc_tcp_server_acceptor* acceptor =
+        static_cast<grpc_tcp_server_acceptor*>(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<grpc_core::ResourceQuota> resource_quota;
+  {
+    void* tmp_quota = config.GetVoidPointer(GRPC_ARG_RESOURCE_QUOTA);
+    GPR_ASSERT(tmp_quota != nullptr);
+    resource_quota =
+        reinterpret_cast<grpc_core::ResourceQuota*>(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<MemoryQuotaBasedMemoryAllocatorFactory>(
+          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<grpc_pollset*>* /*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

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 \

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 \