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 }