Fix a race in transport.

I removed the condition variable here a little while ago to remove a
thundering herd. Unfortunately it introduces a race if we are calling
back an application defined object whilst destroying. Reintroduce the
cv, and guard it's usage closely to avoid the herd (additionally, it's
not needed for stream deletion, so we keep it out of that).

1 files changed, 11 insertions, 0 deletions

diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 8b1fb78917..5921c3806e 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -184,11 +184,13 @@ struct transport {
   gpr_uint8 is_client;
 
   gpr_mu mu;
+  gpr_cv cv;
 
   /* basic state management - what are we doing at the moment? */
   gpr_uint8 reading;
   gpr_uint8 writing;
   gpr_uint8 calling_back;
+  gpr_uint8 destroying;
   error_state error_state;
 
   /* stream indexing */
@@ -362,6 +364,7 @@ static void unref_transport(transport *t) {
 
   gpr_mu_unlock(&t->mu);
   gpr_mu_destroy(&t->mu);
+  gpr_cv_destroy(&t->cv);
 
   /* callback remaining pings: they're not allowed to call into the transpot,
      and maybe they hold resources that need to be freed */
@@ -397,6 +400,7 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup,
   /* one ref is for destroy, the other for when ep becomes NULL */
   gpr_ref_init(&t->refs, 2);
   gpr_mu_init(&t->mu);
+  gpr_cv_init(&t->cv);
   t->metadata_context = mdctx;
   t->str_grpc_timeout =
       grpc_mdstr_from_string(t->metadata_context, "grpc-timeout");
@@ -405,6 +409,7 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup,
   t->error_state = ERROR_STATE_NONE;
   t->next_stream_id = is_client ? 1 : 2;
   t->last_incoming_stream_id = 0;
+  t->destroying = 0;
   t->is_client = is_client;
   t->outgoing_window = DEFAULT_WINDOW;
   t->incoming_window = DEFAULT_WINDOW;
@@ -487,6 +492,7 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup,
   t->cb = sr.callbacks;
   t->cb_user_data = sr.user_data;
   t->calling_back = 0;
+  if (t->destroying) gpr_cv_signal(&t->cv);
   unlock(t);
   unref_transport(t);
 }
@@ -495,6 +501,10 @@ static void destroy_transport(grpc_transport *gt) {
   transport *t = (transport *)gt;
 
   gpr_mu_lock(&t->mu);
+  t->destroying = 1;
+  while (t->calling_back) {
+    gpr_cv_wait(&t->cv, &t->mu, gpr_inf_future);
+  }
   t->cb = NULL;
   gpr_mu_unlock(&t->mu);
 
@@ -754,6 +764,7 @@ static void unlock(transport *t) {
   if (perform_callbacks || call_closed || num_goaways) {
     lock(t);
     t->calling_back = 0;
+    if (t->destroying) gpr_cv_signal(&t->cv);
     unlock(t);
     unref_transport(t);
   }