path: root/src/core/ext/filters/client_channel/request_routing.cc

/*
 *
 * Copyright 2015 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include <grpc/support/port_platform.h>

#include "src/core/ext/filters/client_channel/request_routing.h"

#include <inttypes.h>
#include <limits.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>

#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>

#include "src/core/ext/filters/client_channel/backup_poller.h"
#include "src/core/ext/filters/client_channel/http_connect_handshaker.h"
#include "src/core/ext/filters/client_channel/lb_policy_registry.h"
#include "src/core/ext/filters/client_channel/proxy_mapper_registry.h"
#include "src/core/ext/filters/client_channel/resolver_registry.h"
#include "src/core/ext/filters/client_channel/retry_throttle.h"
#include "src/core/ext/filters/client_channel/server_address.h"
#include "src/core/ext/filters/client_channel/subchannel.h"
#include "src/core/ext/filters/deadline/deadline_filter.h"
#include "src/core/lib/backoff/backoff.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/connected_channel.h"
#include "src/core/lib/channel/status_util.h"
#include "src/core/lib/gpr/string.h"
#include "src/core/lib/gprpp/inlined_vector.h"
#include "src/core/lib/gprpp/manual_constructor.h"
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/iomgr/iomgr.h"
#include "src/core/lib/iomgr/polling_entity.h"
#include "src/core/lib/profiling/timers.h"
#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/slice/slice_string_helpers.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/transport/connectivity_state.h"
#include "src/core/lib/transport/error_utils.h"
#include "src/core/lib/transport/metadata.h"
#include "src/core/lib/transport/metadata_batch.h"
#include "src/core/lib/transport/service_config.h"
#include "src/core/lib/transport/static_metadata.h"
#include "src/core/lib/transport/status_metadata.h"

namespace grpc_core {

//
// RequestRouter::Request::ResolverResultWaiter
//

// Handles waiting for a resolver result.
// Used only for the first call on an idle channel.
class RequestRouter::Request::ResolverResultWaiter {
 public:
  explicit ResolverResultWaiter(Request* request)
      : request_router_(request->request_router_),
        request_(request),
        tracer_enabled_(request_router_->tracer_->enabled()) {
    if (tracer_enabled_) {
      gpr_log(GPR_INFO,
              "request_router=%p request=%p: deferring pick pending resolver "
              "result",
              request_router_, request);
    }
    // Add closure to be run when a resolver result is available.
    GRPC_CLOSURE_INIT(&done_closure_, &DoneLocked, this,
                      grpc_combiner_scheduler(request_router_->combiner_));
    AddToWaitingList();
    // Set cancellation closure, so that we abort if the call is cancelled.
    GRPC_CLOSURE_INIT(&cancel_closure_, &CancelLocked, this,
                      grpc_combiner_scheduler(request_router_->combiner_));
    grpc_call_combiner_set_notify_on_cancel(request->call_combiner_,
                                            &cancel_closure_);
  }

 private:
  // Adds done_closure_ to
  // request_router_->waiting_for_resolver_result_closures_.
  void AddToWaitingList() {
    grpc_closure_list_append(
        &request_router_->waiting_for_resolver_result_closures_, &done_closure_,
        GRPC_ERROR_NONE);
  }

  // Invoked when a resolver result is available.
  static void DoneLocked(void* arg, grpc_error* error) {
    ResolverResultWaiter* self = static_cast<ResolverResultWaiter*>(arg);
    RequestRouter* request_router = self->request_router_;
    // If CancelLocked() has already run, delete ourselves without doing
    // anything.  Note that the call stack may have already been destroyed,
    // so it's not safe to access anything in state_.
    if (GPR_UNLIKELY(self->finished_)) {
      if (self->tracer_enabled_) {
        gpr_log(GPR_INFO,
                "request_router=%p: call cancelled before resolver result",
                request_router);
      }
      Delete(self);
      return;
    }
    // Otherwise, process the resolver result.
    Request* request = self->request_;
    if (GPR_UNLIKELY(error != GRPC_ERROR_NONE)) {
      if (self->tracer_enabled_) {
        gpr_log(GPR_INFO,
                "request_router=%p request=%p: resolver failed to return data",
                request_router, request);
      }
      GRPC_CLOSURE_RUN(request->on_route_done_, GRPC_ERROR_REF(error));
    } else if (GPR_UNLIKELY(request_router->resolver_ == nullptr)) {
      // Shutting down.
      if (self->tracer_enabled_) {
        gpr_log(GPR_INFO, "request_router=%p request=%p: resolver disconnected",
                request_router, request);
      }
      GRPC_CLOSURE_RUN(request->on_route_done_,
                       GRPC_ERROR_CREATE_FROM_STATIC_STRING("Disconnected"));
    } else if (GPR_UNLIKELY(request_router->lb_policy_ == nullptr)) {
      // Transient resolver failure.
      // If call has wait_for_ready=true, try again; otherwise, fail.
      if (*request->pick_.initial_metadata_flags &
          GRPC_INITIAL_METADATA_WAIT_FOR_READY) {
        if (self->tracer_enabled_) {
          gpr_log(GPR_INFO,
                  "request_router=%p request=%p: resolver returned but no LB "
                  "policy; wait_for_ready=true; trying again",
                  request_router, request);
        }
        // Re-add ourselves to the waiting list.
        self->AddToWaitingList();
        // Return early so that we don't set finished_ to true below.
        return;
      } else {
        if (self->tracer_enabled_) {
          gpr_log(GPR_INFO,
                  "request_router=%p request=%p: resolver returned but no LB "
                  "policy; wait_for_ready=false; failing",
                  request_router, request);
        }
        GRPC_CLOSURE_RUN(
            request->on_route_done_,
            grpc_error_set_int(
                GRPC_ERROR_CREATE_FROM_STATIC_STRING("Name resolution failure"),
                GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE));
      }
    } else {
      if (self->tracer_enabled_) {
        gpr_log(GPR_INFO,
                "request_router=%p request=%p: resolver returned, doing LB "
                "pick",
                request_router, request);
      }
      request->ProcessServiceConfigAndStartLbPickLocked();
    }
    self->finished_ = true;
  }

  // Invoked when the call is cancelled.
  // Note: This runs under the client_channel combiner, but will NOT be
  // holding the call combiner.
  static void CancelLocked(void* arg, grpc_error* error) {
    ResolverResultWaiter* self = static_cast<ResolverResultWaiter*>(arg);
    RequestRouter* request_router = self->request_router_;
    // If DoneLocked() has already run, delete ourselves without doing anything.
    if (self->finished_) {
      Delete(self);
      return;
    }
    Request* request = self->request_;
    // If we are being cancelled, immediately invoke on_route_done_
    // to propagate the error back to the caller.
    if (error != GRPC_ERROR_NONE) {
      if (self->tracer_enabled_) {
        gpr_log(GPR_INFO,
                "request_router=%p request=%p: cancelling call waiting for "
                "name resolution",
                request_router, request);
      }
      // Note: Although we are not in the call combiner here, we are
      // basically stealing the call combiner from the pending pick, so
      // it's safe to run on_route_done_ here -- we are essentially
      // calling it here instead of calling it in DoneLocked().
      GRPC_CLOSURE_RUN(request->on_route_done_,
                       GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
                           "Pick cancelled", &error, 1));
    }
    self->finished_ = true;
  }

  RequestRouter* request_router_;
  Request* request_;
  const bool tracer_enabled_;
  grpc_closure done_closure_;
  grpc_closure cancel_closure_;
  bool finished_ = false;
};

//
// RequestRouter::Request::AsyncPickCanceller
//

// Handles the call combiner cancellation callback for an async LB pick.
class RequestRouter::Request::AsyncPickCanceller {
 public:
  explicit AsyncPickCanceller(Request* request)
      : request_router_(request->request_router_),
        request_(request),
        tracer_enabled_(request_router_->tracer_->enabled()) {
    GRPC_CALL_STACK_REF(request->owning_call_, "pick_callback_cancel");
    // Set cancellation closure, so that we abort if the call is cancelled.
    GRPC_CLOSURE_INIT(&cancel_closure_, &CancelLocked, this,
                      grpc_combiner_scheduler(request_router_->combiner_));
    grpc_call_combiner_set_notify_on_cancel(request->call_combiner_,
                                            &cancel_closure_);
  }

  void MarkFinishedLocked() {
    finished_ = true;
    GRPC_CALL_STACK_UNREF(request_->owning_call_, "pick_callback_cancel");
  }

 private:
  // Invoked when the call is cancelled.
  // Note: This runs under the client_channel combiner, but will NOT be
  // holding the call combiner.
  static void CancelLocked(void* arg, grpc_error* error) {
    AsyncPickCanceller* self = static_cast<AsyncPickCanceller*>(arg);
    Request* request = self->request_;
    RequestRouter* request_router = self->request_router_;
    if (!self->finished_) {
      // Note: request_router->lb_policy_ may have changed since we started our
      // pick, in which case we will be cancelling the pick on a policy other
      // than the one we started it on.  However, this will just be a no-op.
      if (error != GRPC_ERROR_NONE && request_router->lb_policy_ != nullptr) {
        if (self->tracer_enabled_) {
          gpr_log(GPR_INFO,
                  "request_router=%p request=%p: cancelling pick from LB "
                  "policy %p",
                  request_router, request, request_router->lb_policy_.get());
        }
        request_router->lb_policy_->CancelPickLocked(&request->pick_,
                                                     GRPC_ERROR_REF(error));
      }
      request->pick_canceller_ = nullptr;
      GRPC_CALL_STACK_UNREF(request->owning_call_, "pick_callback_cancel");
    }
    Delete(self);
  }

  RequestRouter* request_router_;
  Request* request_;
  const bool tracer_enabled_;
  grpc_closure cancel_closure_;
  bool finished_ = false;
};

//
// RequestRouter::Request
//

RequestRouter::Request::Request(grpc_call_stack* owning_call,
                                grpc_call_combiner* call_combiner,
                                grpc_polling_entity* pollent,
                                grpc_metadata_batch* send_initial_metadata,
                                uint32_t* send_initial_metadata_flags,
                                ApplyServiceConfigCallback apply_service_config,
                                void* apply_service_config_user_data,
                                grpc_closure* on_route_done)
    : owning_call_(owning_call),
      call_combiner_(call_combiner),
      pollent_(pollent),
      apply_service_config_(apply_service_config),
      apply_service_config_user_data_(apply_service_config_user_data),
      on_route_done_(on_route_done) {
  pick_.initial_metadata = send_initial_metadata;
  pick_.initial_metadata_flags = send_initial_metadata_flags;
}

RequestRouter::Request::~Request() {
  if (pick_.connected_subchannel != nullptr) {
    pick_.connected_subchannel.reset();
  }
  for (size_t i = 0; i < GRPC_CONTEXT_COUNT; ++i) {
    if (pick_.subchannel_call_context[i].destroy != nullptr) {
      pick_.subchannel_call_context[i].destroy(
          pick_.subchannel_call_context[i].value);
    }
  }
}

// Invoked once resolver results are available.
void RequestRouter::Request::ProcessServiceConfigAndStartLbPickLocked() {
  // Get service config data if needed.
  if (!apply_service_config_(apply_service_config_user_data_)) return;
  // Start LB pick.
  StartLbPickLocked();
}

void RequestRouter::Request::MaybeAddCallToInterestedPartiesLocked() {
  if (!pollent_added_to_interested_parties_) {
    pollent_added_to_interested_parties_ = true;
    grpc_polling_entity_add_to_pollset_set(
        pollent_, request_router_->interested_parties_);
  }
}

void RequestRouter::Request::MaybeRemoveCallFromInterestedPartiesLocked() {
  if (pollent_added_to_interested_parties_) {
    pollent_added_to_interested_parties_ = false;
    grpc_polling_entity_del_from_pollset_set(
        pollent_, request_router_->interested_parties_);
  }
}

// Starts a pick on the LB policy.
void RequestRouter::Request::StartLbPickLocked() {
  if (request_router_->tracer_->enabled()) {
    gpr_log(GPR_INFO,
            "request_router=%p request=%p: starting pick on lb_policy=%p",
            request_router_, this, request_router_->lb_policy_.get());
  }
  GRPC_CLOSURE_INIT(&on_pick_done_, &LbPickDoneLocked, this,
                    grpc_combiner_scheduler(request_router_->combiner_));
  pick_.on_complete = &on_pick_done_;
  GRPC_CALL_STACK_REF(owning_call_, "pick_callback");
  grpc_error* error = GRPC_ERROR_NONE;
  const bool pick_done =
      request_router_->lb_policy_->PickLocked(&pick_, &error);
  if (pick_done) {
    // Pick completed synchronously.
    if (request_router_->tracer_->enabled()) {
      gpr_log(GPR_INFO,
              "request_router=%p request=%p: pick completed synchronously",
              request_router_, this);
    }
    GRPC_CLOSURE_RUN(on_route_done_, error);
    GRPC_CALL_STACK_UNREF(owning_call_, "pick_callback");
  } else {
    // Pick will be returned asynchronously.
    // Add the request's polling entity to the request_router's
    // interested_parties, so that the I/O of the LB policy can be done
    // under it.  It will be removed in LbPickDoneLocked().
    MaybeAddCallToInterestedPartiesLocked();
    // Request notification on call cancellation.
    // We allocate a separate object to track cancellation, since the
    // cancellation closure might still be pending when we need to reuse
    // the memory in which this Request object is stored for a subsequent
    // retry attempt.
    pick_canceller_ = New<AsyncPickCanceller>(this);
  }
}

// Callback invoked by LoadBalancingPolicy::PickLocked() for async picks.
// Unrefs the LB policy and invokes on_route_done_.
void RequestRouter::Request::LbPickDoneLocked(void* arg, grpc_error* error) {
  Request* self = static_cast<Request*>(arg);
  RequestRouter* request_router = self->request_router_;
  if (request_router->tracer_->enabled()) {
    gpr_log(GPR_INFO,
            "request_router=%p request=%p: pick completed asynchronously",
            request_router, self);
  }
  self->MaybeRemoveCallFromInterestedPartiesLocked();
  if (self->pick_canceller_ != nullptr) {
    self->pick_canceller_->MarkFinishedLocked();
  }
  GRPC_CLOSURE_RUN(self->on_route_done_, GRPC_ERROR_REF(error));
  GRPC_CALL_STACK_UNREF(self->owning_call_, "pick_callback");
}

//
// RequestRouter::LbConnectivityWatcher
//

class RequestRouter::LbConnectivityWatcher {
 public:
  LbConnectivityWatcher(RequestRouter* request_router,
                        grpc_connectivity_state state,
                        LoadBalancingPolicy* lb_policy,
                        grpc_channel_stack* owning_stack,
                        grpc_combiner* combiner)
      : request_router_(request_router),
        state_(state),
        lb_policy_(lb_policy),
        owning_stack_(owning_stack) {
    GRPC_CHANNEL_STACK_REF(owning_stack_, "LbConnectivityWatcher");
    GRPC_CLOSURE_INIT(&on_changed_, &OnLbPolicyStateChangedLocked, this,
                      grpc_combiner_scheduler(combiner));
    lb_policy_->NotifyOnStateChangeLocked(&state_, &on_changed_);
  }

  ~LbConnectivityWatcher() {
    GRPC_CHANNEL_STACK_UNREF(owning_stack_, "LbConnectivityWatcher");
  }

 private:
  static void OnLbPolicyStateChangedLocked(void* arg, grpc_error* error) {
    LbConnectivityWatcher* self = static_cast<LbConnectivityWatcher*>(arg);
    // If the notification is not for the current policy, we're stale,
    // so delete ourselves.
    if (self->lb_policy_ != self->request_router_->lb_policy_.get()) {
      Delete(self);
      return;
    }
    // Otherwise, process notification.
    if (self->request_router_->tracer_->enabled()) {
      gpr_log(GPR_INFO, "request_router=%p: lb_policy=%p state changed to %s",
              self->request_router_, self->lb_policy_,
              grpc_connectivity_state_name(self->state_));
    }
    self->request_router_->SetConnectivityStateLocked(
        self->state_, GRPC_ERROR_REF(error), "lb_changed");
    // If shutting down, terminate watch.
    if (self->state_ == GRPC_CHANNEL_SHUTDOWN) {
      Delete(self);
      return;
    }
    // Renew watch.
    self->lb_policy_->NotifyOnStateChangeLocked(&self->state_,
                                                &self->on_changed_);
  }

  RequestRouter* request_router_;
  grpc_connectivity_state state_;
  // LB policy address. No ref held, so not safe to dereference unless
  // it happens to match request_router->lb_policy_.
  LoadBalancingPolicy* lb_policy_;
  grpc_channel_stack* owning_stack_;
  grpc_closure on_changed_;
};

//
// RequestRounter::ReresolutionRequestHandler
//

class RequestRouter::ReresolutionRequestHandler {
 public:
  ReresolutionRequestHandler(RequestRouter* request_router,
                             LoadBalancingPolicy* lb_policy,
                             grpc_channel_stack* owning_stack,
                             grpc_combiner* combiner)
      : request_router_(request_router),
        lb_policy_(lb_policy),
        owning_stack_(owning_stack) {
    GRPC_CHANNEL_STACK_REF(owning_stack_, "ReresolutionRequestHandler");
    GRPC_CLOSURE_INIT(&closure_, &OnRequestReresolutionLocked, this,
                      grpc_combiner_scheduler(combiner));
    lb_policy_->SetReresolutionClosureLocked(&closure_);
  }

 private:
  static void OnRequestReresolutionLocked(void* arg, grpc_error* error) {
    ReresolutionRequestHandler* self =
        static_cast<ReresolutionRequestHandler*>(arg);
    RequestRouter* request_router = self->request_router_;
    // If this invocation is for a stale LB policy, treat it as an LB shutdown
    // signal.
    if (self->lb_policy_ != request_router->lb_policy_.get() ||
        error != GRPC_ERROR_NONE || request_router->resolver_ == nullptr) {
      GRPC_CHANNEL_STACK_UNREF(request_router->owning_stack_,
                               "ReresolutionRequestHandler");
      Delete(self);
      return;
    }
    if (request_router->tracer_->enabled()) {
      gpr_log(GPR_INFO, "request_router=%p: started name re-resolving",
              request_router);
    }
    request_router->resolver_->RequestReresolutionLocked();
    // Give back the closure to the LB policy.
    self->lb_policy_->SetReresolutionClosureLocked(&self->closure_);
  }

  RequestRouter* request_router_;
  // LB policy address. No ref held, so not safe to dereference unless
  // it happens to match request_router->lb_policy_.
  LoadBalancingPolicy* lb_policy_;
  grpc_channel_stack* owning_stack_;
  grpc_closure closure_;
};

//
// RequestRouter
//

RequestRouter::RequestRouter(
    grpc_channel_stack* owning_stack, grpc_combiner* combiner,
    grpc_client_channel_factory* client_channel_factory,
    grpc_pollset_set* interested_parties, TraceFlag* tracer,
    ProcessResolverResultCallback process_resolver_result,
    void* process_resolver_result_user_data, const char* target_uri,
    const grpc_channel_args* args, grpc_error** error)
    : owning_stack_(owning_stack),
      combiner_(combiner),
      client_channel_factory_(client_channel_factory),
      interested_parties_(interested_parties),
      tracer_(tracer),
      process_resolver_result_(process_resolver_result),
      process_resolver_result_user_data_(process_resolver_result_user_data) {
  GRPC_CLOSURE_INIT(&on_resolver_result_changed_,
                    &RequestRouter::OnResolverResultChangedLocked, this,
                    grpc_combiner_scheduler(combiner));
  grpc_connectivity_state_init(&state_tracker_, GRPC_CHANNEL_IDLE,
                               "request_router");
  grpc_channel_args* new_args = nullptr;
  if (process_resolver_result == nullptr) {
    grpc_arg arg = grpc_channel_arg_integer_create(
        const_cast<char*>(GRPC_ARG_SERVICE_CONFIG_DISABLE_RESOLUTION), 0);
    new_args = grpc_channel_args_copy_and_add(args, &arg, 1);
  }
  resolver_ = ResolverRegistry::CreateResolver(
      target_uri, (new_args == nullptr ? args : new_args), interested_parties_,
      combiner_);
  grpc_channel_args_destroy(new_args);
  if (resolver_ == nullptr) {
    *error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("resolver creation failed");
  }
}

RequestRouter::~RequestRouter() {
  if (resolver_ != nullptr) {
    // The only way we can get here is if we never started resolving,
    // because we take a ref to the channel stack when we start
    // resolving and do not release it until the resolver callback is
    // invoked after the resolver shuts down.
    resolver_.reset();
  }
  if (lb_policy_ != nullptr) {
    grpc_pollset_set_del_pollset_set(lb_policy_->interested_parties(),
                                     interested_parties_);
    lb_policy_.reset();
  }
  if (client_channel_factory_ != nullptr) {
    grpc_client_channel_factory_unref(client_channel_factory_);
  }
  grpc_connectivity_state_destroy(&state_tracker_);
}

namespace {

const char* GetChannelConnectivityStateChangeString(
    grpc_connectivity_state state) {
  switch (state) {
    case GRPC_CHANNEL_IDLE:
      return "Channel state change to IDLE";
    case GRPC_CHANNEL_CONNECTING:
      return "Channel state change to CONNECTING";
    case GRPC_CHANNEL_READY:
      return "Channel state change to READY";
    case GRPC_CHANNEL_TRANSIENT_FAILURE:
      return "Channel state change to TRANSIENT_FAILURE";
    case GRPC_CHANNEL_SHUTDOWN:
      return "Channel state change to SHUTDOWN";
  }
  GPR_UNREACHABLE_CODE(return "UNKNOWN");
}

}  // namespace

void RequestRouter::SetConnectivityStateLocked(grpc_connectivity_state state,
                                               grpc_error* error,
                                               const char* reason) {
  if (lb_policy_ != nullptr) {
    if (state == GRPC_CHANNEL_TRANSIENT_FAILURE) {
      // Cancel picks with wait_for_ready=false.
      lb_policy_->CancelMatchingPicksLocked(
          /* mask= */ GRPC_INITIAL_METADATA_WAIT_FOR_READY,
          /* check= */ 0, GRPC_ERROR_REF(error));
    } else if (state == GRPC_CHANNEL_SHUTDOWN) {
      // Cancel all picks.
      lb_policy_->CancelMatchingPicksLocked(/* mask= */ 0, /* check= */ 0,
                                            GRPC_ERROR_REF(error));
    }
  }
  if (tracer_->enabled()) {
    gpr_log(GPR_INFO, "request_router=%p: setting connectivity state to %s",
            this, grpc_connectivity_state_name(state));
  }
  if (channelz_node_ != nullptr) {
    channelz_node_->AddTraceEvent(
        channelz::ChannelTrace::Severity::Info,
        grpc_slice_from_static_string(
            GetChannelConnectivityStateChangeString(state)));
  }
  grpc_connectivity_state_set(&state_tracker_, state, error, reason);
}

void RequestRouter::StartResolvingLocked() {
  if (tracer_->enabled()) {
    gpr_log(GPR_INFO, "request_router=%p: starting name resolution", this);
  }
  GPR_ASSERT(!started_resolving_);
  started_resolving_ = true;
  GRPC_CHANNEL_STACK_REF(owning_stack_, "resolver");
  resolver_->NextLocked(&resolver_result_, &on_resolver_result_changed_);
}

// Invoked from the resolver NextLocked() callback when the resolver
// is shutting down.
void RequestRouter::OnResolverShutdownLocked(grpc_error* error) {
  if (tracer_->enabled()) {
    gpr_log(GPR_INFO, "request_router=%p: shutting down", this);
  }
  if (lb_policy_ != nullptr) {
    if (tracer_->enabled()) {
      gpr_log(GPR_INFO, "request_router=%p: shutting down lb_policy=%p", this,
              lb_policy_.get());
    }
    grpc_pollset_set_del_pollset_set(lb_policy_->interested_parties(),
                                     interested_parties_);
    lb_policy_.reset();
  }
  if (resolver_ != nullptr) {
    // This should never happen; it can only be triggered by a resolver
    // implementation spotaneously deciding to report shutdown without
    // being orphaned.  This code is included just to be defensive.
    if (tracer_->enabled()) {
      gpr_log(GPR_INFO,
              "request_router=%p: spontaneous shutdown from resolver %p", this,
              resolver_.get());
    }
    resolver_.reset();
    SetConnectivityStateLocked(GRPC_CHANNEL_SHUTDOWN,
                               GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
                                   "Resolver spontaneous shutdown", &error, 1),
                               "resolver_spontaneous_shutdown");
  }
  grpc_closure_list_fail_all(&waiting_for_resolver_result_closures_,
                             GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
                                 "Channel disconnected", &error, 1));
  GRPC_CLOSURE_LIST_SCHED(&waiting_for_resolver_result_closures_);
  GRPC_CHANNEL_STACK_UNREF(owning_stack_, "resolver");
  grpc_channel_args_destroy(resolver_result_);
  resolver_result_ = nullptr;
  GRPC_ERROR_UNREF(error);
}

