|
|
|
|
@ -157,9 +157,6 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
|
|
|
|
|
void UpdateLocked(UpdateArgs args) override; |
|
|
|
|
void ResetBackoffLocked() override; |
|
|
|
|
void FillChildRefsForChannelz( |
|
|
|
|
channelz::ChildRefsList* child_subchannels, |
|
|
|
|
channelz::ChildRefsList* child_channels) override; |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
struct LocalityServerlistEntry; |
|
|
|
|
@ -343,6 +340,7 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
void UpdateState(grpc_connectivity_state state, |
|
|
|
|
UniquePtr<SubchannelPicker> picker) override; |
|
|
|
|
void RequestReresolution() override; |
|
|
|
|
void AddTraceEvent(TraceSeverity severity, const char* message) override; |
|
|
|
|
|
|
|
|
|
void set_child(LoadBalancingPolicy* child) { child_ = child; } |
|
|
|
|
|
|
|
|
|
@ -398,8 +396,6 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
const grpc_channel_args* args); |
|
|
|
|
void ShutdownLocked(); |
|
|
|
|
void ResetBackoffLocked(); |
|
|
|
|
void FillChildRefsForChannelz(channelz::ChildRefsList* child_subchannels, |
|
|
|
|
channelz::ChildRefsList* child_channels); |
|
|
|
|
void Orphan() override; |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
@ -415,6 +411,8 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
void UpdateState(grpc_connectivity_state state, |
|
|
|
|
UniquePtr<SubchannelPicker> picker) override; |
|
|
|
|
void RequestReresolution() override; |
|
|
|
|
void AddTraceEvent(TraceSeverity severity, |
|
|
|
|
const char* message) override; |
|
|
|
|
void set_child(LoadBalancingPolicy* child) { child_ = child; } |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
@ -432,9 +430,6 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
|
|
|
|
|
OrphanablePtr<LoadBalancingPolicy> child_policy_; |
|
|
|
|
OrphanablePtr<LoadBalancingPolicy> pending_child_policy_; |
|
|
|
|
// Lock held when modifying the value of child_policy_ or
|
|
|
|
|
// pending_child_policy_.
|
|
|
|
|
Mutex child_policy_mu_; |
|
|
|
|
RefCountedPtr<XdsLb> parent_; |
|
|
|
|
RefCountedPtr<PickerRef> picker_ref_; |
|
|
|
|
grpc_connectivity_state connectivity_state_; |
|
|
|
|
@ -446,17 +441,12 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
const grpc_channel_args* args, XdsLb* parent); |
|
|
|
|
void ShutdownLocked(); |
|
|
|
|
void ResetBackoffLocked(); |
|
|
|
|
void FillChildRefsForChannelz(channelz::ChildRefsList* child_subchannels, |
|
|
|
|
channelz::ChildRefsList* child_channels); |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
void PruneLocalities(const LocalityList& locality_list); |
|
|
|
|
Map<RefCountedPtr<LocalityName>, OrphanablePtr<LocalityEntry>, |
|
|
|
|
LocalityName::Less> |
|
|
|
|
map_; |
|
|
|
|
// Lock held while filling child refs for all localities
|
|
|
|
|
// inside the map
|
|
|
|
|
Mutex child_refs_mu_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
struct LocalityServerlistEntry { |
|
|
|
|
@ -511,10 +501,6 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
// The channel for communicating with the LB server.
|
|
|
|
|
OrphanablePtr<BalancerChannelState> lb_chand_; |
|
|
|
|
OrphanablePtr<BalancerChannelState> pending_lb_chand_; |
|
|
|
|
// Mutex to protect the channel to the LB server. This is used when
|
|
|
|
|
// processing a channelz request.
|
|
|
|
|
// TODO(juanlishen): Replace this with atomic.
|
|
|
|
|
Mutex lb_chand_mu_; |
|
|
|
|
|
|
|
|
|
// Timeout in milliseconds for the LB call. 0 means no deadline.
|
|
|
|
|
int lb_call_timeout_ms_ = 0; |
|
|
|
|
@ -538,9 +524,6 @@ class XdsLb : public LoadBalancingPolicy { |
|
|
|
|
|
|
|
|
|
// The policy to use for the fallback backends.
|
|
|
|
|
RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy_config_; |
|
|
|
|
// Lock held when modifying the value of fallback_policy_ or
|
|
|
|
|
// pending_fallback_policy_.
|
|
|
|
|
Mutex fallback_policy_mu_; |
|
|
|
|
// Non-null iff we are in fallback mode.
|
|
|
|
|
OrphanablePtr<LoadBalancingPolicy> fallback_policy_; |
|
|
|
|
OrphanablePtr<LoadBalancingPolicy> pending_fallback_policy_; |
|
|
|
|
@ -648,7 +631,6 @@ void XdsLb::FallbackHelper::UpdateState(grpc_connectivity_state state, |
|
|
|
|
grpc_pollset_set_del_pollset_set( |
|
|
|
|
parent_->fallback_policy_->interested_parties(), |
|
|
|
|
parent_->interested_parties()); |
|
|
|
|
MutexLock lock(&parent_->fallback_policy_mu_); |
|
|
|
|
parent_->fallback_policy_ = std::move(parent_->pending_fallback_policy_); |
|
|
|
|
} else if (!CalledByCurrentFallback()) { |
|
|
|
|
// This request is from an outdated fallback policy, so ignore it.
|
|
|
|
|
@ -673,6 +655,15 @@ void XdsLb::FallbackHelper::RequestReresolution() { |
|
|
|
|
parent_->channel_control_helper()->RequestReresolution(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::FallbackHelper::AddTraceEvent(TraceSeverity severity, |
|
|
|
|
const char* message) { |
|
|
|
|
if (parent_->shutting_down_ || |
|
|
|
|
(!CalledByPendingFallback() && !CalledByCurrentFallback())) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
parent_->channel_control_helper()->AddTraceEvent(severity, message); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
// serverlist parsing code
|
|
|
|
|
//
|
|
|
|
|
@ -1365,21 +1356,29 @@ grpc_channel_args* BuildBalancerChannelArgs(const grpc_channel_args* args) { |
|
|
|
|
// treated as a stand-alone channel and not inherit this argument from the
|
|
|
|
|
// args of the parent channel.
|
|
|
|
|
GRPC_SSL_TARGET_NAME_OVERRIDE_ARG, |
|
|
|
|
// Don't want to pass down channelz node from parent; the balancer
|
|
|
|
|
// channel will get its own.
|
|
|
|
|
GRPC_ARG_CHANNELZ_CHANNEL_NODE, |
|
|
|
|
}; |
|
|
|
|
// Channel args to add.
|
|
|
|
|
const grpc_arg args_to_add[] = { |
|
|
|
|
// A channel arg indicating the target is a xds load balancer.
|
|
|
|
|
grpc_channel_arg_integer_create( |
|
|
|
|
const_cast<char*>(GRPC_ARG_ADDRESS_IS_XDS_LOAD_BALANCER), 1), |
|
|
|
|
// A channel arg indicating this is an internal channels, aka it is
|
|
|
|
|
// owned by components in Core, not by the user application.
|
|
|
|
|
grpc_channel_arg_integer_create( |
|
|
|
|
const_cast<char*>(GRPC_ARG_CHANNELZ_CHANNEL_IS_INTERNAL_CHANNEL), 1), |
|
|
|
|
}; |
|
|
|
|
InlinedVector<grpc_arg, 2> args_to_add; |
|
|
|
|
// A channel arg indicating the target is a xds load balancer.
|
|
|
|
|
args_to_add.emplace_back(grpc_channel_arg_integer_create( |
|
|
|
|
const_cast<char*>(GRPC_ARG_ADDRESS_IS_XDS_LOAD_BALANCER), 1)); |
|
|
|
|
// The parent channel's channelz uuid.
|
|
|
|
|
channelz::ChannelNode* channelz_node = nullptr; |
|
|
|
|
const grpc_arg* arg = |
|
|
|
|
grpc_channel_args_find(args, GRPC_ARG_CHANNELZ_CHANNEL_NODE); |
|
|
|
|
if (arg != nullptr && arg->type == GRPC_ARG_POINTER && |
|
|
|
|
arg->value.pointer.p != nullptr) { |
|
|
|
|
channelz_node = static_cast<channelz::ChannelNode*>(arg->value.pointer.p); |
|
|
|
|
args_to_add.emplace_back( |
|
|
|
|
channelz::MakeParentUuidArg(channelz_node->uuid())); |
|
|
|
|
} |
|
|
|
|
// Construct channel args.
|
|
|
|
|
grpc_channel_args* new_args = grpc_channel_args_copy_and_add_and_remove( |
|
|
|
|
args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), args_to_add, |
|
|
|
|
GPR_ARRAY_SIZE(args_to_add)); |
|
|
|
|
args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), args_to_add.data(), |
|
|
|
|
args_to_add.size()); |
|
|
|
|
// Make any necessary modifications for security.
|
|
|
|
|
return grpc_lb_policy_xds_modify_lb_channel_args(new_args); |
|
|
|
|
} |
|
|
|
|
@ -1434,18 +1433,12 @@ void XdsLb::ShutdownLocked() { |
|
|
|
|
grpc_pollset_set_del_pollset_set( |
|
|
|
|
pending_fallback_policy_->interested_parties(), interested_parties()); |
|
|
|
|
} |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&fallback_policy_mu_); |
|
|
|
|
fallback_policy_.reset(); |
|
|
|
|
pending_fallback_policy_.reset(); |
|
|
|
|
} |
|
|
|
|
fallback_policy_.reset(); |
|
|
|
|
pending_fallback_policy_.reset(); |
|
|
|
|
// We reset the LB channels here instead of in our destructor because they
|
|
|
|
|
// hold refs to XdsLb.
|
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&lb_chand_mu_); |
|
|
|
|
lb_chand_.reset(); |
|
|
|
|
pending_lb_chand_.reset(); |
|
|
|
|
} |
|
|
|
|
lb_chand_.reset(); |
|
|
|
|
pending_lb_chand_.reset(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
@ -1468,40 +1461,6 @@ void XdsLb::ResetBackoffLocked() { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::FillChildRefsForChannelz(channelz::ChildRefsList* child_subchannels, |
|
|
|
|
channelz::ChildRefsList* child_channels) { |
|
|
|
|
// Delegate to the locality_map_ to fill the children subchannels.
|
|
|
|
|
locality_map_.FillChildRefsForChannelz(child_subchannels, child_channels); |
|
|
|
|
{ |
|
|
|
|
// This must be done holding fallback_policy_mu_, since this method does not
|
|
|
|
|
// run in the combiner.
|
|
|
|
|
MutexLock lock(&fallback_policy_mu_); |
|
|
|
|
if (fallback_policy_ != nullptr) { |
|
|
|
|
fallback_policy_->FillChildRefsForChannelz(child_subchannels, |
|
|
|
|
child_channels); |
|
|
|
|
} |
|
|
|
|
if (pending_fallback_policy_ != nullptr) { |
|
|
|
|
pending_fallback_policy_->FillChildRefsForChannelz(child_subchannels, |
|
|
|
|
child_channels); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
MutexLock lock(&lb_chand_mu_); |
|
|
|
|
if (lb_chand_ != nullptr) { |
|
|
|
|
grpc_core::channelz::ChannelNode* channel_node = |
|
|
|
|
grpc_channel_get_channelz_node(lb_chand_->channel()); |
|
|
|
|
if (channel_node != nullptr) { |
|
|
|
|
child_channels->push_back(channel_node->uuid()); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if (pending_lb_chand_ != nullptr) { |
|
|
|
|
grpc_core::channelz::ChannelNode* channel_node = |
|
|
|
|
grpc_channel_get_channelz_node(pending_lb_chand_->channel()); |
|
|
|
|
if (channel_node != nullptr) { |
|
|
|
|
child_channels->push_back(channel_node->uuid()); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::ProcessAddressesAndChannelArgsLocked( |
|
|
|
|
const ServerAddressList& addresses, const grpc_channel_args& args) { |
|
|
|
|
// Update fallback address list.
|
|
|
|
|
@ -1691,14 +1650,10 @@ void XdsLb::UpdateFallbackPolicyLocked() { |
|
|
|
|
fallback_policy_ == nullptr ? "" : "pending ", |
|
|
|
|
fallback_policy_name); |
|
|
|
|
} |
|
|
|
|
auto new_policy = |
|
|
|
|
CreateFallbackPolicyLocked(fallback_policy_name, update_args.args); |
|
|
|
|
auto& lb_policy = fallback_policy_ == nullptr ? fallback_policy_ |
|
|
|
|
: pending_fallback_policy_; |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&fallback_policy_mu_); |
|
|
|
|
lb_policy = std::move(new_policy); |
|
|
|
|
} |
|
|
|
|
lb_policy = |
|
|
|
|
CreateFallbackPolicyLocked(fallback_policy_name, update_args.args); |
|
|
|
|
policy_to_update = lb_policy.get(); |
|
|
|
|
} else { |
|
|
|
|
// Cases 2a and 3a: update an existing policy.
|
|
|
|
|
@ -1769,7 +1724,6 @@ void XdsLb::LocalityMap::PruneLocalities(const LocalityList& locality_list) { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if (!found) { // Remove entries not present in the locality list
|
|
|
|
|
MutexLock lock(&child_refs_mu_); |
|
|
|
|
iter = map_.erase(iter); |
|
|
|
|
} else |
|
|
|
|
iter++; |
|
|
|
|
@ -1786,7 +1740,6 @@ void XdsLb::LocalityMap::UpdateLocked( |
|
|
|
|
if (iter == map_.end()) { |
|
|
|
|
OrphanablePtr<LocalityEntry> new_entry = MakeOrphanable<LocalityEntry>( |
|
|
|
|
parent->Ref(), locality_serverlist[i]->locality_weight); |
|
|
|
|
MutexLock lock(&child_refs_mu_); |
|
|
|
|
iter = map_.emplace(locality_serverlist[i]->locality_name, |
|
|
|
|
std::move(new_entry)) |
|
|
|
|
.first; |
|
|
|
|
@ -1799,10 +1752,7 @@ void XdsLb::LocalityMap::UpdateLocked( |
|
|
|
|
PruneLocalities(locality_serverlist); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::LocalityMap::ShutdownLocked() { |
|
|
|
|
MutexLock lock(&child_refs_mu_); |
|
|
|
|
map_.clear(); |
|
|
|
|
} |
|
|
|
|
void XdsLb::LocalityMap::ShutdownLocked() { map_.clear(); } |
|
|
|
|
|
|
|
|
|
void XdsLb::LocalityMap::ResetBackoffLocked() { |
|
|
|
|
for (auto& p : map_) { |
|
|
|
|
@ -1810,15 +1760,6 @@ void XdsLb::LocalityMap::ResetBackoffLocked() { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::LocalityMap::FillChildRefsForChannelz( |
|
|
|
|
channelz::ChildRefsList* child_subchannels, |
|
|
|
|
channelz::ChildRefsList* child_channels) { |
|
|
|
|
MutexLock lock(&child_refs_mu_); |
|
|
|
|
for (auto& p : map_) { |
|
|
|
|
p.second->FillChildRefsForChannelz(child_subchannels, child_channels); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
// XdsLb::LocalityMap::LocalityEntry
|
|
|
|
|
//
|
|
|
|
|
@ -1954,14 +1895,9 @@ void XdsLb::LocalityMap::LocalityEntry::UpdateLocked( |
|
|
|
|
gpr_log(GPR_INFO, "[xdslb %p] Creating new %schild policy %s", this, |
|
|
|
|
child_policy_ == nullptr ? "" : "pending ", child_policy_name); |
|
|
|
|
} |
|
|
|
|
auto new_policy = |
|
|
|
|
CreateChildPolicyLocked(child_policy_name, update_args.args); |
|
|
|
|
auto& lb_policy = |
|
|
|
|
child_policy_ == nullptr ? child_policy_ : pending_child_policy_; |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&child_policy_mu_); |
|
|
|
|
lb_policy = std::move(new_policy); |
|
|
|
|
} |
|
|
|
|
lb_policy = CreateChildPolicyLocked(child_policy_name, update_args.args); |
|
|
|
|
policy_to_update = lb_policy.get(); |
|
|
|
|
} else { |
|
|
|
|
// Cases 2a and 3a: update an existing policy.
|
|
|
|
|
@ -1991,11 +1927,8 @@ void XdsLb::LocalityMap::LocalityEntry::ShutdownLocked() { |
|
|
|
|
pending_child_policy_->interested_parties(), |
|
|
|
|
parent_->interested_parties()); |
|
|
|
|
} |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&child_policy_mu_); |
|
|
|
|
child_policy_.reset(); |
|
|
|
|
pending_child_policy_.reset(); |
|
|
|
|
} |
|
|
|
|
child_policy_.reset(); |
|
|
|
|
pending_child_policy_.reset(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::LocalityMap::LocalityEntry::ResetBackoffLocked() { |
|
|
|
|
@ -2005,17 +1938,6 @@ void XdsLb::LocalityMap::LocalityEntry::ResetBackoffLocked() { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::LocalityMap::LocalityEntry::FillChildRefsForChannelz( |
|
|
|
|
channelz::ChildRefsList* child_subchannels, |
|
|
|
|
channelz::ChildRefsList* child_channels) { |
|
|
|
|
MutexLock lock(&child_policy_mu_); |
|
|
|
|
child_policy_->FillChildRefsForChannelz(child_subchannels, child_channels); |
|
|
|
|
if (pending_child_policy_ != nullptr) { |
|
|
|
|
pending_child_policy_->FillChildRefsForChannelz(child_subchannels, |
|
|
|
|
child_channels); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::LocalityMap::LocalityEntry::Orphan() { |
|
|
|
|
ShutdownLocked(); |
|
|
|
|
Unref(); |
|
|
|
|
@ -2070,7 +1992,6 @@ void XdsLb::LocalityMap::LocalityEntry::Helper::UpdateState( |
|
|
|
|
grpc_pollset_set_del_pollset_set( |
|
|
|
|
entry_->child_policy_->interested_parties(), |
|
|
|
|
entry_->parent_->interested_parties()); |
|
|
|
|
MutexLock lock(&entry_->child_policy_mu_); |
|
|
|
|
entry_->child_policy_ = std::move(entry_->pending_child_policy_); |
|
|
|
|
} else if (!CalledByCurrentChild()) { |
|
|
|
|
// This request is from an outdated child, so ignore it.
|
|
|
|
|
@ -2180,6 +2101,15 @@ void XdsLb::LocalityMap::LocalityEntry::Helper::RequestReresolution() { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsLb::LocalityMap::LocalityEntry::Helper::AddTraceEvent( |
|
|
|
|
TraceSeverity severity, const char* message) { |
|
|
|
|
if (entry_->parent_->shutting_down_ || |
|
|
|
|
(!CalledByPendingChild() && !CalledByCurrentChild())) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
entry_->parent_->channel_control_helper()->AddTraceEvent(severity, message); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
// factory
|
|
|
|
|
//
|
|
|
|
|
|
Mark D. Roth
6 years ago