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

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

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;
       }

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)) {

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. */

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;

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;

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 */

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) {

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);
+}

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 */

test/core/bad_client/bad_client.c

@@ -108,8 +108,9 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
   a.validator = validator;
   grpc_server_register_completion_queue(a.server, a.cq);
   grpc_server_start(a.server);
-  transport = grpc_create_chttp2_transport(NULL, sfd.server, NULL, 0, mdctx, 0);
+  transport = grpc_create_chttp2_transport(NULL, sfd.server, mdctx, 0);
   server_setup_transport(&a, transport, mdctx);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 
   /* Bind everything into the same pollset */
   grpc_endpoint_add_to_pollset(sfd.client, grpc_cq_pollset(a.cq));

test/core/end2end/fixtures/chttp2_socket_pair.c

@@ -108,10 +108,10 @@ static void chttp2_init_client_socketpair(grpc_end2end_test_fixture *f,
   sp_client_setup cs;
   cs.client_args = client_args;
   cs.f = f;
-  transport =
-      grpc_create_chttp2_transport(client_args, sfd->client, NULL, 0, mdctx, 1);
+  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx, 1);
   client_setup_transport(&cs, transport, mdctx);
   GPR_ASSERT(f->client);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
 
 static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
@@ -123,9 +123,9 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
   f->server = grpc_server_create_from_filters(NULL, 0, server_args);
   grpc_server_register_completion_queue(f->server, f->cq);
   grpc_server_start(f->server);
-  transport =
-      grpc_create_chttp2_transport(server_args, sfd->server, NULL, 0, mdctx, 0);
+  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx, 0);
   server_setup_transport(f, transport, mdctx);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
 
 static void chttp2_tear_down_socketpair(grpc_end2end_test_fixture *f) {

test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c

@@ -108,10 +108,10 @@ static void chttp2_init_client_socketpair(grpc_end2end_test_fixture *f,
   sp_client_setup cs;
   cs.client_args = client_args;
   cs.f = f;
-  transport =
-      grpc_create_chttp2_transport(client_args, sfd->client, NULL, 0, mdctx, 1);
+  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx, 1);
   client_setup_transport(&cs, transport, mdctx);
   GPR_ASSERT(f->client);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
 
 static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
@@ -123,9 +123,9 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
   f->server = grpc_server_create_from_filters(NULL, 0, server_args);
   grpc_server_register_completion_queue(f->server, f->cq);
   grpc_server_start(f->server);
-  transport =
-      grpc_create_chttp2_transport(server_args, sfd->server, NULL, 0, mdctx, 0);
+  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx, 0);
   server_setup_transport(f, transport, mdctx);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
 
 static void chttp2_tear_down_socketpair(grpc_end2end_test_fixture *f) {

test/core/end2end/fixtures/chttp2_socket_pair_with_grpc_trace.c

@@ -109,10 +109,10 @@ static void chttp2_init_client_socketpair(grpc_end2end_test_fixture *f,
   sp_client_setup cs;
   cs.client_args = client_args;
   cs.f = f;
-  transport =
-      grpc_create_chttp2_transport(client_args, sfd->client, NULL, 0, mdctx, 1);
+  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx, 1);
   client_setup_transport(&cs, transport, mdctx);
   GPR_ASSERT(f->client);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
 
 static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
@@ -124,9 +124,9 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
   f->server = grpc_server_create_from_filters(NULL, 0, server_args);
   grpc_server_register_completion_queue(f->server, f->cq);
   grpc_server_start(f->server);
-  transport =
-      grpc_create_chttp2_transport(server_args, sfd->server, NULL, 0, mdctx, 0);
+  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx, 0);
   server_setup_transport(f, transport, mdctx);
+  grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
 
 static void chttp2_tear_down_socketpair(grpc_end2end_test_fixture *f) {