@ -25,9 +25,9 @@
# include <grpc/grpc.h>
# include "src/core/ext/filters/client_channel/lb_policy.h"
# include "src/core/ext/filters/client_channel/lb_policy/child_policy_handler.h"
# include "src/core/ext/filters/client_channel/lb_policy_factory.h"
# include "src/core/ext/filters/client_channel/lb_policy_registry.h"
# include "src/core/ext/filters/client_channel/lb_policy/child_policy_handler.h"
# include "src/core/lib/channel/channel_args.h"
# include "src/core/lib/gpr/string.h"
# include "src/core/lib/gprpp/orphanable.h"
@ -39,39 +39,38 @@
namespace grpc_core {
TraceFlag grpc_lb_weighted_target _trace ( false , " weighted_target _lb" ) ;
TraceFlag grpc_rds_lb _trace ( false , " rds _lb" ) ;
namespace {
constexpr char kWeightedTarget [ ] = " weighted_target _experimental" ;
constexpr char kRds [ ] = " xds_routing _experimental" ;
// Config for weighted_target LB policy.
class WeightedTarget LbConfig : public LoadBalancingPolicy : : Config {
// Config for rds LB policy.
class Rds LbConfig : public LoadBalancingPolicy : : Config {
public :
struct ChildConfig {
uint32_t weight ;
RefCountedPtr < LoadBalancingPolicy : : Config > config ;
} ;
using Target Map = std : : map < std : : string , ChildConfig > ;
using Action Map = std : : map < std : : string , ChildConfig > ;
explicit WeightedTargetLbConfig ( TargetMap target _map)
: target _map_( std : : move ( target _map) ) { }
explicit RdsLbConfig ( ActionMap action _map)
: action _map_( std : : move ( action _map) ) { }
const char * name ( ) const override { return kWeightedTarget ; }
const char * name ( ) const override { return kRds ; }
const TargetMap & target _map( ) const { return target _map_; }
const ActionMap & action _map( ) const { return action _map_; }
private :
TargetMap target _map_;
ActionMap action _map_;
} ;
// weighted_target LB policy.
class WeightedTarget Lb : public LoadBalancingPolicy {
// rds LB policy.
class Rds Lb : public LoadBalancingPolicy {
public :
explicit WeightedTarget Lb( Args args ) ;
explicit Rds Lb( Args args ) ;
const char * name ( ) const override { return kWeightedTarget ; }
const char * name ( ) const override { return kRds ; }
void UpdateLocked ( UpdateArgs args ) override ;
void ExitIdleLocked ( ) override ;
@ -83,53 +82,48 @@ class WeightedTargetLb : public LoadBalancingPolicy {
public :
explicit ChildPickerWrapper ( std : : unique_ptr < SubchannelPicker > picker )
: picker_ ( std : : move ( picker ) ) { }
PickResult Pick ( PickArgs args ) {
return picker_ - > Pick ( std : : move ( args ) ) ;
}
PickResult Pick ( PickArgs args ) { return picker_ - > Pick ( std : : move ( args ) ) ; }
private :
std : : unique_ptr < SubchannelPicker > picker_ ;
} ;
// Picks a child using stateless WRR and then delegates to that
// child's picker.
class Weighted Picker : public SubchannelPicker {
class Rds Picker : public SubchannelPicker {
public :
// Maintains a weighted list of pickers from each child that is in
// Maintains a rds list of pickers from each child that is in
// ready state. The first element in the pair represents the end of a
// range proportional to the child's weight. The start of the range
// is the previous value in the vector and is 0 for the first element.
using PickerList =
InlinedVector < std : : pair < uint32_t ,
RefCountedPtr < ChildPickerWrapper > > , 1 > ;
using PickerList = InlinedVector < RefCountedPtr < ChildPickerWrapper > , 1 > ;
Weighted Picker( RefCountedPtr < WeightedTarget Lb> parent , PickerList pickers )
RdsPicker ( RefCountedPtr < RdsLb > parent , PickerList pickers )
: parent_ ( std : : move ( parent ) ) , pickers_ ( std : : move ( pickers ) ) { }
~ Weighted Picker( ) { parent_ . reset ( DEBUG_LOCATION , " Weighted Picker" ) ; }
~ Rds Picker( ) { parent_ . reset ( DEBUG_LOCATION , " Rds Picker" ) ; }
PickResult Pick ( PickArgs args ) override ;
private :
RefCountedPtr < WeightedTarget Lb> parent_ ;
RefCountedPtr < Rds Lb> parent_ ;
PickerList pickers_ ;
} ;
// Each WeightedChild holds a ref to its parent WeightedTarget Lb.
class Weighted Child : public InternallyRefCounted < Weighted Child> {
// Each RdsChild holds a ref to its parent Rds Lb.
class Rds Child : public InternallyRefCounted < Rds Child> {
public :
WeightedChild ( RefCountedPtr < WeightedTargetLb > weighted_target_policy ,
const std : : string & name ) ;
~ WeightedChild ( ) ;
RdsChild ( RefCountedPtr < RdsLb > rds_policy , const std : : string & name ) ;
~ RdsChild ( ) ;
void Orphan ( ) override ;
void UpdateLocked ( const WeightedTarget LbConfig: : ChildConfig & config ,
void UpdateLocked ( const Rds LbConfig: : ChildConfig & config ,
const ServerAddressList & addresses ,
const grpc_channel_args * args ) ;
void ExitIdleLocked ( ) ;
void ResetBackoffLocked ( ) ;
void DeactivateLocked ( ) ;
uint32_t weight ( ) const { return weight_ ; }
grpc_connectivity_state connectivity_state ( ) const {
return connectivity_state_ ;
}
@ -140,10 +134,10 @@ class WeightedTargetLb : public LoadBalancingPolicy {
private :
class Helper : public ChannelControlHelper {
public :
explicit Helper ( RefCountedPtr < WeightedChild > weighted _child)
: weighted _child_( std : : move ( weighted _child) ) { }
explicit Helper ( RefCountedPtr < RdsChild > rds _child)
: rds _child_( std : : move ( rds _child) ) { }
~ Helper ( ) { weighted _child_. reset ( DEBUG_LOCATION , " Helper " ) ; }
~ Helper ( ) { rds _child_. reset ( DEBUG_LOCATION , " Helper " ) ; }
RefCountedPtr < SubchannelInterface > CreateSubchannel (
const grpc_channel_args & args ) override ;
@ -153,7 +147,7 @@ class WeightedTargetLb : public LoadBalancingPolicy {
void AddTraceEvent ( TraceSeverity severity , StringView message ) override ;
private :
RefCountedPtr < WeightedChild > weighted _child_;
RefCountedPtr < RdsChild > rds _child_;
} ;
// Methods for dealing with the child policy.
@ -164,13 +158,11 @@ class WeightedTargetLb : public LoadBalancingPolicy {
static void OnDelayedRemovalTimerLocked ( void * arg , grpc_error * error ) ;
// The owning LB policy.
RefCountedPtr < WeightedTargetLb > weighted_target _policy_;
RefCountedPtr < RdsLb > rds _policy_;
// Points to the corresponding key in WeightedTargetLb::target s_.
// Points to the corresponding key in RdsLb::action s_.
const std : : string & name_ ;
uint32_t weight_ ;
OrphanablePtr < LoadBalancingPolicy > child_policy_ ;
RefCountedPtr < ChildPickerWrapper > picker_wrapper_ ;
@ -184,7 +176,7 @@ class WeightedTargetLb : public LoadBalancingPolicy {
bool shutdown_ = false ;
} ;
~ WeightedTarget Lb( ) ;
~ Rds Lb( ) ;
void ShutdownLocked ( ) override ;
@ -193,138 +185,116 @@ class WeightedTargetLb : public LoadBalancingPolicy {
const grpc_millis child_retention_interval_ms_ ;
// Current config from the resolver.
RefCountedPtr < WeightedTarget LbConfig> config_ ;
RefCountedPtr < Rds LbConfig> config_ ;
// Internal state.
bool shutting_down_ = false ;
// Children.
std : : map < std : : string , OrphanablePtr < WeightedChild > > target s_;
std : : map < std : : string , OrphanablePtr < RdsChild > > action s_;
} ;
//
// WeightedTargetLb::Weighted Picker
// RdsLb::Rds Picker
//
WeightedTargetLb : : PickResult WeightedTargetLb : : WeightedPicker : : Pick (
PickArgs args ) {
// Generate a random number in [0, total weight).
const uint32_t key = rand ( ) % pickers_ [ pickers_ . size ( ) - 1 ] . first ;
// Find the index in pickers_ corresponding to key.
size_t mid = 0 ;
size_t start_index = 0 ;
size_t end_index = pickers_ . size ( ) - 1 ;
size_t index = 0 ;
while ( end_index > start_index ) {
mid = ( start_index + end_index ) / 2 ;
if ( pickers_ [ mid ] . first > key ) {
end_index = mid ;
} else if ( pickers_ [ mid ] . first < key ) {
start_index = mid + 1 ;
} else {
index = mid + 1 ;
break ;
}
}
if ( index = = 0 ) index = start_index ;
GPR_ASSERT ( pickers_ [ index ] . first > key ) ;
// Delegate to the child picker.
return pickers_ [ index ] . second - > Pick ( args ) ;
RdsLb : : PickResult RdsLb : : RdsPicker : : Pick ( PickArgs args ) {
gpr_log (
GPR_INFO ,
" donna Picking not implemented yet, just always use the one and only " ) ;
return pickers_ [ 0 ] - > Pick ( args ) ;
}
//
// WeightedTarget Lb
// RdsLb
//
WeightedTargetLb : : WeightedTarget Lb( Args args )
RdsLb : : RdsLb ( Args args )
: LoadBalancingPolicy ( std : : move ( args ) ) ,
// FIXME: new channel arg
// FIXME: new channel arg
child_retention_interval_ms_ ( grpc_channel_args_find_integer (
args . args , GRPC_ARG_LOCALITY_RETENTION_INTERVAL_MS ,
{ GRPC_WEIGHTED_TARGET_CHILD_RETENTION_INTERVAL_MS , 0 , INT_MAX } ) ) { }
WeightedTargetLb : : ~ WeightedTargetLb ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target_trace ) ) {
gpr_log ( GPR_INFO , " [weighted_target_lb %p] destroying weighted_target LB policy " ,
this ) ;
RdsLb : : ~ RdsLb ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb_trace ) ) {
gpr_log ( GPR_INFO , " [rds_lb %p] destroying rds LB policy " , this ) ;
}
}
void WeightedTarget Lb: : ShutdownLocked ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target _trace ) ) {
gpr_log ( GPR_INFO , " [weighted_target _lb %p] shutting down " , this ) ;
void Rds Lb: : ShutdownLocked ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb _trace ) ) {
gpr_log ( GPR_INFO , " [rds _lb %p] shutting down " , this ) ;
}
shutting_down_ = true ;
target s_. clear ( ) ;
action s_. clear ( ) ;
}
void WeightedTarget Lb: : ExitIdleLocked ( ) {
for ( auto & p : target s_) p . second - > ExitIdleLocked ( ) ;
void Rds Lb: : ExitIdleLocked ( ) {
for ( auto & p : action s_) p . second - > ExitIdleLocked ( ) ;
}
void WeightedTarget Lb: : ResetBackoffLocked ( ) {
for ( auto & p : target s_) p . second - > ResetBackoffLocked ( ) ;
void Rds Lb: : ResetBackoffLocked ( ) {
for ( auto & p : action s_) p . second - > ResetBackoffLocked ( ) ;
}
void WeightedTarget Lb: : UpdateLocked ( UpdateArgs args ) {
void Rds Lb: : UpdateLocked ( UpdateArgs args ) {
if ( shutting_down_ ) return ;
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target _trace ) ) {
gpr_log ( GPR_INFO , " [weighted_target _lb %p] Received update " , this ) ;
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb _trace ) ) {
gpr_log ( GPR_INFO , " [rds _lb %p] Received update " , this ) ;
}
// Update config.
config_ = std : : move ( args . config ) ;
// Deactivate the target s not in the new config.
for ( auto it = target s_. begin ( ) ; it ! = target s_. end ( ) ; ) {
// Deactivate the action s not in the new config.
for ( auto it = action s_. begin ( ) ; it ! = action s_. end ( ) ; ) {
const std : : string & name = it - > first ;
Weighted Child* child = it - > second . get ( ) ;
if ( config_ - > target _map( ) . find ( name ) ! = config_ - > target _map( ) . end ( ) ) {
Rds Child* child = it - > second . get ( ) ;
if ( config_ - > action _map( ) . find ( name ) ! = config_ - > action _map( ) . end ( ) ) {
+ + it ;
continue ;
}
if ( child_retention_interval_ms_ = = 0 ) {
it = target s_. erase ( it ) ;
it = action s_. erase ( it ) ;
} else {
child - > DeactivateLocked ( ) ;
+ + it ;
}
}
// Add or update the target s in the new config.
for ( const auto & p : config_ - > target _map( ) ) {
// Add or update the action s in the new config.
for ( const auto & p : config_ - > action _map( ) ) {
const std : : string & name = p . first ;
const WeightedTarget LbConfig: : ChildConfig & config = p . second ;
auto it = target s_. find ( name ) ;
if ( it = = target s_. end ( ) ) {
it = target s_. emplace ( std : : make_pair ( name , nullptr ) ) . first ;
it - > second = MakeOrphanable < WeightedChild > (
Ref ( DEBUG_LOCATION , " Weighted Child" ) , it - > first ) ;
const Rds LbConfig: : ChildConfig & config = p . second ;
auto it = action s_. find ( name ) ;
if ( it = = action s_. end ( ) ) {
it = action s_. emplace ( std : : make_pair ( name , nullptr ) ) . first ;
it - > second =
MakeOrphanable < RdsChild > ( Ref ( DEBUG_LOCATION , " Rds Child" ) , it - > first ) ;
}
it - > second - > UpdateLocked ( config , args . addresses , args . args ) ;
}
}
void WeightedTarget Lb: : UpdateStateLocked ( ) {
void Rds Lb: : UpdateStateLocked ( ) {
// Construct a new picker which maintains a map of all child pickers
// that are ready. Each child is represented by a portion of the range
// proportional to its weight, such that the total range is the sum of the
// weights of all children.
WeightedPicker : : PickerList picker_list ;
uint32_t end = 0 ;
RdsPicker : : PickerList picker_list ;
// Also count the number of children in each state, to determine the
// overall state.
size_t num_connecting = 0 ;
size_t num_idle = 0 ;
size_t num_transient_failures = 0 ;
for ( const auto & p : target s_) {
for ( const auto & p : action s_) {
const auto & child_name = p . first ;
const Weighted Child* child = p . second . get ( ) ;
// Skip the target s that are not in the latest update.
if ( config_ - > target _map( ) . find ( child_name ) = = config_ - > target _map( ) . end ( ) ) {
const Rds Child* child = p . second . get ( ) ;
// Skip the action s that are not in the latest update.
if ( config_ - > action _map( ) . find ( child_name ) = = config_ - > action _map( ) . end ( ) ) {
continue ;
}
switch ( child - > connectivity_state ( ) ) {
case GRPC_CHANNEL_READY : {
end + = child - > weight ( ) ;
picker_list . push_back ( std : : make_pair ( end , child - > picker_wrapper ( ) ) ) ;
picker_list . push_back ( child - > picker_wrapper ( ) ) ;
break ;
}
case GRPC_CHANNEL_CONNECTING : {
@ -354,62 +324,59 @@ void WeightedTargetLb::UpdateStateLocked() {
} else {
connectivity_state = GRPC_CHANNEL_TRANSIENT_FAILURE ;
}
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target _trace ) ) {
gpr_log ( GPR_INFO , " [weighted_target _lb %p] connectivity changed to %s " ,
this , ConnectivityStateName ( connectivity_state ) ) ;
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb _trace ) ) {
gpr_log ( GPR_INFO , " [rds _lb %p] connectivity changed to %s " , this ,
ConnectivityStateName ( connectivity_state ) ) ;
}
std : : unique_ptr < SubchannelPicker > picker ;
switch ( connectivity_state ) {
case GRPC_CHANNEL_READY :
picker = absl : : make_unique < WeightedPicker > (
Ref ( DEBUG_LOCATION , " WeightedPicker " ) , std : : move ( picker_list ) ) ;
picker = absl : : make_unique < RdsPicker > ( Ref ( DEBUG_LOCATION , " RdsPicker " ) ,
std : : move ( picker_list ) ) ;
break ;
case GRPC_CHANNEL_CONNECTING :
case GRPC_CHANNEL_IDLE :
picker = absl : : make_unique < QueuePicker > (
Ref ( DEBUG_LOCATION , " QueuePicker " ) ) ;
picker =
absl : : make_unique < QueuePicker > ( Ref ( DEBUG_LOCATION , " QueuePicker " ) ) ;
break ;
default :
picker = absl : : make_unique < TransientFailurePicker > (
GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" weighted_target : all children report state TRANSIENT_FAILURE" ) ) ;
" rds : all children report state TRANSIENT_FAILURE" ) ) ;
}
channel_control_helper ( ) - > UpdateState ( connectivity_state , std : : move ( picker ) ) ;
}
//
// WeightedTargetLb::Weighted Child
// RdsLb::Rds Child
//
WeightedTargetLb : : WeightedChild : : WeightedChild (
RefCountedPtr < WeightedTargetLb > weighted_target_policy ,
RdsLb : : RdsChild : : RdsChild ( RefCountedPtr < RdsLb > rds_policy ,
const std : : string & name )
: weighted_target _policy_( std : : move ( weighted_target _policy) ) , name_ ( name ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target _trace ) ) {
gpr_log ( GPR_INFO , " [weighted_target_lb %p] created Weighted Child %p for %s " ,
weighted_target _policy_. get ( ) , this , name_ . c_str ( ) ) ;
: rds _policy_( std : : move ( rds _policy) ) , name_ ( name ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb _trace ) ) {
gpr_log ( GPR_INFO , " [rds_lb %p] created Rds Child %p for %s " ,
rds _policy_. get ( ) , this , name_ . c_str ( ) ) ;
}
}
WeightedTargetLb : : WeightedChild : : ~ WeightedChild ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target_trace ) ) {
gpr_log ( GPR_INFO ,
" [weighted_target_lb %p] WeightedChild %p %s: destroying child " ,
weighted_target_policy_ . get ( ) , this , name_ . c_str ( ) ) ;
RdsLb : : RdsChild : : ~ RdsChild ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb_trace ) ) {
gpr_log ( GPR_INFO , " [rds_lb %p] RdsChild %p %s: destroying child " ,
rds_policy_ . get ( ) , this , name_ . c_str ( ) ) ;
}
weighted_target _policy_. reset ( DEBUG_LOCATION , " Weighted Child" ) ;
rds _policy_. reset ( DEBUG_LOCATION , " Rds Child" ) ;
}
void WeightedTargetLb : : WeightedChild : : Orphan ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target_trace ) ) {
gpr_log ( GPR_INFO ,
" [weighted_target_lb %p] WeightedChild %p %s: shutting down child " ,
weighted_target_policy_ . get ( ) , this , name_ . c_str ( ) ) ;
void RdsLb : : RdsChild : : Orphan ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb_trace ) ) {
gpr_log ( GPR_INFO , " [rds_lb %p] RdsChild %p %s: shutting down child " ,
rds_policy_ . get ( ) , this , name_ . c_str ( ) ) ;
}
// Remove the child policy's interested_parties pollset_set from the
// xDS policy.
grpc_pollset_set_del_pollset_set ( child_policy_ - > interested_parties ( ) ,
weighted_target _policy_- > interested_parties ( ) ) ;
rds _policy_- > interested_parties ( ) ) ;
child_policy_ . reset ( ) ;
// Drop our ref to the child's picker, in case it's holding a ref to
// the child.
@ -421,39 +388,35 @@ void WeightedTargetLb::WeightedChild::Orphan() {
Unref ( ) ;
}
OrphanablePtr < LoadBalancingPolicy >
WeightedTargetLb : : WeightedChild : : CreateChildPolicyLocked (
OrphanablePtr < LoadBalancingPolicy > RdsLb : : RdsChild : : CreateChildPolicyLocked (
const grpc_channel_args * args ) {
LoadBalancingPolicy : : Args lb_policy_args ;
lb_policy_args . combiner = weighted_target _policy_- > combiner ( ) ;
lb_policy_args . combiner = rds _policy_- > combiner ( ) ;
lb_policy_args . args = args ;
lb_policy_args . channel_control_helper =
absl : : make_unique < Helper > ( this - > Ref ( DEBUG_LOCATION , " Helper " ) ) ;
OrphanablePtr < LoadBalancingPolicy > lb_policy =
MakeOrphanable < ChildPolicyHandler > ( std : : move ( lb_policy_args ) ,
& grpc_lb_weighted_target _trace ) ;
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target _trace ) ) {
& grpc_rds_lb _trace ) ;
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb _trace ) ) {
gpr_log ( GPR_INFO ,
" [weighted_target_lb %p] Weighted Child %p %s: Created new child "
" [rds_lb %p] Rds Child %p %s: Created new child "
" policy handler %p " ,
weighted_target_policy_ . get ( ) , this , name_ . c_str ( ) ,
lb_policy . get ( ) ) ;
rds_policy_ . get ( ) , this , name_ . c_str ( ) , lb_policy . get ( ) ) ;
}
// Add the xDS's interested_parties pollset_set to that of the newly created
// child policy. This will make the child policy progress upon activity on
// xDS LB, which in turn is tied to the application's call.
grpc_pollset_set_add_pollset_set (
lb_policy - > interested_parties ( ) ,
weighted_target_policy_ - > interested_parties ( ) ) ;
grpc_pollset_set_add_pollset_set ( lb_policy - > interested_parties ( ) ,
rds_policy_ - > interested_parties ( ) ) ;
return lb_policy ;
}
void WeightedTargetLb : : Weighted Child: : UpdateLocked (
const WeightedTargetLbConfig : : ChildConfig & config ,
const ServerAddressList & addresses , const grpc_channel_args * args ) {
if ( weighted_target _policy_- > shutting_down_ ) return ;
void RdsLb : : Rds Child: : UpdateLocked ( const RdsLbConfig : : ChildConfig & config ,
const ServerAddressList & addresses ,
const grpc_channel_args * args ) {
if ( rds _policy_- > shutting_down_ ) return ;
// Update child weight.
weight_ = config . weight ;
// Reactivate if needed.
if ( delayed_removal_timer_callback_pending_ ) {
grpc_timer_cancel ( & delayed_removal_timer_ ) ;
@ -468,205 +431,187 @@ void WeightedTargetLb::WeightedChild::UpdateLocked(
update_args . addresses = addresses ;
update_args . args = grpc_channel_args_copy ( args ) ;
// Update the policy.
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_weighted_target _trace ) ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_rds_lb _trace ) ) {
gpr_log ( GPR_INFO ,
" [weighted_target_lb %p] Weighted Child %p %s: Updating child "
" [rds_lb %p] Rds Child %p %s: Updating child "
" policy handler %p " ,
weighted_target_policy_ . get ( ) , this , name_ . c_str ( ) ,
child_policy_ . get ( ) ) ;
rds_policy_ . get ( ) , this , name_ . c_str ( ) , child_policy_ . get ( ) ) ;
}
child_policy_ - > UpdateLocked ( std : : move ( update_args ) ) ;
}
void WeightedTargetLb : : WeightedChild : : ExitIdleLocked ( ) {
child_policy_ - > ExitIdleLocked ( ) ;
}
void RdsLb : : RdsChild : : ExitIdleLocked ( ) { child_policy_ - > ExitIdleLocked ( ) ; }
void WeightedTargetLb : : Weighted Child: : ResetBackoffLocked ( ) {
void RdsLb : : RdsChild : : ResetBackoffLocked ( ) {
child_policy_ - > ResetBackoffLocked ( ) ;
}
void WeightedTargetLb : : Weighted Child: : DeactivateLocked ( ) {
void RdsLb : : Rds Child: : DeactivateLocked ( ) {
// If already deactivated, don't do that again.
if ( weight_ = = 0 ) return ;
// Set the child weight to 0 so that future picker won't contain this child.
weight_ = 0 ;
// Start a timer to delete the child.
Ref ( DEBUG_LOCATION , " Weighted Child+timer" ) . release ( ) ;
Ref ( DEBUG_LOCATION , " Rds Child+timer" ) . release ( ) ;
GRPC_CLOSURE_INIT ( & on_delayed_removal_timer_ , OnDelayedRemovalTimer , this ,
grpc_schedule_on_exec_ctx ) ;
grpc_timer_init (
& delayed_removal_timer_ ,
ExecCtx : : Get ( ) - > Now ( ) +
weighted_target_policy_ - > child_retention_interval_ms_ ,
ExecCtx : : Get ( ) - > Now ( ) + rds_policy_ - > child_retention_interval_ms_ ,
& on_delayed_removal_timer_ ) ;
delayed_removal_timer_callback_pending_ = true ;
}
void WeightedTargetLb : : WeightedChild : : OnDelayedRemovalTimer ( void * arg ,
grpc_error * error ) {
WeightedChild * self = static_cast < WeightedChild * > ( arg ) ;
self - > weighted_target_policy_ - > combiner ( ) - > Run (
void RdsLb : : RdsChild : : OnDelayedRemovalTimer ( void * arg , grpc_error * error ) {
RdsChild * self = static_cast < RdsChild * > ( arg ) ;
self - > rds_policy_ - > combiner ( ) - > Run (
GRPC_CLOSURE_INIT ( & self - > on_delayed_removal_timer_ ,
OnDelayedRemovalTimerLocked , self , nullptr ) ,
GRPC_ERROR_REF ( error ) ) ;
}
void WeightedTargetLb : : Weighted Child: : OnDelayedRemovalTimerLocked (
void * arg , grpc_error * error ) {
Weighted Child* self = static_cast < Weighted Child* > ( arg ) ;
void RdsLb : : Rds Child: : OnDelayedRemovalTimerLocked ( void * arg ,
grpc_error * error ) {
Rds Child* self = static_cast < Rds Child* > ( arg ) ;
self - > delayed_removal_timer_callback_pending_ = false ;
if ( error = = GRPC_ERROR_NONE & & ! self - > shutdown_ & & self - > weight_ = = 0 ) {
self - > weighted_target_policy_ - > target s_. erase ( self - > name_ ) ;
if ( error = = GRPC_ERROR_NONE & & ! self - > shutdown_ ) {
self - > rds_policy_ - > action s_. erase ( self - > name_ ) ;
}
self - > Unref ( DEBUG_LOCATION , " Weighted Child+timer" ) ;
self - > Unref ( DEBUG_LOCATION , " Rds Child+timer" ) ;
}
//
// WeightedTargetLb::Weighted Child::Helper
// RdsLb::Rds Child::Helper
//
RefCountedPtr < SubchannelInterface >
WeightedTargetLb : : WeightedChild : : Helper : : CreateSubchannel (
RefCountedPtr < SubchannelInterface > RdsLb : : RdsChild : : Helper : : CreateSubchannel (
const grpc_channel_args & args ) {
if ( weighted_child_ - > weighted_target _policy_- > shutting_down_ ) return nullptr ;
return weighted_child_ - > weighted_target _policy_- > channel_control_helper ( ) - > CreateSubchannel (
if ( rds_child_ - > rds _policy_- > shutting_down_ ) return nullptr ;
return rds_child_ - > rds _policy_- > channel_control_helper ( ) - > CreateSubchannel (
args ) ;
}
void WeightedTargetLb : : Weighted Child: : Helper : : UpdateState (
void RdsLb : : Rds Child: : Helper : : UpdateState (
grpc_connectivity_state state , std : : unique_ptr < SubchannelPicker > picker ) {
if ( weighted_child_ - > weighted_target _policy_- > shutting_down_ ) return ;
// Cache the picker in the Weighted Child.
weighted _child_- > picker_wrapper_ =
if ( rds_child_ - > rds _policy_- > shutting_down_ ) return ;
// Cache the picker in the Rds Child.
rds _child_- > picker_wrapper_ =
MakeRefCounted < ChildPickerWrapper > ( std : : move ( picker ) ) ;
// Decide what state to report for aggregation purposes.
// If we haven't seen a failure since the last time we were in state
// READY, then we report the state change as-is. However, once we do see
// a failure, we report TRANSIENT_FAILURE and ignore any subsequent state
// changes until we go back into state READY.
if ( ! weighted _child_- > seen_failure_since_ready_ ) {
if ( ! rds _child_- > seen_failure_since_ready_ ) {
if ( state = = GRPC_CHANNEL_TRANSIENT_FAILURE ) {
weighted _child_- > seen_failure_since_ready_ = true ;
rds _child_- > seen_failure_since_ready_ = true ;
}
} else {
if ( state ! = GRPC_CHANNEL_READY ) return ;
weighted _child_- > seen_failure_since_ready_ = false ;
rds _child_- > seen_failure_since_ready_ = false ;
}
weighted _child_- > connectivity_state_ = state ;
rds _child_- > connectivity_state_ = state ;
// Notify the LB policy.
weighted_child_ - > weighted_target _policy_- > UpdateStateLocked ( ) ;
rds_child_ - > rds _policy_- > UpdateStateLocked ( ) ;
}
void WeightedTargetLb : : Weighted Child: : Helper : : RequestReresolution ( ) {
if ( weighted_child_ - > weighted_target _policy_- > shutting_down_ ) return ;
weighted_child_ - > weighted_target _policy_- > channel_control_helper ( ) - > RequestReresolution ( ) ;
void RdsLb : : Rds Child: : Helper : : RequestReresolution ( ) {
if ( rds_child_ - > rds _policy_- > shutting_down_ ) return ;
rds_child_ - > rds _policy_- > channel_control_helper ( ) - > RequestReresolution ( ) ;
}
void WeightedTargetLb : : Weighted Child: : Helper : : AddTraceEvent ( TraceSeverity severity ,
void RdsLb : : Rds Child: : Helper : : AddTraceEvent ( TraceSeverity severity ,
StringView message ) {
if ( weighted_child_ - > weighted_target _policy_- > shutting_down_ ) return ;
weighted_child_ - > weighted_target _policy_- > channel_control_helper ( ) - > AddTraceEvent (
severity , message ) ;
if ( rds_child_ - > rds _policy_- > shutting_down_ ) return ;
rds_child_ - > rds _policy_- > channel_control_helper ( ) - > AddTraceEvent ( severity ,
message ) ;
}
//
// factory
//
class WeightedTarget LbFactory : public LoadBalancingPolicyFactory {
class Rds LbFactory : public LoadBalancingPolicyFactory {
public :
OrphanablePtr < LoadBalancingPolicy > CreateLoadBalancingPolicy (
LoadBalancingPolicy : : Args args ) const override {
return MakeOrphanable < WeightedTarget Lb> ( std : : move ( args ) ) ;
return MakeOrphanable < Rds Lb> ( std : : move ( args ) ) ;
}
const char * name ( ) const override { return kWeightedTarget ; }
const char * name ( ) const override { return kRds ; }
RefCountedPtr < LoadBalancingPolicy : : Config > ParseLoadBalancingConfig (
const Json & json , grpc_error * * error ) const override {
GPR_DEBUG_ASSERT ( error ! = nullptr & & * error = = GRPC_ERROR_NONE ) ;
if ( json . type ( ) = = Json : : Type : : JSON_NULL ) {
// weighted_target was mentioned as a policy in the deprecated
// rds was mentioned as a policy in the deprecated
// loadBalancingPolicy field or in the client API.
* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:loadBalancingPolicy error:weighted_target policy requires "
" field:loadBalancingPolicy error:rds policy requires "
" configuration. Please use loadBalancingConfig field of service "
" config instead. " ) ;
return nullptr ;
}
std : : vector < grpc_error * > error_list ;
// Weight map.
WeightedTargetLbConfig : : TargetMap target _map;
auto it = json . object_value ( ) . find ( " target s" ) ;
RdsLbConfig : : ActionMap action _map;
auto it = json . object_value ( ) . find ( " action s" ) ;
if ( it = = json . object_value ( ) . end ( ) ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:target s error:required field not present " ) ) ;
} else if ( it - > second . type ( ) ! = Json : : Type : : OBJECT ) {
" field:action s error:required field not present " ) ) ;
} else if ( it - > second . type ( ) ! = Json : : Type : : ARRAY ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:targets error:type should be object " ) ) ;
" field:actions error:type should be array " ) ) ;
} else {
for ( const auto & p : it - > second . object_value ( ) ) {
WeightedTargetLbConfig : : ChildConfig child_config ;
for ( const auto & p : it - > second . array_value ( ) ) {
auto it_name = p . object_value ( ) . find ( " name " ) ;
if ( it_name = = p . object_value ( ) . end ( ) ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:actions error: each action needs a name " ) ) ;
}
auto it_child_policy = p . object_value ( ) . find ( " child_policy " ) ;
if ( it_child_policy = = p . object_value ( ) . end ( ) ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:actions error: each action needs child policies " ) ) ;
}
RdsLbConfig : : ChildConfig child_config ;
std : : vector < grpc_error * > child_errors =
ParseChildConfig ( p . second , & child_config ) ;
ParseChildConfig ( it_child_policy - > second , & child_config ) ;
if ( ! child_errors . empty ( ) ) {
// Can't use GRPC_ERROR_CREATE_FROM_VECTOR() here, because the error
// string is not static in this case.
grpc_error * error = GRPC_ERROR_CREATE_FROM_COPIED_STRING (
absl : : StrCat ( " field:targets key: " , p . first ) . c_str ( ) ) ;
absl : : StrCat ( " field:actions name: " ,
it_name - > second . string_value ( ) )
. c_str ( ) ) ;
for ( grpc_error * child_error : child_errors ) {
error = grpc_error_add_child ( error , child_error ) ;
}
error_list . push_back ( error ) ;
} else {
target_map [ p . first ] = std : : move ( child_config ) ;
action_map [ it_name - > second . string_value ( ) ] = std : : move ( child_config ) ;
}
}
}
if ( ! error_list . empty ( ) ) {
* error = GRPC_ERROR_CREATE_FROM_VECTOR (
" weighted_target _experimental LB policy config" , & error_list ) ;
" rds _experimental LB policy config" , & error_list ) ;
return nullptr ;
}
return MakeRefCounted < WeightedTarget LbConfig> ( std : : move ( target _map) ) ;
return MakeRefCounted < Rds LbConfig> ( std : : move ( action _map) ) ;
}
private :
static std : : vector < grpc_error * > ParseChildConfig (
const Json & json , WeightedTarget LbConfig: : ChildConfig * child_config ) {
const Json & json , Rds LbConfig: : ChildConfig * child_config ) {
std : : vector < grpc_error * > error_list ;
if ( json . type ( ) ! = Json : : Type : : OBJECT ) {
if ( json . type ( ) ! = Json : : Type : : ARRAY ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" value should be of type object " ) ) ;
" value should be of type array " ) ) ;
return error_list ;
}
// Weight.
auto it = json . object_value ( ) . find ( " weight " ) ;
if ( it = = json . object_value ( ) . end ( ) ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" required field \" weight \" not specified " ) ) ;
} else if ( it - > second . type ( ) ! = Json : : Type : : NUMBER ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:weight error:must be of type number " ) ) ;
} else {
child_config - > weight =
gpr_parse_nonnegative_int ( it - > second . string_value ( ) . c_str ( ) ) ;
if ( child_config - > weight = = - 1 ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:weight error:unparseable value " ) ) ;
} else if ( child_config - > weight = = 0 ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:weight error:value must be greater than zero " ) ) ;
}
}
// Child policy.
it = json . object_value ( ) . find ( " childPolicy " ) ;
if ( it ! = json . object_value ( ) . end ( ) ) {
grpc_error * parse_error = GRPC_ERROR_NONE ;
child_config - > config =
LoadBalancingPolicyRegistry : : ParseLoadBalancingConfig ( it - > second ,
& parse_error ) ;
LoadBalancingPolicyRegistry : : ParseLoadBalancingConfig (
json . array_value ( ) , & parse_error ) ;
if ( child_config - > config = = nullptr ) {
GPR_DEBUG_ASSERT ( parse_error ! = GRPC_ERROR_NONE ) ;
std : : vector < grpc_error * > child_errors ;
@ -674,7 +619,6 @@ class WeightedTargetLbFactory : public LoadBalancingPolicyFactory {
error_list . push_back (
GRPC_ERROR_CREATE_FROM_VECTOR ( " field:childPolicy " , & child_errors ) ) ;
}
}
return error_list ;
}
} ;
@ -687,10 +631,10 @@ class WeightedTargetLbFactory : public LoadBalancingPolicyFactory {
// Plugin registration
//
void grpc_lb_policy_weighted_target _init ( ) {
void grpc_lb_policy_rds _init ( ) {
grpc_core : : LoadBalancingPolicyRegistry : : Builder : :
RegisterLoadBalancingPolicyFactory (
absl : : make_unique < grpc_core : : WeightedTarget LbFactory> ( ) ) ;
absl : : make_unique < grpc_core : : Rds LbFactory> ( ) ) ;
}
void grpc_lb_policy_weighted_target _shutdown ( ) { }
void grpc_lb_policy_rds _shutdown ( ) { }