PooledCompletableFuture

A wrapper around CompletableFuture to make working with futures and asynchronous operation easier in Java.

Concepts

Pipelining

Pipelining is where you organise your code to execute in sequential chains of tasks, where each task begins processing as soon as the previous one completes.

Suppose we have a scenario where we need to process data through multiple stages:

1. Retrieve raw data from a remote source
2. Perform computations or transformations on that fetched data
3. Save the processed data to a database Using PooledCompletableFuture, we can pipeline these stages as follows:

importjava.util.concurrent.CompletableFuture;
importjava.util.concurrent.Executor;
importjava.util.concurrent.Executors;
ExecutorcustomExecutor = Executors.newFixedThreadPool(4);
CompletableFuture<String> dataFromRemoteSource = fetchDataFromRemoteSource();
PooledCompletableFuture.of(dataFromRemoteSource, customExecutor)
.thenApply(rawData -> {
PooledCompletableFuture<ProcessedData> processedData = processData(rawData);
returnprocessedData;
})
.thenCompose(x -> x)
.thenApply(processedData -> {
PooledCompletableFuture<Void> saveTask = saveInDatabase(processedData);
returnprocessedData;
})
.thenCompose(x -> x)
.exceptionally(e -> {
System.err.println("An error occurred: " + e.getMessage());
returnnull;
});

In this scenario, where:

• fetchDataFromRemoteSource() is a method from an external library which returns a normal CompletableFuture<String>
• processData(String) and saveInDatabase(ProcessedData) are owned by us, and use PooledCompletableFutures this shows how you can pipeline your execution into a linear sequence of tasks.

Downwind asynchronous execution

Following on from the previous point, a benefit of this library is that when you specify an executor, it follows through downwind in your pipeline until otherwise specified. This means if you specify your execution at the very beginning, any executions happening from instance methods on PooledCompletableFuture occur on the executor you specified originally. Here's how it looks if you want to change what executor you want to use:

importjava.util.concurrent.CompletableFuture;
importjava.util.concurrent.Executor;
importjava.util.concurrent.Executors;
ExecutorcustomExecutor = Executors.newFixedThreadPool(4);
ExecutordatabaseExecutor = Executors.newSingleThreadExecutor();
CompletableFuture<String> dataFromRemoteSource = fetchDataFromRemoteSource();
PooledCompletableFuture.of(dataFromRemoteSource, customExecutor)
.thenApply(rawData -> { // thenApply uses customExecutorPooledCompletableFuture<ProcessedData> processedData = processData(rawData);
returnprocessedData;
})
.async(databaseExecutor) // switching to use the databaseExecutor
.thenCompose(x -> x) // this method (and onwards) uses databaseExecutor
.thenApply(processedData -> { // this method uses databaseExecutorPooledCompletableFuture<Void> saveTask = saveInDatabase(processedData);
returnprocessedData;
})
.thenCompose(x -> x)
.exceptionally(e -> { // if an error is thrown, this method uses databaseExecutorSystem.err.println("An error occurred: " + e.getMessage());
returnnull;
});

Programs with a "Main Thread"

A main thread is a single thread (or a thread pool) where execution of a program primarily happens. In Minecraft, this would be the server thread. The PooledCompletableFuture library allows you to specify the main thread and then use sync() in the pipeline to move back to the main thread seamlessly.

You can achieve this by doing:

importme.emissions.concurrency.PooledExecutors;
PooledExecutors.setMainThread(<yourexecutor>);

For example, with Bukkit:

importme.emissions.concurrency.PooledExecutors;
publicclassSomePluginextendsJavaPlugin {
[...]
@OverridepublicvoidonLoad() {
PooledExecutors.setMainThread(Bukkit.getServer()
.getScheduler()
.getMainThreadExecutor(this)
);
}
[...]
}

Switching to the main thread during pipelining:

importjava.util.concurrent.CompletableFuture;
importjava.util.concurrent.Executor;
importjava.util.concurrent.Executors;
ExecutorcustomExecutor = Executors.newFixedThreadPool(4);
ExecutordatabaseExecutor = Executors.newSingleThreadExecutor();
CompletableFuture<String> dataFromRemoteSource = fetchDataFromRemoteSource();
PooledCompletableFuture.of(dataFromRemoteSource, customExecutor)
.thenApply(rawData -> { // thenApply uses customExecutorPooledCompletableFuture<ProcessedData> processedData = processData(rawData);
returnprocessedData;
})
.async(databaseExecutor) // switching to use the databaseExecutor
.thenCompose(x -> x) // this method (and onwards) uses databaseExecutor
.thenApply(processedData -> { // this method uses databaseExecutorPooledCompletableFuture<Void> saveTask = saveInDatabase(processedData);
returnprocessedData;
})
.thenCompose(x -> x)
.exceptionally(e -> { // if an error is thrown, this method uses databaseExecutorSystem.err.println("An error occurred: " + e.getMessage());
returnnull;
})
.sync() // switch to the main thread you specified
.thenRun(() -> {
// execution here occurs on the main thread!System.out.println("done task!");
})

Pooled Executors

Pooled executors are a concept unique to this library. As standard, this library ships with executors laying dormant for use in asynchronous tasks. See the PooledExecutors class.

This class creates (by default) 50 single thread executors, which any asynchronously executed task will use if not specified otherwise.

NOTE: the reason for generating 50 single thread executors (compared to a thread pool) is that this largely prevents deadlocks

with .join() and .get() -- you can deadlock a thread pool very easily!

importjava.util.concurrent.CompletableFuture;
importjava.util.concurrent.Executor;
importjava.util.concurrent.Executors;
CompletableFuture<String> dataFromRemoteSource = fetchDataFromRemoteSource();
PooledCompletableFuture.of(dataFromRemoteSource) // Immediately uses the pooled executors
.thenApply(rawData -> { // executes on a random pooled executorPooledCompletableFuture<ProcessedData> processedData = processData(rawData);
returnprocessedData;
})
.thenCompose(x -> x) // executes on a random pooled executor
.thenApply(processedData -> { // executes on a random pooled executorPooledCompletableFuture<Void> saveTask = saveInDatabase(processedData);
returnprocessedData;
})
.thenCompose(x -> x) // executes on a random pooled executor
.exceptionally(e -> { // executes on a random pooled executorSystem.err.println("An error occurred: " + e.getMessage());
returnnull;
});

The above shows how pooled executors are used by default.

But how do I switch to using pooled executors mid-pipeline?

importjava.util.concurrent.CompletableFuture;
importjava.util.concurrent.Executor;
importjava.util.concurrent.Executors;
ExecutorcustomExecutor = Executors.newFixedThreadPool(4);
CompletableFuture<String> dataFromRemoteSource = fetchDataFromRemoteSource();
PooledCompletableFuture.of(dataFromRemoteSource, customExecutor)
.thenApply(rawData -> { // thenApply uses customExecutorPooledCompletableFuture<ProcessedData> processedData = processData(rawData);
returnprocessedData;
})
.async() // switching to use pooled executors <--! HERE !-->
.thenCompose(x -> x) // this method (and onwards) uses a random pooled executor
.thenApply(processedData -> { // this method uses uses a random pooled executorPooledCompletableFuture<Void> saveTask = saveInDatabase(processedData);
returnprocessedData;
})
.thenCompose(x -> x) // uses a random pooled executor
.exceptionally(e -> { // if an error is thrown, this method uses uses a random pooled executorSystem.err.println("An error occurred: " + e.getMessage());
returnnull;
})
.sync() // switch to the main thread you specified
.thenRun(() -> {
// execution here occurs on the main thread!System.out.println("done task!");
})