// Creates a new LB policy, replacing any previous one.
// If the new policy is created successfully, sets *connectivity_state and
// *connectivity_error to its initial connectivity state; otherwise,
// leaves them unchanged.
void RequestRouter::CreateNewLbPolicyLocked(
    const char* lb_policy_name, grpc_json* lb_config,
    grpc_connectivity_state* connectivity_state,
    grpc_error** connectivity_error, TraceStringVector* trace_strings) {
  LoadBalancingPolicy::Args lb_policy_args;
  lb_policy_args.combiner = combiner_;
  lb_policy_args.client_channel_factory = client_channel_factory_;
  lb_policy_args.args = resolver_result_;
  lb_policy_args.lb_config = lb_config;
  OrphanablePtr<LoadBalancingPolicy> new_lb_policy =
      LoadBalancingPolicyRegistry::CreateLoadBalancingPolicy(lb_policy_name,
                                                             lb_policy_args);
  if (GPR_UNLIKELY(new_lb_policy == nullptr)) {
    gpr_log(GPR_ERROR, "could not create LB policy \"%s\"", lb_policy_name);
    if (channelz_node_ != nullptr) {
      char* str;
      gpr_asprintf(&str, "Could not create LB policy \'%s\'", lb_policy_name);
      trace_strings->push_back(str);
    }
  } else {
    if (tracer_->enabled()) {
      gpr_log(GPR_INFO, "request_router=%p: created new LB policy \"%s\" (%p)",
              this, lb_policy_name, new_lb_policy.get());
    }
    if (channelz_node_ != nullptr) {
      char* str;
      gpr_asprintf(&str, "Created new LB policy \'%s\'", lb_policy_name);
      trace_strings->push_back(str);
    }
    // Swap out the LB policy and update the fds in interested_parties_.
    if (lb_policy_ != nullptr) {
      if (tracer_->enabled()) {
        gpr_log(GPR_INFO, "request_router=%p: shutting down lb_policy=%p", this,
                lb_policy_.get());
      }
      grpc_pollset_set_del_pollset_set(lb_policy_->interested_parties(),
                                       interested_parties_);
      lb_policy_->HandOffPendingPicksLocked(new_lb_policy.get());
    }
    lb_policy_ = std::move(new_lb_policy);
    grpc_pollset_set_add_pollset_set(lb_policy_->interested_parties(),
                                     interested_parties_);
    // Create re-resolution request handler for the new LB policy.  It
    // will delete itself when no longer needed.
    New<ReresolutionRequestHandler>(this, lb_policy_.get(), owning_stack_,
                                    combiner_);
    // Get the new LB policy's initial connectivity state and start a
    // connectivity watch.
    GRPC_ERROR_UNREF(*connectivity_error);
    *connectivity_state =
        lb_policy_->CheckConnectivityLocked(connectivity_error);
    if (exit_idle_when_lb_policy_arrives_) {
      lb_policy_->ExitIdleLocked();
      exit_idle_when_lb_policy_arrives_ = false;
    }
    // Create new watcher.  It will delete itself when done.
    New<LbConnectivityWatcher>(this, *connectivity_state, lb_policy_.get(),
                               owning_stack_, combiner_);
  }
}

void RequestRouter::MaybeAddTraceMessagesForAddressChangesLocked(
    TraceStringVector* trace_strings) {
  const ServerAddressList* addresses =
      FindServerAddressListChannelArg(resolver_result_);
  const bool resolution_contains_addresses =
      addresses != nullptr && addresses->size() > 0;
  if (!resolution_contains_addresses &&
      previous_resolution_contained_addresses_) {
    trace_strings->push_back(gpr_strdup("Address list became empty"));
  } else if (resolution_contains_addresses &&
             !previous_resolution_contained_addresses_) {
    trace_strings->push_back(gpr_strdup("Address list became non-empty"));
  }
  previous_resolution_contained_addresses_ = resolution_contains_addresses;
}

void RequestRouter::ConcatenateAndAddChannelTraceLocked(
    TraceStringVector* trace_strings) const {
  if (!trace_strings->empty()) {
    gpr_strvec v;
    gpr_strvec_init(&v);
    gpr_strvec_add(&v, gpr_strdup("Resolution event: "));
    bool is_first = 1;
    for (size_t i = 0; i < trace_strings->size(); ++i) {
      if (!is_first) gpr_strvec_add(&v, gpr_strdup(", "));
      is_first = false;
      gpr_strvec_add(&v, (*trace_strings)[i]);
    }
    char* flat;
    size_t flat_len = 0;
    flat = gpr_strvec_flatten(&v, &flat_len);
    channelz_node_->AddTraceEvent(
        grpc_core::channelz::ChannelTrace::Severity::Info,
        grpc_slice_new(flat, flat_len, gpr_free));
    gpr_strvec_destroy(&v);
  }
}

// Callback invoked when a resolver result is available.
void RequestRouter::OnResolverResultChangedLocked(void* arg,
                                                  grpc_error* error) {
  RequestRouter* self = static_cast<RequestRouter*>(arg);
  if (self->tracer_->enabled()) {
    const char* disposition =
        self->resolver_result_ != nullptr
            ? ""
            : (error == GRPC_ERROR_NONE ? " (transient error)"
                                        : " (resolver shutdown)");
    gpr_log(GPR_INFO,
            "request_router=%p: got resolver result: resolver_result=%p "
            "error=%s%s",
            self, self->resolver_result_, grpc_error_string(error),
            disposition);
  }
  // Handle shutdown.
  if (error != GRPC_ERROR_NONE || self->resolver_ == nullptr) {
    self->OnResolverShutdownLocked(GRPC_ERROR_REF(error));
    return;
  }
  // Data used to set the channel's connectivity state.
  bool set_connectivity_state = true;
  // We only want to trace the address resolution in the follow cases:
  // (a) Address resolution resulted in service config change.
  // (b) Address resolution that causes number of backends to go from
  //     zero to non-zero.
  // (c) Address resolution that causes number of backends to go from
  //     non-zero to zero.
  // (d) Address resolution that causes a new LB policy to be created.
  //
  // we track a list of strings to eventually be concatenated and traced.
  TraceStringVector trace_strings;
  grpc_connectivity_state connectivity_state = GRPC_CHANNEL_TRANSIENT_FAILURE;
  grpc_error* connectivity_error =
      GRPC_ERROR_CREATE_FROM_STATIC_STRING("No load balancing policy");
  // resolver_result_ will be null in the case of a transient
  // resolution error.  In that case, we don't have any new result to
  // process, which means that we keep using the previous result (if any).
  if (self->resolver_result_ == nullptr) {
    if (self->tracer_->enabled()) {
      gpr_log(GPR_INFO, "request_router=%p: resolver transient failure", self);
    }
    // Don't override connectivity state if we already have an LB policy.
    if (self->lb_policy_ != nullptr) set_connectivity_state = false;
  } else {
    // Parse the resolver result.
    const char* lb_policy_name = nullptr;
    grpc_json* lb_policy_config = nullptr;
    const bool service_config_changed = self->process_resolver_result_(
        self->process_resolver_result_user_data_, *self->resolver_result_,
        &lb_policy_name, &lb_policy_config);
    GPR_ASSERT(lb_policy_name != nullptr);
    // Check to see if we're already using the right LB policy.
    const bool lb_policy_name_changed =
        self->lb_policy_ == nullptr ||
        strcmp(self->lb_policy_->name(), lb_policy_name) != 0;
    if (self->lb_policy_ != nullptr && !lb_policy_name_changed) {
      // Continue using the same LB policy.  Update with new addresses.
      if (self->tracer_->enabled()) {
        gpr_log(GPR_INFO,
                "request_router=%p: updating existing LB policy \"%s\" (%p)",
                self, lb_policy_name, self->lb_policy_.get());
      }
      self->lb_policy_->UpdateLocked(*self->resolver_result_, lb_policy_config);
      // No need to set the channel's connectivity state; the existing
      // watch on the LB policy will take care of that.
      set_connectivity_state = false;
    } else {
      // Instantiate new LB policy.
      self->CreateNewLbPolicyLocked(lb_policy_name, lb_policy_config,
                                    &connectivity_state, &connectivity_error,
                                    &trace_strings);
    }
    // Add channel trace event.
    if (self->channelz_node_ != nullptr) {
      if (service_config_changed) {
        // TODO(ncteisen): might be worth somehow including a snippet of the
        // config in the trace, at the risk of bloating the trace logs.
        trace_strings.push_back(gpr_strdup("Service config changed"));
      }
      self->MaybeAddTraceMessagesForAddressChangesLocked(&trace_strings);
      self->ConcatenateAndAddChannelTraceLocked(&trace_strings);
    }
    // Clean up.
    grpc_channel_args_destroy(self->resolver_result_);
    self->resolver_result_ = nullptr;
  }
  // Set the channel's connectivity state if needed.
  if (set_connectivity_state) {
    self->SetConnectivityStateLocked(connectivity_state, connectivity_error,
                                     "resolver_result");
  } else {
    GRPC_ERROR_UNREF(connectivity_error);
  }
  // Invoke closures that were waiting for results and renew the watch.
  GRPC_CLOSURE_LIST_SCHED(&self->waiting_for_resolver_result_closures_);
  self->resolver_->NextLocked(&self->resolver_result_,
                              &self->on_resolver_result_changed_);
}

void RequestRouter::RouteCallLocked(Request* request) {
  GPR_ASSERT(request->pick_.connected_subchannel == nullptr);
  request->request_router_ = this;
  if (lb_policy_ != nullptr) {
    // We already have resolver results, so process the service config
    // and start an LB pick.
    request->ProcessServiceConfigAndStartLbPickLocked();
  } else if (resolver_ == nullptr) {
    GRPC_CLOSURE_RUN(request->on_route_done_,
                     GRPC_ERROR_CREATE_FROM_STATIC_STRING("Disconnected"));
  } else {
    // We do not yet have an LB policy, so wait for a resolver result.
    if (!started_resolving_) {
      StartResolvingLocked();
    }
    // Create a new waiter, which will delete itself when done.
    New<Request::ResolverResultWaiter>(request);
    // Add the request's polling entity to the request_router's
    // interested_parties, so that the I/O of the resolver can be done
    // under it.  It will be removed in LbPickDoneLocked().
    request->MaybeAddCallToInterestedPartiesLocked();
  }
}

void RequestRouter::ShutdownLocked(grpc_error* error) {
  if (resolver_ != nullptr) {
    SetConnectivityStateLocked(GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error),
                               "disconnect");
    resolver_.reset();
    if (!started_resolving_) {
      grpc_closure_list_fail_all(&waiting_for_resolver_result_closures_,
                                 GRPC_ERROR_REF(error));
      GRPC_CLOSURE_LIST_SCHED(&waiting_for_resolver_result_closures_);
    }
    if (lb_policy_ != nullptr) {
      grpc_pollset_set_del_pollset_set(lb_policy_->interested_parties(),
                                       interested_parties_);
      lb_policy_.reset();
    }
  }
  GRPC_ERROR_UNREF(error);
}

grpc_connectivity_state RequestRouter::GetConnectivityState() {
  return grpc_connectivity_state_check(&state_tracker_);
}

void RequestRouter::NotifyOnConnectivityStateChange(
    grpc_connectivity_state* state, grpc_closure* closure) {
  grpc_connectivity_state_notify_on_state_change(&state_tracker_, state,
                                                 closure);
}

void RequestRouter::ExitIdleLocked() {
  if (lb_policy_ != nullptr) {
    lb_policy_->ExitIdleLocked();
  } else {
    exit_idle_when_lb_policy_arrives_ = true;
    if (!started_resolving_ && resolver_ != nullptr) {
      StartResolvingLocked();
    }
  }
}

void RequestRouter::ResetConnectionBackoffLocked() {
  if (resolver_ != nullptr) {
    resolver_->ResetBackoffLocked();
    resolver_->RequestReresolutionLocked();
  }
  if (lb_policy_ != nullptr) {
    lb_policy_->ResetBackoffLocked();
  }
}

}  // namespace grpc_core