1 files changed, 57 insertions, 67 deletions

diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c
index 123f46d71d..a63123c6c9 100644
--- a/src/core/iomgr/tcp_windows.c
+++ b/src/core/iomgr/tcp_windows.c
@@ -82,13 +82,11 @@ typedef struct grpc_tcp {
   /* Refcounting how many operations are in progress. */
   gpr_refcount refcount;
 
-  grpc_endpoint_read_cb read_cb;
-  void *read_user_data;
+  grpc_iomgr_closure *read_cb;
+  grpc_iomgr_closure *write_cb;
   gpr_slice read_slice;
-
-  grpc_endpoint_write_cb write_cb;
-  void *write_user_data;
-  gpr_slice_buffer write_slices;
+  gpr_slice_buffer *write_slices;
+  gpr_slice_buffer *read_slices;
 
   /* The IO Completion Port runs from another thread. We need some mechanism
      to protect ourselves when requesting a shutdown. */
@@ -102,7 +100,6 @@ static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); }
 
 static void tcp_unref(grpc_tcp *tcp) {
   if (gpr_unref(&tcp->refcount)) {
-    gpr_slice_buffer_destroy(&tcp->write_slices);
     grpc_winsocket_orphan(tcp->socket);
     gpr_mu_destroy(&tcp->mu);
     gpr_free(tcp->peer_string);
@@ -111,21 +108,16 @@ static void tcp_unref(grpc_tcp *tcp) {
 }
 
 /* Asynchronous callback from the IOCP, or the background thread. */
-static void on_read(void *tcpp, int from_iocp) {
-  grpc_tcp *tcp = (grpc_tcp *)tcpp;
+static int on_read(grpc_tcp *tcp, int from_iocp) {
   grpc_winsocket *socket = tcp->socket;
   gpr_slice sub;
   gpr_slice *slice = NULL;
   size_t nslices = 0;
-  grpc_endpoint_cb_status status;
-  grpc_endpoint_read_cb cb;
+  int success;
   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
@@ -140,8 +132,7 @@ static void on_read(void *tcpp, int from_iocp) {
       gpr_slice_unref(tcp->read_slice);
     }
     tcp_unref(tcp);
-    if (cb) cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
-    return;
+    return 0;
   }
 
   GPR_ASSERT(tcp->socket->read_info.outstanding);
@@ -152,27 +143,33 @@ static void on_read(void *tcpp, int from_iocp) {
       gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message);
       gpr_free(utf8_message);
     }
-    status = GRPC_ENDPOINT_CB_ERROR;
+    success = 0;
   } else {
     if (info->bytes_transfered != 0) {
       sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);
-      status = GRPC_ENDPOINT_CB_OK;
-      slice = ⊂
-      nslices = 1;
+      gpr_slice_buffer_add(tcp->read_slices, sub);
+      success = 1;
     } else {
       gpr_slice_unref(tcp->read_slice);
-      status = GRPC_ENDPOINT_CB_EOF;
+      success = 0;
     }
   }
 
   tcp->socket->read_info.outstanding = 0;
 
+  return success;
+}
+
+static void on_read_cb(void *tcpp, int from_iocp) {
+  grpc_tcp *tcp = tcpp;
+  grpc_iomgr_closure *cb = tcp->read_cb;
+  int success = on_read(tcp, from_iocp);
+  tcp->read_cb = NULL;
   tcp_unref(tcp);
-  cb(opaque, slice, nslices, status);
+  cb->cb(cb->cb_arg, success);
 }
 
