yuqi1129 commented on code in PR #13702:
URL: https://github.com/apache/gravitino/pull/13702#discussion_r4219161949
##########
core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java:
##########
@@ -80,6 +93,24 @@ public class CaffeineEntityCache extends BaseEntityCache {
},
new ThreadPoolExecutor.CallerRunsPolicy());
+ /**
+ * Removes evicted and expired keys from the prefix index. The removal
listener only hands keys to
+ * this executor; the cleanup itself takes the key's segment lock, which is
safe here because this
+ * thread never holds a Caffeine lock.
+ *
+ * <p>The queue is unbounded and the executor is never shut down, so handing
off a key never fails
+ * or runs the cleanup on Caffeine's thread. Each task holds only a key, so
the backlog is bounded
+ * by the number of pending evictions. A segment held across a slow store
read delays the cleanup
+ * behind it, which only lets {@link #size()} over-count and the stale keys
use memory for longer.
+ */
+ private static final ExecutorService INDEX_CLEANUP_EXECUTOR =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread t = new Thread(r, "CaffeineEntityCache-IndexCleanup");
+ t.setDaemon(true);
+ return t;
+ });
Review Comment:
Fixed in b0bccc0ab. Removed the dedicated index-cleanup executor and its
unbounded queue entirely. Index cleanup now uses a per-cache index lock that is
never held across segment-lock acquisition or mutating Caffeine operations.
This avoids both the original lock cycle and cleanup backlog behind a held
segment. Added a regression test with 2,000 distinct insertions while one
segment remains locked; it fails with the previous queued implementation.
##########
core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java:
##########
@@ -252,20 +296,52 @@ public <E, T extends Exception> E withCacheLock(
}
/**
- * Removes the expired entity from the cache. This method is a hook method
for the Cache, when an
- * entry expires, it will call this method.
+ * Removes a key that Caffeine evicted or expired from the prefix index.
Runs on the index cleanup
+ * executor, never on a thread that holds a Caffeine lock.
+ *
+ * <p>A put can add the key back between Caffeine removing it and this
cleanup running. That put
+ * indexes the key under the same segment lock, so the index entry is
removed only when the key is
+ * still absent from the cache; otherwise it belongs to the live entry.
*
- * @param key The key of the expired entity
+ * @param key The key of the evicted or expired entity
*/
@Override
protected void invalidateExpiredItem(EntityCacheKey key) {
segmentedLock.withLock(
key,
() -> {
- cacheIndex.remove(key.toString());
+ if (cacheData.policy().getIfPresentQuietly(key) == null) {
+ cacheIndex.remove(key.toString());
+ }
});
}
+ /**
+ * Hands an evicted or expired key to the index cleanup executor without
blocking.
+ *
+ * @param key The key Caffeine removed
+ * @param cause Why Caffeine removed it
+ */
+ private void scheduleIndexCleanup(EntityCacheKey key, RemovalCause cause) {
+ try {
+ indexCleanupExecutor.execute(
+ () -> {
+ try {
+ invalidateExpiredItem(key);
+ } catch (Throwable t) {
+ LOG.error(
+ "Failed to remove entity key={} from the cache index,
cause={}", key, cause, t);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ // Unreachable with the production executor, which is unbounded and
never shut down. Never
+ // fall
+ // back to running the cleanup here: this thread may hold Caffeine's
eviction lock. A key left
+ // in the index only costs memory; a later invalidation of it or an
ancestor removes it.
+ LOG.error("Failed to schedule cache index cleanup for key={}, cause={}",
key, cause, e);
+ }
+ }
Review Comment:
Fixed in b0bccc0ab by removing the extra task-submission path. Removal
callbacks now perform guarded index cleanup directly and retain
catch(Throwable) with error logging. The independent index lock does not
acquire segment locks or call mutating Caffeine operations, so Caffeine's
inline notification fallback is safe.
##########
core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheEviction.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.gravitino.cache;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.utils.TestUtil;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that Caffeine eviction never waits for an entity segment lock, and
that the prefix index
+ * still tracks evicted keys correctly.
+ *
+ * <p>Caffeine runs maintenance and removal notifications inline when its
executor rejects a task,
+ * and does so while holding its eviction lock. A writer that holds a segment
lock can at the same
+ * time wait for that eviction lock inside {@code cacheData.put}, so the
removal listener must not
+ * take a segment lock on the thread Caffeine calls it on.
+ */
+public class TestCaffeineEntityCacheEviction {
+
+ @Test
+ void testEvictionCompletesWhileSegmentLocksAreHeld() throws Exception {
+ ExecutorService indexCleanup = Executors.newSingleThreadExecutor();
+ ExecutorService evictor = Executors.newSingleThreadExecutor();
+ try {
+ // A direct executor reproduces the rejected-task path: Caffeine
performs eviction and calls
+ // the removal listener inline, under its eviction lock.
+ CaffeineEntityCache cache =
+ new CaffeineEntityCache(countBoundedConfig(1), Runnable::run,
indexCleanup);
+ BaseMetalake first = TestUtil.getTestMetalake(1L, "metalake1", "first");
+ BaseMetalake second = TestUtil.getTestMetalake(2L, "metalake2",
"second");
+ EntityCacheKey firstKey = metalakeKey(first);
+ EntityCacheKey secondKey = metalakeKey(second);
+ cache.put(first);
+
+ // Hold both segment locks, so whichever entry Caffeine evicts, its
listener would block here
+ // if it took the segment lock.
+ cache.withCacheLock(
+ firstKey,
+ () ->
+ cache.withCacheLock(
+ secondKey,
+ () -> {
+ Future<?> eviction =
+ evictor.submit(
+ () -> {
+ // Bypass the segment lock to model Caffeine
evicting on a thread that
+ // does not own the evicted key's segment.
+ cache.getCacheData().put(secondKey, second);
+ cache.getCacheData().cleanUp();
+ });
+ Assertions.assertDoesNotThrow(
+ () -> eviction.get(5, TimeUnit.SECONDS),
+ "eviction must not wait for a segment lock while it
holds Caffeine's"
+ + " eviction lock");
+ }));
+
+ Assertions.assertEquals(1, cache.getCacheData().estimatedSize());
+ // The evicted entry leaves the prefix index once its segment is free
again.
+ Awaitility.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+
cache.getCacheData().policy().getIfPresentQuietly(firstKey) == null ? 0 : 1,
+ cache.size()));
+ } finally {
+ evictor.shutdownNow();
+ indexCleanup.shutdownNow();
+ }
+ }
+
+ @Test
+ void testStaleEvictionDoesNotUnindexReinsertedEntry() {
+ // Defer Caffeine's removal notifications so one can arrive after the key
was put back. They run
+ // later on this thread, which then holds no Caffeine lock, so a direct
index cleanup is safe.
+ Queue<Runnable> deferredNotifications = new ArrayDeque<>();
+ CaffeineEntityCache cache =
+ new CaffeineEntityCache(countBoundedConfig(10),
deferredNotifications::add, Runnable::run);
+ BaseMetalake metalake = TestUtil.getTestMetalake(1L, "metalake1",
"metalake");
+ CatalogEntity catalog =
+ TestUtil.getTestCatalogEntity(2L, "catalog1",
Namespace.of("metalake1"), "hive", "child");
+ EntityCacheKey catalogKey =
+ EntityCacheKey.of(catalog.nameIdentifier(), Entity.EntityType.CATALOG);
+ cache.put(catalog);
+ runAll(deferredNotifications);
+
+ // Evict everything; the notification for the catalog stays queued.
+ cache.getCacheData().policy().eviction().get().setMaximum(0);
Review Comment:
Fixed in b0bccc0ab. Added an explicit cache.getCacheData().cleanUp() after
setMaximum(0) and before asserting the entry is absent. Removal notifications
remain deferred by the controlled executor, allowing the test to
deterministically exercise reinsertion before the stale notification runs.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
