const Router::MetadataMatchCriteria* match_criteria) {

  if (match_criteria != nullptr) {

    for (const auto& selector : subset_selectors) {

      const auto& selector_keys = selector->selectorKeys();

      auto sub_match_criteria = match_criteria->filterMatchCriteria(selector_keys);

      if (sub_match_criteria != nullptr &&

          sub_match_criteria->metadataMatchCriteria().size() == selector_keys.size()) {

        return sub_match_criteria;

      }

    }

  }

  return nullptr;

}

    : lb_config_(lb_config), cluster_info_(cluster_info), stats_(stats), scope_(scope),

      runtime_(runtime), random_(random), time_source_(time_source),

      fallback_policy_(lb_config_.subsetInfo().fallbackPolicy()),

      metadata_fallback_policy_(lb_config_.subsetInfo().metadataFallbackPolicy()),

      default_subset_metadata_(lb_config_.subsetInfo().defaultSubset().fields().begin(),

                               lb_config_.subsetInfo().defaultSubset().fields().end()),

      subset_selectors_(lb_config_.subsetInfo().subsetSelectors()),

      original_priority_set_(priority_set), original_local_priority_set_(local_priority_set),

      locality_weight_aware_(lb_config_.subsetInfo().localityWeightAware()),

      scale_locality_weight_(lb_config_.subsetInfo().scaleLocalityWeight()),

      list_as_any_(lb_config_.subsetInfo().listAsAny()),

      allow_redundant_keys_(lb_config_.subsetInfo().allowRedundantKeys()) {

  ASSERT(lb_config_.subsetInfo().isEnabled());

  if (fallback_policy_ != envoy::config::cluster::v3::Cluster::LbSubsetConfig::NO_FALLBACK) {

    if (fallback_policy_ == envoy::config::cluster::v3::Cluster::LbSubsetConfig::ANY_ENDPOINT) {

      ENVOY_LOG(debug, "subset lb: creating any-endpoint fallback load balancer");

      initSubsetAnyOnce();

      fallback_subset_ = subset_any_;

    } else {

      ENVOY_LOG(debug, "subset lb: creating fallback load balancer for {}",

                describeMetadata(default_subset_metadata_));

      initSubsetDefaultOnce();

      fallback_subset_ = subset_default_;

    }

  }

  if (lb_config_.subsetInfo().panicModeAny()) {

    initSubsetAnyOnce();

    panic_mode_subset_ = subset_any_;

  }

  initSubsetSelectorMap();

  refreshSubsets();

  original_priority_set_callback_handle_ = priority_set.addPriorityUpdateCb(

      [this](uint32_t priority, const HostVector&, const HostVector&) {

        refreshSubsets(priority);

        purgeEmptySubsets(subsets_);

      });

}

SubsetLoadBalancer::~SubsetLoadBalancer() {

  forEachSubset(subsets_, [&](LbSubsetEntryPtr entry) {

    if (entry->active()) {

      stats_.lb_subsets_removed_.inc();

      stats_.lb_subsets_active_.dec();

    }

  });

}

void SubsetLoadBalancer::refreshSubsets() {

  for (auto& host_set : original_priority_set_.hostSetsPerPriority()) {

    update(host_set->priority(), host_set->hosts());

  }

}

void SubsetLoadBalancer::refreshSubsets(uint32_t priority) {

  const auto& host_sets = original_priority_set_.hostSetsPerPriority();

  ASSERT(priority < host_sets.size());

  update(priority, host_sets[priority]->hosts());

}

void SubsetLoadBalancer::initSubsetAnyOnce() {

  if (!subset_any_) {

    subset_any_ = std::make_shared<LbSubsetEntry>();

    subset_any_->lb_subset_ =

        std::make_unique<PriorityLbSubset>(*this, locality_weight_aware_, scale_locality_weight_);

  }

}

void SubsetLoadBalancer::initSubsetDefaultOnce() {

  if (!subset_default_) {

    subset_default_ = std::make_shared<LbSubsetEntry>();

    subset_default_->lb_subset_ =

        std::make_unique<PriorityLbSubset>(*this, locality_weight_aware_, scale_locality_weight_);

  }

}

