1 /*
2 * Copyright (c) 2021, 2023, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25 package java.util.concurrent;
26
27 import java.lang.invoke.MethodHandles;
28 import java.lang.invoke.VarHandle;
29 import java.security.AccessController;
30 import java.security.PrivilegedAction;
31 import java.time.Duration;
32 import java.time.Instant;
33 import java.util.Objects;
34 import java.util.Optional;
35 import java.util.concurrent.locks.ReentrantLock;
36 import java.util.function.Function;
37 import java.util.function.Supplier;
38 import jdk.internal.javac.PreviewFeature;
39 import jdk.internal.misc.ThreadFlock;
40
41 /**
42 * A basic API for <em>structured concurrency</em>. {@code StructuredTaskScope} supports
43 * cases where a task splits into several concurrent subtasks, and where the subtasks must
44 * complete before the main task continues. A {@code StructuredTaskScope} can be used to
45 * ensure that the lifetime of a concurrent operation is confined by a <em>syntax block</em>,
46 * just like that of a sequential operation in structured programming.
47 *
48 * <h2>Basic operation</h2>
49 *
50 * A {@code StructuredTaskScope} is created with one of its public constructors. It defines
51 * the {@link #fork(Callable) fork} method to start a thread to execute a subtask, the {@link
52 * #join() join} method to wait for all subtasks to finish, and the {@link #close() close}
53 * method to close the task scope. The API is intended to be used with the {@code
54 * try-with-resources} statement. The intention is that code in the try <em>block</em>
55 * uses the {@code fork} method to fork threads to execute the subtasks, wait for the
56 * subtasks to finish with the {@code join} method, and then <em>process the results</em>.
57 * A call to the {@code fork} method returns a {@link Subtask Subtask} to representing
58 * the <em>forked subtask</em>. Once {@code join} is called, the {@code Subtask} can be
59 * used to get the result completed successfully, or the exception if the subtask failed.
60 * {@snippet lang=java :
61 * Callable<String> task1 = ...
62 * Callable<Integer> task2 = ...
63 *
64 * try (var scope = new StructuredTaskScope<Object>()) {
65 *
66 * Subtask<String> subtask1 = scope.fork(task1); // @highlight substring="fork"
67 * Subtask<Integer> subtask2 = scope.fork(task2); // @highlight substring="fork"
68 *
69 * scope.join(); // @highlight substring="join"
70 *
71 * ... process results/exceptions ...
72 *
73 * } // close // @highlight substring="close"
74 * }
75 * <p> The following example forks a collection of homogeneous subtasks, waits for all of
76 * them to complete with the {@code join} method, and uses the {@link Subtask.State
77 * Subtask.State} to partition the subtasks into a set of the subtasks that completed
78 * successfully and another for the subtasks that failed.
79 * {@snippet lang=java :
80 * List<Callable<String>> callables = ...
81 *
82 * try (var scope = new StructuredTaskScope<String>()) {
83 *
84 * List<Subtask<String>> subtasks = callables.stream().map(scope::fork).toList();
85 *
86 * scope.join();
87 *
88 * Map<Boolean, Set<Subtask<String>>> map = subtasks.stream()
89 * .collect(Collectors.partitioningBy(h -> h.state() == Subtask.State.SUCCESS,
90 * Collectors.toSet()));
91 *
92 * } // close
93 * }
94 *
95 * <p> To ensure correct usage, the {@code join} and {@code close} methods may only be
96 * invoked by the <em>owner</em> (the thread that opened/created the task scope), and the
97 * {@code close} method throws an exception after closing if the owner did not invoke the
98 * {@code join} method after forking.
99 *
100 * <p> {@code StructuredTaskScope} defines the {@link #shutdown() shutdown} method to shut
101 * down a task scope without closing it. The {@code shutdown()} method <em>cancels</em> all
102 * unfinished subtasks by {@linkplain Thread#interrupt() interrupting} the threads. It
103 * prevents new threads from starting in the task scope. If the owner is waiting in the
104 * {@code join} method then it will wakeup.
105 *
106 * <p> Shutdown is used for <em>short-circuiting</em> and allow subclasses to implement
107 * <em>policy</em> that does not require all subtasks to finish.
108 *
109 * <h2>Subclasses with policies for common cases</h2>
110 *
111 * Two subclasses of {@code StructuredTaskScope} are defined to implement policy for
112 * common cases:
113 * <ol>
114 * <li> {@link ShutdownOnSuccess ShutdownOnSuccess} captures the result of the first
115 * subtask to complete successfully. Once captured, it shuts down the task scope to
116 * interrupt unfinished threads and wakeup the owner. This class is intended for cases
117 * where the result of any subtask will do ("invoke any") and where there is no need to
118 * wait for results of other unfinished subtasks. It defines methods to get the first
119 * result or throw an exception if all subtasks fail.
120 * <li> {@link ShutdownOnFailure ShutdownOnFailure} captures the exception of the first
121 * subtask to fail. Once captured, it shuts down the task scope to interrupt unfinished
122 * threads and wakeup the owner. This class is intended for cases where the results of all
123 * subtasks are required ("invoke all"); if any subtask fails then the results of other
124 * unfinished subtasks are no longer needed. If defines methods to throw an exception if
125 * any of the subtasks fail.
126 * </ol>
127 *
128 * <p> The following are two examples that use the two classes. In both cases, a pair of
129 * subtasks are forked to fetch resources from two URL locations "left" and "right". The
130 * first example creates a ShutdownOnSuccess object to capture the result of the first
131 * subtask to complete successfully, cancelling the other by way of shutting down the task
132 * scope. The main task waits in {@code join} until either subtask completes with a result
133 * or both subtasks fail. It invokes {@link ShutdownOnSuccess#result(Function)
134 * result(Function)} method to get the captured result. If both subtasks fail then this
135 * method throws a {@code WebApplicationException} with the exception from one of the
136 * subtasks as the cause.
137 * {@snippet lang=java :
138 * try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
139 *
140 * scope.fork(() -> fetch(left));
141 * scope.fork(() -> fetch(right));
142 *
143 * scope.join();
144 *
145 * // @link regex="result(?=\()" target="ShutdownOnSuccess#result" :
146 * String result = scope.result(e -> new WebApplicationException(e));
147 *
148 * ...
149 * }
150 * }
151 * The second example creates a ShutdownOnFailure object to capture the exception of the
152 * first subtask to fail, cancelling the other by way of shutting down the task scope. The
153 * main task waits in {@link #joinUntil(Instant)} until both subtasks complete with a
154 * result, either fails, or a deadline is reached. It invokes {@link
155 * ShutdownOnFailure#throwIfFailed(Function) throwIfFailed(Function)} to throw an exception
156 * if either subtask fails. This method is a no-op if both subtasks complete successfully.
157 * The example uses {@link Supplier#get()} to get the result of each subtask. Using
158 * {@code Supplier} instead of {@code Subtask} is preferred for common cases where the
159 * object returned by fork is only used to get the result of a subtask that completed
160 * successfully.
161 * {@snippet lang=java :
162 * Instant deadline = ...
163 *
164 * try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
165 *
166 * Supplier<String> supplier1 = scope.fork(() -> query(left));
167 * Supplier<String> supplier2 = scope.fork(() -> query(right));
168 *
169 * scope.joinUntil(deadline);
170 *
171 * // @link substring="throwIfFailed" target="ShutdownOnFailure#throwIfFailed" :
172 * scope.throwIfFailed(e -> new WebApplicationException(e));
173 *
174 * // both subtasks completed successfully
175 * String result = Stream.of(supplier1, supplier2)
176 * .map(Supplier::get)
177 * .collect(Collectors.joining(", ", "{ ", " }"));
178 *
179 * ...
180 * }
181 * }
182 *
183 * <h2>Extending StructuredTaskScope</h2>
184 *
185 * {@code StructuredTaskScope} can be extended, and the {@link #handleComplete(Subtask)
186 * handleComplete} method overridden, to implement policies other than those implemented
187 * by {@code ShutdownOnSuccess} and {@code ShutdownOnFailure}. A subclass may, for example,
188 * collect the results of subtasks that complete successfully and ignore subtasks that
189 * fail. It may collect exceptions when subtasks fail. It may invoke the {@link #shutdown()
190 * shutdown} method to shut down and cause {@link #join() join} to wakeup when some
191 * condition arises.
192 *
193 * <p> A subclass will typically define methods to make available results, state, or other
194 * outcome to code that executes after the {@code join} method. A subclass that collects
195 * results and ignores subtasks that fail may define a method that returns the results.
196 * A subclass that implements a policy to shut down when a subtask fails may define a
197 * method to get the exception of the first subtask to fail.
198 *
199 * <p> The following is an example of a simple {@code StructuredTaskScope} implementation
200 * that collects homogenous subtasks that complete successfully. It defines the method
201 * "{@code completedSuccessfully()}" that the main task can invoke after it joins.
202 * {@snippet lang=java :
203 * class CollectingScope<T> extends StructuredTaskScope<T> {
204 * private final Queue<Subtask<? extends T>> subtasks = new LinkedTransferQueue<>();
205 *
206 * @Override
207 * protected void handleComplete(Subtask<? extends T> subtask) {
208 * if (subtask.state() == Subtask.State.SUCCESS) {
209 * subtasks.add(subtask);
210 * }
211 * }
212 *
213 * @Override
214 * public CollectingScope<T> join() throws InterruptedException {
215 * super.join();
216 * return this;
217 * }
218 *
219 * public Stream<Subtask<? extends T>> completedSuccessfully() {
220 * // @link substring="ensureOwnerAndJoined" target="ensureOwnerAndJoined" :
221 * super.ensureOwnerAndJoined();
222 * return subtasks.stream();
223 * }
224 * }
225 * }
226 * <p> The implementations of the {@code completedSuccessfully()} method in the example
227 * invokes {@link #ensureOwnerAndJoined()} to ensure that the method can only be invoked
228 * by the owner thread and only after it has joined.
229 *
230 * <h2><a id="TreeStructure">Tree structure</a></h2>
231 *
232 * Task scopes form a tree where parent-child relations are established implicitly when
233 * opening a new task scope:
234 * <ul>
235 * <li> A parent-child relation is established when a thread started in a task scope
236 * opens its own task scope. A thread started in task scope "A" that opens task scope
237 * "B" establishes a parent-child relation where task scope "A" is the parent of task
238 * scope "B".
239 * <li> A parent-child relation is established with nesting. If a thread opens task
240 * scope "B", then opens task scope "C" (before it closes "B"), then the enclosing task
241 * scope "B" is the parent of the nested task scope "C".
242 * </ul>
243 *
244 * The <i>descendants</i> of a task scope are the child task scopes that it is a parent
245 * of, plus the descendants of the child task scopes, recursively.
246 *
247 * <p> The tree structure supports:
248 * <ul>
249 * <li> Inheritance of {@linkplain ScopedValue scoped values} across threads.
250 * <li> Confinement checks. The phrase "threads contained in the task scope" in method
251 * descriptions means threads started in the task scope or descendant scopes.
252 * </ul>
253 *
254 * <p> The following example demonstrates the inheritance of a scoped value. A scoped
255 * value {@code USERNAME} is bound to the value "{@code duke}". A {@code StructuredTaskScope}
256 * is created and its {@code fork} method invoked to start a thread to execute {@code
257 * childTask}. The thread inherits the scoped value <em>bindings</em> captured when
258 * creating the task scope. The code in {@code childTask} uses the value of the scoped
259 * value and so reads the value "{@code duke}".
260 * {@snippet lang=java :
261 * private static final ScopedValue<String> USERNAME = ScopedValue.newInstance();
262 *
263 * // @link substring="runWhere" target="ScopedValue#runWhere(ScopedValue, Object, Runnable)" :
264 * ScopedValue.runWhere(USERNAME, "duke", () -> {
265 * try (var scope = new StructuredTaskScope<String>()) {
266 *
267 * scope.fork(() -> childTask()); // @highlight substring="fork"
268 * ...
269 * }
270 * });
271 *
272 * ...
273 *
274 * String childTask() {
275 * // @link substring="get" target="ScopedValue#get()" :
276 * String name = USERNAME.get(); // "duke"
277 * ...
278 * }
279 * }
280 *
281 * <p> {@code StructuredTaskScope} does not define APIs that exposes the tree structure
282 * at this time.
283 *
284 * <p> Unless otherwise specified, passing a {@code null} argument to a constructor
285 * or method in this class will cause a {@link NullPointerException} to be thrown.
286 *
287 * <h2>Memory consistency effects</h2>
288 *
289 * <p> Actions in the owner thread of, or a thread contained in, the task scope prior to
290 * {@linkplain #fork forking} of a subtask
291 * <a href="{@docRoot}/java.base/java/util/concurrent/package-summary.html#MemoryVisibility">
292 * <i>happen-before</i></a> any actions taken by that subtask, which in turn <i>happen-before</i>
293 * the subtask result is {@linkplain Subtask#get() retrieved} or <i>happen-before</i> any
294 * actions taken in a thread after {@linkplain #join() joining} of the task scope.
295 *
296 * @jls 17.4.5 Happens-before Order
297 *
298 * @param <T> the result type of tasks executed in the task scope
299 * @since 21
300 */
301 @PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
302 public class StructuredTaskScope<T> implements AutoCloseable {
303 private final ThreadFactory factory;
304 private final ThreadFlock flock;
305 private final ReentrantLock shutdownLock = new ReentrantLock();
306
307 // states: OPEN -> SHUTDOWN -> CLOSED
308 private static final int OPEN = 0; // initial state
309 private static final int SHUTDOWN = 1;
310 private static final int CLOSED = 2;
311
312 // state: set to SHUTDOWN by any thread, set to CLOSED by owner, read by any thread
313 private volatile int state;
314
315 // Counters to support checking that the task scope owner joins before closing the task
316 // scope. These counters are accessed only by the owner thread.
317 private int forkRound; // incremented when the first subtask is forked after join
318 private int lastJoinAttempted; // set to the current fork round when join is attempted
319 private int lastJoinCompleted; // set to the current fork round when join completes
320
321 /**
322 * Represents a subtask forked with {@link #fork(Callable)}.
323 * @param <T> the result type
324 * @since 21
325 */
326 @PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
327 public sealed interface Subtask<T> extends Supplier<T> permits SubtaskImpl {
328 /**
329 * {@return the value returning task provided to the {@code fork} method}
330 *
331 * @apiNote Task objects with unique identity may be used for correlation by
332 * implementations of {@link #handleComplete(Subtask) handleComplete}.
333 */
334 Callable<? extends T> task();
335
336 /**
337 * Represents the state of a subtask.
338 * @see Subtask#state()
339 * @since 21
340 */
341 @PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
342 enum State {
343 /**
344 * The subtask result or exception is not available. This state indicates that
345 * the subtask was forked but has not completed, it completed after the task
346 * scope was {@linkplain #shutdown() shut down}, or it was forked after the
347 * task scope was shut down.
348 */
349 UNAVAILABLE,
350 /**
351 * The subtask completed successfully with a result. The {@link Subtask#get()
352 * Subtask.get()} method can be used to obtain the result. This is a terminal
353 * state.
354 */
355 SUCCESS,
356 /**
357 * The subtask failed with an exception. The {@link Subtask#exception()
358 * Subtask.exception()} method can be used to obtain the exception. This is a
359 * terminal state.
360 */
361 FAILED,
362 }
363
364 /**
365 * {@return the state of the subtask}
366 */
367 State state();
368
369 /**
370 * Returns the result of the subtask.
371 *
372 * @return the possibly-null result
373 * @throws IllegalStateException if the subtask has not completed or did not
374 * complete successfully
375 * @see State#SUCCESS
376 */
377 T get();
378
379 /**
380 * {@return the exception thrown by the subtask}
381 *
382 * @throws IllegalStateException if the subtask has not completed or completed
383 * with a result rather than an exception
384 * @see State#FAILED
385 */
386 Throwable exception();
387 }
388
389 /**
390 * Creates a structured task scope with the given name and thread factory. The task
391 * scope is optionally named for the purposes of monitoring and management. The thread
392 * factory is used to {@link ThreadFactory#newThread(Runnable) create} threads when
393 * subtasks are {@linkplain #fork(Callable) forked}. The task scope is owned by the
394 * current thread.
395 *
396 * <p> Construction captures the current thread's {@linkplain ScopedValue scoped value}
397 * bindings for inheritance by threads started in the task scope. The
398 * <a href="#TreeStructure">Tree Structure</a> section in the class description details
399 * how parent-child relations are established implicitly for the purpose of inheritance
400 * of scoped value bindings.
401 *
402 * @param name the name of the task scope, can be null
403 * @param factory the thread factory
404 */
405 public StructuredTaskScope(String name, ThreadFactory factory) {
406 this.factory = Objects.requireNonNull(factory, "'factory' is null");
407 if (name == null)
408 name = Objects.toIdentityString(this);
409 this.flock = ThreadFlock.open(name);
410 }
411
412 /**
413 * Creates an unnamed structured task scope that creates virtual threads. The task
414 * scope is owned by the current thread.
415 *
416 * @implSpec This constructor is equivalent to invoking the 2-arg constructor with a
417 * name of {@code null} and a thread factory that creates virtual threads.
418 */
419 public StructuredTaskScope() {
420 this(null, Thread.ofVirtual().factory());
421 }
422
423 private IllegalStateException newIllegalStateExceptionScopeClosed() {
424 return new IllegalStateException("Task scope is closed");
425 }
426
427 private IllegalStateException newIllegalStateExceptionNoJoin() {
428 return new IllegalStateException("Owner did not join after forking subtasks");
429 }
430
431 /**
432 * Throws IllegalStateException if the scope is closed, returning the state if not
433 * closed.
434 */
435 private int ensureOpen() {
436 int s = state;
437 if (s == CLOSED)
438 throw newIllegalStateExceptionScopeClosed();
439 return s;
440 }
441
442 /**
443 * Throws WrongThreadException if the current thread is not the owner.
444 */
445 private void ensureOwner() {
446 if (Thread.currentThread() != flock.owner())
447 throw new WrongThreadException("Current thread not owner");
448 }
449
450 /**
451 * Throws WrongThreadException if the current thread is not the owner
452 * or a thread contained in the tree.
453 */
454 private void ensureOwnerOrContainsThread() {
455 Thread currentThread = Thread.currentThread();
456 if (currentThread != flock.owner() && !flock.containsThread(currentThread))
457 throw new WrongThreadException("Current thread not owner or thread in the tree");
458 }
459
460 /**
461 * Ensures that the current thread is the owner of this task scope and that it joined
462 * (with {@link #join()} or {@link #joinUntil(Instant)}) after {@linkplain #fork(Callable)
463 * forking} subtasks.
464 *
465 * @apiNote This method can be used by subclasses that define methods to make available
466 * results, state, or other outcome to code intended to execute after the join method.
467 *
468 * @throws WrongThreadException if the current thread is not the task scope owner
469 * @throws IllegalStateException if the task scope is open and task scope owner did
470 * not join after forking
471 */
472 protected final void ensureOwnerAndJoined() {
473 ensureOwner();
474 if (forkRound > lastJoinCompleted) {
475 throw newIllegalStateExceptionNoJoin();
476 }
477 }
478
479 /**
480 * Invoked by a subtask when it completes successfully or fails in this task scope.
481 * This method is not invoked if a subtask completes after the task scope is
482 * {@linkplain #shutdown() shut down}.
483 *
484 * @implSpec The default implementation throws {@code NullPointerException} if the
485 * subtask is {@code null}. It throws {@link IllegalArgumentException} if the subtask
486 * has not completed.
487 *
488 * @apiNote The {@code handleComplete} method should be thread safe. It may be
489 * invoked by several threads concurrently.
490 *
491 * @param subtask the subtask
492 *
493 * @throws IllegalArgumentException if called with a subtask that has not completed
494 */
495 protected void handleComplete(Subtask<? extends T> subtask) {
496 if (subtask.state() == Subtask.State.UNAVAILABLE)
497 throw new IllegalArgumentException();
498 }
499
500 /**
501 * Starts a new thread in this task scope to execute a value-returning task, thus
502 * creating a <em>subtask</em> of this task scope.
503 *
504 * <p> The value-returning task is provided to this method as a {@link Callable}, the
505 * thread executes the task's {@link Callable#call() call} method. The thread is
506 * created with the task scope's {@link ThreadFactory}. It inherits the current thread's
507 * {@linkplain ScopedValue scoped value} bindings. The bindings must match the bindings
508 * captured when the task scope was created.
509 *
510 * <p> This method returns a {@link Subtask Subtask} to represent the <em>forked
511 * subtask</em>. The {@code Subtask} object can be used to obtain the result when
512 * the subtask completes successfully, or the exception when the subtask fails. To
513 * ensure correct usage, the {@link Subtask#get() get()} and {@link Subtask#exception()
514 * exception()} methods may only be called by the task scope owner after it has waited
515 * for all threads to finish with the {@link #join() join} or {@link #joinUntil(Instant)}
516 * methods. When the subtask completes, the thread invokes the {@link
517 * #handleComplete(Subtask) handleComplete} method to consume the completed subtask.
518 * If the task scope is {@linkplain #shutdown() shut down} before the subtask completes
519 * then the {@code handleComplete} method will not be invoked.
520 *
521 * <p> If this task scope is {@linkplain #shutdown() shutdown} (or in the process of
522 * shutting down) then the subtask will not run and the {@code handleComplete} method
523 * will not be invoked.
524 *
525 * <p> This method may only be invoked by the task scope owner or threads contained
526 * in the task scope.
527 *
528 * @implSpec This method may be overridden for customization purposes, wrapping tasks
529 * for example. If overridden, the subclass must invoke {@code super.fork} to start a
530 * new thread in this task scope.
531 *
532 * @param task the value-returning task for the thread to execute
533 * @param <U> the result type
534 * @return the subtask
535 * @throws IllegalStateException if this task scope is closed
536 * @throws WrongThreadException if the current thread is not the task scope owner or a
537 * thread contained in the task scope
538 * @throws StructureViolationException if the current scoped value bindings are not
539 * the same as when the task scope was created
540 * @throws RejectedExecutionException if the thread factory rejected creating a
541 * thread to run the subtask
542 */
543 public <U extends T> Subtask<U> fork(Callable<? extends U> task) {
544 Objects.requireNonNull(task, "'task' is null");
545 int s = ensureOpen(); // throws ISE if closed
546
547 // when forked by the owner, the subtask is forked in the current or next round
548 int round = -1;
549 if (Thread.currentThread() == flock.owner()) {
550 round = forkRound;
551 if (forkRound == lastJoinCompleted) {
552 // new round if first fork after join
553 round++;
554 }
555 }
556
557 var subtask = new SubtaskImpl<U>(this, task);
558 if (s < SHUTDOWN) {
559 // create thread to run task
560 Thread thread = factory.newThread(subtask);
561 if (thread == null) {
562 throw new RejectedExecutionException("Rejected by thread factory");
563 }
564
565 // attempt to start the thread
566 try {
567 flock.start(thread);
568 } catch (IllegalStateException e) {
569 // shutdown by another thread, or underlying flock is shutdown due
570 // to unstructured use
571 }
572 }
573
574 // force owner to join if this is the first fork in the round
575 if (Thread.currentThread() == flock.owner() && round > forkRound) {
576 forkRound = round;
577 }
578
579 // return forked subtask or a subtask that did not run
580 return subtask;
581 }
582
583 /**
584 * Wait for all threads to finish or the task scope to shut down.
585 */
586 private void implJoin(Duration timeout)
587 throws InterruptedException, TimeoutException
588 {
589 ensureOwner();
590 lastJoinAttempted = forkRound;
591 int s = ensureOpen(); // throws ISE if closed
592 if (s == OPEN) {
593 // wait for all threads, wakeup, interrupt, or timeout
594 if (timeout != null) {
595 flock.awaitAll(timeout);
596 } else {
597 flock.awaitAll();
598 }
599 }
600 lastJoinCompleted = forkRound;
601 }
602
603 /**
604 * Wait for all subtasks started in this task scope to finish or the task scope to
605 * shut down.
606 *
607 * <p> This method waits for all subtasks by waiting for all threads {@linkplain
608 * #fork(Callable) started} in this task scope to finish execution. It stops waiting
609 * when all threads finish, the task scope is {@linkplain #shutdown() shut down}, or
610 * the current thread is {@linkplain Thread#interrupt() interrupted}.
611 *
612 * <p> This method may only be invoked by the task scope owner.
613 *
614 * @implSpec This method may be overridden for customization purposes or to return a
615 * more specific return type. If overridden, the subclass must invoke {@code
616 * super.join} to ensure that the method waits for threads in this task scope to
617 * finish.
618 *
619 * @return this task scope
620 * @throws IllegalStateException if this task scope is closed
621 * @throws WrongThreadException if the current thread is not the task scope owner
622 * @throws InterruptedException if interrupted while waiting
623 */
624 public StructuredTaskScope<T> join() throws InterruptedException {
625 try {
626 implJoin(null);
627 } catch (TimeoutException e) {
628 throw new InternalError();
629 }
630 return this;
631 }
632
633 /**
634 * Wait for all subtasks started in this task scope to finish or the task scope to
635 * shut down, up to the given deadline.
636 *
637 * <p> This method waits for all subtasks by waiting for all threads {@linkplain
638 * #fork(Callable) started} in this task scope to finish execution. It stops waiting
639 * when all threads finish, the task scope is {@linkplain #shutdown() shut down}, the
640 * deadline is reached, or the current thread is {@linkplain Thread#interrupt()
641 * interrupted}.
642 *
643 * <p> This method may only be invoked by the task scope owner.
644 *
645 * @implSpec This method may be overridden for customization purposes or to return a
646 * more specific return type. If overridden, the subclass must invoke {@code
647 * super.joinUntil} to ensure that the method waits for threads in this task scope to
648 * finish.
649 *
650 * @param deadline the deadline
651 * @return this task scope
652 * @throws IllegalStateException if this task scope is closed
653 * @throws WrongThreadException if the current thread is not the task scope owner
654 * @throws InterruptedException if interrupted while waiting
655 * @throws TimeoutException if the deadline is reached while waiting
656 */
657 public StructuredTaskScope<T> joinUntil(Instant deadline)
658 throws InterruptedException, TimeoutException
659 {
660 Duration timeout = Duration.between(Instant.now(), deadline);
661 implJoin(timeout);
662 return this;
663 }
664
665 /**
666 * Interrupt all unfinished threads.
667 */
668 private void implInterruptAll() {
669 flock.threads()
670 .filter(t -> t != Thread.currentThread())
671 .forEach(t -> {
672 try {
673 t.interrupt();
674 } catch (Throwable ignore) { }
675 });
676 }
677
678 @SuppressWarnings("removal")
679 private void interruptAll() {
680 if (System.getSecurityManager() == null) {
681 implInterruptAll();
682 } else {
683 PrivilegedAction<Void> pa = () -> {
684 implInterruptAll();
685 return null;
686 };
687 AccessController.doPrivileged(pa);
688 }
689 }
690
691 /**
692 * Shutdown the task scope if not already shutdown. Return true if this method
693 * shutdowns the task scope, false if already shutdown.
694 */
695 private boolean implShutdown() {
696 shutdownLock.lock();
697 try {
698 if (state < SHUTDOWN) {
699 // prevent new threads from starting
700 flock.shutdown();
701
702 // set status before interrupting tasks
703 state = SHUTDOWN;
704
705 // interrupt all unfinished threads
706 interruptAll();
707
708 return true;
709 } else {
710 // already shutdown
711 return false;
712 }
713 } finally {
714 shutdownLock.unlock();
715 }
716 }
717
718 /**
719 * Shut down this task scope without closing it. Shutting down a task scope prevents
720 * new threads from starting, interrupts all unfinished threads, and causes the
721 * {@link #join() join} method to wakeup. Shutdown is useful for cases where the
722 * results of unfinished subtasks are no longer needed. It will typically be called
723 * by the {@link #handleComplete(Subtask)} implementation of a subclass that
724 * implements a policy to discard unfinished tasks once some outcome is reached.
725 *
726 * <p> More specifically, this method:
727 * <ul>
728 * <li> {@linkplain Thread#interrupt() Interrupts} all unfinished threads in the
729 * task scope (except the current thread).
730 * <li> Wakes up the task scope owner if it is waiting in {@link #join()} or {@link
731 * #joinUntil(Instant)}. If the task scope owner is not waiting then its next call to
732 * {@code join} or {@code joinUntil} will return immediately.
733 * </ul>
734 *
735 * <p> The {@linkplain Subtask.State state} of unfinished subtasks that complete at
736 * around the time that the task scope is shutdown is not defined. A subtask that
737 * completes successfully with a result, or fails with an exception, at around
738 * the time that the task scope is shutdown may or may not <i>transition</i> to a
739 * terminal state.
740 *
741 * <p> This method may only be invoked by the task scope owner or threads contained
742 * in the task scope.
743 *
744 * @implSpec This method may be overridden for customization purposes. If overridden,
745 * the subclass must invoke {@code super.shutdown} to ensure that the method shuts
746 * down the task scope.
747 *
748 * @apiNote
749 * There may be threads that have not finished because they are executing code that
750 * did not respond (or respond promptly) to thread interrupt. This method does not wait
751 * for these threads. When the owner invokes the {@link #close() close} method
752 * to close the task scope then it will wait for the remaining threads to finish.
753 *
754 * @throws IllegalStateException if this task scope is closed
755 * @throws WrongThreadException if the current thread is not the task scope owner or
756 * a thread contained in the task scope
757 * @see #isShutdown()
758 */
759 public void shutdown() {
760 ensureOwnerOrContainsThread();
761 int s = ensureOpen(); // throws ISE if closed
762 if (s < SHUTDOWN && implShutdown())
763 flock.wakeup();
764 }
765
766 /**
767 * {@return true if this task scope is shutdown, otherwise false}
768 * @see #shutdown()
769 */
770 public final boolean isShutdown() {
771 return state >= SHUTDOWN;
772 }
773
774 /**
775 * Closes this task scope.
776 *
777 * <p> This method first shuts down the task scope (as if by invoking the {@link
778 * #shutdown() shutdown} method). It then waits for the threads executing any
779 * unfinished tasks to finish. If interrupted, this method will continue to wait for
780 * the threads to finish before completing with the interrupt status set.
781 *
782 * <p> This method may only be invoked by the task scope owner. If the task scope
783 * is already closed then the task scope owner invoking this method has no effect.
784 *
785 * <p> A {@code StructuredTaskScope} is intended to be used in a <em>structured
786 * manner</em>. If this method is called to close a task scope before nested task
787 * scopes are closed then it closes the underlying construct of each nested task scope
788 * (in the reverse order that they were created in), closes this task scope, and then
789 * throws {@link StructureViolationException}.
790 * Similarly, if this method is called to close a task scope while executing with
791 * {@linkplain ScopedValue scoped value} bindings, and the task scope was created
792 * before the scoped values were bound, then {@code StructureViolationException} is
793 * thrown after closing the task scope.
794 * If a thread terminates without first closing task scopes that it owns then
795 * termination will cause the underlying construct of each of its open tasks scopes to
796 * be closed. Closing is performed in the reverse order that the task scopes were
797 * created in. Thread termination may therefore be delayed when the task scope owner
798 * has to wait for threads forked in these task scopes to finish.
799 *
800 * @implSpec This method may be overridden for customization purposes. If overridden,
801 * the subclass must invoke {@code super.close} to close the task scope.
802 *
803 * @throws IllegalStateException thrown after closing the task scope if the task scope
804 * owner did not attempt to join after forking
805 * @throws WrongThreadException if the current thread is not the task scope owner
806 * @throws StructureViolationException if a structure violation was detected
807 */
808 @Override
809 public void close() {
810 ensureOwner();
811 int s = state;
812 if (s == CLOSED)
813 return;
814
815 try {
816 if (s < SHUTDOWN)
817 implShutdown();
818 flock.close();
819 } finally {
820 state = CLOSED;
821 }
822
823 // throw ISE if the owner didn't attempt to join after forking
824 if (forkRound > lastJoinAttempted) {
825 lastJoinCompleted = forkRound; // ensureOwnerAndJoined is a no-op after close
826 throw newIllegalStateExceptionNoJoin();
827 }
828 }
829
830 @Override
831 public String toString() {
832 String name = flock.name();
833 return switch (state) {
834 case OPEN -> name;
835 case SHUTDOWN -> name + "/shutdown";
836 case CLOSED -> name + "/closed";
837 default -> throw new InternalError();
838 };
839 }
840
841 /**
842 * Subtask implementation, runs the task specified to the fork method.
843 */
844 private static final class SubtaskImpl<T> implements Subtask<T>, Runnable {
845 private static final AltResult RESULT_NULL = new AltResult(Subtask.State.SUCCESS);
846
847 private record AltResult(Subtask.State state, Throwable exception) {
848 AltResult(Subtask.State state) {
849 this(state, null);
850 }
851 }
852
853 private final StructuredTaskScope<? super T> scope;
854 private final Callable<? extends T> task;
855 private volatile Object result;
856
857 SubtaskImpl(StructuredTaskScope<? super T> scope, Callable<? extends T> task) {
858 this.scope = scope;
859 this.task = task;
860 }
861
862 @Override
863 public void run() {
864 T result = null;
865 Throwable ex = null;
866 try {
867 result = task.call();
868 } catch (Throwable e) {
869 ex = e;
870 }
871
872 // nothing to do if task scope is shutdown
873 if (scope.isShutdown())
874 return;
875
876 // capture result or exception, invoke handleComplete
877 if (ex == null) {
878 this.result = (result != null) ? result : RESULT_NULL;
879 } else {
880 this.result = new AltResult(State.FAILED, ex);
881 }
882 scope.handleComplete(this);
883 }
884
885 @Override
886 public Callable<? extends T> task() {
887 return task;
888 }
889
890 @Override
891 public Subtask.State state() {
892 Object result = this.result;
893 if (result == null) {
894 return State.UNAVAILABLE;
895 } else if (result instanceof AltResult alt) {
896 // null or failed
897 return alt.state();
898 } else {
899 return State.SUCCESS;
900 }
901 }
902
903 @Override
904 public T get() {
905 Object result = this.result;
906 if (result instanceof AltResult) {
907 if (result == RESULT_NULL) return null;
908 } else if (result != null) {
909 @SuppressWarnings("unchecked")
910 T r = (T) result;
911 return r;
912 }
913 throw new IllegalStateException(
914 "Result is unavailable or subtask did not complete successfully");
915 }
916
917 @Override
918 public Throwable exception() {
919 Object result = this.result;
920 if (result instanceof AltResult alt && alt.state() == State.FAILED) {
921 return alt.exception();
922 }
923 throw new IllegalStateException(
924 "Exception is unavailable or subtask did not complete with exception");
925 }
926
927 @Override
928 public String toString() {
929 String stateAsString = switch (state()) {
930 case UNAVAILABLE -> "[Unavailable]";
931 case SUCCESS -> "[Completed successfully]";
932 case FAILED -> {
933 Throwable ex = ((AltResult) result).exception();
934 yield "[Failed: " + ex + "]";
935 }
936 };
937 return Objects.toIdentityString(this) + stateAsString;
938 }
939 }
940
941 /**
942 * A {@code StructuredTaskScope} that captures the result of the first subtask to
943 * complete {@linkplain Subtask.State#SUCCESS successfully}. Once captured, it
944 * {@linkplain #shutdown() shuts down} the task scope to interrupt unfinished threads
945 * and wakeup the task scope owner. The policy implemented by this class is intended
946 * for cases where the result of any subtask will do ("invoke any") and where the
947 * results of other unfinished subtasks are no longer needed.
948 *
949 * <p> Unless otherwise specified, passing a {@code null} argument to a method
950 * in this class will cause a {@link NullPointerException} to be thrown.
951 *
952 * @apiNote This class implements a policy to shut down the task scope when a subtask
953 * completes successfully. There shouldn't be any need to directly shut down the task
954 * scope with the {@link #shutdown() shutdown} method.
955 *
956 * @param <T> the result type
957 * @since 21
958 */
959 @PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
960 public static final class ShutdownOnSuccess<T> extends StructuredTaskScope<T> {
961 private static final Object RESULT_NULL = new Object();
962 private static final VarHandle FIRST_RESULT;
963 private static final VarHandle FIRST_EXCEPTION;
964 static {
965 try {
966 MethodHandles.Lookup l = MethodHandles.lookup();
967 FIRST_RESULT = l.findVarHandle(ShutdownOnSuccess.class, "firstResult", Object.class);
968 FIRST_EXCEPTION = l.findVarHandle(ShutdownOnSuccess.class, "firstException", Throwable.class);
969 } catch (Exception e) {
970 throw new ExceptionInInitializerError(e);
971 }
972 }
973 private volatile Object firstResult;
974 private volatile Throwable firstException;
975
976 /**
977 * Constructs a new {@code ShutdownOnSuccess} with the given name and thread factory.
978 * The task scope is optionally named for the purposes of monitoring and management.
979 * The thread factory is used to {@link ThreadFactory#newThread(Runnable) create}
980 * threads when subtasks are {@linkplain #fork(Callable) forked}. The task scope
981 * is owned by the current thread.
982 *
983 * <p> Construction captures the current thread's {@linkplain ScopedValue scoped
984 * value} bindings for inheritance by threads started in the task scope. The
985 * <a href="#TreeStructure">Tree Structure</a> section in the class description
986 * details how parent-child relations are established implicitly for the purpose
987 * of inheritance of scoped value bindings.
988 *
989 * @param name the name of the task scope, can be null
990 * @param factory the thread factory
991 */
992 public ShutdownOnSuccess(String name, ThreadFactory factory) {
993 super(name, factory);
994 }
995
996 /**
997 * Constructs a new unnamed {@code ShutdownOnSuccess} that creates virtual threads.
998 *
999 * @implSpec This constructor is equivalent to invoking the 2-arg constructor with
1000 * a name of {@code null} and a thread factory that creates virtual threads.
1001 */
1002 public ShutdownOnSuccess() {
1003 this(null, Thread.ofVirtual().factory());
1004 }
1005
1006 @Override
1007 protected void handleComplete(Subtask<? extends T> subtask) {
1008 if (firstResult != null) {
1009 // already captured a result
1010 return;
1011 }
1012
1013 if (subtask.state() == Subtask.State.SUCCESS) {
1014 // task succeeded
1015 T result = subtask.get();
1016 Object r = (result != null) ? result : RESULT_NULL;
1017 if (FIRST_RESULT.compareAndSet(this, null, r)) {
1018 super.shutdown();
1019 }
1020 } else if (firstException == null) {
1021 // capture the exception thrown by the first subtask that failed
1022 FIRST_EXCEPTION.compareAndSet(this, null, subtask.exception());
1023 }
1024 }
1025
1026 /**
1027 * Wait for a subtask started in this task scope to complete {@linkplain
1028 * Subtask.State#SUCCESS successfully} or all subtasks to complete.
1029 *
1030 * <p> This method waits for all subtasks by waiting for all threads {@linkplain
1031 * #fork(Callable) started} in this task scope to finish execution. It stops waiting
1032 * when all threads finish, a subtask completes successfully, or the current
1033 * thread is {@linkplain Thread#interrupt() interrupted}. It also stops waiting
1034 * if the {@link #shutdown() shutdown} method is invoked directly to shut down
1035 * this task scope.
1036 *
1037 * <p> This method may only be invoked by the task scope owner.
1038 *
1039 * @throws IllegalStateException {@inheritDoc}
1040 * @throws WrongThreadException {@inheritDoc}
1041 */
1042 @Override
1043 public ShutdownOnSuccess<T> join() throws InterruptedException {
1044 super.join();
1045 return this;
1046 }
1047
1048 /**
1049 * Wait for a subtask started in this task scope to complete {@linkplain
1050 * Subtask.State#SUCCESS successfully} or all subtasks to complete, up to the
1051 * given deadline.
1052 *
1053 * <p> This method waits for all subtasks by waiting for all threads {@linkplain
1054 * #fork(Callable) started} in this task scope to finish execution. It stops waiting
1055 * when all threads finish, a subtask completes successfully, the deadline is
1056 * reached, or the current thread is {@linkplain Thread#interrupt() interrupted}.
1057 * It also stops waiting if the {@link #shutdown() shutdown} method is invoked
1058 * directly to shut down this task scope.
1059 *
1060 * <p> This method may only be invoked by the task scope owner.
1061 *
1062 * @throws IllegalStateException {@inheritDoc}
1063 * @throws WrongThreadException {@inheritDoc}
1064 */
1065 @Override
1066 public ShutdownOnSuccess<T> joinUntil(Instant deadline)
1067 throws InterruptedException, TimeoutException
1068 {
1069 super.joinUntil(deadline);
1070 return this;
1071 }
1072
1073 /**
1074 * {@return the result of the first subtask that completed {@linkplain
1075 * Subtask.State#SUCCESS successfully}}
1076 *
1077 * <p> When no subtask completed successfully, but a subtask {@linkplain
1078 * Subtask.State#FAILED failed} then {@code ExecutionException} is thrown with
1079 * the subtask's exception as the {@linkplain Throwable#getCause() cause}.
1080 *
1081 * @throws ExecutionException if no subtasks completed successfully but at least
1082 * one subtask failed
1083 * @throws IllegalStateException if no subtasks completed or the task scope owner
1084 * did not join after forking
1085 * @throws WrongThreadException if the current thread is not the task scope owner
1086 */
1087 public T result() throws ExecutionException {
1088 return result(ExecutionException::new);
1089 }
1090
1091 /**
1092 * Returns the result of the first subtask that completed {@linkplain
1093 * Subtask.State#SUCCESS successfully}, otherwise throws an exception produced
1094 * by the given exception supplying function.
1095 *
1096 * <p> When no subtask completed successfully, but a subtask {@linkplain
1097 * Subtask.State#FAILED failed}, then the exception supplying function is invoked
1098 * with subtask's exception.
1099 *
1100 * @param esf the exception supplying function
1101 * @param <X> type of the exception to be thrown
1102 * @return the result of the first subtask that completed with a result
1103 *
1104 * @throws X if no subtasks completed successfully but at least one subtask failed
1105 * @throws IllegalStateException if no subtasks completed or the task scope owner
1106 * did not join after forking
1107 * @throws WrongThreadException if the current thread is not the task scope owner
1108 */
1109 public <X extends Throwable> T result(Function<Throwable, ? extends X> esf) throws X {
1110 Objects.requireNonNull(esf);
1111 ensureOwnerAndJoined();
1112
1113 Object result = firstResult;
1114 if (result == RESULT_NULL) {
1115 return null;
1116 } else if (result != null) {
1117 @SuppressWarnings("unchecked")
1118 T r = (T) result;
1119 return r;
1120 }
1121
1122 Throwable exception = firstException;
1123 if (exception != null) {
1124 X ex = esf.apply(exception);
1125 Objects.requireNonNull(ex, "esf returned null");
1126 throw ex;
1127 }
1128
1129 throw new IllegalStateException("No completed subtasks");
1130 }
1131 }
1132
1133 /**
1134 * A {@code StructuredTaskScope} that captures the exception of the first subtask to
1135 * {@linkplain Subtask.State#FAILED fail}. Once captured, it {@linkplain #shutdown()
1136 * shuts down} the task scope to interrupt unfinished threads and wakeup the task
1137 * scope owner. The policy implemented by this class is intended for cases where the
1138 * results for all subtasks are required ("invoke all"); if any subtask fails then the
1139 * results of other unfinished subtasks are no longer needed.
1140 *
1141 * <p> Unless otherwise specified, passing a {@code null} argument to a method
1142 * in this class will cause a {@link NullPointerException} to be thrown.
1143 *
1144 * @apiNote This class implements a policy to shut down the task scope when a subtask
1145 * fails. There shouldn't be any need to directly shut down the task scope with the
1146 * {@link #shutdown() shutdown} method.
1147 *
1148 * @since 21
1149 */
1150 @PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY)
1151 public static final class ShutdownOnFailure extends StructuredTaskScope<Object> {
1152 private static final VarHandle FIRST_EXCEPTION;
1153 static {
1154 try {
1155 MethodHandles.Lookup l = MethodHandles.lookup();
1156 FIRST_EXCEPTION = l.findVarHandle(ShutdownOnFailure.class, "firstException", Throwable.class);
1157 } catch (Exception e) {
1158 throw new ExceptionInInitializerError(e);
1159 }
1160 }
1161 private volatile Throwable firstException;
1162
1163 /**
1164 * Constructs a new {@code ShutdownOnFailure} with the given name and thread factory.
1165 * The task scope is optionally named for the purposes of monitoring and management.
1166 * The thread factory is used to {@link ThreadFactory#newThread(Runnable) create}
1167 * threads when subtasks are {@linkplain #fork(Callable) forked}. The task scope
1168 * is owned by the current thread.
1169 *
1170 * <p> Construction captures the current thread's {@linkplain ScopedValue scoped
1171 * value} bindings for inheritance by threads started in the task scope. The
1172 * <a href="#TreeStructure">Tree Structure</a> section in the class description
1173 * details how parent-child relations are established implicitly for the purpose
1174 * of inheritance of scoped value bindings.
1175 *
1176 * @param name the name of the task scope, can be null
1177 * @param factory the thread factory
1178 */
1179 public ShutdownOnFailure(String name, ThreadFactory factory) {
1180 super(name, factory);
1181 }
1182
1183 /**
1184 * Constructs a new unnamed {@code ShutdownOnFailure} that creates virtual threads.
1185 *
1186 * @implSpec This constructor is equivalent to invoking the 2-arg constructor with
1187 * a name of {@code null} and a thread factory that creates virtual threads.
1188 */
1189 public ShutdownOnFailure() {
1190 this(null, Thread.ofVirtual().factory());
1191 }
1192
1193 @Override
1194 protected void handleComplete(Subtask<?> subtask) {
1195 if (subtask.state() == Subtask.State.FAILED
1196 && firstException == null
1197 && FIRST_EXCEPTION.compareAndSet(this, null, subtask.exception())) {
1198 super.shutdown();
1199 }
1200 }
1201
1202 /**
1203 * Wait for all subtasks started in this task scope to complete or for a subtask
1204 * to {@linkplain Subtask.State#FAILED fail}.
1205 *
1206 * <p> This method waits for all subtasks by waiting for all threads {@linkplain
1207 * #fork(Callable) started} in this task scope to finish execution. It stops waiting
1208 * when all threads finish, a subtask fails, or the current thread is {@linkplain
1209 * Thread#interrupt() interrupted}. It also stops waiting if the {@link #shutdown()
1210 * shutdown} method is invoked directly to shut down this task scope.
1211 *
1212 * <p> This method may only be invoked by the task scope owner.
1213 *
1214 * @throws IllegalStateException {@inheritDoc}
1215 * @throws WrongThreadException {@inheritDoc}
1216 */
1217 @Override
1218 public ShutdownOnFailure join() throws InterruptedException {
1219 super.join();
1220 return this;
1221 }
1222
1223 /**
1224 * Wait for all subtasks started in this task scope to complete or for a subtask
1225 * to {@linkplain Subtask.State#FAILED fail}, up to the given deadline.
1226 *
1227 * <p> This method waits for all subtasks by waiting for all threads {@linkplain
1228 * #fork(Callable) started} in this task scope to finish execution. It stops waiting
1229 * when all threads finish, a subtask fails, the deadline is reached, or the current
1230 * thread is {@linkplain Thread#interrupt() interrupted}. It also stops waiting
1231 * if the {@link #shutdown() shutdown} method is invoked directly to shut down
1232 * this task scope.
1233 *
1234 * <p> This method may only be invoked by the task scope owner.
1235 *
1236 * @throws IllegalStateException {@inheritDoc}
1237 * @throws WrongThreadException {@inheritDoc}
1238 */
1239 @Override
1240 public ShutdownOnFailure joinUntil(Instant deadline)
1241 throws InterruptedException, TimeoutException
1242 {
1243 super.joinUntil(deadline);
1244 return this;
1245 }
1246
1247 /**
1248 * Returns the exception of the first subtask that {@linkplain Subtask.State#FAILED
1249 * failed}. If no subtasks failed then an empty {@code Optional} is returned.
1250 *
1251 * @return the exception for the first subtask to fail or an empty optional if no
1252 * subtasks failed
1253 *
1254 * @throws WrongThreadException if the current thread is not the task scope owner
1255 * @throws IllegalStateException if the task scope owner did not join after forking
1256 */
1257 public Optional<Throwable> exception() {
1258 ensureOwnerAndJoined();
1259 return Optional.ofNullable(firstException);
1260 }
1261
1262 /**
1263 * Throws if a subtask {@linkplain Subtask.State#FAILED failed}.
1264 * If any subtask failed with an exception then {@code ExecutionException} is
1265 * thrown with the exception of the first subtask to fail as the {@linkplain
1266 * Throwable#getCause() cause}. This method does nothing if no subtasks failed.
1267 *
1268 * @throws ExecutionException if a subtask failed
1269 * @throws WrongThreadException if the current thread is not the task scope owner
1270 * @throws IllegalStateException if the task scope owner did not join after forking
1271 */
1272 public void throwIfFailed() throws ExecutionException {
1273 throwIfFailed(ExecutionException::new);
1274 }
1275
1276 /**
1277 * Throws the exception produced by the given exception supplying function if a
1278 * subtask {@linkplain Subtask.State#FAILED failed}. If any subtask failed with
1279 * an exception then the function is invoked with the exception of the first
1280 * subtask to fail. The exception returned by the function is thrown. This method
1281 * does nothing if no subtasks failed.
1282 *
1283 * @param esf the exception supplying function
1284 * @param <X> type of the exception to be thrown
1285 *
1286 * @throws X produced by the exception supplying function
1287 * @throws WrongThreadException if the current thread is not the task scope owner
1288 * @throws IllegalStateException if the task scope owner did not join after forking
1289 */
1290 public <X extends Throwable>
1291 void throwIfFailed(Function<Throwable, ? extends X> esf) throws X {
1292 ensureOwnerAndJoined();
1293 Objects.requireNonNull(esf);
1294 Throwable exception = firstException;
1295 if (exception != null) {
1296 X ex = esf.apply(exception);
1297 Objects.requireNonNull(ex, "esf returned null");
1298 throw ex;
1299 }
1300 }
1301 }
1302 }