#19167

Committed 22 Apr 2024 02:48PM UTC coverage: 88.084% (-0.01%) from 88.096%

Build # #19167

Build Type

push

github

Committed by

ejona86

Commit Message

util: Remove deactivation and GracefulSwitchLb from MultiChildLb

It is easy to manage these things outside of MultiChildLb and it makes
the shared code easier and use less memory. In particular, we don't want
to use many instances of GracefulSwitchLb in virtually every policy
simply because it was needed in one or two cases.

Run Details

31195 of 35415 relevant lines covered (88.08%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.06

/../xds/src/main/java/io/grpc/xds/RingHashLoadBalancer.java

/*
 * Copyright 2021 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 static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.SHUTDOWN;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import com.google.common.base.MoreObjects;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Multiset;
import com.google.common.primitives.UnsignedInteger;
import io.grpc.Attributes;
import io.grpc.ConnectivityState;
import io.grpc.EquivalentAddressGroup;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.util.MultiChildLoadBalancer;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import javax.annotation.Nullable;

/**
 * A {@link LoadBalancer} that provides consistent hashing based load balancing to upstream hosts.
 * It implements the "Ketama" hashing that maps hosts onto a circle (the "ring") by hashing its
 * addresses. Each request is routed to a host by hashing some property of the request and finding
 * the nearest corresponding host clockwise around the ring. Each host is placed on the ring some
 * number of times proportional to its weight. With the ring partitioned appropriately, the
 * addition or removal of one host from a set of N hosts will affect only 1/N requests.
 */
final class RingHashLoadBalancer extends MultiChildLoadBalancer {
  private static final Status RPC_HASH_NOT_FOUND =
      Status.INTERNAL.withDescription("RPC hash not found. Probably a bug because xds resolver"
          + " config selector always generates a hash.");
  private static final XxHash64 hashFunc = XxHash64.INSTANCE;

  private final LoadBalancer.Factory lazyLbFactory =
      new LazyLoadBalancer.Factory(pickFirstLbProvider);
  private final XdsLogger logger;
  private final SynchronizationContext syncContext;
  private List<RingEntry> ring;

  RingHashLoadBalancer(Helper helper) {
    super(helper);
    syncContext = checkNotNull(helper.getSynchronizationContext(), "syncContext");
    logger = XdsLogger.withLogId(InternalLogId.allocate("ring_hash_lb", helper.getAuthority()));
    logger.log(XdsLogLevel.INFO, "Created");
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);
    List<EquivalentAddressGroup> addrList = resolvedAddresses.getAddresses();
    Status addressValidityStatus = validateAddrList(addrList);
    if (!addressValidityStatus.isOk()) {
      return addressValidityStatus;
    }

    try {
      resolvingAddresses = true;
      // Subclass handles any special manipulation to create appropriate types of ChildLbStates
      Map<Object, ChildLbState> newChildren = createChildLbMap(resolvedAddresses);

      if (newChildren.isEmpty()) {
        addressValidityStatus = Status.UNAVAILABLE.withDescription(
            "Ring hash lb error: EDS resolution was successful, but there were no valid addresses");
        handleNameResolutionError(addressValidityStatus);
        return addressValidityStatus;
      }

      addMissingChildren(newChildren);
      updateChildrenWithResolvedAddresses(resolvedAddresses, newChildren);

      // Now do the ringhash specific logic with weights and building the ring
      RingHashConfig config = (RingHashConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
      if (config == null) {
        throw new IllegalArgumentException("Missing RingHash configuration");
      }
      Map<EquivalentAddressGroup, Long> serverWeights = new HashMap<>();
      long totalWeight = 0L;
      for (EquivalentAddressGroup eag : addrList) {
        Long weight = eag.getAttributes().get(InternalXdsAttributes.ATTR_SERVER_WEIGHT);
        // Support two ways of server weighing: either multiple instances of the same address
        // or each address contains a per-address weight attribute. If a weight is not provided,
        // each occurrence of the address will be counted a weight value of one.
        if (weight == null) {
          weight = 1L;
        }
        totalWeight += weight;
        EquivalentAddressGroup addrKey = stripAttrs(eag);
        if (serverWeights.containsKey(addrKey)) {
          serverWeights.put(addrKey, serverWeights.get(addrKey) + weight);
        } else {
          serverWeights.put(addrKey, weight);
        }
      }
      // Calculate scale
      long minWeight = Collections.min(serverWeights.values());
      double normalizedMinWeight = (double) minWeight / totalWeight;
      // Scale up the number of hashes per host such that the least-weighted host gets a whole
      // number of hashes on the the ring. Other hosts might not end up with whole numbers, and
      // that's fine (the ring-building algorithm can handle this). This preserves the original
      // implementation's behavior: when weights aren't provided, all hosts should get an equal
      // number of hashes. In the case where this number exceeds the max_ring_size, it's scaled
      // back down to fit.
      double scale = Math.min(
          Math.ceil(normalizedMinWeight * config.minRingSize) / normalizedMinWeight,
          (double) config.maxRingSize);

      // Build the ring
      ring = buildRing(serverWeights, totalWeight, scale);

      // Must update channel picker before return so that new RPCs will not be routed to deleted
      // clusters and resolver can remove them in service config.
      updateOverallBalancingState();

      shutdownRemoved(getRemovedChildren(newChildren.keySet()));
    } finally {
      this.resolvingAddresses = false;
    }

    return Status.OK;
  }


  /**
   * Updates the overall balancing state by aggregating the connectivity states of all subchannels.
   *
   * <p>Aggregation rules (in order of dominance):
   * <ol>
   *   <li>If there is at least one subchannel in READY state, overall state is READY</li>
   *   <li>If there are <em>2 or more</em> subchannels in TRANSIENT_FAILURE, overall state is
   *   TRANSIENT_FAILURE (to allow timely failover to another policy)</li>
   *   <li>If there is at least one subchannel in CONNECTING state, overall state is
   *   CONNECTING</li>
   *   <li> If there is one subchannel in TRANSIENT_FAILURE state and there is
   *    more than one subchannel, report CONNECTING </li>
   *   <li>If there is at least one subchannel in IDLE state, overall state is IDLE</li>
   *   <li>Otherwise, overall state is TRANSIENT_FAILURE</li>
   * </ol>
   */
  @Override
  protected void updateOverallBalancingState() {
    checkState(!getChildLbStates().isEmpty(), "no subchannel has been created");
    if (this.currentConnectivityState == SHUTDOWN) {
      // Ignore changes that happen after shutdown is called
      logger.log(XdsLogLevel.DEBUG, "UpdateOverallBalancingState called after shutdown");
      return;
    }

    // Calculate the current overall state to report
    int numIdle = 0;
    int numReady = 0;
    int numConnecting = 0;
    int numTF = 0;

    forloop:
    for (ChildLbState childLbState : getChildLbStates()) {
      ConnectivityState state = childLbState.getCurrentState();
      switch (state) {
        case READY:
          numReady++;
          break forloop;
        case CONNECTING:
          numConnecting++;
          break;
        case IDLE:
          numIdle++;
          break;
        case TRANSIENT_FAILURE:
          numTF++;
          break;
        default:
          // ignore it
      }
    }

    ConnectivityState overallState;
    if (numReady > 0) {
      overallState = READY;
    } else if (numTF >= 2) {
      overallState = TRANSIENT_FAILURE;
    } else if (numConnecting > 0) {
      overallState = CONNECTING;
    } else if (numTF == 1 && getChildLbStates().size() > 1) {
      overallState = CONNECTING;
    } else if (numIdle > 0) {
      overallState = IDLE;
    } else {
      overallState = TRANSIENT_FAILURE;
    }

    RingHashPicker picker = new RingHashPicker(syncContext, ring, getImmutableChildMap());
    getHelper().updateBalancingState(overallState, picker);
    this.currentConnectivityState = overallState;
  }

  @Override
  protected ChildLbState createChildLbState(Object key, Object policyConfig,
      SubchannelPicker initialPicker, ResolvedAddresses resolvedAddresses) {
    return new RingHashChildLbState((Endpoint)key);
  }

  private Status validateAddrList(List<EquivalentAddressGroup> addrList) {
    if (addrList.isEmpty()) {
      Status unavailableStatus = Status.UNAVAILABLE.withDescription("Ring hash lb error: EDS "
              + "resolution was successful, but returned server addresses are empty.");
      handleNameResolutionError(unavailableStatus);
      return unavailableStatus;
    }

    String dupAddrString = validateNoDuplicateAddresses(addrList);
    if (dupAddrString != null) {
      Status unavailableStatus = Status.UNAVAILABLE.withDescription("Ring hash lb error: EDS "
              + "resolution was successful, but there were duplicate addresses: " + dupAddrString);
      handleNameResolutionError(unavailableStatus);
      return unavailableStatus;
    }

    long totalWeight = 0;
    for (EquivalentAddressGroup eag : addrList) {
      Long weight = eag.getAttributes().get(InternalXdsAttributes.ATTR_SERVER_WEIGHT);

      if (weight == null) {
        weight = 1L;
      }

      if (weight < 0) {
        Status unavailableStatus = Status.UNAVAILABLE.withDescription(
            String.format("Ring hash lb error: EDS resolution was successful, but returned a "
                        + "negative weight for %s.", stripAttrs(eag)));
        handleNameResolutionError(unavailableStatus);
        return unavailableStatus;
      }
      if (weight > UnsignedInteger.MAX_VALUE.longValue()) {
        Status unavailableStatus = Status.UNAVAILABLE.withDescription(
            String.format("Ring hash lb error: EDS resolution was successful, but returned a weight"
                + " too large to fit in an unsigned int for %s.", stripAttrs(eag)));
        handleNameResolutionError(unavailableStatus);
        return unavailableStatus;
      }
      totalWeight += weight;
    }

    if (totalWeight > UnsignedInteger.MAX_VALUE.longValue()) {
      Status unavailableStatus = Status.UNAVAILABLE.withDescription(
          String.format(
              "Ring hash lb error: EDS resolution was successful, but returned a sum of weights too"
                  + " large to fit in an unsigned int (%d).", totalWeight));
      handleNameResolutionError(unavailableStatus);
      return unavailableStatus;
    }

    return Status.OK;
  }

  @Nullable
  private String validateNoDuplicateAddresses(List<EquivalentAddressGroup> addrList) {
    Set<SocketAddress> addresses = new HashSet<>();
    Multiset<String> dups = HashMultiset.create();
    for (EquivalentAddressGroup eag : addrList) {
      for (SocketAddress address : eag.getAddresses()) {
        if (!addresses.add(address)) {
          dups.add(address.toString());
        }
      }
    }

    if (!dups.isEmpty()) {
      return dups.entrySet().stream()
          .map((dup) ->
              String.format("Address: %s, count: %d", dup.getElement(), dup.getCount() + 1))
          .collect(Collectors.joining("; "));
    }

    return null;
  }

  private static List<RingEntry> buildRing(
      Map<EquivalentAddressGroup, Long> serverWeights, long totalWeight, double scale) {
    List<RingEntry> ring = new ArrayList<>();
    double currentHashes = 0.0;
    double targetHashes = 0.0;
    for (Map.Entry<EquivalentAddressGroup, Long> entry : serverWeights.entrySet()) {
      Endpoint endpoint = new Endpoint(entry.getKey());
      double normalizedWeight = (double) entry.getValue() / totalWeight;
      // Per GRFC A61 use the first address for the hash
      StringBuilder sb = new StringBuilder(entry.getKey().getAddresses().get(0).toString());
      sb.append('_');
      int lengthWithoutCounter = sb.length();
      targetHashes += scale * normalizedWeight;
      long i = 0L;
      while (currentHashes < targetHashes) {
        sb.append(i);
        long hash = hashFunc.hashAsciiString(sb.toString());
        ring.add(new RingEntry(hash, endpoint));
        i++;
        currentHashes++;
        sb.setLength(lengthWithoutCounter);
      }
    }
    Collections.sort(ring);
    return Collections.unmodifiableList(ring);
  }

  @SuppressWarnings("ReferenceEquality")
  public static EquivalentAddressGroup stripAttrs(EquivalentAddressGroup eag) {
    if (eag.getAttributes() == Attributes.EMPTY) {
      return eag;
    }
    return new EquivalentAddressGroup(eag.getAddresses());
  }

  private static final class RingHashPicker extends SubchannelPicker {
    private final SynchronizationContext syncContext;
    private final List<RingEntry> ring;
    // Avoid synchronization between pickSubchannel and subchannel's connectivity state change,
    // freeze picker's view of subchannel's connectivity state.
    // TODO(chengyuanzhang): can be more performance-friendly with
    //  IdentityHashMap<Subchannel, ConnectivityStateInfo> and RingEntry contains Subchannel.
    private final Map<Endpoint, SubchannelView> pickableSubchannels;  // read-only

    private RingHashPicker(
        SynchronizationContext syncContext, List<RingEntry> ring,
        ImmutableMap<Object, ChildLbState> subchannels) {
      this.syncContext = syncContext;
      this.ring = ring;
      pickableSubchannels = new HashMap<>(subchannels.size());
      for (Map.Entry<Object, ChildLbState> entry : subchannels.entrySet()) {
        RingHashChildLbState childLbState = (RingHashChildLbState) entry.getValue();
        pickableSubchannels.put((Endpoint)entry.getKey(),
            new SubchannelView(childLbState, childLbState.getCurrentState()));
      }
    }

    // Find the ring entry with hash next to (clockwise) the RPC's hash (binary search).
    private int getTargetIndex(Long requestHash) {
      if (ring.size() <= 1) {
        return 0;
      }

      int low = 0;
      int high = ring.size() - 1;
      int mid = (low + high) / 2;
      do {
        long midVal = ring.get(mid).hash;
        long midValL = mid == 0 ? 0 : ring.get(mid - 1).hash;
        if (requestHash <= midVal && requestHash > midValL) {
          break;
        }
        if (midVal < requestHash) {
          low = mid + 1;
        } else {
          high =  mid - 1;
        }
        mid = (low + high) / 2;
      } while (mid < ring.size() && low <= high);
      return mid;
    }

    @Override
    public PickResult pickSubchannel(PickSubchannelArgs args) {
      Long requestHash = args.getCallOptions().getOption(XdsNameResolver.RPC_HASH_KEY);
      if (requestHash == null) {
        return PickResult.withError(RPC_HASH_NOT_FOUND);
      }

      int targetIndex = getTargetIndex(requestHash);

      // Per gRFC A61, because of sticky-TF with PickFirst's auto reconnect on TF, we ignore
      // all TF subchannels and find the first ring entry in READY, CONNECTING or IDLE.  If
      // CONNECTING or IDLE we return a pick with no results.  Additionally, if that entry is in
      // IDLE, we initiate a connection.
      for (int i = 0; i < ring.size(); i++) {
        int index = (targetIndex + i) % ring.size();
        SubchannelView subchannelView = pickableSubchannels.get(ring.get(index).addrKey);
        RingHashChildLbState childLbState = subchannelView.childLbState;

        if (subchannelView.connectivityState  == READY) {
          return childLbState.getCurrentPicker().pickSubchannel(args);
        }

        // RPCs can be buffered if the next subchannel is pending (per A62). Otherwise, RPCs
        // are failed unless there is a READY connection.
        if (subchannelView.connectivityState == CONNECTING) {
          return PickResult.withNoResult();
        }

        if (subchannelView.connectivityState == IDLE) {
          syncContext.execute(() -> {
            childLbState.getLb().requestConnection();
          });

          return PickResult.withNoResult(); // Indicates that this should be retried after backoff
        }
      }

      // return the pick from the original subchannel hit by hash, which is probably an error
      RingHashChildLbState originalSubchannel =
          pickableSubchannels.get(ring.get(targetIndex).addrKey).childLbState;
      return originalSubchannel.getCurrentPicker().pickSubchannel(args);
    }

  }

  /**
   * An unmodifiable view of a subchannel with state not subject to its real connectivity
   * state changes.
   */
  private static final class SubchannelView {
    private final RingHashChildLbState childLbState;
    private final ConnectivityState connectivityState;

    private SubchannelView(RingHashChildLbState childLbState, ConnectivityState state) {
      this.childLbState = childLbState;
      this.connectivityState = state;
    }
  }

  private static final class RingEntry implements Comparable<RingEntry> {
    private final long hash;
    private final Endpoint addrKey;

    private RingEntry(long hash, Endpoint addrKey) {
      this.hash = hash;
      this.addrKey = addrKey;
    }

    @Override
    public int compareTo(RingEntry entry) {
      return Long.compare(hash, entry.hash);
    }
  }

  /**
   * Configures the ring property. The larger the ring is (that is, the more hashes there are
   * for each provided host) the better the request distribution will reflect the desired weights.
   */
  static final class RingHashConfig {
    final long minRingSize;
    final long maxRingSize;

    RingHashConfig(long minRingSize, long maxRingSize) {
      checkArgument(minRingSize > 0, "minRingSize <= 0");
      checkArgument(maxRingSize > 0, "maxRingSize <= 0");
      checkArgument(minRingSize <= maxRingSize, "minRingSize > maxRingSize");
      this.minRingSize = minRingSize;
      this.maxRingSize = maxRingSize;
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
          .add("minRingSize", minRingSize)
          .add("maxRingSize", maxRingSize)
          .toString();
    }
  }

  class RingHashChildLbState extends MultiChildLoadBalancer.ChildLbState {

    public RingHashChildLbState(Endpoint key) {
      super(key, lazyLbFactory, null, EMPTY_PICKER);
    }

    @Override
    protected ChildLbStateHelper createChildHelper() {
      return new RingHashChildHelper();
    }

    // Need to expose this to the LB class
    @Override
    protected void shutdown() {
      super.shutdown();
    }

    private class RingHashChildHelper extends ChildLbStateHelper {
      @Override
      public void updateBalancingState(final ConnectivityState newState,
                                       final SubchannelPicker newPicker) {
        setCurrentState(newState);
        setCurrentPicker(newPicker);
        
        if (getChildLbState(getKey()) == null) {
          return;
        }

        // 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();
        }
      }
    }
  }
}

