@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); } }
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 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 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); }
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(); }
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();  }