void SubsetLoadBalancer::initSubsetSelectorMap() {

  selectors_ = std::make_shared<SubsetSelectorMap>();

  SubsetSelectorMapPtr selectors;

  for (const auto& subset_selector : subset_selectors_) {

    const auto& selector_keys = subset_selector->selectorKeys();

    const auto& selector_fallback_policy = subset_selector->fallbackPolicy();

    const auto& selector_fallback_keys_subset = subset_selector->fallbackKeysSubset();

    uint32_t pos = 0;

    selectors = selectors_;

    for (const auto& key : selector_keys) {

      const auto& selector_it = selectors->subset_keys_.find(key);

      pos++;

      if (selector_it == selectors->subset_keys_.end()) {

        selectors->subset_keys_.emplace(std::make_pair(key, std::make_shared<SubsetSelectorMap>()));

        const auto& child_selector = selectors->subset_keys_.find(key);

        if (pos == selector_keys.size()) {

          child_selector->second->fallback_params_.fallback_policy_ = selector_fallback_policy;

          child_selector->second->fallback_params_.fallback_keys_subset_ =

              &selector_fallback_keys_subset;

          initSelectorFallbackSubset(selector_fallback_policy);

        }

        selectors = child_selector->second;

      } else {

        selectors = selector_it->second;

      }

    }

    selectors = selectors_;

  }

}

        LbSubsetSelectorFallbackPolicy& fallback_policy) {

  if (fallback_policy ==

          envoy::config::cluster::v3::Cluster::LbSubsetConfig::LbSubsetSelector::ANY_ENDPOINT &&

      subset_any_ == nullptr) {

    ENVOY_LOG(debug, "subset lb: creating any-endpoint fallback load balancer for selector");

    initSubsetAnyOnce();

  } else if (fallback_policy == envoy::config::cluster::v3::Cluster::LbSubsetConfig::

                                    LbSubsetSelector::DEFAULT_SUBSET &&

             subset_default_ == nullptr) {

    ENVOY_LOG(debug, "subset lb: creating default subset fallback load balancer for selector");

    initSubsetDefaultOnce();

  }

}

HostSelectionResponse SubsetLoadBalancer::chooseHost(LoadBalancerContext* context) {

  if (metadata_fallback_policy_ !=

      envoy::config::cluster::v3::

          Cluster_LbSubsetConfig_LbSubsetMetadataFallbackPolicy_FALLBACK_LIST) {

    return chooseHostIteration(context);

  }

  const Protobuf::Value* metadata_fallbacks = getMetadataFallbackList(context);

  if (metadata_fallbacks == nullptr) {

    return chooseHostIteration(context);

  }

  LoadBalancerContextWrapper context_no_metadata_fallback = removeMetadataFallbackList(context);

  return chooseHostWithMetadataFallbacks(&context_no_metadata_fallback,

                                         metadata_fallbacks->list_value().values());

}

                                                    const MetadataFallbacks& metadata_fallbacks) {

  if (metadata_fallbacks.empty()) {

    return chooseHostIteration(context);

  }

  for (const auto& metadata_override : metadata_fallbacks) {

    LoadBalancerContextWrapper context_wrapper(context, metadata_override.struct_value());

    const auto host = chooseHostIteration(&context_wrapper);

    if (host) {

      return host;

    }

  }

}

SubsetLoadBalancer::removeMetadataFallbackList(LoadBalancerContext* context) {

  ASSERT(context->metadataMatchCriteria());

  const auto& match_criteria = context->metadataMatchCriteria()->metadataMatchCriteria();

  std::set<std::string> to_preserve;

  for (const auto& criterion : match_criteria) {

    if (criterion->name() != Config::MetadataEnvoyLbKeys::get().FALLBACK_LIST) {

      to_preserve.emplace(criterion->name());

    }

  }

  return {context, to_preserve};

}

SubsetLoadBalancer::getMetadataFallbackList(LoadBalancerContext* context) const {

  if (context == nullptr) {

    return nullptr;

  }

  const auto& match_criteria = context->metadataMatchCriteria();

  if (match_criteria == nullptr) {

    return nullptr;

  }

  for (const auto& criterion : match_criteria->metadataMatchCriteria()) {

    if (criterion->name() == Config::MetadataEnvoyLbKeys::get().FALLBACK_LIST) {

      return &criterion->value().value();

    } // TODO(MarcinFalkowski): optimization: stop iteration when lexically after 'fallback_list'

  }

  return nullptr;

}

