Merge pull request #553 from ctiller/tsan

Fix a race in transport.

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