-static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
-                               void *arg) {
+static grpc_endpoint_op_status win_read(grpc_endpoint *ep, gpr_slice_buffer *read_slices, grpc_iomgr_closure *cb) {
   grpc_tcp *tcp = (grpc_tcp *)ep;
   grpc_winsocket *handle = tcp->socket;
   grpc_winsocket_callback_info *info = &handle->read_info;
@@ -183,13 +180,11 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
 
   GPR_ASSERT(!tcp->socket->read_info.outstanding);
   if (tcp->shutting_down) {
-    cb(arg, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
-    return;
+    return GRPC_ENDPOINT_ERROR;
   }
-  tcp_ref(tcp);
   tcp->socket->read_info.outstanding = 1;
   tcp->read_cb = cb;
-  tcp->read_user_data = arg;
+  tcp->read_slices = read_slices;
 
   tcp->read_slice = gpr_slice_malloc(8192);
 
@@ -204,9 +199,8 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
   /* Did we get data immediately ? Yay. */
   if (info->wsa_error != WSAEWOULDBLOCK) {
     info->bytes_transfered = bytes_read;
-    /* This might heavily recurse. */
-    on_read(tcp, 1);
-    return;
+    gpr_log(GPR_DEBUG, "immread: %d bytes", bytes_read);
+    return on_read(tcp, 1) ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR;
   }
 
   /* Otherwise, let's retry, by queuing a read. */
@@ -218,12 +212,14 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
     int wsa_error = WSAGetLastError();
     if (wsa_error != WSA_IO_PENDING) {
       info->wsa_error = wsa_error;
-      on_read(tcp, 1);
-      return;
+      gpr_log(GPR_DEBUG, "immread: err=%d", wsa_error);
+      return on_read(tcp, 1) ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR;
     }
   }
 
-  grpc_socket_notify_on_read(tcp->socket, on_read, tcp);
+  tcp_ref(tcp);
+  grpc_socket_notify_on_read(tcp->socket, on_read_cb, tcp);
+  return GRPC_ENDPOINT_PENDING;
 }
 
 /* Asynchronous callback from the IOCP, or the background thread. */
@@ -231,9 +227,8 @@ static void on_write(void *tcpp, int from_iocp) {
   grpc_tcp *tcp = (grpc_tcp *)tcpp;
   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;
-  void *opaque = tcp->write_user_data;
+  grpc_iomgr_closure *cb;
+  int success;
   int do_abort = 0;
 
   gpr_mu_lock(&tcp->mu);
@@ -250,10 +245,11 @@ static void on_write(void *tcpp, int from_iocp) {
   if (do_abort) {
     if (from_iocp) {
       tcp->socket->write_info.outstanding = 0;
-      gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
     }
     tcp_unref(tcp);
-    if (cb) cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN);
+    if (cb) {
+      cb->cb(cb->cb_arg, 0);
+    }
     return;
   }
 
@@ -265,23 +261,22 @@ static void on_write(void *tcpp, int from_iocp) {
       gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
       gpr_free(utf8_message);
     }
-    status = GRPC_ENDPOINT_CB_ERROR;
+    success = 0;
   } else {
-    GPR_ASSERT(info->bytes_transfered == tcp->write_slices.length);
+    GPR_ASSERT(info->bytes_transfered == tcp->write_slices->length);
+    success = 1;
   }
 
-  gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
   tcp->socket->write_info.outstanding = 0;
 
   tcp_unref(tcp);
-  cb(opaque, status);
+  cb->cb(cb->cb_arg, success);
 }
 
 /* Initiates a write. */
-static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
-                                            gpr_slice *slices, size_t nslices,
-                                            grpc_endpoint_write_cb cb,
-                                            void *arg) {
+static grpc_endpoint_op_status win_write(grpc_endpoint *ep,
+                                            gpr_slice_buffer *slices,
+                                            grpc_iomgr_closure *cb) {
   grpc_tcp *tcp = (grpc_tcp *)ep;
   grpc_winsocket *socket = tcp->socket;
   grpc_winsocket_callback_info *info = &socket->write_info;
@@ -294,28 +289,26 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
 
   GPR_ASSERT(!tcp->socket->write_info.outstanding);
   if (tcp->shutting_down) {
-    return GRPC_ENDPOINT_WRITE_ERROR;
+    return GRPC_ENDPOINT_ERROR;
   }
   tcp_ref(tcp);
 
   tcp->socket->write_info.outstanding = 1;
   tcp->write_cb = cb;
-  tcp->write_user_data = arg;
-
-  gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices);
+  tcp->write_slices = slices;
 
-  if (tcp->write_slices.count > GPR_ARRAY_SIZE(local_buffers)) {
-    buffers = (WSABUF *)gpr_malloc(sizeof(WSABUF) * tcp->write_slices.count);
+  if (tcp->write_slices->count > GPR_ARRAY_SIZE(local_buffers)) {
+    buffers = (WSABUF *)gpr_malloc(sizeof(WSABUF) * tcp->write_slices->count);
     allocated = buffers;
   }
 
-  for (i = 0; i < tcp->write_slices.count; i++) {
-    buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices.slices[i]);
-    buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices.slices[i]);
+  for (i = 0; i < tcp->write_slices->count; i++) {
+    buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices->slices[i]);
+    buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices->slices[i]);
   }
 
   /* First, let's try a synchronous, non-blocking write. */
