Optimise shared-blob-cache evictions

docs/changelog/126581.yaml

@@ -0,0 +1,5 @@
+pr: 126581
+summary: Evict from the shared blob cache asynchronously
+area: Searchable Snapshots
+type: enhancement
+issues: []

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/BlobCachePlugin.java

@@ -27,7 +27,8 @@ public List<Setting<?>> getSettings() {
             SharedBlobCacheService.SHARED_CACHE_DECAY_INTERVAL_SETTING,
             SharedBlobCacheService.SHARED_CACHE_MIN_TIME_DELTA_SETTING,
             SharedBlobCacheService.SHARED_CACHE_MMAP,
-            SharedBlobCacheService.SHARED_CACHE_COUNT_READS
+            SharedBlobCacheService.SHARED_CACHE_COUNT_READS,
+            SharedBlobCacheService.SHARED_CACHE_CONCURRENT_EVICTIONS_SETTING
         );
     }
 }

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.RelativeByteSizeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.common.util.concurrent.ThrottledTaskRunner;
 import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.Nullable;
@@ -97,6 +98,13 @@ public class SharedBlobCacheService<KeyType> implements Releasable {
         Setting.Property.NodeScope
     );
 
+    public static final Setting<Integer> SHARED_CACHE_CONCURRENT_EVICTIONS_SETTING = Setting.intSetting(
+        SHARED_CACHE_SETTINGS_PREFIX + "concurrent_evictions",
+        5,
+        1,
+        Setting.Property.NodeScope
+    );
+
     private static Setting.Validator<ByteSizeValue> getPageSizeAlignedByteSizeValueValidator(String settingName) {
         return value -> {
             if (value.getBytes() == -1) {
@@ -283,6 +291,8 @@ private interface Cache<K, T> extends Releasable {
         CacheEntry<T> get(K cacheKey, long fileLength, int region);
 
         int forceEvict(Predicate<K> cacheKeyPredicate);
+
+        void forceEvictAsync(Predicate<K> cacheKey);
     }
 
     private abstract static class CacheEntry<T> {
@@ -328,6 +338,7 @@ private CacheEntry(T chunk) {
     private final Runnable evictIncrementer;
 
     private final LongSupplier relativeTimeInNanosSupplier;
+    private final ThrottledTaskRunner evictionsRunner;
 
     public SharedBlobCacheService(
         NodeEnvironment environment,
@@ -388,6 +399,11 @@ public SharedBlobCacheService(
         this.blobCacheMetrics = blobCacheMetrics;
         this.evictIncrementer = blobCacheMetrics.getEvictedCountNonZeroFrequency()::increment;
         this.relativeTimeInNanosSupplier = relativeTimeInNanosSupplier;
+        this.evictionsRunner = new ThrottledTaskRunner(
+            "shared_blob_cache_evictions",
+            SHARED_CACHE_CONCURRENT_EVICTIONS_SETTING.get(settings),
+            threadPool.generic()
+        );
     }
 
     public static long calculateCacheSize(Settings settings, long totalFsSize) {
@@ -704,6 +720,15 @@ public int forceEvict(Predicate<KeyType> cacheKeyPredicate) {
 
     }
 
+    /**
+     * Evict entries from the cache that match the given predicate asynchronously
+     *
+     * @param cacheKeyPredicate
+     */
+    public void forceEvictAsync(Predicate<KeyType> cacheKeyPredicate) {
+        cache.forceEvictAsync(cacheKeyPredicate);
+    }
+
     // used by tests
     int getFreq(CacheFileRegion<KeyType> cacheFileRegion) {
         if (cache instanceof LFUCache lfuCache) {
@@ -1606,6 +1631,24 @@ public int forceEvict(Predicate<KeyType> cacheKeyPredicate) {
             return evictedCount;
         }
 
+        @Override
+        public void forceEvictAsync(Predicate<KeyType> cacheKeyPredicate) {
+            evictionsRunner.enqueueTask(new ActionListener<>() {
+                @Override
+                public void onResponse(Releasable releasable) {
+                    forceEvict(cacheKeyPredicate);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    // should be impossible, GENERIC pool doesn't reject anything
+                    final String message = "unexpected failure evicting from shared blob cache";
+                    logger.error(message, e);
+                    assert false : new AssertionError(message, e);
+                }
+            });
+        }
+
         private LFUCacheEntry initChunk(LFUCacheEntry entry) {
             assert Thread.holdsLock(entry.chunk);
             RegionKey<KeyType> regionKey = entry.chunk.regionKey;

x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/shared/SharedBlobCacheServiceTests.java

@@ -272,6 +272,44 @@ public void testForceEvictResponse() throws IOException {
         }
     }
 
+    public void testAsynchronousEviction() throws Exception {
+        Settings settings = Settings.builder()
+            .put(NODE_NAME_SETTING.getKey(), "node")
+            .put(SharedBlobCacheService.SHARED_CACHE_SIZE_SETTING.getKey(), ByteSizeValue.ofBytes(size(500)).getStringRep())
+            .put(SharedBlobCacheService.SHARED_CACHE_REGION_SIZE_SETTING.getKey(), ByteSizeValue.ofBytes(size(100)).getStringRep())
+            .put("path.home", createTempDir())
+            .build();
+        final DeterministicTaskQueue taskQueue = new DeterministicTaskQueue();
+        try (
+            NodeEnvironment environment = new NodeEnvironment(settings, TestEnvironment.newEnvironment(settings));
+            var cacheService = new SharedBlobCacheService<>(
+                environment,
+                settings,
+                taskQueue.getThreadPool(),
+                taskQueue.getThreadPool().executor(ThreadPool.Names.GENERIC),
+                BlobCacheMetrics.NOOP
+            )
+        ) {
+            final var cacheKey1 = generateCacheKey();
+            final var cacheKey2 = generateCacheKey();
+            assertEquals(5, cacheService.freeRegionCount());
+            final var region0 = cacheService.get(cacheKey1, size(250), 0);
+            assertEquals(4, cacheService.freeRegionCount());
+            final var region1 = cacheService.get(cacheKey2, size(250), 1);
+            assertEquals(3, cacheService.freeRegionCount());
+            assertFalse(region0.isEvicted());
+            assertFalse(region1.isEvicted());
+            cacheService.forceEvictAsync(ck -> ck == cacheKey1);
+            assertFalse(region0.isEvicted());
+            assertFalse(region1.isEvicted());
+            // run the async task
+            taskQueue.runAllRunnableTasks();
+            assertTrue(region0.isEvicted());
+            assertFalse(region1.isEvicted());
+            assertEquals(4, cacheService.freeRegionCount());
+        }
+    }
+
     public void testDecay() throws IOException {
         // we have 8 regions
         Settings settings = Settings.builder()

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/allocation/SearchableSnapshotIndexEventListener.java

@@ -113,8 +113,10 @@ public void beforeIndexRemoved(IndexService indexService, IndexRemovalReason rea
                             shardId
                         );
                     }
-                    if (sharedBlobCacheService != null) {
-                        sharedBlobCacheService.forceEvict(SearchableSnapshots.forceEvictPredicate(shardId, indexSettings.getSettings()));
+                    if (indexSettings.getIndexMetadata().isPartialSearchableSnapshot() && sharedBlobCacheService != null) {
+                        sharedBlobCacheService.forceEvictAsync(
+                            SearchableSnapshots.forceEvictPredicate(shardId, indexSettings.getSettings())
+                        );
                     }
                 }
             }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/allocation/SearchableSnapshotIndexFoldersDeletionListener.java

@@ -72,8 +72,12 @@ private void markShardAsEvictedInCache(ShardId shardId, IndexSettings indexSetti
             shardId
         );
 
-        final SharedBlobCacheService<CacheKey> sharedBlobCacheService = this.frozenCacheServiceSupplier.get();
-        assert sharedBlobCacheService != null : "frozen cache service not initialized";
-        sharedBlobCacheService.forceEvict(SearchableSnapshots.forceEvictPredicate(shardId, indexSettings.getSettings()));
+        // Only partial searchable snapshots use the SharedBlobCacheService
+        if (indexSettings.getIndexMetadata().isPartialSearchableSnapshot()) {
+            final SharedBlobCacheService<CacheKey> sharedBlobCacheService =
+                SearchableSnapshotIndexFoldersDeletionListener.this.frozenCacheServiceSupplier.get();
+            assert sharedBlobCacheService != null : "frozen cache service not initialized";
+            sharedBlobCacheService.forceEvictAsync(SearchableSnapshots.forceEvictPredicate(shardId, indexSettings.getSettings()));
+        }
     }
 }