Merge pull request #1563 from nicolasnoble/more-win32-fixes

Various Win32 fixes and improvements.

5 files changed, 31 insertions, 12 deletions

diff --git a/src/core/iomgr/endpoint_pair_windows.c b/src/core/iomgr/endpoint_pair_windows.c
index 7c945ebad4..988d622d01 100644
--- a/src/core/iomgr/endpoint_pair_windows.c
+++ b/src/core/iomgr/endpoint_pair_windows.c
@@ -56,6 +56,8 @@ static void create_sockets(SOCKET sv[2]) {
   GPR_ASSERT(lst_sock != INVALID_SOCKET);
 
   memset(&addr, 0, sizeof(addr));
+  addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+  addr.sin_family = AF_INET;
   GPR_ASSERT(bind(lst_sock, (struct sockaddr*)&addr, sizeof(addr)) != SOCKET_ERROR);
   GPR_ASSERT(listen(lst_sock, SOMAXCONN) != SOCKET_ERROR);
   GPR_ASSERT(getsockname(lst_sock, (struct sockaddr*)&addr, &addr_len) != SOCKET_ERROR);
diff --git a/src/core/iomgr/iocp_windows.h b/src/core/iomgr/iocp_windows.h
index fa3f5eee10..4503fdd55e 100644
--- a/src/core/iomgr/iocp_windows.h
+++ b/src/core/iomgr/iocp_windows.h
@@ -40,6 +40,7 @@
 #include "src/core/iomgr/socket_windows.h"
 
 void grpc_iocp_init(void);
+void grpc_iocp_kick(void);
 void grpc_iocp_shutdown(void);
 void grpc_iocp_add_socket(grpc_winsocket *);
 void grpc_iocp_socket_orphan(grpc_winsocket *);
diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c
index 35dbfa1587..ee5150a696 100644
--- a/src/core/iomgr/socket_windows.c
+++ b/src/core/iomgr/socket_windows.c
@@ -59,17 +59,21 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket) {
    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
    mutex hold states, and that'd be unsafe to call them directly. */
-void grpc_winsocket_shutdown(grpc_winsocket *socket) {
+int grpc_winsocket_shutdown(grpc_winsocket *socket) {
+  int callbacks_set = 0;
   gpr_mu_lock(&socket->state_mu);
   if (socket->read_info.cb) {
+    callbacks_set++;
     grpc_iomgr_add_delayed_callback(socket->read_info.cb,
                                     socket->read_info.opaque, 0);
   }
   if (socket->write_info.cb) {
+    callbacks_set++;
     grpc_iomgr_add_delayed_callback(socket->write_info.cb,
                                     socket->write_info.opaque, 0);
   }
   gpr_mu_unlock(&socket->state_mu);
+  return callbacks_set;
 }
 
 /* Abandons a socket. Either we're going to queue it up for garbage collecting
diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h
index 8898def854..b27eb14219 100644
--- a/src/core/iomgr/socket_windows.h
+++ b/src/core/iomgr/socket_windows.h
@@ -100,8 +100,8 @@ typedef struct grpc_winsocket {
 grpc_winsocket *grpc_winsocket_create(SOCKET socket);
 
 /* Initiate an asynchronous shutdown of the socket. Will call off any pending
-   operation to cancel them. */
-void grpc_winsocket_shutdown(grpc_winsocket *socket);
+   operation to cancel them. Returns the number of callbacks that got setup. */
+int grpc_winsocket_shutdown(grpc_winsocket *socket);
 
 /* Abandon a socket. */
 void grpc_winsocket_orphan(grpc_winsocket *socket);
diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c
index f16b4c1268..3341f558a3 100644
--- a/src/core/iomgr/tcp_windows.c
+++ b/src/core/iomgr/tcp_windows.c
@@ -118,12 +118,14 @@ static void on_read(void *tcpp, int from_iocp) {
   gpr_slice *slice = NULL;
   size_t nslices = 0;
   grpc_endpoint_cb_status status;
-  grpc_endpoint_read_cb cb = tcp->read_cb;
+  grpc_endpoint_read_cb cb;
   grpc_winsocket_callback_info *info = &socket->read_info;
   void *opaque = tcp->read_user_data;
   int do_abort = 0;
 
   gpr_mu_lock(&tcp->mu);
+  cb = tcp->read_cb;
+  tcp->read_cb = NULL;
   if (!from_iocp || tcp->shutting_down) {
     /* If we are here with from_iocp set to true, it means we got raced to
     shutting down the endpoint. No actual abort callback will happen
@@ -133,9 +135,12 @@ static void on_read(void *tcpp, int from_iocp) {
   gpr_mu_unlock(&tcp->mu);
 
   if (do_abort) {
-    if (from_iocp) gpr_slice_unref(tcp->read_slice);
+    if (from_iocp) {
+      tcp->socket->read_info.outstanding = 0;
+      gpr_slice_unref(tcp->read_slice);
+    }
     tcp_unref(tcp);
-    cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
+    if (cb) cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
     return;
   }
 
@@ -225,11 +230,13 @@ static void on_write(void *tcpp, int from_iocp) {
   grpc_winsocket *handle = tcp->socket;
   grpc_winsocket_callback_info *info = &handle->write_info;
   grpc_endpoint_cb_status status = GRPC_ENDPOINT_CB_OK;
-  grpc_endpoint_write_cb cb = tcp->write_cb;
+  grpc_endpoint_write_cb cb;
   void *opaque = tcp->write_user_data;
   int do_abort = 0;
 
   gpr_mu_lock(&tcp->mu);
+  cb = tcp->write_cb;
+  tcp->write_cb = NULL;
   if (!from_iocp || tcp->shutting_down) {
     /* If we are here with from_iocp set to true, it means we got raced to
         shutting down the endpoint. No actual abort callback will happen
@@ -238,15 +245,18 @@ static void on_write(void *tcpp, int from_iocp) {
   }
   gpr_mu_unlock(&tcp->mu);
 
-  GPR_ASSERT(tcp->socket->write_info.outstanding);
-
   if (do_abort) {
-    if (from_iocp) gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
+    if (from_iocp) {
+      tcp->socket->write_info.outstanding = 0;
+      gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
+    }
     tcp_unref(tcp);
-    cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN);
+    if (cb) cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN);
     return;
   }
 
+  GPR_ASSERT(tcp->socket->write_info.outstanding);
+
   if (info->wsa_error != 0) {
     char *utf8_message = gpr_format_message(info->wsa_error);
     gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
@@ -361,11 +371,13 @@ static void win_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) {
    concurrent access of the data structure in that regard. */
 static void win_shutdown(grpc_endpoint *ep) {
   grpc_tcp *tcp = (grpc_tcp *) ep;
+  int extra_refs = 0;
   gpr_mu_lock(&tcp->mu);
   /* At that point, what may happen is that we're already inside the IOCP
      callback. See the comments in on_read and on_write. */
   tcp->shutting_down = 1;
-  grpc_winsocket_shutdown(tcp->socket);
+  extra_refs = grpc_winsocket_shutdown(tcp->socket);
+  while (extra_refs--) tcp_ref(tcp);
   gpr_mu_unlock(&tcp->mu);
 }