Fix the memory leak by making StepExecutionIteratorImpl return Future…

…<List<Void>> instead of Future<List<StepExecution>>
jenkinsci · Aug 7, 2024 · 9b6c0ae · 9b6c0ae
1 parent ce222c4
commit 9b6c0ae
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 15 deletions.
diff --git a/src/main/java/org/jenkinsci/plugins/workflow/flow/FlowExecutionList.java b/src/main/java/org/jenkinsci/plugins/workflow/flow/FlowExecutionList.java
@@ -253,24 +253,26 @@ public ListenableFuture<?> apply(final Function<StepExecution, Void> f) {
 
             for (FlowExecution e : FlowExecutionList.get()) {
                 ListenableFuture<List<StepExecution>> execs = e.getCurrentExecutions(false);
-                all.add(execs);
-                Futures.addCallback(execs,new FutureCallback<List<StepExecution>>() {
-                    @Override
-                    public void onSuccess(@NonNull List<StepExecution> result) {
-                        for (StepExecution e : result) {
-                            try {
-                                f.apply(e);
-                            } catch (RuntimeException x) {
-                                LOGGER.log(Level.WARNING, null, x);
-                            }
+                // We transform the futures that return List<StepExecution> into futures that return Void before
+                // passing them to Futures.allAsList so that the combined future only holds strong references to each
+                // FlowExecution until its StepExecutions have been loaded and applied to the function. This prevents
+                // us from leaking references to all processed executions in the case where a single build has a stuck
+                // CpsVmExecutorService that prevents the future returned by getCurrentExecutions from completing.
+                ListenableFuture<Void> results = Futures.transform(execs, (List<StepExecution> result) -> {
+                    for (StepExecution se : result) {
+                        try {
+                            f.apply(se);
+                        } catch (RuntimeException x) {
+                            LOGGER.log(Level.WARNING, null, x);
                         }
                     }
-
-                    @Override
-                    public void onFailure(@NonNull Throwable t) {
-                        LOGGER.log(Level.WARNING, null, t);
-                    }
+                    return null;
+                }, MoreExecutors.directExecutor());
+                ListenableFuture<Void> resultsWithWarningsLogged = Futures.catching(results, Throwable.class, t -> {
+                    LOGGER.log(Level.WARNING, null, t);
+                    return null;
                 }, MoreExecutors.directExecutor());
+                all.add(resultsWithWarningsLogged);
             }
 
             return Futures.allAsList(all);

diff --git a/src/test/java/org/jenkinsci/plugins/workflow/flow/FlowExecutionListTest.java b/src/test/java/org/jenkinsci/plugins/workflow/flow/FlowExecutionListTest.java
@@ -38,12 +38,14 @@
 import hudson.model.TaskListener;
 import hudson.model.queue.QueueTaskFuture;
 import java.io.Serializable;
+import java.lang.ref.WeakReference;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Collections;
 import java.util.Set;
 import java.util.function.Supplier;
 import java.util.logging.Level;
+import jenkins.model.Jenkins;
 import org.hamcrest.Matcher;
 import org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition;
 import org.jenkinsci.plugins.workflow.job.WorkflowJob;
@@ -61,6 +63,7 @@
 import org.jvnet.hudson.test.Issue;
 import org.jvnet.hudson.test.LoggerRule;
 import org.jvnet.hudson.test.JenkinsSessionRule;
+import org.jvnet.hudson.test.MemoryAssert;
 import org.jvnet.hudson.test.TestExtension;
 import org.jvnet.hudson.test.recipes.LocalData;
 import org.kohsuke.stapler.DataBoundConstructor;
@@ -188,6 +191,36 @@ public class FlowExecutionListTest {
         });
     }
 
+    @Ignore("MemoryAssert is unable to clear the SoftReference from AbstractLazyLoadRunMap. Heap dumps show a strong reference from the test execution thread as a local variable, which is not captured by MemoryAssert. Also needs to be moved to a class with no BuildWatcher.")
+    @Test public void stepExecutionIteratorDoesNotLeakBuildsWhenOneIsStuck2() throws Throwable {
+        sessions.then(r -> {
+            WeakReference<Object> buildRef;
+            {
+                var stuck = r.createProject(WorkflowJob.class, "stuck");
+                stuck.setDefinition(new CpsFlowDefinition("parallel(one: { echo 'test message'; Thread.sleep(Integer.MAX_VALUE); })", false));
+                var stuckBuild = stuck.scheduleBuild2(0).waitForStart();
+
+                var notStuck = r.createProject(WorkflowJob.class, "not-stuck");
+                notStuck.setDefinition(new CpsFlowDefinition("semaphore('wait')", true));
+                var notStuckBuild = notStuck.scheduleBuild2(0).waitForStart();
+                buildRef = new WeakReference<>(notStuckBuild);
+
+                SemaphoreStep.waitForStart("wait/1", notStuckBuild);
+                r.waitForMessage("test message", stuckBuild);
+                Thread.sleep(1000); // We need Thread.sleep to be running in the CpsVmExecutorService.
+
+                // Make FlowExecutionList$StepExecutionIteratorImpl.applyAll submit a task to the CpsVmExecutorService
+                // for stuck #1 that will never complete, so the resulting future will never complete.
+                var future = StepExecution.applyAll(e -> null);
+
+                SemaphoreStep.success("wait/1", null);
+                r.waitForCompletion(notStuckBuild);
+                Jenkins.get().getQueue().clearLeftItems(); // Clean up the soft reference immediately.
+            }
+            MemoryAssert.assertGC(buildRef, true);
+        });
+    }
+
     public static class NonResumableStep extends Step implements Serializable {
         public static final long serialVersionUID = 1L;
         @DataBoundConstructor