HostConstSharedPtr SubsetLoadBalancer::chooseHostIteration(LoadBalancerContext* context) {

  if (context) {

    LoadBalancerContext* actual_used_context = context;

    std::unique_ptr<LoadBalancerContextWrapper> actual_used_context_wrapper;

    if (allow_redundant_keys_) {

      auto actual_used_criteria =

          filterCriteriaBySelectors(subset_selectors_, context->metadataMatchCriteria());

      if (actual_used_criteria != nullptr) {

        actual_used_context_wrapper =

            std::make_unique<LoadBalancerContextWrapper>(context, std::move(actual_used_criteria));

        actual_used_context = actual_used_context_wrapper.get();

      }

    }

    bool host_chosen;

    HostConstSharedPtr host = tryChooseHostFromContext(actual_used_context, host_chosen);

    if (host_chosen) {

      return host;

    }

    absl::optional<SubsetSelectorFallbackParamsRef> selector_fallback_params =

        tryFindSelectorFallbackParams(actual_used_context);

    if (selector_fallback_params &&

        selector_fallback_params->get().fallback_policy_ !=

            envoy::config::cluster::v3::Cluster::LbSubsetConfig::LbSubsetSelector::NOT_DEFINED) {

      return chooseHostForSelectorFallbackPolicy(*selector_fallback_params, actual_used_context);

    }

  }

  if (fallback_subset_ == nullptr) {

    return nullptr;

  }

  HostSelectionResponse host_response = fallback_subset_->lb_subset_->chooseHost(context);

  if (host_response.host || host_response.cancelable) {

    stats_.lb_subsets_fallback_.inc();

    return host_response.host;

  }

  if (panic_mode_subset_ != nullptr) {

    HostSelectionResponse host_response = panic_mode_subset_->lb_subset_->chooseHost(context);

    if (host_response.host || host_response.cancelable) {

      stats_.lb_subsets_fallback_panic_.inc();

      return host_response.host;

    }

  }

}

SubsetLoadBalancer::tryFindSelectorFallbackParams(LoadBalancerContext* context) {

  const Router::MetadataMatchCriteria* match_criteria = context->metadataMatchCriteria();

  if (!match_criteria) {

    return absl::nullopt;

  }

  const auto match_criteria_vec = match_criteria->metadataMatchCriteria();

  SubsetSelectorMapPtr selectors = selectors_;

  if (selectors == nullptr) {

  for (uint32_t i = 0; i < match_criteria_vec.size(); i++) {

    const Router::MetadataMatchCriterion& match_criterion = *match_criteria_vec[i];

    const auto& subset_it = selectors->subset_keys_.find(match_criterion.name());

    if (subset_it == selectors->subset_keys_.end()) {

      break;

    }

    if (i + 1 == match_criteria_vec.size()) {

      return subset_it->second->fallback_params_;

    }

    selectors = subset_it->second;

  }

  return absl::nullopt;

}

    const SubsetSelectorFallbackParams& fallback_params, LoadBalancerContext* context) {

  const auto& fallback_policy = fallback_params.fallback_policy_;

  if (fallback_policy ==

          envoy::config::cluster::v3::Cluster::LbSubsetConfig::LbSubsetSelector::ANY_ENDPOINT &&

      subset_any_ != nullptr) {

    return Upstream::LoadBalancer::onlyAllowSynchronousHostSelection(

        subset_any_->lb_subset_->chooseHost(context));

  } else if (fallback_policy == envoy::config::cluster::v3::Cluster::LbSubsetConfig::

                                    LbSubsetSelector::DEFAULT_SUBSET &&

             subset_default_ != nullptr) {

    return Upstream::LoadBalancer::onlyAllowSynchronousHostSelection(

        subset_default_->lb_subset_->chooseHost(context));

  } else if (fallback_policy ==

             envoy::config::cluster::v3::Cluster::LbSubsetConfig::LbSubsetSelector::KEYS_SUBSET) {

    ASSERT(fallback_params.fallback_keys_subset_);

    auto filtered_context = std::make_unique<LoadBalancerContextWrapper>(

        context, *fallback_params.fallback_keys_subset_);

    return chooseHostIteration(filtered_context.get());

  } else {

    return {nullptr};

  }

}

                                                                bool& host_chosen) {

  host_chosen = false;

  const Router::MetadataMatchCriteria* match_criteria = context->metadataMatchCriteria();

  if (!match_criteria) {

    return {nullptr};

  }

  LbSubsetEntryPtr entry = findSubset(match_criteria->metadataMatchCriteria());

  if (entry == nullptr || !entry->active()) {

    return {nullptr};

  }

  host_chosen = true;

  stats_.lb_subsets_selected_.inc();

  return Upstream::LoadBalancer::onlyAllowSynchronousHostSelection(

      entry->lb_subset_->chooseHost(context));

}

    const std::vector<Router::MetadataMatchCriterionConstSharedPtr>& match_criteria) {

  const LbSubsetMap* subsets = &subsets_;

  for (uint32_t i = 0; i < match_criteria.size(); i++) {

    const Router::MetadataMatchCriterion& match_criterion = *match_criteria[i];

    const auto& subset_it = subsets->find(match_criterion.name());

    if (subset_it == subsets->end()) {

      break;

    }

    const ValueSubsetMap& vs_map = subset_it->second;

    const auto& vs_it = vs_map.find(match_criterion.value());

    if (vs_it == vs_map.end()) {

      break;

    }

    const LbSubsetEntryPtr& entry = vs_it->second;

    if (i + 1 == match_criteria.size()) {

      return entry;

    }

    subsets = &entry->children_;

  }

  return nullptr;

}

