FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Rls spec sync (#9437) · Java-Edge/grpc-java@b66250e · GitHub

/ grpc-java Public
forked from grpc/grpc-java

Commit b66250e

Browse files
larry-safran
authored
Rls spec sync (grpc#9437)
rls: Update implementation to match spec. * Cleanup cache if exceeds max size when add an entry. Make cache entry size calculations more accurate * Trigger pending RPC processing if unexpired backoff entries were removed from the cache by triggering helper to call it's parent updateBalancingState with the same state and picker * Introduce minimum time before eviction (5 seconds) * Change default accept ratio for AdaptiveThrottler from 1.2 -> 2.0 * Configuration validation * When checking key names for duplicates also look at headers * Check extra keys for duplicates See analysis of implementation versus spec at https://docs.google.com/spreadsheets/d/18w5s1TEebRumWzk1pvWnjiHFGKc6MW-vt8tRLY4eNs0/
1 parent 618a4de commit b66250e

6 files changed

Lines changed: 122 additions & 25 deletions

File tree

‎interop-testing/src/main/java/io/grpc/testing/integration/RpcBehaviorLoadBalancerProvider.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@
3838
* looks for an "rpc_behavior" field in its configuration and includes the value in the
3939
* "rpc-behavior" metadata entry that is sent to the server. This will cause the test server to
4040
* behave in a predefined way. Endpoint picking logic is delegated to the
41-
* {@link RoundRobinLoadBalancer}.
41+
* io.grpc.util.RoundRobinLoadBalancer.
4242
*
4343
* <p>Initial use case is to prove that a custom load balancer can be configured by the control
4444
* plane via xDS. An interop test will configure this LB and then verify it has been correctly

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ final class AdaptiveThrottler implements Throttler {
4444

4545
private static final int DEFAULT_HISTORY_SECONDS = 30;
4646
private static final int DEFAULT_REQUEST_PADDING = 8;
47-
private static final float DEFAULT_RATIO_FOR_ACCEPT = 1.2f;
47+
private static final float DEFAULT_RATIO_FOR_ACCEPT = 2.0f;
4848

4949
/**
5050
* The duration of history of calls used by Adaptive Throttler.

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

Lines changed: 61 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -81,12 +81,17 @@ final class CachingRlsLbClient {
8181
REQUEST_CONVERTER = new RlsProtoConverters.RouteLookupRequestConverter().reverse();
8282
private static final Converter<RouteLookupResponse, io.grpc.lookup.v1.RouteLookupResponse>
8383
RESPONSE_CONVERTER = new RouteLookupResponseConverter().reverse();
84+
public static final long MIN_EVICTION_TIME_DELTA_NANOS = TimeUnit.SECONDS.toNanos(5);
85+
public static final int BYTES_PER_CHAR = 2;
86+
public static final int STRING_OVERHEAD_BYTES = 38;
87+
/** Minimum bytes for a Java Object. */
88+
public static final int OBJ_OVERHEAD_B = 16;
8489

8590
// All cache status changes (pending, backoff, success) must be under this lock
8691
private final Object lock = new Object();
8792
// LRU cache based on access order (BACKOFF and actual data will be here)
8893
@GuardedBy("lock")
89-
private final LinkedHashLruCache<RouteLookupRequest, CacheEntry> linkedHashLruCache;
94+
private final RlsAsyncLruCache linkedHashLruCache;
9095
// any RPC on the fly will cached in this map
9196
@GuardedBy("lock")
9297
private final Map<RouteLookupRequest, PendingCacheEntry> pendingCallCache = new HashMap<>();
@@ -287,12 +292,12 @@ private CachedRouteLookupResponse handleNewRequest(RouteLookupRequest request) {
287292
try {
288293
RouteLookupResponse response = asyncCall.get();
289294
DataCacheEntry dataEntry = new DataCacheEntry(request, response);
290-
linkedHashLruCache.cache(request, dataEntry);
295+
linkedHashLruCache.cacheAndClean(request, dataEntry);
291296
return CachedRouteLookupResponse.dataEntry(dataEntry);
292297
} catch (Exception e) {
293298
BackoffCacheEntry backoffEntry =
294299
new BackoffCacheEntry(request, Status.fromThrowable(e), backoffProvider.get());
295-
linkedHashLruCache.cache(request, backoffEntry);
300+
linkedHashLruCache.cacheAndClean(request, backoffEntry);
296301
return CachedRouteLookupResponse.backoffEntry(backoffEntry);
297302
}
298303
}
@@ -336,6 +341,10 @@ public void run() {
336341
}
337342
});
338343
}
344+
345+
void triggerPendingRpcProcessing() {
346+
super.updateBalancingState(state, picker);
347+
}
339348
}
340349

