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