void SubsetLoadBalancer::updateFallbackSubset(uint32_t priority, const HostVector& all_hosts) {

  auto update_func = [priority, &all_hosts](LbSubsetPtr& subset, const HostPredicate& predicate) {

    for (const auto& host : all_hosts) {

      if (predicate(*host)) {

        subset->pushHost(priority, host);

      }

    }

    subset->finalize(priority);

  };

  if (subset_any_ != nullptr) {

    update_func(subset_any_->lb_subset_, [](const Host&) { return true; });

  }

  if (subset_default_ != nullptr) {

    HostPredicate predicate = std::bind(&SubsetLoadBalancer::hostMatches, this,

                                        default_subset_metadata_, std::placeholders::_1);

    update_func(subset_default_->lb_subset_, predicate);

  }

  if (fallback_subset_ == nullptr) {

    ENVOY_LOG(debug, "subset lb: fallback load balancer disabled");

    return;

  }

  ASSERT(panic_mode_subset_ == nullptr || panic_mode_subset_ == subset_any_);

}

void SubsetLoadBalancer::initLbSubsetEntryOnce(LbSubsetEntryPtr& entry, bool single_host_subset) {

  ASSERT(entry != nullptr);

  if (entry->initialized()) {

    return;

  }

  if (single_host_subset) {

    entry->lb_subset_ = std::make_unique<SingleHostLbSubset>();

    entry->single_host_subset_ = true;

  } else {

    entry->lb_subset_ =

        std::make_unique<PriorityLbSubset>(*this, locality_weight_aware_, scale_locality_weight_);

    entry->single_host_subset_ = false;

  }

  stats_.lb_subsets_active_.inc();

  stats_.lb_subsets_created_.inc();

}

void SubsetLoadBalancer::processSubsets(uint32_t priority, const HostVector& all_hosts) {

  absl::flat_hash_set<const LbSubsetEntry*> single_host_entries;

  uint64_t collision_count_of_single_host_entries{};

  for (const auto& host : all_hosts) {

    for (const auto& subset_selector : subset_selectors_) {

      const auto& keys = subset_selector->selectorKeys();

      std::vector<SubsetMetadata> all_kvs = extractSubsetMetadata(keys, *host);

      for (const auto& kvs : all_kvs) {

        auto entry = findOrCreateLbSubsetEntry(subsets_, kvs, 0);

        initLbSubsetEntryOnce(entry, subset_selector->singleHostPerSubset());

        if (entry->single_host_subset_) {

          if (single_host_entries.contains(entry.get())) {

            collision_count_of_single_host_entries++;

            continue;

          }

          single_host_entries.emplace(entry.get());

        }

        entry->lb_subset_->pushHost(priority, host);

      }

    }

  }

  if (single_duplicate_stat_ == nullptr) {

    Stats::StatNameManagedStorage name_storage("lb_subsets_single_host_per_subset_duplicate",

                                               scope_.symbolTable());

    single_duplicate_stat_ = &Stats::Utility::gaugeFromElements(

        scope_, {name_storage.statName()}, Stats::Gauge::ImportMode::Accumulate);

  }

  single_duplicate_stat_->set(collision_count_of_single_host_entries);

  forEachSubset(subsets_, [priority](LbSubsetEntryPtr entry) {

    if (entry->initialized()) {

      entry->lb_subset_->finalize(priority);

    }

  });

}

void SubsetLoadBalancer::update(uint32_t priority, const HostVector& all_hosts) {

  updateFallbackSubset(priority, all_hosts);

  processSubsets(priority, all_hosts);

}

bool SubsetLoadBalancer::hostMatches(const SubsetMetadata& kvs, const Host& host) {

  return Config::Metadata::metadataLabelMatch(

      kvs, host.metadata().get(), Config::MetadataFilters::get().ENVOY_LB, list_as_any_);

}

                                          const Host& host) {

  std::vector<SubsetMetadata> all_kvs;

  if (!host.metadata()) {

    return all_kvs;

  }

  const envoy::config::core::v3::Metadata& metadata = *host.metadata();

  const auto& filter_it = metadata.filter_metadata().find(Config::MetadataFilters::get().ENVOY_LB);

  if (filter_it == metadata.filter_metadata().end()) {

    return all_kvs;

  }

  const auto& fields = filter_it->second.fields();

  for (const auto& key : subset_keys) {

    const auto it = fields.find(key);

    if (it == fields.end()) {

      all_kvs.clear();

      break;

    }

    if (list_as_any_ && it->second.kind_case() == Protobuf::Value::kListValue) {

      if (all_kvs.empty()) {

        for (const auto& v : it->second.list_value().values()) {

          all_kvs.emplace_back(SubsetMetadata({make_pair(key, v)}));

        }

      } else {

        std::vector<SubsetMetadata> new_kvs;

        for (const auto& kvs : all_kvs) {

          for (const auto& v : it->second.list_value().values()) {

            auto kv_copy = kvs;

            kv_copy.emplace_back(make_pair(key, v));

            new_kvs.emplace_back(kv_copy);

          }

        }

        all_kvs = new_kvs;

      }

    } else {

      if (all_kvs.empty()) {

        all_kvs.emplace_back(SubsetMetadata({std::make_pair(key, it->second)}));

      } else {

        for (auto& kvs : all_kvs) {

          kvs.emplace_back(std::make_pair(key, it->second));

        }

      }

    }

  }

  return all_kvs;

}

std::string SubsetLoadBalancer::describeMetadata(const SubsetLoadBalancer::SubsetMetadata& kvs) {

  if (kvs.empty()) {

    return "<no metadata>";

  }

  std::ostringstream buf;

  bool first = true;

  for (const auto& it : kvs) {

    if (!first) {

      buf << ", ";

    } else {

      first = false;

    }

    const Protobuf::Value& value = it.second;

    buf << it.first << "=" << MessageUtil::getJsonStringFromMessageOrError(value);

  }

  return buf.str();

}

                                              uint32_t idx) {

  ASSERT(idx < kvs.size());

  const std::string& name = kvs[idx].first;

  const Protobuf::Value& pb_value = kvs[idx].second;

  const HashedValue value(pb_value);

  LbSubsetEntryPtr entry;

  const auto kv_it = subsets.find(name);

  if (kv_it != subsets.end()) {

    ValueSubsetMap& value_subset_map = kv_it->second;

    const auto vs_it = value_subset_map.find(value);

    if (vs_it != value_subset_map.end()) {

      entry = vs_it->second;

    }

  }

  if (!entry) {

    entry = std::make_shared<LbSubsetEntry>();

    if (kv_it != subsets.end()) {

      ValueSubsetMap& value_subset_map = kv_it->second;

      value_subset_map.emplace(value, entry);

    } else {

      ValueSubsetMap value_subset_map = {{value, entry}};

      subsets.emplace(name, value_subset_map);

    }

  }

  idx++;

  if (idx == kvs.size()) {

    return entry;

  }

  return findOrCreateLbSubsetEntry(entry->children_, kvs, idx);

}

                                       std::function<void(LbSubsetEntryPtr&)> cb) {

  for (auto& vsm : subsets) {

    for (auto& em : vsm.second) {

      LbSubsetEntryPtr entry = em.second;

      cb(entry);

      forEachSubset(entry->children_, cb);

    }

  }

}

