You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/01/13 14:19:12 UTC
[GitHub] [beam] scwhittle commented on a change in pull request #16495: [BEAM-13015] Provide caching statistics in the status client.
scwhittle commented on a change in pull request #16495:
URL: https://github.com/apache/beam/pull/16495#discussion_r783999225
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/CachesTest.java
##########
@@ -160,4 +160,20 @@ private void testCache(Cache<String, String> cache) {
assertEquals("value2", cache.computeIfAbsent("key2", (unused) -> "value2"));
assertEquals("value2", cache.peek("key2"));
}
+
+ @Test
+ public void testDescribeStats() throws Exception {
+ Cache<Integer, WeightedValue<String>> cache = Caches.forMaximumBytes(1000 * 1048576L);
+ for (int i = 0; i < 100; ++i) {
+ cache.computeIfAbsent(i, (key) -> WeightedValue.of("value", 1048576L));
+ cache.peek(i);
+ cache.put(100 + i, WeightedValue.of("value", 1048576L));
+ }
+
+ assertThat(cache.describeStats(), containsString("used/max 200/1000 MB"));
Review comment:
would be good to have some test of the correctness of the used bytes as item weight is updated or evictions occur
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/WeightedValue.java
##########
@@ -47,4 +49,26 @@ public long getWeight() {
public T getValue() {
return value;
}
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof WeightedValue)) {
Review comment:
check for null?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/Caches.java
##########
@@ -57,92 +65,58 @@ public static long weigh(Object o) {
/** An eviction listener that reduces the size of entries that are {@link Shrinkable}. */
@VisibleForTesting
- static class ShrinkOnEviction implements RemovalListener<CompositeKey, Object> {
+ static class ShrinkOnEviction implements RemovalListener<CompositeKey, WeightedValue<Object>> {
private final org.apache.beam.vendor.guava.v26_0_jre.com.google.common.cache.Cache<
- CompositeKey, Object>
+ CompositeKey, WeightedValue<Object>>
cache;
+ private final LongAdder weightInBytes;
- ShrinkOnEviction(CacheBuilder<Object, Object> cacheBuilder) {
+ ShrinkOnEviction(
+ CacheBuilder<Object, WeightedValue<Object>> cacheBuilder, LongAdder weightInBytes) {
this.cache = cacheBuilder.removalListener(this).build();
+ this.weightInBytes = weightInBytes;
}
public org.apache.beam.vendor.guava.v26_0_jre.com.google.common.cache.Cache<
- CompositeKey, Object>
+ CompositeKey, WeightedValue<Object>>
getCache() {
return cache;
}
@Override
- public void onRemoval(RemovalNotification<CompositeKey, Object> removalNotification) {
+ public void onRemoval(
+ RemovalNotification<CompositeKey, WeightedValue<Object>> removalNotification) {
+ weightInBytes.add(-removalNotification.getValue().getWeight());
if (removalNotification.wasEvicted()) {
- if (!(removalNotification.getValue() instanceof Cache.Shrinkable)) {
+ if (!(removalNotification.getValue().getValue() instanceof Cache.Shrinkable)) {
return;
}
- Object updatedEntry = ((Shrinkable<?>) removalNotification.getValue()).shrink();
+ Object updatedEntry = ((Shrinkable<?>) removalNotification.getValue().getValue()).shrink();
if (updatedEntry != null) {
- cache.put(removalNotification.getKey(), updatedEntry);
+ cache.put(removalNotification.getKey(), addWeightedValue(updatedEntry, weightInBytes));
}
}
}
}
/** A cache that never stores any values. */
public static <K, V> Cache<K, V> noop() {
- // We specifically use Guava cache since it allows for recursive computeIfAbsent calls
- // preventing deadlock from occurring when a loading function mutates the underlying cache
- return (Cache<K, V>)
- forCache(new ShrinkOnEviction(CacheBuilder.newBuilder().maximumSize(0)).getCache());
+ return forMaximumBytes(0L);
}
/** A cache that never evicts any values. */
public static <K, V> Cache<K, V> eternal() {
- // We specifically use Guava cache since it allows for recursive computeIfAbsent calls
- // preventing deadlock from occurring when a loading function mutates the underlying cache
- return (Cache<K, V>)
- forCache(
- new ShrinkOnEviction(CacheBuilder.newBuilder().maximumSize(Long.MAX_VALUE)).getCache());
+ return forMaximumBytes(Long.MAX_VALUE);
}
/**
* Uses the specified {@link PipelineOptions} to configure and return a cache instance based upon
* parameters within {@link SdkHarnessOptions}.
*/
public static <K, V> Cache<K, V> fromOptions(PipelineOptions options) {
- // We specifically use Guava cache since it allows for recursive computeIfAbsent calls
- // preventing deadlock from occurring when a loading function mutates the underlying cache
- return (Cache<K, V>)
- forCache(
- new ShrinkOnEviction(
- CacheBuilder.newBuilder()
- .maximumWeight(
- options.as(SdkHarnessOptions.class).getMaxCacheMemoryUsageMb()
- * 1024L
- * 1024L
- / WEIGHT_RATIO)
- .weigher(
- new Weigher<Object, Object>() {
-
- @Override
- public int weigh(Object key, Object value) {
- long size;
- if (value instanceof Weighted) {
- size = Caches.weigh(key) + ((Weighted) value).getWeight();
- } else {
- size = Caches.weigh(key) + Caches.weigh(value);
- }
- size = size / WEIGHT_RATIO + 1;
- if (size >= Integer.MAX_VALUE) {
- LOG.warn(
- "Entry with size {} MiBs inserted into the cache. This is larger than the maximum individual entry size of {} MiBs. The cache will under report its memory usage by the difference. This may lead to OutOfMemoryErrors.",
- (size / 1048576L) + 1,
- 2 * WEIGHT_RATIO * 1024);
- return Integer.MAX_VALUE;
- }
- return (int) size;
- }
- }))
- .getCache());
+ return forMaximumBytes(
+ options.as(SdkHarnessOptions.class).getMaxCacheMemoryUsageMb() * 1048576L);
Review comment:
nit: I think 1024* 1024 is easier to eyeball correctness
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org