Implementing CompletableFuture recursively

user_10237736 0
Viewed 74

I have three Tasks:

createSimulationScenarios(), runSimulation(), and getSimulationOutput()

These tasks are done in an iterative manner:

  1. createSimulationScenarios() creates new simulation files
  2. runSimulation() runs a simulation. If there are many simulations they can be executed in parallel batches.
  3. getSimulationOutput() crunches the output data as soon as a simulation is done. Can also be done in parallel as long as the specific simulation is already finished.

Then when all simulations are finished and all outputs are obtained, createSimulationScenarios() creates a new set of simulation files based on the previous batch's output. And so on...

Because each runSimulation() and each getSimulationOutput() can be done in parallel, I thought of using CompletableFuture for the implementation.

for (i = 0; i < noSimulations; ++i) {
    CompletableFuture<Void> runSim = CompletableFuture.runAsync(new runSimulation(), simulationExecutor);
    CompletableFuture<Void> getOutput = runSim.thenCompose(x ->
                                        CompletableFuture.runAsync(new getSimulationOutput(), outputExecutor));
}

However, I am stuck on how to connect this chunk to another CompletableFuture for createSimulationScenarios() with the requirement that it only starts to execute when all simulations and outputs have finished (which is the start of the next iteration).

Any thoughts?

completable-future recursion scheduled-tasks iteration java
2 Answers

Have a look at ForkJoinTask, it seems to fit perfectly for the computation-heavy tasks which scale to run optimally in different environments

RecursiveTask reference shows the example of a simple implementation

user_16496150

Instead of using recursion, you can stick with iteration, but it seems an infinite loop is what you're looking for. This example uses an infinite stream. The CompletableFuture calls will probably need to be changed according to the exact behavior desired (translation: I doubt it will compile, just showing what the code could look like).

Stream.generate(() -> createSimulationScenarios())
    .map(simulations -> IntStream.range(0, simulations.size())
       .mapToObj(i -> CompletableFuture.runAsync(new runSimulation(), simulationExecutor)
          .thenCompose(x -> CompletableFuture.runAsync(new getSimulationOutput(), 
                            outputExecutor)))
       .collect(Collectors.toList())) // force tasks to be submitted
    .forEach(futureList -> futureList.stream().forEach(CompletableFuture::join));
user_57615580
Related