Merge pull request #12427 from dgquintas/grpclb_pick_from_shutdown_rr

grpclb: Don't try to pick from shutdown RR

2 files changed, 51 insertions, 30 deletions

diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
index 03116b420c..1b19650b61 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
@@ -1163,36 +1163,52 @@ static int glb_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
                            "won't work without it. Failing"));
     return 0;
   }
-
   glb_lb_policy *glb_policy = (glb_lb_policy *)pol;
-  bool pick_done;
-
+  bool pick_done = false;
   if (glb_policy->rr_policy != NULL) {
-    if (GRPC_TRACER_ON(grpc_lb_glb_trace)) {
-      gpr_log(GPR_INFO, "grpclb %p about to PICK from RR %p",
-              (void *)glb_policy, (void *)glb_policy->rr_policy);
+    const grpc_connectivity_state rr_connectivity_state =
+        grpc_lb_policy_check_connectivity_locked(exec_ctx,
+                                                 glb_policy->rr_policy, NULL);
+    // The glb_policy->rr_policy may have transitioned to SHUTDOWN but the
+    // callback registered to capture this event
+    // (glb_rr_connectivity_changed_locked) may not have been invoked yet. We
+    // need to make sure we aren't trying to pick from a RR policy instance
+    // that's in shutdown.
+    if (rr_connectivity_state == GRPC_CHANNEL_SHUTDOWN) {
+      if (GRPC_TRACER_ON(grpc_lb_glb_trace)) {
+        gpr_log(GPR_INFO,
+                "grpclb %p NOT picking from from RR %p: RR conn state=%s",
+                (void *)glb_policy, (void *)glb_policy->rr_policy,
+                grpc_connectivity_state_name(rr_connectivity_state));
+      }
+      add_pending_pick(&glb_policy->pending_picks, pick_args, target, context,
+                       on_complete);
+      pick_done = false;
+    } else {  // RR not in shutdown
+      if (GRPC_TRACER_ON(grpc_lb_glb_trace)) {
+        gpr_log(GPR_INFO, "grpclb %p about to PICK from RR %p",
+                (void *)glb_policy, (void *)glb_policy->rr_policy);
+      }
+      GRPC_LB_POLICY_REF(glb_policy->rr_policy, "glb_pick");
+      wrapped_rr_closure_arg *wc_arg =
+          (wrapped_rr_closure_arg *)gpr_zalloc(sizeof(wrapped_rr_closure_arg));
+      GRPC_CLOSURE_INIT(&wc_arg->wrapper_closure, wrapped_rr_closure, wc_arg,
+                        grpc_schedule_on_exec_ctx);
+      wc_arg->rr_policy = glb_policy->rr_policy;
+      wc_arg->target = target;
+      wc_arg->context = context;
+      GPR_ASSERT(glb_policy->client_stats != NULL);
+      wc_arg->client_stats =
+          grpc_grpclb_client_stats_ref(glb_policy->client_stats);
+      wc_arg->wrapped_closure = on_complete;
+      wc_arg->lb_token_mdelem_storage = pick_args->lb_token_mdelem_storage;
+      wc_arg->initial_metadata = pick_args->initial_metadata;
+      wc_arg->free_when_done = wc_arg;
+      pick_done =
+          pick_from_internal_rr_locked(exec_ctx, glb_policy, pick_args,
+                                       false /* force_async */, target, wc_arg);
     }
-    GRPC_LB_POLICY_REF(glb_policy->rr_policy, "glb_pick");
-
-    wrapped_rr_closure_arg *wc_arg =
-        (wrapped_rr_closure_arg *)gpr_zalloc(sizeof(wrapped_rr_closure_arg));
-
-    GRPC_CLOSURE_INIT(&wc_arg->wrapper_closure, wrapped_rr_closure, wc_arg,
-                      grpc_schedule_on_exec_ctx);
-    wc_arg->rr_policy = glb_policy->rr_policy;
-    wc_arg->target = target;
-    wc_arg->context = context;
-    GPR_ASSERT(glb_policy->client_stats != NULL);
-    wc_arg->client_stats =
-        grpc_grpclb_client_stats_ref(glb_policy->client_stats);
-    wc_arg->wrapped_closure = on_complete;
-    wc_arg->lb_token_mdelem_storage = pick_args->lb_token_mdelem_storage;
-    wc_arg->initial_metadata = pick_args->initial_metadata;
-    wc_arg->free_when_done = wc_arg;
-    pick_done =
-        pick_from_internal_rr_locked(exec_ctx, glb_policy, pick_args,
-                                     false /* force_async */, target, wc_arg);
-  } else {
+  } else {  // glb_policy->rr_policy == NULL
     if (GRPC_TRACER_ON(grpc_lb_glb_trace)) {
       gpr_log(GPR_DEBUG,
               "No RR policy in grpclb instance %p. Adding to grpclb's pending "
@@ -1201,7 +1217,6 @@ static int glb_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
     }
     add_pending_pick(&glb_policy->pending_picks, pick_args, target, context,
                      on_complete);
-
     if (!glb_policy->started_picking) {
       start_picking_locked(exec_ctx, glb_policy);
     }
diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
index 8f29c80130..fae40e378b 100644
--- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
+++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
@@ -277,10 +277,11 @@ static int rr_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
                           grpc_call_context_element *context, void **user_data,
                           grpc_closure *on_complete) {
   round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
-  GPR_ASSERT(!p->shutdown);
   if (GRPC_TRACER_ON(grpc_lb_round_robin_trace)) {
-    gpr_log(GPR_INFO, "[RR %p] Trying to pick", (void *)pol);
+    gpr_log(GPR_INFO, "[RR %p] Trying to pick (shutdown: %d)", (void *)pol,
+            p->shutdown);
   }
+  GPR_ASSERT(!p->shutdown);
   if (p->subchannel_list != NULL) {
     const size_t next_ready_index = get_next_ready_subchannel_index_locked(p);
     if (next_ready_index < p->subchannel_list->num_subchannels) {
@@ -393,6 +394,11 @@ static grpc_connectivity_state update_lb_connectivity_status_locked(
                                 "rr_shutdown");
     p->shutdown = true;
     new_state = GRPC_CHANNEL_SHUTDOWN;
+    if (GRPC_TRACER_ON(grpc_lb_round_robin_trace)) {
+      gpr_log(GPR_INFO,
+              "[RR %p] Shutting down: all subchannels have gone into shutdown",
+              (void *)p);
+    }
   } else if (subchannel_list->num_transient_failures ==
              p->subchannel_list->num_subchannels) { /* 4) TRANSIENT_FAILURE */
     grpc_connectivity_state_set(exec_ctx, &p->state_tracker,