Merge pull request #3183 from ctiller/second-coming

Refactor Endpoint API

1 files changed, 119 insertions, 143 deletions

diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c
index 901793ec43..fe3673c607 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. */
@@ -98,82 +96,91 @@ typedef struct grpc_tcp {
   char *peer_string;
 } grpc_tcp;
 
-static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); }
+static void tcp_free(grpc_tcp *tcp) {
+  grpc_winsocket_destroy(tcp->socket);
+  gpr_mu_destroy(&tcp->mu);
+  gpr_free(tcp->peer_string);
+  gpr_free(tcp);
+}
 
+/*#define GRPC_TCP_REFCOUNT_DEBUG*/
+#ifdef GRPC_TCP_REFCOUNT_DEBUG
+#define TCP_UNREF(tcp, reason) tcp_unref((tcp), (reason), __FILE__, __LINE__)
+#define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__)
+static void tcp_unref(grpc_tcp *tcp, const char *reason, const char *file,
+  int line) {
+  gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %d -> %d", tcp,
+    reason, tcp->refcount.count, tcp->refcount.count - 1);
+  if (gpr_unref(&tcp->refcount)) {
+    tcp_free(tcp);
+  }
+}
+
+static void tcp_ref(grpc_tcp *tcp, const char *reason, const char *file,
+  int line) {
+  gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP   ref %p : %s %d -> %d", tcp,
+    reason, tcp->refcount.count, tcp->refcount.count + 1);
+  gpr_ref(&tcp->refcount);
+}
+#else
+#define TCP_UNREF(tcp, reason) tcp_unref((tcp))
+#define TCP_REF(tcp, reason) tcp_ref((tcp))
 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);
-    gpr_free(tcp);
+    tcp_free(tcp);
   }
 }
 
+static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); }
+#endif
+
 /* 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 success) {
   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;
   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
-    though, so we're going to do it from here. */
-    do_abort = 1;
-  }
-  gpr_mu_unlock(&tcp->mu);
-
-  if (do_abort) {
-    if (from_iocp) {
-      tcp->socket->read_info.outstanding = 0;
+  if (success) {
+    if (socket->read_info.wsa_error != 0) {
+      if (socket->read_info.wsa_error != WSAECONNRESET) {
+        char *utf8_message = gpr_format_message(info->wsa_error);
+        gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message);
+        gpr_free(utf8_message);
+      }
+      success = 0;
       gpr_slice_unref(tcp->read_slice);
-    }
-    tcp_unref(tcp);
-    if (cb) cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
-    return;
-  }
-
-  GPR_ASSERT(tcp->socket->read_info.outstanding);
-
-  if (socket->read_info.wsa_error != 0) {
-    if (socket->read_info.wsa_error != WSAECONNRESET) {
-      char *utf8_message = gpr_format_message(info->wsa_error);
-      gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message);
-      gpr_free(utf8_message);
-    }
-    gpr_slice_unref(tcp->read_slice);
-    status = GRPC_ENDPOINT_CB_ERROR;
-  } 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;
     } else {
-      gpr_slice_unref(tcp->read_slice);
-      status = GRPC_ENDPOINT_CB_EOF;
+      if (info->bytes_transfered != 0) {
+        sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);
+        gpr_slice_buffer_add(tcp->read_slices, sub);
+        success = 1;
+      } else {
+        gpr_slice_unref(tcp->read_slice);
+        success = 0;
+      }
     }
   }
 
-  tcp->socket->read_info.outstanding = 0;
+  return success;
+}
 
-  tcp_unref(tcp);
-  cb(opaque, slice, nslices, status);
+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, "read");
+  if (cb) {
+    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;
@@ -182,15 +189,13 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
   DWORD flags = 0;
   WSABUF buffer;
 
-  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;
+  gpr_slice_buffer_reset_and_unref(read_slices);
 
   tcp->read_slice = gpr_slice_malloc(8192);
 
@@ -204,12 +209,14 @@ 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) {
+    int ok;
     info->bytes_transfered = bytes_read;
-    /* This might heavily recurse. */
-    on_read(tcp, 1);
-    return;
+    ok = on_read(tcp, 1);
+    return ok ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR;
   }
 
+  TCP_REF(tcp, "read");
+
   /* Otherwise, let's retry, by queuing a read. */
   memset(&tcp->socket->read_info.overlapped, 0, sizeof(OVERLAPPED));
   status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags,
@@ -218,71 +225,51 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
   if (status != 0) {
     int wsa_error = WSAGetLastError();
     if (wsa_error != WSA_IO_PENDING) {
+      int ok;
       info->wsa_error = wsa_error;
-      on_read(tcp, 1);
-      return;
+      ok = on_read(tcp, 1);
+      return ok ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR;
     }
   }
 
-  grpc_socket_notify_on_read(tcp->socket, on_read, 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. */
-static void on_write(void *tcpp, int from_iocp) {
+static void on_write(void *tcpp, int success) {
   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 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
-        though, so we're going to do it from here. */
-    do_abort = 1;
-  }
   gpr_mu_unlock(&tcp->mu);
 
-  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);
-    return;
-  }
-
-  GPR_ASSERT(tcp->socket->write_info.outstanding);
-
-  if (info->wsa_error != 0) {
-    if (info->wsa_error != WSAECONNRESET) {
-      char *utf8_message = gpr_format_message(info->wsa_error);
-      gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
-      gpr_free(utf8_message);
+  if (success) {
+    if (info->wsa_error != 0) {
+      if (info->wsa_error != WSAECONNRESET) {
+        char *utf8_message = gpr_format_message(info->wsa_error);
+        gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
+        gpr_free(utf8_message);
+      }
+      success = 0;
+    } else {
+      GPR_ASSERT(info->bytes_transfered == tcp->write_slices->length);
     }
-    status = GRPC_ENDPOINT_CB_ERROR;
-  } else {
-    GPR_ASSERT(info->bytes_transfered == tcp->write_slices.length);
   }
 
-  gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
-  tcp->socket->write_info.outstanding = 0;
-
-  tcp_unref(tcp);
-  cb(opaque, status);
+  TCP_UNREF(tcp, "write");
+  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;
@@ -293,30 +280,25 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
   WSABUF *allocated = NULL;
   WSABUF *buffers = local_buffers;
 
-  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;
+  tcp->write_slices = slices;
 
-  gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices);
-
-  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();
 
@@ -324,10 +306,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);
@@ -336,33 +318,30 @@ 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;
   }
 
+  TCP_REF(tcp, "write");
+
   /* 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;
+      TCP_UNREF(tcp, "write");
+      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) {
@@ -387,19 +366,17 @@ static void win_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pss) {
    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;
-  extra_refs = grpc_winsocket_shutdown(tcp->socket);
-  while (extra_refs--) tcp_ref(tcp);
+  grpc_winsocket_shutdown(tcp->socket);
   gpr_mu_unlock(&tcp->mu);
 }
 
 static void win_destroy(grpc_endpoint *ep) {
   grpc_tcp *tcp = (grpc_tcp *)ep;
-  tcp_unref(tcp);
+  TCP_UNREF(tcp, "destroy");
 }
 
 static char *win_get_peer(grpc_endpoint *ep) {
@@ -408,8 +385,8 @@ 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_shutdown,       win_destroy, win_get_peer};
+    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) {
   grpc_tcp *tcp = (grpc_tcp *)gpr_malloc(sizeof(grpc_tcp));
@@ -417,7 +394,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;