Structured Concurrency in Java: Fixing Unstructured Concurrent Code

// Unstructured — what happens on errors or cancellation?
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

Future<User> userFuture = executor.submit(() -> fetchUser(userId));
Future<List<Order>> ordersFuture = executor.submit(() -> fetchOrders(userId));

try {
    User user = userFuture.get(5, SECONDS);
    List<Order> orders = ordersFuture.get(5, SECONDS);
    return new UserProfile(user, orders);
} catch (ExecutionException e) {
    // Which task failed? What about the other task?
    // It's still running. We're leaking it.
    throw new RuntimeException(e.getCause());
} catch (TimeoutException e) {
    // We timed out waiting — but both tasks are still running.
    // We have no clean way to cancel them.
    throw new ServiceException("Timed out");
}

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<User> userTask       = scope.fork(() -> fetchUser(userId));
    Subtask<List<Order>> ordersTask = scope.fork(() -> fetchOrders(userId));

    scope.join();           // Wait for both tasks to complete
    scope.throwIfFailed();  // Rethrow any exception from a failed task

    return new UserProfile(userTask.get(), ordersTask.get());
}
// When the try block exits:
// - All forked tasks are guaranteed to be complete
// - Any running tasks are cancelled
// - Resources are released

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<User>       user    = scope.fork(() -> fetchUser(userId));
    Subtask<List<Order>> orders = scope.fork(() -> fetchOrders(userId));
    Subtask<Preferences> prefs  = scope.fork(() -> fetchPreferences(userId));

    scope.join().throwIfFailed(); // If any fails, others are cancelled

    return new Dashboard(user.get(), orders.get(), prefs.get());
}

try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
    scope.fork(() -> fetchFromPrimaryCache(key));
    scope.fork(() -> fetchFromSecondaryCache(key));
    scope.fork(() -> fetchFromDatabase(key));

    scope.join(); // Returns when any task succeeds

    return scope.result(); // Returns the first successful result
}

class CollectResultsScope<T> extends StructuredTaskScope<T> {
    private final List<T> results = new CopyOnWriteArrayList<>();
    private final List<Throwable> failures = new CopyOnWriteArrayList<>();

    @Override
    protected void handleComplete(Subtask<? extends T> subtask) {
        switch (subtask.state()) {
            case SUCCESS -> results.add(subtask.get());
            case FAILED  -> failures.add(subtask.exception());
            default      -> {} // cancelled
        }
    }

    List<T> results() { return Collections.unmodifiableList(results); }
    List<Throwable> failures() { return Collections.unmodifiableList(failures); }
}

// Usage: collect all results, tolerate some failures
try (var scope = new CollectResultsScope<SearchResult>()) {
    searchSources.forEach(source -> scope.fork(() -> source.search(query)));
    scope.join();
    // Proceed even if some sources failed
    return merge(scope.results());
}

// Parent task cancellation propagates to children
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<String> task1 = scope.fork(() -> {
        // If interrupted, throws InterruptedException
        Thread.sleep(Duration.ofSeconds(10));
        return "done";
    });

    scope.join();
    // When the calling thread is interrupted while waiting in join(),
    // the scope shuts down, task1 receives interrupt, try-block exits cleanly
}

try (var outerScope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<UserProfile> profileTask = outerScope.fork(() -> {
        // Inner scope — its tasks are scoped to this lambda
        try (var innerScope = new StructuredTaskScope.ShutdownOnFailure()) {
            Subtask<User> user   = innerScope.fork(() -> fetchUser(userId));
            Subtask<Settings> s  = innerScope.fork(() -> fetchSettings(userId));
            innerScope.join().throwIfFailed();
            return new UserProfile(user.get(), s.get());
        }
    });

    Subtask<List<Order>> ordersTask = outerScope.fork(() -> fetchOrders(userId));

    outerScope.join().throwIfFailed();
    return new Dashboard(profileTask.get(), ordersTask.get());
}