1	/*
2	* Copyright 2021 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.xds;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static com.google.common.base.Preconditions.checkState;
22	import static io.grpc.ConnectivityState.CONNECTING;
23	import static io.grpc.ConnectivityState.IDLE;
24	import static io.grpc.ConnectivityState.READY;
25	import static io.grpc.ConnectivityState.SHUTDOWN;
26	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
27
28	import com.google.common.base.MoreObjects;
29	import com.google.common.collect.HashMultiset;
30	import com.google.common.collect.ImmutableMap;
31	import com.google.common.collect.Multiset;
32	import com.google.common.primitives.UnsignedInteger;
33	import io.grpc.Attributes;
34	import io.grpc.ConnectivityState;
35	import io.grpc.EquivalentAddressGroup;
36	import io.grpc.InternalLogId;
37	import io.grpc.LoadBalancer;
38	import io.grpc.Status;
39	import io.grpc.SynchronizationContext;
40	import io.grpc.util.MultiChildLoadBalancer;
41	import io.grpc.xds.client.XdsLogger;
42	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
43	import java.net.SocketAddress;
44	import java.util.ArrayList;
45	import java.util.Collections;
46	import java.util.HashMap;
47	import java.util.HashSet;
48	import java.util.List;
49	import java.util.Map;
50	import java.util.Set;
51	import java.util.stream.Collectors;
52	import javax.annotation.Nullable;
53
54	/**
55	* A {@link LoadBalancer} that provides consistent hashing based load balancing to upstream hosts.
56	* It implements the "Ketama" hashing that maps hosts onto a circle (the "ring") by hashing its
57	* addresses. Each request is routed to a host by hashing some property of the request and finding
58	* the nearest corresponding host clockwise around the ring. Each host is placed on the ring some
59	* number of times proportional to its weight. With the ring partitioned appropriately, the
60	* addition or removal of one host from a set of N hosts will affect only 1/N requests.
61	*/
62	final class RingHashLoadBalancer extends MultiChildLoadBalancer {
63	private static final Status RPC_HASH_NOT_FOUND =	1✔
64	Status.INTERNAL.withDescription("RPC hash not found. Probably a bug because xds resolver"	1✔
65	+ " config selector always generates a hash.");
66	private static final XxHash64 hashFunc = XxHash64.INSTANCE;	1✔
67
68	private final LoadBalancer.Factory lazyLbFactory =	1✔
69	new LazyLoadBalancer.Factory(pickFirstLbProvider);
70	private final XdsLogger logger;
71	private final SynchronizationContext syncContext;
72	private List<RingEntry> ring;
73
74	RingHashLoadBalancer(Helper helper) {
75	super(helper);	1✔
76	syncContext = checkNotNull(helper.getSynchronizationContext(), "syncContext");	1✔
77	logger = XdsLogger.withLogId(InternalLogId.allocate("ring_hash_lb", helper.getAuthority()));	1✔
78	logger.log(XdsLogLevel.INFO, "Created");	1✔
79	}	1✔
80
81	@Override
82	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
83	logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);	1✔
84	List<EquivalentAddressGroup> addrList = resolvedAddresses.getAddresses();	1✔
85	Status addressValidityStatus = validateAddrList(addrList);	1✔
86	if (!addressValidityStatus.isOk()) {	1✔
87	return addressValidityStatus;	1✔
88	}
89
90	try {
91	resolvingAddresses = true;	1✔
92	// Subclass handles any special manipulation to create appropriate types of ChildLbStates
93	Map<Object, ChildLbState> newChildren = createChildLbMap(resolvedAddresses);	1✔
94
95	if (newChildren.isEmpty()) {	1✔
96	addressValidityStatus = Status.UNAVAILABLE.withDescription(	×
97	"Ring hash lb error: EDS resolution was successful, but there were no valid addresses");
98	handleNameResolutionError(addressValidityStatus);	×
99	return addressValidityStatus;	×
100	}
101
102	addMissingChildren(newChildren);	1✔
103	updateChildrenWithResolvedAddresses(resolvedAddresses, newChildren);	1✔
104
105	// Now do the ringhash specific logic with weights and building the ring
106	RingHashConfig config = (RingHashConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
107	if (config == null) {	1✔
108	throw new IllegalArgumentException("Missing RingHash configuration");	×
109	}
110	Map<EquivalentAddressGroup, Long> serverWeights = new HashMap<>();	1✔
111	long totalWeight = 0L;	1✔
112	for (EquivalentAddressGroup eag : addrList) {	1✔
113	Long weight = eag.getAttributes().get(InternalXdsAttributes.ATTR_SERVER_WEIGHT);	1✔
114	// Support two ways of server weighing: either multiple instances of the same address
115	// or each address contains a per-address weight attribute. If a weight is not provided,
116	// each occurrence of the address will be counted a weight value of one.
117	if (weight == null) {	1✔
118	weight = 1L;	×
119	}
120	totalWeight += weight;	1✔
121	EquivalentAddressGroup addrKey = stripAttrs(eag);	1✔
122	if (serverWeights.containsKey(addrKey)) {	1✔
123	serverWeights.put(addrKey, serverWeights.get(addrKey) + weight);	×
124	} else {
125	serverWeights.put(addrKey, weight);	1✔
126	}
127	}	1✔
128	// Calculate scale
129	long minWeight = Collections.min(serverWeights.values());	1✔
130	double normalizedMinWeight = (double) minWeight / totalWeight;	1✔
131	// Scale up the number of hashes per host such that the least-weighted host gets a whole
132	// number of hashes on the the ring. Other hosts might not end up with whole numbers, and
133	// that's fine (the ring-building algorithm can handle this). This preserves the original
134	// implementation's behavior: when weights aren't provided, all hosts should get an equal
135	// number of hashes. In the case where this number exceeds the max_ring_size, it's scaled
136	// back down to fit.
137	double scale = Math.min(	1✔
138	Math.ceil(normalizedMinWeight * config.minRingSize) / normalizedMinWeight,	1✔
139	(double) config.maxRingSize);
140
141	// Build the ring
142	ring = buildRing(serverWeights, totalWeight, scale);	1✔
143
144	// Must update channel picker before return so that new RPCs will not be routed to deleted
145	// clusters and resolver can remove them in service config.
146	updateOverallBalancingState();	1✔
147
148	shutdownRemoved(getRemovedChildren(newChildren.keySet()));	1✔
149	} finally {
150	this.resolvingAddresses = false;	1✔
151	}
152
153	return Status.OK;	1✔
154	}
155
156
157	/**
158	* Updates the overall balancing state by aggregating the connectivity states of all subchannels.
159	*
160	* <p>Aggregation rules (in order of dominance):
161	* <ol>
162	* <li>If there is at least one subchannel in READY state, overall state is READY</li>
163	* <li>If there are <em>2 or more</em> subchannels in TRANSIENT_FAILURE, overall state is
164	* TRANSIENT_FAILURE (to allow timely failover to another policy)</li>
165	* <li>If there is at least one subchannel in CONNECTING state, overall state is
166	* CONNECTING</li>
167	* <li> If there is one subchannel in TRANSIENT_FAILURE state and there is
168	* more than one subchannel, report CONNECTING </li>
169	* <li>If there is at least one subchannel in IDLE state, overall state is IDLE</li>
170	* <li>Otherwise, overall state is TRANSIENT_FAILURE</li>
171	* </ol>
172	*/
173	@Override
174	protected void updateOverallBalancingState() {
175	checkState(!getChildLbStates().isEmpty(), "no subchannel has been created");	1✔
176	if (this.currentConnectivityState == SHUTDOWN) {	1✔
177	// Ignore changes that happen after shutdown is called
178	logger.log(XdsLogLevel.DEBUG, "UpdateOverallBalancingState called after shutdown");	×
179	return;	×
180	}
181
182	// Calculate the current overall state to report
183	int numIdle = 0;	1✔
184	int numReady = 0;	1✔
185	int numConnecting = 0;	1✔
186	int numTF = 0;	1✔
187
188	forloop:
189	for (ChildLbState childLbState : getChildLbStates()) {	1✔
190	ConnectivityState state = childLbState.getCurrentState();	1✔
191	switch (state) {	1✔
192	case READY:
193	numReady++;	1✔
194	break forloop;	1✔
195	case CONNECTING:
196	numConnecting++;	1✔
197	break;	1✔
198	case IDLE:
199	numIdle++;	1✔
200	break;	1✔
201	case TRANSIENT_FAILURE:
202	numTF++;	1✔
203	break;	1✔
204	default:
205	// ignore it
206	}
207	}	1✔
208
209	ConnectivityState overallState;
210	if (numReady > 0) {	1✔
211	overallState = READY;	1✔
212	} else if (numTF >= 2) {	1✔
213	overallState = TRANSIENT_FAILURE;	1✔
214	} else if (numConnecting > 0) {	1✔
215	overallState = CONNECTING;	1✔
216	} else if (numTF == 1 && getChildLbStates().size() > 1) {	1✔
217	overallState = CONNECTING;	1✔
218	} else if (numIdle > 0) {	1✔
219	overallState = IDLE;	1✔
220	} else {
221	overallState = TRANSIENT_FAILURE;	×
222	}
223
224	RingHashPicker picker = new RingHashPicker(syncContext, ring, getImmutableChildMap());	1✔
225	getHelper().updateBalancingState(overallState, picker);	1✔
226	this.currentConnectivityState = overallState;	1✔
227	}	1✔
228
229	@Override
230	protected ChildLbState createChildLbState(Object key, Object policyConfig,
231	SubchannelPicker initialPicker, ResolvedAddresses resolvedAddresses) {
232	return new RingHashChildLbState((Endpoint)key);	1✔
233	}
234
235	private Status validateAddrList(List<EquivalentAddressGroup> addrList) {
236	if (addrList.isEmpty()) {	1✔
237	Status unavailableStatus = Status.UNAVAILABLE.withDescription("Ring hash lb error: EDS "	×
238	+ "resolution was successful, but returned server addresses are empty.");
239	handleNameResolutionError(unavailableStatus);	×
240	return unavailableStatus;	×
241	}
242
243	String dupAddrString = validateNoDuplicateAddresses(addrList);	1✔
244	if (dupAddrString != null) {	1✔
245	Status unavailableStatus = Status.UNAVAILABLE.withDescription("Ring hash lb error: EDS "	1✔
246	+ "resolution was successful, but there were duplicate addresses: " + dupAddrString);
247	handleNameResolutionError(unavailableStatus);	1✔
248	return unavailableStatus;	1✔
249	}
250
251	long totalWeight = 0;	1✔
252	for (EquivalentAddressGroup eag : addrList) {	1✔
253	Long weight = eag.getAttributes().get(InternalXdsAttributes.ATTR_SERVER_WEIGHT);	1✔
254
255	if (weight == null) {	1✔
256	weight = 1L;	×
257	}
258
259	if (weight < 0) {	1✔
260	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
261	String.format("Ring hash lb error: EDS resolution was successful, but returned a "	1✔
262	+ "negative weight for %s.", stripAttrs(eag)));	1✔
263	handleNameResolutionError(unavailableStatus);	1✔
264	return unavailableStatus;	1✔
265	}
266	if (weight > UnsignedInteger.MAX_VALUE.longValue()) {	1✔
267	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
268	String.format("Ring hash lb error: EDS resolution was successful, but returned a weight"	1✔
269	+ " too large to fit in an unsigned int for %s.", stripAttrs(eag)));	1✔
270	handleNameResolutionError(unavailableStatus);	1✔
271	return unavailableStatus;	1✔
272	}
273	totalWeight += weight;	1✔
274	}	1✔
275
276	if (totalWeight > UnsignedInteger.MAX_VALUE.longValue()) {	1✔
277	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
278	String.format(	1✔
279	"Ring hash lb error: EDS resolution was successful, but returned a sum of weights too"
280	+ " large to fit in an unsigned int (%d).", totalWeight));	1✔
281	handleNameResolutionError(unavailableStatus);	1✔
282	return unavailableStatus;	1✔
283	}
284
285	return Status.OK;	1✔
286	}
287
288	@Nullable
289	private String validateNoDuplicateAddresses(List<EquivalentAddressGroup> addrList) {
290	Set<SocketAddress> addresses = new HashSet<>();	1✔
291	Multiset<String> dups = HashMultiset.create();	1✔
292	for (EquivalentAddressGroup eag : addrList) {	1✔
293	for (SocketAddress address : eag.getAddresses()) {	1✔
294	if (!addresses.add(address)) {	1✔
295	dups.add(address.toString());	1✔
296	}
297	}	1✔
298	}	1✔
299
300	if (!dups.isEmpty()) {	1✔
301	return dups.entrySet().stream()	1✔
302	.map((dup) ->	1✔
303	String.format("Address: %s, count: %d", dup.getElement(), dup.getCount() + 1))	1✔
304	.collect(Collectors.joining("; "));	1✔
305	}
306
307	return null;	1✔
308	}
309
310	private static List<RingEntry> buildRing(
311	Map<EquivalentAddressGroup, Long> serverWeights, long totalWeight, double scale) {
312	List<RingEntry> ring = new ArrayList<>();	1✔
313	double currentHashes = 0.0;	1✔
314	double targetHashes = 0.0;	1✔
315	for (Map.Entry<EquivalentAddressGroup, Long> entry : serverWeights.entrySet()) {	1✔
316	Endpoint endpoint = new Endpoint(entry.getKey());	1✔
317	double normalizedWeight = (double) entry.getValue() / totalWeight;	1✔
318	// Per GRFC A61 use the first address for the hash
319	StringBuilder sb = new StringBuilder(entry.getKey().getAddresses().get(0).toString());	1✔
320	sb.append('_');	1✔
321	int lengthWithoutCounter = sb.length();	1✔
322	targetHashes += scale * normalizedWeight;	1✔
323	long i = 0L;	1✔
324	while (currentHashes < targetHashes) {	1✔
325	sb.append(i);	1✔
326	long hash = hashFunc.hashAsciiString(sb.toString());	1✔
327	ring.add(new RingEntry(hash, endpoint));	1✔
328	i++;	1✔
329	currentHashes++;	1✔
330	sb.setLength(lengthWithoutCounter);	1✔
331	}	1✔
332	}	1✔
333	Collections.sort(ring);	1✔
334	return Collections.unmodifiableList(ring);	1✔
335	}
336
337	@SuppressWarnings("ReferenceEquality")
338	public static EquivalentAddressGroup stripAttrs(EquivalentAddressGroup eag) {
339	if (eag.getAttributes() == Attributes.EMPTY) {	1✔
340	return eag;	×
341	}
342	return new EquivalentAddressGroup(eag.getAddresses());	1✔
343	}
344
345	private static final class RingHashPicker extends SubchannelPicker {
346	private final SynchronizationContext syncContext;
347	private final List<RingEntry> ring;
348	// Avoid synchronization between pickSubchannel and subchannel's connectivity state change,
349	// freeze picker's view of subchannel's connectivity state.
350	// TODO(chengyuanzhang): can be more performance-friendly with
351	// IdentityHashMap<Subchannel, ConnectivityStateInfo> and RingEntry contains Subchannel.
352	private final Map<Endpoint, SubchannelView> pickableSubchannels; // read-only
353
354	private RingHashPicker(
355	SynchronizationContext syncContext, List<RingEntry> ring,
356	ImmutableMap<Object, ChildLbState> subchannels) {	1✔
357	this.syncContext = syncContext;	1✔
358	this.ring = ring;	1✔
359	pickableSubchannels = new HashMap<>(subchannels.size());	1✔
360	for (Map.Entry<Object, ChildLbState> entry : subchannels.entrySet()) {	1✔
361	RingHashChildLbState childLbState = (RingHashChildLbState) entry.getValue();	1✔
362	pickableSubchannels.put((Endpoint)entry.getKey(),	1✔
363	new SubchannelView(childLbState, childLbState.getCurrentState()));	1✔
364	}	1✔
365	}	1✔
366
367	// Find the ring entry with hash next to (clockwise) the RPC's hash (binary search).
368	private int getTargetIndex(Long requestHash) {
369	if (ring.size() <= 1) {	1✔
370	return 0;	×
371	}
372
373	int low = 0;	1✔
374	int high = ring.size() - 1;	1✔
375	int mid = (low + high) / 2;	1✔
376	do {
377	long midVal = ring.get(mid).hash;	1✔
378	long midValL = mid == 0 ? 0 : ring.get(mid - 1).hash;	1✔
379	if (requestHash <= midVal && requestHash > midValL) {	1✔
380	break;	1✔
381	}
382	if (midVal < requestHash) {	1✔
383	low = mid + 1;	1✔
384	} else {
385	high = mid - 1;	1✔
386	}
387	mid = (low + high) / 2;	1✔
388	} while (mid < ring.size() && low <= high);	1✔
389	return mid;	1✔
390	}
391
392	@Override
393	public PickResult pickSubchannel(PickSubchannelArgs args) {
394	Long requestHash = args.getCallOptions().getOption(XdsNameResolver.RPC_HASH_KEY);	1✔
395	if (requestHash == null) {	1✔
396	return PickResult.withError(RPC_HASH_NOT_FOUND);	×
397	}
398
399	int targetIndex = getTargetIndex(requestHash);	1✔
400
401	// Per gRFC A61, because of sticky-TF with PickFirst's auto reconnect on TF, we ignore
402	// all TF subchannels and find the first ring entry in READY, CONNECTING or IDLE. If
403	// CONNECTING or IDLE we return a pick with no results. Additionally, if that entry is in
404	// IDLE, we initiate a connection.
405	for (int i = 0; i < ring.size(); i++) {	1✔
406	int index = (targetIndex + i) % ring.size();	1✔
407	SubchannelView subchannelView = pickableSubchannels.get(ring.get(index).addrKey);	1✔
408	RingHashChildLbState childLbState = subchannelView.childLbState;	1✔
409
410	if (subchannelView.connectivityState == READY) {	1✔
411	return childLbState.getCurrentPicker().pickSubchannel(args);	1✔
412	}
413
414	// RPCs can be buffered if the next subchannel is pending (per A62). Otherwise, RPCs
415	// are failed unless there is a READY connection.
416	if (subchannelView.connectivityState == CONNECTING) {	1✔
417	return PickResult.withNoResult();	1✔
418	}
419
420	if (subchannelView.connectivityState == IDLE) {	1✔
421	syncContext.execute(() -> {	1✔
422	childLbState.getLb().requestConnection();	1✔
423	});	1✔
424
425	return PickResult.withNoResult(); // Indicates that this should be retried after backoff	1✔
426	}
427	}
428
429	// return the pick from the original subchannel hit by hash, which is probably an error
430	RingHashChildLbState originalSubchannel =	1✔
431	pickableSubchannels.get(ring.get(targetIndex).addrKey).childLbState;	1✔
432	return originalSubchannel.getCurrentPicker().pickSubchannel(args);	1✔
433	}
434
435	}
436
437	/**
438	* An unmodifiable view of a subchannel with state not subject to its real connectivity
439	* state changes.
440	*/
441	private static final class SubchannelView {
442	private final RingHashChildLbState childLbState;
443	private final ConnectivityState connectivityState;
444
445	private SubchannelView(RingHashChildLbState childLbState, ConnectivityState state) {	1✔
446	this.childLbState = childLbState;	1✔
447	this.connectivityState = state;	1✔
448	}	1✔
449	}
450
451	private static final class RingEntry implements Comparable<RingEntry> {
452	private final long hash;
453	private final Endpoint addrKey;
454
455	private RingEntry(long hash, Endpoint addrKey) {	1✔
456	this.hash = hash;	1✔
457	this.addrKey = addrKey;	1✔
458	}	1✔
459
460	@Override
461	public int compareTo(RingEntry entry) {
462	return Long.compare(hash, entry.hash);	1✔
463	}
464	}
465
466	/**
467	* Configures the ring property. The larger the ring is (that is, the more hashes there are
468	* for each provided host) the better the request distribution will reflect the desired weights.
469	*/
470	static final class RingHashConfig {
471	final long minRingSize;
472	final long maxRingSize;
473
474	RingHashConfig(long minRingSize, long maxRingSize) {	1✔
475	checkArgument(minRingSize > 0, "minRingSize <= 0");	1✔
476	checkArgument(maxRingSize > 0, "maxRingSize <= 0");	1✔
477	checkArgument(minRingSize <= maxRingSize, "minRingSize > maxRingSize");	1✔
478	this.minRingSize = minRingSize;	1✔
479	this.maxRingSize = maxRingSize;	1✔
480	}	1✔
481
482	@Override
483	public String toString() {
484	return MoreObjects.toStringHelper(this)	×
485	.add("minRingSize", minRingSize)	×
486	.add("maxRingSize", maxRingSize)	×
487	.toString();	×
488	}
489	}
490
491	class RingHashChildLbState extends MultiChildLoadBalancer.ChildLbState {
492
493	public RingHashChildLbState(Endpoint key) {	1✔
494	super(key, lazyLbFactory, null, EMPTY_PICKER);	1✔
495	}	1✔
496
497	@Override
498	protected ChildLbStateHelper createChildHelper() {
499	return new RingHashChildHelper();	1✔
500	}
501
502	// Need to expose this to the LB class
503	@Override
504	protected void shutdown() {
505	super.shutdown();	1✔
506	}	1✔
507
508	private class RingHashChildHelper extends ChildLbStateHelper {	1✔
509	@Override
510	public void updateBalancingState(final ConnectivityState newState,
511	final SubchannelPicker newPicker) {
512	setCurrentState(newState);	1✔
513	setCurrentPicker(newPicker);	1✔
514
515	if (getChildLbState(getKey()) == null) {	1✔
516	return;	×
517	}
518
519	// If we are already in the process of resolving addresses, the overall balancing state
520	// will be updated at the end of it, and we don't need to trigger that update here.
521	if (!resolvingAddresses) {	1✔
522	updateOverallBalancingState();	1✔
523	}
524	}	1✔
525	}
526	}
527	}

grpc / grpc-java / #19167

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous