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 @SuppressWarnings("this-escape") 406 public StructuredTaskScope(String name, ThreadFactory factory) { 407 this.factory = Objects.requireNonNull(factory, "'factory' is null"); 408 if (name == null) 409 name = Objects.toIdentityString(this); 410 this.flock = ThreadFlock.open(name); 411 } 412 413 /** 414 * Creates an unnamed structured task scope that creates virtual threads. The task 415 * scope is owned by the current thread. 416 * 417 * @implSpec This constructor is equivalent to invoking the 2-arg constructor with a 418 * name of {@code null} and a thread factory that creates virtual threads. 419 */ 420 public StructuredTaskScope() { 421 this(null, Thread.ofVirtual().factory()); 422 } 423 424 private IllegalStateException newIllegalStateExceptionScopeClosed() { 425 return new IllegalStateException("Task scope is closed"); 426 } 427 428 private IllegalStateException newIllegalStateExceptionNoJoin() { 429 return new IllegalStateException("Owner did not join after forking subtasks"); 430 } 431 432 /** 433 * Throws IllegalStateException if the scope is closed, returning the state if not 434 * closed. 435 */ 436 private int ensureOpen() { 437 int s = state; 438 if (s == CLOSED) 439 throw newIllegalStateExceptionScopeClosed(); 440 return s; 441 } 442 443 /** 444 * Throws WrongThreadException if the current thread is not the owner. 445 */ 446 private void ensureOwner() { 447 if (Thread.currentThread() != flock.owner()) 448 throw new WrongThreadException("Current thread not owner"); 449 } 450 451 /** 452 * Throws WrongThreadException if the current thread is not the owner 453 * or a thread contained in the tree. 454 */ 455 private void ensureOwnerOrContainsThread() { 456 Thread currentThread = Thread.currentThread(); 457 if (currentThread != flock.owner() && !flock.containsThread(currentThread)) 458 throw new WrongThreadException("Current thread not owner or thread in the tree"); 459 } 460 461 /** 462 * Ensures that the current thread is the owner of this task scope and that it joined 463 * (with {@link #join()} or {@link #joinUntil(Instant)}) after {@linkplain #fork(Callable) 464 * forking} subtasks. 465 * 466 * @apiNote This method can be used by subclasses that define methods to make available 467 * results, state, or other outcome to code intended to execute after the join method. 468 * 469 * @throws WrongThreadException if the current thread is not the task scope owner 470 * @throws IllegalStateException if the task scope is open and task scope owner did 471 * not join after forking 472 */ 473 protected final void ensureOwnerAndJoined() { 474 ensureOwner(); 475 if (forkRound > lastJoinCompleted) { 476 throw newIllegalStateExceptionNoJoin(); 477 } 478 } 479 480 /** 481 * Invoked by a subtask when it completes successfully or fails in this task scope. 482 * This method is not invoked if a subtask completes after the task scope is 483 * {@linkplain #shutdown() shut down}. 484 * 485 * @implSpec The default implementation throws {@code NullPointerException} if the 486 * subtask is {@code null}. It throws {@link IllegalArgumentException} if the subtask 487 * has not completed. 488 * 489 * @apiNote The {@code handleComplete} method should be thread safe. It may be 490 * invoked by several threads concurrently. 491 * 492 * @param subtask the subtask 493 * 494 * @throws IllegalArgumentException if called with a subtask that has not completed 495 */ 496 protected void handleComplete(Subtask<? extends T> subtask) { 497 if (subtask.state() == Subtask.State.UNAVAILABLE) 498 throw new IllegalArgumentException(); 499 } 500 501 /** 502 * Starts a new thread in this task scope to execute a value-returning task, thus 503 * creating a <em>subtask</em> of this task scope. 504 * 505 * <p> The value-returning task is provided to this method as a {@link Callable}, the 506 * thread executes the task's {@link Callable#call() call} method. The thread is 507 * created with the task scope's {@link ThreadFactory}. It inherits the current thread's 508 * {@linkplain ScopedValue scoped value} bindings. The bindings must match the bindings 509 * captured when the task scope was created. 510 * 511 * <p> This method returns a {@link Subtask Subtask} to represent the <em>forked 512 * subtask</em>. The {@code Subtask} object can be used to obtain the result when 513 * the subtask completes successfully, or the exception when the subtask fails. To 514 * ensure correct usage, the {@link Subtask#get() get()} and {@link Subtask#exception() 515 * exception()} methods may only be called by the task scope owner after it has waited 516 * for all threads to finish with the {@link #join() join} or {@link #joinUntil(Instant)} 517 * methods. When the subtask completes, the thread invokes the {@link 518 * #handleComplete(Subtask) handleComplete} method to consume the completed subtask. 519 * If the task scope is {@linkplain #shutdown() shut down} before the subtask completes 520 * then the {@code handleComplete} method will not be invoked. 521 * 522 * <p> If this task scope is {@linkplain #shutdown() shutdown} (or in the process of 523 * shutting down) then the subtask will not run and the {@code handleComplete} method 524 * will not be invoked. 525 * 526 * <p> This method may only be invoked by the task scope owner or threads contained 527 * in the task scope. 528 * 529 * @implSpec This method may be overridden for customization purposes, wrapping tasks 530 * for example. If overridden, the subclass must invoke {@code super.fork} to start a 531 * new thread in this task scope. 532 * 533 * @param task the value-returning task for the thread to execute 534 * @param <U> the result type 535 * @return the subtask 536 * @throws IllegalStateException if this task scope is closed 537 * @throws WrongThreadException if the current thread is not the task scope owner or a 538 * thread contained in the task scope 539 * @throws StructureViolationException if the current scoped value bindings are not 540 * the same as when the task scope was created 541 * @throws RejectedExecutionException if the thread factory rejected creating a 542 * thread to run the subtask 543 */ 544 public <U extends T> Subtask<U> fork(Callable<? extends U> task) { 545 Objects.requireNonNull(task, "'task' is null"); 546 int s = ensureOpen(); // throws ISE if closed 547 548 // when forked by the owner, the subtask is forked in the current or next round 549 int round = -1; 550 if (Thread.currentThread() == flock.owner()) { 551 round = forkRound; 552 if (forkRound == lastJoinCompleted) { 553 // new round if first fork after join 554 round++; 555 } 556 } 557 558 var subtask = new SubtaskImpl<U>(this, task); 559 if (s < SHUTDOWN) { 560 // create thread to run task 561 Thread thread = factory.newThread(subtask); 562 if (thread == null) { 563 throw new RejectedExecutionException("Rejected by thread factory"); 564 } 565 566 // attempt to start the thread 567 try { 568 flock.start(thread); 569 } catch (IllegalStateException e) { 570 // shutdown by another thread, or underlying flock is shutdown due 571 // to unstructured use 572 } 573 } 574 575 // force owner to join if this is the first fork in the round 576 if (Thread.currentThread() == flock.owner() && round > forkRound) { 577 forkRound = round; 578 } 579 580 // return forked subtask or a subtask that did not run 581 return subtask; 582 } 583 584 /** 585 * Wait for all threads to finish or the task scope to shut down. 586 */ 587 private void implJoin(Duration timeout) 588 throws InterruptedException, TimeoutException 589 { 590 ensureOwner(); 591 lastJoinAttempted = forkRound; 592 int s = ensureOpen(); // throws ISE if closed 593 if (s == OPEN) { 594 // wait for all threads, wakeup, interrupt, or timeout 595 if (timeout != null) { 596 flock.awaitAll(timeout); 597 } else { 598 flock.awaitAll(); 599 } 600 } 601 lastJoinCompleted = forkRound; 602 } 603 604 /** 605 * Wait for all subtasks started in this task scope to finish or the task scope to 606 * shut down. 607 * 608 * <p> This method waits for all subtasks by waiting for all threads {@linkplain 609 * #fork(Callable) started} in this task scope to finish execution. It stops waiting 610 * when all threads finish, the task scope is {@linkplain #shutdown() shut down}, or 611 * the current thread is {@linkplain Thread#interrupt() interrupted}. 612 * 613 * <p> This method may only be invoked by the task scope owner. 614 * 615 * @implSpec This method may be overridden for customization purposes or to return a 616 * more specific return type. If overridden, the subclass must invoke {@code 617 * super.join} to ensure that the method waits for threads in this task scope to 618 * finish. 619 * 620 * @return this task scope 621 * @throws IllegalStateException if this task scope is closed 622 * @throws WrongThreadException if the current thread is not the task scope owner 623 * @throws InterruptedException if interrupted while waiting 624 */ 625 public StructuredTaskScope<T> join() throws InterruptedException { 626 try { 627 implJoin(null); 628 } catch (TimeoutException e) { 629 throw new InternalError(); 630 } 631 return this; 632 } 633 634 /** 635 * Wait for all subtasks started in this task scope to finish or the task scope to 636 * shut down, up to the given deadline. 637 * 638 * <p> This method waits for all subtasks by waiting for all threads {@linkplain 639 * #fork(Callable) started} in this task scope to finish execution. It stops waiting 640 * when all threads finish, the task scope is {@linkplain #shutdown() shut down}, the 641 * deadline is reached, or the current thread is {@linkplain Thread#interrupt() 642 * interrupted}. 643 * 644 * <p> This method may only be invoked by the task scope owner. 645 * 646 * @implSpec This method may be overridden for customization purposes or to return a 647 * more specific return type. If overridden, the subclass must invoke {@code 648 * super.joinUntil} to ensure that the method waits for threads in this task scope to 649 * finish. 650 * 651 * @param deadline the deadline 652 * @return this task scope 653 * @throws IllegalStateException if this task scope is closed 654 * @throws WrongThreadException if the current thread is not the task scope owner 655 * @throws InterruptedException if interrupted while waiting 656 * @throws TimeoutException if the deadline is reached while waiting 657 */ 658 public StructuredTaskScope<T> joinUntil(Instant deadline) 659 throws InterruptedException, TimeoutException 660 { 661 Duration timeout = Duration.between(Instant.now(), deadline); 662 implJoin(timeout); 663 return this; 664 } 665 666 /** 667 * Interrupt all unfinished threads. 668 */ 669 private void implInterruptAll() { 670 flock.threads() 671 .filter(t -> t != Thread.currentThread()) 672 .forEach(t -> { 673 try { 674 t.interrupt(); 675 } catch (Throwable ignore) { } 676 }); 677 } 678 679 @SuppressWarnings("removal") 680 private void interruptAll() { 681 if (System.getSecurityManager() == null) { 682 implInterruptAll(); 683 } else { 684 PrivilegedAction<Void> pa = () -> { 685 implInterruptAll(); 686 return null; 687 }; 688 AccessController.doPrivileged(pa); 689 } 690 } 691 692 /** 693 * Shutdown the task scope if not already shutdown. Return true if this method 694 * shutdowns the task scope, false if already shutdown. 695 */ 696 private boolean implShutdown() { 697 shutdownLock.lock(); 698 try { 699 if (state < SHUTDOWN) { 700 // prevent new threads from starting 701 flock.shutdown(); 702 703 // set status before interrupting tasks 704 state = SHUTDOWN; 705 706 // interrupt all unfinished threads 707 interruptAll(); 708 709 return true; 710 } else { 711 // already shutdown 712 return false; 713 } 714 } finally { 715 shutdownLock.unlock(); 716 } 717 } 718 719 /** 720 * Shut down this task scope without closing it. Shutting down a task scope prevents 721 * new threads from starting, interrupts all unfinished threads, and causes the 722 * {@link #join() join} method to wakeup. Shutdown is useful for cases where the 723 * results of unfinished subtasks are no longer needed. It will typically be called 724 * by the {@link #handleComplete(Subtask)} implementation of a subclass that 725 * implements a policy to discard unfinished tasks once some outcome is reached. 726 * 727 * <p> More specifically, this method: 728 * <ul> 729 * <li> {@linkplain Thread#interrupt() Interrupts} all unfinished threads in the 730 * task scope (except the current thread). 731 * <li> Wakes up the task scope owner if it is waiting in {@link #join()} or {@link 732 * #joinUntil(Instant)}. If the task scope owner is not waiting then its next call to 733 * {@code join} or {@code joinUntil} will return immediately. 734 * </ul> 735 * 736 * <p> The {@linkplain Subtask.State state} of unfinished subtasks that complete at 737 * around the time that the task scope is shutdown is not defined. A subtask that 738 * completes successfully with a result, or fails with an exception, at around 739 * the time that the task scope is shutdown may or may not <i>transition</i> to a 740 * terminal state. 741 * 742 * <p> This method may only be invoked by the task scope owner or threads contained 743 * in the task scope. 744 * 745 * @implSpec This method may be overridden for customization purposes. If overridden, 746 * the subclass must invoke {@code super.shutdown} to ensure that the method shuts 747 * down the task scope. 748 * 749 * @apiNote 750 * There may be threads that have not finished because they are executing code that 751 * did not respond (or respond promptly) to thread interrupt. This method does not wait 752 * for these threads. When the owner invokes the {@link #close() close} method 753 * to close the task scope then it will wait for the remaining threads to finish. 754 * 755 * @throws IllegalStateException if this task scope is closed 756 * @throws WrongThreadException if the current thread is not the task scope owner or 757 * a thread contained in the task scope 758 * @see #isShutdown() 759 */ 760 public void shutdown() { 761 ensureOwnerOrContainsThread(); 762 int s = ensureOpen(); // throws ISE if closed 763 if (s < SHUTDOWN && implShutdown()) 764 flock.wakeup(); 765 } 766 767 /** 768 * {@return true if this task scope is shutdown, otherwise false} 769 * @see #shutdown() 770 */ 771 public final boolean isShutdown() { 772 return state >= SHUTDOWN; 773 } 774 775 /** 776 * Closes this task scope. 777 * 778 * <p> This method first shuts down the task scope (as if by invoking the {@link 779 * #shutdown() shutdown} method). It then waits for the threads executing any 780 * unfinished tasks to finish. If interrupted, this method will continue to wait for 781 * the threads to finish before completing with the interrupt status set. 782 * 783 * <p> This method may only be invoked by the task scope owner. If the task scope 784 * is already closed then the task scope owner invoking this method has no effect. 785 * 786 * <p> A {@code StructuredTaskScope} is intended to be used in a <em>structured 787 * manner</em>. If this method is called to close a task scope before nested task 788 * scopes are closed then it closes the underlying construct of each nested task scope 789 * (in the reverse order that they were created in), closes this task scope, and then 790 * throws {@link StructureViolationException}. 791 * Similarly, if this method is called to close a task scope while executing with 792 * {@linkplain ScopedValue scoped value} bindings, and the task scope was created 793 * before the scoped values were bound, then {@code StructureViolationException} is 794 * thrown after closing the task scope. 795 * If a thread terminates without first closing task scopes that it owns then 796 * termination will cause the underlying construct of each of its open tasks scopes to 797 * be closed. Closing is performed in the reverse order that the task scopes were 798 * created in. Thread termination may therefore be delayed when the task scope owner 799 * has to wait for threads forked in these task scopes to finish. 800 * 801 * @implSpec This method may be overridden for customization purposes. If overridden, 802 * the subclass must invoke {@code super.close} to close the task scope. 803 * 804 * @throws IllegalStateException thrown after closing the task scope if the task scope 805 * owner did not attempt to join after forking 806 * @throws WrongThreadException if the current thread is not the task scope owner 807 * @throws StructureViolationException if a structure violation was detected 808 */ 809 @Override 810 public void close() { 811 ensureOwner(); 812 int s = state; 813 if (s == CLOSED) 814 return; 815 816 try { 817 if (s < SHUTDOWN) 818 implShutdown(); 819 flock.close(); 820 } finally { 821 state = CLOSED; 822 } 823 824 // throw ISE if the owner didn't attempt to join after forking 825 if (forkRound > lastJoinAttempted) { 826 lastJoinCompleted = forkRound; // ensureOwnerAndJoined is a no-op after close 827 throw newIllegalStateExceptionNoJoin(); 828 } 829 } 830 831 @Override 832 public String toString() { 833 String name = flock.name(); 834 return switch (state) { 835 case OPEN -> name; 836 case SHUTDOWN -> name + "/shutdown"; 837 case CLOSED -> name + "/closed"; 838 default -> throw new InternalError(); 839 }; 840 } 841 842 /** 843 * Subtask implementation, runs the task specified to the fork method. 844 */ 845 private static final class SubtaskImpl<T> implements Subtask<T>, Runnable { 846 private static final AltResult RESULT_NULL = new AltResult(Subtask.State.SUCCESS); 847 848 private record AltResult(Subtask.State state, Throwable exception) { 849 AltResult(Subtask.State state) { 850 this(state, null); 851 } 852 } 853 854 private final StructuredTaskScope<? super T> scope; 855 private final Callable<? extends T> task; 856 private volatile Object result; 857 858 SubtaskImpl(StructuredTaskScope<? super T> scope, Callable<? extends T> task) { 859 this.scope = scope; 860 this.task = task; 861 } 862 863 @Override 864 public void run() { 865 T result = null; 866 Throwable ex = null; 867 try { 868 result = task.call(); 869 } catch (Throwable e) { 870 ex = e; 871 } 872 873 // nothing to do if task scope is shutdown 874 if (scope.isShutdown()) 875 return; 876 877 // capture result or exception, invoke handleComplete 878 if (ex == null) { 879 this.result = (result != null) ? result : RESULT_NULL; 880 } else { 881 this.result = new AltResult(State.FAILED, ex); 882 } 883 scope.handleComplete(this); 884 } 885 886 @Override 887 public Callable<? extends T> task() { 888 return task; 889 } 890 891 @Override 892 public Subtask.State state() { 893 Object result = this.result; 894 if (result == null) { 895 return State.UNAVAILABLE; 896 } else if (result instanceof AltResult alt) { 897 // null or failed 898 return alt.state(); 899 } else { 900 return State.SUCCESS; 901 } 902 } 903 904 @Override 905 public T get() { 906 Object result = this.result; 907 if (result instanceof AltResult) { 908 if (result == RESULT_NULL) return null; 909 } else if (result != null) { 910 @SuppressWarnings("unchecked") 911 T r = (T) result; 912 return r; 913 } 914 throw new IllegalStateException( 915 "Result is unavailable or subtask did not complete successfully"); 916 } 917 918 @Override 919 public Throwable exception() { 920 Object result = this.result; 921 if (result instanceof AltResult alt && alt.state() == State.FAILED) { 922 return alt.exception(); 923 } 924 throw new IllegalStateException( 925 "Exception is unavailable or subtask did not complete with exception"); 926 } 927 928 @Override 929 public String toString() { 930 String stateAsString = switch (state()) { 931 case UNAVAILABLE -> "[Unavailable]"; 932 case SUCCESS -> "[Completed successfully]"; 933 case FAILED -> { 934 Throwable ex = ((AltResult) result).exception(); 935 yield "[Failed: " + ex + "]"; 936 } 937 }; 938 return Objects.toIdentityString(this) + stateAsString; 939 } 940 } 941 942 /** 943 * A {@code StructuredTaskScope} that captures the result of the first subtask to 944 * complete {@linkplain Subtask.State#SUCCESS successfully}. Once captured, it 945 * {@linkplain #shutdown() shuts down} the task scope to interrupt unfinished threads 946 * and wakeup the task scope owner. The policy implemented by this class is intended 947 * for cases where the result of any subtask will do ("invoke any") and where the 948 * results of other unfinished subtasks are no longer needed. 949 * 950 * <p> Unless otherwise specified, passing a {@code null} argument to a method 951 * in this class will cause a {@link NullPointerException} to be thrown. 952 * 953 * @apiNote This class implements a policy to shut down the task scope when a subtask 954 * completes successfully. There shouldn't be any need to directly shut down the task 955 * scope with the {@link #shutdown() shutdown} method. 956 * 957 * @param <T> the result type 958 * @since 21 959 */ 960 @PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY) 961 public static final class ShutdownOnSuccess<T> extends StructuredTaskScope<T> { 962 private static final Object RESULT_NULL = new Object(); 963 private static final VarHandle FIRST_RESULT; 964 private static final VarHandle FIRST_EXCEPTION; 965 static { 966 try { 967 MethodHandles.Lookup l = MethodHandles.lookup(); 968 FIRST_RESULT = l.findVarHandle(ShutdownOnSuccess.class, "firstResult", Object.class); 969 FIRST_EXCEPTION = l.findVarHandle(ShutdownOnSuccess.class, "firstException", Throwable.class); 970 } catch (Exception e) { 971 throw new ExceptionInInitializerError(e); 972 } 973 } 974 private volatile Object firstResult; 975 private volatile Throwable firstException; 976 977 /** 978 * Constructs a new {@code ShutdownOnSuccess} with the given name and thread factory. 979 * The task scope is optionally named for the purposes of monitoring and management. 980 * The thread factory is used to {@link ThreadFactory#newThread(Runnable) create} 981 * threads when subtasks are {@linkplain #fork(Callable) forked}. The task scope 982 * is owned by the current thread. 983 * 984 * <p> Construction captures the current thread's {@linkplain ScopedValue scoped 985 * value} bindings for inheritance by threads started in the task scope. The 986 * <a href="#TreeStructure">Tree Structure</a> section in the class description 987 * details how parent-child relations are established implicitly for the purpose 988 * of inheritance of scoped value bindings. 989 * 990 * @param name the name of the task scope, can be null 991 * @param factory the thread factory 992 */ 993 public ShutdownOnSuccess(String name, ThreadFactory factory) { 994 super(name, factory); 995 } 996 997 /** 998 * Constructs a new unnamed {@code ShutdownOnSuccess} that creates virtual threads. 999 * 1000 * @implSpec This constructor is equivalent to invoking the 2-arg constructor with 1001 * a name of {@code null} and a thread factory that creates virtual threads. 1002 */ 1003 public ShutdownOnSuccess() { 1004 this(null, Thread.ofVirtual().factory()); 1005 } 1006 1007 @Override 1008 protected void handleComplete(Subtask<? extends T> subtask) { 1009 if (firstResult != null) { 1010 // already captured a result 1011 return; 1012 } 1013 1014 if (subtask.state() == Subtask.State.SUCCESS) { 1015 // task succeeded 1016 T result = subtask.get(); 1017 Object r = (result != null) ? result : RESULT_NULL; 1018 if (FIRST_RESULT.compareAndSet(this, null, r)) { 1019 super.shutdown(); 1020 } 1021 } else if (firstException == null) { 1022 // capture the exception thrown by the first subtask that failed 1023 FIRST_EXCEPTION.compareAndSet(this, null, subtask.exception()); 1024 } 1025 } 1026 1027 /** 1028 * Wait for a subtask started in this task scope to complete {@linkplain 1029 * Subtask.State#SUCCESS successfully} or all subtasks to complete. 1030 * 1031 * <p> This method waits for all subtasks by waiting for all threads {@linkplain 1032 * #fork(Callable) started} in this task scope to finish execution. It stops waiting 1033 * when all threads finish, a subtask completes successfully, or the current 1034 * thread is {@linkplain Thread#interrupt() interrupted}. It also stops waiting 1035 * if the {@link #shutdown() shutdown} method is invoked directly to shut down 1036 * this task scope. 1037 * 1038 * <p> This method may only be invoked by the task scope owner. 1039 * 1040 * @throws IllegalStateException {@inheritDoc} 1041 * @throws WrongThreadException {@inheritDoc} 1042 */ 1043 @Override 1044 public ShutdownOnSuccess<T> join() throws InterruptedException { 1045 super.join(); 1046 return this; 1047 } 1048 1049 /** 1050 * Wait for a subtask started in this task scope to complete {@linkplain 1051 * Subtask.State#SUCCESS successfully} or all subtasks to complete, up to the 1052 * given deadline. 1053 * 1054 * <p> This method waits for all subtasks by waiting for all threads {@linkplain 1055 * #fork(Callable) started} in this task scope to finish execution. It stops waiting 1056 * when all threads finish, a subtask completes successfully, the deadline is 1057 * reached, or the current thread is {@linkplain Thread#interrupt() interrupted}. 1058 * It also stops waiting if the {@link #shutdown() shutdown} method is invoked 1059 * directly to shut down this task scope. 1060 * 1061 * <p> This method may only be invoked by the task scope owner. 1062 * 1063 * @throws IllegalStateException {@inheritDoc} 1064 * @throws WrongThreadException {@inheritDoc} 1065 */ 1066 @Override 1067 public ShutdownOnSuccess<T> joinUntil(Instant deadline) 1068 throws InterruptedException, TimeoutException 1069 { 1070 super.joinUntil(deadline); 1071 return this; 1072 } 1073 1074 /** 1075 * {@return the result of the first subtask that completed {@linkplain 1076 * Subtask.State#SUCCESS successfully}} 1077 * 1078 * <p> When no subtask completed successfully, but a subtask {@linkplain 1079 * Subtask.State#FAILED failed} then {@code ExecutionException} is thrown with 1080 * the subtask's exception as the {@linkplain Throwable#getCause() cause}. 1081 * 1082 * @throws ExecutionException if no subtasks completed successfully but at least 1083 * one subtask failed 1084 * @throws IllegalStateException if no subtasks completed or the task scope owner 1085 * did not join after forking 1086 * @throws WrongThreadException if the current thread is not the task scope owner 1087 */ 1088 public T result() throws ExecutionException { 1089 return result(ExecutionException::new); 1090 } 1091 1092 /** 1093 * Returns the result of the first subtask that completed {@linkplain 1094 * Subtask.State#SUCCESS successfully}, otherwise throws an exception produced 1095 * by the given exception supplying function. 1096 * 1097 * <p> When no subtask completed successfully, but a subtask {@linkplain 1098 * Subtask.State#FAILED failed}, then the exception supplying function is invoked 1099 * with subtask's exception. 1100 * 1101 * @param esf the exception supplying function 1102 * @param <X> type of the exception to be thrown 1103 * @return the result of the first subtask that completed with a result 1104 * 1105 * @throws X if no subtasks completed successfully but at least one subtask failed 1106 * @throws IllegalStateException if no subtasks completed or the task scope owner 1107 * did not join after forking 1108 * @throws WrongThreadException if the current thread is not the task scope owner 1109 */ 1110 public <X extends Throwable> T result(Function<Throwable, ? extends X> esf) throws X { 1111 Objects.requireNonNull(esf); 1112 ensureOwnerAndJoined(); 1113 1114 Object result = firstResult; 1115 if (result == RESULT_NULL) { 1116 return null; 1117 } else if (result != null) { 1118 @SuppressWarnings("unchecked") 1119 T r = (T) result; 1120 return r; 1121 } 1122 1123 Throwable exception = firstException; 1124 if (exception != null) { 1125 X ex = esf.apply(exception); 1126 Objects.requireNonNull(ex, "esf returned null"); 1127 throw ex; 1128 } 1129 1130 throw new IllegalStateException("No completed subtasks"); 1131 } 1132 } 1133 1134 /** 1135 * A {@code StructuredTaskScope} that captures the exception of the first subtask to 1136 * {@linkplain Subtask.State#FAILED fail}. Once captured, it {@linkplain #shutdown() 1137 * shuts down} the task scope to interrupt unfinished threads and wakeup the task 1138 * scope owner. The policy implemented by this class is intended for cases where the 1139 * results for all subtasks are required ("invoke all"); if any subtask fails then the 1140 * results of other unfinished subtasks are no longer needed. 1141 * 1142 * <p> Unless otherwise specified, passing a {@code null} argument to a method 1143 * in this class will cause a {@link NullPointerException} to be thrown. 1144 * 1145 * @apiNote This class implements a policy to shut down the task scope when a subtask 1146 * fails. There shouldn't be any need to directly shut down the task scope with the 1147 * {@link #shutdown() shutdown} method. 1148 * 1149 * @since 21 1150 */ 1151 @PreviewFeature(feature = PreviewFeature.Feature.STRUCTURED_CONCURRENCY) 1152 public static final class ShutdownOnFailure extends StructuredTaskScope<Object> { 1153 private static final VarHandle FIRST_EXCEPTION; 1154 static { 1155 try { 1156 MethodHandles.Lookup l = MethodHandles.lookup(); 1157 FIRST_EXCEPTION = l.findVarHandle(ShutdownOnFailure.class, "firstException", Throwable.class); 1158 } catch (Exception e) { 1159 throw new ExceptionInInitializerError(e); 1160 } 1161 } 1162 private volatile Throwable firstException; 1163 1164 /** 1165 * Constructs a new {@code ShutdownOnFailure} with the given name and thread factory. 1166 * The task scope is optionally named for the purposes of monitoring and management. 1167 * The thread factory is used to {@link ThreadFactory#newThread(Runnable) create} 1168 * threads when subtasks are {@linkplain #fork(Callable) forked}. The task scope 1169 * is owned by the current thread. 1170 * 1171 * <p> Construction captures the current thread's {@linkplain ScopedValue scoped 1172 * value} bindings for inheritance by threads started in the task scope. The 1173 * <a href="#TreeStructure">Tree Structure</a> section in the class description 1174 * details how parent-child relations are established implicitly for the purpose 1175 * of inheritance of scoped value bindings. 1176 * 1177 * @param name the name of the task scope, can be null 1178 * @param factory the thread factory 1179 */ 1180 public ShutdownOnFailure(String name, ThreadFactory factory) { 1181 super(name, factory); 1182 } 1183 1184 /** 1185 * Constructs a new unnamed {@code ShutdownOnFailure} that creates virtual threads. 1186 * 1187 * @implSpec This constructor is equivalent to invoking the 2-arg constructor with 1188 * a name of {@code null} and a thread factory that creates virtual threads. 1189 */ 1190 public ShutdownOnFailure() { 1191 this(null, Thread.ofVirtual().factory()); 1192 } 1193 1194 @Override 1195 protected void handleComplete(Subtask<?> subtask) { 1196 if (subtask.state() == Subtask.State.FAILED 1197 && firstException == null 1198 && FIRST_EXCEPTION.compareAndSet(this, null, subtask.exception())) { 1199 super.shutdown(); 1200 } 1201 } 1202 1203 /** 1204 * Wait for all subtasks started in this task scope to complete or for a subtask 1205 * to {@linkplain Subtask.State#FAILED fail}. 1206 * 1207 * <p> This method waits for all subtasks by waiting for all threads {@linkplain 1208 * #fork(Callable) started} in this task scope to finish execution. It stops waiting 1209 * when all threads finish, a subtask fails, or the current thread is {@linkplain 1210 * Thread#interrupt() interrupted}. It also stops waiting if the {@link #shutdown() 1211 * shutdown} method is invoked directly to shut down this task scope. 1212 * 1213 * <p> This method may only be invoked by the task scope owner. 1214 * 1215 * @throws IllegalStateException {@inheritDoc} 1216 * @throws WrongThreadException {@inheritDoc} 1217 */ 1218 @Override 1219 public ShutdownOnFailure join() throws InterruptedException { 1220 super.join(); 1221 return this; 1222 } 1223 1224 /** 1225 * Wait for all subtasks started in this task scope to complete or for a subtask 1226 * to {@linkplain Subtask.State#FAILED fail}, up to the given deadline. 1227 * 1228 * <p> This method waits for all subtasks by waiting for all threads {@linkplain 1229 * #fork(Callable) started} in this task scope to finish execution. It stops waiting 1230 * when all threads finish, a subtask fails, the deadline is reached, or the current 1231 * thread is {@linkplain Thread#interrupt() interrupted}. It also stops waiting 1232 * if the {@link #shutdown() shutdown} method is invoked directly to shut down 1233 * this task scope. 1234 * 1235 * <p> This method may only be invoked by the task scope owner. 1236 * 1237 * @throws IllegalStateException {@inheritDoc} 1238 * @throws WrongThreadException {@inheritDoc} 1239 */ 1240 @Override 1241 public ShutdownOnFailure joinUntil(Instant deadline) 1242 throws InterruptedException, TimeoutException 1243 { 1244 super.joinUntil(deadline); 1245 return this; 1246 } 1247 1248 /** 1249 * Returns the exception of the first subtask that {@linkplain Subtask.State#FAILED 1250 * failed}. If no subtasks failed then an empty {@code Optional} is returned. 1251 * 1252 * @return the exception for the first subtask to fail or an empty optional if no 1253 * subtasks failed 1254 * 1255 * @throws WrongThreadException if the current thread is not the task scope owner 1256 * @throws IllegalStateException if the task scope owner did not join after forking 1257 */ 1258 public Optional<Throwable> exception() { 1259 ensureOwnerAndJoined(); 1260 return Optional.ofNullable(firstException); 1261 } 1262 1263 /** 1264 * Throws if a subtask {@linkplain Subtask.State#FAILED failed}. 1265 * If any subtask failed with an exception then {@code ExecutionException} is 1266 * thrown with the exception of the first subtask to fail as the {@linkplain 1267 * Throwable#getCause() cause}. This method does nothing if no subtasks failed. 1268 * 1269 * @throws ExecutionException if a subtask failed 1270 * @throws WrongThreadException if the current thread is not the task scope owner 1271 * @throws IllegalStateException if the task scope owner did not join after forking 1272 */ 1273 public void throwIfFailed() throws ExecutionException { 1274 throwIfFailed(ExecutionException::new); 1275 } 1276 1277 /** 1278 * Throws the exception produced by the given exception supplying function if a 1279 * subtask {@linkplain Subtask.State#FAILED failed}. If any subtask failed with 1280 * an exception then the function is invoked with the exception of the first 1281 * subtask to fail. The exception returned by the function is thrown. This method 1282 * does nothing if no subtasks failed. 1283 * 1284 * @param esf the exception supplying function 1285 * @param <X> type of the exception to be thrown 1286 * 1287 * @throws X produced by the exception supplying function 1288 * @throws WrongThreadException if the current thread is not the task scope owner 1289 * @throws IllegalStateException if the task scope owner did not join after forking 1290 */ 1291 public <X extends Throwable> 1292 void throwIfFailed(Function<Throwable, ? extends X> esf) throws X { 1293 ensureOwnerAndJoined(); 1294 Objects.requireNonNull(esf); 1295 Throwable exception = firstException; 1296 if (exception != null) { 1297 X ex = esf.apply(exception); 1298 Objects.requireNonNull(ex, "esf returned null"); 1299 throw ex; 1300 } 1301 } 1302 } 1303 }