-
Notifications
You must be signed in to change notification settings - Fork 105
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
implement optional concurrency limit for parallel blocks #57
base: master
Are you sure you want to change the base?
Changes from 6 commits
83a5b7c
44a9940
97e38ca
d970754
d5a2e5c
3e69308
a33c97f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,8 +1,9 @@ | ||
/* | ||
* The MIT License | ||
* | ||
* Copyright (c) 2013, CloudBees, Inc., Nicolas De Loof. | ||
* Cisco Systems, Inc., a California corporation | ||
* Copyright (c) 2013-2015, CloudBees, Inc., Nicolas De Loof. | ||
* Cisco Systems, Inc., a California corporation | ||
* SAP SE | ||
* | ||
* Permission is hereby granted, free of charge, to any person obtaining a copy | ||
* of this software and associated documentation files (the "Software"), to deal | ||
|
@@ -41,6 +42,7 @@ import org.codehaus.groovy.control.customizers.ImportCustomizer | |
import java.util.concurrent.* | ||
import java.util.logging.Logger | ||
|
||
import static hudson.model.Result.ABORTED | ||
import static hudson.model.Result.FAILURE | ||
import static hudson.model.Result.SUCCESS | ||
|
||
|
@@ -103,6 +105,9 @@ public class FlowDSL { | |
flowRun.state.result = Executor.currentExecutor().abortResult(); | ||
Executor.currentExecutor().recordCauseOfInterruption(flowRun, listener); | ||
|
||
flowRun.isAborting = true | ||
flowRun.abortUnstartedFutures() | ||
|
||
def graph = flowRun.jobsGraph | ||
graph.vertexSet().each() { ji -> | ||
if (flowRun.project != ji.project) { | ||
|
@@ -395,41 +400,45 @@ public class FlowDelegate { | |
} | ||
|
||
// allows syntax like : parallel(["Kohsuke","Nicolas"].collect { name -> return { build("job1", param1:name) } }) | ||
def List<FlowState> parallel(Collection<? extends Closure> closures) { | ||
parallel(closures as Closure[]) | ||
def List<FlowState> parallel(int maxThreads = 0, Collection<? extends Closure> closures) { | ||
parallel(maxThreads, closures as Closure[]) | ||
} | ||
|
||
// allows collecting job status by name rather than by index | ||
// inspired by https://github.com/caolan/async#parallel | ||
def Map<?, FlowState> parallel(Map<?, ? extends Closure> args) { | ||
def Map<?, FlowState> parallel(int maxThreads = 0, Map<?, ? extends Closure> args) { | ||
def keys = new ArrayList<?>() | ||
def closures = new ArrayList<? extends Closure>() | ||
args.entrySet().each { e -> | ||
keys.add(e.key) | ||
closures.add(e.value) | ||
} | ||
def results = new LinkedHashMap<?, FlowState>() | ||
def flowStates = parallel(closures) // as List<FlowState> | ||
def flowStates = parallel(maxThreads, closures) // as List<FlowState> | ||
flowStates.eachWithIndex { v, i -> results[keys[i]] = v } | ||
results | ||
} | ||
|
||
def List<FlowState> parallel(Closure ... closures) { | ||
def List<FlowState> parallel(int maxThreads = 0, Closure ... closures) { | ||
statusCheck() | ||
// TODO use NamingThreadFactory since Jenkins 1.541 | ||
ExecutorService pool = Executors.newCachedThreadPool(new ThreadFactory() { | ||
ThreadFactory tf = new ThreadFactory() { | ||
public Thread newThread(Runnable r) { | ||
def thread = Executors.defaultThreadFactory().newThread(r); | ||
thread.name = "BuildFlow parallel statement thread for " + flowRun.parent.fullName; | ||
return thread; | ||
} | ||
}); | ||
} | ||
ExecutorService pool = (maxThreads <= 0) ? | ||
Executors.newCachedThreadPool(tf) : Executors.newFixedThreadPool(maxThreads, tf) | ||
Set<Run> upstream = flowRun.state.lastCompleted | ||
Set<Run> lastCompleted = Collections.synchronizedSet(new HashSet<Run>()) | ||
def results = new CopyOnWriteArrayList<FlowState>() | ||
def tasks = new ArrayList<Future<FlowState>>() | ||
|
||
println("parallel {") | ||
def startMsg = "parallel" | ||
if ( maxThreads > 0 ) startMsg += "( "+maxThreads+" )" | ||
println(startMsg + " {") | ||
++indent | ||
|
||
def current_state = flowRun.state | ||
|
@@ -450,26 +459,34 @@ public class FlowDelegate { | |
|
||
tasks.add(pool.submit(track_closure as Callable)) | ||
} | ||
// add the full list of futures to a build wide list | ||
// to account for possible nested parallel blocks | ||
flowRun.addNewFutures(tasks) | ||
|
||
tasks.each {task -> | ||
try { | ||
def final_state = task.get() | ||
Result result = final_state.result | ||
results.add(final_state) | ||
current_state.result = current_state.result.combine(result) | ||
} catch(ExecutionException e) | ||
{ | ||
} | ||
catch(ExecutionException e) { | ||
// TODO perhaps rethrow? | ||
current_state.result = FAILURE | ||
listener.error("Failed to run DSL Script") | ||
e.printStackTrace(listener.getLogger()) | ||
} | ||
catch(CancellationException e) { | ||
current_state.result = ABORTED | ||
} | ||
} | ||
|
||
pool.shutdown() | ||
pool.awaitTermination(1, TimeUnit.DAYS) | ||
current_state.lastCompleted =lastCompleted | ||
} finally { | ||
pool.shutdown() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. please add a comment to explain why we are doing shutdown here and not waiting for task termination. |
||
tasks.each {task -> task.cancel(false)} | ||
flowRun.state = current_state | ||
--indent | ||
println("}") | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,7 +1,8 @@ | ||
/* | ||
* The MIT License | ||
* | ||
* Copyright (c) 2013, CloudBees, Inc., Nicolas De Loof. | ||
* Copyright (c) 2013-2015, CloudBees, Inc., Nicolas De Loof. | ||
* SAP SE | ||
* | ||
* Permission is hereby granted, free of charge, to any person obtaining a copy | ||
* of this software and associated documentation files (the "Software"), to deal | ||
|
@@ -35,11 +36,14 @@ | |
|
||
import java.io.File; | ||
import java.io.IOException; | ||
import java.util.ArrayList; | ||
import java.util.Arrays; | ||
import java.util.Collections; | ||
import java.util.Comparator; | ||
import java.util.LinkedList; | ||
import java.util.List; | ||
import java.util.concurrent.ExecutionException; | ||
import java.util.concurrent.Future; | ||
import java.util.concurrent.atomic.AtomicInteger; | ||
import java.util.logging.Logger; | ||
|
||
|
@@ -67,10 +71,15 @@ public class FlowRun extends Build<BuildFlow, FlowRun> { | |
|
||
private DirectedGraph<JobInvocation, JobEdge> jobsGraph; | ||
|
||
// Note: synchronization necessary | ||
private transient final List<Future<FlowState>> futuresList = new ArrayList<Future<FlowState>>(); | ||
|
||
private transient ThreadLocal<FlowState> state = new ThreadLocal<FlowState>(); | ||
|
||
private transient AtomicInteger buildIndex = new AtomicInteger(1); | ||
|
||
private transient volatile boolean isAborting = false; | ||
|
||
public FlowRun(BuildFlow job, File buildDir) throws IOException { | ||
super(job, buildDir); | ||
setup(job); | ||
|
@@ -156,6 +165,28 @@ public void run() { | |
} | ||
} | ||
|
||
protected void addNewFutures( List<Future<FlowState>> newFutures ) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. please add doc |
||
synchronized(futuresList) { | ||
if ( !isAborting ) { | ||
futuresList.addAll(newFutures); | ||
} | ||
else { | ||
for ( Future f : newFutures ) { | ||
f.cancel(false); | ||
} | ||
} | ||
} | ||
} | ||
|
||
protected void abortUnstartedFutures() { | ||
synchronized(futuresList) { | ||
isAborting = true; | ||
for ( Future f : futuresList ) { | ||
f.cancel(false); | ||
} | ||
} | ||
} | ||
|
||
protected class BuildWithWorkspaceRunnerImpl extends AbstractRunner { | ||
|
||
private final String dsl; | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -137,7 +137,7 @@ class AbortTest extends DSLTestCase { | |
|
||
def flow = future.waitForStart() | ||
// wait for job1 to start | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. please fix comment |
||
while (!job1.building) { | ||
while (!job1.building || !job2.building) { | ||
Thread.sleep(10L) | ||
} | ||
println("job has started") | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
please feel free to add a but I'm hesitant about changing existing notices.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
... note we are in 2016 too.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Because it is basically impossible to separate the copyright ownership of the various pieces, I feel that a single date range and a list of the contributes makes more sense. The original work was done in 2015, but I will update due to review changes.