Remove redundant SubchannelPicker refreshes in RoundRobinLoadBalancer #4607
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -24,6 +24,9 @@ | |
| import static io.grpc.ConnectivityState.TRANSIENT_FAILURE; | ||
|
|
||
| import com.google.common.annotations.VisibleForTesting; | ||
| import com.google.common.base.Objects; | ||
| import com.google.common.base.Preconditions; | ||
|
|
||
| import io.grpc.Attributes; | ||
| import io.grpc.ConnectivityState; | ||
| import io.grpc.ConnectivityStateInfo; | ||
|
|
@@ -42,12 +45,10 @@ | |
| import io.grpc.internal.ServiceConfigUtil; | ||
| import java.util.ArrayList; | ||
| import java.util.Collection; | ||
| import java.util.HashMap; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Queue; | ||
| import java.util.Random; | ||
| import java.util.Set; | ||
|
|
@@ -113,6 +114,9 @@ static final class RoundRobinLoadBalancer extends LoadBalancer { | |
| new HashMap<EquivalentAddressGroup, Subchannel>(); | ||
| private final Random random; | ||
|
|
||
| private ConnectivityState currentState; | ||
| private RoundRobinPicker currentPicker = new EmptyPicker(EMPTY_OK); | ||
|
|
||
| @Nullable | ||
| private StickinessState stickinessState; | ||
|
|
||
|
|
@@ -176,50 +180,92 @@ public void handleResolvedAddressGroups( | |
| // Shutdown subchannels for removed addresses. | ||
| for (EquivalentAddressGroup addressGroup : removedAddrs) { | ||
| Subchannel subchannel = subchannels.remove(addressGroup); | ||
| shutdownSubchannel(subchannel); | ||
| } | ||
|
|
||
| updateBalancingState(); | ||
| } | ||
|
|
||
| @Override | ||
| public void handleNameResolutionError(Status error) { | ||
| // ready pickers aren't affected by status changes | ||
| updateBalancingState(TRANSIENT_FAILURE, | ||
| currentPicker instanceof ReadyPicker ? currentPicker : new EmptyPicker(error)); | ||
| } | ||
|
|
||
| @Override | ||
| public void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) { | ||
| if (subchannels.get(subchannel.getAddresses()) != subchannel) { | ||
| return; | ||
| } | ||
| if (stateInfo.getState() == SHUTDOWN && stickinessState != null) { | ||
| stickinessState.remove(subchannel); | ||
| } | ||
| if (stateInfo.getState() == IDLE) { | ||
| subchannel.requestConnection(); | ||
| } | ||
| getSubchannelStateInfoRef(subchannel).value = stateInfo; | ||
| updateBalancingState(); | ||
| } | ||
|
|
||
| private void shutdownSubchannel(Subchannel subchannel) { | ||
| subchannel.shutdown(); | ||
| getSubchannelStateInfoRef(subchannel).value = | ||
| ConnectivityStateInfo.forNonError(SHUTDOWN); | ||
| if (stickinessState != null) { | ||
| stickinessState.remove(subchannel); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void shutdown() { | ||
| for (Subchannel subchannel : getSubchannels()) { | ||
| shutdownSubchannel(subchannel); | ||
| } | ||
| } | ||
|
|
||
| private static final Status EMPTY_OK = Status.OK.withDescription("no subchannels ready"); | ||
|
|
||
| /** | ||
| * Updates picker with the list of active subchannels (state == READY). | ||
| */ | ||
| @SuppressWarnings("ReferenceEquality") | ||
| private void updateBalancingState() { | ||
| List<Subchannel> activeList = filterNonFailingSubchannels(getSubchannels()); | ||
| if (activeList.isEmpty()) { | ||
| // No READY subchannels, determine aggregate state and error status | ||
| boolean isConnecting = false; | ||
| Status aggStatus = EMPTY_OK; | ||
| for (Subchannel subchannel : getSubchannels()) { | ||
| ConnectivityStateInfo stateInfo = getSubchannelStateInfoRef(subchannel).value; | ||
| // This subchannel IDLE is not because of channel IDLE_TIMEOUT, | ||
| // in which case LB is already shutdown. | ||
| // RRLB will request connection immediately on subchannel IDLE. | ||
| if (stateInfo.getState() == CONNECTING || stateInfo.getState() == IDLE) { | ||
| isConnecting = true; | ||
| } | ||
| if (aggStatus == EMPTY_OK || !aggStatus.isOk()) { | ||
| aggStatus = stateInfo.getStatus(); | ||
| } | ||
| } | ||
| updateBalancingState(isConnecting ? CONNECTING : TRANSIENT_FAILURE, | ||
| // If all subchannels are TRANSIENT_FAILURE, return the Status associated with | ||
| // an arbitrary subchannel, otherwise return OK. | ||
| new EmptyPicker(aggStatus)); | ||
| } else { | ||
| // initialize the Picker to a random start index to ensure that a high frequency of Picker | ||
| // churn does not skew subchannel selection. | ||
| int startIndex = random.nextInt(activeList.size()); | ||
| updateBalancingState(READY, new ReadyPicker(activeList, startIndex, stickinessState)); | ||
|
Contributor
There was a problem hiding this comment. You never call
Contributor
Author
There was a problem hiding this comment. I've now combined/collapsed both the |
||
| } | ||
| } | ||
|
|
||
| private void updateBalancingState(ConnectivityState state, RoundRobinPicker picker) { | ||
| if (state != currentState || !picker.isEquivalentTo(currentPicker)) { | ||
| helper.updateBalancingState(state, picker); | ||
| currentState = state; | ||
| currentPicker = picker; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -229,7 +275,7 @@ private static List<Subchannel> filterNonFailingSubchannels( | |
| Collection<Subchannel> subchannels) { | ||
| List<Subchannel> readySubchannels = new ArrayList<Subchannel>(subchannels.size()); | ||
| for (Subchannel subchannel : subchannels) { | ||
| if (isReady(subchannel)) { | ||
| readySubchannels.add(subchannel); | ||
| } | ||
| } | ||
|
|
@@ -248,43 +294,6 @@ private static Set<EquivalentAddressGroup> stripAttrs(List<EquivalentAddressGrou | |
| return addrs; | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| Collection<Subchannel> getSubchannels() { | ||
| return subchannels.values(); | ||
|
|
@@ -294,6 +303,11 @@ private static Ref<ConnectivityStateInfo> getSubchannelStateInfoRef( | |
| Subchannel subchannel) { | ||
| return checkNotNull(subchannel.getAttributes().get(STATE_INFO), "STATE_INFO"); | ||
| } | ||
|
|
||
| // package-private to avoid synthetic access | ||
| static boolean isReady(Subchannel subchannel) { | ||
| return getSubchannelStateInfoRef(subchannel).value.getState() == READY; | ||
| } | ||
|
|
||
| private static <T> Set<T> setsDifference(Set<T> a, Set<T> b) { | ||
| Set<T> aCopy = new HashSet<T>(a); | ||
|
|
@@ -312,7 +326,8 @@ Map<String, Ref<Subchannel>> getStickinessMapForTest() { | |
| * Holds stickiness related states: The stickiness key, a registry mapping stickiness values to | ||
| * the associated Subchannel Ref, and a map from Subchannel to Subchannel Ref. | ||
| */ | ||
| @VisibleForTesting | ||
| static final class StickinessState { | ||
| static final int MAX_ENTRIES = 1000; | ||
|
|
||
| final Key<String> key; | ||
|
|
@@ -332,7 +347,7 @@ private static final class StickinessState { | |
| */ | ||
| @Nonnull | ||
| Subchannel maybeRegister( | ||
| String stickinessValue, @Nonnull Subchannel subchannel) { | ||
| final Ref<Subchannel> newSubchannelRef = subchannel.getAttributes().get(STICKY_REF); | ||
| while (true) { | ||
| Ref<Subchannel> existingSubchannelRef = | ||
|
|
@@ -344,7 +359,7 @@ Subchannel maybeRegister( | |
| } else { | ||
| // existing entry | ||
| Subchannel existingSubchannel = existingSubchannelRef.value; | ||
| if (existingSubchannel != null && isReady(existingSubchannel)) { | ||
| return existingSubchannel; | ||
| } | ||
| } | ||
|
|
@@ -384,59 +399,49 @@ Subchannel getSubchannel(String stickinessValue) { | |
| } | ||
| } | ||
| } | ||
|
|
||
| // Only subclasses are ReadyPicker or EmptyPicker | ||
| private abstract static class RoundRobinPicker extends SubchannelPicker { | ||
| abstract boolean isEquivalentTo(RoundRobinPicker picker); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| static final class ReadyPicker extends RoundRobinPicker { | ||
| private static final AtomicIntegerFieldUpdater<ReadyPicker> indexUpdater = | ||
| AtomicIntegerFieldUpdater.newUpdater(ReadyPicker.class, "index"); | ||
|
|
||
| private final List<Subchannel> list; // non-empty | ||
| @Nullable | ||
| private final RoundRobinLoadBalancer.StickinessState stickinessState; | ||
| @SuppressWarnings("unused") | ||
| private volatile int index; | ||
|
|
||
| ReadyPicker(List<Subchannel> list, int startIndex, | ||
| @Nullable RoundRobinLoadBalancer.StickinessState stickinessState) { | ||
| Preconditions.checkArgument(!list.isEmpty(), "empty list"); | ||
| this.list = list; | ||
| this.stickinessState = stickinessState; | ||
| this.index = startIndex - 1; | ||
| } | ||
|
|
||
| @Override | ||
| public PickResult pickSubchannel(PickSubchannelArgs args) { | ||
| Subchannel subchannel = null; | ||
| if (stickinessState != null) { | ||
| String stickinessValue = args.getHeaders().get(stickinessState.key); | ||
| if (stickinessValue != null) { | ||
| subchannel = stickinessState.getSubchannel(stickinessValue); | ||
| if (subchannel == null || !RoundRobinLoadBalancer.isReady(subchannel)) { | ||
| subchannel = stickinessState.maybeRegister(stickinessValue, nextSubchannel()); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return PickResult.withSubchannel(subchannel != null ? subchannel : nextSubchannel()); | ||
| } | ||
|
|
||
| private Subchannel nextSubchannel() { | ||
| int size = list.size(); | ||
| int i = indexUpdater.incrementAndGet(this); | ||
| if (i >= size) { | ||
| int oldi = i; | ||
|
|
@@ -451,9 +456,37 @@ List<Subchannel> getList() { | |
| return list; | ||
| } | ||
|
|
||
| @Override | ||
| boolean isEquivalentTo(RoundRobinPicker picker) { | ||
| if (!(picker instanceof ReadyPicker)) { | ||
| return false; | ||
|
|
||
| } | ||
| ReadyPicker other = (ReadyPicker) picker; | ||
| // the lists cannot contain duplicate subchannels | ||
| return other == this || (stickinessState == other.stickinessState | ||
| && list.size() == other.list.size() | ||
| && new HashSet<Subchannel>(list).containsAll(other.list)); | ||
| } | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| static final class EmptyPicker extends RoundRobinPicker { | ||
|
|
||
| private final Status status; | ||
|
|
||
| EmptyPicker(@Nonnull Status status) { | ||
| this.status = Preconditions.checkNotNull(status, "status"); | ||
| } | ||
|
|
||
| @Override | ||
| public PickResult pickSubchannel(PickSubchannelArgs args) { | ||
| return status.isOk() ? PickResult.withNoResult() : PickResult.withError(status); | ||
| } | ||
|
|
||
| @Override | ||
| boolean isEquivalentTo(RoundRobinPicker picker) { | ||
| return picker instanceof EmptyPicker && (Objects.equal(status, ((EmptyPicker) picker).status) | ||
| || (status.isOk() && ((EmptyPicker) picker).status.isOk())); | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This branch is not covered.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've extended a unit test to cover this case - name resolver removal of a sticky subchannel.