mirror of https://github.com/grpc/grpc-java.git
xds: Swap RingHashLb to use lazy child, instead of deactivation
This commit is contained in:
parent
8a21afcc9e
commit
61bf21e2a1
|
|
@ -238,7 +238,7 @@ public abstract class MultiChildLoadBalancer extends LoadBalancer {
|
||||||
|
|
||||||
// Raactivate deactivated children
|
// Raactivate deactivated children
|
||||||
for (ChildLbState reusedChild : reusedChildren) {
|
for (ChildLbState reusedChild : reusedChildren) {
|
||||||
reusedChild.reactivate(reusedChild.getPolicyProvider());
|
reusedChild.reactivate(reusedChild.getPolicyFactory());
|
||||||
}
|
}
|
||||||
|
|
||||||
updateChildrenWithResolvedAddresses(resolvedAddresses, newChildren);
|
updateChildrenWithResolvedAddresses(resolvedAddresses, newChildren);
|
||||||
|
|
@ -388,20 +388,20 @@ public abstract class MultiChildLoadBalancer extends LoadBalancer {
|
||||||
private final Object config;
|
private final Object config;
|
||||||
|
|
||||||
private final GracefulSwitchLoadBalancer lb;
|
private final GracefulSwitchLoadBalancer lb;
|
||||||
private final LoadBalancerProvider policyProvider;
|
private final LoadBalancer.Factory policyFactory;
|
||||||
private ConnectivityState currentState;
|
private ConnectivityState currentState;
|
||||||
private SubchannelPicker currentPicker;
|
private SubchannelPicker currentPicker;
|
||||||
private boolean deactivated;
|
private boolean deactivated;
|
||||||
|
|
||||||
public ChildLbState(Object key, LoadBalancerProvider policyProvider, Object childConfig,
|
public ChildLbState(Object key, LoadBalancer.Factory policyFactory, Object childConfig,
|
||||||
SubchannelPicker initialPicker) {
|
SubchannelPicker initialPicker) {
|
||||||
this(key, policyProvider, childConfig, initialPicker, null, false);
|
this(key, policyFactory, childConfig, initialPicker, null, false);
|
||||||
}
|
}
|
||||||
|
|
||||||
public ChildLbState(Object key, LoadBalancerProvider policyProvider, Object childConfig,
|
public ChildLbState(Object key, LoadBalancer.Factory policyFactory, Object childConfig,
|
||||||
SubchannelPicker initialPicker, ResolvedAddresses resolvedAddrs, boolean deactivated) {
|
SubchannelPicker initialPicker, ResolvedAddresses resolvedAddrs, boolean deactivated) {
|
||||||
this.key = key;
|
this.key = key;
|
||||||
this.policyProvider = policyProvider;
|
this.policyFactory = policyFactory;
|
||||||
this.deactivated = deactivated;
|
this.deactivated = deactivated;
|
||||||
this.currentPicker = initialPicker;
|
this.currentPicker = initialPicker;
|
||||||
this.config = childConfig;
|
this.config = childConfig;
|
||||||
|
|
@ -409,7 +409,7 @@ public abstract class MultiChildLoadBalancer extends LoadBalancer {
|
||||||
this.currentState = deactivated ? IDLE : CONNECTING;
|
this.currentState = deactivated ? IDLE : CONNECTING;
|
||||||
this.resolvedAddresses = resolvedAddrs;
|
this.resolvedAddresses = resolvedAddrs;
|
||||||
if (!deactivated) {
|
if (!deactivated) {
|
||||||
lb.switchTo(policyProvider);
|
lb.switchTo(policyFactory);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -442,7 +442,7 @@ public abstract class MultiChildLoadBalancer extends LoadBalancer {
|
||||||
* This base implementation does nothing but reset the flag. If you really want to both
|
* This base implementation does nothing but reset the flag. If you really want to both
|
||||||
* deactivate and reactivate you should override them both.
|
* deactivate and reactivate you should override them both.
|
||||||
*/
|
*/
|
||||||
protected void reactivate(LoadBalancerProvider policyProvider) {
|
protected void reactivate(LoadBalancer.Factory policyFactory) {
|
||||||
deactivated = false;
|
deactivated = false;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -478,8 +478,8 @@ public abstract class MultiChildLoadBalancer extends LoadBalancer {
|
||||||
return currentPicker;
|
return currentPicker;
|
||||||
}
|
}
|
||||||
|
|
||||||
protected final LoadBalancerProvider getPolicyProvider() {
|
protected final LoadBalancer.Factory getPolicyFactory() {
|
||||||
return policyProvider;
|
return policyFactory;
|
||||||
}
|
}
|
||||||
|
|
||||||
protected final Subchannel getSubchannels(PickSubchannelArgs args) {
|
protected final Subchannel getSubchannels(PickSubchannelArgs args) {
|
||||||
|
|
|
||||||
|
|
@ -210,13 +210,13 @@ class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected void reactivate(LoadBalancerProvider policyProvider) {
|
protected void reactivate(Factory policyFactory) {
|
||||||
if (deletionTimer != null && deletionTimer.isPending()) {
|
if (deletionTimer != null && deletionTimer.isPending()) {
|
||||||
deletionTimer.cancel();
|
deletionTimer.cancel();
|
||||||
logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());
|
logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());
|
||||||
}
|
}
|
||||||
|
|
||||||
super.reactivate(policyProvider);
|
super.reactivate(policyFactory);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,129 @@
|
||||||
|
/*
|
||||||
|
* Copyright 2024 The gRPC Authors
|
||||||
|
*
|
||||||
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
* you may not use this file except in compliance with the License.
|
||||||
|
* You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing, software
|
||||||
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
* See the License for the specific language governing permissions and
|
||||||
|
* limitations under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package io.grpc.xds;
|
||||||
|
|
||||||
|
import com.google.common.base.Preconditions;
|
||||||
|
import io.grpc.ConnectivityState;
|
||||||
|
import io.grpc.LoadBalancer;
|
||||||
|
import io.grpc.Status;
|
||||||
|
import io.grpc.util.ForwardingLoadBalancer;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A load balancer that starts in IDLE instead of CONNECTING. Once it starts connecting, it
|
||||||
|
* instantiates its delegate.
|
||||||
|
*/
|
||||||
|
final class LazyLoadBalancer extends ForwardingLoadBalancer {
|
||||||
|
private LoadBalancer delegate;
|
||||||
|
|
||||||
|
public LazyLoadBalancer(Helper helper, LoadBalancer.Factory delegateFactory) {
|
||||||
|
this.delegate = new LazyDelegate(helper, delegateFactory);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected LoadBalancer delegate() {
|
||||||
|
return delegate;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
|
||||||
|
return delegate.acceptResolvedAddresses(resolvedAddresses);
|
||||||
|
}
|
||||||
|
|
||||||
|
private final class LazyDelegate extends LoadBalancer {
|
||||||
|
private final Helper helper;
|
||||||
|
private final LoadBalancer.Factory delegateFactory;
|
||||||
|
private ResolvedAddresses addresses;
|
||||||
|
private Status error;
|
||||||
|
private boolean updatedBalancingState;
|
||||||
|
|
||||||
|
public LazyDelegate(Helper helper, LoadBalancer.Factory delegateFactory) {
|
||||||
|
this.helper = Preconditions.checkNotNull(helper, "helper");
|
||||||
|
this.delegateFactory = Preconditions.checkNotNull(delegateFactory, "delegateFactory");
|
||||||
|
}
|
||||||
|
|
||||||
|
private LoadBalancer activate() {
|
||||||
|
if (delegate != this) {
|
||||||
|
return delegate;
|
||||||
|
}
|
||||||
|
delegate = delegateFactory.newLoadBalancer(helper);
|
||||||
|
if (addresses != null) {
|
||||||
|
delegate.acceptResolvedAddresses(addresses);
|
||||||
|
}
|
||||||
|
if (error != null) {
|
||||||
|
delegate.handleNameResolutionError(error);
|
||||||
|
}
|
||||||
|
return delegate;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
|
||||||
|
this.addresses = resolvedAddresses;
|
||||||
|
this.error = null;
|
||||||
|
initializeBalancingState();
|
||||||
|
return Status.OK;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void handleNameResolutionError(Status error) {
|
||||||
|
// Preserve addresses, because even old addresses may be used by the real policy
|
||||||
|
this.error = error;
|
||||||
|
initializeBalancingState();
|
||||||
|
}
|
||||||
|
|
||||||
|
private void initializeBalancingState() {
|
||||||
|
if (updatedBalancingState) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
helper.updateBalancingState(ConnectivityState.IDLE, new LazyPicker());
|
||||||
|
updatedBalancingState = true;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void requestConnection() {
|
||||||
|
activate().requestConnection();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void shutdown() {
|
||||||
|
}
|
||||||
|
|
||||||
|
private final class LazyPicker extends SubchannelPicker {
|
||||||
|
@Override
|
||||||
|
public PickResult pickSubchannel(PickSubchannelArgs args) {
|
||||||
|
helper.getSynchronizationContext().execute(LazyDelegate.this::activate);
|
||||||
|
return PickResult.withNoResult();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void requestConnection() {
|
||||||
|
helper.getSynchronizationContext().execute(LazyDelegate.this::requestConnection);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public static final class Factory extends LoadBalancer.Factory {
|
||||||
|
private final LoadBalancer.Factory delegate;
|
||||||
|
|
||||||
|
public Factory(LoadBalancer.Factory delegate) {
|
||||||
|
this.delegate = Preconditions.checkNotNull(delegate, "delegate");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override public LoadBalancer newLoadBalancer(Helper helper) {
|
||||||
|
return new LazyLoadBalancer(helper, delegate);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -35,7 +35,6 @@ import io.grpc.ConnectivityState;
|
||||||
import io.grpc.EquivalentAddressGroup;
|
import io.grpc.EquivalentAddressGroup;
|
||||||
import io.grpc.InternalLogId;
|
import io.grpc.InternalLogId;
|
||||||
import io.grpc.LoadBalancer;
|
import io.grpc.LoadBalancer;
|
||||||
import io.grpc.LoadBalancerProvider;
|
|
||||||
import io.grpc.Status;
|
import io.grpc.Status;
|
||||||
import io.grpc.SynchronizationContext;
|
import io.grpc.SynchronizationContext;
|
||||||
import io.grpc.util.MultiChildLoadBalancer;
|
import io.grpc.util.MultiChildLoadBalancer;
|
||||||
|
|
@ -66,6 +65,8 @@ final class RingHashLoadBalancer extends MultiChildLoadBalancer {
|
||||||
+ " config selector always generates a hash.");
|
+ " config selector always generates a hash.");
|
||||||
private static final XxHash64 hashFunc = XxHash64.INSTANCE;
|
private static final XxHash64 hashFunc = XxHash64.INSTANCE;
|
||||||
|
|
||||||
|
private final LoadBalancer.Factory lazyLbFactory =
|
||||||
|
new LazyLoadBalancer.Factory(pickFirstLbProvider);
|
||||||
private final XdsLogger logger;
|
private final XdsLogger logger;
|
||||||
private final SynchronizationContext syncContext;
|
private final SynchronizationContext syncContext;
|
||||||
private List<RingEntry> ring;
|
private List<RingEntry> ring;
|
||||||
|
|
@ -229,8 +230,7 @@ final class RingHashLoadBalancer extends MultiChildLoadBalancer {
|
||||||
@Override
|
@Override
|
||||||
protected ChildLbState createChildLbState(Object key, Object policyConfig,
|
protected ChildLbState createChildLbState(Object key, Object policyConfig,
|
||||||
SubchannelPicker initialPicker, ResolvedAddresses resolvedAddresses) {
|
SubchannelPicker initialPicker, ResolvedAddresses resolvedAddresses) {
|
||||||
return new RingHashChildLbState((Endpoint)key,
|
return new RingHashChildLbState((Endpoint)key);
|
||||||
getChildAddresses(key, resolvedAddresses, null));
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private Status validateAddrList(List<EquivalentAddressGroup> addrList) {
|
private Status validateAddrList(List<EquivalentAddressGroup> addrList) {
|
||||||
|
|
@ -420,11 +420,7 @@ final class RingHashLoadBalancer extends MultiChildLoadBalancer {
|
||||||
|
|
||||||
if (subchannelView.connectivityState == IDLE) {
|
if (subchannelView.connectivityState == IDLE) {
|
||||||
syncContext.execute(() -> {
|
syncContext.execute(() -> {
|
||||||
if (childLbState.isDeactivated()) {
|
childLbState.getLb().requestConnection();
|
||||||
childLbState.activate();
|
|
||||||
} else {
|
|
||||||
childLbState.getLb().requestConnection();
|
|
||||||
}
|
|
||||||
});
|
});
|
||||||
|
|
||||||
return PickResult.withNoResult(); // Indicates that this should be retried after backoff
|
return PickResult.withNoResult(); // Indicates that this should be retried after backoff
|
||||||
|
|
@ -495,8 +491,8 @@ final class RingHashLoadBalancer extends MultiChildLoadBalancer {
|
||||||
|
|
||||||
class RingHashChildLbState extends MultiChildLoadBalancer.ChildLbState {
|
class RingHashChildLbState extends MultiChildLoadBalancer.ChildLbState {
|
||||||
|
|
||||||
public RingHashChildLbState(Endpoint key, ResolvedAddresses resolvedAddresses) {
|
public RingHashChildLbState(Endpoint key) {
|
||||||
super(key, pickFirstLbProvider, null, EMPTY_PICKER, resolvedAddresses, true);
|
super(key, lazyLbFactory, null, EMPTY_PICKER);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
@ -504,22 +500,6 @@ final class RingHashLoadBalancer extends MultiChildLoadBalancer {
|
||||||
return new RingHashChildHelper();
|
return new RingHashChildHelper();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
protected void reactivate(LoadBalancerProvider policyProvider) {
|
|
||||||
if (!isDeactivated()) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
currentConnectivityState = CONNECTING;
|
|
||||||
getLb().switchTo(pickFirstLbProvider);
|
|
||||||
markReactivated();
|
|
||||||
getLb().acceptResolvedAddresses(this.getResolvedAddresses());
|
|
||||||
logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());
|
|
||||||
}
|
|
||||||
|
|
||||||
public void activate() {
|
|
||||||
reactivate(pickFirstLbProvider);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Need to expose this to the LB class
|
// Need to expose this to the LB class
|
||||||
@Override
|
@Override
|
||||||
protected void shutdown() {
|
protected void shutdown() {
|
||||||
|
|
@ -530,22 +510,19 @@ final class RingHashLoadBalancer extends MultiChildLoadBalancer {
|
||||||
@Override
|
@Override
|
||||||
public void updateBalancingState(final ConnectivityState newState,
|
public void updateBalancingState(final ConnectivityState newState,
|
||||||
final SubchannelPicker newPicker) {
|
final SubchannelPicker newPicker) {
|
||||||
// Subchannel picker and state are saved, but will only be propagated to the channel
|
|
||||||
// when the child instance exits deactivated state.
|
|
||||||
setCurrentState(newState);
|
setCurrentState(newState);
|
||||||
setCurrentPicker(newPicker);
|
setCurrentPicker(newPicker);
|
||||||
|
|
||||||
// If we are already in the process of resolving addresses, the overall balancing state
|
|
||||||
// will be updated at the end of it, and we don't need to trigger that update here.
|
|
||||||
if (getChildLbState(getKey()) == null) {
|
if (getChildLbState(getKey()) == null) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!isDeactivated() && !resolvingAddresses) {
|
// If we are already in the process of resolving addresses, the overall balancing state
|
||||||
|
// will be updated at the end of it, and we don't need to trigger that update here.
|
||||||
|
if (!resolvingAddresses) {
|
||||||
updateOverallBalancingState();
|
updateOverallBalancingState();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -177,15 +177,11 @@ public class RingHashLoadBalancerTest {
|
||||||
|
|
||||||
RingHashChildLbState childLbState =
|
RingHashChildLbState childLbState =
|
||||||
(RingHashChildLbState) loadBalancer.getChildLbStates().iterator().next();
|
(RingHashChildLbState) loadBalancer.getChildLbStates().iterator().next();
|
||||||
assertThat(childLbState.isDeactivated()).isTrue();
|
assertThat(subchannels.get(Collections.singletonList(childLbState.getEag()))).isNull();
|
||||||
|
|
||||||
// Picking subchannel triggers connection.
|
// Picking subchannel triggers connection.
|
||||||
PickSubchannelArgs args = getDefaultPickSubchannelArgs(hashFunc.hashVoid());
|
PickSubchannelArgs args = getDefaultPickSubchannelArgs(hashFunc.hashVoid());
|
||||||
pickerCaptor.getValue().pickSubchannel(args);
|
pickerCaptor.getValue().pickSubchannel(args);
|
||||||
assertThat(childLbState.isDeactivated()).isFalse();
|
|
||||||
String expectedLbType = PickFirstLoadBalancerProvider.isEnabledNewPickFirst()
|
|
||||||
? "PickFirstLeafLoadBalancer" : "PickFirstLoadBalancer";
|
|
||||||
assertThat(childLbState.getLb().delegateType()).isEqualTo(expectedLbType);
|
|
||||||
Subchannel subchannel = subchannels.get(Collections.singletonList(childLbState.getEag()));
|
Subchannel subchannel = subchannels.get(Collections.singletonList(childLbState.getEag()));
|
||||||
InOrder inOrder = Mockito.inOrder(helper, subchannel);
|
InOrder inOrder = Mockito.inOrder(helper, subchannel);
|
||||||
inOrder.verify(subchannel).requestConnection();
|
inOrder.verify(subchannel).requestConnection();
|
||||||
|
|
@ -423,7 +419,8 @@ public class RingHashLoadBalancerTest {
|
||||||
assertThat(addressesAcceptanceStatus.isOk()).isTrue();
|
assertThat(addressesAcceptanceStatus.isOk()).isTrue();
|
||||||
|
|
||||||
// Create subchannel for the first address
|
// Create subchannel for the first address
|
||||||
((RingHashChildLbState)loadBalancer.getChildLbStateEag(servers.get(0))).activate();
|
((RingHashChildLbState) loadBalancer.getChildLbStateEag(servers.get(0))).getCurrentPicker()
|
||||||
|
.pickSubchannel(getDefaultPickSubchannelArgs(hashFunc.hashVoid()));
|
||||||
verifyConnection(1);
|
verifyConnection(1);
|
||||||
|
|
||||||
reset(helper);
|
reset(helper);
|
||||||
|
|
@ -944,7 +941,8 @@ public class RingHashLoadBalancerTest {
|
||||||
|
|
||||||
// Activate them all to create the child LB and subchannel
|
// Activate them all to create the child LB and subchannel
|
||||||
for (ChildLbState childLbState : loadBalancer.getChildLbStates()) {
|
for (ChildLbState childLbState : loadBalancer.getChildLbStates()) {
|
||||||
((RingHashChildLbState)childLbState).activate();
|
childLbState.getCurrentPicker()
|
||||||
|
.pickSubchannel(getDefaultPickSubchannelArgs(hashFunc.hashVoid()));
|
||||||
assertThat(childLbState.getResolvedAddresses().getAttributes().get(IS_PETIOLE_POLICY))
|
assertThat(childLbState.getResolvedAddresses().getAttributes().get(IS_PETIOLE_POLICY))
|
||||||
.isTrue();
|
.isTrue();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue