Skip to content

Commit f7e6d69

Browse files
dapengzhang0dapengzhang0
committed
rls: fix child lb leak when client channel is shutdown (#8750)
When client channel is shutting down, the RlsLoadBalancer is shutting down. However, the child loadbalancers of RlsLoadBalancer are not shut down. This is causing the issue b/209831670
1 parent 69671e1 commit f7e6d69

File tree

7 files changed

+147
-74
lines changed

7 files changed

+147
-74
lines changed

‎rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java‎

Lines changed: 13 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -30,17 +30,14 @@
3030
import io.grpc.ChannelLogger;
3131
import io.grpc.ChannelLogger.ChannelLogLevel;
3232
import io.grpc.ConnectivityState;
33-
import io.grpc.LoadBalancer;
3433
import io.grpc.LoadBalancer.Helper;
3534
import io.grpc.LoadBalancer.PickResult;
3635
import io.grpc.LoadBalancer.PickSubchannelArgs;
3736
import io.grpc.LoadBalancer.ResolvedAddresses;
3837
import io.grpc.LoadBalancer.SubchannelPicker;
39-
import io.grpc.LoadBalancerProvider;
4038
import io.grpc.ManagedChannel;
4139
import io.grpc.ManagedChannelBuilder;
4240
import io.grpc.Metadata;
43-
import io.grpc.NameResolver.ConfigOrError;
4441
import io.grpc.Status;
4542
import io.grpc.SynchronizationContext;
4643
import io.grpc.SynchronizationContext.ScheduledHandle;
@@ -51,7 +48,6 @@
5148
import io.grpc.lookup.v1.RouteLookupServiceGrpc.RouteLookupServiceStub;
5249
import io.grpc.rls.ChildLoadBalancerHelper.ChildLoadBalancerHelperProvider;
5350
import io.grpc.rls.LbPolicyConfiguration.ChildLbStatusListener;
54-
import io.grpc.rls.LbPolicyConfiguration.ChildLoadBalancingPolicy;
5551
import io.grpc.rls.LbPolicyConfiguration.ChildPolicyWrapper;
5652
import io.grpc.rls.LbPolicyConfiguration.RefCountedChildPolicyWrapperFactory;
5753
import io.grpc.rls.LruCache.EvictionListener;
@@ -138,7 +134,8 @@ private CachingRlsLbClient(Builder builder) {
138134
rlsConfig.getCacheSizeBytes(),
139135
builder.evictionListener,
140136
scheduledExecutorService,
141-
timeProvider);
137+
timeProvider,
138+
lock);
142139
logger = helper.getChannelLogger();
143140
String serverHost = null;
144141
try {
@@ -181,7 +178,9 @@ private CachingRlsLbClient(Builder builder) {
181178
new ChildLoadBalancerHelperProvider(helper, new SubchannelStateManagerImpl(), rlsPicker);
182179
refCountedChildPolicyWrapperFactory =
183180
new RefCountedChildPolicyWrapperFactory(
184-
childLbHelperProvider, new BackoffRefreshListener());
181+
lbPolicyConfig.getLoadBalancingPolicy(), childLbResolvedAddressFactory,
182+
childLbHelperProvider,
183+
new BackoffRefreshListener());
185184
logger.log(ChannelLogLevel.DEBUG, "CachingRlsLbClient created");
186185
}
187186

