Skip to content

Commit

Permalink
Merge pull request #20 from jenkinsci/final-parallel-branch-ordering
Browse files Browse the repository at this point in the history 
Ensure that for SimpleBlockVisitor we find the REAL last branch for a parallel [JENKINS-38536]
  • Loading branch information
@svanoort
svanoort committed Nov 30, 2016
2 parents 92ab59a + 7514135 commit eca21af
Show file tree
Hide file tree
Showing 3 changed files with 126 additions and 19 deletions.
7 changes: 7 additions & 0 deletions pom.xml
View file
Expand Up @@ -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>
45 changes: 42 additions & 3 deletions src/main/java/org/jenkinsci/plugins/workflow/graphanalysis/ForkScanner.java
View file
Expand Up @@ -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;
Expand Down Expand Up @@ -328,7 +329,7 @@ ArrayDeque<ParallelBlockStart> leastCommonAncestor(@Nonnull final Set<FlowNode>

// Walk through, merging flownodes one-by-one until everything has merged to one ancestor
boolean mergedAll = false;
// Ends when we merged all branches together, or hit the start of the flow without it
// Ends when we merged all branches together, or hit the start of the flow without it
while (!mergedAll && iterators.size() > 0) {
ListIterator<Filterator<FlowNode>> itIterator = iterators.listIterator();
ListIterator<FlowPiece> pieceIterator = livePieces.listIterator();
Expand Down Expand Up @@ -557,11 +558,49 @@ public static void visitSimpleChunks(@Nonnull Collection<FlowNode> heads, @Nonnu
scanner.visitSimpleChunks(visitor, finder);
}

/** Walk through flows */
/** Ensures we find the last *begun* node when there are multiple heads (parallel branches)
* This means that the simpleBlockVisitor gets the *actual* last node, not just the end of the last declared branch
* (See issue JENKINS-38536)
*/
@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);
// Null timing with multiple heads is probably a node where the GraphListener hasn't fired to add TimingAction yet
long myStart = (ta == null) ? System.currentTimeMillis() : 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);
FlowNode lastStarted = this.findLastStartedNode(branchEnds);
if (lastStarted != null) {
visitor.chunkEnd(lastStarted, null, this);
} else {
throw new IllegalStateException("Flow is inside parallel block, but shows no executing heads!");
}

}
}
while(hasNext()) {
prev = (myCurrent != myNext) ? myCurrent : null;
Expand Down
93 changes: 77 additions & 16 deletions src/test/java/org/jenkinsci/plugins/workflow/graphanalysis/ForkScannerTest.java
View file
Expand Up @@ -29,18 +29,21 @@
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;
import org.jenkinsci.plugins.workflow.graph.BlockStartNode;
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;
import org.junit.Test;
import org.jvnet.hudson.test.BuildWatcher;
import org.jvnet.hudson.test.Issue;
import org.jvnet.hudson.test.JenkinsRule;
import org.junit.Assert;

Expand Down Expand Up @@ -405,25 +408,25 @@ public void testLeastCommonAncestor() throws Exception {
}

@Test
@Issue("JENKINS-38089")
public void testVariousParallelCombos() throws Exception {
WorkflowJob job = r.jenkins.createProject(WorkflowJob.class, "ParallelTimingBug");
job.setDefinition(new CpsFlowDefinition(
"stage 'test' \n" +
" parallel([\n" + // ID 5 is start
" 'unit': {\n" +
" retry(1) {\n" +
" sleep 1;\n" +
" sleep 10; echo 'hello'; \n" +
" }\n" +
" },\n" +
" 'otherunit': {\n" +
" retry(1) {\n" +
" sleep 1;\n" +
" sleep 5; \n" +
" echo 'goodbye' \n" +
" }\n" +
" }\n" + // end of branch:
" ])\n"
// Seemingly gratuitous sleep steps are because original issue required specific timing to reproduce
// TODO test to see if we still need them to reproduce JENKINS-38089
"stage 'test' \n" +
" parallel 'unit': {\n" +
" retry(1) {\n" +
" sleep 1;\n" +
" sleep 10; echo 'hello'; \n" +
" }\n" +
" }, 'otherunit': {\n" +
" retry(1) {\n" +
" sleep 1;\n" +
" sleep 5; \n" +
" echo 'goodbye' \n" +
" }\n" +
" }"
));
/*Node dump follows, format:
[ID]{parent,ids}(millisSinceStartOfRun) flowNodeClassName stepDisplayName [st=startId if a block end node]
Expand Down Expand Up @@ -602,4 +605,62 @@ 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);
}

@Issue("JENKINS-38536")
@Test
public void testParallelCorrectEndNodeForVisitor() throws Exception {
// Verify that SimpleBlockVisitor actually gets the *real* last node not just the last declared branch
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");
jobPauseMiddle.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");
}
}

0 comments on commit eca21af

Please sign in to comment.