Test and fix for wrong order for in-progress parallels [JENKINS-38536]

jenkinsci · Oct 4, 2016 · 2ac612a · 2ac612a
1 parent 42a296e
commit 2ac612a
Show file tree

Hide file tree

Showing 3 changed files with 97 additions and 2 deletions.
diff --git a/pom.xml b/pom.xml
@@ -88,5 +88,12 @@
             <version>1.15</version>
             <scope>test</scope>
         </dependency>
+        <dependency>  <!-- For Semaphore step -->
+            <groupId>org.jenkins-ci.plugins.workflow</groupId>
+            <artifactId>workflow-support</artifactId>
+            <version>2.1</version>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
     </dependencies>
 </project>
diff --git a/src/main/java/org/jenkinsci/plugins/workflow/graphanalysis/ForkScanner.java b/src/main/java/org/jenkinsci/plugins/workflow/graphanalysis/ForkScanner.java
@@ -27,6 +27,7 @@
 import com.google.common.base.Predicate;
 import com.google.common.base.Predicates;
 import org.jenkinsci.plugins.workflow.actions.ThreadNameAction;
+import org.jenkinsci.plugins.workflow.actions.TimingAction;
 import org.jenkinsci.plugins.workflow.graph.BlockEndNode;
 import org.jenkinsci.plugins.workflow.graph.BlockStartNode;
 import org.jenkinsci.plugins.workflow.graph.FlowEndNode;
@@ -553,11 +554,40 @@ public static void visitSimpleChunks(@Nonnull Collection<FlowNode> heads, @Nonnu
         scanner.visitSimpleChunks(visitor, finder);
     }
 
-    /** Walk through flows  */
+    @CheckForNull
+    private static FlowNode findLastStartedNode(@Nonnull List<FlowNode> candidates) {
+        if (candidates.size() == 0) {
+            return null;
+        } else if (candidates.size() == 1) {
+            return candidates.get(0);
+        } else {
+            FlowNode returnOut = candidates.get(0);
+            long startTime = Long.MIN_VALUE;
+            for(FlowNode f : candidates) {
+                TimingAction ta = f.getAction(TimingAction.class);
+                if (ta != null) {
+                    long myStart = ta.getStartTime();
+                    if (myStart > startTime) {
+                        returnOut = f;
+                        startTime = myStart;
+                    }
+                }
+            }
+            return returnOut;
+        }
+    }
+
+    /** Walk through flows */
     public void visitSimpleChunks(@Nonnull SimpleChunkVisitor visitor, @Nonnull ChunkFinder finder) {
         FlowNode prev = null;
         if (finder.isStartInsideChunk() && hasNext()) {
-            visitor.chunkEnd(this.myNext, null, this);
+            if (currentParallelStart == null ) {
+                visitor.chunkEnd(this.myNext, null, this);
+            } else { // Last node is the last started branch
+                List<FlowNode> branchEnds = new ArrayList<FlowNode>(currentParallelStart.unvisited);
+                branchEnds.add(this.myNext);
+                visitor.chunkEnd(this.findLastStartedNode(branchEnds), null, this);
+            }
         }
         while(hasNext()) {
             prev = (myCurrent != myNext) ? myCurrent : null;

diff --git a/src/test/java/org/jenkinsci/plugins/workflow/graphanalysis/ForkScannerTest.java b/src/test/java/org/jenkinsci/plugins/workflow/graphanalysis/ForkScannerTest.java
@@ -29,6 +29,7 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition;
+import org.jenkinsci.plugins.workflow.cps.nodes.StepAtomNode;
 import org.jenkinsci.plugins.workflow.cps.nodes.StepStartNode;
 import org.jenkinsci.plugins.workflow.cps.steps.ParallelStep;
 import org.jenkinsci.plugins.workflow.flow.FlowExecution;
@@ -37,6 +38,7 @@
 import org.jenkinsci.plugins.workflow.graph.FlowNode;
 import org.jenkinsci.plugins.workflow.job.WorkflowJob;
 import org.jenkinsci.plugins.workflow.job.WorkflowRun;
+import org.jenkinsci.plugins.workflow.test.steps.SemaphoreStep;
 import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Rule;
@@ -536,4 +538,60 @@ public void testTripleParallel() throws Exception {
         Assert.assertTrue(pbs.unvisited.contains(exec.getNode("8")));
         Assert.assertTrue(pbs.unvisited.contains(exec.getNode("9")));
     }
+
+    private void testParallelFindsLast(WorkflowJob job, String semaphoreName) throws Exception {
+        ForkScanner scan = new ForkScanner();
+        ChunkFinder labelFinder = new LabelledChunkFinder();
+
+        System.out.println("Testing that semaphore step is always the last step for chunk with "+job.getName());
+        WorkflowRun run  = job.scheduleBuild2(0).getStartCondition().get();
+        SemaphoreStep.waitForStart(semaphoreName+"/1", run);
+
+            /*if (run.getExecution() == null) {
+                Thread.sleep(1000);
+            }*/
+
+        TestVisitor visitor = new TestVisitor();
+        scan.setup(run.getExecution().getCurrentHeads());
+        scan.visitSimpleChunks(visitor, labelFinder);
+        TestVisitor.CallEntry entry = visitor.calls.get(0);
+        Assert.assertEquals(TestVisitor.CallType.CHUNK_END, entry.type);
+        FlowNode lastNode = run.getExecution().getNode(Integer.toString(entry.ids[0]));
+        Assert.assertEquals("Wrong End Node: ("+lastNode.getId()+") "+lastNode.getDisplayName(), "semaphore", lastNode.getDisplayFunctionName());
+
+        SemaphoreStep.success(semaphoreName+"/1", null);
+        r.waitForCompletion(run);
+    }
+
+    @Test
+    public void testParallelCorrectEndNodeForVisitor() throws Exception {
+        WorkflowJob jobPauseFirst = r.jenkins.createProject(WorkflowJob.class, "PauseFirst");
+        jobPauseFirst.setDefinition(new CpsFlowDefinition("" +
+                "stage 'primero'\n" +
+                "parallel 'wait' : {sleep 1; semaphore 'wait1';}, \n" +
+                " 'final': { echo 'succeed';} "
+        ));
+
+        WorkflowJob jobPauseSecond = r.jenkins.createProject(WorkflowJob.class, "PauseSecond");
+        jobPauseSecond.setDefinition(new CpsFlowDefinition("" +
+                "stage 'primero'\n" +
+                "parallel 'success' : {echo 'succeed'}, \n" +
+                " 'pause':{ sleep 1; semaphore 'wait2'; }\n"
+                ));
+
+        WorkflowJob jobPauseMiddle = r.jenkins.createProject(WorkflowJob.class, "PauseMiddle");
+        jobPauseSecond.setDefinition(new CpsFlowDefinition("" +
+                "stage 'primero'\n" +
+                "parallel 'success' : {echo 'succeed'}, \n" +
+                " 'pause':{ sleep 1; semaphore 'wait3'; }, \n" +
+                " 'final': { echo 'succeed-final';} "
+        ));
+
+        ForkScanner scan = new ForkScanner();
+        ChunkFinder labelFinder = new LabelledChunkFinder();
+
+        testParallelFindsLast(jobPauseFirst, "wait1");
+        testParallelFindsLast(jobPauseSecond, "wait2");
+        testParallelFindsLast(jobPauseMiddle, "wait3");
+    }
 }