Enable c-ares queries on Windows

20 files changed, 771 insertions, 71 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 855b921ada..f242ee92bb 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -662,12 +662,8 @@ add_dependencies(buildtests_cxx transport_security_common_api_test)
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 add_dependencies(buildtests_cxx writes_per_rpc_test)
 endif()
-if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 add_dependencies(buildtests_cxx resolver_component_test_unsecure)
-endif()
-if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 add_dependencies(buildtests_cxx resolver_component_test)
-endif()
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 add_dependencies(buildtests_cxx resolver_component_tests_runner_invoker_unsecure)
 endif()
@@ -676,9 +672,7 @@ add_dependencies(buildtests_cxx resolver_component_tests_runner_invoker)
 endif()
 add_dependencies(buildtests_cxx address_sorting_test_unsecure)
 add_dependencies(buildtests_cxx address_sorting_test)
-if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 add_dependencies(buildtests_cxx cancel_ares_query_test)
-endif()
 
 add_custom_target(buildtests
   DEPENDS buildtests_c buildtests_cxx)
@@ -16213,7 +16207,6 @@ target_link_libraries(inproc_nosec_test
 
 endif (gRPC_BUILD_TESTS)
 if (gRPC_BUILD_TESTS)
-if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 
 add_executable(resolver_component_test_unsecure
   test/cpp/naming/resolver_component_test.cc
@@ -16253,10 +16246,8 @@ target_link_libraries(resolver_component_test_unsecure
   ${_gRPC_GFLAGS_LIBRARIES}
 )
 
-endif()
 endif (gRPC_BUILD_TESTS)
 if (gRPC_BUILD_TESTS)
-if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 
 add_executable(resolver_component_test
   test/cpp/naming/resolver_component_test.cc
@@ -16296,7 +16287,6 @@ target_link_libraries(resolver_component_test
   ${_gRPC_GFLAGS_LIBRARIES}
 )
 
-endif()
 endif (gRPC_BUILD_TESTS)
 if (gRPC_BUILD_TESTS)
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
@@ -16467,7 +16457,6 @@ target_link_libraries(address_sorting_test
 
 endif (gRPC_BUILD_TESTS)
 if (gRPC_BUILD_TESTS)
-if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 
 add_executable(cancel_ares_query_test
   test/cpp/naming/cancel_ares_query_test.cc
@@ -16507,7 +16496,6 @@ target_link_libraries(cancel_ares_query_test
   ${_gRPC_GFLAGS_LIBRARIES}
 )
 
-endif()
 endif (gRPC_BUILD_TESTS)
 if (gRPC_BUILD_TESTS)
 
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc
index 0068d0d5f4..fdbd07ebf5 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc
@@ -74,6 +74,8 @@ struct grpc_ares_ev_driver {
   bool shutting_down;
   /** request object that's using this ev driver */
   grpc_ares_request* request;
+  /** Owned by the ev_driver. Creates new GrpcPolledFd's */
+  grpc_core::UniquePtr<grpc_core::GrpcPolledFdFactory> polled_fd_factory;
 };
 
 static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver);
@@ -93,7 +95,7 @@ static void grpc_ares_ev_driver_unref(grpc_ares_ev_driver* ev_driver) {
     GRPC_COMBINER_UNREF(ev_driver->combiner, "free ares event driver");
     ares_destroy(ev_driver->channel);
     grpc_ares_complete_request_locked(ev_driver->request);
-    gpr_free(ev_driver);
+    grpc_core::Delete(ev_driver);
   }
 }
 
@@ -118,13 +120,11 @@ grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver,
                                               grpc_pollset_set* pollset_set,
                                               grpc_combiner* combiner,
                                               grpc_ares_request* request) {
-  *ev_driver = static_cast<grpc_ares_ev_driver*>(
-      gpr_malloc(sizeof(grpc_ares_ev_driver)));
+  *ev_driver = grpc_core::New<grpc_ares_ev_driver>();
   ares_options opts;
   memset(&opts, 0, sizeof(opts));
   opts.flags |= ARES_FLAG_STAYOPEN;
   int status = ares_init_options(&(*ev_driver)->channel, &opts, ARES_OPT_FLAGS);
-  grpc_core::ConfigureAresChannelLocked(&(*ev_driver)->channel);
   gpr_log(GPR_DEBUG, "grpc_ares_ev_driver_create_locked");
   if (status != ARES_SUCCESS) {
     char* err_msg;
@@ -142,6 +142,10 @@ grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver,
   (*ev_driver)->working = false;
   (*ev_driver)->shutting_down = false;
   (*ev_driver)->request = request;
+  (*ev_driver)->polled_fd_factory =
+      grpc_core::NewGrpcPolledFdFactory((*ev_driver)->combiner);
+  (*ev_driver)
+      ->polled_fd_factory->ConfigureAresChannelLocked((*ev_driver)->channel);
   return GRPC_ERROR_NONE;
 }
 
@@ -245,8 +249,9 @@ static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver) {
         // Create a new fd_node if sock[i] is not in the fd_node list.
         if (fdn == nullptr) {
           fdn = static_cast<fd_node*>(gpr_malloc(sizeof(fd_node)));
-          fdn->grpc_polled_fd = grpc_core::NewGrpcPolledFdLocked(
-              socks[i], ev_driver->pollset_set);
+          fdn->grpc_polled_fd =
+              ev_driver->polled_fd_factory->NewGrpcPolledFdLocked(
+                  socks[i], ev_driver->pollset_set, ev_driver->combiner);
           gpr_log(GPR_DEBUG, "new fd: %s", fdn->grpc_polled_fd->GetName());
           fdn->ev_driver = ev_driver;
           fdn->readable_registered = false;
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h
index 2c9db71011..671c537fe7 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h
@@ -81,10 +81,24 @@ class GrpcPolledFd {
   GRPC_ABSTRACT_BASE_CLASS
 };
 
-/* Creates a new wrapped fd for the current platform */
-GrpcPolledFd* NewGrpcPolledFdLocked(ares_socket_t as,
-                                    grpc_pollset_set* driver_pollset_set);
-void ConfigureAresChannelLocked(ares_channel* channel);
+/* A GrpcPolledFdFactory is 1-to-1 with and owned by the
+ * ares event driver. It knows how to create GrpcPolledFd's
+ * for the current platform, and the ares driver uses it for all of
+ * its fd's. */
+class GrpcPolledFdFactory {
+ public:
+  virtual ~GrpcPolledFdFactory() {}
+  /* Creates a new wrapped fd for the current platform */
+  virtual GrpcPolledFd* NewGrpcPolledFdLocked(
+      ares_socket_t as, grpc_pollset_set* driver_pollset_set,
+      grpc_combiner* combiner) GRPC_ABSTRACT;
+  /* Optionally configures the ares channel after creation */
+  virtual void ConfigureAresChannelLocked(ares_channel channel) GRPC_ABSTRACT;
+
+  GRPC_ABSTRACT_BASE_CLASS
+};
+
+UniquePtr<GrpcPolledFdFactory> NewGrpcPolledFdFactory(grpc_combiner* combiner);
 
 }  // namespace grpc_core
 
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc
index fffe9eda8e..aa58e1aaf5 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc
@@ -86,12 +86,20 @@ class GrpcPolledFdPosix : public GrpcPolledFd {
   grpc_pollset_set* driver_pollset_set_;
 };
 
-GrpcPolledFd* NewGrpcPolledFdLocked(ares_socket_t as,
-                                    grpc_pollset_set* driver_pollset_set) {
-  return grpc_core::New<GrpcPolledFdPosix>(as, driver_pollset_set);
-}
+class GrpcPolledFdFactoryPosix : public GrpcPolledFdFactory {
+ public:
+  GrpcPolledFd* NewGrpcPolledFdLocked(ares_socket_t as,
+                                      grpc_pollset_set* driver_pollset_set,
+                                      grpc_combiner* combiner) override {
+    return New<GrpcPolledFdPosix>(as, driver_pollset_set);
+  }
 
-void ConfigureAresChannelLocked(ares_channel* channel) {}
+  void ConfigureAresChannelLocked(ares_channel channel) override {}
+};
+
+UniquePtr<GrpcPolledFdFactory> NewGrpcPolledFdFactory(grpc_combiner* combiner) {
+  return UniquePtr<GrpcPolledFdFactory>(New<GrpcPolledFdFactoryPosix>());
+}
 
 }  // namespace grpc_core
 
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_windows.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_windows.cc
index 5d65ae3ab3..02121aa0ab 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_windows.cc
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_windows.cc
@@ -21,38 +21,516 @@
 #if GRPC_ARES == 1 && defined(GPR_WINDOWS)
 
 #include <ares.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/log_windows.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/time.h>
 #include <string.h>
+#include "src/core/lib/gpr/string.h"
 #include "src/core/lib/gprpp/memory.h"
+#include "src/core/lib/iomgr/combiner.h"
+#include "src/core/lib/iomgr/socket_windows.h"
+#include "src/core/lib/iomgr/tcp_windows.h"
+#include "src/core/lib/slice/slice_internal.h"
 
 #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h"
+#include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h"
+
+/* TODO(apolcyn): remove this hack after fixing upstream.
+ * Our grpc/c-ares code on Windows uses the ares_set_socket_functions API,
+ * which uses "struct iovec" type, which on Windows is defined inside of
+ * a c-ares header that is not public.
+ * See https://github.com/c-ares/c-ares/issues/206. */
+struct iovec {
+  void* iov_base;
+  size_t iov_len;
+};
 
 namespace grpc_core {
 
-/* TODO: fill in the body of GrpcPolledFdWindows to enable c-ares on Windows.
-   This dummy implementation only allows grpc to compile on windows with
-   GRPC_ARES=1. */
+/* c-ares creates its own sockets and is meant to read them when readable and
+ * write them when writeable. To fit this socket usage model into the grpc
+ * windows poller (which gives notifications when attempted reads and writes are
+ * actually fulfilled rather than possible), this GrpcPolledFdWindows class
+ * takes advantage of the ares_set_socket_functions API and acts as a virtual
+ * socket. It holds its own read and write buffers which are written to and read
+ * from c-ares and are used with the grpc windows poller, and it, e.g.,
+ * manufactures virtual socket error codes when it e.g. needs to tell the c-ares
+ * library to wait for an async read. */
 class GrpcPolledFdWindows : public GrpcPolledFd {
  public:
-  GrpcPolledFdWindows() { abort(); }
-  ~GrpcPolledFdWindows() { abort(); }
+  enum WriteState {
+    WRITE_IDLE,
+    WRITE_REQUESTED,
+    WRITE_PENDING,
+    WRITE_WAITING_FOR_VERIFICATION_UPON_RETRY,
+  };
+
+  GrpcPolledFdWindows(ares_socket_t as, grpc_combiner* combiner)
+      : read_buf_(grpc_empty_slice()),
+        write_buf_(grpc_empty_slice()),
+        write_state_(WRITE_IDLE),
+        gotten_into_driver_list_(false) {
+    gpr_asprintf(&name_, "c-ares socket: %" PRIdPTR, as);
+    winsocket_ = grpc_winsocket_create(as, name_);
+    combiner_ = GRPC_COMBINER_REF(combiner, name_);
+    GRPC_CLOSURE_INIT(&outer_read_closure_,
+                      &GrpcPolledFdWindows::OnIocpReadable, this,
+                      grpc_combiner_scheduler(combiner_));
+    GRPC_CLOSURE_INIT(&outer_write_closure_,
+                      &GrpcPolledFdWindows::OnIocpWriteable, this,
+                      grpc_combiner_scheduler(combiner_));
+  }
+
+  ~GrpcPolledFdWindows() {
+    GRPC_COMBINER_UNREF(combiner_, name_);
+    grpc_slice_unref_internal(read_buf_);
+    grpc_slice_unref_internal(write_buf_);
+    GPR_ASSERT(read_closure_ == nullptr);
+    GPR_ASSERT(write_closure_ == nullptr);
+    grpc_winsocket_destroy(winsocket_);
+    gpr_free(name_);
+  }
+
+  void ScheduleAndNullReadClosure(grpc_error* error) {
+    GRPC_CLOSURE_SCHED(read_closure_, error);
+    read_closure_ = nullptr;
+  }
+
+  void ScheduleAndNullWriteClosure(grpc_error* error) {
+    GRPC_CLOSURE_SCHED(write_closure_, error);
+    write_closure_ = nullptr;
+  }
+
   void RegisterForOnReadableLocked(grpc_closure* read_closure) override {
-    abort();
+    GPR_ASSERT(read_closure_ == nullptr);
+    read_closure_ = read_closure;
+    GPR_ASSERT(GRPC_SLICE_LENGTH(read_buf_) == 0);
+    grpc_slice_unref_internal(read_buf_);
+    read_buf_ = GRPC_SLICE_MALLOC(4192);
+    WSABUF buffer;
+    buffer.buf = (char*)GRPC_SLICE_START_PTR(read_buf_);
+    buffer.len = GRPC_SLICE_LENGTH(read_buf_);
+    memset(&winsocket_->read_info.overlapped, 0, sizeof(OVERLAPPED));
+    recv_from_source_addr_len_ = sizeof(recv_from_source_addr_);
+    DWORD flags = 0;
+    if (WSARecvFrom(grpc_winsocket_wrapped_socket(winsocket_), &buffer, 1,
+                    nullptr, &flags, (sockaddr*)recv_from_source_addr_,
+                    &recv_from_source_addr_len_,
+                    &winsocket_->read_info.overlapped, nullptr)) {
+      char* msg = gpr_format_message(WSAGetLastError());
+      grpc_error* error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
+      GRPC_CARES_TRACE_LOG(
+          "RegisterForOnReadableLocked: WSARecvFrom error:|%s|. fd:|%s|", msg,
+          GetName());
+      gpr_free(msg);
+      if (WSAGetLastError() != WSA_IO_PENDING) {
+        ScheduleAndNullReadClosure(error);
+        return;
+      }
+    }
+    grpc_socket_notify_on_read(winsocket_, &outer_read_closure_);
   }
+
   void RegisterForOnWriteableLocked(grpc_closure* write_closure) override {
+    GRPC_CARES_TRACE_LOG(
+        "RegisterForOnWriteableLocked. fd:|%s|. Current write state: %d",
+        GetName(), write_state_);
+    GPR_ASSERT(write_closure_ == nullptr);
+    write_closure_ = write_closure;
+    switch (write_state_) {
+      case WRITE_IDLE:
+        ScheduleAndNullWriteClosure(GRPC_ERROR_NONE);
+        break;
+      case WRITE_REQUESTED:
+        write_state_ = WRITE_PENDING;
+        SendWriteBuf(nullptr, &winsocket_->write_info.overlapped);
+        grpc_socket_notify_on_write(winsocket_, &outer_write_closure_);
+        break;
+      case WRITE_PENDING:
+      case WRITE_WAITING_FOR_VERIFICATION_UPON_RETRY:
+        abort();
+    }
+  }
+
+  bool IsFdStillReadableLocked() override {
+    return GRPC_SLICE_LENGTH(read_buf_) > 0;
+  }
+
+  void ShutdownLocked(grpc_error* error) override {
+    grpc_winsocket_shutdown(winsocket_);
+  }
+
+  ares_socket_t GetWrappedAresSocketLocked() override {
+    return grpc_winsocket_wrapped_socket(winsocket_);
+  }
+
+  const char* GetName() override { return name_; }
+
+  ares_ssize_t RecvFrom(void* data, ares_socket_t data_len, int flags,
+                        struct sockaddr* from, ares_socklen_t* from_len) {
+    GRPC_CARES_TRACE_LOG(
+        "RecvFrom called on fd:|%s|. Current read buf length:|%d|", GetName(),
+        GRPC_SLICE_LENGTH(read_buf_));
+    if (GRPC_SLICE_LENGTH(read_buf_) == 0) {
+      WSASetLastError(WSAEWOULDBLOCK);
+      return -1;
+    }
+    ares_ssize_t bytes_read = 0;
+    for (size_t i = 0; i < GRPC_SLICE_LENGTH(read_buf_) && i < data_len; i++) {
+      ((char*)data)[i] = GRPC_SLICE_START_PTR(read_buf_)[i];
+      bytes_read++;
+    }
+    read_buf_ = grpc_slice_sub_no_ref(read_buf_, bytes_read,
+                                      GRPC_SLICE_LENGTH(read_buf_));
+    /* c-ares overloads this recv_from virtual socket function to receive
+     * data on both UDP and TCP sockets, and from is nullptr for TCP. */
+    if (from != nullptr) {
+      GPR_ASSERT(*from_len <= recv_from_source_addr_len_);
+      memcpy(from, &recv_from_source_addr_, recv_from_source_addr_len_);
+      *from_len = recv_from_source_addr_len_;
+    }
+    return bytes_read;
+  }
+
+  grpc_slice FlattenIovec(const struct iovec* iov, int iov_count) {
+    int total = 0;
+    for (int i = 0; i < iov_count; i++) {
+      total += iov[i].iov_len;
+    }
+    grpc_slice out = GRPC_SLICE_MALLOC(total);
+    size_t cur = 0;
+    for (int i = 0; i < iov_count; i++) {
+      for (int k = 0; k < iov[i].iov_len; k++) {
+        GRPC_SLICE_START_PTR(out)[cur++] = ((char*)iov[i].iov_base)[k];
+      }
+    }
+    return out;
+  }
+
+  int SendWriteBuf(LPDWORD bytes_sent_ptr, LPWSAOVERLAPPED overlapped) {
+    WSABUF buf;
+    buf.len = GRPC_SLICE_LENGTH(write_buf_);
+    buf.buf = (char*)GRPC_SLICE_START_PTR(write_buf_);
+    DWORD flags = 0;
+    int out = WSASend(grpc_winsocket_wrapped_socket(winsocket_), &buf, 1,
+                      bytes_sent_ptr, flags, overlapped, nullptr);
+    GRPC_CARES_TRACE_LOG(
+        "WSASend: name:%s. buf len:%d. bytes sent: %d. overlapped %p. return "
+        "val: %d",
+        GetName(), buf.len, *bytes_sent_ptr, overlapped, out);
+    return out;
+  }
+
+  ares_ssize_t TrySendWriteBufSyncNonBlocking() {
+    GPR_ASSERT(write_state_ == WRITE_IDLE);
+    ares_ssize_t total_sent;
+    DWORD bytes_sent = 0;
+    if (SendWriteBuf(&bytes_sent, nullptr) != 0) {
+      char* msg = gpr_format_message(WSAGetLastError());
+      GRPC_CARES_TRACE_LOG(
+          "TrySendWriteBufSyncNonBlocking: SendWriteBuf error:|%s|. fd:|%s|",
+          msg, GetName());
+      gpr_free(msg);
+      if (WSAGetLastError() == WSA_IO_PENDING) {
+        WSASetLastError(WSAEWOULDBLOCK);
+        write_state_ = WRITE_REQUESTED;
+      }
+    }
+    write_buf_ = grpc_slice_sub_no_ref(write_buf_, bytes_sent,
+                                       GRPC_SLICE_LENGTH(write_buf_));
+    return bytes_sent;
+  }
+
+  ares_ssize_t SendV(const struct iovec* iov, int iov_count) {
+    GRPC_CARES_TRACE_LOG("SendV called on fd:|%s|. Current write state: %d",
+                         GetName(), write_state_);
+    switch (write_state_) {
+      case WRITE_IDLE:
+        GPR_ASSERT(GRPC_SLICE_LENGTH(write_buf_) == 0);
+        grpc_slice_unref_internal(write_buf_);
+        write_buf_ = FlattenIovec(iov, iov_count);
+        return TrySendWriteBufSyncNonBlocking();
+      case WRITE_REQUESTED:
+      case WRITE_PENDING:
+        WSASetLastError(WSAEWOULDBLOCK);
+        return -1;
+      case WRITE_WAITING_FOR_VERIFICATION_UPON_RETRY:
+        grpc_slice currently_attempted = FlattenIovec(iov, iov_count);
+        GPR_ASSERT(GRPC_SLICE_LENGTH(currently_attempted) >=
+                   GRPC_SLICE_LENGTH(write_buf_));
+        ares_ssize_t total_sent = 0;
+        for (size_t i = 0; i < GRPC_SLICE_LENGTH(write_buf_); i++) {
+          GPR_ASSERT(GRPC_SLICE_START_PTR(currently_attempted)[i] ==
+                     GRPC_SLICE_START_PTR(write_buf_)[i]);
+          total_sent++;
+        }
+        grpc_slice_unref_internal(write_buf_);
+        write_buf_ =
+            grpc_slice_sub_no_ref(currently_attempted, total_sent,
+                                  GRPC_SLICE_LENGTH(currently_attempted));
+        write_state_ = WRITE_IDLE;
+        total_sent += TrySendWriteBufSyncNonBlocking();
+        return total_sent;
+    }
     abort();
   }
-  bool IsFdStillReadableLocked() override { abort(); }
-  void ShutdownLocked(grpc_error* error) override { abort(); }
-  ares_socket_t GetWrappedAresSocketLocked() override { abort(); }
-  const char* GetName() override { abort(); }
+
+  int Connect(const struct sockaddr* target, ares_socklen_t target_len) {
+    SOCKET s = grpc_winsocket_wrapped_socket(winsocket_);
+    GRPC_CARES_TRACE_LOG("Connect: fd:|%s|", GetName());
+    int out =
+        WSAConnect(s, target, target_len, nullptr, nullptr, nullptr, nullptr);
+    if (out != 0) {
+      char* msg = gpr_format_message(WSAGetLastError());
+      GRPC_CARES_TRACE_LOG("Connect error code:|%d|, msg:|%s|. fd:|%s|",
+                           WSAGetLastError(), msg, GetName());
+      gpr_free(msg);
+      // c-ares expects a posix-style connect API
+      out = -1;
+    }
+    return out;
+  }
+
+  static void OnIocpReadable(void* arg, grpc_error* error) {
+    GrpcPolledFdWindows* polled_fd = static_cast<GrpcPolledFdWindows*>(arg);
+    polled_fd->OnIocpReadableInner(error);
+  }
+
+  void OnIocpReadableInner(grpc_error* error) {
+    if (error == GRPC_ERROR_NONE) {
+      if (winsocket_->read_info.wsa_error != 0) {
+        /* WSAEMSGSIZE would be due to receiving more data
+         * than our read buffer's fixed capacity. Assume that
+         * the connection is TCP and read the leftovers
+         * in subsequent c-ares reads. */
+        if (winsocket_->read_info.wsa_error != WSAEMSGSIZE) {
+          GRPC_ERROR_UNREF(error);
+          char* msg = gpr_format_message(winsocket_->read_info.wsa_error);
+          GRPC_CARES_TRACE_LOG(
+              "OnIocpReadableInner. winsocket error:|%s|. fd:|%s|", msg,
+              GetName());
+          error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
+          gpr_free(msg);
+        }
+      }
+    }
+    if (error == GRPC_ERROR_NONE) {
+      read_buf_ = grpc_slice_sub_no_ref(read_buf_, 0,
+                                        winsocket_->read_info.bytes_transfered);
+    } else {
+      grpc_slice_unref_internal(read_buf_);
+      read_buf_ = grpc_empty_slice();
+    }
+    GRPC_CARES_TRACE_LOG(
+        "OnIocpReadable finishing. read buf length now:|%d|. :fd:|%s|",
+        GRPC_SLICE_LENGTH(read_buf_), GetName());
+    ScheduleAndNullReadClosure(error);
+  }
+
+  static void OnIocpWriteable(void* arg, grpc_error* error) {
+    GrpcPolledFdWindows* polled_fd = static_cast<GrpcPolledFdWindows*>(arg);
+    polled_fd->OnIocpWriteableInner(error);
+  }
+
+  void OnIocpWriteableInner(grpc_error* error) {
+    GRPC_CARES_TRACE_LOG("OnIocpWriteableInner. fd:|%s|", GetName());
+    if (error == GRPC_ERROR_NONE) {
+      if (winsocket_->write_info.wsa_error != 0) {
+        char* msg = gpr_format_message(winsocket_->write_info.wsa_error);
+        GRPC_CARES_TRACE_LOG(
+            "OnIocpWriteableInner. winsocket error:|%s|. fd:|%s|", msg,
+            GetName());
+        GRPC_ERROR_UNREF(error);
+        error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
+        gpr_free(msg);
+      }
+    }
+    GPR_ASSERT(write_state_ == WRITE_PENDING);
+    if (error == GRPC_ERROR_NONE) {
+      write_state_ = WRITE_WAITING_FOR_VERIFICATION_UPON_RETRY;
+      write_buf_ = grpc_slice_sub_no_ref(
+          write_buf_, 0, winsocket_->write_info.bytes_transfered);
+    } else {
+      grpc_slice_unref_internal(write_buf_);
+      write_buf_ = grpc_empty_slice();
+    }
+    ScheduleAndNullWriteClosure(error);
+  }
+
+  bool gotten_into_driver_list() const { return gotten_into_driver_list_; }
+  void set_gotten_into_driver_list() { gotten_into_driver_list_ = true; }
+
+  grpc_combiner* combiner_;
+  char recv_from_source_addr_[200];
+  ares_socklen_t recv_from_source_addr_len_;
+  grpc_slice read_buf_;
+  grpc_slice write_buf_;
+  grpc_closure* read_closure_ = nullptr;
+  grpc_closure* write_closure_ = nullptr;
+  grpc_closure outer_read_closure_;
+  grpc_closure outer_write_closure_;
+  grpc_winsocket* winsocket_;
+  WriteState write_state_;
+  char* name_ = nullptr;
+  bool gotten_into_driver_list_;
 };
 
-GrpcPolledFd* NewGrpcPolledFdLocked(ares_socket_t as,
-                                    grpc_pollset_set* driver_pollset_set) {
-  return nullptr;
-}
+struct SockToPolledFdEntry {
+  SockToPolledFdEntry(SOCKET s, GrpcPolledFdWindows* fd)
+      : socket(s), polled_fd(fd) {}
+  SOCKET socket;
+  GrpcPolledFdWindows* polled_fd;
+  SockToPolledFdEntry* next = nullptr;
+};
+
+/* A SockToPolledFdMap can make ares_socket_t types (SOCKET's on windows)
+ * to GrpcPolledFdWindow's, and is used to find the appropriate
+ * GrpcPolledFdWindows to handle a virtual socket call when c-ares makes that
+ * socket call on the ares_socket_t type. Instances are owned by and one-to-one
+ * with a GrpcPolledFdWindows factory and event driver */
+class SockToPolledFdMap {
+ public:
+  SockToPolledFdMap(grpc_combiner* combiner) {
+    combiner_ = GRPC_COMBINER_REF(combiner, "sock to polled fd map");
+  }
+
+  ~SockToPolledFdMap() {
+    GPR_ASSERT(head_ == nullptr);
+    GRPC_COMBINER_UNREF(combiner_, "sock to polled fd map");
+  }
+
+  void AddNewSocket(SOCKET s, GrpcPolledFdWindows* polled_fd) {
+    SockToPolledFdEntry* new_node = New<SockToPolledFdEntry>(s, polled_fd);
+    new_node->next = head_;
+    head_ = new_node;
+  }
+
+  GrpcPolledFdWindows* LookupPolledFd(SOCKET s) {
+    for (SockToPolledFdEntry* node = head_; node != nullptr;
+         node = node->next) {
+      if (node->socket == s) {
+        GPR_ASSERT(node->polled_fd != nullptr);
+        return node->polled_fd;
+      }
+    }
+    abort();
+  }
+
+  void RemoveEntry(SOCKET s) {
+    GPR_ASSERT(head_ != nullptr);
+    SockToPolledFdEntry** prev = &head_;
+    for (SockToPolledFdEntry* node = head_; node != nullptr;
+         node = node->next) {
+      if (node->socket == s) {
+        *prev = node->next;
+        Delete(node);
+        return;
+      }
+      prev = &node->next;
+    }
+    abort();
+  }
+
+  /* These virtual socket functions are called from within the c-ares
+   * library. These methods generally dispatch those socket calls to the
+   * appropriate methods. The virtual "socket" and "close" methods are
+   * special and instead create/add and remove/destroy GrpcPolledFdWindows
+   * objects.
+   */
+  static ares_socket_t Socket(int af, int type, int protocol, void* user_data) {
+    SockToPolledFdMap* map = static_cast<SockToPolledFdMap*>(user_data);
+    SOCKET s = WSASocket(af, type, protocol, nullptr, 0, WSA_FLAG_OVERLAPPED);
+    if (s == INVALID_SOCKET) {
+      return s;
+    }
+    grpc_tcp_set_non_block(s);
+    GrpcPolledFdWindows* polled_fd =
+        New<GrpcPolledFdWindows>(s, map->combiner_);
+    map->AddNewSocket(s, polled_fd);
+    return s;
+  }
+
+  static int Connect(ares_socket_t as, const struct sockaddr* target,
+                     ares_socklen_t target_len, void* user_data) {
+    SockToPolledFdMap* map = static_cast<SockToPolledFdMap*>(user_data);
+    GrpcPolledFdWindows* polled_fd = map->LookupPolledFd(as);
+    return polled_fd->Connect(target, target_len);
+  }
+
+  static ares_ssize_t SendV(ares_socket_t as, const struct iovec* iov,
+                            int iovec_count, void* user_data) {
+    SockToPolledFdMap* map = static_cast<SockToPolledFdMap*>(user_data);
+    GrpcPolledFdWindows* polled_fd = map->LookupPolledFd(as);
+    return polled_fd->SendV(iov, iovec_count);
+  }
+
+  static ares_ssize_t RecvFrom(ares_socket_t as, void* data, size_t data_len,
+                               int flags, struct sockaddr* from,
+                               ares_socklen_t* from_len, void* user_data) {
+    SockToPolledFdMap* map = static_cast<SockToPolledFdMap*>(user_data);
+    GrpcPolledFdWindows* polled_fd = map->LookupPolledFd(as);
+    return polled_fd->RecvFrom(data, data_len, flags, from, from_len);
+  }
+
+  static int CloseSocket(SOCKET s, void* user_data) {
+    SockToPolledFdMap* map = static_cast<SockToPolledFdMap*>(user_data);
+    GrpcPolledFdWindows* polled_fd = map->LookupPolledFd(s);
+    map->RemoveEntry(s);
+    // If a gRPC polled fd has not made it in to the driver's list yet, then
+    // the driver has not and will never see this socket.
+    if (!polled_fd->gotten_into_driver_list()) {
+      polled_fd->ShutdownLocked(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+          "Shut down c-ares fd before without it ever having made it into the "
+          "driver's list"));
+      return 0;
+    }
+    return 0;
+  }
+
+ private:
+  SockToPolledFdEntry* head_ = nullptr;
+  grpc_combiner* combiner_;
+};
+
+const struct ares_socket_functions custom_ares_sock_funcs = {
+    &SockToPolledFdMap::Socket /* socket */,
+    &SockToPolledFdMap::CloseSocket /* close */,
+    &SockToPolledFdMap::Connect /* connect */,
+    &SockToPolledFdMap::RecvFrom /* recvfrom */,
+    &SockToPolledFdMap::SendV /* sendv */,
+};
+
+class GrpcPolledFdFactoryWindows : public GrpcPolledFdFactory {
+ public:
+  GrpcPolledFdFactoryWindows(grpc_combiner* combiner)
+      : sock_to_polled_fd_map_(combiner) {}
+
+  GrpcPolledFd* NewGrpcPolledFdLocked(ares_socket_t as,
+                                      grpc_pollset_set* driver_pollset_set,
+                                      grpc_combiner* combiner) override {
+    GrpcPolledFdWindows* polled_fd = sock_to_polled_fd_map_.LookupPolledFd(as);
+    // Set a flag so that the virtual socket "close" method knows it
+    // doesn't need to call ShutdownLocked, since now the driver will.
+    polled_fd->set_gotten_into_driver_list();
+    return polled_fd;
+  }
 
-void ConfigureAresChannelLocked(ares_channel* channel) { abort(); }
+  void ConfigureAresChannelLocked(ares_channel channel) override {
+    ares_set_socket_functions(channel, &custom_ares_sock_funcs,
+                              &sock_to_polled_fd_map_);
+  }
+
+ private:
+  SockToPolledFdMap sock_to_polled_fd_map_;
+};
+
+UniquePtr<GrpcPolledFdFactory> NewGrpcPolledFdFactory(grpc_combiner* combiner) {
+  return UniquePtr<GrpcPolledFdFactory>(
+      New<GrpcPolledFdFactoryWindows>(combiner));
+}
 
 }  // namespace grpc_core
 
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc
index b3d6437e9a..485998f5e4 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc
@@ -49,6 +49,8 @@ static gpr_mu g_init_mu;
 grpc_core::TraceFlag grpc_trace_cares_address_sorting(false,
                                                       "cares_address_sorting");
 
+grpc_core::TraceFlag grpc_trace_cares_resolver(false, "cares_resolver");
+
 struct grpc_ares_request {
   /** indicates the DNS server to use, if specified */
   struct ares_addr_port_node dns_server_addr;
diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h
index 17eaa7ccf0..ca5779e1d7 100644
--- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h
+++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h
@@ -28,6 +28,13 @@
 
 extern grpc_core::TraceFlag grpc_trace_cares_address_sorting;
 
+extern grpc_core::TraceFlag grpc_trace_cares_resolver;
+
+#define GRPC_CARES_TRACE_LOG(format, ...)                         \
+  if (grpc_trace_cares_resolver.enabled()) {                      \
+    gpr_log(GPR_DEBUG, "(c-ares resolver) " format, __VA_ARGS__); \
+  }
+
 typedef struct grpc_ares_request grpc_ares_request;
 
 /* Asynchronously resolve \a name. Use \a default_port if a port isn't
diff --git a/src/core/lib/iomgr/iocp_windows.cc b/src/core/lib/iomgr/iocp_windows.cc
index ce77231036..ad325fe215 100644
--- a/src/core/lib/iomgr/iocp_windows.cc
+++ b/src/core/lib/iomgr/iocp_windows.cc
@@ -89,10 +89,15 @@ grpc_iocp_work_status grpc_iocp_work(grpc_millis deadline) {
   } else {
     abort();
   }
-  success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes,
-                                   FALSE, &flags);
-  info->bytes_transfered = bytes;
-  info->wsa_error = success ? 0 : WSAGetLastError();
+  if (socket->shutdown_called) {
+    info->bytes_transfered = 0;
+    info->wsa_error = WSA_OPERATION_ABORTED;
+  } else {
+    success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes,
+                                     FALSE, &flags);
+    info->bytes_transfered = bytes;
+    info->wsa_error = success ? 0 : WSAGetLastError();
+  }
   GPR_ASSERT(overlapped == &info->overlapped);
   grpc_socket_become_ready(socket, info);
   return GRPC_IOCP_WORK_WORK;
diff --git a/src/core/lib/iomgr/socket_windows.cc b/src/core/lib/iomgr/socket_windows.cc
index 4ad31cb35d..999c6646ad 100644
--- a/src/core/lib/iomgr/socket_windows.cc
+++ b/src/core/lib/iomgr/socket_windows.cc
@@ -52,6 +52,10 @@ grpc_winsocket* grpc_winsocket_create(SOCKET socket, const char* name) {
   return r;
 }
 
+SOCKET grpc_winsocket_wrapped_socket(grpc_winsocket* socket) {
+  return socket->socket;
+}
+
 /* Schedule a shutdown of the socket operations. Will call the pending
    operations to abort them. We need to do that this way because of the
    various callsites of that function, which happens to be in various
diff --git a/src/core/lib/iomgr/socket_windows.h b/src/core/lib/iomgr/socket_windows.h
index b09b9da562..46d7d58356 100644
--- a/src/core/lib/iomgr/socket_windows.h
+++ b/src/core/lib/iomgr/socket_windows.h
@@ -92,6 +92,8 @@ typedef struct grpc_winsocket {
    it will be responsible for closing it. */
 grpc_winsocket* grpc_winsocket_create(SOCKET socket, const char* name);
 
+SOCKET grpc_winsocket_wrapped_socket(grpc_winsocket* socket);
+
 /* Initiate an asynchronous shutdown of the socket. Will call off any pending
    operation to cancel them. */
 void grpc_winsocket_shutdown(grpc_winsocket* socket);
diff --git a/src/core/lib/iomgr/tcp_windows.cc b/src/core/lib/iomgr/tcp_windows.cc
index 5d316d477b..b3cb442f18 100644
--- a/src/core/lib/iomgr/tcp_windows.cc
+++ b/src/core/lib/iomgr/tcp_windows.cc
@@ -53,7 +53,7 @@
 
 extern grpc_core::TraceFlag grpc_tcp_trace;
 
-static grpc_error* set_non_block(SOCKET sock) {
+grpc_error* grpc_tcp_set_non_block(SOCKET sock) {
   int status;
   uint32_t param = 1;
   DWORD ret;
@@ -90,7 +90,7 @@ static grpc_error* enable_loopback_fast_path(SOCKET sock) {
 
 grpc_error* grpc_tcp_prepare_socket(SOCKET sock) {
   grpc_error* err;
-  err = set_non_block(sock);
+  err = grpc_tcp_set_non_block(sock);
   if (err != GRPC_ERROR_NONE) return err;
   err = set_dualstack(sock);
   if (err != GRPC_ERROR_NONE) return err;
diff --git a/src/core/lib/iomgr/tcp_windows.h b/src/core/lib/iomgr/tcp_windows.h
index 161a545a2a..04ef8102b6 100644
--- a/src/core/lib/iomgr/tcp_windows.h
+++ b/src/core/lib/iomgr/tcp_windows.h
@@ -46,6 +46,8 @@ grpc_endpoint* grpc_tcp_create(grpc_winsocket* socket,
 
 grpc_error* grpc_tcp_prepare_socket(SOCKET sock);
 
+grpc_error* grpc_tcp_set_non_block(SOCKET sock);
+
 #endif
 
 #endif /* GRPC_CORE_LIB_IOMGR_TCP_WINDOWS_H */
diff --git a/templates/test/cpp/naming/resolver_component_tests_defs.include b/templates/test/cpp/naming/resolver_component_tests_defs.include
index bc981dc83e..b34845e01a 100644
--- a/templates/test/cpp/naming/resolver_component_tests_defs.include
+++ b/templates/test/cpp/naming/resolver_component_tests_defs.include
@@ -22,6 +22,7 @@ import tempfile
 import os
 import time
 import signal
+import platform
 
 
 argp = argparse.ArgumentParser(description='Run c-ares resolver tests')
@@ -43,6 +44,11 @@ args = argp.parse_args()
 def test_runner_log(msg):
   sys.stderr.write('\n%s: %s\n' % (__file__, msg))
 
+def python_args(arg_list):
+  if platform.system() == 'Windows':
+    return [sys.executable] + arg_list
+  return arg_list
+
 cur_resolver = os.environ.get('GRPC_DNS_RESOLVER')
 if cur_resolver and cur_resolver != 'ares':
   test_runner_log(('WARNING: cur resolver set to %s. This set of tests '
@@ -50,26 +56,27 @@ if cur_resolver and cur_resolver != 'ares':
   test_runner_log('Exit 1 without running tests.')
   sys.exit(1)
 os.environ.update({'GRPC_DNS_RESOLVER': 'ares'})
+os.environ.update({'GRPC_TRACE': 'cares_resolver'})
 
 def wait_until_dns_server_is_up(args,
                                 dns_server_subprocess,
                                 dns_server_subprocess_output):
   for i in range(0, 30):
     test_runner_log('Health check: attempt to connect to DNS server over TCP.')
-    tcp_connect_subprocess = subprocess.Popen([
+    tcp_connect_subprocess = subprocess.Popen(python_args([
         args.tcp_connect_bin_path,
         '--server_host', '127.0.0.1',
         '--server_port', str(args.dns_server_port),
-        '--timeout', str(1)])
+        '--timeout', str(1)]))
     tcp_connect_subprocess.communicate()
     if tcp_connect_subprocess.returncode == 0:
       test_runner_log(('Health check: attempt to make an A-record '
                        'query to DNS server.'))
-      dns_resolver_subprocess = subprocess.Popen([
+      dns_resolver_subprocess = subprocess.Popen(python_args([
           args.dns_resolver_bin_path,
           '--qname', 'health-check-local-dns-server-is-alive.resolver-tests.grpctestingexp',
           '--server_host', '127.0.0.1',
-          '--server_port', str(args.dns_server_port)],
+          '--server_port', str(args.dns_server_port)]),
           stdout=subprocess.PIPE)
       dns_resolver_stdout, _ = dns_resolver_subprocess.communicate()
       if dns_resolver_subprocess.returncode == 0:
@@ -91,10 +98,10 @@ def wait_until_dns_server_is_up(args,
 
 dns_server_subprocess_output = tempfile.mktemp()
 with open(dns_server_subprocess_output, 'w') as l:
-  dns_server_subprocess = subprocess.Popen([
+  dns_server_subprocess = subprocess.Popen(python_args([
       args.dns_server_bin_path,
       '--port', str(args.dns_server_port),
-      '--records_config_path', args.records_config_path],
+      '--records_config_path', args.records_config_path]),
       stdin=subprocess.PIPE,
       stdout=l,
       stderr=l)
diff --git a/test/cpp/naming/cancel_ares_query_test.cc b/test/cpp/naming/cancel_ares_query_test.cc
index 0d59bf6fb6..dec7c171dc 100644
--- a/test/cpp/naming/cancel_ares_query_test.cc
+++ b/test/cpp/naming/cancel_ares_query_test.cc
@@ -45,11 +45,14 @@
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
 
-// TODO: pull in different headers when enabling this
-// test on windows. Also set BAD_SOCKET_RETURN_VAL
-// to INVALID_SOCKET on windows.
+#ifdef GPR_WINDOWS
+#include "src/core/lib/iomgr/sockaddr_windows.h"
+#include "src/core/lib/iomgr/socket_windows.h"
+#define BAD_SOCKET_RETURN_VAL INVALID_SOCKET
+#else
 #include "src/core/lib/iomgr/sockaddr_posix.h"
 #define BAD_SOCKET_RETURN_VAL -1
+#endif
 
 namespace {
 
@@ -91,7 +94,13 @@ class FakeNonResponsiveDNSServer {
       abort();
     }
   }
-  ~FakeNonResponsiveDNSServer() { close(socket_); }
+  ~FakeNonResponsiveDNSServer() {
+#ifdef GPR_WINDOWS
+    closesocket(socket_);
+#else
+    close(socket_);
+#endif
+  }
 
  private:
   int socket_;
@@ -193,6 +202,38 @@ TEST(CancelDuringAresQuery, TestCancelActiveDNSQuery) {
   TestCancelActiveDNSQuery(&args);
 }
 
+#ifdef GPR_WINDOWS
+
+void MaybePollArbitraryPollsetTwice() {
+  grpc_pollset* pollset = (grpc_pollset*)gpr_zalloc(grpc_pollset_size());
+  gpr_mu* mu;
+  grpc_pollset_init(pollset, &mu);
+  grpc_pollset_worker* worker = nullptr;
+  // Make a zero timeout poll
+  gpr_mu_lock(mu);
+  GRPC_LOG_IF_ERROR(
+      "pollset_work",
+      grpc_pollset_work(pollset, &worker, grpc_core::ExecCtx::Get()->Now()));
+  gpr_mu_unlock(mu);
+  grpc_core::ExecCtx::Get()->Flush();
+  // Make a second zero-timeout poll (in case the first one
+  // short-circuited by picking up a previous "kick")
+  gpr_mu_lock(mu);
+  GRPC_LOG_IF_ERROR(
+      "pollset_work",
+      grpc_pollset_work(pollset, &worker, grpc_core::ExecCtx::Get()->Now()));
+  gpr_mu_unlock(mu);
+  grpc_core::ExecCtx::Get()->Flush();
+  grpc_pollset_destroy(pollset);
+  gpr_free(pollset);
+}
+
+#else
+
+void MaybePollArbitraryPollsetTwice() {}
+
+#endif
+
 TEST(CancelDuringAresQuery, TestFdsAreDeletedFromPollsetSet) {
   grpc_core::ExecCtx exec_ctx;
   ArgsStruct args;
@@ -209,6 +250,12 @@ TEST(CancelDuringAresQuery, TestFdsAreDeletedFromPollsetSet) {
   // this test. This test only cares about what happens to fd's that c-ares
   // opens.
   TestCancelActiveDNSQuery(&args);
+  // This test relies on the assumption that cancelling a c-ares query
+  // will flush out all callbacks on the current exec ctx, which is true
+  // on posix platforms but not on Windows, because fd shutdown on Windows
+  // requires a trip through the polling loop to schedule the callback.
+  // So we need to do extra polling work on Windows to free things up.
+  MaybePollArbitraryPollsetTwice();
   EXPECT_EQ(grpc_iomgr_count_objects_for_testing(), 0u);
   grpc_pollset_set_destroy(fake_other_pollset_set);
 }
diff --git a/test/cpp/naming/gen_build_yaml.py b/test/cpp/naming/gen_build_yaml.py
index 5dad2ea7af..1c9d0676b8 100755
--- a/test/cpp/naming/gen_build_yaml.py
+++ b/test/cpp/naming/gen_build_yaml.py
@@ -68,7 +68,7 @@ def main():
               'gtest': False,
               'run': False,
               'src': ['test/cpp/naming/resolver_component_test.cc'],
-              'platforms': ['linux', 'posix', 'mac'],
+              'platforms': ['linux', 'posix', 'mac', 'windows'],
               'deps': [
                   'grpc++_test_util' + unsecure_build_config_suffix,
                   'grpc_test_util' + unsecure_build_config_suffix,
@@ -129,7 +129,7 @@ def main():
           'gtest': True,
           'run': True,
           'src': ['test/cpp/naming/cancel_ares_query_test.cc'],
-          'platforms': ['linux', 'posix', 'mac'],
+          'platforms': ['linux', 'posix', 'mac', 'windows'],
           'deps': [
               'grpc++_test_util',
               'grpc_test_util',
diff --git a/test/cpp/naming/manual_run_resolver_component_test.py b/test/cpp/naming/manual_run_resolver_component_test.py
new file mode 100644
index 0000000000..fb2157741a
--- /dev/null
+++ b/test/cpp/naming/manual_run_resolver_component_test.py
@@ -0,0 +1,36 @@
+#!/usr/bin/env python
+# Copyright 2015 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.
+
+import os
+import subprocess
+import sys
+
+# The c-ares test suite doesn't get ran regularly on Windows, but
+# this script provides a way to run a lot of the tests manually.
+_MSBUILD_CONFIG = os.environ['CONFIG']
+os.chdir(os.path.join('..', '..', os.getcwd()))
+# This port is arbitrary, but it needs to be available.
+_DNS_SERVER_PORT = 15353
+
+subprocess.call([
+    sys.executable,
+    'test\\cpp\\naming\\resolver_component_tests_runner.py',
+    '--test_bin_path', 'cmake\\build\\%s\\resolver_component_test.exe' % _MSBUILD_CONFIG,
+    '--dns_server_bin_path', 'test\\cpp\\naming\\utils\\dns_server.py',
+    '--records_config_path', 'test\\cpp\\naming\\resolver_test_record_groups.yaml',
+    '--dns_server_port', str(_DNS_SERVER_PORT),
+    '--dns_resolver_bin_path', 'test\\cpp\\naming\\utils\\dns_resolver.py',
+    '--tcp_connect_bin_path', 'test\\cpp\\naming\\utils\\tcp_connect.py',
+])
diff --git a/test/cpp/naming/resolver_component_test.cc b/test/cpp/naming/resolver_component_test.cc
index 6ac548120c..3dc6e7178c 100644
--- a/test/cpp/naming/resolver_component_test.cc
+++ b/test/cpp/naming/resolver_component_test.cc
@@ -16,6 +16,8 @@
  *
  */
 
+#include <grpc/support/port_platform.h>
+
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
@@ -55,8 +57,15 @@
 // TODO: pull in different headers when enabling this
 // test on windows. Also set BAD_SOCKET_RETURN_VAL
 // to INVALID_SOCKET on windows.
+#ifdef GPR_WINDOWS
+#include "src/core/lib/iomgr/sockaddr_windows.h"
+#include "src/core/lib/iomgr/socket_windows.h"
+#include "src/core/lib/iomgr/tcp_windows.h"
+#define BAD_SOCKET_RETURN_VAL INVALID_SOCKET
+#else
 #include "src/core/lib/iomgr/sockaddr_posix.h"
 #define BAD_SOCKET_RETURN_VAL -1
+#endif
 
 using grpc::SubProcess;
 using std::vector;
@@ -241,6 +250,62 @@ void CheckLBPolicyResultLocked(grpc_channel_args* channel_args,
   }
 }
 
+#ifdef GPR_WINDOWS
+void OpenAndCloseSocketsStressLoop(int dummy_port, gpr_event* done_ev) {
+  sockaddr_in6 addr;
+  memset(&addr, 0, sizeof(addr));
+  addr.sin6_family = AF_INET6;
+  addr.sin6_port = htons(dummy_port);
+  ((char*)&addr.sin6_addr)[15] = 1;
+  for (;;) {
+    if (gpr_event_get(done_ev)) {
+      return;
+    }
+    std::vector<int> sockets;
+    for (size_t i = 0; i < 50; i++) {
+      SOCKET s = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, nullptr, 0,
+                           WSA_FLAG_OVERLAPPED);
+      ASSERT_TRUE(s != BAD_SOCKET_RETURN_VAL)
+          << "Failed to create TCP ipv6 socket";
+      gpr_log(GPR_DEBUG, "Opened socket: %d", s);
+      char val = 1;
+      ASSERT_TRUE(setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val)) !=
+                  SOCKET_ERROR)
+          << "Failed to set socketopt reuseaddr. WSA error: " +
+                 std::to_string(WSAGetLastError());
+      ASSERT_TRUE(grpc_tcp_set_non_block(s) == GRPC_ERROR_NONE)
+          << "Failed to set socket non-blocking";
+      ASSERT_TRUE(bind(s, (const sockaddr*)&addr, sizeof(addr)) != SOCKET_ERROR)
+          << "Failed to bind socket " + std::to_string(s) +
+                 " to [::1]:" + std::to_string(dummy_port) +
+                 ". WSA error: " + std::to_string(WSAGetLastError());
+      ASSERT_TRUE(listen(s, 1) != SOCKET_ERROR)
+          << "Failed to listen on socket " + std::to_string(s) +
+                 ". WSA error: " + std::to_string(WSAGetLastError());
+      sockets.push_back(s);
+    }
+    // Do a non-blocking accept followed by a close on all of those sockets.
+    // Do this in a separate loop to try to induce a time window to hit races.
+    for (size_t i = 0; i < sockets.size(); i++) {
+      gpr_log(GPR_DEBUG, "non-blocking accept then close on %d", sockets[i]);
+      ASSERT_TRUE(accept(sockets[i], nullptr, nullptr) == INVALID_SOCKET)
+          << "Accept on dummy socket unexpectedly accepted actual connection.";
+      ASSERT_TRUE(WSAGetLastError() == WSAEWOULDBLOCK)
+          << "OpenAndCloseSocketsStressLoop accept on socket " +
+                 std::to_string(sockets[i]) +
+                 " failed in "
+                 "an unexpected way. "
+                 "WSA error: " +
+                 std::to_string(WSAGetLastError()) +
+                 ". Socket use-after-close bugs are likely.";
+      ASSERT_TRUE(closesocket(sockets[i]) != SOCKET_ERROR)
+          << "Failed to close socket: " + std::to_string(sockets[i]) +
+                 ". WSA error: " + std::to_string(WSAGetLastError());
+    }
+  }
+  return;
+}
+#else
 void OpenAndCloseSocketsStressLoop(int dummy_port, gpr_event* done_ev) {
   // The goal of this loop is to catch socket
   // "use after close" bugs within the c-ares resolver by acting
@@ -311,6 +376,7 @@ void OpenAndCloseSocketsStressLoop(int dummy_port, gpr_event* done_ev) {
     }
   }
 }
+#endif
 
 void CheckResolverResultLocked(void* argsp, grpc_error* err) {
   EXPECT_EQ(err, GRPC_ERROR_NONE);
@@ -372,9 +438,9 @@ void RunResolvesRelevantRecordsTest(void (*OnDoneLocked)(void* arg,
   args.expected_lb_policy = FLAGS_expected_lb_policy;
   // maybe build the address with an authority
   char* whole_uri = nullptr;
-  GPR_ASSERT(asprintf(&whole_uri, "dns://%s/%s",
-                      FLAGS_local_dns_server_address.c_str(),
-                      FLAGS_target_name.c_str()));
+  GPR_ASSERT(gpr_asprintf(&whole_uri, "dns://%s/%s",
+                          FLAGS_local_dns_server_address.c_str(),
+                          FLAGS_target_name.c_str()));
   // create resolver and resolve
   grpc_core::OrphanablePtr<grpc_core::Resolver> resolver =
       grpc_core::ResolverRegistry::CreateResolver(whole_uri, nullptr,
diff --git a/test/cpp/naming/resolver_component_tests_runner.py b/test/cpp/naming/resolver_component_tests_runner.py
index 69386ebeb0..1873eec35b 100755
--- a/test/cpp/naming/resolver_component_tests_runner.py
+++ b/test/cpp/naming/resolver_component_tests_runner.py
@@ -22,6 +22,7 @@ import tempfile
 import os
 import time
 import signal
+import platform
 
 
 argp = argparse.ArgumentParser(description='Run c-ares resolver tests')
@@ -43,6 +44,11 @@ args = argp.parse_args()
 def test_runner_log(msg):
   sys.stderr.write('\n%s: %s\n' % (__file__, msg))
 
+def python_args(arg_list):
+  if platform.system() == 'Windows':
+    return [sys.executable] + arg_list
+  return arg_list
+
 cur_resolver = os.environ.get('GRPC_DNS_RESOLVER')
 if cur_resolver and cur_resolver != 'ares':
   test_runner_log(('WARNING: cur resolver set to %s. This set of tests '
@@ -50,26 +56,27 @@ if cur_resolver and cur_resolver != 'ares':
   test_runner_log('Exit 1 without running tests.')
   sys.exit(1)
 os.environ.update({'GRPC_DNS_RESOLVER': 'ares'})
+os.environ.update({'GRPC_TRACE': 'cares_resolver'})
 
 def wait_until_dns_server_is_up(args,
                                 dns_server_subprocess,
                                 dns_server_subprocess_output):
   for i in range(0, 30):
     test_runner_log('Health check: attempt to connect to DNS server over TCP.')
-    tcp_connect_subprocess = subprocess.Popen([
+    tcp_connect_subprocess = subprocess.Popen(python_args([
         args.tcp_connect_bin_path,
         '--server_host', '127.0.0.1',
         '--server_port', str(args.dns_server_port),
-        '--timeout', str(1)])
+        '--timeout', str(1)]))
     tcp_connect_subprocess.communicate()
     if tcp_connect_subprocess.returncode == 0:
       test_runner_log(('Health check: attempt to make an A-record '
                        'query to DNS server.'))
-      dns_resolver_subprocess = subprocess.Popen([
+      dns_resolver_subprocess = subprocess.Popen(python_args([
           args.dns_resolver_bin_path,
           '--qname', 'health-check-local-dns-server-is-alive.resolver-tests.grpctestingexp',
           '--server_host', '127.0.0.1',
-          '--server_port', str(args.dns_server_port)],
+          '--server_port', str(args.dns_server_port)]),
           stdout=subprocess.PIPE)
       dns_resolver_stdout, _ = dns_resolver_subprocess.communicate()
       if dns_resolver_subprocess.returncode == 0:
@@ -91,10 +98,10 @@ def wait_until_dns_server_is_up(args,
 
 dns_server_subprocess_output = tempfile.mktemp()
 with open(dns_server_subprocess_output, 'w') as l:
-  dns_server_subprocess = subprocess.Popen([
+  dns_server_subprocess = subprocess.Popen(python_args([
       args.dns_server_bin_path,
       '--port', str(args.dns_server_port),
-      '--records_config_path', args.records_config_path],
+      '--records_config_path', args.records_config_path]),
       stdin=subprocess.PIPE,
       stdout=l,
       stderr=l)
@@ -112,6 +119,18 @@ wait_until_dns_server_is_up(args,
                             dns_server_subprocess_output)
 num_test_failures = 0
 
+test_runner_log('Run test with target: %s' % 'no-srv-ipv4-single-target.resolver-tests-version-4.grpctestingexp.')
+current_test_subprocess = subprocess.Popen([
+  args.test_bin_path,
+  '--target_name', 'no-srv-ipv4-single-target.resolver-tests-version-4.grpctestingexp.',
+  '--expected_addrs', '5.5.5.5:443,False',
+  '--expected_chosen_service_config', '',
+  '--expected_lb_policy', '',
+  '--local_dns_server_address', '127.0.0.1:%d' % args.dns_server_port])
+current_test_subprocess.communicate()
+if current_test_subprocess.returncode != 0:
+  num_test_failures += 1
+
 test_runner_log('Run test with target: %s' % 'srv-ipv4-single-target.resolver-tests-version-4.grpctestingexp.')
 current_test_subprocess = subprocess.Popen([
   args.test_bin_path,
diff --git a/test/cpp/naming/resolver_test_record_groups.yaml b/test/cpp/naming/resolver_test_record_groups.yaml
index 6c4f89d09b..3c51a00c7b 100644
--- a/test/cpp/naming/resolver_test_record_groups.yaml
+++ b/test/cpp/naming/resolver_test_record_groups.yaml
@@ -1,6 +1,14 @@
 resolver_tests_common_zone_name: resolver-tests-version-4.grpctestingexp.
 resolver_component_tests:
 - expected_addrs:
+  - {address: '5.5.5.5:443', is_balancer: false}
+  expected_chosen_service_config: null
+  expected_lb_policy: null
+  record_to_resolve: no-srv-ipv4-single-target
+  records:
+    no-srv-ipv4-single-target:
+    - {TTL: '2100', data: 5.5.5.5, type: A}
+- expected_addrs:
   - {address: '1.2.3.4:1234', is_balancer: true}
   expected_chosen_service_config: null
   expected_lb_policy: null
diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json
index 5815f82fef..cf3b54e044 100644
--- a/tools/run_tests/generated/tests.json
+++ b/tools/run_tests/generated/tests.json
@@ -5784,7 +5784,8 @@
     "ci_platforms": [
       "linux", 
       "mac", 
-      "posix"
+      "posix", 
+      "windows"
     ], 
     "cpu_cost": 1.0, 
     "exclude_configs": [], 
@@ -5796,7 +5797,8 @@
     "platforms": [
       "linux", 
       "mac", 
-      "posix"
+      "posix", 
+      "windows"
     ], 
     "uses_polling": true
   },