Skip to content

Commit 7767fbe

Browse files
markdrothmarkdroth
committed
Hide ConnectedSubchannel from LB policy API.
1 parent bf46c11 commit 7767fbe

File tree

14 files changed

+514
-298
lines changed

14 files changed

+514
-298
lines changed

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

Lines changed: 319 additions & 101 deletions
Large diffs are not rendered by default.

‎src/core/ext/filters/client_channel/lb_policy.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> {
128128

129129
/// Used only if type is PICK_COMPLETE. Will be set to the selected
130130
/// subchannel, or nullptr if the LB policy decides to drop the call.
131-
RefCountedPtr<ConnectedSubchannelInterface> connected_subchannel;
131+
RefCountedPtr<SubchannelInterface> subchannel;
132132

133133
/// Used only if type is PICK_TRANSIENT_FAILURE.
134134
/// Error to be set when returning a transient failure.

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

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -575,13 +575,12 @@ GrpcLb::PickResult GrpcLb::Picker::Pick(PickArgs args) {
575575
result = child_picker_->Pick(args);
576576
// If pick succeeded, add LB token to initial metadata.
577577
if (result.type == PickResult::PICK_COMPLETE &&
578-
result.connected_subchannel != nullptr) {
578+
result.subchannel != nullptr) {
579579
const grpc_arg* arg = grpc_channel_args_find(
580-
result.connected_subchannel->args(), GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN);
580+
result.subchannel->channel_args(), GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN);
581581
if (arg == nullptr) {
582-
gpr_log(GPR_ERROR,
583-
"[grpclb %p picker %p] No LB token for connected subchannel %p",
584-
parent_, this, result.connected_subchannel.get());
582+
gpr_log(GPR_ERROR, "[grpclb %p picker %p] No LB token for subchannel %p",
583+
parent_, this, result.subchannel.get());
585584
abort();
586585
}
587586
grpc_mdelem lb_token = {reinterpret_cast<uintptr_t>(arg->value.pointer.p)};

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

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,6 @@
2828
#include "src/core/ext/filters/client_channel/subchannel.h"
2929
#include "src/core/lib/channel/channel_args.h"
3030
#include "src/core/lib/gprpp/sync.h"
31-
#include "src/core/lib/iomgr/combiner.h"
3231
#include "src/core/lib/iomgr/sockaddr_utils.h"
3332
#include "src/core/lib/transport/connectivity_state.h"
3433

@@ -85,9 +84,8 @@ class PickFirst : public LoadBalancingPolicy {
8584
public:
8685
PickFirstSubchannelList(PickFirst* policy, TraceFlag* tracer,
8786
const ServerAddressList& addresses,
88-
grpc_combiner* combiner,
8987
const grpc_channel_args& args)
90-
: SubchannelList(policy, tracer, addresses, combiner,
88+
: SubchannelList(policy, tracer, addresses,
9189
policy->channel_control_helper(), args) {
9290
// Need to maintain a ref to the LB policy as long as we maintain
9391
// any references to subchannels, since the subchannels'
@@ -111,19 +109,18 @@ class PickFirst : public LoadBalancingPolicy {
111109

112110
class Picker : public SubchannelPicker {
113111
public:
114-
explicit Picker(
115-
RefCountedPtr<ConnectedSubchannelInterface> connected_subchannel)
116-
: connected_subchannel_(std::move(connected_subchannel)) {}
112+
explicit Picker(RefCountedPtr<SubchannelInterface> subchannel)
113+
: subchannel_(std::move(subchannel)) {}
117114

118115
PickResult Pick(PickArgs args) override {
119116
PickResult result;
120117
result.type = PickResult::PICK_COMPLETE;
121-
result.connected_subchannel = connected_subchannel_;
118+
result.subchannel = subchannel_;
122119
return result;
123120
}
124121

125122
private:
126-
RefCountedPtr<ConnectedSubchannelInterface> connected_subchannel_;
123+
RefCountedPtr<SubchannelInterface> subchannel_;
127124
};
128125

129126
void ShutdownLocked() override;
@@ -166,6 +163,9 @@ void PickFirst::ShutdownLocked() {
166163
void PickFirst::ExitIdleLocked() {
167164
if (shutdown_) return;
168165
if (idle_) {
166+
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_pick_first_trace)) {
167+
gpr_log(GPR_INFO, "Pick First %p exiting idle", this);
168+
}
169169
idle_ = false;
170170
if (subchannel_list_ == nullptr ||
171171
subchannel_list_->num_subchannels() == 0) {
@@ -200,7 +200,7 @@ void PickFirst::UpdateLocked(UpdateArgs args) {
200200
grpc_channel_args* new_args =
201201
grpc_channel_args_copy_and_add(args.args, &new_arg, 1);
202202
auto subchannel_list = MakeOrphanable<PickFirstSubchannelList>(
203-
this, &grpc_lb_pick_first_trace, args.addresses, combiner(), *new_args);
203+
this, &grpc_lb_pick_first_trace, args.addresses, *new_args);
204204
grpc_channel_args_destroy(new_args);
205205
if (subchannel_list->num_subchannels() == 0) {
206206
// Empty update or no valid subchannels. Unsubscribe from all current
@@ -350,8 +350,8 @@ void PickFirst::PickFirstSubchannelData::ProcessConnectivityChangeLocked(
350350
// some connectivity state notifications.
351351
if (connectivity_state == GRPC_CHANNEL_READY) {
352352
p->channel_control_helper()->UpdateState(
353-
GRPC_CHANNEL_READY, UniquePtr<SubchannelPicker>(New<Picker>(
354-
connected_subchannel()->Ref())));
353+
GRPC_CHANNEL_READY,
354+
UniquePtr<SubchannelPicker>(New<Picker>(subchannel()->Ref())));
355355
} else { // CONNECTING
356356
p->channel_control_helper()->UpdateState(
357357
connectivity_state,
@@ -445,13 +445,13 @@ void PickFirst::PickFirstSubchannelData::ProcessUnselectedReadyLocked() {
445445
p->subchannel_list_ = std::move(p->latest_pending_subchannel_list_);
446446
}
447447
// Cases 1 and 2.
448-
p->selected_ = this;
449-
p->channel_control_helper()->UpdateState(
450-
GRPC_CHANNEL_READY,
451-
UniquePtr<SubchannelPicker>(New<Picker>(connected_subchannel()->Ref())));
452448
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_pick_first_trace)) {
453449
gpr_log(GPR_INFO, "Pick First %p selected subchannel %p", p, subchannel());
454450
}
451+
p->selected_ = this;
452+
p->channel_control_helper()->UpdateState(
453+
GRPC_CHANNEL_READY,
454+
UniquePtr<SubchannelPicker>(New<Picker>(subchannel()->Ref())));
455455
}
456456

457457
void PickFirst::PickFirstSubchannelData::

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

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,6 @@
3838
#include "src/core/lib/debug/trace.h"
3939
#include "src/core/lib/gprpp/ref_counted_ptr.h"
4040
#include "src/core/lib/gprpp/sync.h"
41-
#include "src/core/lib/iomgr/combiner.h"
4241
#include "src/core/lib/iomgr/sockaddr_utils.h"
4342
#include "src/core/lib/transport/connectivity_state.h"
4443
#include "src/core/lib/transport/static_metadata.h"
@@ -106,9 +105,8 @@ class RoundRobin : public LoadBalancingPolicy {
106105
public:
107106
RoundRobinSubchannelList(RoundRobin* policy, TraceFlag* tracer,
108107
const ServerAddressList& addresses,
109-
grpc_combiner* combiner,
110108
const grpc_channel_args& args)
111-
: SubchannelList(policy, tracer, addresses, combiner,
109+
: SubchannelList(policy, tracer, addresses,
112110
policy->channel_control_helper(), args) {
113111
// Need to maintain a ref to the LB policy as long as we maintain
114112
// any references to subchannels, since the subchannels'
@@ -155,7 +153,7 @@ class RoundRobin : public LoadBalancingPolicy {
155153
RoundRobin* parent_;
156154

157155
size_t last_picked_index_;
158-
InlinedVector<RefCountedPtr<ConnectedSubchannelInterface>, 10> subchannels_;
156+
InlinedVector<RefCountedPtr<SubchannelInterface>, 10> subchannels_;
159157
};
160158

161159
void ShutdownLocked() override;
@@ -180,10 +178,9 @@ RoundRobin::Picker::Picker(RoundRobin* parent,
180178
RoundRobinSubchannelList* subchannel_list)
181179
: parent_(parent) {
182180
for (size_t i = 0; i < subchannel_list->num_subchannels(); ++i) {
183-
auto* connected_subchannel =
184-
subchannel_list->subchannel(i)->connected_subchannel();
185-
if (connected_subchannel != nullptr) {
186-
subchannels_.push_back(connected_subchannel->Ref());
181+
RoundRobinSubchannelData* sd = subchannel_list->subchannel(i);
182+
if (sd->connectivity_state() == GRPC_CHANNEL_READY) {
183+
subchannels_.push_back(sd->subchannel()->Ref());
187184
}
188185
}
189186
// For discussion on why we generate a random starting index for
@@ -204,14 +201,13 @@ RoundRobin::PickResult RoundRobin::Picker::Pick(PickArgs args) {
204201
last_picked_index_ = (last_picked_index_ + 1) % subchannels_.size();
205202
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_round_robin_trace)) {
206203
gpr_log(GPR_INFO,
207-
"[RR %p picker %p] returning index %" PRIuPTR
208-
", connected_subchannel=%p",
204+
"[RR %p picker %p] returning index %" PRIuPTR ", subchannel=%p",
209205
parent_, this, last_picked_index_,
210206
subchannels_[last_picked_index_].get());
211207
}
212208
PickResult result;
213209
result.type = PickResult::PICK_COMPLETE;
214-
result.connected_subchannel = subchannels_[last_picked_index_];
210+
result.subchannel = subchannels_[last_picked_index_];
215211
return result;
216212
}
217213

@@ -424,7 +420,7 @@ void RoundRobin::UpdateLocked(UpdateArgs args) {
424420
}
425421
}
426422
latest_pending_subchannel_list_ = MakeOrphanable<RoundRobinSubchannelList>(
427-
this, &grpc_lb_round_robin_trace, args.addresses, combiner(), *args.args);
423+
this, &grpc_lb_round_robin_trace, args.addresses, *args.args);
428424
if (latest_pending_subchannel_list_->num_subchannels() == 0) {
429425
// If the new list is empty, immediately promote the new list to the
430426
// current list and transition to TRANSIENT_FAILURE.

0 commit comments

Comments
 (0)