Move pollset_set ownership from channel to transport setup

4 files changed, 42 insertions, 23 deletions

diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 6c21ca34ef..26b6841f24 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -65,7 +65,6 @@ typedef struct {
   call_data **waiting_children;
   size_t waiting_child_count;
   size_t waiting_child_capacity;
-  grpc_pollset_set waiting_pollsets;
 
   /* transport setup for this channel */
   grpc_transport_setup *transport_setup;
@@ -212,7 +211,6 @@ static void cc_start_transport_op(grpc_call_element *elem,
           if (!chand->transport_setup_initiated) {
             chand->transport_setup_initiated = 1;
             initiate_transport_setup = 1;
-            grpc_pollset_set_init(&chand->waiting_pollsets);
           }
           /* add this call to the waiting set to be resumed once we have a child
              channel stack, growing the waiting set if needed */
@@ -223,16 +221,15 @@ static void cc_start_transport_op(grpc_call_element *elem,
                 chand->waiting_children,
                 chand->waiting_child_capacity * sizeof(call_data *));
           }
-          grpc_pollset_set_add_pollset(&chand->waiting_pollsets, op->bind_pollset);
           calld->s.waiting_op = *op;
           chand->waiting_children[chand->waiting_child_count++] = calld;
           gpr_mu_unlock(&chand->mu);
 
           /* finally initiate transport setup if needed */
           if (initiate_transport_setup) {
-            grpc_transport_setup_initiate(chand->transport_setup,
-                                          &chand->waiting_pollsets);
+            grpc_transport_setup_initiate(chand->transport_setup);
           }
+          grpc_transport_setup_add_interested_party(chand->transport_setup, op->bind_pollset);
         }
       }
       break;
