LocalComputationManager implements ComputationManager { private ExecutionReport execute(Path workingDir, List<CommandExecution> commandExecutionList, Map<String, String> variables, ComputationParameters computationParameters, ExecutionMonitor monitor) throws InterruptedException { List<ExecutionError> errors = new ArrayList<>(); ExecutorService executionSubmitter = Executors.newCachedThreadPool(); for (CommandExecution commandExecution : commandExecutionList) { Command command = commandExecution.getCommand(); CountDownLatch latch = new CountDownLatch(commandExecution.getExecutionCount()); IntStream.range(0, commandExecution.getExecutionCount()).forEach(idx -> executionSubmitter.execute(() -> { try { enter(); logExecutingCommand(workingDir, command, idx); preProcess(workingDir, command, idx); int exitValue = process(workingDir, commandExecution, idx, variables, computationParameters); postProcess(workingDir, commandExecution, idx, exitValue, errors, monitor); } catch (Exception e) { LOGGER.warn(e.getMessage()); } finally { latch.countDown(); exit(); } }) ); latch.await(); } executionSubmitter.shutdown(); if (!executionSubmitter.awaitTermination(20, TimeUnit.SECONDS)) { executionSubmitter.shutdownNow(); if (!executionSubmitter.awaitTermination(20, TimeUnit.SECONDS)) { LOGGER.error("Thread pool did not terminate"); } } return new DefaultExecutionReport(errors); } LocalComputationManager();  LocalComputationManager(Executor executor);  LocalComputationManager(PlatformConfig platformConfig);  LocalComputationManager(Path localDir);  LocalComputationManager(LocalComputationConfig config);  LocalComputationManager(LocalComputationConfig config, Executor executor);  LocalComputationManager(LocalComputationConfig config, LocalCommandExecutor localCommandExecutor, Executor executor); static ComputationManager getDefault(); @Override String getVersion(); @Override Path getLocalDir(); @Override OutputStream newCommonFile(String fileName); @Override CompletableFuture<R> execute(ExecutionEnvironment environment, ExecutionHandler<R> handler); @Override CompletableFuture<R> execute(ExecutionEnvironment environment, ExecutionHandler<R> handler, ComputationParameters parameters); @Override ComputationResourcesStatus getResourcesStatus(); @Override Executor getExecutor(); @Override void close();  }
@Test public void test1() throws Exception { LocalCommandExecutor localCommandExecutor = new AbstractLocalCommandExecutor() { @Override void nonZeroLog(List<String> cmdLs, int exitCode) { } @Override public int execute(String program, List<String> args, Path outFile, Path errFile, Path workingDir, Map<String, String> env) throws IOException, InterruptedException { assertEquals("prog1", program); assertEquals(ImmutableList.of("file1", "file2", "file3"), args); assertEquals(ImmutableMap.of("var1", "val1"), env); assertTrue(Files.exists(workingDir)); assertEquals(workingDir.resolve("prog1_cmd_0.out").toString(), outFile.toString()); assertEquals(workingDir.resolve("prog1_cmd_0.err").toString(), errFile.toString()); assertTrue(Files.exists(workingDir.resolve("file1"))); assertTrue(Files.exists(workingDir.resolve("file2"))); assertTrue(Files.exists(workingDir.resolve("file3"))); return 1; } }; try (ComputationManager computationManager = new LocalComputationManager(config, localCommandExecutor, ForkJoinPool.commonPool())) { computationManager.execute(new ExecutionEnvironment(ImmutableMap.of("var1", "val1"), PREFIX, false), new AbstractExecutionHandler<Object>() { @Override public List<CommandExecution> before(Path workingDir) throws IOException { Files.createFile(workingDir.resolve("file1")); try (OutputStream os = new GZIPOutputStream(Files.newOutputStream(workingDir.resolve("file2.gz")))) { } try (ZipOutputStream os = new ZipOutputStream(Files.newOutputStream(workingDir.resolve("file3.zip")))) { os.putNextEntry(new ZipEntry("file3")); os.closeEntry(); } Command command = new SimpleCommandBuilder() .id("prog1_cmd") .program("prog1") .args("file1", "file2", "file3") .timeout(60) .inputFiles(new InputFile("file1"), new InputFile("file2.gz", FilePreProcessor.FILE_GUNZIP), new InputFile("file3.zip", FilePreProcessor.ARCHIVE_UNZIP)) .build(); return Collections.singletonList(new CommandExecution(command, 1)); } @Override public Object after(Path workingDir, ExecutionReport report) throws IOException { assertEquals(1, report.getErrors().size()); assertEquals("prog1_cmd", report.getErrors().get(0).getCommand().getId()); assertEquals(0, report.getErrors().get(0).getIndex()); assertEquals(1, report.getErrors().get(0).getExitCode()); return null; } }).join(); } }
@Test(expected = ExecutionException.class) public void hangingIssue() throws Exception { LocalCommandExecutor localCommandExecutor = new AbstractLocalCommandExecutor() { @Override void nonZeroLog(List<String> cmdLs, int exitCode) { } @Override public int execute(String program, List<String> args, Path outFile, Path errFile, Path workingDir, Map<String, String> env) throws IOException, InterruptedException { return 0; } }; try (ComputationManager computationManager = new LocalComputationManager(config, localCommandExecutor, ForkJoinPool.commonPool())) { CompletableFuture<Object> result = computationManager.execute(ExecutionEnvironment.createDefault(), new AbstractExecutionHandler<Object>() { @Override public List<CommandExecution> before(Path workingDir) { throw new AssertionError("Oups"); } }); result.get(100, TimeUnit.MILLISECONDS); } }
@Test(expected = CancellationException.class) public void cancelBeforeExecutionShouldThrowAndNotExecute() throws Exception { LocalCommandExecutor localCommandExecutor = new AbstractLocalCommandExecutor() { @Override void nonZeroLog(List<String> cmdLs, int exitCode) { } @Override public int execute(String program, List<String> args, Path outFile, Path errFile, Path workingDir, Map<String, String> env) throws IOException, InterruptedException { fail(); return 0; } }; CountDownLatch waitForBefore = new CountDownLatch(1); CountDownLatch waitForCancel = new CountDownLatch(1); try (ComputationManager computationManager = new LocalComputationManager(config, localCommandExecutor, ForkJoinPool.commonPool())) { Lock lock = new ReentrantLock(); lock.lock(); CompletableFuture<Object> result = computationManager.execute(ExecutionEnvironment.createDefault(), new AbstractExecutionHandler<Object>() { @Override public List<CommandExecution> before(Path workingDir) { waitForBefore.countDown(); awaitUninterruptibly(waitForCancel); return dummyExecutions(); } @Override public Object after(Path workingDir, ExecutionReport report) throws IOException { fail(); return super.after(workingDir, report); } }); waitForBefore.await(); result.cancel(true); waitForCancel.countDown(); result.get(); } }
@Test public void cancelDuringExecutionShouldThrowAndEventuallyStopExecution() throws InterruptedException, ExecutionException, IOException { CountDownLatch waitForExecution = new CountDownLatch(1); CountDownLatch execution = new CountDownLatch(1); CountDownLatch waitForInterruption = new CountDownLatch(1); MutableBoolean stopped = new MutableBoolean(false); LocalCommandExecutor localCommandExecutor = new AbstractLocalCommandExecutor() { @Override void nonZeroLog(List<String> cmdLs, int exitCode) { } @Override public int execute(String program, List<String> args, Path outFile, Path errFile, Path workingDir, Map<String, String> env) throws IOException, InterruptedException { waitForExecution.countDown(); execution.await(); return 0; } @Override public void stop(Path workingDir) { stopped.setTrue(); waitForInterruption.countDown(); } @Override public void stopForcibly(Path workingDir) { stopped.setTrue(); waitForInterruption.countDown(); } }; try (ComputationManager computationManager = new LocalComputationManager(config, localCommandExecutor, ForkJoinPool.commonPool())) { CompletableFuture<Object> result = computationManager.execute(ExecutionEnvironment.createDefault(), new AbstractExecutionHandler<Object>() { @Override public List<CommandExecution> before(Path workingDir) { return dummyExecutions(); } @Override public Object after(Path workingDir, ExecutionReport report) throws IOException { fail(); return super.after(workingDir, report); } }); waitForExecution.await(); result.cancel(true); result.get(); fail("Should not happen: result has been cancelled"); } catch (CancellationException exc) { } waitForInterruption.await(10, TimeUnit.SECONDS); assertTrue(stopped.isTrue()); }