Merge github.com:grpc/grpc into reuse_port

4 files changed, 85 insertions, 25 deletions

diff --git a/test/core/iomgr/endpoint_tests.c b/test/core/iomgr/endpoint_tests.c
index 55e79189cb..b79c22e42a 100644
--- a/test/core/iomgr/endpoint_tests.c
+++ b/test/core/iomgr/endpoint_tests.c
@@ -189,13 +189,15 @@ static void read_and_write_test(grpc_endpoint_test_config config,
   grpc_endpoint_test_fixture f =
       begin_test(config, "read_and_write_test", slice_size);
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  gpr_log(GPR_DEBUG, "num_bytes=%d write_size=%d slice_size=%d shutdown=%d",
+  gpr_log(GPR_DEBUG, "num_bytes=%" PRIuPTR " write_size=%" PRIuPTR
+                     " slice_size=%" PRIuPTR " shutdown=%d",
           num_bytes, write_size, slice_size, shutdown);
 
   if (shutdown) {
     gpr_log(GPR_INFO, "Start read and write shutdown test");
   } else {
-    gpr_log(GPR_INFO, "Start read and write test with %d bytes, slice size %d",
+    gpr_log(GPR_INFO, "Start read and write test with %" PRIuPTR
+                      " bytes, slice size %" PRIuPTR,
             num_bytes, slice_size);
   }
 
@@ -221,7 +223,7 @@ static void read_and_write_test(grpc_endpoint_test_config config,
      even when bytes_written is unsigned. */
   state.bytes_written -= state.current_write_size;
   read_and_write_test_write_handler(&exec_ctx, &state, GRPC_ERROR_NONE);
-  grpc_exec_ctx_finish(&exec_ctx);
+  grpc_exec_ctx_flush(&exec_ctx);
 
   grpc_endpoint_read(&exec_ctx, state.read_ep, &state.incoming,
                      &state.done_read);
@@ -232,7 +234,7 @@ static void read_and_write_test(grpc_endpoint_test_config config,
     gpr_log(GPR_DEBUG, "shutdown write");
     grpc_endpoint_shutdown(&exec_ctx, state.write_ep);
   }
-  grpc_exec_ctx_finish(&exec_ctx);
+  grpc_exec_ctx_flush(&exec_ctx);
 
   gpr_mu_lock(g_mu);
   while (!state.read_done || !state.write_done) {
@@ -244,7 +246,7 @@ static void read_and_write_test(grpc_endpoint_test_config config,
                           gpr_now(GPR_CLOCK_MONOTONIC), deadline)));
   }
   gpr_mu_unlock(g_mu);
-  grpc_exec_ctx_finish(&exec_ctx);
+  grpc_exec_ctx_flush(&exec_ctx);
 
   end_test(config);
   gpr_slice_buffer_destroy(&state.outgoing);
@@ -254,11 +256,67 @@ static void read_and_write_test(grpc_endpoint_test_config config,
   grpc_exec_ctx_finish(&exec_ctx);
 }
 
+static void inc_on_failure(grpc_exec_ctx *exec_ctx, void *arg,
+                           grpc_error *error) {
+  *(int *)arg += (error != GRPC_ERROR_NONE);
+}
+
+static void wait_for_fail_count(grpc_exec_ctx *exec_ctx, int *fail_count,
+                                int want_fail_count) {
+  grpc_exec_ctx_flush(exec_ctx);
+  for (int i = 0; i < 5 && *fail_count < want_fail_count; i++) {
+    grpc_pollset_worker *worker = NULL;
+    gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME);
+    gpr_timespec deadline =
+        gpr_time_add(now, gpr_time_from_seconds(1, GPR_TIMESPAN));
+    gpr_mu_lock(g_mu);
+    GPR_ASSERT(GRPC_LOG_IF_ERROR(
+        "pollset_work",
+        grpc_pollset_work(exec_ctx, g_pollset, &worker, now, deadline)));
+    gpr_mu_unlock(g_mu);
+    grpc_exec_ctx_flush(exec_ctx);
+  }
+  GPR_ASSERT(*fail_count == want_fail_count);
+}
+
+static void multiple_shutdown_test(grpc_endpoint_test_config config) {
+  grpc_endpoint_test_fixture f =
+      begin_test(config, "multiple_shutdown_test", 128);
+  int fail_count = 0;
+
+  gpr_slice_buffer slice_buffer;
+  gpr_slice_buffer_init(&slice_buffer);
+
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_endpoint_add_to_pollset(&exec_ctx, f.client_ep, g_pollset);
+  grpc_endpoint_read(&exec_ctx, f.client_ep, &slice_buffer,
+                     grpc_closure_create(inc_on_failure, &fail_count));
+  wait_for_fail_count(&exec_ctx, &fail_count, 0);
+  grpc_endpoint_shutdown(&exec_ctx, f.client_ep);
+  wait_for_fail_count(&exec_ctx, &fail_count, 1);
+  grpc_endpoint_read(&exec_ctx, f.client_ep, &slice_buffer,
+                     grpc_closure_create(inc_on_failure, &fail_count));
+  wait_for_fail_count(&exec_ctx, &fail_count, 2);
+  gpr_slice_buffer_add(&slice_buffer, gpr_slice_from_copied_string("a"));
+  grpc_endpoint_write(&exec_ctx, f.client_ep, &slice_buffer,
+                      grpc_closure_create(inc_on_failure, &fail_count));
+  wait_for_fail_count(&exec_ctx, &fail_count, 3);
+  grpc_endpoint_shutdown(&exec_ctx, f.client_ep);
+  wait_for_fail_count(&exec_ctx, &fail_count, 3);
+
+  gpr_slice_buffer_destroy(&slice_buffer);
+
+  grpc_endpoint_destroy(&exec_ctx, f.client_ep);
+  grpc_endpoint_destroy(&exec_ctx, f.server_ep);
+  grpc_exec_ctx_finish(&exec_ctx);
+}
+
 void grpc_endpoint_tests(grpc_endpoint_test_config config,
                          grpc_pollset *pollset, gpr_mu *mu) {
   size_t i;
   g_pollset = pollset;
   g_mu = mu;
+  multiple_shutdown_test(config);
   read_and_write_test(config, 10000000, 100000, 8192, false);
   read_and_write_test(config, 1000000, 100000, 1, false);
   read_and_write_test(config, 100000000, 100000, 1, true);
@@ -266,4 +324,5 @@ void grpc_endpoint_tests(grpc_endpoint_test_config config,
     read_and_write_test(config, 40320, i, i, false);
   }
   g_pollset = NULL;
+  g_mu = NULL;
 }
diff --git a/test/core/iomgr/fd_posix_test.c b/test/core/iomgr/fd_posix_test.c
index ab5c279ac6..62dc24d85a 100644
--- a/test/core/iomgr/fd_posix_test.c
+++ b/test/core/iomgr/fd_posix_test.c
@@ -52,6 +52,7 @@
 
 #include "src/core/lib/iomgr/ev_posix.h"
 #include "src/core/lib/iomgr/iomgr.h"
+#include "src/core/lib/iomgr/socket_utils_posix.h"
 #include "test/core/util/test_config.h"
 
 static gpr_mu *g_mu;
@@ -68,17 +69,15 @@ static void create_test_socket(int port, int *socket_fd,
                                struct sockaddr_in *sin) {
   int fd;
   int one = 1;
-  int buf_size = BUF_SIZE;
+  int buffer_size_bytes = BUF_SIZE;
   int flags;
 
   fd = socket(AF_INET, SOCK_STREAM, 0);
   setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
   /* Reset the size of socket send buffer to the minimal value to facilitate
      buffer filling up and triggering notify_on_write  */
-  GPR_ASSERT(
-      setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof(buf_size)) != -1);
-  GPR_ASSERT(
-      setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof(buf_size)) != -1);
+  GPR_ASSERT(grpc_set_socket_sndbuf(fd, buffer_size_bytes) == GRPC_ERROR_NONE);
+  GPR_ASSERT(grpc_set_socket_rcvbuf(fd, buffer_size_bytes) == GRPC_ERROR_NONE);
   /* Make fd non-blocking */
   flags = fcntl(fd, F_GETFL, 0);
   GPR_ASSERT(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);
@@ -405,7 +404,7 @@ static void test_grpc_fd(void) {
   client_wait_and_shutdown(&cl);
   server_wait_and_shutdown(&sv);
   GPR_ASSERT(sv.read_bytes_total == cl.write_bytes_total);
-  gpr_log(GPR_INFO, "Total read bytes %d", sv.read_bytes_total);
+  gpr_log(GPR_INFO, "Total read bytes %" PRIdPTR, sv.read_bytes_total);
 }
 
 typedef struct fd_change_data {
diff --git a/test/core/iomgr/tcp_posix_test.c b/test/core/iomgr/tcp_posix_test.c
index 0a351f7422..42614567ca 100644
--- a/test/core/iomgr/tcp_posix_test.c
+++ b/test/core/iomgr/tcp_posix_test.c
@@ -152,7 +152,7 @@ static void read_cb(grpc_exec_ctx *exec_ctx, void *user_data,
   read_bytes = count_slices(state->incoming.slices, state->incoming.count,
                             &current_data);
   state->read_bytes += read_bytes;
-  gpr_log(GPR_INFO, "Read %d bytes of %d", read_bytes,
+  gpr_log(GPR_INFO, "Read %" PRIuPTR " bytes of %" PRIuPTR, read_bytes,
           state->target_read_bytes);
   if (state->read_bytes >= state->target_read_bytes) {
     gpr_mu_unlock(g_mu);
@@ -171,8 +171,8 @@ static void read_test(size_t num_bytes, size_t slice_size) {
   gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
-  gpr_log(GPR_INFO, "Read test of size %d, slice size %d", num_bytes,
-          slice_size);
+  gpr_log(GPR_INFO, "Read test of size %" PRIuPTR ", slice size %" PRIuPTR,
+          num_bytes, slice_size);
 
   create_sockets(sv);
 
@@ -180,7 +180,7 @@ static void read_test(size_t num_bytes, size_t slice_size) {
   grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset);
 
   written_bytes = fill_socket_partial(sv[0], num_bytes);
-  gpr_log(GPR_INFO, "Wrote %d bytes", written_bytes);
+  gpr_log(GPR_INFO, "Wrote %" PRIuPTR " bytes", written_bytes);
 
   state.ep = ep;
   state.read_bytes = 0;
@@ -219,7 +219,7 @@ static void large_read_test(size_t slice_size) {
   gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
-  gpr_log(GPR_INFO, "Start large read test, slice size %d", slice_size);
+  gpr_log(GPR_INFO, "Start large read test, slice size %" PRIuPTR, slice_size);
 
   create_sockets(sv);
 
@@ -228,7 +228,7 @@ static void large_read_test(size_t slice_size) {
   grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset);
 
   written_bytes = fill_socket(sv[0]);
-  gpr_log(GPR_INFO, "Wrote %d bytes", written_bytes);
+  gpr_log(GPR_INFO, "Wrote %" PRIuPTR " bytes", written_bytes);
 
   state.ep = ep;
   state.read_bytes = 0;
@@ -353,8 +353,9 @@ static void write_test(size_t num_bytes, size_t slice_size) {
   gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
-  gpr_log(GPR_INFO, "Start write test with %d bytes, slice size %d", num_bytes,
-          slice_size);
+  gpr_log(GPR_INFO,
+          "Start write test with %" PRIuPTR " bytes, slice size %" PRIuPTR,
+          num_bytes, slice_size);
 
   create_sockets(sv);
 
@@ -416,8 +417,9 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) {
   int fd_released_done = 0;
   grpc_closure_init(&fd_released_cb, &on_fd_released, &fd_released_done);
 
-  gpr_log(GPR_INFO, "Release fd read_test of size %d, slice size %d", num_bytes,
-          slice_size);
+  gpr_log(GPR_INFO,
+          "Release fd read_test of size %" PRIuPTR ", slice size %" PRIuPTR,
+          num_bytes, slice_size);
 
   create_sockets(sv);
 
@@ -426,7 +428,7 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) {
   grpc_endpoint_add_to_pollset(&exec_ctx, ep, g_pollset);
 
   written_bytes = fill_socket_partial(sv[0], num_bytes);
-  gpr_log(GPR_INFO, "Wrote %d bytes", written_bytes);
+  gpr_log(GPR_INFO, "Wrote %" PRIuPTR " bytes", written_bytes);
 
   state.ep = ep;
   state.read_bytes = 0;
@@ -532,8 +534,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);
diff --git a/test/core/iomgr/workqueue_test.c b/test/core/iomgr/workqueue_test.c
index 9a359cd799..76ecfae74b 100644
--- a/test/core/iomgr/workqueue_test.c
+++ b/test/core/iomgr/workqueue_test.c
@@ -73,7 +73,7 @@ static void test_add_closure(void) {
   grpc_pollset_worker *worker = NULL;
   grpc_closure_init(&c, must_succeed, &done);
 
-  grpc_workqueue_push(&exec_ctx, wq, &c, GRPC_ERROR_NONE);
+  grpc_workqueue_enqueue(&exec_ctx, wq, &c, GRPC_ERROR_NONE);
   grpc_workqueue_add_to_pollset(&exec_ctx, wq, g_pollset);
 
   gpr_mu_lock(g_mu);
@@ -103,7 +103,7 @@ static void test_flush(void) {
   grpc_pollset_worker *worker = NULL;
   grpc_closure_init(&c, must_succeed, &done);
 
-  grpc_exec_ctx_push(&exec_ctx, &c, GRPC_ERROR_NONE, NULL);
+  grpc_exec_ctx_sched(&exec_ctx, &c, GRPC_ERROR_NONE, NULL);
   grpc_workqueue_flush(&exec_ctx, wq);
   grpc_workqueue_add_to_pollset(&exec_ctx, wq, g_pollset);