grpclb: Don't try to pick from shutdown RR

src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c

@@ -1179,35 +1179,56 @@ 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 transition 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));
+      }
+      // Attempt to switch over to a pending update.
+      rr_handover_locked(exec_ctx, glb_policy);
+      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 =
+          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 = 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 "
@@ -1216,11 +1237,10 @@ 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);
-
+    pick_done = false;
     if (!glb_policy->started_picking) {
       start_picking_locked(exec_ctx, glb_policy);
     }
-    pick_done = false;
   }
   return pick_done;
 }

src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c

@@ -420,10 +420,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) {
@@ -534,7 +535,12 @@ static grpc_connectivity_state update_lb_connectivity_status_locked(
                                 GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error),
                                 "rr_shutdown");
     p->shutdown = true;
-    new_state = GRPC_CHANNEL_SHUTDOWN;
+    if (GRPC_TRACER_ON(grpc_lb_round_robin_trace)) {
+      new_state = GRPC_CHANNEL_SHUTDOWN;
+      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,
@@ -755,6 +761,10 @@ static void rr_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
     return;
   }
   grpc_lb_addresses *addresses = arg->value.pointer.p;
+  if (GRPC_TRACER_ON(grpc_lb_round_robin_trace)) {
+    gpr_log(GPR_INFO, "[RR %p] Update with %lu addressees (shutdown: %d)",
+            (void *)p, (unsigned long)addresses->num_addresses, p->shutdown);
+  }
   rr_subchannel_list *subchannel_list =
       rr_subchannel_list_create(p, addresses->num_addresses);
   if (addresses->num_addresses == 0) {