Make {endpoint,fd}_shutdown idempotent

src/core/lib/iomgr/endpoint.h

@@ -82,7 +82,7 @@ char *grpc_endpoint_get_peer(grpc_endpoint *ep);
 void grpc_endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                          gpr_slice_buffer *slices, grpc_closure *cb);
 
-/* Causes any pending read/write callbacks to run immediately with
+/* Causes any pending and future read/write callbacks to run immediately with
    success==0 */
 void grpc_endpoint_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep);
 void grpc_endpoint_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep);

src/core/lib/iomgr/ev_poll_and_epoll_posix.c

@@ -515,7 +515,9 @@ static void fd_unref(grpc_fd *fd) { unref_by(fd, 2); }
 
 static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
                              grpc_closure **st, grpc_closure *closure) {
-  if (*st == CLOSURE_NOT_READY) {
+  if (fd->shutdown) {
+    grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
+  } else if (*st == CLOSURE_NOT_READY) {
     /* not ready ==> switch to a waiting state by setting the closure */
     *st = closure;
   } else if (*st == CLOSURE_READY) {
@@ -557,13 +559,24 @@ static void set_read_notifier_pollset_locked(
 
 static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
   gpr_mu_lock(&fd->mu);
-  GPR_ASSERT(!fd->shutdown);
+  /* only shutdown once */
+  if (!fd->shutdown) {
     fd->shutdown = 1;
+    /* signal read/write closed to OS so that future operations fail */
+    shutdown(fd->fd, SHUT_RDWR);
     set_ready_locked(exec_ctx, fd, &fd->read_closure);
     set_ready_locked(exec_ctx, fd, &fd->write_closure);
+  }
   gpr_mu_unlock(&fd->mu);
 }
 
+static bool fd_is_shutdown(grpc_fd *fd) {
+  gpr_mu_lock(&fd->mu);
+  bool r = fd->shutdown;
+  gpr_mu_unlock(&fd->mu);
+  return r;
+}
+
 static void fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
                               grpc_closure *closure) {
   gpr_mu_lock(&fd->mu);
@@ -1938,6 +1951,7 @@ static const grpc_event_engine_vtable vtable = {
     .fd_wrapped_fd = fd_wrapped_fd,
     .fd_orphan = fd_orphan,
     .fd_shutdown = fd_shutdown,
+    .fd_is_shutdown = fd_is_shutdown,
     .fd_notify_on_read = fd_notify_on_read,
     .fd_notify_on_write = fd_notify_on_write,
     .fd_get_read_notifier_pollset = fd_get_read_notifier_pollset,

src/core/lib/iomgr/ev_poll_posix.c

@@ -421,7 +421,9 @@ static void fd_unref(grpc_fd *fd) { unref_by(fd, 2); }
 
 static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
                              grpc_closure **st, grpc_closure *closure) {
-  if (*st == CLOSURE_NOT_READY) {
+  if (fd->shutdown) {
+    grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
+  } else if (*st == CLOSURE_NOT_READY) {
     /* not ready ==> switch to a waiting state by setting the closure */
     *st = closure;
   } else if (*st == CLOSURE_READY) {
@@ -463,13 +465,24 @@ static void set_read_notifier_pollset_locked(
 
 static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
   gpr_mu_lock(&fd->mu);
-  GPR_ASSERT(!fd->shutdown);
+  /* only shutdown once */
+  if (!fd->shutdown) {
     fd->shutdown = 1;
+    /* signal read/write closed to OS so that future operations fail */
+    shutdown(fd->fd, SHUT_RDWR);
     set_ready_locked(exec_ctx, fd, &fd->read_closure);
     set_ready_locked(exec_ctx, fd, &fd->write_closure);
+  }
   gpr_mu_unlock(&fd->mu);
 }
 
+static bool fd_is_shutdown(grpc_fd *fd) {
+  gpr_mu_lock(&fd->mu);
+  bool r = fd->shutdown;
+  gpr_mu_unlock(&fd->mu);
+  return r;
+}
+
 static void fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
                               grpc_closure *closure) {
   gpr_mu_lock(&fd->mu);
@@ -1172,6 +1185,7 @@ static const grpc_event_engine_vtable vtable = {
     .fd_wrapped_fd = fd_wrapped_fd,
     .fd_orphan = fd_orphan,
     .fd_shutdown = fd_shutdown,
+    .fd_is_shutdown = fd_is_shutdown,
     .fd_notify_on_read = fd_notify_on_read,
     .fd_notify_on_write = fd_notify_on_write,
     .fd_get_read_notifier_pollset = fd_get_read_notifier_pollset,

src/core/lib/iomgr/ev_posix.c

@@ -153,6 +153,10 @@ void grpc_fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
   g_event_engine->fd_shutdown(exec_ctx, fd);
 }
 
+bool grpc_fd_is_shutdown(grpc_fd *fd) {
+  return g_event_engine->fd_is_shutdown(fd);
+}
+
 void grpc_fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
                             grpc_closure *closure) {
   g_event_engine->fd_notify_on_read(exec_ctx, fd, closure);

src/core/lib/iomgr/ev_posix.h

@@ -55,6 +55,7 @@ typedef struct grpc_event_engine_vtable {
                             grpc_closure *closure);
   void (*fd_notify_on_write)(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
                              grpc_closure *closure);
+  bool (*fd_is_shutdown)(grpc_fd *fd);
   grpc_pollset *(*fd_get_read_notifier_pollset)(grpc_exec_ctx *exec_ctx,
                                                 grpc_fd *fd);
 
@@ -116,7 +117,10 @@ int grpc_fd_wrapped_fd(grpc_fd *fd);
 void grpc_fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure *on_done,
                     int *release_fd, const char *reason);
 
-/* Cause any current callbacks to error out with GRPC_CALLBACK_CANCELLED. */
+/* Has grpc_fd_shutdown been called on an fd? */
+bool grpc_fd_is_shutdown(grpc_fd *fd);
+
+/* Cause any current and future callbacks to fail. */
 void grpc_fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd);
 
 /* Register read interest, causing read_cb to be called once when fd becomes

src/core/lib/iomgr/tcp_posix.c

@@ -408,7 +408,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
 
   if (buf->length == 0) {
     GPR_TIMER_END("tcp_write", 0);
-    grpc_exec_ctx_enqueue(exec_ctx, cb, true, NULL);
+    grpc_exec_ctx_enqueue(exec_ctx, cb, !grpc_fd_is_shutdown(tcp->em_fd), NULL);
     return;
   }
   tcp->outgoing_buffer = buf;

test/core/iomgr/endpoint_tests.c

@@ -253,35 +253,39 @@ static void read_and_write_test(grpc_endpoint_test_config config,
   grpc_exec_ctx_finish(&exec_ctx);
 }
 
-static void must_fail(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
+static void inc_on_failure(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
-  GPR_ASSERT(!success);
+  *(int *)arg += (success == false);
-  ++*(int *)arg;
 }
 
 static void multiple_shutdown_test(grpc_endpoint_test_config config) {
-  grpc_endpoint_test_fixture f = begin_test(config, "read_and_write_test", 128);
+  grpc_endpoint_test_fixture f =
+      begin_test(config, "multiple_shutdown_test", 128);
   int fail_count = 0;
 
-  gpr_slice_buffer incoming;
+  gpr_slice_buffer slice_buffer;
-  gpr_slice_buffer_init(&incoming);
+  gpr_slice_buffer_init(&slice_buffer);
 
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  grpc_endpoint_read(&exec_ctx, f.client_ep, &incoming,
+  grpc_endpoint_read(&exec_ctx, f.client_ep, &slice_buffer,
-                     grpc_closure_create(must_fail, &fail_count));
+                     grpc_closure_create(inc_on_failure, &fail_count));
   grpc_exec_ctx_flush(&exec_ctx);
   GPR_ASSERT(fail_count == 0);
   grpc_endpoint_shutdown(&exec_ctx, f.client_ep);
   grpc_exec_ctx_flush(&exec_ctx);
   GPR_ASSERT(fail_count == 1);
-  grpc_endpoint_read(&exec_ctx, f.client_ep, &incoming,
+  grpc_endpoint_read(&exec_ctx, f.client_ep, &slice_buffer,
-                     grpc_closure_create(must_fail, &fail_count));
+                     grpc_closure_create(inc_on_failure, &fail_count));
   grpc_exec_ctx_flush(&exec_ctx);
   GPR_ASSERT(fail_count == 2);
+  grpc_endpoint_write(&exec_ctx, f.client_ep, &slice_buffer,
+                      grpc_closure_create(inc_on_failure, &fail_count));
+  grpc_exec_ctx_flush(&exec_ctx);
+  GPR_ASSERT(fail_count == 3);
   grpc_endpoint_shutdown(&exec_ctx, f.client_ep);
   grpc_exec_ctx_flush(&exec_ctx);
-  GPR_ASSERT(fail_count == 2);
+  GPR_ASSERT(fail_count == 3);
 
-  gpr_slice_buffer_destroy(&incoming);
+  gpr_slice_buffer_destroy(&slice_buffer);
 
   grpc_endpoint_destroy(&exec_ctx, f.client_ep);
   grpc_endpoint_destroy(&exec_ctx, f.server_ep);
@@ -293,12 +297,12 @@ void grpc_endpoint_tests(grpc_endpoint_test_config config,
   size_t i;
   g_pollset = pollset;
   g_mu = mu;
+  multiple_shutdown_test(config);
   read_and_write_test(config, 10000000, 100000, 8192, 0);
   read_and_write_test(config, 1000000, 100000, 1, 0);
   read_and_write_test(config, 100000000, 100000, 1, 1);
   for (i = 1; i < 1000; i = GPR_MAX(i + 1, i * 5 / 4)) {
     read_and_write_test(config, 40320, i, i, 0);
   }
-  multiple_shutdown_test(config);
   g_pollset = NULL;
 }

test/core/iomgr/tcp_posix_test.c

@@ -517,8 +517,8 @@ int main(int argc, char **argv) {
   grpc_init();
   g_pollset = gpr_malloc(grpc_pollset_size());
   grpc_pollset_init(g_pollset, &g_mu);
-  run_tests();
   grpc_endpoint_tests(configs[0], g_pollset, g_mu);
+  run_tests();
   grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
   grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
   grpc_exec_ctx_finish(&exec_ctx);