util: Use acceptResolvedAddresses() for MultiChildLb children

util/src/main/java/io/grpc/util/MultiChildLoadBalancer.java

@@ -107,21 +107,22 @@
    */
   @Override
   public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
+    logger.log(Level.FINE, "Received resolution result: {0}", resolvedAddresses);
     try {
       resolvingAddresses = true;
 
       // process resolvedAddresses to update children
-      AcceptResolvedAddrRetVal acceptRetVal = acceptResolvedAddressesInternal(resolvedAddresses);
-      if (!acceptRetVal.status.isOk()) {
-        return acceptRetVal.status;
+      Map<Object, ResolvedAddresses> newChildAddresses = createChildAddressesMap(resolvedAddresses);
+
+      // Handle error case
+      if (newChildAddresses.isEmpty()) {
+        Status unavailableStatus = Status.UNAVAILABLE.withDescription(
+            "NameResolver returned no usable address. " + resolvedAddresses);
+        handleNameResolutionError(unavailableStatus);
+        return unavailableStatus;
       }
 
-      // Update the picker and our connectivity state
-      updateOverallBalancingState();
-
-      // shutdown removed children
-      shutdownRemoved(acceptRetVal.removedChildren);
-      return acceptRetVal.status;
+      return updateChildrenWithResolvedAddresses(newChildAddresses);
     } finally {
       resolvingAddresses = false;
     }
@@ -149,31 +150,7 @@
     childLbStates.clear();
   }
 
-  /**
-   *   This does the work to update the child map and calculate which children have been removed.
-   *   You must call {@link #updateOverallBalancingState} to update the picker
-   *   and call {@link #shutdownRemoved(List)} to shutdown the endpoints that have been removed.
-    */
-  protected final AcceptResolvedAddrRetVal acceptResolvedAddressesInternal(
-      ResolvedAddresses resolvedAddresses) {
-    logger.log(Level.FINE, "Received resolution result: {0}", resolvedAddresses);
-
-    Map<Object, ResolvedAddresses> newChildAddresses = createChildAddressesMap(resolvedAddresses);
-
-    // Handle error case
-    if (newChildAddresses.isEmpty()) {
-      Status unavailableStatus = Status.UNAVAILABLE.withDescription(
-          "NameResolver returned no usable address. " + resolvedAddresses);
-      handleNameResolutionError(unavailableStatus);
-      return new AcceptResolvedAddrRetVal(unavailableStatus, null);
-    }
-
-    List<ChildLbState> removed = updateChildrenWithResolvedAddresses(newChildAddresses);
-    return new AcceptResolvedAddrRetVal(Status.OK, removed);
-  }
-
-  /** Returns removed children. */
-  private List<ChildLbState> updateChildrenWithResolvedAddresses(
+  private Status updateChildrenWithResolvedAddresses(
       Map<Object, ResolvedAddresses> newChildAddresses) {
     // Create a map with the old values
     Map<Object, ChildLbState> oldStatesMap =
@@ -183,6 +160,7 @@
     }
 
     // Move ChildLbStates from the map to a new list (preserving the new map's order)
+    Status status = Status.OK;
     List<ChildLbState> newChildLbStates = new ArrayList<>(newChildAddresses.size());
     for (Map.Entry<Object, ResolvedAddresses> entry : newChildAddresses.entrySet()) {
       ChildLbState childLbState = oldStatesMap.remove(entry.getKey());
@@ -191,21 +169,23 @@
       }
       newChildLbStates.add(childLbState);
       if (entry.getValue() != null) {
-        childLbState.lb.handleResolvedAddresses(entry.getValue()); // update child LB
+        // update child LB
+        Status newStatus = childLbState.lb.acceptResolvedAddresses(entry.getValue());
+        if (!newStatus.isOk()) {
+          status = newStatus;
+        }
       }
     }
 
     childLbStates = newChildLbStates;
-    // Remaining entries in map are orphaned
-    return new ArrayList<>(oldStatesMap.values());
-  }
+    // Update the picker and our connectivity state
+    updateOverallBalancingState();
 
-  protected final void shutdownRemoved(List<ChildLbState> removedChildren) {
-    // Do shutdowns after updating picker to reduce the chance of failing an RPC by picking a
-    // subchannel that has been shutdown.
-    for (ChildLbState childLbState : removedChildren) {
+    // Remaining entries in map are orphaned
+    for (ChildLbState childLbState : oldStatesMap.values()) {
       childLbState.shutdown();
     }
+    return status;
   }
 
   @Nullable
@@ -406,14 +386,4 @@
       return addrs.toString();
     }
   }
-
-  protected static class AcceptResolvedAddrRetVal {
-    public final Status status;
-    public final List<ChildLbState> removedChildren;
-
-    public AcceptResolvedAddrRetVal(Status status, List<ChildLbState> removedChildren) {
-      this.status = status;
-      this.removedChildren = removedChildren;
-    }
-  }
 }

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

@@ -87,62 +87,46 @@
       return addressValidityStatus;
     }
 
-    try {
-      resolvingAddresses = true;
-      AcceptResolvedAddrRetVal acceptRetVal = acceptResolvedAddressesInternal(resolvedAddresses);
-      if (!acceptRetVal.status.isOk()) {
-        return acceptRetVal.status;
-      }
-
-      // 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");
+    // 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(XdsAttributes.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;
       }
-      Map<EquivalentAddressGroup, Long> serverWeights = new HashMap<>();
-      long totalWeight = 0L;
-      for (EquivalentAddressGroup eag : addrList) {
-        Long weight = eag.getAttributes().get(XdsAttributes.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);
-        }
+      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(acceptRetVal.removedChildren);
-    } finally {
-      this.resolvingAddresses = false;
     }
-
-    return Status.OK;
+    // 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);
+
+    return super.acceptResolvedAddresses(resolvedAddresses);
   }
 
 

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

@@ -170,31 +170,17 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     }
     config =
             (WeightedRoundRobinLoadBalancerConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
-    AcceptResolvedAddrRetVal acceptRetVal;
-    try {
-      resolvingAddresses = true;
-      acceptRetVal = acceptResolvedAddressesInternal(resolvedAddresses);
-      if (!acceptRetVal.status.isOk()) {
-        return acceptRetVal.status;
-      }
-
-      if (weightUpdateTimer != null && weightUpdateTimer.isPending()) {
-        weightUpdateTimer.cancel();
-      }
-      updateWeightTask.run();
 
-      createAndApplyOrcaListeners();
+    if (weightUpdateTimer != null && weightUpdateTimer.isPending()) {
+      weightUpdateTimer.cancel();
+    }
+    updateWeightTask.run();
 
-      // 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();
+    Status status = super.acceptResolvedAddresses(resolvedAddresses);
 
-      shutdownRemoved(acceptRetVal.removedChildren);
-    } finally {
-      resolvingAddresses = false;
-    }
+    createAndApplyOrcaListeners();
 
-    return acceptRetVal.status;
+    return status;
   }
 
   /**