< prev index next > src/java.base/share/classes/java/util/concurrent/StructuredTaskScope.java
Print this page
* questions.
*/
package java.util.concurrent;
import java.time.Duration;
- import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Stream;
import jdk.internal.javac.PreviewFeature;
/**
* An API for <em>structured concurrency</em>. {@code StructuredTaskScope} supports cases
* questions.
*/
package java.util.concurrent;
import java.time.Duration;
import java.util.function.Predicate;
import java.util.function.Supplier;
+ import java.util.function.UnaryOperator;
import java.util.stream.Stream;
import jdk.internal.javac.PreviewFeature;
/**
* An API for <em>structured concurrency</em>. {@code StructuredTaskScope} supports cases
* {@link Thread} to run the subtask. The thread executing the task does not continue
* beyond the {@code close} method until all threads started to execute subtasks have finished.
* To ensure correct usage, the {@code fork}, {@code join} and {@code close} methods may
* only be invoked by the <em>owner thread</em> (the thread that opened the {@code
* StructuredTaskScope}), the {@code fork} method may not be called after {@code join},
! * the {@code join} method may only be invoked once, and the {@code close} method throws
! * an exception after closing if the owner did not invoke the {@code join} method after
! * forking subtasks.
*
* <p> As a first example, consider a task that splits into two subtasks to concurrently
* fetch resources from two URL locations "left" and "right". Both subtasks may complete
* successfully, one subtask may succeed and the other may fail, or both subtasks may
* fail. The task in this example is interested in the successful result from both
* {@link Thread} to run the subtask. The thread executing the task does not continue
* beyond the {@code close} method until all threads started to execute subtasks have finished.
* To ensure correct usage, the {@code fork}, {@code join} and {@code close} methods may
* only be invoked by the <em>owner thread</em> (the thread that opened the {@code
* StructuredTaskScope}), the {@code fork} method may not be called after {@code join},
! * the {@code join} method may only be invoked once to get the outcome, and the
! * {@code close} method throws an exception after closing if the owner did not invoke the
! * {@code join} method after forking subtasks.
*
* <p> As a first example, consider a task that splits into two subtasks to concurrently
* fetch resources from two URL locations "left" and "right". Both subtasks may complete
* successfully, one subtask may succeed and the other may fail, or both subtasks may
* fail. The task in this example is interested in the successful result from both
* may be more appropriate to handle this in the subtask itself rather than the subtask
* failing and the scope owner handling the exception.
*
* <h2>Configuration</h2>
*
- *
* A {@code StructuredTaskScope} is opened with {@linkplain Configuration configuration}
! * that consists of a {@link ThreadFactory} to create threads, an optional name for
! * monitoring and management purposes, and an optional timeout.
*
* <p> The {@link #open()} and {@link #open(Joiner)} methods create a {@code StructuredTaskScope}
* with the <a id="DefaultConfiguration"> <em>default configuration</em></a>. The default
! * configuration has a {@code ThreadFactory} that creates unnamed
! * <a href="{@docRoot}/java.base/java/lang/Thread.html#virtual-threads">virtual threads</a>,
! * is unnamed for monitoring and management purposes, and has no timeout.
! *
! * <p> The 2-arg {@link #open(Joiner, Function) open} method can be used to create a
! * {@code StructuredTaskScope} that uses a different {@code ThreadFactory}, has a name for
! * the purposes of monitoring and management, or has a timeout that cancels the scope if
! * the timeout expires before or while waiting for subtasks to complete. The {@code open}
! * method is called with a {@linkplain Function function} that is applied to the default
- * configuration and returns a {@link Configuration Configuration} for the
* {@code StructuredTaskScope} under construction.
*
* <p> The following example opens a new {@code StructuredTaskScope} with a {@code
* ThreadFactory} that creates virtual threads {@linkplain Thread#setName(String) named}
* "duke-0", "duke-1" ...
* may be more appropriate to handle this in the subtask itself rather than the subtask
* failing and the scope owner handling the exception.
*
* <h2>Configuration</h2>
*
* A {@code StructuredTaskScope} is opened with {@linkplain Configuration configuration}
! * that consists of a {@link ThreadFactory} to create threads, an optional name for the
! * scope, and an optional timeout. The name is intended for monitoring and management
+ * purposes.
*
* <p> The {@link #open()} and {@link #open(Joiner)} methods create a {@code StructuredTaskScope}
* with the <a id="DefaultConfiguration"> <em>default configuration</em></a>. The default
! * configuration has a {@code ThreadFactory} that creates unnamed {@linkplain
! * Thread##virtual-threads virtual threads}, does not name the scope, and has no timeout.
! *
! * <p> The 2-arg {@link #open(Joiner, UnaryOperator) open} method can be used to create a
! * {@code StructuredTaskScope} that uses a different {@code ThreadFactory}, is named for
! * monitoring and management purposes, or has a timeout that cancels the scope if the
! * timeout expires before or while waiting for subtasks to complete. The {@code open}
! * method is called with a {@linkplain UnaryOperator operator} that is applied to the
! * default configuration and returns a {@link Configuration Configuration} for the
* {@code StructuredTaskScope} under construction.
*
* <p> The following example opens a new {@code StructuredTaskScope} with a {@code
* ThreadFactory} that creates virtual threads {@linkplain Thread#setName(String) named}
* "duke-0", "duke-1" ...
* }
*}
*
* <p> A second example sets a timeout, represented by a {@link Duration}. The timeout
* starts when the new scope is opened. If the timeout expires before the {@code join}
! * method has completed then the scope is <a href="#Cancallation">cancelled</a>. This
! * interrupts the threads executing the two subtasks and causes the {@link #join() join}
! * method to throw {@link TimeoutException}.
* {@snippet lang=java :
* Duration timeout = Duration.ofSeconds(10);
*
* // @link substring="allSuccessfulOrThrow" target="Joiner#allSuccessfulOrThrow()" :
* try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(),
* }
*}
*
* <p> A second example sets a timeout, represented by a {@link Duration}. The timeout
* starts when the new scope is opened. If the timeout expires before the {@code join}
! * method has completed then the scope is {@linkplain ##Cancallation cancelled} (this
! * interrupts the threads executing the two subtasks), and the {@code join} method
! * throws {@link TimeoutException TimeoutException}.
* {@snippet lang=java :
* Duration timeout = Duration.ofSeconds(10);
*
* // @link substring="allSuccessfulOrThrow" target="Joiner#allSuccessfulOrThrow()" :
* try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow(),
* }
*
* });
* }
*
! * <p> A scoped value inherited into a subtask may be
! * <a href="{@docRoot}/java.base/java/lang/ScopedValue.html#rebind">rebound</a> 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 {@code ScopedValue} reverts to its previous
- * value, the value inherited from the thread executing the task.
*
* <p> A subtask may execute code that itself opens a new {@code StructuredTaskScope}.
* A task executing in thread T1 opens a {@code 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
* }
*
* });
* }
*
! * <p> A scoped value inherited into a subtask may be {@linkplain ScopedValue##rebind
! * 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 {@code ScopedValue} reverts
! * to its previous value, the value inherited from the thread executing the task.
*
* <p> A subtask may execute code that itself opens a new {@code StructuredTaskScope}.
* A task executing in thread T1 opens a {@code 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
* include (or may be the same) as the bindings that were inherited from T1. In effect,
* scoped values are inherited into a tree of subtasks, not just one level of subtask.
*
* <h2>Memory consistency effects</h2>
*
! * <p> Actions in the owner thread of a {@code StructuredTaskScope} prior to
! * {@linkplain #fork forking} of a subtask
! * <a href="{@docRoot}/java.base/java/util/concurrent/package-summary.html#MemoryVisibility">
- * <i>happen-before</i></a> any actions taken by that subtask, which in turn
* <i>happen-before</i> the subtask result is {@linkplain Subtask#get() retrieved}.
*
* <h2>General exceptions</h2>
*
* <p> Unless otherwise specified, passing a {@code null} argument to a method in this
* include (or may be the same) as the bindings that were inherited from T1. In effect,
* scoped values are inherited into a tree of subtasks, not just one level of subtask.
*
* <h2>Memory consistency effects</h2>
*
! * <p> Actions in the owner thread of a {@code StructuredTaskScope} prior to {@linkplain
! * #fork forking} of a subtask {@linkplain java.util.concurrent##MemoryVisibility
! * <i>happen-before</i>} any actions taken by that subtask, which in turn
* <i>happen-before</i> the subtask result is {@linkplain Subtask#get() retrieved}.
*
* <h2>General exceptions</h2>
*
* <p> Unless otherwise specified, passing a {@code null} argument to a method in this
* was forked with {@link #fork(Callable) fork(Callable)} then the result from the
* {@link Callable#call() call} method is returned. If the subtask was forked with
* {@link #fork(Runnable) fork(Runnable)} then {@code null} is returned.
*
* <p> Code executing in the scope owner thread can use this method to get the
! * result of a successful subtask only after it has {@linkplain #join() joined}.
*
* <p> Code executing in the {@code Joiner} {@link Joiner#onComplete(Subtask)
* onComplete} method should test that the {@linkplain #state() subtask state} is
* {@link State#SUCCESS SUCCESS} before using this method to get the result.
*
* @return the possibly-null result
! * @throws IllegalStateException if the subtask has not completed, did not complete
! * successfully, or the current thread is the scope owner invoking this
! * method before {@linkplain #join() joining}
* @see State#SUCCESS
*/
T get();
/**
* was forked with {@link #fork(Callable) fork(Callable)} then the result from the
* {@link Callable#call() call} method is returned. If the subtask was forked with
* {@link #fork(Runnable) fork(Runnable)} then {@code null} is returned.
*
* <p> Code executing in the scope owner thread can use this method to get the
! * result of a successful subtask after it has {@linkplain #join() joined}.
*
* <p> Code executing in the {@code Joiner} {@link Joiner#onComplete(Subtask)
* onComplete} method should test that the {@linkplain #state() subtask state} is
* {@link State#SUCCESS SUCCESS} before using this method to get the result.
*
+ * <p> This method may be invoked by any thread after the scope owner has joined.
+ * The only case where this method can be used to get the result before the scope
+ * owner has joined is when called from the {@code onComplete(Subtask)} method.
+ *
* @return the possibly-null result
! * @throws IllegalStateException if the subtask has not completed or did not
! * complete successfully, or this method if invoked outside the context of the
! * {@code onComplete(Subtask)} method before the owner thread has joined
* @see State#SUCCESS
*/
T get();
/**
* exception or error thrown by the {@link Callable#call() call} method is returned.
* If the subtask was forked with {@link #fork(Runnable) fork(Runnable)} then the
* exception or error thrown by the {@link Runnable#run() run} method is returned.
*
* <p> Code executing in the scope owner thread can use this method to get the
! * exception thrown by a failed subtask only after it has {@linkplain #join() joined}.
*
* <p> Code executing in a {@code Joiner} {@link Joiner#onComplete(Subtask)
* onComplete} method should test that the {@linkplain #state() subtask state} is
* {@link State#FAILED FAILED} before using this method to get the exception.
*
! * @throws IllegalStateException if the subtask has not completed, completed with
! * a result, or the current thread is the scope owner invoking this method
! * before {@linkplain #join() joining}
* @see State#FAILED
*/
Throwable exception();
}
* exception or error thrown by the {@link Callable#call() call} method is returned.
* If the subtask was forked with {@link #fork(Runnable) fork(Runnable)} then the
* exception or error thrown by the {@link Runnable#run() run} method is returned.
*
* <p> Code executing in the scope owner thread can use this method to get the
! * exception thrown by a failed subtask after it has {@linkplain #join() joined}.
*
* <p> Code executing in a {@code Joiner} {@link Joiner#onComplete(Subtask)
* onComplete} method should test that the {@linkplain #state() subtask state} is
* {@link State#FAILED FAILED} before using this method to get the exception.
*
! * <p> This method may be invoked by any thread after the scope owner has joined.
! * The only case where this method can be used to get the exception before the scope
! * owner has joined is when called from the {@code onComplete(Subtask)} method.
+ *
+ * @throws IllegalStateException if the subtask has not completed or completed
+ * with a result, or this method if invoked outside the context of the {@code
+ * onComplete(Subtask)} method before the owner thread has joined
* @see State#FAILED
*/
Throwable exception();
}
* join} to return. It causes {@code join} to throw if all subtasks fail.
* <li> {@link #awaitAllSuccessfulOrThrow() awaitAllSuccessfulOrThrow()} creates a
* {@code Joiner} that waits for all successful subtasks. It cancels the scope and
* causes {@code join} to throw if any subtask fails.
* <li> {@link #awaitAll() awaitAll()} creates a {@code Joiner} that waits for all
! * subtasks. It does not cancel the scope or cause {@code join} to throw.
* </ul>
*
* <p> In addition to the methods to create {@code Joiner} objects for common cases,
* the {@link #allUntil(Predicate) allUntil(Predicate)} method is defined to create a
* {@code Joiner} that yields a stream of all subtasks. It is created with a {@link
* join} to return. It causes {@code join} to throw if all subtasks fail.
* <li> {@link #awaitAllSuccessfulOrThrow() awaitAllSuccessfulOrThrow()} creates a
* {@code Joiner} that waits for all successful subtasks. It cancels the scope and
* causes {@code join} to throw if any subtask fails.
* <li> {@link #awaitAll() awaitAll()} creates a {@code Joiner} that waits for all
! * subtasks to complete. It does not cancel the scope or cause {@code join} to throw.
* </ul>
*
* <p> In addition to the methods to create {@code Joiner} objects for common cases,
* the {@link #allUntil(Predicate) allUntil(Predicate)} method is defined to create a
* {@code Joiner} that yields a stream of all subtasks. It is created with a {@link
* scope based on some condition.
*
* <p> More advanced policies can be developed by implementing the {@code Joiner}
* interface. The {@link #onFork(Subtask)} method is invoked when subtasks are forked.
* The {@link #onComplete(Subtask)} method is invoked when subtasks complete with a
! * result or exception. These methods return a {@code boolean} to indicate if scope
* should be cancelled. These methods can be used to collect subtasks, results, or
* exceptions, and control when to cancel the scope. The {@link #result()} method
* must be implemented to produce the result (or exception) for the {@code join}
* method.
*
* <p> Unless otherwise specified, passing a {@code null} argument to a method
* in this class will cause a {@link NullPointerException} to be thrown.
*
* @implSpec Implementations of this interface must be thread safe. The {@link
* #onComplete(Subtask)} method defined by this interface may be invoked by several
! * threads concurrently.
*
* @apiNote It is very important that a new {@code Joiner} object is created for each
* {@code StructuredTaskScope}. {@code Joiner} objects should never be shared with
! * different scopes or re-used after a task is closed.
*
* <p> Designing a {@code Joiner} should take into account the code at the use-site
* where the results from the {@link StructuredTaskScope#join() join} method are
* processed. It should be clear what the {@code Joiner} does vs. the application
* code at the use-site. In general, the {@code Joiner} implementation is not the
* scope based on some condition.
*
* <p> More advanced policies can be developed by implementing the {@code Joiner}
* interface. The {@link #onFork(Subtask)} method is invoked when subtasks are forked.
* The {@link #onComplete(Subtask)} method is invoked when subtasks complete with a
! * result or exception. These methods return a {@code boolean} to indicate if the scope
* should be cancelled. These methods can be used to collect subtasks, results, or
* exceptions, and control when to cancel the scope. The {@link #result()} method
* must be implemented to produce the result (or exception) for the {@code join}
* method.
*
+ * <p> If a {@code StructuredTaskScope} is opened with a {@linkplain
+ * Configuration#withTimeout(Duration) timeout}, and the timeout expires before or
+ * while waiting in {@link StructuredTaskScope#join() join()}, then the scope is
+ * {@linkplain StructuredTaskScope##Cancallation cancelled}, and the {@code Joiners}'s
+ * {@link #onTimeout()} method is invoked to notify the {@code Joiner} and optionally
+ * throw {@link TimeoutException TimeoutException}. If the {@code onTimeout()} method
+ * does not throw then the {@code join()} method will invoke the {@link #result()}
+ * method to produce a result. This result may be based on the outcome of subtasks
+ * that completed before the timeout expired.
+ *
* <p> Unless otherwise specified, passing a {@code null} argument to a method
* in this class will cause a {@link NullPointerException} to be thrown.
*
* @implSpec Implementations of this interface must be thread safe. The {@link
* #onComplete(Subtask)} method defined by this interface may be invoked by several
! * threads concurrently. The {@link #onTimeout()} method may be invoked at around
+ * the same time that subtasks complete.
*
* @apiNote It is very important that a new {@code Joiner} object is created for each
* {@code StructuredTaskScope}. {@code Joiner} objects should never be shared with
! * different scopes or re-used after a scope is closed.
*
* <p> Designing a {@code Joiner} should take into account the code at the use-site
* where the results from the {@link StructuredTaskScope#join() join} method are
* processed. It should be clear what the {@code Joiner} does vs. the application
* code at the use-site. In general, the {@code Joiner} implementation is not the
* @param <R> the result type of the scope
* @since 25
* @see #open(Joiner)
*/
@PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
- @FunctionalInterface
interface Joiner<T, R> {
/**
* Invoked by {@link #fork(Callable) fork(Callable)} and {@link #fork(Runnable)
! * fork(Runnable)} when forking a subtask. The method is invoked from the task
! * owner thread. The method is invoked before a thread is created to run the
- * subtask.
*
* @implSpec The default implementation throws {@code NullPointerException} if the
* subtask is {@code null}. It throws {@code IllegalArgumentException} if the
* subtask is not in the {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state, it
* otherwise returns {@code false}.
* @param <R> the result type of the scope
* @since 25
* @see #open(Joiner)
*/
@PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
interface Joiner<T, R> {
/**
* Invoked by {@link #fork(Callable) fork(Callable)} and {@link #fork(Runnable)
! * fork(Runnable)} when forking a subtask. The method is invoked before a thread
! * is created to run the subtask.
*
* @implSpec The default implementation throws {@code NullPointerException} if the
* subtask is {@code null}. It throws {@code IllegalArgumentException} if the
* subtask is not in the {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state, it
* otherwise returns {@code false}.
throw new IllegalArgumentException("Subtask has not completed");
}
return false;
}
+ /**
+ * Invoked by the {@link #join() join()} method if the scope was opened with a
+ * timeout, and the timeout expires before or while waiting in the {@code join}
+ * method.
+ *
+ * @implSpec The default implementation throws {@link TimeoutException TimeoutException}.
+ *
+ * @apiNote This method is intended for {@code Joiner} implementations that do not
+ * throw {@link TimeoutException TimeoutException}, or require a notification when
+ * the timeout expires before or while waiting in {@code join}.
+ *
+ * <p> This method is invoked by the {@code join} method. It should not be
+ * invoked directly.
+ *
+ * @throws TimeoutException for {@code join} to throw
+ * @since 26
+ */
+ default void onTimeout() {
+ throw new TimeoutException();
+ }
+
/**
* Invoked by the {@link #join() join()} method to produce the result (or exception)
* after waiting for all subtasks to complete or the scope cancelled. The result
* from this method is returned by the {@code join} method. If this method throws,
* then {@code join} throws {@link FailedException} with the exception thrown by
R result() throws Throwable;
/**
* {@return a new Joiner object that yields a stream of all subtasks when all
* subtasks complete successfully}
! * The {@code Joiner} <a href="StructuredTaskScope.html#Cancallation">cancels</a>
* the scope and causes {@code join} to throw if any subtask fails.
*
! * <p> If all subtasks complete successfully, the joiner's {@link Joiner#result()}
! * method returns a stream of all subtasks in the order that they were forked.
! * If any subtask failed then the {@code result} method throws the exception from
! * the first subtask to fail.
*
* @apiNote Joiners returned by this method are suited to cases where all subtasks
* return a result of the same type. Joiners returned by {@link
* #awaitAllSuccessfulOrThrow()} are suited to cases where the subtasks return
* results of different types.
R result() throws Throwable;
/**
* {@return a new Joiner object that yields a stream of all subtasks when all
* subtasks complete successfully}
! * The {@code Joiner} {@linkplain StructuredTaskScope##Cancallation cancels}
* the scope and causes {@code join} to throw if any subtask fails.
*
! * <p> If all subtasks complete successfully then the joiner's {@link
! * Joiner#result()} method returns a stream of all subtasks, in the order that they
! * were forked, for the {@link StructuredTaskScope#join() join()} to return. If
! * the scope was opened with a {@linkplain Configuration#withTimeout(Duration)
+ * timeout}, and the timeout expires before or while waiting for all subtasks to
+ * complete, then the {@code join} method throws {@code TimeoutException}.
*
* @apiNote Joiners returned by this method are suited to cases where all subtasks
* return a result of the same type. Joiners returned by {@link
* #awaitAllSuccessfulOrThrow()} are suited to cases where the subtasks return
* results of different types.
/**
* {@return a new Joiner object that yields the result of any subtask that
* completed successfully}
* The {@code Joiner} causes {@code join} to throw if all subtasks fail.
*
! * <p> The joiner's {@link Joiner#result()} method returns the result of a subtask
! * that completed successfully. If all subtasks fail then the {@code result} method
! * throws the exception from one of the failed subtasks. The {@code result} method
! * throws {@code NoSuchElementException} if no subtasks were forked.
*
* @param <T> the result type of subtasks
*/
static <T> Joiner<T, T> anySuccessfulResultOrThrow() {
return new Joiners.AnySuccessful<>();
}
/**
* {@return a new Joiner object that waits for subtasks to complete successfully}
! * The {@code Joiner} <a href="StructuredTaskScope.html#Cancallation">cancels</a>
* the scope and causes {@code join} to throw if any subtask fails.
*
* <p> The joiner's {@link Joiner#result() result} method returns {@code null}
* if all subtasks complete successfully, or throws the exception from the first
! * subtask to fail.
*
* @apiNote Joiners returned by this method are suited to cases where subtasks
* return results of different types. Joiners returned by {@link #allSuccessfulOrThrow()}
* are suited to cases where the subtasks return a result of the same type.
*
/**
* {@return a new Joiner object that yields the result of any subtask that
* completed successfully}
* The {@code Joiner} causes {@code join} to throw if all subtasks fail.
*
! * <p> The joiner's {@link Joiner#result()} method returns the result of a subtask,
! * that completed successfully, for the {@link StructuredTaskScope#join() join()}
! * to return. If all subtasks fail then the {@code result} method throws the
! * exception from one of the failed subtasks. The {@code result} method throws
+ * {@code NoSuchElementException} if no subtasks were forked. If the scope was
+ * opened with a {@linkplain Configuration#withTimeout(Duration) timeout}, and
+ * the timeout expires before or while waiting for any subtask to complete
+ * successfully, then the {@code join} method throws {@code TimeoutException}.
*
* @param <T> the result type of subtasks
*/
static <T> Joiner<T, T> anySuccessfulResultOrThrow() {
return new Joiners.AnySuccessful<>();
}
/**
* {@return a new Joiner object that waits for subtasks to complete successfully}
! * The {@code Joiner} {@linkplain StructuredTaskScope##Cancallation cancels}
* the scope and causes {@code join} to throw if any subtask fails.
*
* <p> The joiner's {@link Joiner#result() result} method returns {@code null}
* if all subtasks complete successfully, or throws the exception from the first
! * subtask to fail. If the scope was opened with a {@linkplain
+ * Configuration#withTimeout(Duration) timeout}, and the timeout expires before or
+ * while waiting for all subtasks to complete, then the {@code join} method throws
+ * {@code TimeoutException}.
*
* @apiNote Joiners returned by this method are suited to cases where subtasks
* return results of different types. Joiners returned by {@link #allSuccessfulOrThrow()}
* are suited to cases where the subtasks return a result of the same type.
*
/**
* {@return a new Joiner object that waits for all subtasks to complete}
* The {@code Joiner} does not cancel the scope if a subtask fails.
*
* <p> The joiner's {@link Joiner#result() result} method returns {@code null}.
+ * If the scope was opened with a {@linkplain Configuration#withTimeout(Duration)
+ * timeout}, and the timeout expires before or while waiting for all subtasks to
+ * complete, then the {@code join} method throws {@code TimeoutException}.
*
* @apiNote This Joiner is useful for cases where subtasks make use of
* <em>side-effects</em> rather than return results or fail with exceptions.
* The {@link #fork(Runnable) fork(Runnable)} method can be used to fork subtasks
* that do not return a result.
/**
* {@return a new Joiner object that yields a stream of all subtasks when all
* subtasks complete or a predicate returns {@code true} to cancel the scope}
*
! * <p> The joiner's {@link Joiner#onComplete(Subtask)} method invokes the
! * predicate's {@link Predicate#test(Object) test} method with the subtask that
! * completed successfully or failed with an exception. If the {@code test} method
! * returns {@code true} then <a href="StructuredTaskScope.html#Cancallation">
! * the scope is cancelled</a>. The {@code test} method must be thread safe as it
* may be invoked concurrently from several threads. If the {@code test} method
* completes with an exception or error, then the thread that executed the subtask
* invokes the {@linkplain Thread.UncaughtExceptionHandler uncaught exception handler}
* with the exception or error before the thread terminates.
*
/**
* {@return a new Joiner object that yields a stream of all subtasks when all
* subtasks complete or a predicate returns {@code true} to cancel the scope}
*
! * <p> The joiner's {@link #onComplete(Subtask)} method invokes the predicate's
! * {@link Predicate#test(Object) test} method with the subtask that completed
! * successfully or failed with an exception. If the {@code test} method
! * returns {@code true} then {@linkplain StructuredTaskScope##Cancallation
! * the scope is cancelled}. The {@code test} method must be thread safe as it
* may be invoked concurrently from several threads. If the {@code test} method
* completes with an exception or error, then the thread that executed the subtask
* invokes the {@linkplain Thread.UncaughtExceptionHandler uncaught exception handler}
* with the exception or error before the thread terminates.
*
* in fork order. The stream may contain subtasks that have completed
* (in {@link Subtask.State#SUCCESS SUCCESS} or {@link Subtask.State#FAILED FAILED}
* state) or subtasks in the {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state
* if the scope was cancelled before all subtasks were forked or completed.
*
* <p> The following example uses this method to create a {@code Joiner} that
! * <a href="StructuredTaskScope.html#Cancallation">cancels</a> the scope when
! * two or more subtasks fail.
* {@snippet lang=java :
* class CancelAfterTwoFailures<T> implements Predicate<Subtask<? extends T>> {
* private final AtomicInteger failedCount = new AtomicInteger();
* @Override
* public boolean test(Subtask<? extends T> subtask) {
* in fork order. The stream may contain subtasks that have completed
* (in {@link Subtask.State#SUCCESS SUCCESS} or {@link Subtask.State#FAILED FAILED}
* state) or subtasks in the {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state
* if the scope was cancelled before all subtasks were forked or completed.
*
+ * <p> The joiner's {@link #onTimeout()} method does nothing. If configured with
+ * a {@linkplain Configuration#withTimeout(Duration) timeout}, and the timeout
+ * expires before or while waiting in {@link StructuredTaskScope#join() join},
+ * then the {@link #result()} method returns the stream of all subtasks.
+ * Subtasks that did not complete before the timeout expired will be in the
+ * {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state.
+ *
* <p> The following example uses this method to create a {@code Joiner} that
! * {@linkplain StructuredTaskScope##Cancallation cancels} the scope when two or
! * more subtasks fail.
* {@snippet lang=java :
* class CancelAfterTwoFailures<T> implements Predicate<Subtask<? extends T>> {
* private final AtomicInteger failedCount = new AtomicInteger();
* @Override
* public boolean test(Subtask<? extends T> subtask) {
* return scope.join().toList();
* }
* }
* }
*
+ * <p> The following example uses {@code allUntil} to get the results of all
+ * subtasks that complete successfully within a timeout period.
+ * {@snippet lang=java :
+ * <T> List<T> invokeAll(Collection<Callable<T>> tasks, Duration timeout) throws InterruptedException {
+ * try (var scope = StructuredTaskScope.open(Joiner.<T>allUntil(_ -> false), cf -> cf.withTimeout(timeout))) {
+ * tasks.forEach(scope::fork);
+ * return scope.join()
+ * .filter(s -> s.state() == Subtask.State.SUCCESS)
+ * .map(Subtask::get)
+ * .toList();
+ * }
+ * }
+ * }
+ *
* @param isDone the predicate to evaluate completed subtasks
* @param <T> the result type of subtasks
*/
static <T> Joiner<T, Stream<Subtask<T>>> allUntil(Predicate<Subtask<? extends T>> isDone) {
return new Joiners.AllSubtasks<>(isDone);
/**
* Represents the configuration for a {@code StructuredTaskScope}.
*
* <p> The configuration for a {@code StructuredTaskScope} consists of a {@link
! * ThreadFactory} to create threads, an optional name for the purposes of monitoring
! * and management, and an optional timeout.
- *
- * <p> Creating a {@code StructuredTaskScope} with {@link #open()} or {@link #open(Joiner)}
- * uses the <a href="StructuredTaskScope.html#DefaultConfiguration">default
- * configuration</a>. The default configuration consists of a thread factory that
- * creates unnamed <a href="{@docRoot}/java.base/java/lang/Thread.html#virtual-threads">
- * virtual threads</a>, no name for monitoring and management purposes, and no timeout.
*
! * <p> Creating a {@code StructuredTaskScope} with its 2-arg {@link #open(Joiner, Function)
! * open} method allows a different configuration to be used. The function specified
* to the {@code open} method is applied to the default configuration and returns the
! * configuration for the {@code StructuredTaskScope} under construction. The function
* can use the {@code with-} prefixed methods defined here to specify the components
* of the configuration to use.
*
* <p> Unless otherwise specified, passing a {@code null} argument to a method
* in this class will cause a {@link NullPointerException} to be thrown.
/**
* Represents the configuration for a {@code StructuredTaskScope}.
*
* <p> The configuration for a {@code StructuredTaskScope} consists of a {@link
! * ThreadFactory} to create threads, an optional name for the scope, and an optional
! * timeout. The name is intended for monitoring and management purposes.
*
! * <p> Creating a {@code StructuredTaskScope} with its 2-arg {@link #open(Joiner, UnaryOperator)
! * open} method allows a different configuration to be used. The operator specified
* to the {@code open} method is applied to the default configuration and returns the
! * configuration for the {@code StructuredTaskScope} under construction. The operator
* can use the {@code with-} prefixed methods defined here to specify the components
* of the configuration to use.
*
* <p> Unless otherwise specified, passing a {@code null} argument to a method
* in this class will cause a {@link NullPointerException} to be thrown.
* {@return a new {@code Configuration} object with the given thread factory}
* The other components are the same as this object. The thread factory is used by
* a scope to create threads when {@linkplain #fork(Callable) forking} subtasks.
* @param threadFactory the thread factory
*
! * @apiNote The thread factory will typically create
! * <a href="{@docRoot}/java.base/java/lang/Thread.html#virtual-threads">virtual threads</a>,
! * maybe with names for monitoring purposes, an {@linkplain Thread.UncaughtExceptionHandler
! * uncaught exception handler}, or other properties configured.
*
* @see #fork(Callable)
*/
Configuration withThreadFactory(ThreadFactory threadFactory);
/**
! * {@return a new {@code Configuration} object with the given name}
* The other components are the same as this object. A scope is optionally
* named for the purposes of monitoring and management.
* @param name the name
*/
Configuration withName(String name);
* {@return a new {@code Configuration} object with the given thread factory}
* The other components are the same as this object. The thread factory is used by
* a scope to create threads when {@linkplain #fork(Callable) forking} subtasks.
* @param threadFactory the thread factory
*
! * @apiNote The thread factory will typically create {@linkplain Thread##virtual-threads
! * virtual threads}, maybe with names for monitoring purposes, an {@linkplain
! * Thread.UncaughtExceptionHandler uncaught exception handler}, or other properties
! * configured.
*
* @see #fork(Callable)
*/
Configuration withThreadFactory(ThreadFactory threadFactory);
/**
! * {@return a new {@code Configuration} object with the given scope name}
* The other components are the same as this object. A scope is optionally
* named for the purposes of monitoring and management.
* @param name the name
*/
Configuration withName(String name);
* @apiNote Applications using deadlines, expressed as an {@link java.time.Instant},
* can use {@link Duration#between Duration.between(Instant.now(), deadline)} to
* compute the timeout for this method.
*
* @see #join()
+ * @see Joiner#onTimeout()
*/
Configuration withTimeout(Duration timeout);
}
/**
super(cause);
}
}
/**
! * Exception thrown by {@link #join()} if the scope was created with a timeout and
! * the timeout expired before or while waiting in {@code join}.
*
* @since 25
* @see Configuration#withTimeout(Duration)
*/
@PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
final class TimeoutException extends RuntimeException {
@java.io.Serial
static final long serialVersionUID = 705788143955048766L;
super(cause);
}
}
/**
! * Exception thrown by {@link #join()} if the scope was opened with a timeout,
! * the timeout expired before or while waiting in {@code join}, and the {@link
+ * Joiner#onTimeout() Joiner.onTimeout} method throws this exception.
*
* @since 25
* @see Configuration#withTimeout(Duration)
+ * @see Joiner#onTimeout()
*/
@PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
final class TimeoutException extends RuntimeException {
@java.io.Serial
static final long serialVersionUID = 705788143955048766L;
TimeoutException() { }
}
/**
* Opens a new {@code StructuredTaskScope} to use the given {@code Joiner} object and
! * with configuration that is the result of applying the given function to the
! * <a href="#DefaultConfiguration">default configuration</a>.
*
! * <p> The {@code configFunction} is called with the default configuration and returns
! * the configuration for the new scope. The function may, for example, set the
* {@linkplain Configuration#withThreadFactory(ThreadFactory) ThreadFactory} or set a
! * {@linkplain Configuration#withTimeout(Duration) timeout}. If the function completes
! * with an exception or error then it is propagated by this method. If the function
* returns {@code null} then {@code NullPointerException} is thrown.
*
* <p> If a {@code ThreadFactory} is set then its {@link ThreadFactory#newThread(Runnable)
* newThread} method will be called to create threads when {@linkplain #fork(Callable)
* forking} subtasks in this scope. If a {@code ThreadFactory} is not set then
* forking subtasks will create an unnamed virtual thread for each subtask.
*
* <p> If a {@linkplain Configuration#withTimeout(Duration) timeout} is set then it
* starts when the scope is opened. If the timeout expires before the scope has
! * {@linkplain #join() joined} then the scope is <a href="#Cancallation">cancelled</a>
! * and the {@code join} method throws {@link TimeoutException}.
*
* <p> The new scope is owned by the current thread. Only code executing in this
* thread can {@linkplain #fork(Callable) fork}, {@linkplain #join() join}, or
* {@linkplain #close close} the scope.
*
* <p> Construction captures the current thread's {@linkplain ScopedValue scoped
* value} bindings for inheritance by threads started in the scope.
*
* @param joiner the joiner
! * @param configFunction a function to produce the configuration
* @return a new scope
* @param <T> the result type of subtasks executed in the scope
* @param <R> the result type of the scope
! * @since 25
*/
static <T, R> StructuredTaskScope<T, R> open(Joiner<? super T, ? extends R> joiner,
! Function<Configuration, Configuration> configFunction) {
! return StructuredTaskScopeImpl.open(joiner, configFunction);
}
/**
* Opens a new {@code StructuredTaskScope}to use the given {@code Joiner} object. The
! * scope is created with the <a href="#DefaultConfiguration">default configuration</a>.
* The default configuration has a {@code ThreadFactory} that creates unnamed
! * <a href="{@docRoot}/java.base/java/lang/Thread.html#virtual-threads">virtual threads</a>,
! * is unnamed for monitoring and management purposes, and has no timeout.
*
* @implSpec
* This factory method is equivalent to invoking the 2-arg open method with the given
! * joiner and the {@linkplain Function#identity() identity function}.
*
* @param joiner the joiner
* @return a new scope
* @param <T> the result type of subtasks executed in the scope
* @param <R> the result type of the scope
* @since 25
*/
static <T, R> StructuredTaskScope<T, R> open(Joiner<? super T, ? extends R> joiner) {
! return open(joiner, Function.identity());
}
/**
* Opens a new {@code StructuredTaskScope} that can be used to fork subtasks that return
* results of any type. The scope's {@link #join()} method waits for all subtasks to
TimeoutException() { }
}
/**
* Opens a new {@code StructuredTaskScope} to use the given {@code Joiner} object and
! * with configuration that is the result of applying the given operator to the
! * {@linkplain ##DefaultConfiguration default configuration}.
*
! * <p> The {@code configOperator} is called with the default configuration and returns
! * the configuration for the new scope. The operator may, for example, set the
* {@linkplain Configuration#withThreadFactory(ThreadFactory) ThreadFactory} or set a
! * {@linkplain Configuration#withTimeout(Duration) timeout}. If the operator completes
! * with an exception or error then it is propagated by this method. If the operator
* returns {@code null} then {@code NullPointerException} is thrown.
*
* <p> If a {@code ThreadFactory} is set then its {@link ThreadFactory#newThread(Runnable)
* newThread} method will be called to create threads when {@linkplain #fork(Callable)
* forking} subtasks in this scope. If a {@code ThreadFactory} is not set then
* forking subtasks will create an unnamed virtual thread for each subtask.
*
* <p> If a {@linkplain Configuration#withTimeout(Duration) timeout} is set then it
* starts when the scope is opened. If the timeout expires before the scope has
! * {@linkplain #join() joined} then the scope is {@linkplain ##Cancallation cancelled}
! * and the {@code Joiner}'s {@link Joiner#onTimeout()} method is invoked to throw
+ * optionally throw {@link TimeoutException TimeoutException}.
*
* <p> The new scope is owned by the current thread. Only code executing in this
* thread can {@linkplain #fork(Callable) fork}, {@linkplain #join() join}, or
* {@linkplain #close close} the scope.
*
* <p> Construction captures the current thread's {@linkplain ScopedValue scoped
* value} bindings for inheritance by threads started in the scope.
*
* @param joiner the joiner
! * @param configOperator the operator to produce the configuration
* @return a new scope
* @param <T> the result type of subtasks executed in the scope
* @param <R> the result type of the scope
! * @since 26
*/
static <T, R> StructuredTaskScope<T, R> open(Joiner<? super T, ? extends R> joiner,
! UnaryOperator<Configuration> configOperator) {
! return StructuredTaskScopeImpl.open(joiner, configOperator);
}
/**
* Opens a new {@code StructuredTaskScope}to use the given {@code Joiner} object. The
! * scope is created with the {@linkplain ##DefaultConfiguration default configuration}.
* The default configuration has a {@code ThreadFactory} that creates unnamed
! * {@linkplain Thread##irtual-threads virtual threads}, does not name the scope, and
! * has no timeout.
*
* @implSpec
* This factory method is equivalent to invoking the 2-arg open method with the given
! * joiner and the {@linkplain UnaryOperator#identity() identity operator}.
*
* @param joiner the joiner
* @return a new scope
* @param <T> the result type of subtasks executed in the scope
* @param <R> the result type of the scope
* @since 25
*/
static <T, R> StructuredTaskScope<T, R> open(Joiner<? super T, ? extends R> joiner) {
! return open(joiner, UnaryOperator.identity());
}
/**
* Opens a new {@code StructuredTaskScope} that can be used to fork subtasks that return
* results of any type. The scope's {@link #join()} method waits for all subtasks to
*
* <p> The {@code join} method returns {@code null} if all subtasks complete successfully.
* It throws {@link FailedException} if any subtask fails, with the exception from
* the first subtask to fail as the cause.
*
! * <p> The scope is created with the <a href="#DefaultConfiguration">default
! * configuration</a>. The default configuration has a {@code ThreadFactory} that creates
! * unnamed <a href="{@docRoot}/java.base/java/lang/Thread.html#virtual-threads">virtual
! * threads</a>, is unnamed for monitoring and management purposes, and has no timeout.
*
* @implSpec
* This factory method is equivalent to invoking the 2-arg open method with a joiner
* created with {@link Joiner#awaitAllSuccessfulOrThrow() awaitAllSuccessfulOrThrow()}
! * and the {@linkplain Function#identity() identity function}.
*
* @param <T> the result type of subtasks
* @return a new scope
* @since 25
*/
static <T> StructuredTaskScope<T, Void> open() {
! return open(Joiner.awaitAllSuccessfulOrThrow(), Function.identity());
}
/**
* Fork 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.
*
* <p> The {@code join} method returns {@code null} if all subtasks complete successfully.
* It throws {@link FailedException} if any subtask fails, with the exception from
* the first subtask to fail as the cause.
*
! * <p> The scope is created with the {@linkplain ##DefaultConfiguration default
! * configuration}. The default configuration has a {@code ThreadFactory} that creates
! * unnamed {@linkplain Thread##virtual-threads virtual threads}, does not name the
! * scope, and has no timeout.
*
* @implSpec
* This factory method is equivalent to invoking the 2-arg open method with a joiner
* created with {@link Joiner#awaitAllSuccessfulOrThrow() awaitAllSuccessfulOrThrow()}
! * and the {@linkplain UnaryOperator#identity() identity operator}.
*
* @param <T> the result type of subtasks
* @return a new scope
* @since 25
*/
static <T> StructuredTaskScope<T, Void> open() {
! return open(Joiner.awaitAllSuccessfulOrThrow(), UnaryOperator.identity());
}
/**
* Fork 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.
* <p> This method first creates a {@link Subtask Subtask} object to represent the
* <em>forked subtask</em>. It invokes the joiner's {@link Joiner#onFork(Subtask) onFork}
* method with the subtask in the {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state.
* If the {@code onFork} completes with an exception or error then it is propagated by
* the {@code fork} method without creating a thread. If the scope is already
! * <a href="#Cancallation">cancelled</a>, or {@code onFork} returns {@code true} to
* cancel the scope, then this method returns the {@code Subtask}, in the
* {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state, without creating a thread to
* execute the subtask.
*
* <p> If the scope is not cancelled, and the {@code onFork} method returns {@code false},
* <p> This method first creates a {@link Subtask Subtask} object to represent the
* <em>forked subtask</em>. It invokes the joiner's {@link Joiner#onFork(Subtask) onFork}
* method with the subtask in the {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state.
* If the {@code onFork} completes with an exception or error then it is propagated by
* the {@code fork} method without creating a thread. If the scope is already
! * {@linkplain ##Cancallation cancelled}, or {@code onFork} returns {@code true} to
* cancel the scope, then this method returns the {@code Subtask}, in the
* {@link Subtask.State#UNAVAILABLE UNAVAILABLE} state, without creating a thread to
* execute the subtask.
*
* <p> If the scope is not cancelled, and the {@code onFork} method returns {@code false},
*/
<U extends T> Subtask<U> fork(Runnable task);
/**
* Returns the result, or throws, after waiting for all subtasks to complete or
! * the scope to be <a href="#Cancallation">cancelled</a>.
*
* <p> This method waits for all subtasks started in this scope to complete or the
! * scope to be cancelled. If a {@linkplain Configuration#withTimeout(Duration) timeout}
! * is configured and the timeout expires before or while waiting, then the scope is
! * cancelled and {@link TimeoutException TimeoutException} is thrown. Once finished
! * waiting, the {@code Joiner}'s {@link Joiner#result() result()} method is invoked
! * to get the result or throw an exception. If the {@code result()} method throws
! * then this method throws {@code FailedException} with the exception as the cause.
*
! * <p> This method may only be invoked by the scope owner, and only once.
*
* @return the result
* @throws WrongThreadException if the current thread is not the scope owner
* @throws IllegalStateException if already joined or this scope is closed
* @throws FailedException if the <i>outcome</i> is an exception, thrown with the
* exception from {@link Joiner#result() Joiner.result()} as the cause
! * @throws TimeoutException if a timeout is set and the timeout expires before or
! * while waiting
* @throws InterruptedException if interrupted while waiting
* @since 25
*/
R join() throws InterruptedException;
/**
! * {@return {@code true} if this scope is <a href="#Cancallation">cancelled</a> or in
* the process of being cancelled, otherwise {@code false}}
*
* <p> Cancelling the scope prevents new threads from starting in the scope and
* {@linkplain Thread#interrupt() interrupts} threads executing unfinished subtasks.
* It may take some time before the interrupted threads finish execution; this
*/
<U extends T> Subtask<U> fork(Runnable task);
/**
* Returns the result, or throws, after waiting for all subtasks to complete or
! * the scope to be {@linkplain ##Cancallation cancelled}.
*
* <p> This method waits for all subtasks started in this scope to complete or the
! * scope to be cancelled. Once finished waiting, the {@code Joiner}'s {@link
! * Joiner#result() result()} method is invoked to get the result or throw an exception.
! * If the {@code result()} method throws then {@code join()} throws
! * {@code FailedException} with the exception from the {@code Joiner} as the cause.
! *
! * <p> If a {@linkplain Configuration#withTimeout(Duration) timeout} is configured,
+ * and the timeout expires before or while waiting, then the scope is cancelled and
+ * the {@code Joiner}'s {@link Joiner#onTimeout() onTimeout()} method is invoked
+ * before calling the {@code Joiner}'s {@code result()} method. If the {@code onTimeout()}
+ * method throws {@link TimeoutException TimeoutException} (or throws any exception
+ * or error), then it is propagated by this method. If the {@code onTimeout()} method
+ * does not throw then the {@code Joiner}'s {@code result()} method is invoked to
+ * get the result or throw.
*
! * <p> This method may only be invoked by the scope owner. Once the result or
+ * exception outcome is obtained, this method may not be invoked again. The only
+ * case where the method may be called again is where {@code InterruptedException}
+ * is thrown while waiting.
*
* @return the result
* @throws WrongThreadException if the current thread is not the scope owner
* @throws IllegalStateException if already joined or this scope is closed
* @throws FailedException if the <i>outcome</i> is an exception, thrown with the
* exception from {@link Joiner#result() Joiner.result()} as the cause
! * @throws TimeoutException if a timeout is set, the timeout expires before or while
! * waiting, and {@link Joiner#onTimeout() Joiner.onTimeout()} throws this exception
* @throws InterruptedException if interrupted while waiting
* @since 25
*/
R join() throws InterruptedException;
/**
! * {@return {@code true} if this scope is {@linkplain ##Cancallation cancelled} or in
* the process of being cancelled, otherwise {@code false}}
*
* <p> Cancelling the scope prevents new threads from starting in the scope and
* {@linkplain Thread#interrupt() interrupts} threads executing unfinished subtasks.
* It may take some time before the interrupted threads finish execution; this
boolean isCancelled();
/**
* Closes this scope.
*
! * <p> This method first <a href="#Cancallation">cancels</a> 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 then it
* will continue to wait until the threads finish, before completing with the interrupt
* status set.
*
boolean isCancelled();
/**
* Closes this scope.
*
! * <p> This method first {@linkplain ##Cancallation 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 then it
* will continue to wait until the threads finish, before completing with the interrupt
* status set.
*
* @throws WrongThreadException if the current thread is not the scope owner
* @throws StructureViolationException if a structure violation was detected
*/
@Override
void close();
! }
* @throws WrongThreadException if the current thread is not the scope owner
* @throws StructureViolationException if a structure violation was detected
*/
@Override
void close();
! }
< prev index next >