Interface StructuredTaskScope<T, R, R_X extends Throwable>

Type Parameters:
T - the result type of subtasks forked in the scope
R - the type of the result returned by the join() method
R_X - the type of the exception thrown by the join() method
All Superinterfaces:
AutoCloseable
public sealed interface StructuredTaskScope<T, R, R_X extends Throwable> extends AutoCloseable
An API for structured concurrency. StructuredTaskScope supports cases where execution of a task (a unit of work) splits into several concurrent subtasks, and where the subtasks must complete before the task continues. A StructuredTaskScope can be used to ensure that the lifetime of a concurrent operation is confined by a syntactic block, similar to that of a sequential operation in structured programming.

StructuredTaskScope defines the static open() method to create and open a new StructuredTaskScope. It defines the close() method to close it. The API is designed to be used with the try-with-resources statement where a StructuredTaskScope is opened as a resource and then closed automatically. The code inside the try block forks subtasks with the fork(Callable) or fork(Runnable) methods. Each call to fork(Callable) starts a new Thread (typically a virtual thread) to execute a value-returning method as a subtask. Each call to fork(Runnable) starts a new Thread to execute a method that does not return a result as a subtask. The threads executing subtasks run concurrently with each other and with the code inside the try block. After forking all subtasks, the code inside the block uses the join() method to wait for all subtasks to finish (or for another outcome) as a single operation. The code after the join() method processes the outcome. Execution does not continue beyond the try block (or close method) until all threads started in the scope to execute subtasks have finished.

As a first example, consider a "main" task that splits into two subtasks to concurrently fetch values from two remote services. The main task aggregates the results of both subtasks. The example invokes fork(Callable) to fork the two subtasks. Each call to fork(Callable) returns a Subtask as a handle to the forked subtask. Both subtasks may succeed, one may succeed and the other fail, or both may fail. 

   try (var scope = StructuredTaskScope.open()) {

       Subtask<String> subtask1 = scope.fork(() -> fetchFromRemoteService1());
       Subtask<Integer> subtask2 = scope.fork(() -> fetchFromRemoteService2());

       // throws ExecutionException if either subtask fails
       scope.join();

       // both subtasks completed successfully
       var result = new MyResult(subtask1.get(), subtask2.get());

   } // close

The main task in the example is interested in the successful result of both subtasks. It waits in the join() method for both subtasks to complete successfully or for either subtask to fail. If both subtasks complete successfully then the join() method completes normally and the task uses the Subtask.get() method to get the result of each subtask. If one of the subtasks fails then the other subtask is cancelled, and the join() method throws ExecutionException with the exception from the failed subtask as the cause.

The close() method always waits for threads executing subtasks to finish, even if the join() method throws, so that execution cannot continue beyond the close() method until all subtasks have finished.

To ensure correct usage, the following restrictions are enforced:

  • The fork methods, the join() method and close() method may only be invoked by the owner thread (the thread that opened the StructuredTaskScope).
  • The join() method must be invoked to get the outcome after forking subtasks (the close method throws an exception after closing if the owner did not invoke the join method after forking subtasks).
  • Neither fork method may be called after join() is invoked.

To allow for cancellation, subtasks must be coded so that they finish as soon as possible when interrupted. Subtasks that do not respond to interrupt, e.g. block on methods that are not interruptible, may delay the close() method indefinitely.

In the example, the subtasks produce results of different types (String and Integer). In other cases where the subtasks produce results of the same type, the scope can be opened with that type as the subtask result type. For example, StructuredTaskScope.<String>open() opens a scope that can only be used to fork subtasks that produce a String result.

Joiners

The join() method in the example above completes normally, and returns null, if all subtasks succeed. It throws ExecutionException if any subtask fails. Other policies and outcomes are possible by creating a StructuredTaskScope with a Joiner that implements the desired policy and outcome. A Joiner handles subtasks as they are forked and when they complete, and produces the outcome for the join() method. Instead of null, a Joiner may cause join() to return the result of a specific subtask, a collection of results, or an object constructed from the results of some or all subtasks. When the outcome is an exception, a Joiner may cause join() to throw an exception other than ExecutionException.

