Implementation of PWF without the annoying ThreadPoolExecutor. :)
Review by: spoon (pair prog)
git-svn-id: https://google-web-toolkit.googlecode.com/svn/releases/1.6@4163 8db76d5a-ed1c-0410-87a9-c151d255dfc7
diff --git a/dev/core/src/com/google/gwt/dev/ExternalPermutationWorkerFactory.java b/dev/core/src/com/google/gwt/dev/ExternalPermutationWorkerFactory.java
index 392afea..137144d 100644
--- a/dev/core/src/com/google/gwt/dev/ExternalPermutationWorkerFactory.java
+++ b/dev/core/src/com/google/gwt/dev/ExternalPermutationWorkerFactory.java
@@ -342,6 +342,7 @@
List<PermutationWorker> toReturn = new ArrayList<PermutationWorker>(
numWorkers);
+ // TODO(spoon): clean up already-launched processes if we get an exception?
for (int i = 0; i < numWorkers; i++) {
String cookie = launchExternalWorker(logger, sock.getLocalPort());
cookies.add(cookie);
diff --git a/dev/core/src/com/google/gwt/dev/PermutationWorkerFactory.java b/dev/core/src/com/google/gwt/dev/PermutationWorkerFactory.java
index 4448c3d..0322072 100644
--- a/dev/core/src/com/google/gwt/dev/PermutationWorkerFactory.java
+++ b/dev/core/src/com/google/gwt/dev/PermutationWorkerFactory.java
@@ -17,7 +17,6 @@
import com.google.gwt.core.ext.TreeLogger;
import com.google.gwt.core.ext.UnableToCompleteException;
-import com.google.gwt.dev.jjs.InternalCompilerException;
import com.google.gwt.dev.jjs.UnifiedAst;
import com.google.gwt.dev.util.Util;
@@ -26,16 +25,9 @@
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
-import java.util.LinkedList;
import java.util.List;
-import java.util.Queue;
-import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
/**
* Represents a factory for implementations of an endpoint that will invoke
@@ -46,73 +38,167 @@
public abstract class PermutationWorkerFactory {
/**
- * This callable is responsible for compiling one permutation. It waits for an
- * available worker, uses it, and returns it to the pool when finished.
+ * Coordinates the actions of a set of {@link PermutationWorker}s, running
+ * each in its own thread.
*/
- private static class CompileOnePermutation implements Callable<ResultStatus> {
- private final BlockingQueue<PermutationWorker> availableWorkers;
- private final TreeLogger logger;
- private final Permutation permutation;
- private final File resultFile;
+ private static class Manager {
- public CompileOnePermutation(TreeLogger logger, Permutation permutation,
- File resultFile, BlockingQueue<PermutationWorker> availableWorkers) {
- this.logger = logger;
- this.permutation = permutation;
- this.resultFile = resultFile;
- this.availableWorkers = availableWorkers;
+ private static final Work POISON_PILL = new Work(null, null, null);
+
+ private static enum Result {
+ SUCCESS, FAIL, WORKER_DEATH
}
- public ResultStatus call() {
- // Find a free worker
- PermutationWorker worker;
+ /**
+ * Runs a {@link PermutationWorker} on its own thread.
+ */
+ private class WorkerThread implements Runnable {
+ private final PermutationWorker worker;
+
+ public WorkerThread(PermutationWorker worker) {
+ this.worker = worker;
+ }
+
+ public void run() {
+ Result threadDeathResult = Result.FAIL;
+ try {
+ while (true) {
+ Work work = workQueue.take();
+ if (work == POISON_PILL) {
+ return;
+ }
+ TreeLogger logger = work.getLogger();
+ try {
+ PermutationResult result = worker.compile(logger, work.getPerm());
+ Util.writeObjectAsFile(logger, work.getResultFile(), result);
+ logger.log(TreeLogger.DEBUG, "Successfully compiled permutation");
+ resultsQueue.put(Result.SUCCESS);
+ } catch (TransientWorkerException e) {
+ logger.log(TreeLogger.DEBUG,
+ "Worker died, will retry Permutation", e);
+ workQueue.add(work);
+ threadDeathResult = Result.WORKER_DEATH;
+ return;
+ } catch (UnableToCompleteException e) {
+ logger.log(TreeLogger.ERROR,
+ "Unrecoverable exception, shutting down", e);
+ threadDeathResult = Result.FAIL;
+ return;
+ }
+ }
+ } catch (InterruptedException e) {
+ return;
+ } finally {
+ // Record why I died.
+ try {
+ resultsQueue.put(threadDeathResult);
+ } catch (InterruptedException ignored) {
+ }
+ }
+ }
+ }
+
+ public static void run(TreeLogger logger, List<Work> work,
+ List<PermutationWorker> workers) throws UnableToCompleteException {
+ new Manager().doRun(logger, work, workers);
+ }
+
+ /**
+ * The queue of work to do.
+ */
+ BlockingQueue<Work> workQueue;
+
+ /**
+ * The queue of work to do.
+ */
+ BlockingQueue<Result> resultsQueue;
+
+ private Manager() {
+ }
+
+ private void doRun(TreeLogger logger, List<Work> work,
+ List<PermutationWorker> workers) throws UnableToCompleteException {
+
+ // Initialize state.
+ workQueue = new LinkedBlockingQueue<Work>(work);
+ resultsQueue = new LinkedBlockingQueue<Result>();
+
+ List<Thread> threads = new ArrayList<Thread>(workers.size());
try {
- worker = availableWorkers.take();
+ for (PermutationWorker worker : workers) {
+ Thread thread = new Thread(new WorkerThread(worker));
+ threads.add(thread);
+ thread.start();
+ }
+
+ int workToDo = work.size();
+ int aliveWorkers = workers.size();
+ waitForWorkers : while (workToDo > 0 && aliveWorkers > 0) {
+ Result take = resultsQueue.take();
+ switch (take) {
+ case SUCCESS:
+ --workToDo;
+ break;
+ case FAIL:
+ break waitForWorkers;
+ case WORKER_DEATH:
+ --aliveWorkers;
+ break;
+ default:
+ throw new IncompatibleClassChangeError(Result.class.toString());
+ }
+ }
+
+ workQueue.clear();
+ for (int i = 0; i < aliveWorkers; ++i) {
+ workQueue.add(POISON_PILL);
+ }
+
+ if (workToDo > 0) {
+ logger.log(TreeLogger.ERROR,
+ "Not all permutation were compiled , completed ("
+ + (work.size() - workToDo) + "/" + work.size() + ")");
+ throw new UnableToCompleteException();
+ }
} catch (InterruptedException e) {
- logger.log(TreeLogger.DEBUG, "Worker interrupted", e);
- return ResultStatus.HARD_FAILURE;
- }
-
- if (worker == noMoreWorkersWorker) {
- // Shutting down
- return ResultStatus.HARD_FAILURE;
- }
-
- // Invoke the worker
- try {
- PermutationResult result = worker.compile(logger, permutation);
- Util.writeObjectAsFile(logger, resultFile, result);
- logger.log(TreeLogger.DEBUG, "Successfully compiled permutation");
- availableWorkers.add(worker);
- return ResultStatus.SUCCESS;
- } catch (TransientWorkerException e) {
- logger.log(TreeLogger.DEBUG, "Worker died, will retry Permutation", e);
- return ResultStatus.TRANSIENT_FAILURE;
- } catch (UnableToCompleteException e) {
- logger.log(TreeLogger.ERROR, "Unrecoverable exception, shutting down",
- e);
- return ResultStatus.HARD_FAILURE;
+ logger.log(TreeLogger.ERROR,
+ "Exiting without results due to interruption", e);
+ throw new UnableToCompleteException();
+ } finally {
+ // Interrupt any outstanding threads.
+ for (Thread thread : threads) {
+ thread.interrupt();
+ }
}
}
}
- private static enum ResultStatus {
- /**
- * A failure bad enough to merit shutting down the compilation.
- */
- HARD_FAILURE,
+ /**
+ * Represents work to do.
+ */
+ private static class Work {
+ private final TreeLogger logger;
+ private final Permutation perm;
+ private final File resultFile;
- /**
- * A successful compile.
- */
- SUCCESS,
+ public Work(TreeLogger logger, Permutation perm, File resultFile) {
+ this.logger = logger;
+ this.perm = perm;
+ this.resultFile = resultFile;
+ }
- /**
- * A worker died while processing this permutation, but it is worth trying
- * to compile it with another worker.
- */
- TRANSIENT_FAILURE
- };
+ public TreeLogger getLogger() {
+ return logger;
+ }
+
+ public Permutation getPerm() {
+ return perm;
+ }
+
+ public File getResultFile() {
+ return resultFile;
+ }
+ }
/**
* The name of the system property used to define the workers.
@@ -128,21 +214,6 @@
private static List<PermutationWorkerFactory> lazyFactories;
- private static final PermutationWorker noMoreWorkersWorker = new PermutationWorker() {
-
- public PermutationResult compile(TreeLogger logger, Permutation permutation)
- throws TransientWorkerException, UnableToCompleteException {
- throw new UnableToCompleteException();
- }
-
- public String getName() {
- return "Marker worker indicating no more workers";
- }
-
- public void shutdown() {
- }
- };
-
/**
* Compiles all Permutations in a Precompilation and returns an array of Files
* that can be consumed by Link using the system-default
@@ -175,103 +246,35 @@
assert Arrays.asList(precompilation.getPermutations()).containsAll(
Arrays.asList(permutations));
- // We may have a mixed collection of workers from different factories
+ // Create the work.
+ List<Work> work = new ArrayList<Work>(permutations.length);
+ for (int i = 0; i < permutations.length; ++i) {
+ Permutation perm = permutations[i];
+ TreeLogger permLogger = logger.branch(TreeLogger.DEBUG,
+ "Worker permutation " + perm.getId() + " of " + permutations.length);
+ work.add(new Work(permLogger, perm, resultFiles[i]));
+ }
+
+ // Create the workers.
List<PermutationWorker> workers = new ArrayList<PermutationWorker>();
-
- /*
- * We can have errors below this point, there's a finally block to handle
- * cleanup of workers.
- */
try {
- createWorkers(logger, precompilation.getUnifiedAst(),
- permutations.length, localWorkers, workers);
- ExecutorService executor = Executors.newFixedThreadPool(workers.size());
+ createWorkers(logger, precompilation.getUnifiedAst(), work.size(),
+ localWorkers, workers);
- // List of available workers.
- // The extra space is for inserting nulls at shutdown time
- BlockingQueue<PermutationWorker> availableWorkers = new ArrayBlockingQueue<PermutationWorker>(
- 2 * workers.size());
- availableWorkers.addAll(workers);
-
- try {
-
- // Submit all tasks to the executor
-
- // The permutation compiles not yet finished
- Queue<CompileOnePermutation> tasksOutstanding = new LinkedList<CompileOnePermutation>();
-
- // The futures for the results of those compiles
- Queue<Future<ResultStatus>> resultFutures = new LinkedList<Future<ResultStatus>>();
-
- for (int i = 0; i < permutations.length; ++i) {
- TreeLogger permLogger = logger.branch(TreeLogger.DEBUG,
- "Worker permutation " + permutations[i].getId() + " of "
- + permutations.length);
- CompileOnePermutation task = new CompileOnePermutation(permLogger,
- permutations[i], resultFiles[i], availableWorkers);
- tasksOutstanding.add(task);
- resultFutures.add(executor.submit(task));
- }
-
- // Count the number of dead workers
- int numDeadWorkers = 0;
- int successCount = 0;
-
- while (!resultFutures.isEmpty() && numDeadWorkers < workers.size()) {
- assert resultFutures.size() == tasksOutstanding.size();
-
- CompileOnePermutation task = tasksOutstanding.remove();
- Future<ResultStatus> future = resultFutures.remove();
- ResultStatus result;
- try {
- result = future.get();
- } catch (InterruptedException e) {
- logger.log(TreeLogger.ERROR,
- "Exiting without results due to interruption", e);
- throw new UnableToCompleteException();
- } catch (ExecutionException e) {
- logger.log(TreeLogger.ERROR, "A compilation failed", e);
- throw new UnableToCompleteException();
- }
-
- if (result == ResultStatus.SUCCESS) {
- ++successCount;
- } else if (result == ResultStatus.TRANSIENT_FAILURE) {
- // A worker died. Resubmit for the remaining workers.
- ++numDeadWorkers;
- tasksOutstanding.add(task);
- resultFutures.add(executor.submit(task));
- } else if (result == ResultStatus.HARD_FAILURE) {
- // Shut down.
- break;
- } else {
- throw new InternalCompilerException("Unknown result type");
- }
- }
-
- // Too many permutations is a coding error
- assert successCount <= permutations.length;
-
- if (successCount < permutations.length) {
- // Likely as not, all of the workers died
- logger.log(TreeLogger.ERROR, "Not all permutation were compiled "
- + successCount + " of " + permutations.length);
- throw new UnableToCompleteException();
- }
- } finally {
- // Shut down the executor
- executor.shutdown();
-
- // Inform any residual CompileOnePermutation's that there aren't any
- // more workers
- for (int i = 0; i < workers.size(); i++) {
- availableWorkers.add(noMoreWorkersWorker);
+ // Get it done!
+ Manager.run(logger, work, workers);
+ } finally {
+ Throwable caught = null;
+ for (PermutationWorker worker : workers) {
+ try {
+ worker.shutdown();
+ } catch (Throwable e) {
+ caught = e;
}
}
- } finally {
- // Shut down all workers
- for (PermutationWorker worker : workers) {
- worker.shutdown();
+ if (caught != null) {
+ throw new RuntimeException(
+ "One of the workers threw an exception while shutting down", caught);
}
}
}
