#20104

Committed 04 Dec 2025 03:25PM UTC coverage: 88.647% (-0.005%) from 88.652%

Build # #20104

Build Type

push

github

Committed by

ejona86

Commit Message

rls: Avoid missed config update from reentrancy

Since ChildPolicyWrapper() called into the child before
childPolicyMap.put(), it is possible for that child to call back into
RLS and further update state without that child being known. When CDS
is_dynamic=true (since ca99a8c47), it registers the cluster with
XdsDependencyManager, which adds a watch to XdsClient. If XdsClient
already has the results cached then the watch callback can be enqueued
immediately onto the syncContext and execute still within the
constructor.

Calling into the child with the lock held isn't great, as it allows for
this type of reentrancy bug. But that'll take larger changes to fix.

b/464116731

Run Details

35162 of 39665 relevant lines covered (88.65%)

0.89 hits per line

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

79.35

/../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.Status;
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 ChildLoadBalancingPolicy childPolicy;
    private ResolvedAddressFactory childLbResolvedAddressFactory;

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

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

    Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
      this.childLbResolvedAddressFactory = childLbResolvedAddressFactory;
      Status status = Status.OK;
      for (RefCountedChildPolicyWrapper wrapper : childPolicyMap.values()) {
        Status newStatus =
            wrapper.childPolicyWrapper.acceptResolvedAddressFactory(childLbResolvedAddressFactory);
        if (!newStatus.isOk()) {
          status = newStatus;
        }
      }
      return status;
    }

    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, childLbHelperProvider);
        pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);
        childPolicyMap.put(target, pooledChildPolicyWrapper);
        childPolicyWrapper.start(childLbResolvedAddressFactory);
        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 final Object childLbConfig;
    private volatile SubchannelPicker picker;
    private ConnectivityState state;

    public ChildPolicyWrapper(
        String target,
        ChildLoadBalancingPolicy childPolicy,
        ChildLoadBalancerHelperProvider childLbHelperProvider) {
      this.target = target;
      this.helper = new ChildPolicyReportingHelper(childLbHelperProvider);
      LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();
      final ConfigOrError lbConfig =
          lbProvider
              .parseLoadBalancingPolicyConfig(
                  childPolicy.getEffectiveChildPolicy(target));
      this.lb = lbProvider.newLoadBalancer(helper);
      this.childLbConfig = lbConfig.getConfig();
      helper.getChannelLogger().log(
          ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", childLbConfig);
    }

    void start(ResolvedAddressFactory childLbResolvedAddressFactory) {
      helper.getSynchronizationContext().execute(
          new Runnable() {
            @Override
            public void run() {
              if (!acceptResolvedAddressFactory(childLbResolvedAddressFactory).isOk()) {
                helper.refreshNameResolution();
              }
              lb.requestConnection();
            }
          });
    }

    Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
      helper.getSynchronizationContext().throwIfNotInThisSynchronizationContext();
      return lb.acceptResolvedAddresses(childLbResolvedAddressFactory.create(childLbConfig));
    }

    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;

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

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

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

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

grpc / grpc-java / #20104

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