[JENKINS-26034] add option to fail fast on parallel branch failure.

Added an option "failFast: true" to the parallel step options that will terminate all running branches inside a parallel step if one of the branches fails.
jenkinsci · Mar 13, 2015 · 76f6422 · 76f6422
1 parent 40b3dd0
commit 76f6422
Show file tree

Hide file tree

Showing 4 changed files with 105 additions and 10 deletions.
diff --git a/aggregator/src/test/java/org/jenkinsci/plugins/workflow/steps/parallel/ParallelStepTest.java b/aggregator/src/test/java/org/jenkinsci/plugins/workflow/steps/parallel/ParallelStepTest.java
@@ -23,6 +23,7 @@
 import org.jenkinsci.plugins.workflow.support.visualization.table.FlowGraphTable;
 import org.jenkinsci.plugins.workflow.support.visualization.table.FlowGraphTable.Row;
 import org.jenkinsci.plugins.workflow.test.steps.SemaphoreStep;
+import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runners.model.Statement;
 import org.jvnet.hudson.test.Issue;
@@ -100,6 +101,47 @@ public void failure_in_subflow_will_cause_join_to_fail() throws Exception {
         });
     }
 
+
+    /**
+     * Failure in a branch will cause the join to fail.
+     */
+    @Test @Issue("JENKINS-26034")
+    public void failure_in_subflow_will_fail_fast() throws Exception {
+        story.addStep(new Statement() {
+            @Override public void evaluate() throws Throwable {
+                p = jenkins().createProject(WorkflowJob.class, "demo");
+                p.setDefinition(new CpsFlowDefinition(join(
+                    "import "+AbortException.class.getName(),
+                    "import "+ParallelStepException.class.getName(),
+
+                    "node {",
+                    "  try {",
+                    "    parallel(",
+                    "      b: { error 'died' },",
+
+                        // make sure this branch takes longer than a
+                    "      a: { sleep 10; writeFile text: '', file: 'b.done' },",
+                    "      failFast: true",
+                    "    )",
+                    "    assert false;",
+                    "  } catch (ParallelStepException e) {",
+                    "    assert e.name=='b'",
+                    "    assert e.cause instanceof AbortException",
+                    "  }",
+                    "}"
+                )));
+
+                startBuilding().get();
+                assertBuildCompletedSuccessfully();
+                Assert.assertFalse("b should have aborted", jenkins().getWorkspaceFor(p).child("b.done").exists());
+
+                buildTable();
+                shouldHaveParallelStepsInTheOrder("b","a");
+            }
+        });
+    }
+
+
     @Test
     public void localMethodCallWithinBranch() {
         story.addStep(new Statement() {

diff --git a/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/steps/ParallelStep.java b/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/steps/ParallelStep.java
@@ -15,6 +15,7 @@
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -30,13 +31,18 @@
  * @author Kohsuke Kawaguchi
  */
 public class ParallelStep extends Step {
+
+    /** should a failure in a parallel branch early terminate other branches. */
+    private final boolean failFast;
+
     /**
      * All the sub-workflows as {@link Closure}s, keyed by their names.
      */
     /*package*/ final Map<String,Closure> closures;
 
-    public ParallelStep(Map<String,Closure> closures) {
+    public ParallelStep(Map<String,Closure> closures, boolean failFast) {
         this.closures = closures;
+        this.failFast = failFast;
     }
 
     @Override
@@ -45,24 +51,44 @@ public StepExecution start(StepContext context) throws Exception {
         return new ParallelStepExecution(this, context);
     }
 
+    /*package*/ boolean isFailFast() {
+        return failFast;
+    }
+
+
     @PersistIn(PROGRAM)
     static class ResultHandler implements Serializable {
         private final StepContext context;
+        private final ParallelStepExecution stepExecution;
+        private final boolean failFast;
+        /** Have we called stop on the StepExecution? */
+        private boolean stopSent = false;
+
         /**
          * Collect the results of sub-workflows as they complete.
          * The key set is fully populated from the beginning.
          */
         private final Map<String,Outcome> outcomes = new HashMap<String, Outcome>();
 
-        ResultHandler(StepContext context) {
+        ResultHandler(StepContext context, ParallelStepExecution parallelStepExecution, boolean failFast) {
             this.context = context;
+            this.stepExecution = parallelStepExecution;
+            this.failFast = failFast;
         }
 
         Callback callbackFor(String name) {
             outcomes.put(name, null);
             return new Callback(this, name);
         }
 
+        private void stopSent() {
+            stopSent = true;
+        }
+
+        private boolean isStopSent() {
+            return stopSent;
+        }
+
         private static class Callback extends BodyExecutionCallback {
 
             private final ResultHandler handler;
@@ -91,8 +117,19 @@ private void checkAllDone() {
                 for (Entry<String,Outcome> e : handler.outcomes.entrySet()) {
                     Outcome o = e.getValue();
 
-                    if (o==null)
-                        return; // some of the results are not yet ready
+                    if (o==null) {
+                        // some of the results are not yet ready
+                        if (handler.failFast && ! handler.isStopSent()) {
+                            handler.stopSent();
+                            try {
+                                handler.stepExecution.stop(new InterruptedException("Interupted due to early termination of parallel step."));
+                            }
+                            catch (Exception ignored) {
+                                // ignored.
+                            }
+                        }
+                        return;
+                    }
                     if (o.isFailure()) {
                         failure= e;
                     } else {
@@ -124,15 +161,29 @@ public String getFunctionName() {
 
         @Override
         public Step newInstance(Map<String,Object> arguments) {
+            boolean earlyTermination = false;
+            Map<String,Closure<?>> closures = new LinkedHashMap<String, Closure<?>>();
             for (Entry<String,Object> e : arguments.entrySet()) {
-                if (!(e.getValue() instanceof Closure))
+                if ((e.getValue() instanceof Closure)) {
+                    closures.put(e.getKey(), (Closure<?>)e.getValue());
+                }
+                else if ("earlyTermination".equals(e.getKey())) {
+                    earlyTermination = Boolean.valueOf(e.getValue().toString());
+                }
+                else {
                     throw new IllegalArgumentException("Expected a closure but found "+e.getKey()+"="+e.getValue());
+                }
             }
-            return new ParallelStep((Map)arguments);
+            return new ParallelStep((Map)closures, earlyTermination);
         }
 
         @Override public Map<String,Object> defineArguments(Step step) throws UnsupportedOperationException {
-            return new TreeMap<String,Object>(((ParallelStep) step).closures);
+            ParallelStep ps = (ParallelStep) step;
+            Map<String,Object> retVal = new TreeMap<String,Object>(ps.closures);
+            if (ps.failFast) {
+                retVal.put("earlyTermination", Boolean.TRUE);
+            }
+            return retVal;
         }
 
         @Override

diff --git a/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/steps/ParallelStepExecution.java b/cps/src/main/java/org/jenkinsci/plugins/workflow/cps/steps/ParallelStepExecution.java
@@ -39,7 +39,7 @@ public boolean start() throws Exception {
         CpsStepContext cps = (CpsStepContext) getContext();
         CpsThread t = CpsThread.current();
 
-        ResultHandler r = new ResultHandler(cps);
+        ResultHandler r = new ResultHandler(cps, this, parallelStep.isFailFast());
 
         for (Entry<String,Closure> e : parallelStep.closures.entrySet()) {
             BodyExecution body = cps.newBodyInvoker(t.getGroup().export(e.getValue()))

diff --git a/cps/src/main/resources/org/jenkinsci/plugins/workflow/cps/steps/ParallelStep/config.jelly b/cps/src/main/resources/org/jenkinsci/plugins/workflow/cps/steps/ParallelStep/config.jelly
@@ -27,12 +27,14 @@ THE SOFTWARE.
 <j:jelly xmlns:j="jelly:core" xmlns:f="/lib/form">
     <f:block>
         <p>
-            (No visual configuration.) Takes a map from branch names to closures:
+            (No visual configuration.) Takes a map from branch names to closures and an optional argument <code>failFast</code>
+             which will terminate all branches upon a failure in any other branch:
         </p>
         <pre>parallel firstBranch: {
     // do something
 }, secondBranch: {
     // do something else
-}</pre>
+},
+failFast: true|false</pre>
     </f:block>
 </j:jelly>