CompletableFuture.allOff completes even if one CompletableFuture in its list is not yet finished

user_2682526 0
Viewed 319

I have 2 CompletableFutures. The task2 should only start once task1 finishes. Then, I need to wait for all tasks to finish. In my code below, the program ends after task1 ends. The task2 starts but did not finish. Any ideas why this happens? Also, why is it that the list only contains 1 entry while in the code, I added 2?

Code:

public void testFutures () throws Exception {
    List<CompletableFuture<Void>> futures = new ArrayList<>();
    CompletableFuture<Void> task1 = CompletableFuture.supplyAsync( () -> {
      System.out.println(" task1 start");
      try {
        Thread.sleep(5000L);
      } catch (InterruptedException ex) {
        ex.printStackTrace();
      }
      System.out.println(" task1 done");
      return null;

    });

    task1.whenComplete( (x, y) -> {
      CompletableFuture<Void> task2 = CompletableFuture.supplyAsync( () -> {
        System.out.println(" task2 start");
        try {
          Thread.sleep(2000L);
        } catch (InterruptedException ex) {
          ex.printStackTrace();
        }
        System.out.println(" task2 done");
        return null;
      });
      futures.add(task2);
    });
    futures.add(task1);
    // wait for the calls to finish
    try {
      CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete( (x, y) -> {
        System.out.println(" all tasks done " + futures.size());
      }).get();

    } catch (Exception e) {
      e.printStackTrace();
    }
  }

Output:

 task1 start
 task1 done
 all tasks done 1
 task2 start
completable-future asynchronous java
2 Answers

Let's first clean your code.

Let's define a method that will do the sleeping, so that it does not muddy the water:

private static void sleep(int seconds) {
    try {
        Thread.sleep(TimeUnit.SECONDS.toMillis(seconds));
    } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
    }
}

Then let's separate the tasks and use proper methods:

private static CompletableFuture<Void> task1() {

    return CompletableFuture.runAsync(() -> {
        System.out.println(" task1 start");
        sleep(5);
        System.out.println(" task1 done");
    });
}

private static CompletableFuture<Void> task2() {
    return CompletableFuture.runAsync(() -> {
        System.out.println(" task2 start");
        sleep(2);
        System.out.println(" task2 done");
    });
}

You need to understand that chaining of CompletableFuture methods already do exactly what you want, they run the next stage, after the previous one has ended. You can make your code far, far more easy with:

public static void main(String[] args) throws Exception {
    testFutures();
}

private static void testFutures() throws Exception {

    CompletableFuture<Void> both = task1().thenCompose(ignoreMe -> task2());
    both.get();
    System.out.println("both done");

}
user_10593720

You have two problems.

First, you've created a race condition as to when task2 gets added to your list of futures. At the time you execute this line—

CompletableFuture.allOf(...).get();

—which I'll call the terminating getter, you only have task1 in the list. See for yourself by outputting its size:

// wait for the calls to finish
try {
    System.out.println("# of futures: " + futures.size()); // 1

task2 still runs eventually, because you scheduled it with whenComplete(). But it's not your terminating getter that triggers it.

Recall that I said it's a race condition. To demonstrate this for yourself, add a sleep() before the terminating getter, like so:

try {
  Thread.sleep(6000L);
} catch (InterruptedException ex) {
  ex.printStackTrace();
}
// wait for the calls to finish
try {
    System.out.println("# of futures: " + futures.size()); // 2

Then you'll have given it enough time to add task2.

But here's the thing. Now is the terminating getter triggering both tasks?

Still no! And that's the second problem: You almost always want to use one of the thenRun(), thenAccept(), thenApply(), thenCompose() methods. These methods chain your futures, i.e. make each stage dependent on the previous, so that your terminating getter actually waits for the entire chain to complete. whenComplete() is a special method that kicks off a totally unrelated pipeline and is thus unaffected by the terminating get().

In your case, you want to use thenRun(), like this:

    task1.thenRun( ignore -> {

Okay, so how do we combine all that?

public static void testFutures () throws Exception {
    
    CompletableFuture<Void> task1 = CompletableFuture.supplyAsync( () -> {
      System.out.println(" task1 start");
      try {
        Thread.sleep(5000L);
      } catch (InterruptedException ex) {
        ex.printStackTrace();
      }
      System.out.println(" task1 done");
      return null;
    });

    CompletableFuture<Void> futuresChain = task1.thenRun( () -> {
      System.out.println(" task2 start");
      try {
        Thread.sleep(2000L);
      } catch (InterruptedException ex) {
        ex.printStackTrace();
      }
      System.out.println(" task2 done");
    });
    
    // wait for the calls to finish
    try {
      futuresChain.thenRun( () -> {
        System.out.println(" all tasks done ");
      }).toCompletableFuture().get();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

Output:

 task1 start
 task1 done
 task2 start
 task2 done
 all tasks done

You see, you only need to supplyAsync() for the first task. You want to run task2 sequentially after that task, so thenRun() will do the scheduling (the supplyAsync()ing) for you. So you don't need an array of futures either. The allOf() is for when you want to run tasks in parallel, and wait for all of them to finish.

user_9259130
Related