Chiebot-Mirror
/
grpc
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

More LB policy API improvements.

Browse Source
pull/19049/head
Mark D. Roth 6 years ago
parent d43604a8b1
commit 31ec9a74ae
15 changed files with 291 additions and 237 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 164
      src/core/ext/filters/client_channel/client_channel.cc
  2. 18
      src/core/ext/filters/client_channel/lb_policy.cc
  3. 154
      src/core/ext/filters/client_channel/lb_policy.h
  4. 43
      src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
  5. 14
      src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc
  6. 17
      src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
  7. 56
      src/core/ext/filters/client_channel/lb_policy/xds/xds.cc
  8. 2
      src/core/ext/filters/client_channel/lb_policy_factory.h
  9. 2
      src/core/ext/filters/client_channel/lb_policy_registry.cc
  10. 2
      src/core/ext/filters/client_channel/lb_policy_registry.h
  11. 2
      src/core/ext/filters/client_channel/resolver_result_parsing.cc
  12. 6
      src/core/ext/filters/client_channel/resolver_result_parsing.h
  13. 6
      src/core/ext/filters/client_channel/resolving_lb_policy.cc
  14. 8
      src/core/ext/filters/client_channel/resolving_lb_policy.h
  15. 34
      test/core/util/test_lb_policies.cc

164
src/core/ext/filters/client_channel/client_channel.cc
Unescape View File