@@ -476,8 +473,6 @@ grpc_transport_setup_result grpc_client_channel_transport_setup_complete(
   chand->waiting_child_count = 0;
   chand->waiting_child_capacity = 0;
 
-  grpc_pollset_set_destroy(&chand->waiting_pollsets);
-
   call_ops = gpr_malloc(sizeof(*call_ops) * waiting_child_count);
 
   for (i = 0; i < waiting_child_count; i++) {
diff --git a/src/core/channel/client_setup.c b/src/core/channel/client_setup.c
index 8234ea2438..df1e375bc9 100644
--- a/src/core/channel/client_setup.c
+++ b/src/core/channel/client_setup.c
@@ -61,7 +61,7 @@ struct grpc_client_setup {
 struct grpc_client_setup_request {
   /* pointer back to the setup object */
   grpc_client_setup *setup;
-  grpc_pollset_set *interested_parties;
+  grpc_pollset_set interested_parties;
   gpr_timespec deadline;
 };
 
@@ -71,7 +71,7 @@ gpr_timespec grpc_client_setup_request_deadline(grpc_client_setup_request *r) {
 
 grpc_pollset_set *grpc_client_setup_get_interested_parties(
     grpc_client_setup_request *r) {
-  return r->interested_parties;
+  return &r->interested_parties;
 }
 
 static void destroy_setup(grpc_client_setup *s) {
@@ -82,15 +82,19 @@ static void destroy_setup(grpc_client_setup *s) {
   gpr_free(s);
 }
 
+static void destroy_request(grpc_client_setup_request *r) {
+  grpc_pollset_set_destroy(&r->interested_parties);
+  gpr_free(r);
+}
+
 /* initiate handshaking */
-static void setup_initiate(grpc_transport_setup *sp,
-                           grpc_pollset_set *interested_parties) {
+static void setup_initiate(grpc_transport_setup *sp) {
   grpc_client_setup *s = (grpc_client_setup *)sp;
   grpc_client_setup_request *r = gpr_malloc(sizeof(grpc_client_setup_request));
   int in_alarm = 0;
 
   r->setup = s;
-  r->interested_parties = interested_parties;
+  grpc_pollset_set_init(&r->interested_parties);
   /* TODO(klempner): Actually set a deadline */
   r->deadline = gpr_inf_future;
 
@@ -112,10 +116,24 @@ static void setup_initiate(grpc_transport_setup *sp,
   if (!in_alarm) {
     s->initiate(s->user_data, r);
   } else {
-    gpr_free(r);
+    destroy_request(r);
   }
 }
 
+static void setup_add_interested_party(grpc_transport_setup *sp, grpc_pollset *pollset) {
+  grpc_client_setup *s = (grpc_client_setup *)sp;
+
+  gpr_mu_lock(&s->mu);
+  if (!s->active_request) {
+    gpr_mu_unlock(&s->mu);
+    return;
+  }
+
+  grpc_pollset_set_add_pollset(&s->active_request->interested_parties, pollset);
+
+  gpr_mu_unlock(&s->mu);
+}
+
 /* cancel handshaking: cancel all requests, and shutdown (the caller promises
    not to initiate again) */
 static void setup_cancel(grpc_transport_setup *sp) {
@@ -165,6 +183,7 @@ void grpc_client_setup_cb_end(grpc_client_setup_request *r) {
 
 /* vtable for transport setup */
 static const grpc_transport_setup_vtable setup_vtable = {setup_initiate,
+                                                         setup_add_interested_party,
                                                          setup_cancel};
 
 void grpc_client_setup_create_and_attach(
@@ -217,7 +236,7 @@ static void backoff_alarm_done(void *arg /* grpc_client_setup */, int success) {
     if (0 == --s->refs) {
       gpr_mu_unlock(&s->mu);
       destroy_setup(s);
-      gpr_free(r);
+      destroy_request(r);
       return;
     } else {
       gpr_mu_unlock(&s->mu);
@@ -242,11 +261,11 @@ void grpc_client_setup_request_finish(grpc_client_setup_request *r,
   if (!retry && 0 == --s->refs) {
     gpr_mu_unlock(&s->mu);
     destroy_setup(s);
-    gpr_free(r);
+    destroy_request(r);
     return;
   }
 
-  gpr_free(r);
+  destroy_request(r);
 
   if (retry) {
     /* TODO(klempner): Replace these values with further consideration. 2x is
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index b30b13d297..0bceeabcc3 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -82,9 +82,12 @@ void grpc_transport_setup_cancel(grpc_transport_setup *setup) {
   setup->vtable->cancel(setup);
 }
 
-void grpc_transport_setup_initiate(grpc_transport_setup *setup,
-                                   grpc_pollset_set *interested_parties) {
-  setup->vtable->initiate(setup, interested_parties);
+void grpc_transport_setup_initiate(grpc_transport_setup *setup) {
+  setup->vtable->initiate(setup);
+}
+
+void grpc_transport_setup_add_interested_party(grpc_transport_setup *setup, grpc_pollset *pollset) {
+  setup->vtable->add_interested_party(setup, pollset);
 }
 
 void grpc_transport_op_finish_with_failure(grpc_transport_op *op) {
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index ef1bda8d90..5303cbee7d 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -194,8 +194,8 @@ typedef struct grpc_transport_setup grpc_transport_setup;
 typedef struct grpc_transport_setup_vtable grpc_transport_setup_vtable;
 
 struct grpc_transport_setup_vtable {
-  void (*initiate)(grpc_transport_setup *setup,
-                   grpc_pollset_set *interested_parties);
+  void (*initiate)(grpc_transport_setup *setup);
+  void (*add_interested_party)(grpc_transport_setup *setup, grpc_pollset *pollset);
   void (*cancel)(grpc_transport_setup *setup);
 };
 
@@ -211,8 +211,10 @@ struct grpc_transport_setup {
    setup construction time.
    This *may* be implemented as a no-op if the setup process monitors something
    continuously. */
-void grpc_transport_setup_initiate(grpc_transport_setup *setup,
-                                   grpc_pollset_set *interested_parties);
+void grpc_transport_setup_initiate(grpc_transport_setup *setup);
+
+void grpc_transport_setup_add_interested_party(grpc_transport_setup *setup, grpc_pollset *pollset);
+
 /* Cancel transport setup. After this returns, no new transports should be
    created, and all pending transport setup callbacks should be completed.
    After this call completes, setup should be considered invalid (this can be