Merge branch 'just-say-goodbye-when-we-are-done' into screw-you-guys-im-taking-my-own-lock

10 files changed, 54 insertions, 28 deletions

diff --git a/src/core/client_config/README.md b/src/core/client_config/README.md
index 7cb19cd130..d7aed27223 100644
--- a/src/core/client_config/README.md
+++ b/src/core/client_config/README.md
@@ -42,3 +42,19 @@ Their behavior is specified by a set of grpc channel filters defined at their
 construction. To customize this behavior, resolvers build grpc_subchannel_factory 
 objects, which use the decorator pattern to customize construction arguments for 
 concrete grpc_subchannel instances.
+
+
+Naming for GRPC
+===============
+
+Names in GRPC are represented by a URI.
+
+The following schemes are currently supported:
+
+dns:///host:port - dns schemes are currently supported so long as authority is
+                   empty (authority based dns resolution is expected in a future
+                   release)
+
+unix:path        - the unix scheme is used to create and connect to unix domain 
+                   sockets - the authority must be empty, and the path represents
+                   the absolute or relative path to the desired socket
diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c
index 3d57e3136a..73da624aff 100644
--- a/src/core/client_config/lb_policies/pick_first.c
+++ b/src/core/client_config/lb_policies/pick_first.c
@@ -155,8 +155,6 @@ loop:
   switch (p->checking_connectivity) {
     case GRPC_CHANNEL_READY:
       p->selected = p->subchannels[p->checking_subchannel];
-      GPR_ASSERT(grpc_subchannel_check_connectivity(p->selected) ==
-                 GRPC_CHANNEL_READY);
       while ((pp = p->pending_picks)) {
         p->pending_picks = pp->next;
         *pp->target = p->selected;
@@ -185,6 +183,7 @@ loop:
       GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel],
                p->subchannels[p->num_subchannels - 1]);
       p->num_subchannels--;
+      GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first");
       if (p->num_subchannels == 0) {
         while ((pp = p->pending_picks)) {
           p->pending_picks = pp->next;
@@ -197,7 +196,6 @@ loop:
         p->checking_subchannel %= p->num_subchannels;
         p->checking_connectivity = grpc_subchannel_check_connectivity(
             p->subchannels[p->checking_subchannel]);
-        GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first");
         add_interested_parties_locked(p);
         goto loop;
       }
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index 15ed8e75e6..12496440de 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -249,7 +249,8 @@ static void basic_do_promote(void *args, int success) {
   pollset->in_flight_cbs--;
   if (pollset->shutting_down) {
     /* We don't care about this pollset anymore. */
-    if (pollset->in_flight_cbs == 0 && pollset->counter == 0) {
+    if (pollset->in_flight_cbs == 0 && pollset->counter == 0 && !pollset->called_shutdown) {
+      pollset->called_shutdown = 1;
       do_shutdown_cb = 1;
     }
   } else if (grpc_fd_is_orphaned(fd)) {
diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c
index 6a99324da6..8a7ada07af 100644
--- a/src/core/security/server_secure_chttp2.c
+++ b/src/core/security/server_secure_chttp2.c
@@ -99,9 +99,10 @@ static void on_secure_transport_setup_done(void *statep,
     if (!state->is_shutdown) {
       mdctx = grpc_mdctx_create();
       transport = grpc_create_chttp2_transport(
-          grpc_server_get_channel_args(state->server), secure_endpoint, NULL, 0,
-          mdctx, 0);
+          grpc_server_get_channel_args(state->server), secure_endpoint, mdctx,
+          0);
       setup_transport(state, transport, mdctx);
+      grpc_chttp2_transport_start_reading(transport, NULL, 0);
     } else {
       /* We need to consume this here, because the server may already have gone
        * away. */
diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c
index 09b4fb782b..e205f0a9f8 100644
--- a/src/core/surface/channel_create.c
+++ b/src/core/surface/channel_create.c
@@ -72,7 +72,8 @@ static void connected(void *arg, grpc_endpoint *tcp) {
   grpc_iomgr_closure *notify;
   if (tcp != NULL) {
     c->result->transport = grpc_create_chttp2_transport(
-        c->args.channel_args, tcp, NULL, 0, c->args.metadata_context, 1);
+        c->args.channel_args, tcp, c->args.metadata_context, 1);
+    grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0);
     GPR_ASSERT(c->result->transport);
     c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *));
     c->result->filters[0] = &grpc_http_client_filter;
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c
index 76fc862621..34ee3f8400 100644
--- a/src/core/surface/secure_channel_create.c
+++ b/src/core/surface/secure_channel_create.c
@@ -82,9 +82,9 @@ static void on_secure_transport_setup_done(void *arg,
     gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status);
     memset(c->result, 0, sizeof(*c->result));
   } else {
-    c->result->transport =
-        grpc_create_chttp2_transport(c->args.channel_args, secure_endpoint,
-                                     NULL, 0, c->args.metadata_context, 1);
+    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);
     c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *) * 2);
     c->result->filters[0] = &grpc_client_auth_filter;
     c->result->filters[1] = &grpc_http_client_filter;
diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c
index 9c02c3ef29..78c53466b3 100644
--- a/src/core/surface/server_chttp2.c
+++ b/src/core/surface/server_chttp2.c
@@ -61,8 +61,9 @@ static void new_transport(void *server, grpc_endpoint *tcp) {
    */
   grpc_mdctx *mdctx = grpc_mdctx_create();
   grpc_transport *transport = grpc_create_chttp2_transport(
-      grpc_server_get_channel_args(server), tcp, NULL, 0, mdctx, 0);
+      grpc_server_get_channel_args(server), tcp, mdctx, 0);
   setup_transport(server, transport, mdctx);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
 
 /* Server callback: start listening on our ports */
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index fdcc300099..a78654334e 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -97,12 +97,8 @@ int grpc_chttp2_unlocking_check_writes(
       grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
     }
 
-    /* we should either exhaust window or have no ops left, but not both */
-    if (stream_global->outgoing_sopb->nops == 0) {
-      stream_global->outgoing_sopb = NULL;
-      grpc_chttp2_schedule_closure(transport_global,
-                                   stream_global->send_done_closure, 1);
-    } else if (stream_global->outgoing_window > 0) {
+    if (stream_global->outgoing_window > 0 &&
+        stream_global->outgoing_sopb->nops != 0) {
       grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
     }
   }
@@ -201,6 +197,11 @@ void grpc_chttp2_cleanup_writing(
 
   while (grpc_chttp2_list_pop_written_stream(
       transport_global, transport_writing, &stream_global, &stream_writing)) {
+    if (stream_global->outgoing_sopb->nops == 0) {
+      stream_global->outgoing_sopb = NULL;
+      grpc_chttp2_schedule_closure(transport_global,
+                                   stream_global->send_done_closure, 1);
+    }
     if (stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
       stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE;
       if (!transport_global->is_client) {
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 391577c948..cb7b3e1548 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -201,8 +201,8 @@ static void ref_transport(grpc_chttp2_transport *t) { gpr_ref(&t->refs); }
 
 static void init_transport(grpc_chttp2_transport *t,
                            const grpc_channel_args *channel_args,
-                           grpc_endpoint *ep, gpr_slice *slices, size_t nslices,
-                           grpc_mdctx *mdctx, int is_client) {
+                           grpc_endpoint *ep, grpc_mdctx *mdctx,
+                           int is_client) {
   size_t i;
   int j;
 
@@ -311,9 +311,6 @@ static void init_transport(grpc_chttp2_transport *t,
       }
     }
   }
-
-  REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */
-  recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK);
 }
 
 static void destroy_transport(grpc_transport *gt) {
@@ -690,7 +687,7 @@ static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) {
     grpc_chttp2_goaway_append(
         t->global.last_incoming_stream_id,
         grpc_chttp2_grpc_status_to_http2_error(op->goaway_status),
-        *op->goaway_message, &t->global.qbuf);
+        gpr_slice_ref(*op->goaway_message), &t->global.qbuf);
     if (!grpc_chttp2_has_streams(t)) {
       close_transport_locked(t);
     }
@@ -1052,9 +1049,16 @@ static const grpc_transport_vtable vtable = {
     perform_transport_op,       destroy_stream, destroy_transport};
 
 grpc_transport *grpc_create_chttp2_transport(
-    const grpc_channel_args *channel_args, grpc_endpoint *ep, gpr_slice *slices,
-    size_t nslices, grpc_mdctx *mdctx, int is_client) {
+    const grpc_channel_args *channel_args, grpc_endpoint *ep, grpc_mdctx *mdctx,
+    int is_client) {
   grpc_chttp2_transport *t = gpr_malloc(sizeof(grpc_chttp2_transport));
-  init_transport(t, channel_args, ep, slices, nslices, mdctx, is_client);
+  init_transport(t, channel_args, ep, mdctx, is_client);
   return &t->base;
 }
+
+void grpc_chttp2_transport_start_reading(grpc_transport *transport,
+                                         gpr_slice *slices, size_t nslices) {
+  grpc_chttp2_transport *t = (grpc_chttp2_transport *)transport;
+  REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */
+  recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK);
+}
diff --git a/src/core/transport/chttp2_transport.h b/src/core/transport/chttp2_transport.h
index 1747792b95..fa0d6e4151 100644
--- a/src/core/transport/chttp2_transport.h
+++ b/src/core/transport/chttp2_transport.h
@@ -41,7 +41,10 @@ extern int grpc_http_trace;
 extern int grpc_flowctl_trace;
 
 grpc_transport *grpc_create_chttp2_transport(
-    const grpc_channel_args *channel_args, grpc_endpoint *ep, gpr_slice *slices,
-    size_t nslices, grpc_mdctx *metadata_context, int is_client);
+    const grpc_channel_args *channel_args, grpc_endpoint *ep,
+    grpc_mdctx *metadata_context, int is_client);
+
+void grpc_chttp2_transport_start_reading(grpc_transport *transport,
+                                         gpr_slice *slices, size_t nslices);
 
 #endif /* GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_TRANSPORT_H */