#19908

Committed 16 Jul 2025 07:54PM UTC coverage: 88.593% (+0.07%) from 88.528%

Build # #19908

Build Type

push

github

Committed by

ejona86

Commit Message

Revert "xds: Convert CdsLb to XdsDepManager"

This reverts commit 297ab05ef.

b/430347751 shows multiple concerning behaviors in the xDS stack with
the new A74 config update model. XdsDepManager and CdsLB2 still seem to
be working correctly, but the change is exacerbated issues in other
parts of the stack, like RingHashConfig not having equals fixed in
a8de9f07ab.

Revert only for the v1.74.x release, leaving it on master.

Run Details

34647 of 39108 relevant lines covered (88.59%)

0.89 hits per line

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

78.86

/../rls/src/main/java/io/grpc/rls/LbPolicyConfiguration.java

/*
 * Copyright 2020 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.rls;

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 com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import io.grpc.ChannelLogger.ChannelLogLevel;
import io.grpc.ConnectivityState;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer.Subchannel;
import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.NameResolver.ConfigOrError;
import io.grpc.internal.ObjectPool;
import io.grpc.rls.ChildLoadBalancerHelper.ChildLoadBalancerHelperProvider;
import io.grpc.rls.RlsProtoData.RouteLookupConfig;
import io.grpc.util.ForwardingLoadBalancerHelper;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import javax.annotation.Nullable;

/** Configuration for RLS load balancing policy. */
final class LbPolicyConfiguration {

  private final RouteLookupConfig routeLookupConfig;
  @Nullable
  private final Map<String, ?> routeLookupChannelServiceConfig;
  private final ChildLoadBalancingPolicy policy;

  LbPolicyConfiguration(
      RouteLookupConfig routeLookupConfig, @Nullable Map<String, ?> routeLookupChannelServiceConfig,
      ChildLoadBalancingPolicy policy) {
    this.routeLookupConfig = checkNotNull(routeLookupConfig, "routeLookupConfig");
    this.routeLookupChannelServiceConfig = routeLookupChannelServiceConfig;
    this.policy = checkNotNull(policy, "policy");
  }

  RouteLookupConfig getRouteLookupConfig() {
    return routeLookupConfig;
  }

  @Nullable
  Map<String, ?> getRouteLookupChannelServiceConfig() {
    return routeLookupChannelServiceConfig;
  }

  ChildLoadBalancingPolicy getLoadBalancingPolicy() {
    return policy;
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
    LbPolicyConfiguration that = (LbPolicyConfiguration) o;
    return Objects.equals(routeLookupConfig, that.routeLookupConfig)
        && Objects.equals(routeLookupChannelServiceConfig, that.routeLookupChannelServiceConfig)
        && Objects.equals(policy, that.policy);
  }

