Merge pull request #3338 from ctiller/bettererer-ports

Better port selection

11 files changed, 64 insertions, 12 deletions

diff --git a/src/core/client_config/connector.c b/src/core/client_config/connector.c
index a8cd5fc149..c1e583e4a5 100644
--- a/src/core/client_config/connector.c
+++ b/src/core/client_config/connector.c
@@ -47,3 +47,7 @@ void grpc_connector_connect(grpc_connector *connector,
                             grpc_iomgr_closure *notify) {
   connector->vtable->connect(connector, in_args, out_args, notify);
 }
+
+void grpc_connector_shutdown(grpc_connector *connector) {
+  connector->vtable->shutdown(connector);
+}
diff --git a/src/core/client_config/connector.h b/src/core/client_config/connector.h
index 39f3467990..01aa716412 100644
--- a/src/core/client_config/connector.h
+++ b/src/core/client_config/connector.h
@@ -70,6 +70,9 @@ typedef struct {
 struct grpc_connector_vtable {
   void (*ref)(grpc_connector *connector);
   void (*unref)(grpc_connector *connector);
+  /** Implementation of grpc_connector_shutdown */
+  void (*shutdown)(grpc_connector *connector);
+  /** Implementation of grpc_connector_connect */
   void (*connect)(grpc_connector *connector,
                   const grpc_connect_in_args *in_args,
                   grpc_connect_out_args *out_args, grpc_iomgr_closure *notify);
@@ -77,9 +80,12 @@ struct grpc_connector_vtable {
 
 void grpc_connector_ref(grpc_connector *connector);
 void grpc_connector_unref(grpc_connector *connector);
+/** Connect using the connector: max one outstanding call at a time */
 void grpc_connector_connect(grpc_connector *connector,
                             const grpc_connect_in_args *in_args,
                             grpc_connect_out_args *out_args,
                             grpc_iomgr_closure *notify);
+/** Cancel any pending connection */
+void grpc_connector_shutdown(grpc_connector *connector);
 
 #endif
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index ca52c75beb..876d2aa418 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -439,6 +439,10 @@ void grpc_subchannel_process_transport_op(grpc_subchannel *c,
   if (cancel_alarm) {
     grpc_alarm_cancel(&c->alarm);
   }
+
+  if (op->disconnect) {
+    grpc_connector_shutdown(c->connector);
+  }
 }
 
 static void on_state_changed(void *p, int iomgr_success) {
diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h
index 2e36c69134..29a0b67d2d 100644
--- a/src/core/client_config/subchannel.h
+++ b/src/core/client_config/subchannel.h
@@ -43,6 +43,7 @@ typedef struct grpc_subchannel grpc_subchannel;
 typedef struct grpc_subchannel_call grpc_subchannel_call;
 typedef struct grpc_subchannel_args grpc_subchannel_args;
 
+#define GRPC_SUBCHANNEL_REFCOUNT_DEBUG
 #ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG
 #define GRPC_SUBCHANNEL_REF(p, r) \
   grpc_subchannel_ref((p), __FILE__, __LINE__, (r))
diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c
index 2d08a77a70..38a543e36e 100644
--- a/src/core/iomgr/fd_posix.c
+++ b/src/core/iomgr/fd_posix.c
@@ -213,10 +213,9 @@ void grpc_fd_orphan(grpc_fd *fd, grpc_iomgr_closure *on_done,
                     const char *reason) {
   fd->on_done_closure = on_done;
   shutdown(fd->fd, SHUT_RDWR);
-  REF_BY(fd, 1, reason); /* remove active status, but keep referenced */
   gpr_mu_lock(&fd->watcher_mu);
+  REF_BY(fd, 1, reason); /* remove active status, but keep referenced */
   if (!has_watchers(fd)) {
-    GPR_ASSERT(!fd->closed);
     fd->closed = 1;
     close(fd->fd);
     if (fd->on_done_closure) {
diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
index 1dd03992ae..d6ca5d1f71 100644
--- a/src/core/iomgr/iomgr.c
+++ b/src/core/iomgr/iomgr.c
@@ -34,16 +34,18 @@
 #include "src/core/iomgr/iomgr.h"
 
 #include <stdlib.h>
+#include <string.h>
 
-#include "src/core/iomgr/iomgr_internal.h"
-#include "src/core/iomgr/alarm_internal.h"
-#include "src/core/support/string.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include <grpc/support/sync.h>
 #include <grpc/support/thd.h>
 
+#include "src/core/iomgr/iomgr_internal.h"
+#include "src/core/iomgr/alarm_internal.h"
+#include "src/core/support/string.h"
+
 static gpr_mu g_mu;
 static gpr_cv g_rcv;
 static grpc_iomgr_closure *g_cbs_head = NULL;
@@ -179,6 +181,8 @@ void grpc_iomgr_shutdown(void) {
   }
   gpr_mu_unlock(&g_mu);
 
+  memset(&g_root_object, 0, sizeof(g_root_object));
+
   grpc_kick_poller();
   gpr_event_wait(&g_background_callback_executor_done,
                  gpr_inf_future(GPR_CLOCK_REALTIME));
diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c
index 8f62ce2954..481bdc4ede 100644
--- a/src/core/iomgr/pollset_multipoller_with_epoll.c
+++ b/src/core/iomgr/pollset_multipoller_with_epoll.c
@@ -72,7 +72,7 @@ static void finally_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
      to this pollset whilst adding, but that should be benign. */
   GPR_ASSERT(grpc_fd_begin_poll(fd, pollset, 0, 0, &watcher) == 0);
   if (watcher.fd != NULL) {
-    ev.events = EPOLLIN | EPOLLOUT | EPOLLET;
+    ev.events = (uint32_t)(EPOLLIN | EPOLLOUT | EPOLLET);
     ev.data.ptr = fd;
     err = epoll_ctl(h->epoll_fd, EPOLL_CTL_ADD, fd->fd, &ev);
     if (err < 0) {
diff --git a/src/core/support/string.c b/src/core/support/string.c
index af0389ea83..e0ffeb8a4a 100644
--- a/src/core/support/string.c
+++ b/src/core/support/string.c
@@ -101,7 +101,7 @@ static void asciidump(dump_out *out, const char *buf, size_t len) {
     dump_out_append(out, '\'');
   }
   for (cur = beg; cur != end; ++cur) {
-    dump_out_append(out, isprint(*cur) ? *(char *)cur : '.');
+    dump_out_append(out, (char)(isprint(*cur) ? *(char *)cur : '.'));
   }
   if (!out_was_empty) {
     dump_out_append(out, '\'');
diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c
index 9e2cf1cf66..d323d0d74f 100644
--- a/src/core/surface/channel_create.c
+++ b/src/core/surface/channel_create.c
@@ -88,6 +88,8 @@ static void connected(void *arg, grpc_endpoint *tcp) {
   grpc_iomgr_add_callback(notify);
 }
 
+static void connector_shutdown(grpc_connector *con) {}
+
 static void connector_connect(grpc_connector *con,
                               const grpc_connect_in_args *args,
                               grpc_connect_out_args *result,
@@ -103,7 +105,7 @@ static void connector_connect(grpc_connector *con,
 }
 
 static const grpc_connector_vtable connector_vtable = {
-    connector_ref, connector_unref, connector_connect};
+    connector_ref, connector_unref, connector_shutdown, connector_connect};
 
 typedef struct {
   grpc_subchannel_factory base;
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c
index 9b554eeb70..52c5e93988 100644
--- a/src/core/surface/secure_channel_create.c
+++ b/src/core/surface/secure_channel_create.c
@@ -61,6 +61,9 @@ typedef struct {
   grpc_iomgr_closure *notify;
   grpc_connect_in_args args;
   grpc_connect_out_args *result;
+
+  gpr_mu mu;
+  grpc_endpoint *connecting_endpoint;
 } connector;
 
 static void connector_ref(grpc_connector *con) {
@@ -81,10 +84,20 @@ static void on_secure_transport_setup_done(void *arg,
                                            grpc_endpoint *secure_endpoint) {
   connector *c = arg;
   grpc_iomgr_closure *notify;
-  if (status != GRPC_SECURITY_OK) {
+  gpr_mu_lock(&c->mu);
+  if (c->connecting_endpoint == NULL) {
+    memset(c->result, 0, sizeof(*c->result));
+    gpr_mu_unlock(&c->mu);
+  } else if (status != GRPC_SECURITY_OK) {
+    GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint);
     gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status);
     memset(c->result, 0, sizeof(*c->result));
+    c->connecting_endpoint = NULL;
+    gpr_mu_unlock(&c->mu);
   } else {
+    GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint);
+    c->connecting_endpoint = NULL;
+    gpr_mu_unlock(&c->mu);
     c->result->transport = grpc_create_chttp2_transport(
         c->args.channel_args, secure_endpoint, c->args.metadata_context, 1);
     grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0);
@@ -102,6 +115,10 @@ static void connected(void *arg, grpc_endpoint *tcp) {
   connector *c = arg;
   grpc_iomgr_closure *notify;
   if (tcp != NULL) {
+    gpr_mu_lock(&c->mu);
+    GPR_ASSERT(c->connecting_endpoint == NULL);
+    c->connecting_endpoint = tcp;
+    gpr_mu_unlock(&c->mu);
     grpc_setup_secure_transport(&c->security_connector->base, tcp,
                                 on_secure_transport_setup_done, c);
   } else {
@@ -112,6 +129,18 @@ static void connected(void *arg, grpc_endpoint *tcp) {
   }
 }
 
+static void connector_shutdown(grpc_connector *con) {
+  connector *c = (connector *)con;
+  grpc_endpoint *ep;
+  gpr_mu_lock(&c->mu);
+  ep = c->connecting_endpoint;
+  c->connecting_endpoint = NULL;
+  gpr_mu_unlock(&c->mu);
+  if (ep) {
+    grpc_endpoint_shutdown(ep);
+  }
+}
+
 static void connector_connect(grpc_connector *con,
                               const grpc_connect_in_args *args,
                               grpc_connect_out_args *result,
@@ -122,12 +151,15 @@ static void connector_connect(grpc_connector *con,
   c->notify = notify;
   c->args = *args;
   c->result = result;
+  gpr_mu_lock(&c->mu);
+  GPR_ASSERT(c->connecting_endpoint == NULL);
+  gpr_mu_unlock(&c->mu);
   grpc_tcp_client_connect(connected, c, args->interested_parties, args->addr,
                           args->addr_len, args->deadline);
 }
 
 static const grpc_connector_vtable connector_vtable = {
-    connector_ref, connector_unref, connector_connect};
+    connector_ref, connector_unref, connector_shutdown, connector_connect};
 
 typedef struct {
   grpc_subchannel_factory base;
diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c
index a592ce7d28..f26f446787 100644
--- a/src/core/transport/chttp2/parsing.c
+++ b/src/core/transport/chttp2/parsing.c
@@ -486,7 +486,7 @@ static int init_skip_frame_parser(
     transport_parsing->hpack_parser.on_header_user_data = NULL;
     transport_parsing->hpack_parser.is_boundary = is_eoh;
     transport_parsing->hpack_parser.is_eof =
-        is_eoh ? transport_parsing->header_eof : 0;
+        (gpr_uint8)(is_eoh ? transport_parsing->header_eof : 0);
   } else {
     transport_parsing->parser = skip_parser;
   }
@@ -696,7 +696,7 @@ static int init_header_frame_parser(
   transport_parsing->hpack_parser.on_header_user_data = transport_parsing;
   transport_parsing->hpack_parser.is_boundary = is_eoh;
   transport_parsing->hpack_parser.is_eof =
-      is_eoh ? transport_parsing->header_eof : 0;
+      (gpr_uint8)(is_eoh ? transport_parsing->header_eof : 0);
   if (!is_continuation && (transport_parsing->incoming_frame_flags &
                            GRPC_CHTTP2_FLAG_HAS_PRIORITY)) {
     grpc_chttp2_hpack_parser_set_has_priority(&transport_parsing->hpack_parser);