Cancellation: A Joiner may cancel the scope (sometimes called "short-circuiting") when some condition is reached, for example, when a subtask fails, and the outcomes (results or exceptions) of other subtasks that are still executing are not required. Cancelling the scope prevents new threads from being started in the scope, cancels subtasks in the scope that have not completed execution (this will interrupt the threads executing the subtasks) and causes the join() method to wake up with the outcome. In the above example, the outcome is that join() completes normally when all subtasks succeed. The scope is cancelled if any subtask fails and join() throws ExecutionException with the exception from a failed subtask as the cause. Other Joiner implementations may cancel the scope for other reasons, and may cause join() to throw a different exception when the outcome is an exception.

The Joiner interface defines static factory methods to create a Joiner for a number of common cases. The interface can be implemented when a more advanced or custom policy is required. A Joiner that returns a non-null result may remove the need for bookkeeping and the need to keep a reference to Subtask objects returned by the fork(Callable) method.

Now consider an example where a main task splits into two subtasks and where the main task is interested in the successful outcome of either subtask. The example uses Joiner.anySuccessfulOrThrow() to create a Joiner that produces the result of any subtask that completes successfully. 

   try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulOrThrow())) {

       scope.fork(callable1);
       scope.fork(callable2);

       // throws ExecutionException if both subtasks fail
       String anyResult = scope.join();

   } // close

The main task waits in the join() method for either subtask to complete successfully or for both subtasks to fail. If one of the subtasks completes successfully, then the Joiner causes the other subtask to be cancelled (this will interrupt the thread executing the subtask), and the join() method returns the result of the successful subtask. Cancelling the other subtask avoids the task waiting for a result that it doesn't care about. If both subtasks fail, then the join() method throws ExecutionException with the exception from one of the subtasks as the cause. Joiner.anySuccessfulOrThrow(Function) can be used with a function that produces an exception other than ExecutionException to throw when all subtasks fail.

Configuration

A StructuredTaskScope is opened with configuration that consists of a ThreadFactory to create threads, an optional name for the scope, and an optional timeout. The name is intended for monitoring and management purposes.

The open() and open(Joiner) methods create a StructuredTaskScope with the default configuration. The default configuration has a ThreadFactory that creates unnamed virtual threads, does not name the scope, and has no timeout.

The open(UnaryOperator) and open(Joiner, UnaryOperator) methods can be used to create a StructuredTaskScope that uses a different ThreadFactory, is named for monitoring and management purposes, or has a timeout that cancels the scope if the timeout expires before the join() method is invoked or completes. The open methods are called with an operator that is applied to the default configuration and returns a Configuration for the StructuredTaskScope under construction.

The following example opens a new StructuredTaskScope with a ThreadFactory that creates virtual threads named "duke-0", "duke-1" ... 

    ThreadFactory factory = Thread.ofVirtual().name("duke-", 0).factory();

    try (var scope = StructuredTaskScope.open(cf -> cf.withThreadFactory(factory))) {

        var subtask1 = scope.fork( .. );   // runs in a virtual thread with name "duke-0"
        var subtask2 = scope.fork( .. );   // runs in a virtual thread with name "duke-1"

        scope.join();

        var result = new MyResult(subtask1.get(), subtask2.get());

     }

A second example sets a timeout, represented by a Duration. The timeout starts when the new scope is opened. If the timeout expires before the join() method is invoked or completes then the scope is cancelled (this interrupts the threads executing the subtasks that have not completed), and the join() method throws ExecutionException with a CancelledByTimeoutException as the cause. 

   Duration timeout = Duration.ofSeconds(10);

   try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(),
                                             cf -> cf.withTimeout(timeout))) {

       scope.fork(callable1);   // subtask takes a really long time
       scope.fork(callable2);

       // throws ExecutionException with CancelledByTimeoutException as cause
       List<String> results = scope.join();

   }

Exception handling

The outcome of the join() method is a result or exception. When the outcome is an exception, its cause will typically be the exception from a failed subtask or CancelledByTimeoutException if a timeout was configured.

In some cases it may be useful to add a catch block to the try-with-resources statement to handle the exception. The following example uses the open(UnaryOperator) method to open a scope with a timeout configured. The join() method in this example throws ExecutionException if any subtask fails or the timeout expires before the join() method is invoked or completes. The exception cause is the exception from a failed subtask or CancelledByTimeoutException. The example uses the switch statement to select based on the cause. 

   try (var scope = StructuredTaskScope.open(cf -> cf.withTimeout(timeout))) {

       ..

   } catch (ExecutionException e) {
       switch (e.getCause()) {
           case CancelledByTimeoutException ->
           case IOException ioe -> ..
           default -> ..
       }
   }

In other cases it may not be useful to catch the exception but instead leave it to propagate to the configured uncaught exception handler for logging purposes.

For cases where a specific exception triggers the use of a default result then it may be more appropriate to handle this in the subtask itself rather than the subtask failing and the scope owner handling the exception.

The join() method throws InterruptedException if the owner thread is interrupted while waiting in the join() method. The Thread Interruption section of the Thread specification provides guidance on handling this exception.

Inheritance of scoped value bindings

ScopedValue supports the execution of a method with a ScopedValue bound to a value for the bounded period of execution of the method by the current thread. It allows a value to be safely and efficiently shared to methods without using method parameters.

When used in conjunction with a StructuredTaskScope, a ScopedValue can also safely and efficiently share a value to methods executed by subtasks forked in the scope. When a ScopedValue object is bound to a value in the thread executing a "main" task then that binding is inherited by the threads created to execute subtasks. The thread executing the main task does not continue beyond the close() method until all threads executing the subtasks have finished. This ensures that the ScopedValue is not reverted to being unbound (or its previous value) while subtasks are executing. In addition to providing a safe and efficient means to inherit a value into subtasks, the inheritance allows sequential code using ScopedValue to be refactored to use structured concurrency.

To ensure correctness, opening a new StructuredTaskScope captures the current thread's scoped value bindings. These are the scoped value bindings that are inherited by the threads created to execute subtasks in the scope. Forking a subtask checks that the bindings in effect at the time that the subtask is forked match the bindings when the StructuredTaskScope was created. This check ensures that a subtask does not inherit a binding that is reverted in the main task before the subtask has completed.

A ScopedValue that is shared across threads requires that the value be an immutable object or for all access to the value to be appropriately synchronized.

The following example demonstrates the inheritance of scoped value bindings. The scoped value USERNAME is bound to the value "duke" for the bounded period of a lambda expression by the thread executing it. The code in the block opens a StructuredTaskScope and forks two subtasks, then waits in the join() method and aggregates the results from both subtasks. If code executed by the threads running subtask1 and subtask2 uses ScopedValue.get() to get the value of USERNAME, then the value "duke" will be returned. 

    private static final ScopedValue<String> USERNAME = ScopedValue.newInstance();

    MyResult result = ScopedValue.where(USERNAME, "duke").call(() -> {

        try (var scope = StructuredTaskScope.open()) {

            Subtask<String> subtask1 = scope.fork( .. );    // inherits binding
            Subtask<Integer> subtask2 = scope.fork( .. );   // inherits binding

            scope.join();
            return new MyResult(subtask1.get(), subtask2.get());
        }

    });

A scoped value inherited into a subtask may be rebound to a new value in the subtask for the bounded execution of some method executed in the subtask. When the method completes, the value of the ScopedValue reverts to its previous value, the value inherited from the thread executing the main task.

A subtask may execute code that itself opens a new StructuredTaskScope. A main task executing in thread T1 opens a StructuredTaskScope and forks a subtask that runs in thread T2. The scoped value bindings captured when T1 opens the scope are inherited into T2. The subtask (in thread T2) executes code that opens a new StructuredTaskScope and forks a (sub-)subtask that runs in thread T3. The scoped value bindings captured when T2 opens the scope are inherited into T3. These include the bindings that were inherited from T1. In effect, scoped values are inherited into a tree of subtasks, not just one level of subtask.

Memory consistency effects

Actions in the owner thread of a StructuredTaskScope prior to forking of a subtask happen-before any actions taken by the thread that executes the subtask, which in turn happen-before actions in any thread that successfully obtains the subtask outcome with Subtask.get() or Subtask.exception(). If a subtask's outcome contributes to the result or exception from join(), then any actions taken by the thread executing that subtask happen-before the owner thread returns from join() with the outcome.

General exceptions

Unless otherwise specified, passing a null argument to a method in this class will cause a NullPointerException to be thrown.

See Java Language Specification:
17.4.5 Happens-before Order
Since:
28

  • Method Details

    • open

      static <T, R, R_X extends Throwable> StructuredTaskScope<T,R,R_X> open(StructuredTaskScope.Joiner<? super T, ? extends R, R_X> joiner, UnaryOperator<StructuredTaskScope.Configuration> configOperator)
      Opens a new StructuredTaskScope that uses the given Joiner object and the Configuration that is the result of applying the given operator to the default configuration.

      The Joiner specified to this method implements the desired policy and produces the outcome (result or exception) for the join() method when all subtasks forked in the scope complete execution or the scope is cancelled.

      This method invokes the operator's apply method with the default configuration to produce the configuration for the new scope:

      • If the apply method returns a configuration with a ThreadFactory, the ThreadFactory's newThread(Runnable) method is invoked to create threads when forking subtasks in the scope. If a ThreadFactory is not set then forking subtasks will create an unnamed virtual thread for each subtask.
      • If the apply method returns a configuration with a timeout, the timeout starts when the scope is opened. If the timeout expires before join() is invoked or completes then the scope is cancelled asynchronously. Whether the join() method returns a result or throws an exception when a timeout occurs is Joiner-specific. If the outcome is an exception then it will be thrown with a CancelledByTimeoutException as the cause.
      • If the apply method returns a configuration with a name, the new scope is created with this name for monitoring and management purposes.
      • If the apply method throws an exception or error then it is propagated by this method.
      • If the apply method returns null then NullPointerException is thrown.

      The new scope is owned by the current thread. Only code executing in this thread can fork, join, or close the scope.

      Construction captures the current thread's scoped value bindings for inheritance by threads forked in the scope.

      Type Parameters:
      T - the result type of subtasks forked in the scope
      R - the type of the result returned by the join() method
      R_X - the type of the exception thrown by the join() method
      Parameters:
      joiner - the Joiner
      configOperator - the operator to produce the configuration
      Returns:
      a new scope

    • open

      static <T, R, R_X extends Throwable> StructuredTaskScope<T,R,R_X> open(StructuredTaskScope.Joiner<? super T, ? extends R, R_X> joiner)
      Opens a new StructuredTaskScope that uses the given Joiner object. The Joiner implements the desired policy and produces the outcome (result or exception) for the join() method when all subtasks forked in the scope complete execution or the scope is cancelled.

      The scope is created with the default configuration. The default configuration has a ThreadFactory that creates unnamed virtual threads, does not name the scope, and has no timeout.

      This method is equivalent to invoking the 2-arg open method with the given Joiner and the identity operator.

      The new scope is owned by the current thread. Only code executing in this thread can fork, join, or close the scope.

      Construction captures the current thread's scoped value bindings for inheritance by threads forked in the scope.

      Type Parameters:
      T - the result type of subtasks forked in the scope
      R - the type of the result returned by the join() method
      R_X - the type of the exception thrown by the join() method
      Parameters:
      joiner - the Joiner
      Returns:
      a new scope

    • open

      static <T> StructuredTaskScope<T, Void, ExecutionException> open(UnaryOperator<StructuredTaskScope.Configuration> configOperator)
      Opens a new StructuredTaskScope that uses the Configuration that is the result of applying the given operator to the default configuration. This method invokes the operator's apply method to produce the configuration for the new scope, as specified by the 2-arg open method.

      The join() method of the new scope waits for all subtasks to succeed or any subtask to fail. The join() method returns null if all subtasks complete successfully. It throws ExecutionException if any subtask fails, with the exception from a failed subtask as the cause. If a timeout is configured, and the timeout expires before the join() method is invoked or completes, it throws ExecutionException with a CancelledByTimeoutException as the cause.

      This method is equivalent to invoking the 2-arg open method with a Joiner created with Joiner.awaitAllSuccessfulOrThrow() and the given configuration operator.

      The new scope is owned by the current thread. Only code executing in this thread can fork, join, or close the scope.

      Construction captures the current thread's scoped value bindings for inheritance by threads forked in the scope.

      Type Parameters:
      T - the result type of subtasks forked in the scope
      Parameters:
      configOperator - the operator to produce the configuration
      Returns:
      a new scope

    • open

      static <T> StructuredTaskScope<T, Void, ExecutionException> open()
      Opens a new StructuredTaskScope where join() waits for all subtasks to succeed or any subtask to fail. The join() method returns null if all subtasks complete successfully. It throws ExecutionException if any subtask fails, with the exception from a failed subtask as the cause.

      The scope is created with the default configuration. The default configuration has a ThreadFactory that creates unnamed virtual threads, does not name the scope, and has no timeout.

      This method is equivalent to invoking the 2-arg open method with a Joiner created with Joiner.awaitAllSuccessfulOrThrow() and the identity operator.

      The new scope is owned by the current thread. Only code executing in this thread can fork, join, or close the scope.

      Construction captures the current thread's scoped value bindings for inheritance by threads forked in the scope.

      Type Parameters:
      T - the result type of subtasks
      Returns:
      a new scope

    • fork

      <U extends T> StructuredTaskScope.Subtask<U> fork(Callable<? extends U> task)
      Forks a subtask by starting a new thread in this scope to execute a value-returning method. The new thread executes the subtask concurrently with the current thread. The parameter to this method is a Callable. The new thread executes its call() method. The thread inherits the current thread's scoped value bindings that must match the bindings captured when the scope was opened.

      If the scope was opened with a function to produce the Configuration for the scope, and a ThreadFactory is set, then its newThread(Runnable) method is invoked to create the thread that will execute the subtask. RejectedExecutionException is thrown if the newThread(Runnable) method returns null. If the newThread(Runnable) method throws an exception or error then it is propagated by this method.

      If a ThreadFactory is not set, the fork(Callable) method creates an unnamed virtual thread to execute the subtask.

      This method returns a Subtask object as a handle to the forked subtask. If the scope is cancelled, the Subtask is returned in the UNAVAILABLE state without creating a thread. In some usages, the Subtask object will be used by the "main" task (the scope owner) to get the subtask's outcome (result or exception) after it has invoked join(). In other usages, the scope is created with a Joiner that produces the outcome for the main task to process after joining. A Joiner that produces a result reduces the need for bookkeeping and the need for the main task to retain references to Subtask objects for correlation purposes.

      After the scope owner has joined to obtain the scope's outcome, Subtask.get() and Subtask.exception() can be used to get the subtask's result or exception.

      This method may only be invoked by the scope owner.

      Joiner Implementers:

      If the scope was opened with a Joiner, the Joiner's onFork(Subtask) method is invoked with the newly created Subtask object in the UNAVAILABLE state and before a thread is created to execute it. If the onFork(Subtask) method returns true then the scope is cancelled if not already cancelled. This causes the fork(Callable) method to return the Subtask in the UNAVAILABLE state without creating a thread. If the onFork(Subtask) method returns false, and the scope is not cancelled, then the fork(Callable) method continues on to create a thread to execute the subtask. If the onFork(Subtask) method completes with an exception or error, then it is propagated by the fork(Callable) method without creating a thread to execute the subtask.

      If the subtask executes and completes (successfully or with an exception) before the scope is cancelled, then the thread invokes the Joiner's onComplete(Subtask) method with the subtask in the SUCCESS or FAILED state. If the onComplete(Subtask) method returns true, then the scope is cancelled, if not already cancelled. If the Joiner's onComplete(Subtask) method completes with an exception or error, then the thread executes the uncaught exception handler before the thread terminates.

      The subtask is not executed if the fork method is unable to create a thread to execute the subtask. In that case, the fork method invokes the Joiner's onComplete(Subtask) method directly with the subtask in the FAILED state. The Joiner can invoke the subtask's exception() method to get the exception or error encountered when attempting to create the thread.

      Type Parameters:
      U - the result type
      Parameters:
      task - the value-returning task for the thread to execute
      Returns:
      the subtask
      Throws:
      WrongThreadException - if the current thread is not the scope owner
      IllegalStateException - if the owner has already joined or the scope is closed
      StructureViolationException - if the current scoped value bindings are not the same as when the scope was created
      RejectedExecutionException - if the thread factory rejected creating a thread to execute the subtask
      See Also:

    • fork

      <U extends T> StructuredTaskScope.Subtask<U> fork(Runnable task)
      Forks a subtask by starting a new thread in this scope to execute a method that does not return a result.

      This method works exactly the same as fork(Callable) except that the parameter to this method is a Runnable. The new thread executes its run() method, and Subtask.get() returns null if the subtask completes successfully.

      Type Parameters:
      U - the result type
      Parameters:
      task - the task for the thread to execute
      Returns:
      the subtask
      Throws:
      WrongThreadException - if the current thread is not the scope owner
      IllegalStateException - if the owner has already joined or the scope is closed
      StructureViolationException - if the current scoped value bindings are not the same as when the scope was created
      RejectedExecutionException - if the thread factory rejected creating a thread to execute the subtask

    • join

      R join() throws R_X, InterruptedException
      Returns a result or throws an exception after waiting for all subtasks to complete or the scope to be cancelled.

      If the scope was opened with the open() or open(UnaryOperator) method then join() waits for all subtasks to succeed or any subtask to fail. It returns null if all subtasks complete successfully. It throws ExecutionException if any subtask fails, with the exception from any failed subtask as the cause. If a timeout is configured and the timeout expires before join() is invoked or completes, it throws ExecutionException with a CancelledByTimeoutException as the cause.

      If the scope was opened with a Joiner, the Joiner is invoked after waiting or cancellation to produce the outcome (result or exception). This includes the timeout case where a timeout is configured and the timeout expires before this method completes.

      This method may only be invoked by the scope owner. It may only be invoked once to get the result, exception or timeout outcome, unless the previous invocation resulted in an InterruptedException being thrown.

      Joiner Implementers:

      When all subtasks complete, or the scope is cancelled, this method invokes the Joiner's result() or timeout() method to produce the outcome (result or exception) for the join() method. The result() method is invoked for the "no timeout" case. The timeout() method is invoked for the timeout case. If the outcome for the timeout case is an exception then it is thrown with a CancelledByTimeoutException as the cause.

      API Note:
      When the outcome is an exception, the stack trace will be the stack trace of the call to the join() method. Its cause will typically be the exception thrown by a failed subtask.
      Returns:
      the result
      Throws:
      WrongThreadException - if the current thread is not the scope owner
      IllegalStateException - if already joined or this scope is closed
      R_X - when the outcome is an exception
      InterruptedException - if the current thread is interrupted while waiting. The current thread's interrupted status is cleared when this exception is thrown.
      See Also:

    • isCancelled

      boolean isCancelled()
      Returns true if this scope is cancelled or in the process of being cancelled, otherwise false.

      Cancelling the scope prevents new threads from being started in the scope and interrupts threads executing unfinished subtasks. It may take some time before the interrupted threads finish execution; this method may return true before all threads have been interrupted or before all threads have finished.

      API Note:
      A task with a lengthy "forking phase" (the code in the try block that forks subtasks before the join() method is invoked) may use this method to avoid doing work in cases where the scope is cancelled by the completion of a previously forked subtask or a timeout.
      Returns:
      true if this scope is cancelled or in the process of being cancelled, otherwise false

    • close

      void close()
      Closes this scope.

      If the join() method has not completed with an outcome, this method first cancels the scope, if not already cancelled. This interrupts the threads executing unfinished subtasks. This method then waits for all threads to finish. If interrupted while waiting, it continues to wait until the threads finish, then completes with the interrupted status set. If the scope is already closed then this method has no effect.

      This method may only be invoked by the scope owner.

      A StructuredTaskScope is intended to be used in a structured manner. If this method is called to close a scope before nested scopes are closed then it closes the underlying construct of each nested scope (in the reverse order that they were created in), closes this scope, and then throws StructureViolationException. Similarly, if this method is called to close a scope while executing with scoped value bindings, and the scope was created before the scoped values were bound, then StructureViolationException is thrown after closing the scope. If a thread terminates without first closing scopes that it owns then termination will cause the underlying construct of each of its open scopes to be closed. Closing is performed in the reverse order that the scopes were created in. Thread termination may therefore be delayed when the scope owner has to wait for threads forked in these scopes to finish.

      Specified by:
      close in interface AutoCloseable
      Throws:
      IllegalStateException - thrown after closing the scope if the scope owner did not attempt to join after forking
      WrongThreadException - if the current thread is not the scope owner
      StructureViolationException - if a structure violation was detected