Avoid capturing SnapshotsInProgress$Entry in queue (elastic#88726)

Today each time there's shards to snapshot we enqueue a lambda which captures the current `SnapshotsInProgress$Entry`. This is a pretty heavyweight object, possibly several MB in size, most of which is not necessary to capture, and with concurrent snapshots across thousands of shards we may enqueue many hundreds of slightly different such objects. With this commit we compute a more efficient representation of the work to be done by each task in the queue instead. Relates elastic#77466 Backport of elastic#88707
dnhatn · Jul 22, 2022 · b3eb52d · b3eb52d
1 parent 17fde1e
commit b3eb52d
Show file tree

Hide file tree

Showing 2 changed files with 72 additions and 55 deletions.
diff --git a/docs/changelog/88707.yaml b/docs/changelog/88707.yaml
@@ -0,0 +1,5 @@
+pr: 88707
+summary: Avoid capturing `SnapshotsInProgress$Entry` in queue
+area: Snapshot/Restore
+type: bug
+issues: []
diff --git a/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java b/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java
@@ -56,6 +56,7 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -228,7 +229,24 @@ private void startNewSnapshots(List<SnapshotsInProgress.Entry> snapshotsInProgre
                 }
                 if (startedShards != null && startedShards.isEmpty() == false) {
                     shardSnapshots.computeIfAbsent(snapshot, s -> new HashMap<>()).putAll(startedShards);
-                    startNewShards(entry, startedShards);
+
+                    final List<Runnable> shardSnapshotTasks = new ArrayList<>(startedShards.size());
+                    for (final Map.Entry<ShardId, IndexShardSnapshotStatus> shardEntry : startedShards.entrySet()) {
+                        final ShardId shardId = shardEntry.getKey();
+                        final IndexShardSnapshotStatus snapshotStatus = shardEntry.getValue();
+                        final IndexId indexId = entry.indices().get(shardId.getIndexName());
+                        assert indexId != null;
+                        assert SnapshotsService.useShardGenerations(entry.version())
+                            || ShardGenerations.fixShardGeneration(snapshotStatus.generation()) == null
+                            : "Found non-null, non-numeric shard generation ["
+                                + snapshotStatus.generation()
+                                + "] for snapshot with old-format compatibility";
+                        shardSnapshotTasks.add(
+                            newShardSnapshotTask(shardId, snapshot, indexId, entry.userMetadata(), snapshotStatus, entry.version())
+                        );
+                    }
+
+                    threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(() -> shardSnapshotTasks.forEach(Runnable::run));
                 }
             } else if (entryState == State.ABORTED) {
                 // Abort all running shards for this snapshot
@@ -251,63 +269,57 @@ private void startNewSnapshots(List<SnapshotsInProgress.Entry> snapshotsInProgre
         }
     }
 
