mirror of
https://github.com/sbt/sbt.git
synced 2026-09-01 02:27:13 +02:00
Third draft to execute the forked tests in parallel.
This feature is not activated by default. To enable it set `testForkedParallel` to `true`. The test-agent then executes the tests in a thread pool. For now it has a fixed size set to the number of available processors. The concurrent restrictions configuration should be used.
This commit is contained in:
Bruno Bieth
Mark Harrah
@@ -0,0 +1,21 @@
|
||||
package sbt;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
public final class ForkConfiguration implements Serializable {
|
||||
private boolean ansiCodesSupported;
|
||||
private boolean parallel;
|
||||
|
||||
public ForkConfiguration(boolean ansiCodesSupported, boolean parallel) {
|
||||
this.ansiCodesSupported = ansiCodesSupported;
|
||||
this.parallel = parallel;
|
||||
}
|
||||
|
||||
public boolean isAnsiCodesSupported() {
|
||||
return ansiCodesSupported;
|
||||
}
|
||||
|
||||
public boolean isParallel() {
|
||||
return parallel;
|
||||
}
|
||||
}
|
||||
@@ -12,10 +12,15 @@ import java.io.Serializable;
|
||||
import java.net.Socket;
|
||||
import java.net.InetAddress;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.*;
|
||||
|
||||
public class ForkMain {
|
||||
|
||||
// serializables
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
static class SubclassFingerscan implements SubclassFingerprint, Serializable {
|
||||
private boolean isModule;
|
||||
private String superclassName;
|
||||
@@ -29,6 +34,7 @@ public class ForkMain {
|
||||
public String superclassName() { return superclassName; }
|
||||
public boolean requireNoArgConstructor() { return requireNoArgConstructor; }
|
||||
}
|
||||
|
||||
static class AnnotatedFingerscan implements AnnotatedFingerprint, Serializable {
|
||||
private boolean isModule;
|
||||
private String annotationName;
|
||||
@@ -39,6 +45,54 @@ public class ForkMain {
|
||||
public boolean isModule() { return isModule; }
|
||||
public String annotationName() { return annotationName; }
|
||||
}
|
||||
|
||||
static class ForkEvent implements Event, Serializable {
|
||||
private String fullyQualifiedName;
|
||||
private Fingerprint fingerprint;
|
||||
private Selector selector;
|
||||
private Status status;
|
||||
private OptionalThrowable throwable;
|
||||
private long duration;
|
||||
|
||||
ForkEvent(Event e) {
|
||||
fullyQualifiedName = e.fullyQualifiedName();
|
||||
Fingerprint rawFingerprint = e.fingerprint();
|
||||
|
||||
if (rawFingerprint instanceof SubclassFingerprint)
|
||||
this.fingerprint = new SubclassFingerscan((SubclassFingerprint) rawFingerprint);
|
||||
else
|
||||
this.fingerprint = new AnnotatedFingerscan((AnnotatedFingerprint) rawFingerprint);
|
||||
|
||||
selector = e.selector();
|
||||
checkSerializableSelector(selector);
|
||||
status = e.status();
|
||||
OptionalThrowable originalThrowable = e.throwable();
|
||||
|
||||
if (originalThrowable.isDefined())
|
||||
this.throwable = new OptionalThrowable(new ForkError(originalThrowable.get()));
|
||||
else
|
||||
this.throwable = originalThrowable;
|
||||
|
||||
this.duration = e.duration();
|
||||
}
|
||||
|
||||
public String fullyQualifiedName() { return fullyQualifiedName; }
|
||||
public Fingerprint fingerprint() { return fingerprint; }
|
||||
public Selector selector() { return selector; }
|
||||
public Status status() { return status; }
|
||||
public OptionalThrowable throwable() { return throwable; }
|
||||
public long duration() { return duration; }
|
||||
|
||||
static void checkSerializableSelector(Selector selector) {
|
||||
if (! (selector instanceof Serializable)) {
|
||||
throw new UnsupportedOperationException("Selector implementation must be Serializable, but " + selector.getClass().getName() + " is not.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
static class ForkError extends Exception {
|
||||
private String originalMessage;
|
||||
private ForkError cause;
|
||||
@@ -50,62 +104,50 @@ public class ForkMain {
|
||||
public String getMessage() { return originalMessage; }
|
||||
public Exception getCause() { return cause; }
|
||||
}
|
||||
|
||||
static Selector forkSelector(Selector selector) {
|
||||
if (selector instanceof Serializable)
|
||||
return selector;
|
||||
else
|
||||
throw new UnsupportedOperationException("Selector implementation must be Serializable, but " + selector.getClass().getName() + " is not.");
|
||||
}
|
||||
|
||||
static class ForkEvent implements Event, Serializable {
|
||||
private String fullyQualifiedName;
|
||||
private Fingerprint fingerprint;
|
||||
private Selector selector;
|
||||
private Status status;
|
||||
private OptionalThrowable throwable;
|
||||
private long duration;
|
||||
ForkEvent(Event e) {
|
||||
fullyQualifiedName = e.fullyQualifiedName();
|
||||
Fingerprint rawFingerprint = e.fingerprint();
|
||||
if (rawFingerprint instanceof SubclassFingerprint)
|
||||
this.fingerprint = new SubclassFingerscan((SubclassFingerprint) rawFingerprint);
|
||||
else
|
||||
this.fingerprint = new AnnotatedFingerscan((AnnotatedFingerprint) rawFingerprint);
|
||||
selector = forkSelector(e.selector());
|
||||
status = e.status();
|
||||
OptionalThrowable originalThrowable = e.throwable();
|
||||
if (originalThrowable.isDefined())
|
||||
this.throwable = new OptionalThrowable(new ForkError(originalThrowable.get()));
|
||||
else
|
||||
this.throwable = originalThrowable;
|
||||
this.duration = e.duration();
|
||||
}
|
||||
public String fullyQualifiedName() { return fullyQualifiedName; }
|
||||
public Fingerprint fingerprint() { return fingerprint; }
|
||||
public Selector selector() { return selector; }
|
||||
public Status status() { return status; }
|
||||
public OptionalThrowable throwable() { return throwable; }
|
||||
public long duration() { return duration; }
|
||||
}
|
||||
|
||||
|
||||
// main
|
||||
// ----------------------------------------------------------------------------------------------------------------
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
Socket socket = new Socket(InetAddress.getByName(null), Integer.valueOf(args[0]));
|
||||
final ObjectInputStream is = new ObjectInputStream(socket.getInputStream());
|
||||
final ObjectOutputStream os = new ObjectOutputStream(socket.getOutputStream());
|
||||
// Must flush the header that the constructor writes, otherwise the ObjectInputStream on the other end may block indefinitely
|
||||
os.flush();
|
||||
|
||||
try {
|
||||
new Run().run(is, os);
|
||||
} finally {
|
||||
try {
|
||||
new Run().run(is, os);
|
||||
} finally {
|
||||
is.close();
|
||||
os.close();
|
||||
} finally {
|
||||
System.exit(0);
|
||||
}
|
||||
} finally {
|
||||
System.exit(0);
|
||||
}
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
private static class Run {
|
||||
|
||||
void run(ObjectInputStream is, ObjectOutputStream os) throws Exception {
|
||||
try {
|
||||
runTests(is, os);
|
||||
} catch (RunAborted e) {
|
||||
internalError(e);
|
||||
} catch (Throwable t) {
|
||||
try {
|
||||
logError(os, "Uncaught exception when running tests: " + t.toString());
|
||||
write(os, new ForkError(t));
|
||||
} catch (Throwable t2) {
|
||||
internalError(t2);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
boolean matches(Fingerprint f1, Fingerprint f2) {
|
||||
if (f1 instanceof SubclassFingerprint && f2 instanceof SubclassFingerprint) {
|
||||
final SubclassFingerprint sf1 = (SubclassFingerprint) f1;
|
||||
@@ -118,9 +160,11 @@ public class ForkMain {
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
class RunAborted extends RuntimeException {
|
||||
RunAborted(Exception e) { super(e); }
|
||||
}
|
||||
|
||||
synchronized void write(ObjectOutputStream os, Object obj) {
|
||||
try {
|
||||
os.writeObject(obj);
|
||||
@@ -129,29 +173,50 @@ public class ForkMain {
|
||||
throw new RunAborted(e);
|
||||
}
|
||||
}
|
||||
void logError(ObjectOutputStream os, String message) {
|
||||
write(os, new Object[]{ForkTags.Error, message});
|
||||
|
||||
void log(ObjectOutputStream os, String message, ForkTags level) {
|
||||
write(os, new Object[]{level, message});
|
||||
}
|
||||
void logDebug(ObjectOutputStream os, String message) {
|
||||
write(os, new Object[]{ForkTags.Debug, message});
|
||||
|
||||
void logDebug(ObjectOutputStream os, String message) { log(os, message, ForkTags.Debug); }
|
||||
void logInfo(ObjectOutputStream os, String message) { log(os, message, ForkTags.Info); }
|
||||
void logWarn(ObjectOutputStream os, String message) { log(os, message, ForkTags.Warn); }
|
||||
void logError(ObjectOutputStream os, String message) { log(os, message, ForkTags.Error); }
|
||||
|
||||
Logger remoteLogger(final boolean ansiCodesSupported, final ObjectOutputStream os) {
|
||||
return new Logger() {
|
||||
public boolean ansiCodesSupported() { return ansiCodesSupported; }
|
||||
public void error(String s) { logError(os, s); }
|
||||
public void warn(String s) { logWarn(os, s); }
|
||||
public void info(String s) { logInfo(os, s); }
|
||||
public void debug(String s) { logDebug(os, s); }
|
||||
public void trace(Throwable t) { write(os, new ForkError(t)); }
|
||||
};
|
||||
}
|
||||
|
||||
void writeEvents(ObjectOutputStream os, TaskDef taskDef, ForkEvent[] events) {
|
||||
write(os, new Object[]{taskDef.fullyQualifiedName(), events});
|
||||
}
|
||||
|
||||
ExecutorService executorService(ForkConfiguration config, ObjectOutputStream os) {
|
||||
if(config.isParallel()) {
|
||||
int nbThreads = Runtime.getRuntime().availableProcessors();
|
||||
logDebug(os, "Create a test executor with a thread pool of " + nbThreads + " threads.");
|
||||
// more options later...
|
||||
// TODO we might want to configure the blocking queue with size #proc
|
||||
return Executors.newFixedThreadPool(nbThreads);
|
||||
} else {
|
||||
logDebug(os, "Create a single-thread test executor");
|
||||
return Executors.newSingleThreadExecutor();
|
||||
}
|
||||
}
|
||||
|
||||
void runTests(ObjectInputStream is, final ObjectOutputStream os) throws Exception {
|
||||
final boolean ansiCodesSupported = is.readBoolean();
|
||||
final ForkConfiguration config = (ForkConfiguration) is.readObject();
|
||||
ExecutorService executor = executorService(config, os);
|
||||
final TaskDef[] tests = (TaskDef[]) is.readObject();
|
||||
int nFrameworks = is.readInt();
|
||||
Logger[] loggers = {
|
||||
new Logger() {
|
||||
public boolean ansiCodesSupported() { return ansiCodesSupported; }
|
||||
public void error(String s) { logError(os, s); }
|
||||
public void warn(String s) { write(os, new Object[]{ForkTags.Warn, s}); }
|
||||
public void info(String s) { write(os, new Object[]{ForkTags.Info, s}); }
|
||||
public void debug(String s) { write(os, new Object[]{ForkTags.Debug, s}); }
|
||||
public void trace(Throwable t) { write(os, new ForkError(t)); }
|
||||
}
|
||||
};
|
||||
Logger[] loggers = { remoteLogger(config.isAnsiCodesSupported(), os) };
|
||||
|
||||
for (int i = 0; i < nFrameworks; i++) {
|
||||
final String[] implClassNames = (String[]) is.readObject();
|
||||
@@ -186,89 +251,66 @@ public class ForkMain {
|
||||
final Runner runner = framework.runner(frameworkArgs, remoteFrameworkArgs, getClass().getClassLoader());
|
||||
Task[] tasks = runner.tasks(filteredTests.toArray(new TaskDef[filteredTests.size()]));
|
||||
logDebug(os, "Runner for " + framework.getClass().getName() + " produced " + tasks.length + " initial tasks for " + filteredTests.size() + " tests.");
|
||||
for (Task task : tasks)
|
||||
runTestSafe(task, runner, loggers, os);
|
||||
|
||||
runTestTasks(executor, tasks, loggers, os);
|
||||
|
||||
runner.done();
|
||||
}
|
||||
write(os, ForkTags.Done);
|
||||
is.readObject();
|
||||
}
|
||||
class NestedTask {
|
||||
private String parentName;
|
||||
private Task task;
|
||||
NestedTask(String parentName, Task task) {
|
||||
this.parentName = parentName;
|
||||
this.task = task;
|
||||
}
|
||||
public String getParentName() {
|
||||
return parentName;
|
||||
}
|
||||
public Task getTask() {
|
||||
return task;
|
||||
}
|
||||
}
|
||||
void runTestSafe(Task task, Runner runner, Logger[] loggers, ObjectOutputStream os) {
|
||||
TaskDef taskDef = task.taskDef();
|
||||
try {
|
||||
List<NestedTask> nestedTasks = new ArrayList<NestedTask>();
|
||||
for (Task nt : runTest(taskDef, task, loggers, os))
|
||||
nestedTasks.add(new NestedTask(taskDef.fullyQualifiedName(), nt));
|
||||
while (true) {
|
||||
List<NestedTask> newNestedTasks = new ArrayList<NestedTask>();
|
||||
int nestedTasksLength = nestedTasks.size();
|
||||
for (int i = 0; i < nestedTasksLength; i++) {
|
||||
NestedTask nestedTask = nestedTasks.get(i);
|
||||
String nestedParentName = nestedTask.getParentName() + "-" + i;
|
||||
for (Task nt : runTest(nestedTask.getTask().taskDef(), nestedTask.getTask(), loggers, os)) {
|
||||
newNestedTasks.add(new NestedTask(nestedParentName, nt));
|
||||
}
|
||||
}
|
||||
if (newNestedTasks.size() == 0)
|
||||
break;
|
||||
else {
|
||||
nestedTasks = newNestedTasks;
|
||||
|
||||
void runTestTasks(ExecutorService executor, Task[] tasks, Logger[] loggers, ObjectOutputStream os) {
|
||||
if( tasks.length > 0 ) {
|
||||
List<Future<Task[]>> futureNestedTasks = new ArrayList<Future<Task[]>>();
|
||||
for( Task task : tasks ) {
|
||||
futureNestedTasks.add(runTest(executor, task, loggers, os));
|
||||
}
|
||||
|
||||
// Note: this could be optimized further, we could have a callback once a test finishes that executes immediately the nested tasks
|
||||
// At the moment, I'm especially interested in JUnit, which doesn't have nested tasks.
|
||||
List<Task> nestedTasks = new ArrayList<Task>();
|
||||
for( Future<Task[]> futureNestedTask : futureNestedTasks ) {
|
||||
try {
|
||||
nestedTasks.addAll( Arrays.asList(futureNestedTask.get()));
|
||||
} catch (Exception e) {
|
||||
logError(os, "Failed to execute task " + futureNestedTask);
|
||||
}
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
writeEvents(os, taskDef, new ForkEvent[] { testError(os, taskDef, "Uncaught exception when running " + taskDef.fullyQualifiedName() + ": " + t.toString(), t) });
|
||||
runTestTasks(executor, nestedTasks.toArray(new Task[nestedTasks.size()]), loggers, os);
|
||||
}
|
||||
}
|
||||
Task[] runTest(TaskDef taskDef, Task task, Logger[] loggers, ObjectOutputStream os) {
|
||||
ForkEvent[] events;
|
||||
Task[] nestedTasks;
|
||||
try {
|
||||
final List<ForkEvent> eventList = new ArrayList<ForkEvent>();
|
||||
EventHandler handler = new EventHandler() { public void handle(Event e){ eventList.add(new ForkEvent(e)); } };
|
||||
logDebug(os, " Running " + taskDef);
|
||||
nestedTasks = task.execute(handler, loggers);
|
||||
if(nestedTasks.length > 0 || eventList.size() > 0)
|
||||
logDebug(os, " Produced " + nestedTasks.length + " nested tasks and " + eventList.size() + " events.");
|
||||
events = eventList.toArray(new ForkEvent[eventList.size()]);
|
||||
}
|
||||
catch (Throwable t) {
|
||||
nestedTasks = new Task[0];
|
||||
events = new ForkEvent[] { testError(os, taskDef, "Uncaught exception when running " + taskDef.fullyQualifiedName() + ": " + t.toString(), t) };
|
||||
}
|
||||
writeEvents(os, taskDef, events);
|
||||
return nestedTasks;
|
||||
}
|
||||
void run(ObjectInputStream is, ObjectOutputStream os) throws Exception {
|
||||
try {
|
||||
runTests(is, os);
|
||||
} catch (RunAborted e) {
|
||||
internalError(e);
|
||||
} catch (Throwable t) {
|
||||
try {
|
||||
logError(os, "Uncaught exception when running tests: " + t.toString());
|
||||
write(os, new ForkError(t));
|
||||
} catch (Throwable t2) {
|
||||
internalError(t2);
|
||||
|
||||
Future<Task[]> runTest(ExecutorService executor, final Task task, final Logger[] loggers, final ObjectOutputStream os) {
|
||||
return executor.submit(new Callable<Task[]>() {
|
||||
@Override
|
||||
public Task[] call() {
|
||||
ForkEvent[] events;
|
||||
Task[] nestedTasks;
|
||||
TaskDef taskDef = task.taskDef();
|
||||
try {
|
||||
final List<ForkEvent> eventList = new ArrayList<ForkEvent>();
|
||||
EventHandler handler = new EventHandler() { public void handle(Event e){ eventList.add(new ForkEvent(e)); } };
|
||||
logDebug(os, " Running " + taskDef);
|
||||
nestedTasks = task.execute(handler, loggers);
|
||||
if(nestedTasks.length > 0 || eventList.size() > 0)
|
||||
logDebug(os, " Produced " + nestedTasks.length + " nested tasks and " + eventList.size() + " events.");
|
||||
events = eventList.toArray(new ForkEvent[eventList.size()]);
|
||||
}
|
||||
catch (Throwable t) {
|
||||
nestedTasks = new Task[0];
|
||||
events = new ForkEvent[] { testError(os, taskDef, "Uncaught exception when running " + taskDef.fullyQualifiedName() + ": " + t.toString(), t) };
|
||||
}
|
||||
writeEvents(os, taskDef, events);
|
||||
return nestedTasks;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void internalError(Throwable t) {
|
||||
System.err.println("Internal error when running tests: " + t.toString());
|
||||
}
|
||||
|
||||
ForkEvent testEvent(final String fullyQualifiedName, final Fingerprint fingerprint, final Selector selector, final Status r, final ForkError err, final long duration) {
|
||||
final OptionalThrowable throwable;
|
||||
if (err == null)
|
||||
@@ -280,21 +322,18 @@ public class ForkMain {
|
||||
public Fingerprint fingerprint() { return fingerprint; }
|
||||
public Selector selector() { return selector; }
|
||||
public Status status() { return r; }
|
||||
public OptionalThrowable throwable() {
|
||||
return throwable;
|
||||
public OptionalThrowable throwable() {
|
||||
return throwable;
|
||||
}
|
||||
public long duration() {
|
||||
return duration;
|
||||
}
|
||||
});
|
||||
}
|
||||
ForkEvent testError(ObjectOutputStream os, TaskDef taskDef, String message) {
|
||||
logError(os, message);
|
||||
return testEvent(taskDef.fullyQualifiedName(), taskDef.fingerprint(), new SuiteSelector(), Status.Error, null, 0);
|
||||
}
|
||||
|
||||
ForkEvent testError(ObjectOutputStream os, TaskDef taskDef, String message, Throwable t) {
|
||||
logError(os, message);
|
||||
ForkError fe = new ForkError(t);
|
||||
ForkError fe = new ForkError(t);
|
||||
write(os, fe);
|
||||
return testEvent(taskDef.fullyQualifiedName(), taskDef.fingerprint(), new SuiteSelector(), Status.Error, fe, 0);
|
||||
}
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
package sbt;
|
||||
|
||||
import sbt.testing.*;
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* Adapts the old {@link org.scalatools.testing.Framework} interface into the new
|
||||
* {@link sbt.testing.Framework}
|
||||
*/
|
||||
final class FrameworkWrapper implements Framework {
|
||||
|
||||
private org.scalatools.testing.Framework oldFramework;
|
||||
|
||||
Reference in New Issue
Block a user