Merge remote-tracking branch 'origin/races' into thread_pool

4 files changed, 83 insertions, 64 deletions

diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c
index 25a4db3c2d..28c18f1d58 100644
--- a/src/core/ext/filters/client_channel/client_channel.c
+++ b/src/core/ext/filters/client_channel/client_channel.c
@@ -754,11 +754,6 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
  * PER-CALL FUNCTIONS
  */
 
-#define GET_CALL(call_data) \
-  ((grpc_subchannel_call *)(gpr_atm_acq_load(&(call_data)->subchannel_call)))
-
-#define CANCELLED_CALL ((grpc_subchannel_call *)1)
-
 /** Call data.  Holds a pointer to grpc_subchannel_call and the
     associated machinery to create such a pointer.
     Handles queueing of stream ops until a call object is ready, waiting
@@ -779,11 +774,9 @@ typedef struct client_channel_call_data {
   grpc_server_retry_throttle_data *retry_throttle_data;
   method_parameters *method_params;
 
-  grpc_error *cancel_error;
-
-  /** either 0 for no call, 1 for cancelled, or a pointer to a
-      grpc_subchannel_call */
-  gpr_atm subchannel_call;
+  /** either 0 for no call, a pointer to a grpc_subchannel_call (if the lowest
+      bit is 0), or a pointer to an error (if the lowest bit is 1) */
+  gpr_atm subchannel_call_or_error;
   gpr_arena *arena;
 
   bool pick_pending;
@@ -805,10 +798,24 @@ typedef struct client_channel_call_data {
   grpc_closure *original_on_complete;
 } call_data;
 
+typedef struct {
+  grpc_subchannel_call *subchannel_call;
+  grpc_error *error;
+} call_or_error;
+
+static call_or_error get_call_or_error(call_data *p) {
+  gpr_atm c = gpr_atm_acq_load(&p->subchannel_call_or_error);
+  if (c == 0)
+    return (call_or_error){NULL, NULL};
+  else if (c & 1)
+    return (call_or_error){NULL, (grpc_error *)((c) & ~(gpr_atm)1)};
+  else
+    return (call_or_error){(grpc_subchannel_call *)c, NULL};
+}
+
 grpc_subchannel_call *grpc_client_channel_get_subchannel_call(
     grpc_call_element *call_elem) {
-  grpc_subchannel_call *scc = GET_CALL((call_data *)call_elem->call_data);
-  return scc == CANCELLED_CALL ? NULL : scc;
+  return get_call_or_error(call_elem->call_data).subchannel_call;
 }
 
 static void add_waiting_locked(call_data *calld,
@@ -840,18 +847,18 @@ static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
     return;
   }
 
-  grpc_subchannel_call *call = GET_CALL(calld);
+  call_or_error call = get_call_or_error(calld);
   grpc_transport_stream_op_batch **ops = calld->waiting_ops;
   size_t nops = calld->waiting_ops_count;
-  if (call == CANCELLED_CALL) {
-    fail_locked(exec_ctx, calld, GRPC_ERROR_CANCELLED);
+  if (call.error != GRPC_ERROR_NONE) {
+    fail_locked(exec_ctx, calld, GRPC_ERROR_REF(call.error));
     return;
   }
   calld->waiting_ops = NULL;
   calld->waiting_ops_count = 0;
   calld->waiting_ops_capacity = 0;
   for (size_t i = 0; i < nops; i++) {
-    grpc_subchannel_call_process_op(exec_ctx, call, ops[i]);
+    grpc_subchannel_call_process_op(exec_ctx, call.subchannel_call, ops[i]);
   }
   gpr_free(ops);
 }
@@ -912,16 +919,20 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
   calld->pick_pending = false;
   grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent,
                                            chand->interested_parties);
+  call_or_error coe = get_call_or_error(calld);
   if (calld->connected_subchannel == NULL) {
-    gpr_atm_no_barrier_store(&calld->subchannel_call, 1);
-    fail_locked(exec_ctx, calld,
-                GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
-                    "Failed to create subchannel", &error, 1));
-  } else if (GET_CALL(calld) == CANCELLED_CALL) {
+    grpc_error *failure = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
+        "Failed to create subchannel", &error, 1);
+    gpr_atm_no_barrier_store(&calld->subchannel_call_or_error,
+                             1 | (gpr_atm)GRPC_ERROR_REF(failure));
+    fail_locked(exec_ctx, calld, failure);
+  } else if (coe.error != GRPC_ERROR_NONE) {
     /* already cancelled before subchannel became ready */
+    grpc_error *child_errors[] = {error, coe.error};
     grpc_error *cancellation_error =
         GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
-            "Cancelled before creating subchannel", &error, 1);
+            "Cancelled before creating subchannel", child_errors,
+            GPR_ARRAY_SIZE(child_errors));
     /* if due to deadline, attach the deadline exceeded status to the error */
     if (gpr_time_cmp(calld->deadline, gpr_now(GPR_CLOCK_MONOTONIC)) < 0) {
       cancellation_error =
@@ -941,8 +952,8 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
         .context = calld->subchannel_call_context};
     grpc_error *new_error = grpc_connected_subchannel_create_call(
         exec_ctx, calld->connected_subchannel, &call_args, &subchannel_call);
-    gpr_atm_rel_store(&calld->subchannel_call,
-                      (gpr_atm)(uintptr_t)subchannel_call);
+    gpr_atm_rel_store(&calld->subchannel_call_or_error,
+                      (gpr_atm)subchannel_call);
     if (new_error != GRPC_ERROR_NONE) {
       new_error = grpc_error_add_child(new_error, error);
       fail_locked(exec_ctx, calld, new_error);
@@ -955,8 +966,9 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
 
 static char *cc_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
   call_data *calld = elem->call_data;
-  grpc_subchannel_call *subchannel_call = GET_CALL(calld);
-  if (subchannel_call == NULL || subchannel_call == CANCELLED_CALL) {
+  grpc_subchannel_call *subchannel_call =
+      get_call_or_error(calld).subchannel_call;
+  if (subchannel_call == NULL) {
     return NULL;
   } else {
     return grpc_subchannel_call_get_peer(exec_ctx, subchannel_call);
@@ -1115,25 +1127,26 @@ static void start_transport_stream_op_batch_locked_inner(
     grpc_call_element *elem) {
   channel_data *chand = elem->channel_data;
   call_data *calld = elem->call_data;
-  grpc_subchannel_call *call;
 
   /* need to recheck that another thread hasn't set the call */
-  call = GET_CALL(calld);
-  if (call == CANCELLED_CALL) {
+  call_or_error coe = get_call_or_error(calld);
+  if (coe.error != GRPC_ERROR_NONE) {
     grpc_transport_stream_op_batch_finish_with_failure(
-        exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
+        exec_ctx, op, GRPC_ERROR_REF(coe.error));
     /* early out */
     return;
   }
-  if (call != NULL) {
-    grpc_subchannel_call_process_op(exec_ctx, call, op);
+  if (coe.subchannel_call != NULL) {
+    grpc_subchannel_call_process_op(exec_ctx, coe.subchannel_call, op);
     /* early out */
     return;
   }
   /* if this is a cancellation, then we can raise our cancelled flag */
   if (op->cancel_stream) {
-    if (!gpr_atm_rel_cas(&calld->subchannel_call, 0,
-                         (gpr_atm)(uintptr_t)CANCELLED_CALL)) {
+    grpc_error *error = GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error);
+    if (!gpr_atm_rel_cas(&calld->subchannel_call_or_error, 0,
+                         1 | (gpr_atm)error)) {
+      GRPC_ERROR_UNREF(error);
       /* recurse to retry */
       start_transport_stream_op_batch_locked_inner(exec_ctx, op, elem);
       /* early out */
@@ -1144,19 +1157,13 @@ static void start_transport_stream_op_batch_locked_inner(
          cancelled before any ops are passed down (e.g., if the deadline
          is in the past when the call starts), we can return the right
          error to the caller when the first op does get passed down. */
-      calld->cancel_error =
-          GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error);
       if (calld->pick_pending) {
-        cancel_pick_locked(
-            exec_ctx, elem,
-            GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
+        cancel_pick_locked(exec_ctx, elem, GRPC_ERROR_REF(error));
       } else {
-        fail_locked(exec_ctx, calld,
-                    GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
+        fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error));
       }
-      grpc_transport_stream_op_batch_finish_with_failure(
-          exec_ctx, op,
-          GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
+      grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, op,
+                                                         GRPC_ERROR_REF(error));
       /* early out */
       return;
     }
@@ -1196,8 +1203,8 @@ static void start_transport_stream_op_batch_locked_inner(
         .context = calld->subchannel_call_context};
     grpc_error *error = grpc_connected_subchannel_create_call(
         exec_ctx, calld->connected_subchannel, &call_args, &subchannel_call);
-    gpr_atm_rel_store(&calld->subchannel_call,
-                      (gpr_atm)(uintptr_t)subchannel_call);
+    gpr_atm_rel_store(&calld->subchannel_call_or_error,
+                      (gpr_atm)subchannel_call);
     if (error != GRPC_ERROR_NONE) {
       fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error));
       grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, op, error);
@@ -1276,17 +1283,17 @@ static void cc_start_transport_stream_op_batch(
                                                                op);
   }
   /* try to (atomically) get the call */
-  grpc_subchannel_call *call = GET_CALL(calld);
+  call_or_error coe = get_call_or_error(calld);
   GPR_TIMER_BEGIN("cc_start_transport_stream_op_batch", 0);
-  if (call == CANCELLED_CALL) {
+  if (coe.error != GRPC_ERROR_NONE) {
     grpc_transport_stream_op_batch_finish_with_failure(
-        exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
+        exec_ctx, op, GRPC_ERROR_REF(coe.error));
     GPR_TIMER_END("cc_start_transport_stream_op_batch", 0);
     /* early out */
     return;
   }
-  if (call != NULL) {
-    grpc_subchannel_call_process_op(exec_ctx, call, op);
+  if (coe.subchannel_call != NULL) {
+    grpc_subchannel_call_process_op(exec_ctx, coe.subchannel_call, op);
     GPR_TIMER_END("cc_start_transport_stream_op_batch", 0);
     /* early out */
     return;
@@ -1334,12 +1341,14 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
   if (calld->method_params != NULL) {
     method_parameters_unref(calld->method_params);
   }
-  GRPC_ERROR_UNREF(calld->cancel_error);
-  grpc_subchannel_call *call = GET_CALL(calld);
-  if (call != NULL && call != CANCELLED_CALL) {
-    grpc_subchannel_call_set_cleanup_closure(call, then_schedule_closure);
+  call_or_error coe = get_call_or_error(calld);
+  GRPC_ERROR_UNREF(coe.error);
+  if (coe.subchannel_call != NULL) {
+    grpc_subchannel_call_set_cleanup_closure(coe.subchannel_call,
+                                             then_schedule_closure);
     then_schedule_closure = NULL;
-    GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call");
+    GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, coe.subchannel_call,
+                               "client_channel_destroy_call");
   }
   GPR_ASSERT(!calld->pick_pending);
   GPR_ASSERT(calld->waiting_ops_count == 0);
diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c
index 1af3393a62..dd14bf1d02 100644
--- a/src/core/ext/filters/client_channel/subchannel.c
+++ b/src/core/ext/filters/client_channel/subchannel.c
@@ -283,6 +283,7 @@ static void disconnect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
 void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
                            grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
   gpr_atm old_refs;
+  // add a weak ref and subtract a strong ref (atomically)
   old_refs = ref_mutate(c, (gpr_atm)1 - (gpr_atm)(1 << INTERNAL_REF_BITS),
                         1 REF_MUTATE_PURPOSE("STRONG_UNREF"));
   if ((old_refs & STRONG_REF_MASK) == (1 << INTERNAL_REF_BITS)) {
@@ -656,7 +657,6 @@ static bool publish_transport_locked(grpc_exec_ctx *exec_ctx,
     gpr_free(sw_subchannel);
     grpc_channel_stack_destroy(exec_ctx, stk);
     gpr_free(con);
-    GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
     return false;
   }
 
diff --git a/src/core/ext/filters/client_channel/subchannel_index.c b/src/core/ext/filters/client_channel/subchannel_index.c
index f6ef4a845e..b25dbfcf51 100644
--- a/src/core/ext/filters/client_channel/subchannel_index.c
+++ b/src/core/ext/filters/client_channel/subchannel_index.c
@@ -183,8 +183,11 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
   enter_ctx(exec_ctx);
 
   grpc_subchannel *c = NULL;
+  bool need_to_unref_constructed;
 
   while (c == NULL) {
+    need_to_unref_constructed = false;
+
     // Compare and swap loop:
     // - take a reference to the current index
     gpr_mu_lock(&g_mu);
@@ -194,8 +197,11 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
     // - Check to see if a subchannel already exists
     c = gpr_avl_get(index, key);
     if (c != NULL) {
+      c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(c, "index_register");
+    }
+    if (c != NULL) {
       // yes -> we're done
-      GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, constructed, "index_register");
+      need_to_unref_constructed = true;
     } else {
       // no -> update the avl and compare/swap
       gpr_avl updated =
@@ -219,6 +225,10 @@ grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx,
 
   leave_ctx(exec_ctx);
 
+  if (need_to_unref_constructed) {
+    GRPC_SUBCHANNEL_UNREF(exec_ctx, constructed, "index_register");
+  }
+
   return c;
 }
 
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 054f9b66f7..10b187496e 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -1563,12 +1563,6 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
   grpc_chttp2_transport *t = op->handler_private.extra_arg;
   grpc_error *close_transport = op->disconnect_with_error;
 
-  if (op->on_connectivity_state_change != NULL) {
-    grpc_connectivity_state_notify_on_state_change(
-        exec_ctx, &t->channel_callback.state_tracker, op->connectivity_state,
-        op->on_connectivity_state_change);
-  }
-
   if (op->goaway_error) {
     send_goaway(exec_ctx, t, op->goaway_error);
   }
@@ -1592,6 +1586,12 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
                      op->send_ping);
   }
 
+  if (op->on_connectivity_state_change != NULL) {
+    grpc_connectivity_state_notify_on_state_change(
+        exec_ctx, &t->channel_callback.state_tracker, op->connectivity_state,
+        op->on_connectivity_state_change);
+  }
+
   if (close_transport != GRPC_ERROR_NONE) {
     close_transport_locked(exec_ctx, t, close_transport);
   }