  @Override
  public int hashCode() {
    return Objects.hash(routeLookupConfig, routeLookupChannelServiceConfig, policy);
  }

  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
        .add("routeLookupConfig", routeLookupConfig)
        .add("routeLookupChannelServiceConfig", routeLookupChannelServiceConfig)
        .add("policy", policy)
        .toString();
  }

  /** ChildLoadBalancingPolicy is an elected child policy to delegate requests. */
  static final class ChildLoadBalancingPolicy {

    private final Map<String, Object> effectiveRawChildPolicy;
    private final LoadBalancerProvider effectiveLbProvider;
    private final String targetFieldName;

    @VisibleForTesting
    ChildLoadBalancingPolicy(
        String targetFieldName,
        Map<String, Object> effectiveRawChildPolicy,
        LoadBalancerProvider effectiveLbProvider) {
      checkArgument(
          targetFieldName != null && !targetFieldName.isEmpty(),
          "targetFieldName cannot be empty or null");
      this.targetFieldName = targetFieldName;
      this.effectiveRawChildPolicy =
          checkNotNull(effectiveRawChildPolicy, "effectiveRawChildPolicy");
      this.effectiveLbProvider = checkNotNull(effectiveLbProvider, "effectiveLbProvider");
    }

    /** Creates ChildLoadBalancingPolicy. */
    @SuppressWarnings("unchecked")
    static ChildLoadBalancingPolicy create(
        String childPolicyConfigTargetFieldName, List<Map<String, ?>> childPolicies)
        throws InvalidChildPolicyConfigException {
      Map<String, Object> effectiveChildPolicy = null;
      LoadBalancerProvider effectiveLbProvider = null;
      List<String> policyTried = new ArrayList<>();

      LoadBalancerRegistry lbRegistry = LoadBalancerRegistry.getDefaultRegistry();
      for (Map<String, ?> childPolicy : childPolicies) {
        if (childPolicy.isEmpty()) {
          continue;
        }
        if (childPolicy.size() != 1) {
          throw
              new InvalidChildPolicyConfigException(
                  "childPolicy should have exactly one loadbalancing policy");
        }
        String policyName = childPolicy.keySet().iterator().next();
        LoadBalancerProvider provider = lbRegistry.getProvider(policyName);
        if (provider != null) {
          effectiveLbProvider = provider;
          effectiveChildPolicy = Collections.unmodifiableMap(childPolicy);
          break;
        }
        policyTried.add(policyName);
      }
      if (effectiveChildPolicy == null) {
        throw
            new InvalidChildPolicyConfigException(
                String.format("no valid childPolicy found, policy tried: %s", policyTried));
      }
      return
          new ChildLoadBalancingPolicy(
              childPolicyConfigTargetFieldName,
              (Map<String, Object>) effectiveChildPolicy.values().iterator().next(),
              effectiveLbProvider);
    }

    /** Creates a child load balancer config for given target from elected raw child policy. */
    Map<String, ?> getEffectiveChildPolicy(String target) {
      Map<String, Object> childPolicy = new HashMap<>(effectiveRawChildPolicy);
      childPolicy.put(targetFieldName, target);
      return childPolicy;
    }

    /** Returns the elected child {@link LoadBalancerProvider}. */
    LoadBalancerProvider getEffectiveLbProvider() {
      return effectiveLbProvider;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (o == null || getClass() != o.getClass()) {
        return false;
      }
      ChildLoadBalancingPolicy that = (ChildLoadBalancingPolicy) o;
      return Objects.equals(effectiveRawChildPolicy, that.effectiveRawChildPolicy)
          && Objects.equals(effectiveLbProvider, that.effectiveLbProvider)
          && Objects.equals(targetFieldName, that.targetFieldName);
    }

    @Override
    public int hashCode() {
      return Objects.hash(effectiveRawChildPolicy, effectiveLbProvider, targetFieldName);
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
          .add("effectiveRawChildPolicy", effectiveRawChildPolicy)
          .add("effectiveLbProvider", effectiveLbProvider)
          .add("childPolicyConfigTargetFieldName", targetFieldName)
          .toString();
    }
  }

  /** Factory for {@link ChildPolicyWrapper}. Not thread-safe. */
  static final class RefCountedChildPolicyWrapperFactory {
    @VisibleForTesting
    final Map<String /* target */, RefCountedChildPolicyWrapper> childPolicyMap =
        new HashMap<>();

    private final ChildLoadBalancerHelperProvider childLbHelperProvider;
    private final ChildLbStatusListener childLbStatusListener;
    private final ChildLoadBalancingPolicy childPolicy;
    private final ResolvedAddressFactory childLbResolvedAddressFactory;

    public RefCountedChildPolicyWrapperFactory(
        ChildLoadBalancingPolicy childPolicy,
        ResolvedAddressFactory childLbResolvedAddressFactory,
        ChildLoadBalancerHelperProvider childLbHelperProvider,
        ChildLbStatusListener childLbStatusListener) {
      this.childPolicy = checkNotNull(childPolicy, "childPolicy");
      this.childLbResolvedAddressFactory =
          checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");
      this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");
      this.childLbStatusListener = checkNotNull(childLbStatusListener, "childLbStatusListener");
    }

    void init() {
      childLbHelperProvider.init();
    }

    ChildPolicyWrapper createOrGet(String target) {
      // TODO(creamsoup) check if the target is valid or not
      RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);
      if (pooledChildPolicyWrapper == null) {
        ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(
            target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider,
            childLbStatusListener);
        pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);
        childPolicyMap.put(target, pooledChildPolicyWrapper);
        return pooledChildPolicyWrapper.getObject();
      } else {
        ChildPolicyWrapper childPolicyWrapper = pooledChildPolicyWrapper.getObject();
        if (childPolicyWrapper.getPicker() != null) {
          childPolicyWrapper.refreshState();
        }
        return childPolicyWrapper;
      }
    }

    List<ChildPolicyWrapper> createOrGet(List<String> targets) {
      List<ChildPolicyWrapper> retVal = new ArrayList<>();
      for (String target : targets) {
        retVal.add(createOrGet(target));
      }
      return retVal;
    }

    void release(ChildPolicyWrapper childPolicyWrapper) {
      checkNotNull(childPolicyWrapper, "childPolicyWrapper");
      String target = childPolicyWrapper.getTarget();
      RefCountedChildPolicyWrapper existing = childPolicyMap.get(target);
      checkState(existing != null, "Cannot access already released object");
      existing.returnObject(childPolicyWrapper);
      if (existing.isReleased()) {
        childPolicyMap.remove(target);
      }
    }
  }

  /**
   * ChildPolicyWrapper is a wrapper class for child load balancing policy with associated helper /
   * utility classes to manage the child policy.
   */
  static final class ChildPolicyWrapper {

    private final String target;
    private final ChildPolicyReportingHelper helper;
    private final LoadBalancer lb;
    private volatile SubchannelPicker picker;
    private ConnectivityState state;

    public ChildPolicyWrapper(
        String target,
        ChildLoadBalancingPolicy childPolicy,
        final ResolvedAddressFactory childLbResolvedAddressFactory,
        ChildLoadBalancerHelperProvider childLbHelperProvider,
        ChildLbStatusListener childLbStatusListener) {
      this.target = target;
      this.helper =
          new ChildPolicyReportingHelper(childLbHelperProvider, childLbStatusListener);
      LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();
      final ConfigOrError lbConfig =
          lbProvider
              .parseLoadBalancingPolicyConfig(
                  childPolicy.getEffectiveChildPolicy(target));
      this.lb = lbProvider.newLoadBalancer(helper);
      helper.getChannelLogger().log(
          ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", lbConfig.getConfig());
      helper.getSynchronizationContext().execute(
          new Runnable() {
            @Override
            public void run() {
              if (!lb.acceptResolvedAddresses(
                  childLbResolvedAddressFactory.create(lbConfig.getConfig())).isOk()) {
                helper.refreshNameResolution();
              }
              lb.requestConnection();
            }
          });
    }

    String getTarget() {
      return target;
    }

    SubchannelPicker getPicker() {
      return picker;
    }

    ChildPolicyReportingHelper getHelper() {
      return helper;
    }

    public ConnectivityState getState() {
      return state;
    }

    void refreshState() {
      helper.getSynchronizationContext().execute(
          new Runnable() {
            @Override
            public void run() {
              helper.updateBalancingState(state, picker);
            }
          }
      );
    }

    void shutdown() {
      helper.getSynchronizationContext().execute(
          new Runnable() {
            @Override
            public void run() {
              lb.shutdown();
            }
          }
      );
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
          .add("target", target)
          .add("picker", picker)
          .add("state", state)
          .toString();
    }

    /**
     * A delegating {@link io.grpc.LoadBalancer.Helper} maintains status of {@link
     * ChildPolicyWrapper} when {@link Subchannel} status changed. This helper is used between child
     * policy and parent load-balancer where each picker in child policy is governed by a governing
     * picker (RlsPicker). The governing picker will be reported back to the parent load-balancer.
     */
    final class ChildPolicyReportingHelper extends ForwardingLoadBalancerHelper {

      private final ChildLoadBalancerHelper delegate;
      private final ChildLbStatusListener listener;

      ChildPolicyReportingHelper(
          ChildLoadBalancerHelperProvider childHelperProvider,
          ChildLbStatusListener listener) {
        checkNotNull(childHelperProvider, "childHelperProvider");
        this.delegate = childHelperProvider.forTarget(getTarget());
        this.listener = checkNotNull(listener, "listener");
      }

      @Override
      protected Helper delegate() {
        return delegate;
      }

      @Override
      public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
        picker = newPicker;
        state = newState;
        super.updateBalancingState(newState, newPicker);
        listener.onStatusChanged(newState);
      }
    }
  }

  /** Listener for child lb status change events. */
  interface ChildLbStatusListener {

    /** Notifies when child lb status changes. */
    void onStatusChanged(ConnectivityState newState);
  }

  private static final class RefCountedChildPolicyWrapper
      implements ObjectPool<ChildPolicyWrapper> {

    private long refCnt;
    @Nullable
    private ChildPolicyWrapper childPolicyWrapper;

    private RefCountedChildPolicyWrapper(ChildPolicyWrapper childPolicyWrapper) {
      this.childPolicyWrapper = checkNotNull(childPolicyWrapper, "childPolicyWrapper");
    }

    @Override
    public ChildPolicyWrapper getObject() {
      checkState(!isReleased(), "ChildPolicyWrapper is already released");
      refCnt++;
      return childPolicyWrapper;
    }

    @Override
    @Nullable
    public ChildPolicyWrapper returnObject(Object object) {
      checkState(
          !isReleased(),
          "cannot return already released ChildPolicyWrapper, this is possibly a bug.");
      checkState(
          childPolicyWrapper == object,
          "returned object doesn't match the pooled childPolicyWrapper");
      long newCnt = --refCnt;
      checkState(newCnt != -1, "Cannot return never pooled childPolicyWrapper");
      if (newCnt == 0) {
        childPolicyWrapper.shutdown();
        childPolicyWrapper = null;
      }
      return null;
    }

    boolean isReleased() {
      return childPolicyWrapper == null;
    }

    static RefCountedChildPolicyWrapper of(ChildPolicyWrapper childPolicyWrapper) {
      return new RefCountedChildPolicyWrapper(childPolicyWrapper);
    }

    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
          .add("object", childPolicyWrapper)
          .add("refCnt", refCnt)
          .toString();
    }
  }

  /** Exception thrown when attempting to parse child policy encountered parsing issue. */
  static final class InvalidChildPolicyConfigException extends Exception {

    private static final long serialVersionUID = 0L;

    InvalidChildPolicyConfigException(String message) {
      super(message);
    }

    @Override
    public synchronized Throwable fillInStackTrace() {
      // no stack trace above this point
      return this;
    }
  }
}