@ -105,7 +105,6 @@ namespace {
class ChannelData { class ChannelData {
public: public:
struct QueuedPick { struct QueuedPick {
LoadBalancingPolicy::PickArgs pick;
grpc_call_element* elem; grpc_call_element* elem;
QueuedPick* next = nullptr; QueuedPick* next = nullptr;
}; };
@ -223,7 +222,7 @@ class ChannelData {
static bool ProcessResolverResultLocked( static bool ProcessResolverResultLocked(
void* arg, Resolver::Result* result, const char** lb_policy_name, void* arg, Resolver::Result* result, const char** lb_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig>* lb_policy_config, RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config,
grpc_error** service_config_error); grpc_error** service_config_error);
grpc_error* DoPingLocked(grpc_transport_op* op); grpc_error* DoPingLocked(grpc_transport_op* op);
@ -236,7 +235,7 @@ class ChannelData {
const Resolver::Result& resolver_result, const Resolver::Result& resolver_result,
const internal::ClientChannelGlobalParsedConfig* parsed_service_config, const internal::ClientChannelGlobalParsedConfig* parsed_service_config,
UniquePtr<char>* lb_policy_name, UniquePtr<char>* lb_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig>* lb_policy_config); RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config);
// //
// Fields set at construction and never modified. // Fields set at construction and never modified.
@ -314,6 +313,16 @@ class CallData {
private: private:
class QueuedPickCanceller; class QueuedPickCanceller;
class LbCallState : public LoadBalancingPolicy::CallState {
public:
explicit LbCallState(CallData* calld) : calld_(calld) {}
void* Alloc(size_t size) override { return calld_->arena_->Alloc(size); }
private:
CallData* calld_;
};
// State used for starting a retryable batch on a subchannel call. // State used for starting a retryable batch on a subchannel call.
// This provides its own grpc_transport_stream_op_batch and other data // This provides its own grpc_transport_stream_op_batch and other data
// structures needed to populate the ops in the batch. // structures needed to populate the ops in the batch.
@ -449,8 +458,9 @@ class CallData {
grpc_call_element* elem, SubchannelCallBatchData* batch_data, grpc_call_element* elem, SubchannelCallBatchData* batch_data,
SubchannelCallRetryState* retry_state); SubchannelCallRetryState* retry_state);
static void MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy( static void RecvTrailingMetadataReadyForLoadBalancingPolicy(
const LoadBalancingPolicy::PickArgs& pick, void* arg, grpc_error* error);
void MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy(
grpc_transport_stream_op_batch* batch); grpc_transport_stream_op_batch* batch);
// Returns the index into pending_batches_ to be used for batch. // Returns the index into pending_batches_ to be used for batch.
@ -640,8 +650,19 @@ class CallData {
bool pick_queued_ = false; bool pick_queued_ = false;
bool service_config_applied_ = false; bool service_config_applied_ = false;
QueuedPickCanceller* pick_canceller_ = nullptr; QueuedPickCanceller* pick_canceller_ = nullptr;
LbCallState lb_call_state_;
RefCountedPtr<ConnectedSubchannel> connected_subchannel_;
void (*lb_recv_trailing_metadata_ready_)(
void* user_data, grpc_metadata_batch* recv_trailing_metadata,
LoadBalancingPolicy::CallState* call_state) = nullptr;
void* lb_recv_trailing_metadata_ready_user_data_ = nullptr;
grpc_closure pick_closure_; grpc_closure pick_closure_;
// For intercepting recv_trailing_metadata_ready for the LB policy.
grpc_metadata_batch* recv_trailing_metadata_ = nullptr;
grpc_closure recv_trailing_metadata_ready_;
grpc_closure* original_recv_trailing_metadata_ready_ = nullptr;
grpc_polling_entity* pollent_ = nullptr; grpc_polling_entity* pollent_ = nullptr;
// Batches are added to this list when received from above. // Batches are added to this list when received from above.
@ -1143,7 +1164,7 @@ void ChannelData::ProcessLbPolicy(
const Resolver::Result& resolver_result, const Resolver::Result& resolver_result,
const internal::ClientChannelGlobalParsedConfig* parsed_service_config, const internal::ClientChannelGlobalParsedConfig* parsed_service_config,
UniquePtr<char>* lb_policy_name, UniquePtr<char>* lb_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig>* lb_policy_config) { RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config) {
// Prefer the LB policy name found in the service config. // Prefer the LB policy name found in the service config.
if (parsed_service_config != nullptr && if (parsed_service_config != nullptr &&
parsed_service_config->parsed_lb_config() != nullptr) { parsed_service_config->parsed_lb_config() != nullptr) {
@ -1191,7 +1212,7 @@ void ChannelData::ProcessLbPolicy(
// resolver result update. // resolver result update.
bool ChannelData::ProcessResolverResultLocked( bool ChannelData::ProcessResolverResultLocked(
void* arg, Resolver::Result* result, const char** lb_policy_name, void* arg, Resolver::Result* result, const char** lb_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig>* lb_policy_config, RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config,
grpc_error** service_config_error) { grpc_error** service_config_error) {
ChannelData* chand = static_cast<ChannelData*>(arg); ChannelData* chand = static_cast<ChannelData*>(arg);
RefCountedPtr<ServiceConfig> service_config; RefCountedPtr<ServiceConfig> service_config;
@ -1312,19 +1333,18 @@ grpc_error* ChannelData::DoPingLocked(grpc_transport_op* op) {
if (grpc_connectivity_state_check(&state_tracker_) != GRPC_CHANNEL_READY) { if (grpc_connectivity_state_check(&state_tracker_) != GRPC_CHANNEL_READY) {
return GRPC_ERROR_CREATE_FROM_STATIC_STRING("channel not connected"); return GRPC_ERROR_CREATE_FROM_STATIC_STRING("channel not connected");
} }
LoadBalancingPolicy::PickArgs pick; LoadBalancingPolicy::PickResult result =
grpc_error* error = GRPC_ERROR_NONE; picker_->Pick(LoadBalancingPolicy::PickArgs());
picker_->Pick(&pick, &error); if (result.connected_subchannel != nullptr) {
if (pick.connected_subchannel != nullptr) { result.connected_subchannel->Ping(op->send_ping.on_initiate,
pick.connected_subchannel->Ping(op->send_ping.on_initiate, op->send_ping.on_ack);
op->send_ping.on_ack);
} else { } else {
if (error == GRPC_ERROR_NONE) { if (result.error == GRPC_ERROR_NONE) {
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( result.error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"LB policy dropped call on ping"); "LB policy dropped call on ping");
} }
} }
return error; return result.error;
} }
void ChannelData::StartTransportOpLocked(void* arg, grpc_error* ignored) { void ChannelData::StartTransportOpLocked(void* arg, grpc_error* ignored) {
@ -1505,6 +1525,7 @@ CallData::CallData(grpc_call_element* elem, const ChannelData& chand,
owning_call_(args.call_stack), owning_call_(args.call_stack),
call_combiner_(args.call_combiner), call_combiner_(args.call_combiner),
call_context_(args.context), call_context_(args.context),
lb_call_state_(this),
pending_send_initial_metadata_(false), pending_send_initial_metadata_(false),
pending_send_message_(false), pending_send_message_(false),
pending_send_trailing_metadata_(false), pending_send_trailing_metadata_(false),
@ -1737,18 +1758,30 @@ void CallData::FreeCachedSendOpDataForCompletedBatch(
// LB recv_trailing_metadata_ready handling // LB recv_trailing_metadata_ready handling
// //
void CallData::RecvTrailingMetadataReadyForLoadBalancingPolicy(
void* arg, grpc_error* error) {
CallData* calld = static_cast<CallData*>(arg);
// Invoke callback to LB policy.
calld->lb_recv_trailing_metadata_ready_(
calld->lb_recv_trailing_metadata_ready_user_data_,
calld->recv_trailing_metadata_, &calld->lb_call_state_);
// Chain to original callback.
GRPC_CLOSURE_RUN(calld->original_recv_trailing_metadata_ready_,
GRPC_ERROR_REF(error));
}
void CallData::MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy( void CallData::MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy(
const LoadBalancingPolicy::PickArgs& pick,
grpc_transport_stream_op_batch* batch) { grpc_transport_stream_op_batch* batch) {
if (pick.recv_trailing_metadata_ready != nullptr) { if (lb_recv_trailing_metadata_ready_ != nullptr) {
*pick.original_recv_trailing_metadata_ready = recv_trailing_metadata_ =
batch->payload->recv_trailing_metadata.recv_trailing_metadata;
original_recv_trailing_metadata_ready_ =
batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready; batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
GRPC_CLOSURE_INIT(&recv_trailing_metadata_ready_,
RecvTrailingMetadataReadyForLoadBalancingPolicy, this,
grpc_schedule_on_exec_ctx);
batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready = batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
pick.recv_trailing_metadata_ready; &recv_trailing_metadata_ready_;
if (pick.recv_trailing_metadata != nullptr) {
*pick.recv_trailing_metadata =
batch->payload->recv_trailing_metadata.recv_trailing_metadata;
}
} }
} }
@ -1894,8 +1927,7 @@ void CallData::PendingBatchesFail(
grpc_transport_stream_op_batch* batch = pending->batch; grpc_transport_stream_op_batch* batch = pending->batch;
if (batch != nullptr) { if (batch != nullptr) {
if (batch->recv_trailing_metadata) { if (batch->recv_trailing_metadata) {
MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy(pick_.pick, MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy(batch);
batch);
} }
batch->handler_private.extra_arg = this; batch->handler_private.extra_arg = this;
GRPC_CLOSURE_INIT(&batch->handler_private.closure, GRPC_CLOSURE_INIT(&batch->handler_private.closure,
@ -1949,8 +1981,7 @@ void CallData::PendingBatchesResume(grpc_call_element* elem) {
grpc_transport_stream_op_batch* batch = pending->batch; grpc_transport_stream_op_batch* batch = pending->batch;
if (batch != nullptr) { if (batch != nullptr) {
if (batch->recv_trailing_metadata) { if (batch->recv_trailing_metadata) {
MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy(pick_.pick, MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy(batch);
batch);
} }
batch->handler_private.extra_arg = subchannel_call_.get(); batch->handler_private.extra_arg = subchannel_call_.get();
GRPC_CLOSURE_INIT(&batch->handler_private.closure, GRPC_CLOSURE_INIT(&batch->handler_private.closure,
@ -2011,7 +2042,7 @@ void CallData::DoRetry(grpc_call_element* elem,
GPR_ASSERT(retry_policy != nullptr); GPR_ASSERT(retry_policy != nullptr);
// Reset subchannel call and connected subchannel. // Reset subchannel call and connected subchannel.
subchannel_call_.reset(); subchannel_call_.reset();
pick_.pick.connected_subchannel.reset(); connected_subchannel_.reset();
// Compute backoff delay. // Compute backoff delay.
grpc_millis next_attempt_time; grpc_millis next_attempt_time;
if (server_pushback_ms >= 0) { if (server_pushback_ms >= 0) {
@ -2868,7 +2899,7 @@ void CallData::AddRetriableRecvTrailingMetadataOp(
.recv_trailing_metadata_ready = .recv_trailing_metadata_ready =
&retry_state->recv_trailing_metadata_ready; &retry_state->recv_trailing_metadata_ready;
MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy( MaybeInjectRecvTrailingMetadataReadyForLoadBalancingPolicy(
pick_.pick, &batch_data->batch); &batch_data->batch);
} }
void CallData::StartInternalRecvTrailingMetadata(grpc_call_element* elem) { void CallData::StartInternalRecvTrailingMetadata(grpc_call_element* elem) {
@ -3135,8 +3166,7 @@ void CallData::CreateSubchannelCall(grpc_call_element* elem) {
// need to use a separate call context for each subchannel call. // need to use a separate call context for each subchannel call.
call_context_, call_combiner_, parent_data_size}; call_context_, call_combiner_, parent_data_size};
grpc_error* error = GRPC_ERROR_NONE; grpc_error* error = GRPC_ERROR_NONE;
subchannel_call_ = subchannel_call_ = connected_subchannel_->CreateCall(call_args, &error);
pick_.pick.connected_subchannel->CreateCall(call_args, &error);
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO, "chand=%p calld=%p: create subchannel_call=%p: error=%s", gpr_log(GPR_INFO, "chand=%p calld=%p: create subchannel_call=%p: error=%s",
chand, this, subchannel_call_.get(), grpc_error_string(error)); chand, this, subchannel_call_.get(), grpc_error_string(error));
@ -3297,13 +3327,14 @@ void CallData::MaybeApplyServiceConfigToCallLocked(grpc_call_element* elem) {
} }
} }
const char* PickResultName(LoadBalancingPolicy::PickResult result) { const char* PickResultTypeName(
switch (result) { LoadBalancingPolicy::PickResult::ResultType type) {
case LoadBalancingPolicy::PICK_COMPLETE: switch (type) {
case LoadBalancingPolicy::PickResult::PICK_COMPLETE:
return "COMPLETE"; return "COMPLETE";
case LoadBalancingPolicy::PICK_QUEUE: case LoadBalancingPolicy::PickResult::PICK_QUEUE:
return "QUEUE"; return "QUEUE";
case LoadBalancingPolicy::PICK_TRANSIENT_FAILURE: case LoadBalancingPolicy::PickResult::PICK_TRANSIENT_FAILURE:
return "TRANSIENT_FAILURE"; return "TRANSIENT_FAILURE";
} }
GPR_UNREACHABLE_CODE(return "UNKNOWN"); GPR_UNREACHABLE_CODE(return "UNKNOWN");
@ -3313,8 +3344,10 @@ void CallData::StartPickLocked(void* arg, grpc_error* error) {
grpc_call_element* elem = static_cast<grpc_call_element*>(arg); grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
CallData* calld = static_cast<CallData*>(elem->call_data); CallData* calld = static_cast<CallData*>(elem->call_data);
ChannelData* chand = static_cast<ChannelData*>(elem->channel_data); ChannelData* chand = static_cast<ChannelData*>(elem->channel_data);
GPR_ASSERT(calld->pick_.pick.connected_subchannel == nullptr); GPR_ASSERT(calld->connected_subchannel_ == nullptr);
GPR_ASSERT(calld->subchannel_call_ == nullptr); GPR_ASSERT(calld->subchannel_call_ == nullptr);
// Apply service config to call if needed.
calld->MaybeApplyServiceConfigToCallLocked(elem);
// If this is a retry, use the send_initial_metadata payload that // If this is a retry, use the send_initial_metadata payload that
// we've cached; otherwise, use the pending batch. The // we've cached; otherwise, use the pending batch. The
// send_initial_metadata batch will be the first pending batch in the // send_initial_metadata batch will be the first pending batch in the
@ -3325,58 +3358,58 @@ void CallData::StartPickLocked(void* arg, grpc_error* error) {
// allocate the subchannel batch earlier so that we can give the // allocate the subchannel batch earlier so that we can give the
// subchannel's copy of the metadata batch (which is copied for each // subchannel's copy of the metadata batch (which is copied for each
// attempt) to the LB policy instead the one from the parent channel. // attempt) to the LB policy instead the one from the parent channel.
calld->pick_.pick.initial_metadata = LoadBalancingPolicy::PickArgs pick_args;
pick_args.call_state = &calld->lb_call_state_;
pick_args.initial_metadata =
calld->seen_send_initial_metadata_ calld->seen_send_initial_metadata_
? &calld->send_initial_metadata_ ? &calld->send_initial_metadata_
: calld->pending_batches_[0] : calld->pending_batches_[0]
.batch->payload->send_initial_metadata.send_initial_metadata; .batch->payload->send_initial_metadata.send_initial_metadata;
uint32_t* send_initial_metadata_flags = // Grab initial metadata flags so that we can check later if the call has
// wait_for_ready enabled.
const uint32_t send_initial_metadata_flags =
calld->seen_send_initial_metadata_ calld->seen_send_initial_metadata_
? &calld->send_initial_metadata_flags_ ? calld->send_initial_metadata_flags_
: &calld->pending_batches_[0] : calld->pending_batches_[0]
.batch->payload->send_initial_metadata .batch->payload->send_initial_metadata
.send_initial_metadata_flags; .send_initial_metadata_flags;
// Apply service config to call if needed.
calld->MaybeApplyServiceConfigToCallLocked(elem);
// When done, we schedule this closure to leave the data plane combiner. // When done, we schedule this closure to leave the data plane combiner.
GRPC_CLOSURE_INIT(&calld->pick_closure_, PickDone, elem, GRPC_CLOSURE_INIT(&calld->pick_closure_, PickDone, elem,
grpc_schedule_on_exec_ctx); grpc_schedule_on_exec_ctx);
// Attempt pick. // Attempt pick.
error = GRPC_ERROR_NONE; auto result = chand->picker()->Pick(pick_args);
auto pick_result = chand->picker()->Pick(&calld->pick_.pick, &error);
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
"chand=%p calld=%p: LB pick returned %s (connected_subchannel=%p, " "chand=%p calld=%p: LB pick returned %s (connected_subchannel=%p, "
"error=%s)", "error=%s)",
chand, calld, PickResultName(pick_result), chand, calld, PickResultTypeName(result.type),
calld->pick_.pick.connected_subchannel.get(), result.connected_subchannel.get(), grpc_error_string(result.error));
grpc_error_string(error));
} }
switch (pick_result) { switch (result.type) {
case LoadBalancingPolicy::PICK_TRANSIENT_FAILURE: { case LoadBalancingPolicy::PickResult::PICK_TRANSIENT_FAILURE: {
// If we're shutting down, fail all RPCs. // If we're shutting down, fail all RPCs.
grpc_error* disconnect_error = chand->disconnect_error(); grpc_error* disconnect_error = chand->disconnect_error();
if (disconnect_error != GRPC_ERROR_NONE) { if (disconnect_error != GRPC_ERROR_NONE) {
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(result.error);
GRPC_CLOSURE_SCHED(&calld->pick_closure_, GRPC_CLOSURE_SCHED(&calld->pick_closure_,
GRPC_ERROR_REF(disconnect_error)); GRPC_ERROR_REF(disconnect_error));
break; break;
} }
// If wait_for_ready is false, then the error indicates the RPC // If wait_for_ready is false, then the error indicates the RPC
// attempt's final status. // attempt's final status.
if ((*send_initial_metadata_flags & if ((send_initial_metadata_flags &
GRPC_INITIAL_METADATA_WAIT_FOR_READY) == 0) { GRPC_INITIAL_METADATA_WAIT_FOR_READY) == 0) {
// Retry if appropriate; otherwise, fail. // Retry if appropriate; otherwise, fail.
grpc_status_code status = GRPC_STATUS_OK; grpc_status_code status = GRPC_STATUS_OK;
grpc_error_get_status(error, calld->deadline_, &status, nullptr, grpc_error_get_status(result.error, calld->deadline_, &status, nullptr,
nullptr, nullptr); nullptr, nullptr);
if (!calld->enable_retries_ || if (!calld->enable_retries_ ||
!calld->MaybeRetry(elem, nullptr /* batch_data */, status, !calld->MaybeRetry(elem, nullptr /* batch_data */, status,
nullptr /* server_pushback_md */)) { nullptr /* server_pushback_md */)) {
grpc_error* new_error = grpc_error* new_error =
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Failed to pick subchannel", &error, 1); "Failed to pick subchannel", &result.error, 1);
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(result.error);
GRPC_CLOSURE_SCHED(&calld->pick_closure_, new_error); GRPC_CLOSURE_SCHED(&calld->pick_closure_, new_error);
} }
if (calld->pick_queued_) calld->RemoveCallFromQueuedPicksLocked(elem); if (calld->pick_queued_) calld->RemoveCallFromQueuedPicksLocked(elem);
@ -3384,19 +3417,24 @@ void CallData::StartPickLocked(void* arg, grpc_error* error) {
} }
// If wait_for_ready is true, then queue to retry when we get a new // If wait_for_ready is true, then queue to retry when we get a new
// picker. // picker.
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(result.error);
} }
// Fallthrough // Fallthrough
case LoadBalancingPolicy::PICK_QUEUE: case LoadBalancingPolicy::PickResult::PICK_QUEUE:
if (!calld->pick_queued_) calld->AddCallToQueuedPicksLocked(elem); if (!calld->pick_queued_) calld->AddCallToQueuedPicksLocked(elem);
break; break;
default: // PICK_COMPLETE default: // PICK_COMPLETE
// Handle drops. // Handle drops.
if (GPR_UNLIKELY(calld->pick_.pick.connected_subchannel == nullptr)) { if (GPR_UNLIKELY(result.connected_subchannel == nullptr)) {
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( result.error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"Call dropped by load balancing policy"); "Call dropped by load balancing policy");
} }
GRPC_CLOSURE_SCHED(&calld->pick_closure_, error); calld->connected_subchannel_ = std::move(result.connected_subchannel);
calld->lb_recv_trailing_metadata_ready_ =
result.recv_trailing_metadata_ready;
calld->lb_recv_trailing_metadata_ready_user_data_ =
result.recv_trailing_metadata_ready_user_data;
GRPC_CLOSURE_SCHED(&calld->pick_closure_, result.error);
if (calld->pick_queued_) calld->RemoveCallFromQueuedPicksLocked(elem); if (calld->pick_queued_) calld->RemoveCallFromQueuedPicksLocked(elem);
} }
} }

18
src/core/ext/filters/client_channel/lb_policy.cc
Unescape View File

@ -105,7 +105,7 @@ LoadBalancingPolicy::UpdateArgs& LoadBalancingPolicy::UpdateArgs::operator=(
// //
LoadBalancingPolicy::PickResult LoadBalancingPolicy::QueuePicker::Pick( LoadBalancingPolicy::PickResult LoadBalancingPolicy::QueuePicker::Pick(
PickArgs* pick, grpc_error** error) { PickArgs args) {
// We invoke the parent's ExitIdleLocked() via a closure instead // We invoke the parent's ExitIdleLocked() via a closure instead
// of doing it directly here, for two reasons: // of doing it directly here, for two reasons:
// 1. ExitIdleLocked() may cause the policy's state to change and // 1. ExitIdleLocked() may cause the policy's state to change and
@ -125,7 +125,9 @@ LoadBalancingPolicy::PickResult LoadBalancingPolicy::QueuePicker::Pick(
grpc_combiner_scheduler(parent_->combiner())), grpc_combiner_scheduler(parent_->combiner())),
GRPC_ERROR_NONE); GRPC_ERROR_NONE);
} }
return PICK_QUEUE; PickResult result;
result.type = PickResult::PICK_QUEUE;
return result;
} }
void LoadBalancingPolicy::QueuePicker::CallExitIdle(void* arg, void LoadBalancingPolicy::QueuePicker::CallExitIdle(void* arg,
@ -135,4 +137,16 @@ void LoadBalancingPolicy::QueuePicker::CallExitIdle(void* arg,
parent->Unref(); parent->Unref();
} }
//
// LoadBalancingPolicy::TransientFailurePicker
//
LoadBalancingPolicy::PickResult
LoadBalancingPolicy::TransientFailurePicker::Pick(PickArgs args) {
PickResult result;
result.type = PickResult::PICK_TRANSIENT_FAILURE;
result.error = GRPC_ERROR_REF(error_);
return result;
}
} // namespace grpc_core } // namespace grpc_core

154
src/core/ext/filters/client_channel/lb_policy.h
Unescape View File

@ -32,21 +32,9 @@
#include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/iomgr/polling_entity.h"
#include "src/core/lib/transport/connectivity_state.h" #include "src/core/lib/transport/connectivity_state.h"
extern grpc_core::DebugOnlyTraceFlag grpc_trace_lb_policy_refcount;
namespace grpc_core { namespace grpc_core {
/// Interface for parsed forms of load balancing configs found in a service extern DebugOnlyTraceFlag grpc_trace_lb_policy_refcount;
/// config.
class ParsedLoadBalancingConfig : public RefCounted<ParsedLoadBalancingConfig> {
public:
virtual ~ParsedLoadBalancingConfig() = default;
// Returns the load balancing policy name
virtual const char* name() const GRPC_ABSTRACT;
GRPC_ABSTRACT_BASE_CLASS;
};
/// Interface for load balancing policies. /// Interface for load balancing policies.
/// ///
@ -89,66 +77,77 @@ class ParsedLoadBalancingConfig : public RefCounted<ParsedLoadBalancingConfig> {
// interested_parties() hooks from the API. // interested_parties() hooks from the API.
class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> { class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> {
public: public:
/// Interface for accessing per-call state.
class CallState {
public:
CallState() = default;
virtual ~CallState() = default;
/// Allocates memory associated with the call, which will be
/// automatically freed when the call is complete.
/// It is more efficient to use this than to allocate memory directly
/// for allocations that need to be made on a per-call basis.
virtual void* Alloc(size_t size) GRPC_ABSTRACT;
GRPC_ABSTRACT_BASE_CLASS
};
/// Arguments used when picking a subchannel for an RPC. /// Arguments used when picking a subchannel for an RPC.
struct PickArgs { struct PickArgs {
///
/// Input parameters.
///
/// Initial metadata associated with the picking call. /// Initial metadata associated with the picking call.
/// The LB policy may use the existing metadata to influence its routing /// The LB policy may use the existing metadata to influence its routing
/// decision, and it may add new metadata elements to be sent with the /// decision, and it may add new metadata elements to be sent with the
/// call to the chosen backend. /// call to the chosen backend.
// TODO(roth): Provide a more generic metadata API here. // TODO(roth): Provide a more generic metadata API here.
grpc_metadata_batch* initial_metadata = nullptr; grpc_metadata_batch* initial_metadata = nullptr;
/// Storage for LB token in \a initial_metadata, or nullptr if not used. /// An interface for accessing call state. Can be used to allocate
// TODO(roth): Remove this from the API. Maybe have the LB policy /// data associated with the call in an efficient way.
// allocate this on the arena instead? CallState* call_state;
grpc_linked_mdelem lb_token_mdelem_storage;
///
/// Output parameters.
///
/// Will be set to the selected subchannel, or nullptr on failure or when
/// the LB policy decides to drop the call.
RefCountedPtr<ConnectedSubchannel> connected_subchannel;
/// Callback set by lb policy to be notified of trailing metadata.
/// The callback must be scheduled on grpc_schedule_on_exec_ctx.
// TODO(roth): Provide a cleaner callback API.
grpc_closure* recv_trailing_metadata_ready = nullptr;
/// The address that will be set to point to the original
/// recv_trailing_metadata_ready callback, to be invoked by the LB
/// policy's recv_trailing_metadata_ready callback when complete.
/// Must be non-null if recv_trailing_metadata_ready is non-null.
// TODO(roth): Consider making the recv_trailing_metadata closure a
// synchronous callback, in which case it is not responsible for
// chaining to the next callback, so this can be removed from the API.
grpc_closure** original_recv_trailing_metadata_ready = nullptr;
/// If this is not nullptr, then the client channel will point it to the
/// call's trailing metadata before invoking recv_trailing_metadata_ready.
/// If this is nullptr, then the callback will still be called.
/// The lb does not have ownership of the metadata.
// TODO(roth): If we make this a synchronous callback, then this can
// be passed to the callback as a parameter and can be removed from
// the API here.
grpc_metadata_batch** recv_trailing_metadata = nullptr;
}; };
/// The result of picking a subchannel for an RPC. /// The result of picking a subchannel for an RPC.
enum PickResult { struct PickResult {
// Pick complete. If connected_subchannel is non-null, client channel enum ResultType {
// can immediately proceed with the call on connected_subchannel; /// Pick complete. If connected_subchannel is non-null, client channel
// otherwise, call should be dropped. /// can immediately proceed with the call on connected_subchannel;
PICK_COMPLETE, /// otherwise, call should be dropped.
// Pick cannot be completed until something changes on the control PICK_COMPLETE,
// plane. Client channel will queue the pick and try again the /// Pick cannot be completed until something changes on the control
// next time the picker is updated. /// plane. Client channel will queue the pick and try again the
PICK_QUEUE, /// next time the picker is updated.
// LB policy is in transient failure. If the pick is wait_for_ready, PICK_QUEUE,
// client channel will wait for the next picker and try again; /// LB policy is in transient failure. If the pick is wait_for_ready,
// otherwise, the call will be failed immediately (although it may /// client channel will wait for the next picker and try again;
// be retried if the client channel is configured to do so). /// otherwise, the call will be failed immediately (although it may
// The Pick() method will set its error parameter if this value is /// be retried if the client channel is configured to do so).
// returned. /// The Pick() method will set its error parameter if this value is
PICK_TRANSIENT_FAILURE, /// returned.
PICK_TRANSIENT_FAILURE,
};
ResultType type;
/// Used only if type is PICK_COMPLETE. Will be set to the selected
/// subchannel, or nullptr if the LB policy decides to drop the call.
RefCountedPtr<ConnectedSubchannel> connected_subchannel;
/// Used only if type is PICK_TRANSIENT_FAILURE.
/// Error to be set when returning a transient failure.
// TODO(roth): Replace this with something similar to grpc::Status,
// so that we don't expose grpc_error to this API.
grpc_error* error = GRPC_ERROR_NONE;
/// Used only if type is PICK_COMPLETE.
/// Callback set by lb policy to be notified of trailing metadata.
/// The user_data argument will be set to the
/// recv_trailing_metadata_ready_user_data field.
/// recv_trailing_metadata will be set to the metadata, which may be
/// modified by the callback. The callback does not take ownership,
/// however, so any data that needs to be used after returning must
/// be copied.
void (*recv_trailing_metadata_ready)(
void* user_data, grpc_metadata_batch* recv_trailing_metadata,
CallState* call_state) = nullptr;
void* recv_trailing_metadata_ready_user_data = nullptr;
}; };
/// A subchannel picker is the object used to pick the subchannel to /// A subchannel picker is the object used to pick the subchannel to
@ -162,17 +161,14 @@ class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> {
/// live in the LB policy object itself. /// live in the LB policy object itself.
/// ///
/// Currently, pickers are always accessed from within the /// Currently, pickers are always accessed from within the
/// client_channel combiner, so they do not have to be thread-safe. /// client_channel data plane combiner, so they do not have to be
// TODO(roth): In a subsequent PR, split the data plane work (i.e., /// thread-safe.
// the interaction with the picker) and the control plane work (i.e.,
// the interaction with the LB policy) into two different
// synchronization mechanisms, to avoid lock contention between the two.
class SubchannelPicker { class SubchannelPicker {
public: public:
SubchannelPicker() = default; SubchannelPicker() = default;
virtual ~SubchannelPicker() = default; virtual ~SubchannelPicker() = default;
virtual PickResult Pick(PickArgs* pick, grpc_error** error) GRPC_ABSTRACT; virtual PickResult Pick(PickArgs args) GRPC_ABSTRACT;
GRPC_ABSTRACT_BASE_CLASS GRPC_ABSTRACT_BASE_CLASS
}; };
@ -208,11 +204,24 @@ class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> {
GRPC_ABSTRACT_BASE_CLASS GRPC_ABSTRACT_BASE_CLASS
}; };
/// Interface for configuration data used by an LB policy implementation.
/// Individual implementations will create a subclass that adds methods to
/// return the parameters they need.
class Config : public RefCounted<Config> {
public:
virtual ~Config() = default;
// Returns the load balancing policy name
virtual const char* name() const GRPC_ABSTRACT;
GRPC_ABSTRACT_BASE_CLASS
};
/// Data passed to the UpdateLocked() method when new addresses and /// Data passed to the UpdateLocked() method when new addresses and
/// config are available. /// config are available.
struct UpdateArgs { struct UpdateArgs {
ServerAddressList addresses; ServerAddressList addresses;
RefCountedPtr<ParsedLoadBalancingConfig> config; RefCountedPtr<Config> config;
const grpc_channel_args* args = nullptr; const grpc_channel_args* args = nullptr;
// TODO(roth): Remove everything below once channel args is // TODO(roth): Remove everything below once channel args is
@ -291,7 +300,7 @@ class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> {
explicit QueuePicker(RefCountedPtr<LoadBalancingPolicy> parent) explicit QueuePicker(RefCountedPtr<LoadBalancingPolicy> parent)
: parent_(std::move(parent)) {} : parent_(std::move(parent)) {}
PickResult Pick(PickArgs* pick, grpc_error** error) override; PickResult Pick(PickArgs args) override;
private: private:
static void CallExitIdle(void* arg, grpc_error* error); static void CallExitIdle(void* arg, grpc_error* error);
@ -306,10 +315,7 @@ class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> {
explicit TransientFailurePicker(grpc_error* error) : error_(error) {} explicit TransientFailurePicker(grpc_error* error) : error_(error) {}
~TransientFailurePicker() override { GRPC_ERROR_UNREF(error_); } ~TransientFailurePicker() override { GRPC_ERROR_UNREF(error_); }
PickResult Pick(PickArgs* pick, grpc_error** error) override { PickResult Pick(PickArgs args) override;
*error = GRPC_ERROR_REF(error_);
return PICK_TRANSIENT_FAILURE;
}
private: private:
grpc_error* error_; grpc_error* error_;

43
src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
Unescape View File

@ -118,19 +118,19 @@ namespace {
constexpr char kGrpclb[] = "grpclb"; constexpr char kGrpclb[] = "grpclb";
class ParsedGrpcLbConfig : public ParsedLoadBalancingConfig { class ParsedGrpcLbConfig : public LoadBalancingPolicy::Config {
public: public:
explicit ParsedGrpcLbConfig( explicit ParsedGrpcLbConfig(
RefCountedPtr<ParsedLoadBalancingConfig> child_policy) RefCountedPtr<LoadBalancingPolicy::Config> child_policy)
: child_policy_(std::move(child_policy)) {} : child_policy_(std::move(child_policy)) {}
const char* name() const override { return kGrpclb; } const char* name() const override { return kGrpclb; }
RefCountedPtr<ParsedLoadBalancingConfig> child_policy() const { RefCountedPtr<LoadBalancingPolicy::Config> child_policy() const {
return child_policy_; return child_policy_;
} }
private: private:
RefCountedPtr<ParsedLoadBalancingConfig> child_policy_; RefCountedPtr<LoadBalancingPolicy::Config> child_policy_;
}; };
class GrpcLb : public LoadBalancingPolicy { class GrpcLb : public LoadBalancingPolicy {
@ -274,7 +274,7 @@ class GrpcLb : public LoadBalancingPolicy {
child_picker_(std::move(child_picker)), child_picker_(std::move(child_picker)),
client_stats_(std::move(client_stats)) {} client_stats_(std::move(client_stats)) {}
PickResult Pick(PickArgs* pick, grpc_error** error) override; PickResult Pick(PickArgs args) override;
private: private:
// Storing the address for logging, but not holding a ref. // Storing the address for logging, but not holding a ref.
@ -394,7 +394,7 @@ class GrpcLb : public LoadBalancingPolicy {
// until it reports READY, at which point it will be moved to child_policy_. // until it reports READY, at which point it will be moved to child_policy_.
OrphanablePtr<LoadBalancingPolicy> pending_child_policy_; OrphanablePtr<LoadBalancingPolicy> pending_child_policy_;
// The child policy config. // The child policy config.
RefCountedPtr<ParsedLoadBalancingConfig> child_policy_config_; RefCountedPtr<LoadBalancingPolicy::Config> child_policy_config_;
// Child policy in state READY. // Child policy in state READY.
bool child_policy_ready_ = false; bool child_policy_ready_ = false;
}; };
@ -561,7 +561,8 @@ const char* GrpcLb::Serverlist::ShouldDrop() {
// GrpcLb::Picker // GrpcLb::Picker
// //
GrpcLb::PickResult GrpcLb::Picker::Pick(PickArgs* pick, grpc_error** error) { GrpcLb::PickResult GrpcLb::Picker::Pick(PickArgs args) {
PickResult result;
// Check if we should drop the call. // Check if we should drop the call.
const char* drop_token = serverlist_->ShouldDrop(); const char* drop_token = serverlist_->ShouldDrop();
if (drop_token != nullptr) { if (drop_token != nullptr) {
@ -573,26 +574,28 @@ GrpcLb::PickResult GrpcLb::Picker::Pick(PickArgs* pick, grpc_error** error) {
if (client_stats_ != nullptr) { if (client_stats_ != nullptr) {
client_stats_->AddCallDropped(drop_token); client_stats_->AddCallDropped(drop_token);
} }
return PICK_COMPLETE; result.type = PickResult::PICK_COMPLETE;
return result;
} }
// Forward pick to child policy. // Forward pick to child policy.
PickResult result = child_picker_->Pick(pick, error); result = child_picker_->Pick(args);
// If pick succeeded, add LB token to initial metadata. // If pick succeeded, add LB token to initial metadata.
if (result == PickResult::PICK_COMPLETE && if (result.type == PickResult::PICK_COMPLETE &&
pick->connected_subchannel != nullptr) { result.connected_subchannel != nullptr) {
const grpc_arg* arg = grpc_channel_args_find( const grpc_arg* arg = grpc_channel_args_find(
pick->connected_subchannel->args(), GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN); result.connected_subchannel->args(), GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN);
if (arg == nullptr) { if (arg == nullptr) {
gpr_log(GPR_ERROR, gpr_log(GPR_ERROR,
"[grpclb %p picker %p] No LB token for connected subchannel " "[grpclb %p picker %p] No LB token for connected subchannel %p",
"pick %p", parent_, this, result.connected_subchannel.get());
parent_, this, pick);
abort(); abort();
} }
grpc_mdelem lb_token = {reinterpret_cast<uintptr_t>(arg->value.pointer.p)}; grpc_mdelem lb_token = {reinterpret_cast<uintptr_t>(arg->value.pointer.p)};
GPR_ASSERT(!GRPC_MDISNULL(lb_token)); GPR_ASSERT(!GRPC_MDISNULL(lb_token));
grpc_linked_mdelem* mdelem_storage = static_cast<grpc_linked_mdelem*>(
args.call_state->Alloc(sizeof(grpc_linked_mdelem)));
GPR_ASSERT(grpc_metadata_batch_add_tail( GPR_ASSERT(grpc_metadata_batch_add_tail(
pick->initial_metadata, &pick->lb_token_mdelem_storage, args.initial_metadata, mdelem_storage,
GRPC_MDELEM_REF(lb_token)) == GRPC_ERROR_NONE); GRPC_MDELEM_REF(lb_token)) == GRPC_ERROR_NONE);
GrpcLbClientStats* client_stats = static_cast<GrpcLbClientStats*>( GrpcLbClientStats* client_stats = static_cast<GrpcLbClientStats*>(
grpc_mdelem_get_user_data(lb_token, GrpcLbClientStats::Destroy)); grpc_mdelem_get_user_data(lb_token, GrpcLbClientStats::Destroy));
@ -1800,15 +1803,15 @@ class GrpcLbFactory : public LoadBalancingPolicyFactory {
const char* name() const override { return kGrpclb; } const char* name() const override { return kGrpclb; }
RefCountedPtr<ParsedLoadBalancingConfig> ParseLoadBalancingConfig( RefCountedPtr<LoadBalancingPolicy::Config> ParseLoadBalancingConfig(
const grpc_json* json, grpc_error** error) const override { const grpc_json* json, grpc_error** error) const override {
GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE); GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE);
if (json == nullptr) { if (json == nullptr) {
return RefCountedPtr<ParsedLoadBalancingConfig>( return RefCountedPtr<LoadBalancingPolicy::Config>(
New<ParsedGrpcLbConfig>(nullptr)); New<ParsedGrpcLbConfig>(nullptr));
} }
InlinedVector<grpc_error*, 2> error_list; InlinedVector<grpc_error*, 2> error_list;
RefCountedPtr<ParsedLoadBalancingConfig> child_policy; RefCountedPtr<LoadBalancingPolicy::Config> child_policy;
for (const grpc_json* field = json->child; field != nullptr; for (const grpc_json* field = json->child; field != nullptr;
field = field->next) { field = field->next) {
if (field->key == nullptr) continue; if (field->key == nullptr) continue;
@ -1826,7 +1829,7 @@ class GrpcLbFactory : public LoadBalancingPolicyFactory {
} }
} }
if (error_list.empty()) { if (error_list.empty()) {
return RefCountedPtr<ParsedLoadBalancingConfig>( return RefCountedPtr<LoadBalancingPolicy::Config>(
New<ParsedGrpcLbConfig>(std::move(child_policy))); New<ParsedGrpcLbConfig>(std::move(child_policy)));
} else { } else {
*error = GRPC_ERROR_CREATE_FROM_VECTOR("GrpcLb Parser", &error_list); *error = GRPC_ERROR_CREATE_FROM_VECTOR("GrpcLb Parser", &error_list);

14
src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc
Unescape View File

@ -115,9 +115,11 @@ class PickFirst : public LoadBalancingPolicy {
explicit Picker(RefCountedPtr<ConnectedSubchannel> connected_subchannel) explicit Picker(RefCountedPtr<ConnectedSubchannel> connected_subchannel)
: connected_subchannel_(std::move(connected_subchannel)) {} : connected_subchannel_(std::move(connected_subchannel)) {}
PickResult Pick(PickArgs* pick, grpc_error** error) override { PickResult Pick(PickArgs args) override {
pick->connected_subchannel = connected_subchannel_; PickResult result;
return PICK_COMPLETE; result.type = PickResult::PICK_COMPLETE;
result.connected_subchannel = connected_subchannel_;
return result;
} }
private: private:
@ -527,7 +529,7 @@ void PickFirst::PickFirstSubchannelData::
} }
} }
class ParsedPickFirstConfig : public ParsedLoadBalancingConfig { class ParsedPickFirstConfig : public LoadBalancingPolicy::Config {
public: public:
const char* name() const override { return kPickFirst; } const char* name() const override { return kPickFirst; }
}; };
@ -545,12 +547,12 @@ class PickFirstFactory : public LoadBalancingPolicyFactory {
const char* name() const override { return kPickFirst; } const char* name() const override { return kPickFirst; }
RefCountedPtr<ParsedLoadBalancingConfig> ParseLoadBalancingConfig( RefCountedPtr<LoadBalancingPolicy::Config> ParseLoadBalancingConfig(
const grpc_json* json, grpc_error** error) const override { const grpc_json* json, grpc_error** error) const override {
if (json != nullptr) { if (json != nullptr) {
GPR_DEBUG_ASSERT(strcmp(json->key, name()) == 0); GPR_DEBUG_ASSERT(strcmp(json->key, name()) == 0);
} }
return RefCountedPtr<ParsedLoadBalancingConfig>( return RefCountedPtr<LoadBalancingPolicy::Config>(
New<ParsedPickFirstConfig>()); New<ParsedPickFirstConfig>());
} }
}; };

17
src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
Unescape View File

@ -149,7 +149,7 @@ class RoundRobin : public LoadBalancingPolicy {
public: public:
Picker(RoundRobin* parent, RoundRobinSubchannelList* subchannel_list); Picker(RoundRobin* parent, RoundRobinSubchannelList* subchannel_list);
PickResult Pick(PickArgs* pick, grpc_error** error) override; PickResult Pick(PickArgs args) override;
private: private:
// Using pointer value only, no ref held -- do not dereference! // Using pointer value only, no ref held -- do not dereference!
@ -220,8 +220,7 @@ RoundRobin::Picker::Picker(RoundRobin* parent,
} }
} }
RoundRobin::PickResult RoundRobin::Picker::Pick(PickArgs* pick, RoundRobin::PickResult RoundRobin::Picker::Pick(PickArgs args) {
grpc_error** error) {
last_picked_index_ = (last_picked_index_ + 1) % subchannels_.size(); last_picked_index_ = (last_picked_index_ + 1) % subchannels_.size();
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_round_robin_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_round_robin_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
@ -230,8 +229,10 @@ RoundRobin::PickResult RoundRobin::Picker::Pick(PickArgs* pick,
parent_, this, last_picked_index_, parent_, this, last_picked_index_,
subchannels_[last_picked_index_].get()); subchannels_[last_picked_index_].get());
} }
pick->connected_subchannel = subchannels_[last_picked_index_]; PickResult result;
return PICK_COMPLETE; result.type = PickResult::PICK_COMPLETE;
result.connected_subchannel = subchannels_[last_picked_index_];
return result;
} }
// //
@ -503,7 +504,7 @@ void RoundRobin::UpdateLocked(UpdateArgs args) {
} }
} }
class ParsedRoundRobinConfig : public ParsedLoadBalancingConfig { class ParsedRoundRobinConfig : public LoadBalancingPolicy::Config {
public: public:
const char* name() const override { return kRoundRobin; } const char* name() const override { return kRoundRobin; }
}; };
@ -521,12 +522,12 @@ class RoundRobinFactory : public LoadBalancingPolicyFactory {
const char* name() const override { return kRoundRobin; } const char* name() const override { return kRoundRobin; }
RefCountedPtr<ParsedLoadBalancingConfig> ParseLoadBalancingConfig( RefCountedPtr<LoadBalancingPolicy::Config> ParseLoadBalancingConfig(
const grpc_json* json, grpc_error** error) const override { const grpc_json* json, grpc_error** error) const override {
if (json != nullptr) { if (json != nullptr) {
GPR_DEBUG_ASSERT(strcmp(json->key, name()) == 0); GPR_DEBUG_ASSERT(strcmp(json->key, name()) == 0);
} }
return RefCountedPtr<ParsedLoadBalancingConfig>( return RefCountedPtr<LoadBalancingPolicy::Config>(
New<ParsedRoundRobinConfig>()); New<ParsedRoundRobinConfig>());
} }
}; };

56
src/core/ext/filters/client_channel/lb_policy/xds/xds.cc
Unescape View File

@ -120,11 +120,11 @@ constexpr char kXds[] = "xds_experimental";
constexpr char kDefaultLocalityName[] = "xds_default_locality"; constexpr char kDefaultLocalityName[] = "xds_default_locality";
constexpr uint32_t kDefaultLocalityWeight = 3; constexpr uint32_t kDefaultLocalityWeight = 3;
class ParsedXdsConfig : public ParsedLoadBalancingConfig { class ParsedXdsConfig : public LoadBalancingPolicy::Config {
public: public:
ParsedXdsConfig(const char* balancer_name, ParsedXdsConfig(const char* balancer_name,
RefCountedPtr<ParsedLoadBalancingConfig> child_policy, RefCountedPtr<LoadBalancingPolicy::Config> child_policy,
RefCountedPtr<ParsedLoadBalancingConfig> fallback_policy) RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy)
: balancer_name_(balancer_name), : balancer_name_(balancer_name),
child_policy_(std::move(child_policy)), child_policy_(std::move(child_policy)),
fallback_policy_(std::move(fallback_policy)) {} fallback_policy_(std::move(fallback_policy)) {}
@ -133,18 +133,18 @@ class ParsedXdsConfig : public ParsedLoadBalancingConfig {
const char* balancer_name() const { return balancer_name_; }; const char* balancer_name() const { return balancer_name_; };
RefCountedPtr<ParsedLoadBalancingConfig> child_policy() const { RefCountedPtr<LoadBalancingPolicy::Config> child_policy() const {
return child_policy_; return child_policy_;
} }
RefCountedPtr<ParsedLoadBalancingConfig> fallback_policy() const { RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy() const {
return fallback_policy_; return fallback_policy_;
} }
private: private:
const char* balancer_name_ = nullptr; const char* balancer_name_ = nullptr;
RefCountedPtr<ParsedLoadBalancingConfig> child_policy_; RefCountedPtr<LoadBalancingPolicy::Config> child_policy_;
RefCountedPtr<ParsedLoadBalancingConfig> fallback_policy_; RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy_;
}; };
class XdsLb : public LoadBalancingPolicy { class XdsLb : public LoadBalancingPolicy {
@ -300,9 +300,7 @@ class XdsLb : public LoadBalancingPolicy {
public: public:
explicit PickerRef(UniquePtr<SubchannelPicker> picker) explicit PickerRef(UniquePtr<SubchannelPicker> picker)
: picker_(std::move(picker)) {} : picker_(std::move(picker)) {}
PickResult Pick(PickArgs* pick, grpc_error** error) { PickResult Pick(PickArgs args) { return picker_->Pick(args); }
return picker_->Pick(pick, error);
}
private: private:
UniquePtr<SubchannelPicker> picker_; UniquePtr<SubchannelPicker> picker_;
@ -322,12 +320,11 @@ class XdsLb : public LoadBalancingPolicy {
: client_stats_(std::move(client_stats)), : client_stats_(std::move(client_stats)),
pickers_(std::move(pickers)) {} pickers_(std::move(pickers)) {}
PickResult Pick(PickArgs* pick, grpc_error** error) override; PickResult Pick(PickArgs args) override;
private: private:
// Calls the picker of the locality that the key falls within // Calls the picker of the locality that the key falls within
PickResult PickFromLocality(const uint32_t key, PickArgs* pick, PickResult PickFromLocality(const uint32_t key, PickArgs args);
grpc_error** error);
RefCountedPtr<XdsLbClientStats> client_stats_; RefCountedPtr<XdsLbClientStats> client_stats_;
PickerList pickers_; PickerList pickers_;
}; };
@ -363,7 +360,7 @@ class XdsLb : public LoadBalancingPolicy {
~LocalityEntry() = default; ~LocalityEntry() = default;
void UpdateLocked(xds_grpclb_serverlist* serverlist, void UpdateLocked(xds_grpclb_serverlist* serverlist,
ParsedLoadBalancingConfig* child_policy_config, LoadBalancingPolicy::Config* child_policy_config,
const grpc_channel_args* args); const grpc_channel_args* args);
void ShutdownLocked(); void ShutdownLocked();
void ResetBackoffLocked(); void ResetBackoffLocked();
@ -410,7 +407,7 @@ class XdsLb : public LoadBalancingPolicy {
}; };
void UpdateLocked(const LocalityList& locality_list, void UpdateLocked(const LocalityList& locality_list,
ParsedLoadBalancingConfig* child_policy_config, LoadBalancingPolicy::Config* child_policy_config,
const grpc_channel_args* args, XdsLb* parent); const grpc_channel_args* args, XdsLb* parent);
void ShutdownLocked(); void ShutdownLocked();
void ResetBackoffLocked(); void ResetBackoffLocked();
@ -506,7 +503,7 @@ class XdsLb : public LoadBalancingPolicy {
grpc_closure lb_on_fallback_; grpc_closure lb_on_fallback_;
// The policy to use for the fallback backends. // The policy to use for the fallback backends.
RefCountedPtr<ParsedLoadBalancingConfig> fallback_policy_config_; RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy_config_;
// Lock held when modifying the value of fallback_policy_ or // Lock held when modifying the value of fallback_policy_ or
// pending_fallback_policy_. // pending_fallback_policy_.
Mutex fallback_policy_mu_; Mutex fallback_policy_mu_;
@ -515,7 +512,7 @@ class XdsLb : public LoadBalancingPolicy {
OrphanablePtr<LoadBalancingPolicy> pending_fallback_policy_; OrphanablePtr<LoadBalancingPolicy> pending_fallback_policy_;
// The policy to use for the backends. // The policy to use for the backends.
RefCountedPtr<ParsedLoadBalancingConfig> child_policy_config_; RefCountedPtr<LoadBalancingPolicy::Config> child_policy_config_;
// Map of policies to use in the backend // Map of policies to use in the backend
LocalityMap locality_map_; LocalityMap locality_map_;
// TODO(mhaidry) : Add support for multiple maps of localities // TODO(mhaidry) : Add support for multiple maps of localities
@ -530,25 +527,24 @@ class XdsLb : public LoadBalancingPolicy {
// XdsLb::Picker // XdsLb::Picker
// //
XdsLb::PickResult XdsLb::Picker::Pick(PickArgs* pick, grpc_error** error) { XdsLb::PickResult XdsLb::Picker::Pick(PickArgs args) {
// TODO(roth): Add support for drop handling. // TODO(roth): Add support for drop handling.
// Generate a random number between 0 and the total weight // Generate a random number between 0 and the total weight
const uint32_t key = const uint32_t key =
(rand() * pickers_[pickers_.size() - 1].first) / RAND_MAX; (rand() * pickers_[pickers_.size() - 1].first) / RAND_MAX;
// Forward pick to whichever locality maps to the range in which the // Forward pick to whichever locality maps to the range in which the
// random number falls in. // random number falls in.
PickResult result = PickFromLocality(key, pick, error); PickResult result = PickFromLocality(key, args);
// If pick succeeded, add client stats. // If pick succeeded, add client stats.
if (result == PickResult::PICK_COMPLETE && if (result.type == PickResult::PICK_COMPLETE &&
pick->connected_subchannel != nullptr && client_stats_ != nullptr) { result.connected_subchannel != nullptr && client_stats_ != nullptr) {
// TODO(roth): Add support for client stats. // TODO(roth): Add support for client stats.
} }
return result; return result;
} }
XdsLb::PickResult XdsLb::Picker::PickFromLocality(const uint32_t key, XdsLb::PickResult XdsLb::Picker::PickFromLocality(const uint32_t key,
PickArgs* pick, PickArgs args) {
grpc_error** error) {
size_t mid = 0; size_t mid = 0;
size_t start_index = 0; size_t start_index = 0;
size_t end_index = pickers_.size() - 1; size_t end_index = pickers_.size() - 1;
@ -566,7 +562,7 @@ XdsLb::PickResult XdsLb::Picker::PickFromLocality(const uint32_t key,
} }
if (index == 0) index = start_index; if (index == 0) index = start_index;
GPR_ASSERT(pickers_[index].first > key); GPR_ASSERT(pickers_[index].first > key);
return pickers_[index].second->Pick(pick, error); return pickers_[index].second->Pick(args);
} }
// //
@ -1744,7 +1740,7 @@ void XdsLb::LocalityMap::PruneLocalities(const LocalityList& locality_list) {
void XdsLb::LocalityMap::UpdateLocked( void XdsLb::LocalityMap::UpdateLocked(
const LocalityList& locality_serverlist, const LocalityList& locality_serverlist,
ParsedLoadBalancingConfig* child_policy_config, LoadBalancingPolicy::Config* child_policy_config,
const grpc_channel_args* args, XdsLb* parent) { const grpc_channel_args* args, XdsLb* parent) {
if (parent->shutting_down_) return; if (parent->shutting_down_) return;
for (size_t i = 0; i < locality_serverlist.size(); i++) { for (size_t i = 0; i < locality_serverlist.size(); i++) {
@ -1839,7 +1835,7 @@ XdsLb::LocalityMap::LocalityEntry::CreateChildPolicyLocked(
void XdsLb::LocalityMap::LocalityEntry::UpdateLocked( void XdsLb::LocalityMap::LocalityEntry::UpdateLocked(
xds_grpclb_serverlist* serverlist, xds_grpclb_serverlist* serverlist,
ParsedLoadBalancingConfig* child_policy_config, LoadBalancingPolicy::Config* child_policy_config,
const grpc_channel_args* args_in) { const grpc_channel_args* args_in) {
if (parent_->shutting_down_) return; if (parent_->shutting_down_) return;
// Construct update args. // Construct update args.
@ -2158,7 +2154,7 @@ class XdsFactory : public LoadBalancingPolicyFactory {
const char* name() const override { return kXds; } const char* name() const override { return kXds; }
RefCountedPtr<ParsedLoadBalancingConfig> ParseLoadBalancingConfig( RefCountedPtr<LoadBalancingPolicy::Config> ParseLoadBalancingConfig(
const grpc_json* json, grpc_error** error) const override { const grpc_json* json, grpc_error** error) const override {
GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE); GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE);
if (json == nullptr) { if (json == nullptr) {
@ -2174,8 +2170,8 @@ class XdsFactory : public LoadBalancingPolicyFactory {
InlinedVector<grpc_error*, 3> error_list; InlinedVector<grpc_error*, 3> error_list;
const char* balancer_name = nullptr; const char* balancer_name = nullptr;
RefCountedPtr<ParsedLoadBalancingConfig> child_policy; RefCountedPtr<LoadBalancingPolicy::Config> child_policy;
RefCountedPtr<ParsedLoadBalancingConfig> fallback_policy; RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy;
for (const grpc_json* field = json->child; field != nullptr; for (const grpc_json* field = json->child; field != nullptr;
field = field->next) { field = field->next) {
if (field->key == nullptr) continue; if (field->key == nullptr) continue;
@ -2221,7 +2217,7 @@ class XdsFactory : public LoadBalancingPolicyFactory {
"field:balancerName error:not found")); "field:balancerName error:not found"));
} }
if (error_list.empty()) { if (error_list.empty()) {
return RefCountedPtr<ParsedLoadBalancingConfig>(New<ParsedXdsConfig>( return RefCountedPtr<LoadBalancingPolicy::Config>(New<ParsedXdsConfig>(
balancer_name, std::move(child_policy), std::move(fallback_policy))); balancer_name, std::move(child_policy), std::move(fallback_policy)));
} else { } else {
*error = GRPC_ERROR_CREATE_FROM_VECTOR("Xds Parser", &error_list); *error = GRPC_ERROR_CREATE_FROM_VECTOR("Xds Parser", &error_list);

2
src/core/ext/filters/client_channel/lb_policy_factory.h
Unescape View File

@ -37,7 +37,7 @@ class LoadBalancingPolicyFactory {
/// Caller does NOT take ownership of result. /// Caller does NOT take ownership of result.
virtual const char* name() const GRPC_ABSTRACT; virtual const char* name() const GRPC_ABSTRACT;
virtual RefCountedPtr<ParsedLoadBalancingConfig> ParseLoadBalancingConfig( virtual RefCountedPtr<LoadBalancingPolicy::Config> ParseLoadBalancingConfig(
const grpc_json* json, grpc_error** error) const GRPC_ABSTRACT; const grpc_json* json, grpc_error** error) const GRPC_ABSTRACT;
virtual ~LoadBalancingPolicyFactory() {} virtual ~LoadBalancingPolicyFactory() {}

2
src/core/ext/filters/client_channel/lb_policy_registry.cc
Unescape View File

@ -176,7 +176,7 @@ grpc_json* ParseLoadBalancingConfigHelper(const grpc_json* lb_config_array,
} }
} // namespace } // namespace
RefCountedPtr<ParsedLoadBalancingConfig> RefCountedPtr<LoadBalancingPolicy::Config>
LoadBalancingPolicyRegistry::ParseLoadBalancingConfig(const grpc_json* json, LoadBalancingPolicyRegistry::ParseLoadBalancingConfig(const grpc_json* json,
grpc_error** error) { grpc_error** error) {
GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE); GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE);

2
src/core/ext/filters/client_channel/lb_policy_registry.h
Unescape View File

@ -56,7 +56,7 @@ class LoadBalancingPolicyRegistry {
/// Returns a parsed object of the load balancing policy to be used from a /// Returns a parsed object of the load balancing policy to be used from a
/// LoadBalancingConfig array \a json. /// LoadBalancingConfig array \a json.
static RefCountedPtr<ParsedLoadBalancingConfig> ParseLoadBalancingConfig( static RefCountedPtr<LoadBalancingPolicy::Config> ParseLoadBalancingConfig(
const grpc_json* json, grpc_error** error); const grpc_json* json, grpc_error** error);
}; };

2
src/core/ext/filters/client_channel/resolver_result_parsing.cc
Unescape View File

@ -268,7 +268,7 @@ ClientChannelServiceConfigParser::ParseGlobalParams(const grpc_json* json,
grpc_error** error) { grpc_error** error) {
GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE); GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE);
InlinedVector<grpc_error*, 4> error_list; InlinedVector<grpc_error*, 4> error_list;
RefCountedPtr<ParsedLoadBalancingConfig> parsed_lb_config; RefCountedPtr<LoadBalancingPolicy::Config> parsed_lb_config;
UniquePtr<char> lb_policy_name; UniquePtr<char> lb_policy_name;
Optional<ClientChannelGlobalParsedConfig::RetryThrottling> retry_throttling; Optional<ClientChannelGlobalParsedConfig::RetryThrottling> retry_throttling;
const char* health_check_service_name = nullptr; const char* health_check_service_name = nullptr;

6
src/core/ext/filters/client_channel/resolver_result_parsing.h
Unescape View File

@ -45,7 +45,7 @@ class ClientChannelGlobalParsedConfig : public ServiceConfig::ParsedConfig {
}; };
ClientChannelGlobalParsedConfig( ClientChannelGlobalParsedConfig(
RefCountedPtr<ParsedLoadBalancingConfig> parsed_lb_config, RefCountedPtr<LoadBalancingPolicy::Config> parsed_lb_config,
UniquePtr<char> parsed_deprecated_lb_policy, UniquePtr<char> parsed_deprecated_lb_policy,
const Optional<RetryThrottling>& retry_throttling, const Optional<RetryThrottling>& retry_throttling,
const char* health_check_service_name) const char* health_check_service_name)
@ -58,7 +58,7 @@ class ClientChannelGlobalParsedConfig : public ServiceConfig::ParsedConfig {
return retry_throttling_; return retry_throttling_;
} }
RefCountedPtr<ParsedLoadBalancingConfig> parsed_lb_config() const { RefCountedPtr<LoadBalancingPolicy::Config> parsed_lb_config() const {
return parsed_lb_config_; return parsed_lb_config_;
} }
@ -71,7 +71,7 @@ class ClientChannelGlobalParsedConfig : public ServiceConfig::ParsedConfig {
} }
private: private:
RefCountedPtr<ParsedLoadBalancingConfig> parsed_lb_config_; RefCountedPtr<LoadBalancingPolicy::Config> parsed_lb_config_;
UniquePtr<char> parsed_deprecated_lb_policy_; UniquePtr<char> parsed_deprecated_lb_policy_;
Optional<RetryThrottling> retry_throttling_; Optional<RetryThrottling> retry_throttling_;
const char* health_check_service_name_; const char* health_check_service_name_;

6
src/core/ext/filters/client_channel/resolving_lb_policy.cc
Unescape View File

@ -184,7 +184,7 @@ class ResolvingLoadBalancingPolicy::ResolvingControlHelper
ResolvingLoadBalancingPolicy::ResolvingLoadBalancingPolicy( ResolvingLoadBalancingPolicy::ResolvingLoadBalancingPolicy(
Args args, TraceFlag* tracer, UniquePtr<char> target_uri, Args args, TraceFlag* tracer, UniquePtr<char> target_uri,
UniquePtr<char> child_policy_name, UniquePtr<char> child_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig> child_lb_config, RefCountedPtr<LoadBalancingPolicy::Config> child_lb_config,
grpc_error** error) grpc_error** error)
: LoadBalancingPolicy(std::move(args)), : LoadBalancingPolicy(std::move(args)),
tracer_(tracer), tracer_(tracer),
@ -333,7 +333,7 @@ void ResolvingLoadBalancingPolicy::OnResolverError(grpc_error* error) {
void ResolvingLoadBalancingPolicy::CreateOrUpdateLbPolicyLocked( void ResolvingLoadBalancingPolicy::CreateOrUpdateLbPolicyLocked(
const char* lb_policy_name, const char* lb_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig> lb_policy_config, RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config,
Resolver::Result result, TraceStringVector* trace_strings) { Resolver::Result result, TraceStringVector* trace_strings) {
// If the child policy name changes, we need to create a new child // If the child policy name changes, we need to create a new child
// policy. When this happens, we leave child_policy_ as-is and store // policy. When this happens, we leave child_policy_ as-is and store
@ -530,7 +530,7 @@ void ResolvingLoadBalancingPolicy::OnResolverResultChangedLocked(
const bool resolution_contains_addresses = result.addresses.size() > 0; const bool resolution_contains_addresses = result.addresses.size() > 0;
// Process the resolver result. // Process the resolver result.
const char* lb_policy_name = nullptr; const char* lb_policy_name = nullptr;
RefCountedPtr<ParsedLoadBalancingConfig> lb_policy_config; RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config;
bool service_config_changed = false; bool service_config_changed = false;
char* service_config_error_string = nullptr; char* service_config_error_string = nullptr;
if (process_resolver_result_ != nullptr) { if (process_resolver_result_ != nullptr) {

8
src/core/ext/filters/client_channel/resolving_lb_policy.h
Unescape View File

@ -57,7 +57,7 @@ class ResolvingLoadBalancingPolicy : public LoadBalancingPolicy {
ResolvingLoadBalancingPolicy( ResolvingLoadBalancingPolicy(
Args args, TraceFlag* tracer, UniquePtr<char> target_uri, Args args, TraceFlag* tracer, UniquePtr<char> target_uri,
UniquePtr<char> child_policy_name, UniquePtr<char> child_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig> child_lb_config, RefCountedPtr<LoadBalancingPolicy::Config> child_lb_config,
grpc_error** error); grpc_error** error);
// Private ctor, to be used by client_channel only! // Private ctor, to be used by client_channel only!
@ -70,7 +70,7 @@ class ResolvingLoadBalancingPolicy : public LoadBalancingPolicy {
// should set the channel to be in TRANSIENT_FAILURE. // should set the channel to be in TRANSIENT_FAILURE.
typedef bool (*ProcessResolverResultCallback)( typedef bool (*ProcessResolverResultCallback)(
void* user_data, Resolver::Result* result, const char** lb_policy_name, void* user_data, Resolver::Result* result, const char** lb_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig>* lb_policy_config, RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config,
grpc_error** service_config_error); grpc_error** service_config_error);
// If error is set when this returns, then construction failed, and // If error is set when this returns, then construction failed, and
// the caller may not use the new object. // the caller may not use the new object.
@ -109,7 +109,7 @@ class ResolvingLoadBalancingPolicy : public LoadBalancingPolicy {
void OnResolverError(grpc_error* error); void OnResolverError(grpc_error* error);
void CreateOrUpdateLbPolicyLocked( void CreateOrUpdateLbPolicyLocked(
const char* lb_policy_name, const char* lb_policy_name,
RefCountedPtr<ParsedLoadBalancingConfig> lb_policy_config, RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config,
Resolver::Result result, TraceStringVector* trace_strings); Resolver::Result result, TraceStringVector* trace_strings);
OrphanablePtr<LoadBalancingPolicy> CreateLbPolicyLocked( OrphanablePtr<LoadBalancingPolicy> CreateLbPolicyLocked(
const char* lb_policy_name, const grpc_channel_args& args, const char* lb_policy_name, const grpc_channel_args& args,
@ -126,7 +126,7 @@ class ResolvingLoadBalancingPolicy : public LoadBalancingPolicy {
ProcessResolverResultCallback process_resolver_result_ = nullptr; ProcessResolverResultCallback process_resolver_result_ = nullptr;
void* process_resolver_result_user_data_ = nullptr; void* process_resolver_result_user_data_ = nullptr;
UniquePtr<char> child_policy_name_; UniquePtr<char> child_policy_name_;
RefCountedPtr<ParsedLoadBalancingConfig> child_lb_config_; RefCountedPtr<LoadBalancingPolicy::Config> child_lb_config_;
// Resolver and associated state. // Resolver and associated state.
OrphanablePtr<Resolver> resolver_; OrphanablePtr<Resolver> resolver_;

34
test/core/util/test_lb_policies.cc
Unescape View File

@ -120,10 +120,12 @@ class InterceptRecvTrailingMetadataLoadBalancingPolicy
cb_(cb), cb_(cb),
user_data_(user_data) {} user_data_(user_data) {}
PickResult Pick(PickArgs* pick, grpc_error** error) override { PickResult Pick(PickArgs args) override {
PickResult result = delegate_picker_->Pick(pick, error); PickResult result = delegate_picker_->Pick(args);
if (result == PICK_COMPLETE && pick->connected_subchannel != nullptr) { if (result.type == PickResult::PICK_COMPLETE &&
New<TrailingMetadataHandler>(pick, cb_, user_data_); // deletes itself result.connected_subchannel != nullptr) {
new (args.call_state->Alloc(sizeof(TrailingMetadataHandler)))
TrailingMetadataHandler(&result, cb_, user_data_);
} }
return result; return result;
} }
@ -169,35 +171,27 @@ class InterceptRecvTrailingMetadataLoadBalancingPolicy
class TrailingMetadataHandler { class TrailingMetadataHandler {
public: public:
TrailingMetadataHandler(PickArgs* pick, TrailingMetadataHandler(PickResult* result,
InterceptRecvTrailingMetadataCallback cb, InterceptRecvTrailingMetadataCallback cb,
void* user_data) void* user_data)
: cb_(cb), user_data_(user_data) { : cb_(cb), user_data_(user_data) {
GRPC_CLOSURE_INIT(&recv_trailing_metadata_ready_, result->recv_trailing_metadata_ready = &RecordRecvTrailingMetadata;
RecordRecvTrailingMetadata, this, result->recv_trailing_metadata_ready_user_data = this;
grpc_schedule_on_exec_ctx);
pick->recv_trailing_metadata_ready = &recv_trailing_metadata_ready_;
pick->original_recv_trailing_metadata_ready =
&original_recv_trailing_metadata_ready_;
pick->recv_trailing_metadata = &recv_trailing_metadata_;
} }
private: private:
static void RecordRecvTrailingMetadata(void* arg, grpc_error* err) { static void RecordRecvTrailingMetadata(
void* arg, grpc_metadata_batch* recv_trailing_metadata,
CallState* call_state) {
TrailingMetadataHandler* self = TrailingMetadataHandler* self =
static_cast<TrailingMetadataHandler*>(arg); static_cast<TrailingMetadataHandler*>(arg);
GPR_ASSERT(self->recv_trailing_metadata_ != nullptr); GPR_ASSERT(recv_trailing_metadata != nullptr);
self->cb_(self->user_data_); self->cb_(self->user_data_);
GRPC_CLOSURE_SCHED(self->original_recv_trailing_metadata_ready_, self->~TrailingMetadataHandler();
GRPC_ERROR_REF(err));
Delete(self);
} }
InterceptRecvTrailingMetadataCallback cb_; InterceptRecvTrailingMetadataCallback cb_;
void* user_data_; void* user_data_;
grpc_closure recv_trailing_metadata_ready_;
grpc_closure* original_recv_trailing_metadata_ready_ = nullptr;
grpc_metadata_batch* recv_trailing_metadata_ = nullptr;
}; };
}; };

Write Preview
Loading…
Cancel
Save