WorkStealingPool and ThreadPoolExecutor yields different results when used with CompletableFuture

user_5208889 0
Viewed 410

When executing the following code:

ExecutorService executorService = Executors.newWorkStealingPool(20);

Function<String, CompletableFuture<String>> requestTask =
        url -> CompletableFuture.supplyAsync(() -> {
                    System.out.println("Request " + requestCount++ + " was sent");
                    HttpClient.get(url);
                    return url;
                    }, executorService);

Function<String, String> extractName = s -> s.replaceAll("(https|http|://|\\.com|www\\.|\\.io)", "");

CompletableFuture[] futures = urls.stream() // urls list contains 14 urls
        .map(requestTask)
        .map(future -> future.thenApply(extractName))
        .map(future -> future.thenAccept(System.out::println))
        .toArray(CompletableFuture[]::new);

CompletableFuture.allOf(futures);
executorService.shutdown();

the result is the following:

Request 0 was sent
Request 1 was sent
Request 2 was sent
Request 3 was sent
Request 4 was sent
Request 5 was sent
Process finished with exit code 0

However, when Executors.newWorkStealingPool(20) is replaced with Executors.newFixedThreadPool(20) all the requests are sent. What is the reason for this behavior?

completable-future executorservice concurrency java
1 Answers

Not all requests are sent because (spoiler alert!) the JVM simply terminates.

As you may know, the conditions for JVM termination are:

either of the following occurs:

  • The exit method of class Runtime has been called and the security manager has permitted the exit operation to take place.
  • All threads that are not daemon threads have died, either by returning from the call to the run method or by throwing an exception that propagates beyond the run method.

Obviously it is not the first case, so it must be the second one.

The first thing to note is that you let your main() method exit:

  • you call CompletableFuture.allOf() but you don't do anything with the result so it does not block (no join() call);
  • calling ExecutorService.shutDown() only tells the executor to shutdown, it does not wait for it.

Initially, the main thread is the only non-daemon thread, so this should be sufficient for the JVM to exit. But this is where the 2 executors make a difference:

  • newFixedThreadPool() is implemented by a ThreadPoolExecutor, which uses the Executors.defaultThreadFactory(), which creates non-daemon threads;
  • newWorkStealingPool() is implemented by a ForkJoinPool, which calls setDaemon(true) on all threads it creates¹.

Unfortunately it is not documented, but basically, this boils down to Why does the following application terminate immediately when using ForkJoinPool, but not when I use ThreadPoolExecutor?

So two possible solutions for your problem:

  • call join() after allOf()
  • call awaitTermination() on the executor after shutdown()

¹ As noted by teppic in the comments, this was not documented in Java 8, but it is now since Java 9.

user_5250360
Related