1	/*
2	* Copyright 2020 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.rls;
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
23	import com.google.common.annotations.VisibleForTesting;
24	import com.google.common.base.MoreObjects;
25	import io.grpc.ChannelLogger.ChannelLogLevel;
26	import io.grpc.ConnectivityState;
27	import io.grpc.LoadBalancer;
28	import io.grpc.LoadBalancer.Helper;
29	import io.grpc.LoadBalancer.Subchannel;
30	import io.grpc.LoadBalancer.SubchannelPicker;
31	import io.grpc.LoadBalancerProvider;
32	import io.grpc.LoadBalancerRegistry;
33	import io.grpc.NameResolver.ConfigOrError;
34	import io.grpc.internal.ObjectPool;
35	import io.grpc.rls.ChildLoadBalancerHelper.ChildLoadBalancerHelperProvider;
36	import io.grpc.rls.RlsProtoData.RouteLookupConfig;
37	import io.grpc.util.ForwardingLoadBalancerHelper;
38	import java.util.ArrayList;
39	import java.util.Collections;
40	import java.util.HashMap;
41	import java.util.List;
42	import java.util.Map;
43	import java.util.Objects;
44	import javax.annotation.Nullable;
45
46	/** Configuration for RLS load balancing policy. */
47	final class LbPolicyConfiguration {
48
49	private final RouteLookupConfig routeLookupConfig;
50	@Nullable
51	private final Map<String, ?> routeLookupChannelServiceConfig;
52	private final ChildLoadBalancingPolicy policy;
53
54	LbPolicyConfiguration(
55	RouteLookupConfig routeLookupConfig, @Nullable Map<String, ?> routeLookupChannelServiceConfig,
56	ChildLoadBalancingPolicy policy) {	1✔
57	this.routeLookupConfig = checkNotNull(routeLookupConfig, "routeLookupConfig");	1✔
58	this.routeLookupChannelServiceConfig = routeLookupChannelServiceConfig;	1✔
59	this.policy = checkNotNull(policy, "policy");	1✔
60	}	1✔
61
62	RouteLookupConfig getRouteLookupConfig() {
63	return routeLookupConfig;	1✔
64	}
65
66	@Nullable
67	Map<String, ?> getRouteLookupChannelServiceConfig() {
68	return routeLookupChannelServiceConfig;	1✔
69	}
70
71	ChildLoadBalancingPolicy getLoadBalancingPolicy() {
72	return policy;	1✔
73	}
74
75	@Override
76	public boolean equals(Object o) {
77	if (this == o) {	1✔
78	return true;	×
79	}
80	if (o == null \|\| getClass() != o.getClass()) {	1✔
81	return false;	1✔
82	}
83	LbPolicyConfiguration that = (LbPolicyConfiguration) o;	×
84	return Objects.equals(routeLookupConfig, that.routeLookupConfig)	×
85	&& Objects.equals(routeLookupChannelServiceConfig, that.routeLookupChannelServiceConfig)	×
86	&& Objects.equals(policy, that.policy);	×
87	}
88
89	@Override
90	public int hashCode() {
91	return Objects.hash(routeLookupConfig, routeLookupChannelServiceConfig, policy);	×
92	}
93
94	@Override
95	public String toString() {
96	return MoreObjects.toStringHelper(this)	×
97	.add("routeLookupConfig", routeLookupConfig)	×
98	.add("routeLookupChannelServiceConfig", routeLookupChannelServiceConfig)	×
99	.add("policy", policy)	×
100	.toString();	×
101	}
102
103	/** ChildLoadBalancingPolicy is an elected child policy to delegate requests. */
104	static final class ChildLoadBalancingPolicy {
105
106	private final Map<String, Object> effectiveRawChildPolicy;
107	private final LoadBalancerProvider effectiveLbProvider;
108	private final String targetFieldName;
109
110	@VisibleForTesting
111	ChildLoadBalancingPolicy(
112	String targetFieldName,
113	Map<String, Object> effectiveRawChildPolicy,
114	LoadBalancerProvider effectiveLbProvider) {	1✔
115	checkArgument(	1✔
116	targetFieldName != null && !targetFieldName.isEmpty(),	1✔
117	"targetFieldName cannot be empty or null");
118	this.targetFieldName = targetFieldName;	1✔
119	this.effectiveRawChildPolicy =	1✔
120	checkNotNull(effectiveRawChildPolicy, "effectiveRawChildPolicy");	1✔
121	this.effectiveLbProvider = checkNotNull(effectiveLbProvider, "effectiveLbProvider");	1✔
122	}	1✔
123
124	/** Creates ChildLoadBalancingPolicy. */
125	@SuppressWarnings("unchecked")
126	static ChildLoadBalancingPolicy create(
127	String childPolicyConfigTargetFieldName, List<Map<String, ?>> childPolicies)
128	throws InvalidChildPolicyConfigException {
129	Map<String, Object> effectiveChildPolicy = null;	1✔
130	LoadBalancerProvider effectiveLbProvider = null;	1✔
131	List<String> policyTried = new ArrayList<>();	1✔
132
133	LoadBalancerRegistry lbRegistry = LoadBalancerRegistry.getDefaultRegistry();	1✔
134	for (Map<String, ?> childPolicy : childPolicies) {	1✔
135	if (childPolicy.isEmpty()) {	1✔
136	continue;	×
137	}
138	if (childPolicy.size() != 1) {	1✔
139	throw	1✔
140	new InvalidChildPolicyConfigException(
141	"childPolicy should have exactly one loadbalancing policy");
142	}
143	String policyName = childPolicy.keySet().iterator().next();	1✔
144	LoadBalancerProvider provider = lbRegistry.getProvider(policyName);	1✔
145	if (provider != null) {	1✔
146	effectiveLbProvider = provider;	1✔
147	effectiveChildPolicy = Collections.unmodifiableMap(childPolicy);	1✔
148	break;	1✔
149	}
150	policyTried.add(policyName);	×
151	}	×
152	if (effectiveChildPolicy == null) {	1✔
153	throw	1✔
154	new InvalidChildPolicyConfigException(
155	String.format("no valid childPolicy found, policy tried: %s", policyTried));	1✔
156	}
157	return	1✔
158	new ChildLoadBalancingPolicy(
159	childPolicyConfigTargetFieldName,
160	(Map<String, Object>) effectiveChildPolicy.values().iterator().next(),	1✔
161	effectiveLbProvider);
162	}
163
164	/** Creates a child load balancer config for given target from elected raw child policy. */
165	Map<String, ?> getEffectiveChildPolicy(String target) {
166	Map<String, Object> childPolicy = new HashMap<>(effectiveRawChildPolicy);	1✔
167	childPolicy.put(targetFieldName, target);	1✔
168	return childPolicy;	1✔
169	}
170
171	/** Returns the elected child {@link LoadBalancerProvider}. */
172	LoadBalancerProvider getEffectiveLbProvider() {
173	return effectiveLbProvider;	1✔
174	}
175
176	@Override
177	public boolean equals(Object o) {
178	if (this == o) {	×
179	return true;	×
180	}
181	if (o == null \|\| getClass() != o.getClass()) {	×
182	return false;	×
183	}
184	ChildLoadBalancingPolicy that = (ChildLoadBalancingPolicy) o;	×
185	return Objects.equals(effectiveRawChildPolicy, that.effectiveRawChildPolicy)	×
186	&& Objects.equals(effectiveLbProvider, that.effectiveLbProvider)	×
187	&& Objects.equals(targetFieldName, that.targetFieldName);	×
188	}
189
190	@Override
191	public int hashCode() {
192	return Objects.hash(effectiveRawChildPolicy, effectiveLbProvider, targetFieldName);	×
193	}
194
195	@Override
196	public String toString() {
197	return MoreObjects.toStringHelper(this)	×
198	.add("effectiveRawChildPolicy", effectiveRawChildPolicy)	×
199	.add("effectiveLbProvider", effectiveLbProvider)	×
200	.add("childPolicyConfigTargetFieldName", targetFieldName)	×
201	.toString();	×
202	}
203	}
204
205	/** Factory for {@link ChildPolicyWrapper}. Not thread-safe. */
206	static final class RefCountedChildPolicyWrapperFactory {
207	@VisibleForTesting	1✔
208	final Map<String /* target */, RefCountedChildPolicyWrapper> childPolicyMap =
209	new HashMap<>();
210
211	private final ChildLoadBalancerHelperProvider childLbHelperProvider;
212	private final ChildLbStatusListener childLbStatusListener;
213	private final ChildLoadBalancingPolicy childPolicy;
214	private final ResolvedAddressFactory childLbResolvedAddressFactory;
215
216	public RefCountedChildPolicyWrapperFactory(
217	ChildLoadBalancingPolicy childPolicy,
218	ResolvedAddressFactory childLbResolvedAddressFactory,
219	ChildLoadBalancerHelperProvider childLbHelperProvider,
220	ChildLbStatusListener childLbStatusListener) {	1✔
221	this.childPolicy = checkNotNull(childPolicy, "childPolicy");	1✔
222	this.childLbResolvedAddressFactory =	1✔
223	checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");	1✔
224	this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");	1✔
225	this.childLbStatusListener = checkNotNull(childLbStatusListener, "childLbStatusListener");	1✔
226	}	1✔
227
228	void init() {
229	childLbHelperProvider.init();	1✔
230	}	1✔
231
232	ChildPolicyWrapper createOrGet(String target) {
233	// TODO(creamsoup) check if the target is valid or not
234	RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);	1✔
235	if (pooledChildPolicyWrapper == null) {	1✔
236	ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(	1✔
237	target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider,
238	childLbStatusListener);
239	pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);	1✔
240	childPolicyMap.put(target, pooledChildPolicyWrapper);	1✔
241	return pooledChildPolicyWrapper.getObject();	1✔
242	} else {
243	ChildPolicyWrapper childPolicyWrapper = pooledChildPolicyWrapper.getObject();	1✔
244	if (childPolicyWrapper.getPicker() != null) {	1✔
245	childPolicyWrapper.refreshState();	1✔
246	}
247	return childPolicyWrapper;	1✔
248	}
249	}
250
251	List<ChildPolicyWrapper> createOrGet(List<String> targets) {
252	List<ChildPolicyWrapper> retVal = new ArrayList<>();	1✔
253	for (String target : targets) {	1✔
254	retVal.add(createOrGet(target));	1✔
255	}	1✔
256	return retVal;	1✔
257	}
258
259	void release(ChildPolicyWrapper childPolicyWrapper) {
260	checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
261	String target = childPolicyWrapper.getTarget();	1✔
262	RefCountedChildPolicyWrapper existing = childPolicyMap.get(target);	1✔
263	checkState(existing != null, "Cannot access already released object");	1✔
264	existing.returnObject(childPolicyWrapper);	1✔
265	if (existing.isReleased()) {	1✔
266	childPolicyMap.remove(target);	1✔
267	}
268	}	1✔
269	}
270
271	/**
272	* ChildPolicyWrapper is a wrapper class for child load balancing policy with associated helper /
273	* utility classes to manage the child policy.
274	*/
275	static final class ChildPolicyWrapper {
276
277	private final String target;
278	private final ChildPolicyReportingHelper helper;
279	private final LoadBalancer lb;
280	private volatile SubchannelPicker picker;
281	private ConnectivityState state;
282
283	public ChildPolicyWrapper(
284	String target,
285	ChildLoadBalancingPolicy childPolicy,
286	final ResolvedAddressFactory childLbResolvedAddressFactory,
287	ChildLoadBalancerHelperProvider childLbHelperProvider,
288	ChildLbStatusListener childLbStatusListener) {	1✔
289	this.target = target;	1✔
290	this.helper =	1✔
291	new ChildPolicyReportingHelper(childLbHelperProvider, childLbStatusListener);
292	LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();	1✔
293	final ConfigOrError lbConfig =	1✔
294	lbProvider
295	.parseLoadBalancingPolicyConfig(	1✔
296	childPolicy.getEffectiveChildPolicy(target));	1✔
297	this.lb = lbProvider.newLoadBalancer(helper);	1✔
298	helper.getChannelLogger().log(	1✔
299	ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", lbConfig.getConfig());	1✔
300	helper.getSynchronizationContext().execute(	1✔
301	new Runnable() {	1✔
302	@Override
303	public void run() {
304	if (!lb.acceptResolvedAddresses(	1✔
305	childLbResolvedAddressFactory.create(lbConfig.getConfig())).isOk()) {	1✔
306	helper.refreshNameResolution();	1✔
307	}
308	lb.requestConnection();	1✔
309	}	1✔
310	});
311	}	1✔
312
313	String getTarget() {
314	return target;	1✔
315	}
316
317	SubchannelPicker getPicker() {
318	return picker;	1✔
319	}
320
321	ChildPolicyReportingHelper getHelper() {
322	return helper;	1✔
323	}
324
325	public ConnectivityState getState() {
326	return state;	1✔
327	}
328
329	void refreshState() {
330	helper.getSynchronizationContext().execute(	1✔
331	new Runnable() {	1✔
332	@Override
333	public void run() {
334	helper.updateBalancingState(state, picker);	1✔
335	}	1✔
336	}
337	);
338	}	1✔
339
340	void shutdown() {
341	helper.getSynchronizationContext().execute(	1✔
342	new Runnable() {	1✔
343	@Override
344	public void run() {
345	lb.shutdown();	1✔
346	}	1✔
347	}
348	);
349	}	1✔
350
351	@Override
352	public String toString() {
353	return MoreObjects.toStringHelper(this)	×
354	.add("target", target)	×
355	.add("picker", picker)	×
356	.add("state", state)	×
357	.toString();	×
358	}
359
360	/**
361	* A delegating {@link io.grpc.LoadBalancer.Helper} maintains status of {@link
362	* ChildPolicyWrapper} when {@link Subchannel} status changed. This helper is used between child
363	* policy and parent load-balancer where each picker in child policy is governed by a governing
364	* picker (RlsPicker). The governing picker will be reported back to the parent load-balancer.
365	*/
366	final class ChildPolicyReportingHelper extends ForwardingLoadBalancerHelper {
367
368	private final ChildLoadBalancerHelper delegate;
369	private final ChildLbStatusListener listener;
370
371	ChildPolicyReportingHelper(
372	ChildLoadBalancerHelperProvider childHelperProvider,
373	ChildLbStatusListener listener) {	1✔
374	checkNotNull(childHelperProvider, "childHelperProvider");	1✔
375	this.delegate = childHelperProvider.forTarget(getTarget());	1✔
376	this.listener = checkNotNull(listener, "listener");	1✔
377	}	1✔
378
379	@Override
380	protected Helper delegate() {
381	return delegate;	1✔
382	}
383
384	@Override
385	public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
386	picker = newPicker;	1✔
387	state = newState;	1✔
388	super.updateBalancingState(newState, newPicker);	1✔
389	listener.onStatusChanged(newState);	1✔
390	}	1✔
391	}
392	}
393
394	/** Listener for child lb status change events. */
395	interface ChildLbStatusListener {
396
397	/** Notifies when child lb status changes. */
398	void onStatusChanged(ConnectivityState newState);
399	}
400
401	private static final class RefCountedChildPolicyWrapper
402	implements ObjectPool<ChildPolicyWrapper> {
403
404	private long refCnt;
405	@Nullable
406	private ChildPolicyWrapper childPolicyWrapper;
407
408	private RefCountedChildPolicyWrapper(ChildPolicyWrapper childPolicyWrapper) {	1✔
409	this.childPolicyWrapper = checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
410	}	1✔
411
412	@Override
413	public ChildPolicyWrapper getObject() {
414	checkState(!isReleased(), "ChildPolicyWrapper is already released");	1✔
415	refCnt++;	1✔
416	return childPolicyWrapper;	1✔
417	}
418
419	@Override
420	@Nullable
421	public ChildPolicyWrapper returnObject(Object object) {
422	checkState(	1✔
423	!isReleased(),	1✔
424	"cannot return already released ChildPolicyWrapper, this is possibly a bug.");
425	checkState(	1✔
426	childPolicyWrapper == object,
427	"returned object doesn't match the pooled childPolicyWrapper");
428	long newCnt = --refCnt;	1✔
429	checkState(newCnt != -1, "Cannot return never pooled childPolicyWrapper");	1✔
430	if (newCnt == 0) {	1✔
431	childPolicyWrapper.shutdown();	1✔
432	childPolicyWrapper = null;	1✔
433	}
434	return null;	1✔
435	}
436
437	boolean isReleased() {
438	return childPolicyWrapper == null;	1✔
439	}
440
441	static RefCountedChildPolicyWrapper of(ChildPolicyWrapper childPolicyWrapper) {
442	return new RefCountedChildPolicyWrapper(childPolicyWrapper);	1✔
443	}
444
445	@Override
446	public String toString() {
447	return MoreObjects.toStringHelper(this)	×
448	.add("object", childPolicyWrapper)	×
449	.add("refCnt", refCnt)	×
450	.toString();	×
451	}
452	}
453
454	/** Exception thrown when attempting to parse child policy encountered parsing issue. */
455	static final class InvalidChildPolicyConfigException extends Exception {
456
457	private static final long serialVersionUID = 0L;
458
459	InvalidChildPolicyConfigException(String message) {
460	super(message);	1✔
461	}	1✔
462
463	@Override
464	public synchronized Throwable fillInStackTrace() {
465	// no stack trace above this point
466	return this;	1✔
467	}
468	}
469	}

grpc / grpc-java / #19908

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