@@ -536,6 +535,7 @@ final class DataCacheEntry extends CacheEntry {
536535
private final long staleTime;
537536
private final ChildPolicyWrapper childPolicyWrapper;
538537

538+
// GuardedBy CachingRlsLbClient.lock
539539
DataCacheEntry(RouteLookupRequest request, final RouteLookupResponse response) {
540540
super(request);
541541
this.response = checkNotNull(response, "response");
@@ -546,29 +546,6 @@ final class DataCacheEntry extends CacheEntry {
546546
long now = timeProvider.currentTimeNanos();
547547
expireTime = now + maxAgeNanos;
548548
staleTime = now + staleAgeNanos;
549-
550-
if (childPolicyWrapper.getPicker() != null) {
551-
childPolicyWrapper.refreshState();
552-
} else {
553-
createChildLbPolicy();
554-
}
555-
}
556-
557-
private void createChildLbPolicy() {
558-
ChildLoadBalancingPolicy childPolicy = lbPolicyConfig.getLoadBalancingPolicy();
559-
LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();
560-
ConfigOrError lbConfig =
561-
lbProvider
562-
.parseLoadBalancingPolicyConfig(
563-
childPolicy.getEffectiveChildPolicy(childPolicyWrapper.getTarget()));
564-
565-
LoadBalancer lb = lbProvider.newLoadBalancer(childPolicyWrapper.getHelper());
566-
logger.log(
567-
ChannelLogLevel.DEBUG,
568-
"RLS child lb created. config: {0}",
569-
lbConfig.getConfig());
570-
lb.handleResolvedAddresses(childLbResolvedAddressFactory.create(lbConfig.getConfig()));
571-
lb.requestConnection();
572549
}
573550

574551
/**
@@ -637,7 +614,9 @@ boolean isStaled(long now) {
637614

638615
@Override
639616
void cleanup() {
640-
refCountedChildPolicyWrapperFactory.release(childPolicyWrapper);
617+
synchronized (lock) {
618+
refCountedChildPolicyWrapperFactory.release(childPolicyWrapper);
619+
}
641620
}
642621

643622
@Override
@@ -856,14 +835,15 @@ private static final class RlsAsyncLruCache
856835

857836
RlsAsyncLruCache(long maxEstimatedSizeBytes,
858837
@Nullable EvictionListener<RouteLookupRequest, CacheEntry> evictionListener,
859-
ScheduledExecutorService ses, TimeProvider timeProvider) {
838+
ScheduledExecutorService ses, TimeProvider timeProvider, Object lock) {
860839
super(
861840
maxEstimatedSizeBytes,
862841
new AutoCleaningEvictionListener(evictionListener),
863842
1,
864843
TimeUnit.MINUTES,
865844
ses,
866-
timeProvider);
845+
timeProvider,
846+
lock);
867847
}
868848

869849
@Override
@@ -985,27 +965,9 @@ private void startFallbackChildPolicy() {
985965
}
986966
fallbackChildPolicyWrapper = refCountedChildPolicyWrapperFactory.createOrGet(defaultTarget);
987967
}
988-
LoadBalancerProvider lbProvider =
989-
lbPolicyConfig.getLoadBalancingPolicy().getEffectiveLbProvider();
990-
final LoadBalancer lb =
991-
lbProvider.newLoadBalancer(fallbackChildPolicyWrapper.getHelper());
992-
final ConfigOrError lbConfig =
993-
lbProvider
994-
.parseLoadBalancingPolicyConfig(
995-
lbPolicyConfig
996-
.getLoadBalancingPolicy()
997-
.getEffectiveChildPolicy(defaultTarget));
998-
helper.getSynchronizationContext().execute(
999-
new Runnable() {
1000-
@Override
1001-
public void run() {
1002-
lb.handleResolvedAddresses(
1003-
childLbResolvedAddressFactory.create(lbConfig.getConfig()));
1004-
lb.requestConnection();
1005-
}
1006-
});
1007968
}
1008969

970+
// GuardedBy CachingRlsLbClient.lock
1009971
void close() {
1010972
if (fallbackChildPolicyWrapper != null) {
1011973
refCountedChildPolicyWrapperFactory.release(fallbackChildPolicyWrapper);

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

Lines changed: 63 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,15 @@
2222

2323
import com.google.common.annotations.VisibleForTesting;
2424
import com.google.common.base.MoreObjects;
25+
import io.grpc.ChannelLogger.ChannelLogLevel;
2526
import io.grpc.ConnectivityState;
27+
import io.grpc.LoadBalancer;
2628
import io.grpc.LoadBalancer.Helper;
2729
import io.grpc.LoadBalancer.Subchannel;
2830
import io.grpc.LoadBalancer.SubchannelPicker;
2931
import io.grpc.LoadBalancerProvider;
3032
import io.grpc.LoadBalancerRegistry;
33+
import io.grpc.NameResolver.ConfigOrError;
3134
import io.grpc.internal.ObjectPool;
3235
import io.grpc.rls.ChildLoadBalancerHelper.ChildLoadBalancerHelperProvider;
3336
import io.grpc.rls.RlsProtoData.RouteLookupConfig;
@@ -191,33 +194,49 @@ public String toString() {
191194

192195
/** Factory for {@link ChildPolicyWrapper}. */
193196
static final class RefCountedChildPolicyWrapperFactory {
197+
// GuardedBy CachingRlsLbClient.lock
194198
@VisibleForTesting
195199
final Map<String /* target */, RefCountedChildPolicyWrapper> childPolicyMap =
196200
new HashMap<>();
197201

198202
private final ChildLoadBalancerHelperProvider childLbHelperProvider;
199203
private final ChildLbStatusListener childLbStatusListener;
204+
private final ChildLoadBalancingPolicy childPolicy;
205+
private final ResolvedAddressFactory childLbResolvedAddressFactory;
200206

201207
public RefCountedChildPolicyWrapperFactory(
208+
ChildLoadBalancingPolicy childPolicy,
209+
ResolvedAddressFactory childLbResolvedAddressFactory,
202210
ChildLoadBalancerHelperProvider childLbHelperProvider,
203211
ChildLbStatusListener childLbStatusListener) {
212+
this.childPolicy = checkNotNull(childPolicy, "childPolicy");
213+
this.childLbResolvedAddressFactory =
214+
checkNotNull(childLbResolvedAddressFactory, "childLbResolvedAddressFactory");
204215
this.childLbHelperProvider = checkNotNull(childLbHelperProvider, "childLbHelperProvider");
205216
this.childLbStatusListener = checkNotNull(childLbStatusListener, "childLbStatusListener");
206217
}
207218

219+
// GuardedBy CachingRlsLbClient.lock
208220
ChildPolicyWrapper createOrGet(String target) {
209221
// TODO(creamsoup) check if the target is valid or not
210222
RefCountedChildPolicyWrapper pooledChildPolicyWrapper = childPolicyMap.get(target);
211223
if (pooledChildPolicyWrapper == null) {
212-
ChildPolicyWrapper childPolicyWrapper =
213-
new ChildPolicyWrapper(target, childLbHelperProvider, childLbStatusListener);
224+
ChildPolicyWrapper childPolicyWrapper = new ChildPolicyWrapper(
225+
target, childPolicy, childLbResolvedAddressFactory, childLbHelperProvider,
226+
childLbStatusListener);
214227
pooledChildPolicyWrapper = RefCountedChildPolicyWrapper.of(childPolicyWrapper);
215228
childPolicyMap.put(target, pooledChildPolicyWrapper);
229+
return pooledChildPolicyWrapper.getObject();
230+
} else {
231+
ChildPolicyWrapper childPolicyWrapper = pooledChildPolicyWrapper.getObject();
232+
if (childPolicyWrapper.getPicker() != null) {
233+
childPolicyWrapper.refreshState();
234+
}
235+
return childPolicyWrapper;
216236
}
217-
218-
return pooledChildPolicyWrapper.getObject();
219237
}
220238

239+
// GuardedBy CachingRlsLbClient.lock
221240
void release(ChildPolicyWrapper childPolicyWrapper) {
222241
checkNotNull(childPolicyWrapper, "childPolicyWrapper");
223242
String target = childPolicyWrapper.getTarget();
@@ -238,16 +257,36 @@ static final class ChildPolicyWrapper {
238257

239258
private final String target;
240259
private final ChildPolicyReportingHelper helper;
260+
private final LoadBalancer lb;
241261
private volatile SubchannelPicker picker;
242262
private ConnectivityState state;
243263

244264
public ChildPolicyWrapper(
245265
String target,
266+
ChildLoadBalancingPolicy childPolicy,
267+
final ResolvedAddressFactory childLbResolvedAddressFactory,
246268
ChildLoadBalancerHelperProvider childLbHelperProvider,
247269
ChildLbStatusListener childLbStatusListener) {
248270
this.target = target;
249271
this.helper =
250272
new ChildPolicyReportingHelper(childLbHelperProvider, childLbStatusListener);
273+
LoadBalancerProvider lbProvider = childPolicy.getEffectiveLbProvider();
274+
final ConfigOrError lbConfig =
275+
lbProvider
276+
.parseLoadBalancingPolicyConfig(
277+
childPolicy.getEffectiveChildPolicy(target));
278+
this.lb = lbProvider.newLoadBalancer(helper);
279+
helper.getChannelLogger().log(
280+
ChannelLogLevel.DEBUG, "RLS child lb created. config: {0}", lbConfig.getConfig());
281+
helper.getSynchronizationContext().execute(
282+
new Runnable() {
283+
@Override
284+
public void run() {
285+
lb.handleResolvedAddresses(
286+
childLbResolvedAddressFactory.create(lbConfig.getConfig()));
287+
lb.requestConnection();
288+
}
289+
});
251290
}
252291

253292
String getTarget() {
@@ -263,7 +302,25 @@ ChildPolicyReportingHelper getHelper() {
263302
}
264303

265304
void refreshState() {
266-
helper.updateBalancingState(state, picker);
305+
helper.getSynchronizationContext().execute(
306+
new Runnable() {
307+
@Override
308+
public void run() {
309+
helper.updateBalancingState(state, picker);
310+
}
311+
}
312+
);
313+
}
314+
315+
void shutdown() {
316+
helper.getSynchronizationContext().execute(
317+
new Runnable() {
318+
@Override
319+
public void run() {
320+
lb.shutdown();
321+
}
322+
}
323+
);
267324
}
268325

269326
@Override
@@ -346,6 +403,7 @@ public ChildPolicyWrapper returnObject(Object object) {
346403
long newCnt = refCnt.decrementAndGet();
347404
checkState(newCnt != -1, "Cannot return never pooled childPolicyWrapper");
348405
if (newCnt == 0) {
406+
childPolicyWrapper.shutdown();
349407
childPolicyWrapper = null;
350408
}
351409
return null;

‎rls/src/main/java/io/grpc/rls/LinkedHashLruCache.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@
4848
@ThreadSafe
4949
abstract class LinkedHashLruCache<K, V> implements LruCache<K, V> {
5050

51-
private final Object lock = new Object();
51+
private final Object lock;
5252

5353
@GuardedBy("lock")
5454
private final LinkedHashMap<K, SizedValue> delegate;
@@ -64,9 +64,11 @@ abstract class LinkedHashLruCache<K, V> implements LruCache<K, V> {
6464
int cleaningInterval,
6565
TimeUnit cleaningIntervalUnit,
6666
ScheduledExecutorService ses,
67-
final TimeProvider timeProvider) {
67+
final TimeProvider timeProvider,
68+
Object lock) {
6869
checkState(estimatedMaxSizeBytes > 0, "max estimated cache size should be positive");
6970
this.estimatedMaxSizeBytes = estimatedMaxSizeBytes;
71+
this.lock = checkNotNull(lock, "lock");
7072
this.evictionListener = new SizeHandlingEvictionListener(evictionListener);
7173
this.timeProvider = checkNotNull(timeProvider, "timeProvider");
7274
delegate = new LinkedHashMap<K, SizedValue>(
@@ -200,14 +202,15 @@ private V invalidate(K key, EvictionType cause) {
200202
}
201203

202204
@Override
203-
public final void invalidateAll(Iterable<K> keys) {
204-
checkNotNull(keys, "keys");
205+
public final void invalidateAll() {
205206
synchronized (lock) {
206-
for (K key : keys) {
207-
SizedValue existing = delegate.remove(key);
208-
if (existing != null) {
209-
evictionListener.onEviction(key, existing, EvictionType.EXPLICIT);
207+
Iterator<Map.Entry<K, SizedValue>> iterator = delegate.entrySet().iterator();
208+
while (iterator.hasNext()) {
209+
Map.Entry<K, SizedValue> entry = iterator.next();
210+
if (entry.getValue() != null) {
211+
evictionListener.onEviction(entry.getKey(), entry.getValue(), EvictionType.EXPLICIT);
210212
}
213+
iterator.remove();
211214
}
212215
}
213216
}
@@ -291,13 +294,10 @@ private boolean cleanupExpiredEntries(int maxExpiredEntries, long now) {
291294
public final void close() {
292295
synchronized (lock) {
293296
periodicCleaner.stop();
294-
doClose();
295-
delegate.clear();
297+
invalidateAll();
296298
}
297299
}
298300

299-
protected void doClose() {}
300-
301301
/** Periodically cleans up the AsyncRequestCache. */
302302
private final class PeriodicCleaner {
303303

‎rls/src/main/java/io/grpc/rls/LruCache.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,10 +49,10 @@ interface LruCache<K, V> {
4949
V invalidate(K key);
5050

5151
/**
52-
* Invalidates cache entries for given keys. This operation will trigger {@link EvictionListener}
52+
* Invalidates cache entries for all keys. This operation will trigger {@link EvictionListener}
5353
* with {@link EvictionType#EXPLICIT}.
5454
*/
55-
void invalidateAll(Iterable<K> keys);
55+
void invalidateAll();
5656

5757
/** Returns {@code true} if given key is cached. */
5858
@CheckReturnValue

0 commit comments

Comments
 (0)