void SubsetLoadBalancer::purgeEmptySubsets(LbSubsetMap& subsets) {

  for (auto subset_it = subsets.begin(); subset_it != subsets.end();) {

    for (auto it = subset_it->second.begin(); it != subset_it->second.end();) {

      LbSubsetEntryPtr entry = it->second;

      purgeEmptySubsets(entry->children_);

      if (entry->active() || entry->hasChildren()) {

        it++;

        continue;

      }

      if (entry->initialized()) {

        stats_.lb_subsets_active_.dec();

        stats_.lb_subsets_removed_.inc();

      }

      auto next_it = std::next(it);

      subset_it->second.erase(it);

      it = next_it;

    }

    if (subset_it->second.empty()) {

      auto next_subset_it = std::next(subset_it);

      subsets.erase(subset_it);

      subset_it = next_subset_it;

    } else {

      subset_it++;

    }

  }

}

    : original_priority_set_(subset_lb.original_priority_set_),

      original_local_priority_set_(subset_lb.original_local_priority_set_),

      locality_weight_aware_(locality_weight_aware), scale_locality_weight_(scale_locality_weight) {

  getOrCreateHostSet(0);

  thread_aware_lb_ =

      subset_lb.lb_config_.createLoadBalancer(subset_lb.cluster_info_, *this, subset_lb.runtime_,

                                              subset_lb.random_, subset_lb.time_source_);

  ASSERT(thread_aware_lb_ != nullptr);

  THROW_IF_NOT_OK(thread_aware_lb_->initialize());

  lb_ = thread_aware_lb_->factory()->create({*this, original_local_priority_set_});

  triggerCallbacks();

}

                                                const HostVector& hosts_removed) {

  auto cached_predicate = [&matching_hosts](const auto& host) {

    return matching_hosts.count(&host) == 1;

  };

  auto hosts = std::make_shared<HostVector>();

  hosts->reserve(original_host_set_.hosts().size());

  for (const auto& host : original_host_set_.hosts()) {

    if (cached_predicate(*host)) {

      hosts->emplace_back(host);

    }

  }

  auto healthy_hosts = std::make_shared<HealthyHostVector>();

  healthy_hosts->get().reserve(original_host_set_.healthyHosts().size());

  for (const auto& host : original_host_set_.healthyHosts()) {

    if (cached_predicate(*host)) {

      healthy_hosts->get().emplace_back(host);

    }

  }

  auto degraded_hosts = std::make_shared<DegradedHostVector>();

  degraded_hosts->get().reserve(original_host_set_.degradedHosts().size());

  for (const auto& host : original_host_set_.degradedHosts()) {

  auto excluded_hosts = std::make_shared<ExcludedHostVector>();

  excluded_hosts->get().reserve(original_host_set_.excludedHosts().size());

  for (const auto& host : original_host_set_.excludedHosts()) {

  HostsPerLocalityConstSharedPtr hosts_per_locality;

  if (original_host_set_.hostsPerLocality().get().size() == 1) {

    hosts_per_locality = std::make_shared<HostsPerLocalityImpl>(

        *hosts, original_host_set_.hostsPerLocality().hasLocalLocality());

  } else {

    hosts_per_locality = original_host_set_.hostsPerLocality().filter({cached_predicate})[0];

  }

  auto healthy_hosts_per_locality =

      original_host_set_.healthyHostsPerLocality().filter({cached_predicate})[0];

  auto degraded_hosts_per_locality =

      original_host_set_.degradedHostsPerLocality().filter({cached_predicate})[0];

  auto excluded_hosts_per_locality =

      original_host_set_.excludedHostsPerLocality().filter({cached_predicate})[0];

  HostSetImpl::updateHosts(

      HostSetImpl::updateHostsParams(

          hosts, hosts_per_locality, healthy_hosts, healthy_hosts_per_locality, degraded_hosts,

          degraded_hosts_per_locality, excluded_hosts, excluded_hosts_per_locality),

      determineLocalityWeights(*hosts_per_locality), hosts_added, hosts_removed,

      original_host_set_.weightedPriorityHealth(), original_host_set_.overprovisioningFactor());

}

    const HostsPerLocality& hosts_per_locality) const {

  if (locality_weight_aware_) {

    if (scale_locality_weight_) {

      const auto& original_hosts_per_locality = original_host_set_.hostsPerLocality().get();

      if (!original_host_set_.localityWeights()) {

        return {};

      }

      const auto& original_weights = *original_host_set_.localityWeights();

      auto scaled_locality_weights = std::make_shared<LocalityWeights>(original_weights.size());

      for (uint32_t i = 0; i < original_weights.size(); ++i) {

        if (original_hosts_per_locality[i].empty()) {

        (*scaled_locality_weights)[i] =

            std::round(float((original_weights[i] * hosts_per_locality.get()[i].size())) /

                       original_hosts_per_locality[i].size());

      }

      return scaled_locality_weights;

    } else {

      return original_host_set_.localityWeights();

    }

  }

  return {};

}

    absl::optional<uint32_t> overprovisioning_factor) {

  RELEASE_ASSERT(priority < original_priority_set_.hostSetsPerPriority().size(), "");

  const HostSetPtr& host_set = original_priority_set_.hostSetsPerPriority()[priority];

  ASSERT(!overprovisioning_factor.has_value() ||

         overprovisioning_factor.value() == host_set->overprovisioningFactor());

  ASSERT(!weighted_priority_health.has_value() ||

         weighted_priority_health.value() == host_set->weightedPriorityHealth());

  return HostSetImplPtr{

      new HostSubsetImpl(*host_set, locality_weight_aware_, scale_locality_weight_)};

}

                                                    const HostVector& hosts_removed) {

  const auto& host_subset = getOrCreateHostSet(priority);

  updateSubset(priority, matching_hosts, hosts_added, hosts_removed);

  if (host_subset.hosts().empty() != empty_) {

    empty_ = true;

    for (auto& host_set : hostSetsPerPriority()) {

      empty_ &= host_set->hosts().empty();

    }

  }

  if (thread_aware_lb_ != nullptr && thread_aware_lb_->factory()->recreateOnHostChange()) {

    lb_ = thread_aware_lb_->factory()->create({*this, original_local_priority_set_});

  }

}

void SubsetLoadBalancer::PriorityLbSubset::finalize(uint32_t priority) {

  while (host_sets_.size() <= priority) {

    host_sets_.push_back({HostHashSet(), HostHashSet()});

  }

  auto& [old_hosts, new_hosts] = host_sets_[priority];

  HostVector added;

  HostVector removed;

  for (const auto& host : old_hosts) {

    if (new_hosts.count(host) == 0) {

      removed.emplace_back(host);

    }

  }

  for (const auto& host : new_hosts) {

    if (old_hosts.count(host) == 0) {

      added.emplace_back(host);

    }

  }

  subset_.update(priority, new_hosts, added, removed);

  old_hosts.swap(new_hosts);

  new_hosts.clear();

}

    : wrapped_(wrapped) {

  ASSERT(wrapped->metadataMatchCriteria());

  metadata_match_ =

      wrapped->metadataMatchCriteria()->filterMatchCriteria(filtered_metadata_match_criteria_names);

}

    : wrapped_(wrapped) {

  ASSERT(wrapped->metadataMatchCriteria());

  metadata_match_ =

      wrapped->metadataMatchCriteria()->mergeMatchCriteria(metadata_match_criteria_override);

}