341350
/**
@@ -488,14 +497,15 @@ private void transitionToDataEntry(RouteLookupResponse routeLookupResponse) {
488497
ChannelLogLevel.DEBUG,
489498
"Transition to data cache: routeLookupResponse={0}",
490499
routeLookupResponse);
491-
linkedHashLruCache.cache(request, new DataCacheEntry(request, routeLookupResponse));
500+
linkedHashLruCache.cacheAndClean(request, new DataCacheEntry(request, routeLookupResponse));
492501
}
493502
}
494503

495504
private void transitionToBackOff(Status status) {
496505
synchronized (lock) {
497506
logger.log(ChannelLogLevel.DEBUG, "Transition to back off: status={0}", status);
498-
linkedHashLruCache.cache(request, new BackoffCacheEntry(request, status, backoffPolicy));
507+
linkedHashLruCache.cacheAndClean(request,
508+
new BackoffCacheEntry(request, status, backoffPolicy));
499509
}
500510
}
501511

@@ -525,11 +535,20 @@ final boolean isExpired() {
525535
abstract boolean isExpired(long now);
526536

527537
abstract void cleanup();
538+
539+
protected long getMinEvictionTime() {
540+
return 0L;
541+
}
542+
543+
protected void triggerPendingRpcProcessing() {
544+
helper.triggerPendingRpcProcessing();
545+
}
528546
}
529547

530548
/** Implementation of {@link CacheEntry} contains valid data. */
531549
final class DataCacheEntry extends CacheEntry {
532550
private final RouteLookupResponse response;
551+
private final long minEvictionTime;
533552
private final long expireTime;
534553
private final long staleTime;
535554
private final List<ChildPolicyWrapper> childPolicyWrappers;
@@ -543,6 +562,7 @@ final class DataCacheEntry extends CacheEntry {
543562
refCountedChildPolicyWrapperFactory
544563
.createOrGet(response.targets());
545564
long now = ticker.read();
565+
minEvictionTime = now + MIN_EVICTION_TIME_DELTA_NANOS;
546566
expireTime = now + maxAgeNanos;
547567
staleTime = now + staleAgeNanos;
548568
}
@@ -574,13 +594,13 @@ void maybeRefresh() {
574594
// async call returned finished future is most likely throttled
575595
try {
576596
RouteLookupResponse response = asyncCall.get();
577-
linkedHashLruCache.cache(request, new DataCacheEntry(request, response));
597+
linkedHashLruCache.cacheAndClean(request, new DataCacheEntry(request, response));
578598
} catch (InterruptedException e) {
579599
Thread.currentThread().interrupt();
580600
} catch (Exception e) {
581601
BackoffCacheEntry backoffEntry =
582602
new BackoffCacheEntry(request, Status.fromThrowable(e), backoffProvider.get());
583-
linkedHashLruCache.cache(request, backoffEntry);
603+
linkedHashLruCache.cacheAndClean(request, backoffEntry);
584604
}
585605
}
586606
}
@@ -611,11 +631,19 @@ String getHeaderData() {
611631
return response.getHeaderData();
612632
}
613633

634+
// Assume UTF-16 (2 bytes) and overhead of a String object is 38 bytes
635+
int calcStringSize(String target) {
636+
return target.length() * BYTES_PER_CHAR + STRING_OVERHEAD_BYTES;
637+
}
638+
614639
@Override
615640
int getSizeBytes() {
616-
// size of strings and java object overhead, actual memory usage is more than this.
617-
return
618-
(response.targets().get(0).length() + response.getHeaderData().length()) * 2 + 38 * 2;
641+
int targetSize = 0;
642+
for (String target : response.targets()) {
643+
targetSize += calcStringSize(target);
644+
}
645+
return targetSize + calcStringSize(response.getHeaderData()) + OBJ_OVERHEAD_B // response size
646+
+ Long.SIZE * 2 + OBJ_OVERHEAD_B; // Other fields
619647
}
620648

621649
@Override
@@ -627,6 +655,11 @@ boolean isStaled(long now) {
627655
return staleTime - now <= 0;
628656
}
629657

658+
@Override
659+
protected long getMinEvictionTime() {
660+
return minEvictionTime;
661+
}
662+
630663
@Override
631664
void cleanup() {
632665
synchronized (lock) {
@@ -700,11 +733,11 @@ private void transitionToPending() {
700733
} else {
701734
try {
702735
RouteLookupResponse response = call.get();
703-
linkedHashLruCache.cache(request, new DataCacheEntry(request, response));
736+
linkedHashLruCache.cacheAndClean(request, new DataCacheEntry(request, response));
704737
} catch (InterruptedException e) {
705738
Thread.currentThread().interrupt();
706739
} catch (Exception e) {
707-
linkedHashLruCache.cache(
740+
linkedHashLruCache.cacheAndClean(
708741
request,
709742
new BackoffCacheEntry(request, Status.fromThrowable(e), backoffPolicy));
710743
}
@@ -718,7 +751,7 @@ Status getStatus() {
718751

719752
@Override
720753
int getSizeBytes() {
721-
return 0;
754+
return OBJ_OVERHEAD_B * 3 + Long.SIZE + 8; // 3 java objects, 1 long and a boolean
722755
}
723756

724757
@Override
@@ -876,8 +909,22 @@ protected int estimateSizeOf(RouteLookupRequest key, CacheEntry value) {
876909
@Override
877910
protected boolean shouldInvalidateEldestEntry(
878911
RouteLookupRequest eldestKey, CacheEntry eldestValue) {
912+
if (eldestValue.getMinEvictionTime() > now()) {
913+
return false;
914+
}
915+
879916
// eldest entry should be evicted if size limit exceeded
880-
return true;
917+
return this.estimatedSizeBytes() > this.estimatedMaxSizeBytes();
918+
}
919+
920+
public CacheEntry cacheAndClean(RouteLookupRequest key, CacheEntry value) {
921+
CacheEntry newEntry = cache(key, value);
922+
923+
// force cleanup if new entry pushed cache over max size (in bytes)
924+
if (fitToLimit()) {
925+
value.triggerPendingRpcProcessing();
926+
}
927+
return newEntry;
881928
}
882929
}
883930

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

Lines changed: 31 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,10 @@ protected int estimateSizeOf(K key, V value) {
118118
return 1;
119119
}
120120

121+
protected long estimatedMaxSizeBytes() {
122+
return estimatedMaxSizeBytes;
123+
}
124+
121125
/** Updates size for given key if entry exists. It is useful if the cache value is mutated. */
122126
public void updateEntrySize(K key) {
123127
synchronized (lock) {
@@ -233,30 +237,50 @@ public final List<V> values() {
233237
}
234238
}
235239

240+
protected long now() {
241+
return ticker.read();
242+
}
243+
236244
/**
237-
* Resizes cache. If new size is smaller than current estimated size, it will free up space by
245+
* Cleans up cache if needed to fit into max size bytes by
238246
* removing expired entries and removing oldest entries by LRU order.
247+
* Returns TRUE if any unexpired entries were removed
239248
*/
240-
public final void resize(int newSizeBytes) {
241-
long now = ticker.read();
249+
protected final boolean fitToLimit() {
250+
boolean removedAnyUnexpired = false;
242251
synchronized (lock) {
243-
this.estimatedMaxSizeBytes = newSizeBytes;
244-
if (estimatedSizeBytes.get() <= newSizeBytes) {
252+
if (estimatedSizeBytes.get() <= estimatedMaxSizeBytes) {
245253
// new size is larger no need to do cleanup
246-
return;
254+
return false;
247255
}
248256
// cleanup expired entries
249-
cleanupExpiredEntries(now);
257+
cleanupExpiredEntries(now());
250258

251259
// cleanup eldest entry until new size limit
252260
Iterator<Map.Entry<K, SizedValue>> lruIter = delegate.entrySet().iterator();
253261
while (lruIter.hasNext() && estimatedMaxSizeBytes < this.estimatedSizeBytes.get()) {
254262
Map.Entry<K, SizedValue> entry = lruIter.next();
263+
if (!shouldInvalidateEldestEntry(entry.getKey(), entry.getValue().value)) {
264+
break; // Violates some constraint like minimum age so stop our cleanup
265+
}
255266
lruIter.remove();
256267
// eviction listener will update the estimatedSizeBytes
257268
evictionListener.onEviction(entry.getKey(), entry.getValue(), EvictionType.SIZE);
269+
removedAnyUnexpired = true;
258270
}
259271
}
272+
return removedAnyUnexpired;
273+
}
274+
275+
/**
276+
* Resizes cache. If new size is smaller than current estimated size, it will free up space by
277+
* removing expired entries and removing oldest entries by LRU order.
278+
*/
279+
public final void resize(long newSizeBytes) {
280+
synchronized (lock) {
281+
this.estimatedMaxSizeBytes = newSizeBytes;
282+
fitToLimit();
283+
}
260284
}
261285

262286
@Override

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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,13 +112,28 @@ protected RouteLookupConfig doForward(Map<String, ?> json) {
112112
ImmutableList<GrpcKeyBuilder> grpcKeybuilders =
113113
GrpcKeyBuilderConverter.covertAll(
114114
checkNotNull(JsonUtil.getListOfObjects(json, "grpcKeybuilders"), "grpcKeybuilders"));
115+
116+
// Validate grpc_keybuilders
115117
checkArgument(!grpcKeybuilders.isEmpty(), "must have at least one GrpcKeyBuilder");
116118
Set<Name> names = new HashSet<>();
117119
for (GrpcKeyBuilder keyBuilder : grpcKeybuilders) {
118120
for (Name name : keyBuilder.names()) {
119121
checkArgument(names.add(name), "duplicate names in grpc_keybuilders: " + name);
120122
}
123+
124+
Set<String> keys = new HashSet<>();
125+
for (NameMatcher header : keyBuilder.headers()) {
126+
checkKeys(keys, header.key(), "header");
127+
}
128+
for (String key : keyBuilder.constantKeys().keySet()) {
129+
checkKeys(keys, key, "constant");
130+
}
131+
String extraKeyStr = keyToString(keyBuilder.extraKeys());
132+
checkArgument(keys.add(extraKeyStr),
133+
"duplicate extra key in grpc_keybuilders: " + extraKeyStr);
121134
}
135+
136+
// Validate lookup_service
122137
String lookupService = JsonUtil.getString(json, "lookupService");
123138
checkArgument(!Strings.isNullOrEmpty(lookupService), "lookupService must not be empty");
124139
try {
@@ -157,6 +172,11 @@ protected RouteLookupConfig doForward(Map<String, ?> json) {
157172
.build();
158173
}
159174

175+
private static String keyToString(ExtraKeys extraKeys) {
176+
return String.format("host: %s, service: %s, method: %s",
177+
extraKeys.host(), extraKeys.service(), extraKeys.method());
178+
}
179+
160180
private static <T> T orDefault(@Nullable T value, T defaultValue) {
161181
if (value == null) {
162182
return checkNotNull(defaultValue, "defaultValue");
@@ -170,6 +190,12 @@ protected Map<String, Object> doBackward(RouteLookupConfig routeLookupConfig) {
170190
}
171191
}
172192

193+
private static void checkKeys(Set<String> keys, String key, String keyType) {
194+
checkArgument(key != null, "unset " + keyType + " key");
195+
checkArgument(!key.isEmpty(), "Empty string for " + keyType + " key");
196+
checkArgument(keys.add(key), "duplicate " + keyType + " key in grpc_keybuilders: " + key);
197+
}
198+
173199
private static final class GrpcKeyBuilderConverter {
174200
public static ImmutableList<GrpcKeyBuilder> covertAll(List<Map<String, ?>> keyBuilders) {
175201
ImmutableList.Builder<GrpcKeyBuilder> keyBuilderList = ImmutableList.builder();

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,8 @@ private static Map<String, GrpcKeyBuilder> createKeyBuilderTable(
5252
Map<String, GrpcKeyBuilder> table = new HashMap<>();
5353
for (GrpcKeyBuilder grpcKeyBuilder : config.grpcKeybuilders()) {
5454
for (Name name : grpcKeyBuilder.names()) {
55-
boolean hasMethod = name.method() == null || name.method().isEmpty();
56-
String method = hasMethod ? "*" : name.method();
55+
boolean noMethod = name.method() == null || name.method().isEmpty();
56+
String method = noMethod ? "*" : name.method();
5757
String path = "/" + name.service() + "/" + method;
5858
table.put(path, grpcKeyBuilder);
5959
}

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL