src/core/ext/filters/client_channel/lb_policy.cc

   1 /*
   2  *
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <grpc/support/port_platform.h>
  20
  21 #include "src/core/ext/filters/client_channel/lb_policy.h"
  22
  23 #include "src/core/ext/filters/client_channel/lb_policy_registry.h"
  24 #include "src/core/lib/iomgr/combiner.h"
  25
  26 namespace grpc_core {
  27
  28 DebugOnlyTraceFlag grpc_trace_lb_policy_refcount(false, "lb_policy_refcount");
  29
  30 //
  31 // LoadBalancingPolicy
  32 //
  33
  34 LoadBalancingPolicy::LoadBalancingPolicy(Args args, intptr_t initial_refcount)
  35     : InternallyRefCounted(&grpc_trace_lb_policy_refcount, initial_refcount),
  36       combiner_(GRPC_COMBINER_REF(args.combiner, "lb_policy")),
  37       interested_parties_(grpc_pollset_set_create()),
  38       channel_control_helper_(std::move(args.channel_control_helper)) {}
  39
  40 LoadBalancingPolicy::~LoadBalancingPolicy() {
  41   grpc_pollset_set_destroy(interested_parties_);
  42   GRPC_COMBINER_UNREF(combiner_, "lb_policy");
  43 }
  44
  45 void LoadBalancingPolicy::Orphan() {
  46   // Invoke ShutdownAndUnrefLocked() inside of the combiner.
  47   // TODO(roth): Is this actually needed?  We should already be in the
  48   // combiner here.  Note that if we directly call ShutdownLocked(),
  49   // then we can probably remove the hack whereby the helper is
  50   // destroyed at shutdown instead of at destruction.
  51   GRPC_CLOSURE_SCHED(
  52       GRPC_CLOSURE_CREATE(&LoadBalancingPolicy::ShutdownAndUnrefLocked, this,
  53                           grpc_combiner_scheduler(combiner_)),
  54       GRPC_ERROR_NONE);
  55 }
  56
  57 void LoadBalancingPolicy::ShutdownAndUnrefLocked(void* arg,
  58                                                  grpc_error* ignored) {
  59   LoadBalancingPolicy* policy = static_cast<LoadBalancingPolicy*>(arg);
  60   policy->ShutdownLocked();
  61   policy->channel_control_helper_.reset();
  62   policy->Unref();
  63 }
  64
  65 grpc_json* LoadBalancingPolicy::ParseLoadBalancingConfig(
  66     const grpc_json* lb_config_array) {
  67   if (lb_config_array == nullptr || lb_config_array->type != GRPC_JSON_ARRAY) {
  68     return nullptr;
  69   }
  70   // Find the first LB policy that this client supports.
  71   for (const grpc_json* lb_config = lb_config_array->child;
  72        lb_config != nullptr; lb_config = lb_config->next) {
  73     if (lb_config->type != GRPC_JSON_OBJECT) return nullptr;
  74     grpc_json* policy = nullptr;
  75     for (grpc_json* field = lb_config->child; field != nullptr;
  76          field = field->next) {
  77       if (field->key == nullptr || field->type != GRPC_JSON_OBJECT)
  78         return nullptr;
  79       if (policy != nullptr) return nullptr;  // Violate "oneof" type.
  80       policy = field;
  81     }
  82     if (policy == nullptr) return nullptr;
  83     // If we support this policy, then select it.
  84     if (LoadBalancingPolicyRegistry::LoadBalancingPolicyExists(policy->key)) {
  85       return policy;
  86     }
  87   }
  88   return nullptr;
  89 }
  90
  91 //
  92 // LoadBalancingPolicy::UpdateArgs
  93 //
  94
  95 LoadBalancingPolicy::UpdateArgs::UpdateArgs(const UpdateArgs& other) {
  96   addresses = other.addresses;
  97   config = other.config;
  98   args = grpc_channel_args_copy(other.args);
  99 }
 100
 101 LoadBalancingPolicy::UpdateArgs::UpdateArgs(UpdateArgs&& other) {
 102   addresses = std::move(other.addresses);
 103   config = std::move(other.config);
 104   // TODO(roth): Use std::move() once channel args is converted to C++.
 105   args = other.args;
 106   other.args = nullptr;
 107 }
 108
 109 LoadBalancingPolicy::UpdateArgs& LoadBalancingPolicy::UpdateArgs::operator=(
 110     const UpdateArgs& other) {
 111   addresses = other.addresses;
 112   config = other.config;
 113   grpc_channel_args_destroy(args);
 114   args = grpc_channel_args_copy(other.args);
 115   return *this;
 116 }
 117
 118 LoadBalancingPolicy::UpdateArgs& LoadBalancingPolicy::UpdateArgs::operator=(
 119     UpdateArgs&& other) {
 120   addresses = std::move(other.addresses);
 121   config = std::move(other.config);
 122   // TODO(roth): Use std::move() once channel args is converted to C++.
 123   grpc_channel_args_destroy(args);
 124   args = other.args;
 125   other.args = nullptr;
 126   return *this;
 127 }
 128
 129 //
 130 // LoadBalancingPolicy::QueuePicker
 131 //
 132
 133 LoadBalancingPolicy::PickResult LoadBalancingPolicy::QueuePicker::Pick(
 134     PickArgs* pick, grpc_error** error) {
 135   // We invoke the parent's ExitIdleLocked() via a closure instead
 136   // of doing it directly here, for two reasons:
 137   // 1. ExitIdleLocked() may cause the policy's state to change and
 138   //    a new picker to be delivered to the channel.  If that new
 139   //    picker is delivered before ExitIdleLocked() returns, then by
 140   //    the time this function returns, the pick will already have
 141   //    been processed, and we'll be trying to re-process the same
 142   //    pick again, leading to a crash.
 143   // 2. In a subsequent PR, we will split the data plane and control
 144   //    plane synchronization into separate combiners, at which
 145   //    point this will need to hop from the data plane combiner into
 146   //    the control plane combiner.
 147   if (!exit_idle_called_) {
 148     exit_idle_called_ = true;
 149     parent_->Ref().release();  // ref held by closure.
 150     GRPC_CLOSURE_SCHED(
 151         GRPC_CLOSURE_CREATE(&CallExitIdle, parent_.get(),
 152                             grpc_combiner_scheduler(parent_->combiner())),
 153         GRPC_ERROR_NONE);
 154   }
 155   return PICK_QUEUE;
 156 }
 157
 158 void LoadBalancingPolicy::QueuePicker::CallExitIdle(void* arg,
 159                                                     grpc_error* error) {
 160   LoadBalancingPolicy* parent = static_cast<LoadBalancingPolicy*>(arg);
 161   parent->ExitIdleLocked();
 162   parent->Unref();
 163 }
 164
 165 }  // namespace grpc_core