A Coroutines Framework for Java

Inspired by coroutines in Go and Kotlin, we provide a new Java framework for suspending coroutines. Other than existing Java solutions, it is implemented as a simple dependency without using JNI or the need to perform bytecode manipulation from separate Java Agents. It also applies the pattern of structured concurrency, which has been inspired by the noteworthy essay: Notes on structured concurrency, or: Go statement considered harmful by Nathaniel J. Smith. The project is available on GitHub under the Apache 2.0 license and the documentation of the usage and features can be found on the project page. An extended introductory article is also available here.

Because Java doesn't have language support for coroutines, the declaration of coroutines needs to be done through an API. Fortunately, the functional language features available since Java 8, like lambda expressions and method references, together with static imports, allow a concise declaration and execution of coroutines. The underlying implementation is based on Java's CompletableFuture and standard thread pools. Even if a suspension is not needed, the framework can be used as a simpler and structured interface to concurrency.

The application of coroutines is split into two parts: declaration and execution. The declaration is done through a simple builder pattern that composes coroutine from single steps. Such steps can either be simple code executions or suspending functionality, like waiting for I/O to complete or communication with other coroutines that run in parallel. The framework contains several standard coroutine step implementations, including the execution of functional expressions (like lambdas). The following shows a simple coroutine example:

import static de.esoco.coroutine.Coroutine.*;
import static de.esoco.coroutine.step.CodeExecution.*;

Coroutine<String, Integer> parseInteger = first(apply(String::trim))
                                          .then(apply(s -> Integer.valueOf(s)));

This also demonstrates how static imports can be used to simplify the declaration. first() is a factory method that creates a new coroutine instance. It provides a fluent interface for the declaration but a constructor could also be used. The instance method then()extends a coroutine with another execution step. It should be noted that coroutines are always immutable. Extending them with new steps will always create a new coroutine instance. Coroutines are declared with generic types for their input and output values because they can be used similar to Java's Function interface but in asynchronous executions.

The argument to the coroutine builder methods are instances of CoroutineStep. This can either be a custom implementation or one of the predefined step implementations in the framework. The example above shows the most basic CodeExecution step, which provides factory methods, like apply(Function) for steps, that execute functional interfaces.

The executions of coroutines always occur in a CoroutineScope. Scopes provide the environment for structured concurrency and make sure that a set of concurrent executions is always tracked, including cases where exceptions are thrown. The following code shows the execution of a million coroutines in a scope. Launching that amount of parallel threads would either be much slower or could even cause an out of memory error.

import static de.esoco.coroutine.Coroutine.launch;

Coroutine<?, ?> crunchNumbers =
    first(run(() -> Range.from(1).to(10).forEach(Math::sqrt)));

launch(scope -> {
    for (int i = 0; i < 1_000_000; i++) {
        crunchNumbers.runAsync(scope);
    }
});

Again, a static import is used to allow a short notation for the launching of the scope. The lauch() method creates a new scope and executes a functional interface in it with the scope as it's argument. The lambda expression then runs the coroutine asynchronously in the configured thread pool.

The most important property of the scope is that the code after the launch block will only continue executing after all the launched coroutines have finished execution or thrown an error (in which case the scope will also throw an exception). That makes it impossible to ignore the execution state of coroutines and requires the handling of all errors that might occur.

In some cases, it may be more desirable to have an object to track the execution of a scope instead of blocking the current thread. This is especially the case if the coroutines in a scope are used to generate some result that needs to be processed later. For that purpose, the produce() method can be used instead of launch(). That method returns immediately with an instance of Future that can then be used to track the execution of a scope.

The coroutines framework provides a lot more functionality, like continuations, to track single executions, execution contexts, control structures, suspending coroutine steps for asynchronous I/O and non-blocking communication through channels, and advanced concepts, like selection.