-  status = WSASend(socket->socket, buffers, tcp->write_slices.count,
+  status = WSASend(socket->socket, buffers, tcp->write_slices->count,
                    &bytes_sent, 0, NULL, NULL);
   info->wsa_error = status == 0 ? 0 : WSAGetLastError();
 
@@ -323,10 +316,10 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
      connection that has its send queue filled up. But if we don't, then we can
      avoid doing an async write operation at all. */
   if (info->wsa_error != WSAEWOULDBLOCK) {
-    grpc_endpoint_write_status ret = GRPC_ENDPOINT_WRITE_ERROR;
+    grpc_endpoint_op_status ret = GRPC_ENDPOINT_ERROR;
     if (status == 0) {
-      ret = GRPC_ENDPOINT_WRITE_DONE;
-      GPR_ASSERT(bytes_sent == tcp->write_slices.length);
+      ret = GRPC_ENDPOINT_DONE;
+      GPR_ASSERT(bytes_sent == tcp->write_slices->length);
     } else {
       if (socket->read_info.wsa_error != WSAECONNRESET) {
         char *utf8_message = gpr_format_message(info->wsa_error);
@@ -335,7 +328,6 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
       }
     }
     if (allocated) gpr_free(allocated);
-    gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
     tcp->socket->write_info.outstanding = 0;
     tcp_unref(tcp);
     return ret;
@@ -344,24 +336,23 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
   /* If we got a WSAEWOULDBLOCK earlier, then we need to re-do the same
      operation, this time asynchronously. */
   memset(&socket->write_info.overlapped, 0, sizeof(OVERLAPPED));
-  status = WSASend(socket->socket, buffers, tcp->write_slices.count,
+  status = WSASend(socket->socket, buffers, tcp->write_slices->count,
                    &bytes_sent, 0, &socket->write_info.overlapped, NULL);
   if (allocated) gpr_free(allocated);
 
   if (status != 0) {
     int wsa_error = WSAGetLastError();
     if (wsa_error != WSA_IO_PENDING) {
-      gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
       tcp->socket->write_info.outstanding = 0;
       tcp_unref(tcp);
-      return GRPC_ENDPOINT_WRITE_ERROR;
+      return GRPC_ENDPOINT_ERROR;
     }
   }
 
   /* As all is now setup, we can now ask for the IOCP notification. It may
      trigger the callback immediately however, but no matter. */
   grpc_socket_notify_on_write(socket, on_write, tcp);
-  return GRPC_ENDPOINT_WRITE_PENDING;
+  return GRPC_ENDPOINT_PENDING;
 }
 
 static void win_add_to_pollset(grpc_endpoint *ep, grpc_pollset *ps) {
@@ -407,7 +398,7 @@ static char *win_get_peer(grpc_endpoint *ep) {
 }
 
 static grpc_endpoint_vtable vtable = {
-    win_notify_on_read, win_write,   win_add_to_pollset, win_add_to_pollset_set,
+    win_read, win_write,   win_add_to_pollset, win_add_to_pollset_set,
     win_shutdown,       win_destroy, win_get_peer};
 
 grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) {
@@ -416,7 +407,6 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) {
   tcp->base.vtable = &vtable;
   tcp->socket = socket;
   gpr_mu_init(&tcp->mu);
-  gpr_slice_buffer_init(&tcp->write_slices);
   gpr_ref_init(&tcp->refcount, 1);
   tcp->peer_string = gpr_strdup(peer_string);
   return &tcp->base;