-    private void startNewShards(SnapshotsInProgress.Entry entry, Map<ShardId, IndexShardSnapshotStatus> startedShards) {
-        threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(() -> {
-            final Snapshot snapshot = entry.snapshot();
-            for (final Map.Entry<ShardId, IndexShardSnapshotStatus> shardEntry : startedShards.entrySet()) {
-                final ShardId shardId = shardEntry.getKey();
-                final IndexShardSnapshotStatus snapshotStatus = shardEntry.getValue();
-                final IndexId indexId = entry.indices().get(shardId.getIndexName());
-                assert indexId != null;
-                assert SnapshotsService.useShardGenerations(entry.version())
-                    || ShardGenerations.fixShardGeneration(snapshotStatus.generation()) == null
-                    : "Found non-null, non-numeric shard generation ["
-                        + snapshotStatus.generation()
-                        + "] for snapshot with old-format compatibility";
-                snapshot(
-                    shardId,
-                    snapshot,
-                    indexId,
-                    entry.userMetadata(),
-                    snapshotStatus,
-                    entry.version(),
-                    new ActionListener<ShardSnapshotResult>() {
-                        @Override
-                        public void onResponse(ShardSnapshotResult shardSnapshotResult) {
-                            final ShardGeneration newGeneration = shardSnapshotResult.getGeneration();
-                            assert newGeneration != null;
-                            assert newGeneration.equals(snapshotStatus.generation());
-                            if (logger.isDebugEnabled()) {
-                                final IndexShardSnapshotStatus.Copy lastSnapshotStatus = snapshotStatus.asCopy();
-                                logger.debug(
-                                    "[{}][{}] completed snapshot to [{}] with status [{}] at generation [{}]",
-                                    shardId,
-                                    snapshot,
-                                    snapshot.getRepository(),
-                                    lastSnapshotStatus,
-                                    snapshotStatus.generation()
-                                );
-                            }
-                            notifySuccessfulSnapshotShard(snapshot, shardId, shardSnapshotResult);
-                        }
+    private Runnable newShardSnapshotTask(
+        final ShardId shardId,
+        final Snapshot snapshot,
+        final IndexId indexId,
+        final Map<String, Object> userMetadata,
+        final IndexShardSnapshotStatus snapshotStatus,
+        final Version entryVersion
+    ) {
+        // separate method to make sure this lambda doesn't capture any heavy local objects like a SnapshotsInProgress.Entry
+        return () -> snapshot(
+            shardId,
+            snapshot,
+            indexId,
+            userMetadata,
+            snapshotStatus,
+            entryVersion,
+            new ActionListener<ShardSnapshotResult>() {
+                @Override
+                public void onResponse(ShardSnapshotResult shardSnapshotResult) {
+                    final ShardGeneration newGeneration = shardSnapshotResult.getGeneration();
+                    assert newGeneration != null;
+                    assert newGeneration.equals(snapshotStatus.generation());
+                    if (logger.isDebugEnabled()) {
+                        final IndexShardSnapshotStatus.Copy lastSnapshotStatus = snapshotStatus.asCopy();
+                        logger.debug(
+                            "[{}][{}] completed snapshot to [{}] with status [{}] at generation [{}]",
+                            shardId,
+                            snapshot,
+                            snapshot.getRepository(),
+                            lastSnapshotStatus,
+                            snapshotStatus.generation()
+                        );
+                    }
+                    notifySuccessfulSnapshotShard(snapshot, shardId, shardSnapshotResult);
+                }
 
-                        @Override
-                        public void onFailure(Exception e) {
-                            final String failure;
-                            if (e instanceof AbortedSnapshotException) {
-                                failure = "aborted";
-                                logger.debug(() -> new ParameterizedMessage("[{}][{}] aborted shard snapshot", shardId, snapshot), e);
-                            } else {
-                                failure = summarizeFailure(e);
-                                logger.warn(() -> new ParameterizedMessage("[{}][{}] failed to snapshot shard", shardId, snapshot), e);
-                            }
-                            snapshotStatus.moveToFailed(threadPool.absoluteTimeInMillis(), failure);
-                            notifyFailedSnapshotShard(snapshot, shardId, failure, snapshotStatus.generation());
-                        }
+                @Override
+                public void onFailure(Exception e) {
+                    final String failure;
+                    if (e instanceof AbortedSnapshotException) {
+                        failure = "aborted";
+                        logger.debug(() -> new ParameterizedMessage("[{}][{}] aborted shard snapshot", shardId, snapshot), e);
+                    } else {
+                        failure = summarizeFailure(e);
+                        logger.warn(() -> new ParameterizedMessage("[{}][{}] failed to snapshot shard", shardId, snapshot), e);
                     }
-                );
+                    snapshotStatus.moveToFailed(threadPool.absoluteTimeInMillis(), failure);
+                    notifyFailedSnapshotShard(snapshot, shardId, failure, snapshotStatus.generation());
+                }
             }
-        });
+        );
     }
 
     // package private for testing