Merge pull request #213 from jglick/many-steps-JENKINS-30055

[JENKINS-30055] Handle a large number of steps in quick succession
jenkinsci · Jan 11, 2016 · 0fdf534 · 0fdf534
2 parents 97ce55c + 3686605
commit 0fdf534
Show file tree

Hide file tree

Showing 14 changed files with 338 additions and 180 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -4,6 +4,7 @@ Only noting significant user changes, not internal code cleanups and minor bug f
 
 ## 1.13 (upcoming)
 
+* [JENKINS-30055](https://issues.jenkins-ci.org/browse/JENKINS-30055): poor performance and file handle leaks when running a script with an enormous number of steps in quick succession
 * [JENKINS-26126](https://issues.jenkins-ci.org/browse/JENKINS-26126) (partial): introspect Workflow steps to generate static reference documentation (link from _Snippet Generator_). Planned to be used for IDE auto-completion as well.
 * [JENKINS-31614](https://issues.jenkins-ci.org/browse/JENKINS-31614): avoiding various deadlocks involving `Queue`.
 * [JENKINS-31897](https://issues.jenkins-ci.org/browse/JENKINS-31897): parameters with default values may now be omitted from the `parameters` option to the `build` step.

diff --git a/aggregator/src/test/java/org/jenkinsci/plugins/workflow/ScalabilityTest.java b/aggregator/src/test/java/org/jenkinsci/plugins/workflow/ScalabilityTest.java
@@ -0,0 +1,70 @@
+/*
+ * The MIT License
+ *
+ * Copyright 2015 CloudBees, Inc.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package org.jenkinsci.plugins.workflow;
+
+import java.util.List;
+import org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition;
+import org.jenkinsci.plugins.workflow.flow.FlowExecution;
+import org.jenkinsci.plugins.workflow.graph.FlowNode;
+import org.jenkinsci.plugins.workflow.job.WorkflowJob;
+import org.jenkinsci.plugins.workflow.job.WorkflowRun;
+import static org.junit.Assert.*;
+import org.junit.Test;
+import org.junit.Rule;
+import org.junit.runners.model.Statement;
+import org.jvnet.hudson.test.Issue;
+import org.jvnet.hudson.test.RestartableJenkinsRule;
+
+public class ScalabilityTest {
+
+    @Rule public RestartableJenkinsRule story = new RestartableJenkinsRule();
+
+    @Issue("JENKINS-30055")
+    @Test public void manySteps() {
+        story.addStep(new Statement() {
+            @Override public void evaluate() throws Throwable {
+                WorkflowJob p = story.j.jenkins.createProject(WorkflowJob.class, "p");
+                p.setDefinition(new CpsFlowDefinition("for (int i = 0; i < 10000; i++) {echo \"iteration #${i}\"}", true));
+                WorkflowRun b = story.j.assertBuildStatusSuccess(p.scheduleBuild2(0));
+                story.j.assertLogContains("iteration #9876", b); // arbitrary iteration close to the end
+            }
+        });
+        story.addStep(new Statement() {
+            @Override public void evaluate() throws Throwable {
+                WorkflowJob p = story.j.jenkins.getItemByFullName("p", WorkflowJob.class);
+                WorkflowRun b = p.getLastBuild();
+                story.j.assertLogContains("iteration #5432", b);
+                FlowExecution execution = b.getExecution();
+                assertNotNull(execution);
+                FlowNode node = execution.getNode("5678");
+                assertNotNull(node);
+                List<FlowNode> parents = node.getParents();
+                assertEquals(1, parents.size());
+                assertEquals("5677", parents.get(0).getId());
+            }
+        });
+    }
+
+}
diff --git a/api/src/main/java/org/jenkinsci/plugins/workflow/flow/FlowExecution.java b/api/src/main/java/org/jenkinsci/plugins/workflow/flow/FlowExecution.java
@@ -158,8 +158,17 @@ public String getUrl() throws IOException {
      */
     public abstract void interrupt(Result r, CauseOfInterruption... causes) throws IOException, InterruptedException;
 
+    /**
+     * Add a listener to changes in the flow graph structure.
+     * @param listener a listener to add
+     */
     public abstract void addListener(GraphListener listener);
 
+    /**
+     * Reverse of {@link #addListener}.
+     */
+    public /*abstract*/ void removeListener(GraphListener listener) {}
+
     /**
      * Checks whether this flow execution has finished executing completely.
      */

diff --git a/api/src/main/java/org/jenkinsci/plugins/workflow/flow/GraphListener.java b/api/src/main/java/org/jenkinsci/plugins/workflow/flow/GraphListener.java
@@ -38,4 +38,12 @@ public interface GraphListener {
      * One of the use cases of this listener is to persist the state of {@link FlowExecution}.
      */
     void onNewHead(FlowNode node);
+
+    /**
+     * Listener which should be notified of events immediately as they occur.
+     * You must be very careful not to acquire locks or block.
+     * If you do not implement this marker interface, you will receive notifications in batched deliveries.
+     */
+    interface Synchronous extends GraphListener {}
+
 }
diff --git a/api/src/main/java/org/jenkinsci/plugins/workflow/graph/BlockEndNode.java b/api/src/main/java/org/jenkinsci/plugins/workflow/graph/BlockEndNode.java
@@ -24,29 +24,50 @@
 
 package org.jenkinsci.plugins.workflow.graph;
 
+import java.io.IOException;
 import org.jenkinsci.plugins.workflow.flow.FlowExecution;
 
 import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.Nonnull;
 
 /**
- * @author Kohsuke Kawaguchi
- * @author Jesse Glick
+ * End of a block.
  * @see BlockStartNode
  */
 public abstract class BlockEndNode<START extends BlockStartNode> extends FlowNode {
-    private final START start;
+    private transient START start;
+    private final String startId;
 
     public BlockEndNode(FlowExecution exec, String id, START start, FlowNode... parents) {
         super(exec, id, parents);
         this.start = start;
+        startId = start.getId();
     }
 
     public BlockEndNode(FlowExecution exec, String id, START start, List<FlowNode> parents) {
         super(exec, id, parents);
         this.start = start;
+        startId = start.getId();
     }
 
-    public START getStartNode() {
+    /**
+     * Returns the matching start node.
+     * @return an earlier node matching this block
+     * @throws IllegalStateException if the start node could not be reloaded after deserialization
+     */
+    public @Nonnull START getStartNode() {
+        if (start == null) {
+            try {
+                start = (START) getExecution().getNode(startId);
+                if (start == null) {
+                    throw new IllegalStateException("Matching start node " + startId + " lost from deserialization");
+                }
+            } catch (IOException x) {
+                throw new IllegalStateException("Could not load matching start node: " + x);
+            }
+        }
         return start;
     }
 

diff --git a/api/src/main/java/org/jenkinsci/plugins/workflow/graph/FlowNode.java b/api/src/main/java/org/jenkinsci/plugins/workflow/graph/FlowNode.java
@@ -35,6 +35,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.logging.Level;
 import static java.util.logging.Level.*;
 import java.util.logging.Logger;
 import javax.annotation.CheckForNull;
@@ -53,26 +54,37 @@
  */
 @ExportedBean
 public abstract class FlowNode extends Actionable implements Saveable {
-    private final List<FlowNode> parents;
+    private transient List<FlowNode> parents;
+    private final List<String> parentIds;
 
     private final String id;
 
     // this is a copy-on-write array so synchronization isn't needed between reader & writer.
     @edu.umd.cs.findbugs.annotations.SuppressWarnings("IS2_INCONSISTENT_SYNC")
-    private transient CopyOnWriteArrayList<Action> actions;
+    private transient CopyOnWriteArrayList<Action> actions = new CopyOnWriteArrayList<Action>();
 
     private transient final FlowExecution exec;
 
     protected FlowNode(FlowExecution exec, String id, List<FlowNode> parents) {
         this.id = id;
         this.exec = exec;
         this.parents = ImmutableList.copyOf(parents);
+        parentIds = ids();
     }
 
     protected FlowNode(FlowExecution exec, String id, FlowNode... parents) {
         this.id = id;
         this.exec = exec;
         this.parents = ImmutableList.copyOf(parents);
+        parentIds = ids();
+    }
+
+    private List<String> ids() {
+        List<String> ids = new ArrayList<String>(parents.size());
+        for (FlowNode n : parents) {
+            ids.add(n.id);
+        }
+        return ids;
     }
 
     /**
@@ -103,9 +115,24 @@ public final boolean isRunning() {
      * Returns a read-only view of parents.
      */
     public List<FlowNode> getParents() {
+        if (parents == null) {
+            parents = loadParents(parentIds);
+        }
         return parents;
     }
 
+    private List<FlowNode> loadParents(List<String> parentIds) {
+        List<FlowNode> _parents = new ArrayList<FlowNode>(parentIds.size());
+        for (String parentId : parentIds) {
+            try {
+                _parents.add(exec.getNode(parentId));
+            } catch (IOException x) {
+                LOGGER.log(Level.WARNING, "failed to load parents of " + id, x);
+            }
+        }
+        return _parents;
+    }
+
     @Restricted(DoNotUse.class)
     @Exported(name="parents")
     public List<String> getParentIds() {

diff --git a/api/src/main/java/org/jenkinsci/plugins/workflow/graph/ForkNode.java b/api/src/main/java/org/jenkinsci/plugins/workflow/graph/ForkNode.java
@@ -27,9 +27,9 @@
 import org.jenkinsci.plugins.workflow.flow.FlowExecution;
 
 /**
- * @author Kohsuke Kawaguchi
- * @author Jesse Glick
+ * @deprecated unused
  */
+@Deprecated
 public class ForkNode extends BlockStartNode {
     public ForkNode(FlowExecution storage, String id, FlowNode... parents) {
         super(storage, id, parents);

diff --git a/api/src/main/java/org/jenkinsci/plugins/workflow/graph/JoinNode.java b/api/src/main/java/org/jenkinsci/plugins/workflow/graph/JoinNode.java
@@ -29,9 +29,9 @@
 import java.util.List;
 
 /**
- * @author Kohsuke Kawaguchi
- * @author Jesse Glick
+ * @deprecated unused
  */
+@Deprecated
 public class JoinNode extends BlockEndNode<ForkNode> {
     public JoinNode(FlowExecution exec, String id, ForkNode forkNode, List<FlowNode> parents) {
         super(exec, id, forkNode, parents);

diff --git a/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/CpsFlowExecution.java b/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/CpsFlowExecution.java
@@ -93,7 +93,10 @@
 
 import static com.thoughtworks.xstream.io.ExtendedHierarchicalStreamWriterHelper.*;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import hudson.BulkChange;
 import hudson.init.Terminator;
+import hudson.model.Queue;
+import hudson.model.Saveable;
 import hudson.model.User;
 import hudson.security.ACL;
 import java.beans.Introspector;
@@ -714,6 +717,12 @@ public void addListener(GraphListener listener) {
         listeners.add(listener);
     }
 
+    @Override public void removeListener(GraphListener listener) {
+        if (listeners != null) {
+            listeners.remove(listener);
+        }
+    }
+
     @Override
     public void interrupt(Result result, CauseOfInterruption... causes) throws IOException, InterruptedException {
         setResult(result);
@@ -793,10 +802,36 @@ synchronized FlowHead getFirstHead() {
         return heads.firstEntry().getValue();
     }
 
-    void notifyListeners(FlowNode node) {
+    void notifyListeners(List<FlowNode> nodes, boolean synchronous) {
         if (listeners != null) {
-            for (GraphListener listener : listeners) {
-                listener.onNewHead(node);
+            Saveable s = Saveable.NOOP;
+            try {
+                Queue.Executable exec = owner.getExecutable();
+                if (exec instanceof Saveable) {
+                    s = (Saveable) exec;
+                }
+            } catch (IOException x) {
+                LOGGER.log(Level.WARNING, "failed to notify listeners of changes to " + nodes + " in " + this, x);
+            }
+            BulkChange bc = new BulkChange(s);
+            try {
+                for (FlowNode node : nodes) {
+                    for (GraphListener listener : listeners) {
+                        if (listener instanceof GraphListener.Synchronous == synchronous) {
+                            listener.onNewHead(node);
+                        }
+                    }
+                }
+            } finally {
+                if (synchronous) {
+                    bc.abort(); // hack to skip save—we are holding a lock
+                } else {
+                    try {
+                        bc.commit();
+                    } catch (IOException x) {
+                        LOGGER.log(Level.WARNING, null, x);
+                    }
+                }
             }
         }
     }

diff --git a/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/CpsThreadGroup.java b/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/CpsThreadGroup.java
@@ -42,7 +42,10 @@
 import java.io.Serializable;
 import java.nio.file.Files;
 import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.TreeMap;
@@ -307,6 +310,8 @@ private void run() throws IOException {
         }
     }
 
+    private transient List<FlowNode> nodesToNotify;
+    private static final Object nodesToNotifyLock = new Object();
     /**
      * Notifies listeners of the new {@link FlowHead}.
      *
@@ -315,9 +320,24 @@ private void run() throws IOException {
     @CpsVmThreadOnly
     /*package*/ void notifyNewHead(final FlowNode head) {
         assertVmThread();
+        execution.notifyListeners(Collections.singletonList(head), true);
+        synchronized (nodesToNotifyLock) {
+            if (nodesToNotify == null) {
+                nodesToNotify = new ArrayList<FlowNode>();
+            }
+            nodesToNotify.add(head);
+        }
         runner.execute(new Runnable() {
             public void run() {
-                execution.notifyListeners(head);
+                List<FlowNode> _nodesToNotify;
+                synchronized (nodesToNotifyLock) {
+                    if (nodesToNotify == null) {
+                        return;
+                    }
+                    _nodesToNotify = nodesToNotify;
+                    nodesToNotify = null;
+                }
+                execution.notifyListeners(_nodesToNotify, false);
             }
         });
     }

diff --git a/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/FlowHead.java b/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/FlowHead.java
@@ -30,6 +30,7 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
+import java.util.Collections;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.jenkinsci.plugins.workflow.actions.ErrorAction;
@@ -119,7 +120,9 @@ void setNewHead(FlowNode v) {
             } else {
                 // in recovering from error and such situation, we sometimes need to grow the graph
                 // without running the program.
-                execution.notifyListeners(v);
+                // TODO can CpsThreadGroup.notifyNewHead be used instead to notify both kinds of listeners?
+                execution.notifyListeners(Collections.singletonList(v), true);
+                execution.notifyListeners(Collections.singletonList(v), false);
             }
         } catch (IOException e) {
             LOGGER.log(Level.FINE, "Failed to record new head: " + v, e);