#20114

Committed 11 Dec 2025 04:45AM UTC coverage: 88.549% (-0.02%) from 88.569%

Build # #20114

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

35016 of 39544 relevant lines covered (88.55%)

0.89 hits per line

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

79.68

/../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 ChildLbStatusListener childLbStatusListener;
    private final ChildLoadBalancingPolicy childPolicy;
    private 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();
    }

    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, childLbStatusListener);
        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,
        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);
      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;
      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.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 ChildLbStatusListener childLbStatusListener;
214	private final ChildLoadBalancingPolicy childPolicy;
215	private ResolvedAddressFactory childLbResolvedAddressFactory;
216
217	public RefCountedChildPolicyWrapperFactory(
218	ChildLoadBalancingPolicy childPolicy,
219	ResolvedAddressFactory childLbResolvedAddressFactory,
220	ChildLoadBalancerHelperProvider childLbHelperProvider,
221	ChildLbStatusListener childLbStatusListener) {	1✔
222	this.childPolicy = checkNotNull(childPolicy, "childPolicy");	1✔
223	this.childLbResolvedAddressFactory =	1✔
224	checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");	1✔
225	this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");	1✔
226	this.childLbStatusListener = checkNotNull(childLbStatusListener, "childLbStatusListener");	1✔
227	}	1✔
228
229	void init() {
230	childLbHelperProvider.init();	1✔
231	}	1✔
232
233	Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
234	this.childLbResolvedAddressFactory = childLbResolvedAddressFactory;	1✔
235	Status status = Status.OK;	1✔
236	for (RefCountedChildPolicyWrapper wrapper : childPolicyMap.values()) {	1✔
237	Status newStatus =	1✔
238	wrapper.childPolicyWrapper.acceptResolvedAddressFactory(childLbResolvedAddressFactory);	1✔
239	if (!newStatus.isOk()) {	1✔
240	status = newStatus;	×
241	}
242	}	1✔
243	return status;	1✔
244	}
245
246	ChildPolicyWrapper createOrGet(String target) {
247	// TODO(creamsoup) check if the target is valid or not
248	RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);	1✔
249	if (pooledChildPolicyWrapper == null) {	1✔
250	ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(	1✔
251	target, childPolicy, childLbHelperProvider, childLbStatusListener);
252	pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);	1✔
253	childPolicyMap.put(target, pooledChildPolicyWrapper);	1✔
254	childPolicyWrapper.start(childLbResolvedAddressFactory);	1✔
255	return pooledChildPolicyWrapper.getObject();	1✔
256	} else {
257	ChildPolicyWrapper childPolicyWrapper = pooledChildPolicyWrapper.getObject();	1✔
258	if (childPolicyWrapper.getPicker() != null) {	1✔
259	childPolicyWrapper.refreshState();	1✔
260	}
261	return childPolicyWrapper;	1✔
262	}
263	}
264
265	List<ChildPolicyWrapper> createOrGet(List<String> targets) {
266	List<ChildPolicyWrapper> retVal = new ArrayList<>();	1✔
267	for (String target : targets) {	1✔
268	retVal.add(createOrGet(target));	1✔
269	}	1✔
270	return retVal;	1✔
271	}
272
273	void release(ChildPolicyWrapper childPolicyWrapper) {
274	checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
275	String target = childPolicyWrapper.getTarget();	1✔
276	RefCountedChildPolicyWrapper existing = childPolicyMap.get(target);	1✔
277	checkState(existing != null, "Cannot access already released object");	1✔
278	existing.returnObject(childPolicyWrapper);	1✔
279	if (existing.isReleased()) {	1✔
280	childPolicyMap.remove(target);	1✔
281	}
282	}	1✔
283	}
284
285	/**
286	* ChildPolicyWrapper is a wrapper class for child load balancing policy with associated helper /
287	* utility classes to manage the child policy.
288	*/
289	static final class ChildPolicyWrapper {
290
291	private final String target;
292	private final ChildPolicyReportingHelper helper;
293	private final LoadBalancer lb;
294	private final Object childLbConfig;
295	private volatile SubchannelPicker picker;
296	private ConnectivityState state;
297
298	public ChildPolicyWrapper(
299	String target,
300	ChildLoadBalancingPolicy childPolicy,
301	ChildLoadBalancerHelperProvider childLbHelperProvider,
302	ChildLbStatusListener childLbStatusListener) {	1✔
303	this.target = target;	1✔
304	this.helper =	1✔
305	new ChildPolicyReportingHelper(childLbHelperProvider, childLbStatusListener);
306	LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();	1✔
307	final ConfigOrError lbConfig =	1✔
308	lbProvider
309	.parseLoadBalancingPolicyConfig(	1✔
310	childPolicy.getEffectiveChildPolicy(target));	1✔
311	this.lb = lbProvider.newLoadBalancer(helper);	1✔
312	this.childLbConfig = lbConfig.getConfig();	1✔
313	helper.getChannelLogger().log(	1✔
314	ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", childLbConfig);
315	}	1✔
316
317	void start(ResolvedAddressFactory childLbResolvedAddressFactory) {
318	helper.getSynchronizationContext().execute(	1✔
319	new Runnable() {	1✔
320	@Override
321	public void run() {
322	if (!acceptResolvedAddressFactory(childLbResolvedAddressFactory).isOk()) {	1✔
323	helper.refreshNameResolution();	1✔
324	}
325	lb.requestConnection();	1✔
326	}	1✔
327	});
328	}	1✔
329
330	Status acceptResolvedAddressFactory(ResolvedAddressFactory childLbResolvedAddressFactory) {
331	helper.getSynchronizationContext().throwIfNotInThisSynchronizationContext();	1✔
332	return lb.acceptResolvedAddresses(childLbResolvedAddressFactory.create(childLbConfig));	1✔
333	}
334
335	String getTarget() {
336	return target;	1✔
337	}
338
339	SubchannelPicker getPicker() {
340	return picker;	1✔
341	}
342
343	ChildPolicyReportingHelper getHelper() {
344	return helper;	1✔
345	}
346
347	public ConnectivityState getState() {
348	return state;	1✔
349	}
350
351	void refreshState() {
352	helper.getSynchronizationContext().execute(	1✔
353	new Runnable() {	1✔
354	@Override
355	public void run() {
356	helper.updateBalancingState(state, picker);	1✔
357	}	1✔
358	}
359	);
360	}	1✔
361
362	void shutdown() {
363	helper.getSynchronizationContext().execute(	1✔
364	new Runnable() {	1✔
365	@Override
366	public void run() {
367	lb.shutdown();	1✔
368	}	1✔
369	}
370	);
371	}	1✔
372
373	@Override
374	public String toString() {
375	return MoreObjects.toStringHelper(this)	×
376	.add("target", target)	×
377	.add("picker", picker)	×
378	.add("state", state)	×
379	.toString();	×
380	}
381
382	/**
383	* A delegating {@link io.grpc.LoadBalancer.Helper} maintains status of {@link
384	* ChildPolicyWrapper} when {@link Subchannel} status changed. This helper is used between child
385	* policy and parent load-balancer where each picker in child policy is governed by a governing
386	* picker (RlsPicker). The governing picker will be reported back to the parent load-balancer.
387	*/
388	final class ChildPolicyReportingHelper extends ForwardingLoadBalancerHelper {
389
390	private final ChildLoadBalancerHelper delegate;
391	private final ChildLbStatusListener listener;
392
393	ChildPolicyReportingHelper(
394	ChildLoadBalancerHelperProvider childHelperProvider,
395	ChildLbStatusListener listener) {	1✔
396	checkNotNull(childHelperProvider, "childHelperProvider");	1✔
397	this.delegate = childHelperProvider.forTarget(getTarget());	1✔
398	this.listener = checkNotNull(listener, "listener");	1✔
399	}	1✔
400
401	@Override
402	protected Helper delegate() {
403	return delegate;	1✔
404	}
405
406	@Override
407	public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
408	picker = newPicker;	1✔
409	state = newState;	1✔
410	super.updateBalancingState(newState, newPicker);	1✔
411	listener.onStatusChanged(newState);	1✔
412	}	1✔
413	}
414	}
415
416	/** Listener for child lb status change events. */
417	interface ChildLbStatusListener {
418
419	/** Notifies when child lb status changes. */
420	void onStatusChanged(ConnectivityState newState);
421	}
422
423	private static final class RefCountedChildPolicyWrapper
424	implements ObjectPool<ChildPolicyWrapper> {
425
426	private long refCnt;
427	@Nullable
428	private ChildPolicyWrapper childPolicyWrapper;
429
430	private RefCountedChildPolicyWrapper(ChildPolicyWrapper childPolicyWrapper) {	1✔
431	this.childPolicyWrapper = checkNotNull(childPolicyWrapper, "childPolicyWrapper");	1✔
432	}	1✔
433
434	@Override
435	public ChildPolicyWrapper getObject() {
436	checkState(!isReleased(), "ChildPolicyWrapper is already released");	1✔
437	refCnt++;	1✔
438	return childPolicyWrapper;	1✔
439	}
440
441	@Override
442	@Nullable
443	public ChildPolicyWrapper returnObject(Object object) {
444	checkState(	1✔
445	!isReleased(),	1✔
446	"cannot return already released ChildPolicyWrapper, this is possibly a bug.");
447	checkState(	1✔
448	childPolicyWrapper == object,
449	"returned object doesn't match the pooled childPolicyWrapper");
450	long newCnt = --refCnt;	1✔
451	checkState(newCnt != -1, "Cannot return never pooled childPolicyWrapper");	1✔
452	if (newCnt == 0) {	1✔
453	childPolicyWrapper.shutdown();	1✔
454	childPolicyWrapper = null;	1✔
455	}
456	return null;	1✔
457	}
458
459	boolean isReleased() {
460	return childPolicyWrapper == null;	1✔
461	}
462
463	static RefCountedChildPolicyWrapper of(ChildPolicyWrapper childPolicyWrapper) {
464	return new RefCountedChildPolicyWrapper(childPolicyWrapper);	1✔
465	}
466
467	@Override
468	public String toString() {
469	return MoreObjects.toStringHelper(this)	×
470	.add("object", childPolicyWrapper)	×
471	.add("refCnt", refCnt)	×
472	.toString();	×
473	}
474	}
475
476	/** Exception thrown when attempting to parse child policy encountered parsing issue. */
477	static final class InvalidChildPolicyConfigException extends Exception {
478
479	private static final long serialVersionUID = 0L;
480
481	InvalidChildPolicyConfigException(String message) {
482	super(message);	1✔
483	}	1✔
484
485	@Override
486	public synchronized Throwable fillInStackTrace() {
487	// no stack trace above this point
488	return this;	1✔
489	}
490	}
491	}

grpc / grpc-java / #20114

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