1 /* 2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. 3 * 4 * This code is free software; you can redistribute it and/or modify it 5 * under the terms of the GNU General Public License version 2 only, as 6 * published by the Free Software Foundation. Oracle designates this 7 * particular file as subject to the "Classpath" exception as provided 8 * by Oracle in the LICENSE file that accompanied this code. 9 * 10 * This code is distributed in the hope that it will be useful, but WITHOUT 11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 13 * version 2 for more details (a copy is included in the LICENSE file that 14 * accompanied this code). 15 * 16 * You should have received a copy of the GNU General Public License version 17 * 2 along with this work; if not, write to the Free Software Foundation, 18 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. 19 * 20 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA 21 * or visit www.oracle.com if you need additional information or have any 22 * questions. 23 */ 24 25 /* 26 * This file is available under and governed by the GNU General Public 27 * License version 2 only, as published by the Free Software Foundation. 28 * However, the following notice accompanied the original version of this 29 * file: 30 * 31 * Written by Doug Lea with assistance from members of JCP JSR-166 32 * Expert Group and released to the public domain, as explained at 33 * http://creativecommons.org/publicdomain/zero/1.0/ 34 */ 35 36 package java.util.concurrent; 37 38 import java.util.concurrent.locks.LockSupport; 39 import java.util.function.BiConsumer; 40 import java.util.function.BiFunction; 41 import java.util.function.Consumer; 42 import java.util.function.Function; 43 import java.util.function.Supplier; 44 45 /** 46 * A {@link Future} that may be explicitly completed (setting its 47 * value and status), and may be used as a {@link CompletionStage}, 48 * supporting dependent functions and actions that trigger upon its 49 * completion. 50 * 51 * <p>When two or more threads attempt to 52 * {@link #complete complete}, 53 * {@link #completeExceptionally completeExceptionally}, or 54 * {@link #cancel cancel} 55 * a CompletableFuture, only one of them succeeds. 56 * 57 * <p>In addition to these and related methods for directly 58 * manipulating status and results, CompletableFuture implements 59 * interface {@link CompletionStage} with the following policies: <ul> 60 * 61 * <li>Actions supplied for dependent completions of 62 * <em>non-async</em> methods may be performed by the thread that 63 * completes the current CompletableFuture, or by any other caller of 64 * a completion method. 65 * 66 * <li>All <em>async</em> methods without an explicit Executor 67 * argument are performed using the {@link ForkJoinPool#commonPool()} 68 * (unless it does not support a parallelism level of at least two, in 69 * which case, a new Thread is created to run each task). 70 * To simplify monitoring, debugging, 71 * and tracking, all generated asynchronous tasks are instances of the 72 * marker interface {@link AsynchronousCompletionTask}. Operations 73 * with time-delays can use adapter methods defined in this class, for 74 * example: {@code supplyAsync(supplier, delayedExecutor(timeout, 75 * timeUnit))}. To support methods with delays and timeouts, this 76 * class maintains at most one daemon thread for triggering and 77 * cancelling actions, not for running them. 78 * 79 * <li>All CompletionStage methods are implemented independently of 80 * other public methods, so the behavior of one method is not impacted 81 * by overrides of others in subclasses. 82 * 83 * </ul> 84 * 85 * <p>CompletableFuture also implements {@link Future} with the following 86 * policies: <ul> 87 * 88 * <li>Since (unlike {@link FutureTask}) this class has no direct 89 * control over the computation that causes it to be completed, 90 * cancellation is treated as just another form of exceptional 91 * completion. Method {@link #cancel cancel} has the same effect as 92 * {@code completeExceptionally(new CancellationException())}. Method 93 * {@link #isCompletedExceptionally} can be used to determine if a 94 * CompletableFuture completed in any exceptional fashion. 95 * 96 * <li>In case of exceptional completion with a CompletionException, 97 * methods {@link #get()} and {@link #get(long, TimeUnit)} throw an 98 * {@link ExecutionException} with the same cause as held in the 99 * corresponding CompletionException. To simplify usage in most 100 * contexts, this class also defines methods {@link #join()} and 101 * {@link #getNow} that instead throw the CompletionException directly 102 * in these cases. 103 * </ul> 104 * 105 * <p>Arguments used to pass a completion result (that is, for 106 * parameters of type {@code T}) for methods accepting them may be 107 * null, but passing a null value for any other parameter will result 108 * in a {@link NullPointerException} being thrown. 109 * 110 * @author Doug Lea 111 * @since 1.8 112 * @param <T> The result type returned by this future's {@code join} 113 * and {@code get} methods 114 */ 115 public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { 116 117 /* 118 * Overview: 119 * 120 * A CompletableFuture may have dependent completion actions, 121 * collected in a linked stack. It atomically completes by CASing 122 * a result field, and then pops off and runs those actions. This 123 * applies across normal vs exceptional outcomes, sync vs async 124 * actions, binary triggers, and various forms of completions. 125 * 126 * Non-nullness of field result (set via CAS) indicates done. An 127 * AltResult is used to box null as a result, as well as to hold 128 * exceptions. Using a single field makes completion simple to 129 * detect and trigger. Encoding and decoding is straightforward 130 * but adds to the sprawl of trapping and associating exceptions 131 * with targets. Minor simplifications rely on (static) NIL (to 132 * box null results) being the only AltResult with a null 133 * exception field, so we don't usually need explicit comparisons. 134 * Even though some of the generics casts are unchecked (see 135 * SuppressWarnings annotations), they are placed to be 136 * appropriate even if checked. 137 * 138 * Dependent actions are represented by Completion objects linked 139 * as Treiber stacks headed by field "stack". There are Completion 140 * classes for each kind of action, grouped into single-input 141 * (UniCompletion), two-input (BiCompletion), projected 142 * (BiCompletions using either (not both) of two inputs), shared 143 * (CoCompletion, used by the second of two sources), zero-input 144 * source actions, and Signallers that unblock waiters. Class 145 * Completion extends ForkJoinTask to enable async execution 146 * (adding no space overhead because we exploit its "tag" methods 147 * to maintain claims). It is also declared as Runnable to allow 148 * usage with arbitrary executors. 149 * 150 * Support for each kind of CompletionStage relies on a separate 151 * class, along with two CompletableFuture methods: 152 * 153 * * A Completion class with name X corresponding to function, 154 * prefaced with "Uni", "Bi", or "Or". Each class contains 155 * fields for source(s), actions, and dependent. They are 156 * boringly similar, differing from others only with respect to 157 * underlying functional forms. We do this so that users don't 158 * encounter layers of adapters in common usages. 159 * 160 * * Boolean CompletableFuture method x(...) (for example 161 * uniApply) takes all of the arguments needed to check that an 162 * action is triggerable, and then either runs the action or 163 * arranges its async execution by executing its Completion 164 * argument, if present. The method returns true if known to be 165 * complete. 166 * 167 * * Completion method tryFire(int mode) invokes the associated x 168 * method with its held arguments, and on success cleans up. 169 * The mode argument allows tryFire to be called twice (SYNC, 170 * then ASYNC); the first to screen and trap exceptions while 171 * arranging to execute, and the second when called from a 172 * task. (A few classes are not used async so take slightly 173 * different forms.) The claim() callback suppresses function 174 * invocation if already claimed by another thread. 175 * 176 * * CompletableFuture method xStage(...) is called from a public 177 * stage method of CompletableFuture x. It screens user 178 * arguments and invokes and/or creates the stage object. If 179 * not async and x is already complete, the action is run 180 * immediately. Otherwise a Completion c is created, pushed to 181 * x's stack (unless done), and started or triggered via 182 * c.tryFire. This also covers races possible if x completes 183 * while pushing. Classes with two inputs (for example BiApply) 184 * deal with races across both while pushing actions. The 185 * second completion is a CoCompletion pointing to the first, 186 * shared so that at most one performs the action. The 187 * multiple-arity methods allOf and anyOf do this pairwise to 188 * form trees of completions. 189 * 190 * Note that the generic type parameters of methods vary according 191 * to whether "this" is a source, dependent, or completion. 192 * 193 * Method postComplete is called upon completion unless the target 194 * is guaranteed not to be observable (i.e., not yet returned or 195 * linked). Multiple threads can call postComplete, which 196 * atomically pops each dependent action, and tries to trigger it 197 * via method tryFire, in NESTED mode. Triggering can propagate 198 * recursively, so NESTED mode returns its completed dependent (if 199 * one exists) for further processing by its caller (see method 200 * postFire). 201 * 202 * Blocking methods get() and join() rely on Signaller Completions 203 * that wake up waiting threads. The mechanics are similar to 204 * Treiber stack wait-nodes used in FutureTask, Phaser, and 205 * SynchronousQueue. See their internal documentation for 206 * algorithmic details. 207 * 208 * Without precautions, CompletableFutures would be prone to 209 * garbage accumulation as chains of Completions build up, each 210 * pointing back to its sources. So we null out fields as soon as 211 * possible. The screening checks needed anyway harmlessly ignore 212 * null arguments that may have been obtained during races with 213 * threads nulling out fields. We also try to unlink fired 214 * Completions from stacks that might never be popped (see method 215 * postFire). Completion fields need not be declared as final or 216 * volatile because they are only visible to other threads upon 217 * safe publication. 218 */ 219 220 volatile Object result; // Either the result or boxed AltResult 221 volatile Completion stack; // Top of Treiber stack of dependent actions 222 internalComplete(Object r)223 final boolean internalComplete(Object r) { // CAS from null to r 224 return U.compareAndSwapObject(this, RESULT, null, r); 225 } 226 casStack(Completion cmp, Completion val)227 final boolean casStack(Completion cmp, Completion val) { 228 return U.compareAndSwapObject(this, STACK, cmp, val); 229 } 230 231 /** Returns true if successfully pushed c onto stack. */ tryPushStack(Completion c)232 final boolean tryPushStack(Completion c) { 233 Completion h = stack; 234 lazySetNext(c, h); 235 return U.compareAndSwapObject(this, STACK, h, c); 236 } 237 238 /** Unconditionally pushes c onto stack, retrying if necessary. */ pushStack(Completion c)239 final void pushStack(Completion c) { 240 do {} while (!tryPushStack(c)); 241 } 242 243 /* ------------- Encoding and decoding outcomes -------------- */ 244 245 static final class AltResult { // See above 246 final Throwable ex; // null only for NIL AltResult(Throwable x)247 AltResult(Throwable x) { this.ex = x; } 248 } 249 250 /** The encoding of the null value. */ 251 static final AltResult NIL = new AltResult(null); 252 253 /** Completes with the null value, unless already completed. */ completeNull()254 final boolean completeNull() { 255 return U.compareAndSwapObject(this, RESULT, null, 256 NIL); 257 } 258 259 /** Returns the encoding of the given non-exceptional value. */ encodeValue(T t)260 final Object encodeValue(T t) { 261 return (t == null) ? NIL : t; 262 } 263 264 /** Completes with a non-exceptional result, unless already completed. */ completeValue(T t)265 final boolean completeValue(T t) { 266 return U.compareAndSwapObject(this, RESULT, null, 267 (t == null) ? NIL : t); 268 } 269 270 /** 271 * Returns the encoding of the given (non-null) exception as a 272 * wrapped CompletionException unless it is one already. 273 */ encodeThrowable(Throwable x)274 static AltResult encodeThrowable(Throwable x) { 275 return new AltResult((x instanceof CompletionException) ? x : 276 new CompletionException(x)); 277 } 278 279 /** Completes with an exceptional result, unless already completed. */ completeThrowable(Throwable x)280 final boolean completeThrowable(Throwable x) { 281 return U.compareAndSwapObject(this, RESULT, null, 282 encodeThrowable(x)); 283 } 284 285 /** 286 * Returns the encoding of the given (non-null) exception as a 287 * wrapped CompletionException unless it is one already. May 288 * return the given Object r (which must have been the result of a 289 * source future) if it is equivalent, i.e. if this is a simple 290 * relay of an existing CompletionException. 291 */ encodeThrowable(Throwable x, Object r)292 static Object encodeThrowable(Throwable x, Object r) { 293 if (!(x instanceof CompletionException)) 294 x = new CompletionException(x); 295 else if (r instanceof AltResult && x == ((AltResult)r).ex) 296 return r; 297 return new AltResult(x); 298 } 299 300 /** 301 * Completes with the given (non-null) exceptional result as a 302 * wrapped CompletionException unless it is one already, unless 303 * already completed. May complete with the given Object r 304 * (which must have been the result of a source future) if it is 305 * equivalent, i.e. if this is a simple propagation of an 306 * existing CompletionException. 307 */ completeThrowable(Throwable x, Object r)308 final boolean completeThrowable(Throwable x, Object r) { 309 return U.compareAndSwapObject(this, RESULT, null, 310 encodeThrowable(x, r)); 311 } 312 313 /** 314 * Returns the encoding of the given arguments: if the exception 315 * is non-null, encodes as AltResult. Otherwise uses the given 316 * value, boxed as NIL if null. 317 */ encodeOutcome(T t, Throwable x)318 Object encodeOutcome(T t, Throwable x) { 319 return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x); 320 } 321 322 /** 323 * Returns the encoding of a copied outcome; if exceptional, 324 * rewraps as a CompletionException, else returns argument. 325 */ encodeRelay(Object r)326 static Object encodeRelay(Object r) { 327 Throwable x; 328 return (((r instanceof AltResult) && 329 (x = ((AltResult)r).ex) != null && 330 !(x instanceof CompletionException)) ? 331 new AltResult(new CompletionException(x)) : r); 332 } 333 334 /** 335 * Completes with r or a copy of r, unless already completed. 336 * If exceptional, r is first coerced to a CompletionException. 337 */ completeRelay(Object r)338 final boolean completeRelay(Object r) { 339 return U.compareAndSwapObject(this, RESULT, null, 340 encodeRelay(r)); 341 } 342 343 /** 344 * Reports result using Future.get conventions. 345 */ reportGet(Object r)346 private static <T> T reportGet(Object r) 347 throws InterruptedException, ExecutionException { 348 if (r == null) // by convention below, null means interrupted 349 throw new InterruptedException(); 350 if (r instanceof AltResult) { 351 Throwable x, cause; 352 if ((x = ((AltResult)r).ex) == null) 353 return null; 354 if (x instanceof CancellationException) 355 throw (CancellationException)x; 356 if ((x instanceof CompletionException) && 357 (cause = x.getCause()) != null) 358 x = cause; 359 throw new ExecutionException(x); 360 } 361 @SuppressWarnings("unchecked") T t = (T) r; 362 return t; 363 } 364 365 /** 366 * Decodes outcome to return result or throw unchecked exception. 367 */ reportJoin(Object r)368 private static <T> T reportJoin(Object r) { 369 if (r instanceof AltResult) { 370 Throwable x; 371 if ((x = ((AltResult)r).ex) == null) 372 return null; 373 if (x instanceof CancellationException) 374 throw (CancellationException)x; 375 if (x instanceof CompletionException) 376 throw (CompletionException)x; 377 throw new CompletionException(x); 378 } 379 @SuppressWarnings("unchecked") T t = (T) r; 380 return t; 381 } 382 383 /* ------------- Async task preliminaries -------------- */ 384 385 /** 386 * A marker interface identifying asynchronous tasks produced by 387 * {@code async} methods. This may be useful for monitoring, 388 * debugging, and tracking asynchronous activities. 389 * 390 * @since 1.8 391 */ 392 public static interface AsynchronousCompletionTask { 393 } 394 395 private static final boolean USE_COMMON_POOL = 396 (ForkJoinPool.getCommonPoolParallelism() > 1); 397 398 /** 399 * Default executor -- ForkJoinPool.commonPool() unless it cannot 400 * support parallelism. 401 */ 402 private static final Executor ASYNC_POOL = USE_COMMON_POOL ? 403 ForkJoinPool.commonPool() : new ThreadPerTaskExecutor(); 404 405 /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */ 406 static final class ThreadPerTaskExecutor implements Executor { execute(Runnable r)407 public void execute(Runnable r) { new Thread(r).start(); } 408 } 409 410 /** 411 * Null-checks user executor argument, and translates uses of 412 * commonPool to ASYNC_POOL in case parallelism disabled. 413 */ screenExecutor(Executor e)414 static Executor screenExecutor(Executor e) { 415 if (!USE_COMMON_POOL && e == ForkJoinPool.commonPool()) 416 return ASYNC_POOL; 417 if (e == null) throw new NullPointerException(); 418 return e; 419 } 420 421 // Modes for Completion.tryFire. Signedness matters. 422 static final int SYNC = 0; 423 static final int ASYNC = 1; 424 static final int NESTED = -1; 425 426 /** 427 * Spins before blocking in waitingGet 428 */ 429 static final int SPINS = (Runtime.getRuntime().availableProcessors() > 1 ? 430 1 << 8 : 0); 431 432 /* ------------- Base Completion classes and operations -------------- */ 433 434 @SuppressWarnings("serial") 435 abstract static class Completion extends ForkJoinTask<Void> 436 implements Runnable, AsynchronousCompletionTask { 437 volatile Completion next; // Treiber stack link 438 439 /** 440 * Performs completion action if triggered, returning a 441 * dependent that may need propagation, if one exists. 442 * 443 * @param mode SYNC, ASYNC, or NESTED 444 */ tryFire(int mode)445 abstract CompletableFuture<?> tryFire(int mode); 446 447 /** Returns true if possibly still triggerable. Used by cleanStack. */ isLive()448 abstract boolean isLive(); 449 run()450 public final void run() { tryFire(ASYNC); } exec()451 public final boolean exec() { tryFire(ASYNC); return false; } getRawResult()452 public final Void getRawResult() { return null; } setRawResult(Void v)453 public final void setRawResult(Void v) {} 454 } 455 lazySetNext(Completion c, Completion next)456 static void lazySetNext(Completion c, Completion next) { 457 U.putOrderedObject(c, NEXT, next); 458 } 459 460 /** 461 * Pops and tries to trigger all reachable dependents. Call only 462 * when known to be done. 463 */ postComplete()464 final void postComplete() { 465 /* 466 * On each step, variable f holds current dependents to pop 467 * and run. It is extended along only one path at a time, 468 * pushing others to avoid unbounded recursion. 469 */ 470 CompletableFuture<?> f = this; Completion h; 471 while ((h = f.stack) != null || 472 (f != this && (h = (f = this).stack) != null)) { 473 CompletableFuture<?> d; Completion t; 474 if (f.casStack(h, t = h.next)) { 475 if (t != null) { 476 if (f != this) { 477 pushStack(h); 478 continue; 479 } 480 h.next = null; // detach 481 } 482 f = (d = h.tryFire(NESTED)) == null ? this : d; 483 } 484 } 485 } 486 487 /** Traverses stack and unlinks dead Completions. */ cleanStack()488 final void cleanStack() { 489 for (Completion p = null, q = stack; q != null;) { 490 Completion s = q.next; 491 if (q.isLive()) { 492 p = q; 493 q = s; 494 } 495 else if (p == null) { 496 casStack(q, s); 497 q = stack; 498 } 499 else { 500 p.next = s; 501 if (p.isLive()) 502 q = s; 503 else { 504 p = null; // restart 505 q = stack; 506 } 507 } 508 } 509 } 510 511 /* ------------- One-input Completions -------------- */ 512 513 /** A Completion with a source, dependent, and executor. */ 514 @SuppressWarnings("serial") 515 abstract static class UniCompletion<T,V> extends Completion { 516 Executor executor; // executor to use (null if none) 517 CompletableFuture<V> dep; // the dependent to complete 518 CompletableFuture<T> src; // source for action 519 UniCompletion(Executor executor, CompletableFuture<V> dep, CompletableFuture<T> src)520 UniCompletion(Executor executor, CompletableFuture<V> dep, 521 CompletableFuture<T> src) { 522 this.executor = executor; this.dep = dep; this.src = src; 523 } 524 525 /** 526 * Returns true if action can be run. Call only when known to 527 * be triggerable. Uses FJ tag bit to ensure that only one 528 * thread claims ownership. If async, starts as task -- a 529 * later call to tryFire will run action. 530 */ claim()531 final boolean claim() { 532 Executor e = executor; 533 if (compareAndSetForkJoinTaskTag((short)0, (short)1)) { 534 if (e == null) 535 return true; 536 executor = null; // disable 537 e.execute(this); 538 } 539 return false; 540 } 541 isLive()542 final boolean isLive() { return dep != null; } 543 } 544 545 /** Pushes the given completion (if it exists) unless done. */ push(UniCompletion<?,?> c)546 final void push(UniCompletion<?,?> c) { 547 if (c != null) { 548 while (result == null && !tryPushStack(c)) 549 lazySetNext(c, null); // clear on failure 550 } 551 } 552 553 /** 554 * Post-processing by dependent after successful UniCompletion 555 * tryFire. Tries to clean stack of source a, and then either runs 556 * postComplete or returns this to caller, depending on mode. 557 */ postFire(CompletableFuture<?> a, int mode)558 final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) { 559 if (a != null && a.stack != null) { 560 if (mode < 0 || a.result == null) 561 a.cleanStack(); 562 else 563 a.postComplete(); 564 } 565 if (result != null && stack != null) { 566 if (mode < 0) 567 return this; 568 else 569 postComplete(); 570 } 571 return null; 572 } 573 574 @SuppressWarnings("serial") 575 static final class UniApply<T,V> extends UniCompletion<T,V> { 576 Function<? super T,? extends V> fn; UniApply(Executor executor, CompletableFuture<V> dep, CompletableFuture<T> src, Function<? super T,? extends V> fn)577 UniApply(Executor executor, CompletableFuture<V> dep, 578 CompletableFuture<T> src, 579 Function<? super T,? extends V> fn) { 580 super(executor, dep, src); this.fn = fn; 581 } tryFire(int mode)582 final CompletableFuture<V> tryFire(int mode) { 583 CompletableFuture<V> d; CompletableFuture<T> a; 584 if ((d = dep) == null || 585 !d.uniApply(a = src, fn, mode > 0 ? null : this)) 586 return null; 587 dep = null; src = null; fn = null; 588 return d.postFire(a, mode); 589 } 590 } 591 uniApply(CompletableFuture<S> a, Function<? super S,? extends T> f, UniApply<S,T> c)592 final <S> boolean uniApply(CompletableFuture<S> a, 593 Function<? super S,? extends T> f, 594 UniApply<S,T> c) { 595 Object r; Throwable x; 596 if (a == null || (r = a.result) == null || f == null) 597 return false; 598 tryComplete: if (result == null) { 599 if (r instanceof AltResult) { 600 if ((x = ((AltResult)r).ex) != null) { 601 completeThrowable(x, r); 602 break tryComplete; 603 } 604 r = null; 605 } 606 try { 607 if (c != null && !c.claim()) 608 return false; 609 @SuppressWarnings("unchecked") S s = (S) r; 610 completeValue(f.apply(s)); 611 } catch (Throwable ex) { 612 completeThrowable(ex); 613 } 614 } 615 return true; 616 } 617 uniApplyStage( Executor e, Function<? super T,? extends V> f)618 private <V> CompletableFuture<V> uniApplyStage( 619 Executor e, Function<? super T,? extends V> f) { 620 if (f == null) throw new NullPointerException(); 621 CompletableFuture<V> d = newIncompleteFuture(); 622 if (e != null || !d.uniApply(this, f, null)) { 623 UniApply<T,V> c = new UniApply<T,V>(e, d, this, f); 624 push(c); 625 c.tryFire(SYNC); 626 } 627 return d; 628 } 629 630 @SuppressWarnings("serial") 631 static final class UniAccept<T> extends UniCompletion<T,Void> { 632 Consumer<? super T> fn; UniAccept(Executor executor, CompletableFuture<Void> dep, CompletableFuture<T> src, Consumer<? super T> fn)633 UniAccept(Executor executor, CompletableFuture<Void> dep, 634 CompletableFuture<T> src, Consumer<? super T> fn) { 635 super(executor, dep, src); this.fn = fn; 636 } tryFire(int mode)637 final CompletableFuture<Void> tryFire(int mode) { 638 CompletableFuture<Void> d; CompletableFuture<T> a; 639 if ((d = dep) == null || 640 !d.uniAccept(a = src, fn, mode > 0 ? null : this)) 641 return null; 642 dep = null; src = null; fn = null; 643 return d.postFire(a, mode); 644 } 645 } 646 uniAccept(CompletableFuture<S> a, Consumer<? super S> f, UniAccept<S> c)647 final <S> boolean uniAccept(CompletableFuture<S> a, 648 Consumer<? super S> f, UniAccept<S> c) { 649 Object r; Throwable x; 650 if (a == null || (r = a.result) == null || f == null) 651 return false; 652 tryComplete: if (result == null) { 653 if (r instanceof AltResult) { 654 if ((x = ((AltResult)r).ex) != null) { 655 completeThrowable(x, r); 656 break tryComplete; 657 } 658 r = null; 659 } 660 try { 661 if (c != null && !c.claim()) 662 return false; 663 @SuppressWarnings("unchecked") S s = (S) r; 664 f.accept(s); 665 completeNull(); 666 } catch (Throwable ex) { 667 completeThrowable(ex); 668 } 669 } 670 return true; 671 } 672 uniAcceptStage(Executor e, Consumer<? super T> f)673 private CompletableFuture<Void> uniAcceptStage(Executor e, 674 Consumer<? super T> f) { 675 if (f == null) throw new NullPointerException(); 676 CompletableFuture<Void> d = newIncompleteFuture(); 677 if (e != null || !d.uniAccept(this, f, null)) { 678 UniAccept<T> c = new UniAccept<T>(e, d, this, f); 679 push(c); 680 c.tryFire(SYNC); 681 } 682 return d; 683 } 684 685 @SuppressWarnings("serial") 686 static final class UniRun<T> extends UniCompletion<T,Void> { 687 Runnable fn; UniRun(Executor executor, CompletableFuture<Void> dep, CompletableFuture<T> src, Runnable fn)688 UniRun(Executor executor, CompletableFuture<Void> dep, 689 CompletableFuture<T> src, Runnable fn) { 690 super(executor, dep, src); this.fn = fn; 691 } tryFire(int mode)692 final CompletableFuture<Void> tryFire(int mode) { 693 CompletableFuture<Void> d; CompletableFuture<T> a; 694 if ((d = dep) == null || 695 !d.uniRun(a = src, fn, mode > 0 ? null : this)) 696 return null; 697 dep = null; src = null; fn = null; 698 return d.postFire(a, mode); 699 } 700 } 701 uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c)702 final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) { 703 Object r; Throwable x; 704 if (a == null || (r = a.result) == null || f == null) 705 return false; 706 if (result == null) { 707 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) 708 completeThrowable(x, r); 709 else 710 try { 711 if (c != null && !c.claim()) 712 return false; 713 f.run(); 714 completeNull(); 715 } catch (Throwable ex) { 716 completeThrowable(ex); 717 } 718 } 719 return true; 720 } 721 uniRunStage(Executor e, Runnable f)722 private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) { 723 if (f == null) throw new NullPointerException(); 724 CompletableFuture<Void> d = newIncompleteFuture(); 725 if (e != null || !d.uniRun(this, f, null)) { 726 UniRun<T> c = new UniRun<T>(e, d, this, f); 727 push(c); 728 c.tryFire(SYNC); 729 } 730 return d; 731 } 732 733 @SuppressWarnings("serial") 734 static final class UniWhenComplete<T> extends UniCompletion<T,T> { 735 BiConsumer<? super T, ? super Throwable> fn; UniWhenComplete(Executor executor, CompletableFuture<T> dep, CompletableFuture<T> src, BiConsumer<? super T, ? super Throwable> fn)736 UniWhenComplete(Executor executor, CompletableFuture<T> dep, 737 CompletableFuture<T> src, 738 BiConsumer<? super T, ? super Throwable> fn) { 739 super(executor, dep, src); this.fn = fn; 740 } tryFire(int mode)741 final CompletableFuture<T> tryFire(int mode) { 742 CompletableFuture<T> d; CompletableFuture<T> a; 743 if ((d = dep) == null || 744 !d.uniWhenComplete(a = src, fn, mode > 0 ? null : this)) 745 return null; 746 dep = null; src = null; fn = null; 747 return d.postFire(a, mode); 748 } 749 } 750 uniWhenComplete(CompletableFuture<T> a, BiConsumer<? super T,? super Throwable> f, UniWhenComplete<T> c)751 final boolean uniWhenComplete(CompletableFuture<T> a, 752 BiConsumer<? super T,? super Throwable> f, 753 UniWhenComplete<T> c) { 754 Object r; T t; Throwable x = null; 755 if (a == null || (r = a.result) == null || f == null) 756 return false; 757 if (result == null) { 758 try { 759 if (c != null && !c.claim()) 760 return false; 761 if (r instanceof AltResult) { 762 x = ((AltResult)r).ex; 763 t = null; 764 } else { 765 @SuppressWarnings("unchecked") T tr = (T) r; 766 t = tr; 767 } 768 f.accept(t, x); 769 if (x == null) { 770 internalComplete(r); 771 return true; 772 } 773 } catch (Throwable ex) { 774 if (x == null) 775 x = ex; 776 else if (x != ex) 777 x.addSuppressed(ex); 778 } 779 completeThrowable(x, r); 780 } 781 return true; 782 } 783 uniWhenCompleteStage( Executor e, BiConsumer<? super T, ? super Throwable> f)784 private CompletableFuture<T> uniWhenCompleteStage( 785 Executor e, BiConsumer<? super T, ? super Throwable> f) { 786 if (f == null) throw new NullPointerException(); 787 CompletableFuture<T> d = newIncompleteFuture(); 788 if (e != null || !d.uniWhenComplete(this, f, null)) { 789 UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f); 790 push(c); 791 c.tryFire(SYNC); 792 } 793 return d; 794 } 795 796 @SuppressWarnings("serial") 797 static final class UniHandle<T,V> extends UniCompletion<T,V> { 798 BiFunction<? super T, Throwable, ? extends V> fn; UniHandle(Executor executor, CompletableFuture<V> dep, CompletableFuture<T> src, BiFunction<? super T, Throwable, ? extends V> fn)799 UniHandle(Executor executor, CompletableFuture<V> dep, 800 CompletableFuture<T> src, 801 BiFunction<? super T, Throwable, ? extends V> fn) { 802 super(executor, dep, src); this.fn = fn; 803 } tryFire(int mode)804 final CompletableFuture<V> tryFire(int mode) { 805 CompletableFuture<V> d; CompletableFuture<T> a; 806 if ((d = dep) == null || 807 !d.uniHandle(a = src, fn, mode > 0 ? null : this)) 808 return null; 809 dep = null; src = null; fn = null; 810 return d.postFire(a, mode); 811 } 812 } 813 uniHandle(CompletableFuture<S> a, BiFunction<? super S, Throwable, ? extends T> f, UniHandle<S,T> c)814 final <S> boolean uniHandle(CompletableFuture<S> a, 815 BiFunction<? super S, Throwable, ? extends T> f, 816 UniHandle<S,T> c) { 817 Object r; S s; Throwable x; 818 if (a == null || (r = a.result) == null || f == null) 819 return false; 820 if (result == null) { 821 try { 822 if (c != null && !c.claim()) 823 return false; 824 if (r instanceof AltResult) { 825 x = ((AltResult)r).ex; 826 s = null; 827 } else { 828 x = null; 829 @SuppressWarnings("unchecked") S ss = (S) r; 830 s = ss; 831 } 832 completeValue(f.apply(s, x)); 833 } catch (Throwable ex) { 834 completeThrowable(ex); 835 } 836 } 837 return true; 838 } 839 uniHandleStage( Executor e, BiFunction<? super T, Throwable, ? extends V> f)840 private <V> CompletableFuture<V> uniHandleStage( 841 Executor e, BiFunction<? super T, Throwable, ? extends V> f) { 842 if (f == null) throw new NullPointerException(); 843 CompletableFuture<V> d = newIncompleteFuture(); 844 if (e != null || !d.uniHandle(this, f, null)) { 845 UniHandle<T,V> c = new UniHandle<T,V>(e, d, this, f); 846 push(c); 847 c.tryFire(SYNC); 848 } 849 return d; 850 } 851 852 @SuppressWarnings("serial") 853 static final class UniExceptionally<T> extends UniCompletion<T,T> { 854 Function<? super Throwable, ? extends T> fn; UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src, Function<? super Throwable, ? extends T> fn)855 UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src, 856 Function<? super Throwable, ? extends T> fn) { 857 super(null, dep, src); this.fn = fn; 858 } tryFire(int mode)859 final CompletableFuture<T> tryFire(int mode) { // never ASYNC 860 // assert mode != ASYNC; 861 CompletableFuture<T> d; CompletableFuture<T> a; 862 if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this)) 863 return null; 864 dep = null; src = null; fn = null; 865 return d.postFire(a, mode); 866 } 867 } 868 uniExceptionally(CompletableFuture<T> a, Function<? super Throwable, ? extends T> f, UniExceptionally<T> c)869 final boolean uniExceptionally(CompletableFuture<T> a, 870 Function<? super Throwable, ? extends T> f, 871 UniExceptionally<T> c) { 872 Object r; Throwable x; 873 if (a == null || (r = a.result) == null || f == null) 874 return false; 875 if (result == null) { 876 try { 877 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) { 878 if (c != null && !c.claim()) 879 return false; 880 completeValue(f.apply(x)); 881 } else 882 internalComplete(r); 883 } catch (Throwable ex) { 884 completeThrowable(ex); 885 } 886 } 887 return true; 888 } 889 uniExceptionallyStage( Function<Throwable, ? extends T> f)890 private CompletableFuture<T> uniExceptionallyStage( 891 Function<Throwable, ? extends T> f) { 892 if (f == null) throw new NullPointerException(); 893 CompletableFuture<T> d = newIncompleteFuture(); 894 if (!d.uniExceptionally(this, f, null)) { 895 UniExceptionally<T> c = new UniExceptionally<T>(d, this, f); 896 push(c); 897 c.tryFire(SYNC); 898 } 899 return d; 900 } 901 902 @SuppressWarnings("serial") 903 static final class UniRelay<T> extends UniCompletion<T,T> { // for Compose UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src)904 UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src) { 905 super(null, dep, src); 906 } tryFire(int mode)907 final CompletableFuture<T> tryFire(int mode) { 908 CompletableFuture<T> d; CompletableFuture<T> a; 909 if ((d = dep) == null || !d.uniRelay(a = src)) 910 return null; 911 src = null; dep = null; 912 return d.postFire(a, mode); 913 } 914 } 915 uniRelay(CompletableFuture<T> a)916 final boolean uniRelay(CompletableFuture<T> a) { 917 Object r; 918 if (a == null || (r = a.result) == null) 919 return false; 920 if (result == null) // no need to claim 921 completeRelay(r); 922 return true; 923 } 924 uniCopyStage()925 private CompletableFuture<T> uniCopyStage() { 926 Object r; 927 CompletableFuture<T> d = newIncompleteFuture(); 928 if ((r = result) != null) 929 d.completeRelay(r); 930 else { 931 UniRelay<T> c = new UniRelay<T>(d, this); 932 push(c); 933 c.tryFire(SYNC); 934 } 935 return d; 936 } 937 uniAsMinimalStage()938 private MinimalStage<T> uniAsMinimalStage() { 939 Object r; 940 if ((r = result) != null) 941 return new MinimalStage<T>(encodeRelay(r)); 942 MinimalStage<T> d = new MinimalStage<T>(); 943 UniRelay<T> c = new UniRelay<T>(d, this); 944 push(c); 945 c.tryFire(SYNC); 946 return d; 947 } 948 949 @SuppressWarnings("serial") 950 static final class UniCompose<T,V> extends UniCompletion<T,V> { 951 Function<? super T, ? extends CompletionStage<V>> fn; UniCompose(Executor executor, CompletableFuture<V> dep, CompletableFuture<T> src, Function<? super T, ? extends CompletionStage<V>> fn)952 UniCompose(Executor executor, CompletableFuture<V> dep, 953 CompletableFuture<T> src, 954 Function<? super T, ? extends CompletionStage<V>> fn) { 955 super(executor, dep, src); this.fn = fn; 956 } tryFire(int mode)957 final CompletableFuture<V> tryFire(int mode) { 958 CompletableFuture<V> d; CompletableFuture<T> a; 959 if ((d = dep) == null || 960 !d.uniCompose(a = src, fn, mode > 0 ? null : this)) 961 return null; 962 dep = null; src = null; fn = null; 963 return d.postFire(a, mode); 964 } 965 } 966 uniCompose( CompletableFuture<S> a, Function<? super S, ? extends CompletionStage<T>> f, UniCompose<S,T> c)967 final <S> boolean uniCompose( 968 CompletableFuture<S> a, 969 Function<? super S, ? extends CompletionStage<T>> f, 970 UniCompose<S,T> c) { 971 Object r; Throwable x; 972 if (a == null || (r = a.result) == null || f == null) 973 return false; 974 tryComplete: if (result == null) { 975 if (r instanceof AltResult) { 976 if ((x = ((AltResult)r).ex) != null) { 977 completeThrowable(x, r); 978 break tryComplete; 979 } 980 r = null; 981 } 982 try { 983 if (c != null && !c.claim()) 984 return false; 985 @SuppressWarnings("unchecked") S s = (S) r; 986 CompletableFuture<T> g = f.apply(s).toCompletableFuture(); 987 if (g.result == null || !uniRelay(g)) { 988 UniRelay<T> copy = new UniRelay<T>(this, g); 989 g.push(copy); 990 copy.tryFire(SYNC); 991 if (result == null) 992 return false; 993 } 994 } catch (Throwable ex) { 995 completeThrowable(ex); 996 } 997 } 998 return true; 999 } 1000 uniComposeStage( Executor e, Function<? super T, ? extends CompletionStage<V>> f)1001 private <V> CompletableFuture<V> uniComposeStage( 1002 Executor e, Function<? super T, ? extends CompletionStage<V>> f) { 1003 if (f == null) throw new NullPointerException(); 1004 Object r, s; Throwable x; 1005 CompletableFuture<V> d = newIncompleteFuture(); 1006 if (e == null && (r = result) != null) { 1007 if (r instanceof AltResult) { 1008 if ((x = ((AltResult)r).ex) != null) { 1009 d.result = encodeThrowable(x, r); 1010 return d; 1011 } 1012 r = null; 1013 } 1014 try { 1015 @SuppressWarnings("unchecked") T t = (T) r; 1016 CompletableFuture<V> g = f.apply(t).toCompletableFuture(); 1017 if ((s = g.result) != null) 1018 d.completeRelay(s); 1019 else { 1020 UniRelay<V> c = new UniRelay<V>(d, g); 1021 g.push(c); 1022 c.tryFire(SYNC); 1023 } 1024 return d; 1025 } catch (Throwable ex) { 1026 d.result = encodeThrowable(ex); 1027 return d; 1028 } 1029 } 1030 UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f); 1031 push(c); 1032 c.tryFire(SYNC); 1033 return d; 1034 } 1035 1036 /* ------------- Two-input Completions -------------- */ 1037 1038 /** A Completion for an action with two sources */ 1039 @SuppressWarnings("serial") 1040 abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> { 1041 CompletableFuture<U> snd; // second source for action BiCompletion(Executor executor, CompletableFuture<V> dep, CompletableFuture<T> src, CompletableFuture<U> snd)1042 BiCompletion(Executor executor, CompletableFuture<V> dep, 1043 CompletableFuture<T> src, CompletableFuture<U> snd) { 1044 super(executor, dep, src); this.snd = snd; 1045 } 1046 } 1047 1048 /** A Completion delegating to a BiCompletion */ 1049 @SuppressWarnings("serial") 1050 static final class CoCompletion extends Completion { 1051 BiCompletion<?,?,?> base; CoCompletion(BiCompletion<?,?,?> base)1052 CoCompletion(BiCompletion<?,?,?> base) { this.base = base; } tryFire(int mode)1053 final CompletableFuture<?> tryFire(int mode) { 1054 BiCompletion<?,?,?> c; CompletableFuture<?> d; 1055 if ((c = base) == null || (d = c.tryFire(mode)) == null) 1056 return null; 1057 base = null; // detach 1058 return d; 1059 } isLive()1060 final boolean isLive() { 1061 BiCompletion<?,?,?> c; 1062 return (c = base) != null && c.dep != null; 1063 } 1064 } 1065 1066 /** Pushes completion to this and b unless both done. */ bipush(CompletableFuture<?> b, BiCompletion<?,?,?> c)1067 final void bipush(CompletableFuture<?> b, BiCompletion<?,?,?> c) { 1068 if (c != null) { 1069 Object r; 1070 while ((r = result) == null && !tryPushStack(c)) 1071 lazySetNext(c, null); // clear on failure 1072 if (b != null && b != this && b.result == null) { 1073 Completion q = (r != null) ? c : new CoCompletion(c); 1074 while (b.result == null && !b.tryPushStack(q)) 1075 lazySetNext(q, null); // clear on failure 1076 } 1077 } 1078 } 1079 1080 /** Post-processing after successful BiCompletion tryFire. */ postFire(CompletableFuture<?> a, CompletableFuture<?> b, int mode)1081 final CompletableFuture<T> postFire(CompletableFuture<?> a, 1082 CompletableFuture<?> b, int mode) { 1083 if (b != null && b.stack != null) { // clean second source 1084 if (mode < 0 || b.result == null) 1085 b.cleanStack(); 1086 else 1087 b.postComplete(); 1088 } 1089 return postFire(a, mode); 1090 } 1091 1092 @SuppressWarnings("serial") 1093 static final class BiApply<T,U,V> extends BiCompletion<T,U,V> { 1094 BiFunction<? super T,? super U,? extends V> fn; BiApply(Executor executor, CompletableFuture<V> dep, CompletableFuture<T> src, CompletableFuture<U> snd, BiFunction<? super T,? super U,? extends V> fn)1095 BiApply(Executor executor, CompletableFuture<V> dep, 1096 CompletableFuture<T> src, CompletableFuture<U> snd, 1097 BiFunction<? super T,? super U,? extends V> fn) { 1098 super(executor, dep, src, snd); this.fn = fn; 1099 } tryFire(int mode)1100 final CompletableFuture<V> tryFire(int mode) { 1101 CompletableFuture<V> d; 1102 CompletableFuture<T> a; 1103 CompletableFuture<U> b; 1104 if ((d = dep) == null || 1105 !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this)) 1106 return null; 1107 dep = null; src = null; snd = null; fn = null; 1108 return d.postFire(a, b, mode); 1109 } 1110 } 1111 biApply(CompletableFuture<R> a, CompletableFuture<S> b, BiFunction<? super R,? super S,? extends T> f, BiApply<R,S,T> c)1112 final <R,S> boolean biApply(CompletableFuture<R> a, 1113 CompletableFuture<S> b, 1114 BiFunction<? super R,? super S,? extends T> f, 1115 BiApply<R,S,T> c) { 1116 Object r, s; Throwable x; 1117 if (a == null || (r = a.result) == null || 1118 b == null || (s = b.result) == null || f == null) 1119 return false; 1120 tryComplete: if (result == null) { 1121 if (r instanceof AltResult) { 1122 if ((x = ((AltResult)r).ex) != null) { 1123 completeThrowable(x, r); 1124 break tryComplete; 1125 } 1126 r = null; 1127 } 1128 if (s instanceof AltResult) { 1129 if ((x = ((AltResult)s).ex) != null) { 1130 completeThrowable(x, s); 1131 break tryComplete; 1132 } 1133 s = null; 1134 } 1135 try { 1136 if (c != null && !c.claim()) 1137 return false; 1138 @SuppressWarnings("unchecked") R rr = (R) r; 1139 @SuppressWarnings("unchecked") S ss = (S) s; 1140 completeValue(f.apply(rr, ss)); 1141 } catch (Throwable ex) { 1142 completeThrowable(ex); 1143 } 1144 } 1145 return true; 1146 } 1147 biApplyStage( Executor e, CompletionStage<U> o, BiFunction<? super T,? super U,? extends V> f)1148 private <U,V> CompletableFuture<V> biApplyStage( 1149 Executor e, CompletionStage<U> o, 1150 BiFunction<? super T,? super U,? extends V> f) { 1151 CompletableFuture<U> b; 1152 if (f == null || (b = o.toCompletableFuture()) == null) 1153 throw new NullPointerException(); 1154 CompletableFuture<V> d = newIncompleteFuture(); 1155 if (e != null || !d.biApply(this, b, f, null)) { 1156 BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f); 1157 bipush(b, c); 1158 c.tryFire(SYNC); 1159 } 1160 return d; 1161 } 1162 1163 @SuppressWarnings("serial") 1164 static final class BiAccept<T,U> extends BiCompletion<T,U,Void> { 1165 BiConsumer<? super T,? super U> fn; BiAccept(Executor executor, CompletableFuture<Void> dep, CompletableFuture<T> src, CompletableFuture<U> snd, BiConsumer<? super T,? super U> fn)1166 BiAccept(Executor executor, CompletableFuture<Void> dep, 1167 CompletableFuture<T> src, CompletableFuture<U> snd, 1168 BiConsumer<? super T,? super U> fn) { 1169 super(executor, dep, src, snd); this.fn = fn; 1170 } tryFire(int mode)1171 final CompletableFuture<Void> tryFire(int mode) { 1172 CompletableFuture<Void> d; 1173 CompletableFuture<T> a; 1174 CompletableFuture<U> b; 1175 if ((d = dep) == null || 1176 !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this)) 1177 return null; 1178 dep = null; src = null; snd = null; fn = null; 1179 return d.postFire(a, b, mode); 1180 } 1181 } 1182 biAccept(CompletableFuture<R> a, CompletableFuture<S> b, BiConsumer<? super R,? super S> f, BiAccept<R,S> c)1183 final <R,S> boolean biAccept(CompletableFuture<R> a, 1184 CompletableFuture<S> b, 1185 BiConsumer<? super R,? super S> f, 1186 BiAccept<R,S> c) { 1187 Object r, s; Throwable x; 1188 if (a == null || (r = a.result) == null || 1189 b == null || (s = b.result) == null || f == null) 1190 return false; 1191 tryComplete: if (result == null) { 1192 if (r instanceof AltResult) { 1193 if ((x = ((AltResult)r).ex) != null) { 1194 completeThrowable(x, r); 1195 break tryComplete; 1196 } 1197 r = null; 1198 } 1199 if (s instanceof AltResult) { 1200 if ((x = ((AltResult)s).ex) != null) { 1201 completeThrowable(x, s); 1202 break tryComplete; 1203 } 1204 s = null; 1205 } 1206 try { 1207 if (c != null && !c.claim()) 1208 return false; 1209 @SuppressWarnings("unchecked") R rr = (R) r; 1210 @SuppressWarnings("unchecked") S ss = (S) s; 1211 f.accept(rr, ss); 1212 completeNull(); 1213 } catch (Throwable ex) { 1214 completeThrowable(ex); 1215 } 1216 } 1217 return true; 1218 } 1219 biAcceptStage( Executor e, CompletionStage<U> o, BiConsumer<? super T,? super U> f)1220 private <U> CompletableFuture<Void> biAcceptStage( 1221 Executor e, CompletionStage<U> o, 1222 BiConsumer<? super T,? super U> f) { 1223 CompletableFuture<U> b; 1224 if (f == null || (b = o.toCompletableFuture()) == null) 1225 throw new NullPointerException(); 1226 CompletableFuture<Void> d = newIncompleteFuture(); 1227 if (e != null || !d.biAccept(this, b, f, null)) { 1228 BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f); 1229 bipush(b, c); 1230 c.tryFire(SYNC); 1231 } 1232 return d; 1233 } 1234 1235 @SuppressWarnings("serial") 1236 static final class BiRun<T,U> extends BiCompletion<T,U,Void> { 1237 Runnable fn; BiRun(Executor executor, CompletableFuture<Void> dep, CompletableFuture<T> src, CompletableFuture<U> snd, Runnable fn)1238 BiRun(Executor executor, CompletableFuture<Void> dep, 1239 CompletableFuture<T> src, 1240 CompletableFuture<U> snd, 1241 Runnable fn) { 1242 super(executor, dep, src, snd); this.fn = fn; 1243 } tryFire(int mode)1244 final CompletableFuture<Void> tryFire(int mode) { 1245 CompletableFuture<Void> d; 1246 CompletableFuture<T> a; 1247 CompletableFuture<U> b; 1248 if ((d = dep) == null || 1249 !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this)) 1250 return null; 1251 dep = null; src = null; snd = null; fn = null; 1252 return d.postFire(a, b, mode); 1253 } 1254 } 1255 biRun(CompletableFuture<?> a, CompletableFuture<?> b, Runnable f, BiRun<?,?> c)1256 final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b, 1257 Runnable f, BiRun<?,?> c) { 1258 Object r, s; Throwable x; 1259 if (a == null || (r = a.result) == null || 1260 b == null || (s = b.result) == null || f == null) 1261 return false; 1262 if (result == null) { 1263 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) 1264 completeThrowable(x, r); 1265 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null) 1266 completeThrowable(x, s); 1267 else 1268 try { 1269 if (c != null && !c.claim()) 1270 return false; 1271 f.run(); 1272 completeNull(); 1273 } catch (Throwable ex) { 1274 completeThrowable(ex); 1275 } 1276 } 1277 return true; 1278 } 1279 biRunStage(Executor e, CompletionStage<?> o, Runnable f)1280 private CompletableFuture<Void> biRunStage(Executor e, CompletionStage<?> o, 1281 Runnable f) { 1282 CompletableFuture<?> b; 1283 if (f == null || (b = o.toCompletableFuture()) == null) 1284 throw new NullPointerException(); 1285 CompletableFuture<Void> d = newIncompleteFuture(); 1286 if (e != null || !d.biRun(this, b, f, null)) { 1287 BiRun<T,?> c = new BiRun<>(e, d, this, b, f); 1288 bipush(b, c); 1289 c.tryFire(SYNC); 1290 } 1291 return d; 1292 } 1293 1294 @SuppressWarnings("serial") 1295 static final class BiRelay<T,U> extends BiCompletion<T,U,Void> { // for And BiRelay(CompletableFuture<Void> dep, CompletableFuture<T> src, CompletableFuture<U> snd)1296 BiRelay(CompletableFuture<Void> dep, 1297 CompletableFuture<T> src, 1298 CompletableFuture<U> snd) { 1299 super(null, dep, src, snd); 1300 } tryFire(int mode)1301 final CompletableFuture<Void> tryFire(int mode) { 1302 CompletableFuture<Void> d; 1303 CompletableFuture<T> a; 1304 CompletableFuture<U> b; 1305 if ((d = dep) == null || !d.biRelay(a = src, b = snd)) 1306 return null; 1307 src = null; snd = null; dep = null; 1308 return d.postFire(a, b, mode); 1309 } 1310 } 1311 biRelay(CompletableFuture<?> a, CompletableFuture<?> b)1312 boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) { 1313 Object r, s; Throwable x; 1314 if (a == null || (r = a.result) == null || 1315 b == null || (s = b.result) == null) 1316 return false; 1317 if (result == null) { 1318 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) 1319 completeThrowable(x, r); 1320 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null) 1321 completeThrowable(x, s); 1322 else 1323 completeNull(); 1324 } 1325 return true; 1326 } 1327 1328 /** Recursively constructs a tree of completions. */ andTree(CompletableFuture<?>[] cfs, int lo, int hi)1329 static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs, 1330 int lo, int hi) { 1331 CompletableFuture<Void> d = new CompletableFuture<Void>(); 1332 if (lo > hi) // empty 1333 d.result = NIL; 1334 else { 1335 CompletableFuture<?> a, b; 1336 int mid = (lo + hi) >>> 1; 1337 if ((a = (lo == mid ? cfs[lo] : 1338 andTree(cfs, lo, mid))) == null || 1339 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] : 1340 andTree(cfs, mid+1, hi))) == null) 1341 throw new NullPointerException(); 1342 if (!d.biRelay(a, b)) { 1343 BiRelay<?,?> c = new BiRelay<>(d, a, b); 1344 a.bipush(b, c); 1345 c.tryFire(SYNC); 1346 } 1347 } 1348 return d; 1349 } 1350 1351 /* ------------- Projected (Ored) BiCompletions -------------- */ 1352 1353 /** Pushes completion to this and b unless either done. */ orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c)1354 final void orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c) { 1355 if (c != null) { 1356 while ((b == null || b.result == null) && result == null) { 1357 if (tryPushStack(c)) { 1358 if (b != null && b != this && b.result == null) { 1359 Completion q = new CoCompletion(c); 1360 while (result == null && b.result == null && 1361 !b.tryPushStack(q)) 1362 lazySetNext(q, null); // clear on failure 1363 } 1364 break; 1365 } 1366 lazySetNext(c, null); // clear on failure 1367 } 1368 } 1369 } 1370 1371 @SuppressWarnings("serial") 1372 static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> { 1373 Function<? super T,? extends V> fn; OrApply(Executor executor, CompletableFuture<V> dep, CompletableFuture<T> src, CompletableFuture<U> snd, Function<? super T,? extends V> fn)1374 OrApply(Executor executor, CompletableFuture<V> dep, 1375 CompletableFuture<T> src, 1376 CompletableFuture<U> snd, 1377 Function<? super T,? extends V> fn) { 1378 super(executor, dep, src, snd); this.fn = fn; 1379 } tryFire(int mode)1380 final CompletableFuture<V> tryFire(int mode) { 1381 CompletableFuture<V> d; 1382 CompletableFuture<T> a; 1383 CompletableFuture<U> b; 1384 if ((d = dep) == null || 1385 !d.orApply(a = src, b = snd, fn, mode > 0 ? null : this)) 1386 return null; 1387 dep = null; src = null; snd = null; fn = null; 1388 return d.postFire(a, b, mode); 1389 } 1390 } 1391 orApply(CompletableFuture<R> a, CompletableFuture<S> b, Function<? super R, ? extends T> f, OrApply<R,S,T> c)1392 final <R,S extends R> boolean orApply(CompletableFuture<R> a, 1393 CompletableFuture<S> b, 1394 Function<? super R, ? extends T> f, 1395 OrApply<R,S,T> c) { 1396 Object r; Throwable x; 1397 if (a == null || b == null || 1398 ((r = a.result) == null && (r = b.result) == null) || f == null) 1399 return false; 1400 tryComplete: if (result == null) { 1401 try { 1402 if (c != null && !c.claim()) 1403 return false; 1404 if (r instanceof AltResult) { 1405 if ((x = ((AltResult)r).ex) != null) { 1406 completeThrowable(x, r); 1407 break tryComplete; 1408 } 1409 r = null; 1410 } 1411 @SuppressWarnings("unchecked") R rr = (R) r; 1412 completeValue(f.apply(rr)); 1413 } catch (Throwable ex) { 1414 completeThrowable(ex); 1415 } 1416 } 1417 return true; 1418 } 1419 orApplyStage( Executor e, CompletionStage<U> o, Function<? super T, ? extends V> f)1420 private <U extends T,V> CompletableFuture<V> orApplyStage( 1421 Executor e, CompletionStage<U> o, 1422 Function<? super T, ? extends V> f) { 1423 CompletableFuture<U> b; 1424 if (f == null || (b = o.toCompletableFuture()) == null) 1425 throw new NullPointerException(); 1426 CompletableFuture<V> d = newIncompleteFuture(); 1427 if (e != null || !d.orApply(this, b, f, null)) { 1428 OrApply<T,U,V> c = new OrApply<T,U,V>(e, d, this, b, f); 1429 orpush(b, c); 1430 c.tryFire(SYNC); 1431 } 1432 return d; 1433 } 1434 1435 @SuppressWarnings("serial") 1436 static final class OrAccept<T,U extends T> extends BiCompletion<T,U,Void> { 1437 Consumer<? super T> fn; OrAccept(Executor executor, CompletableFuture<Void> dep, CompletableFuture<T> src, CompletableFuture<U> snd, Consumer<? super T> fn)1438 OrAccept(Executor executor, CompletableFuture<Void> dep, 1439 CompletableFuture<T> src, 1440 CompletableFuture<U> snd, 1441 Consumer<? super T> fn) { 1442 super(executor, dep, src, snd); this.fn = fn; 1443 } tryFire(int mode)1444 final CompletableFuture<Void> tryFire(int mode) { 1445 CompletableFuture<Void> d; 1446 CompletableFuture<T> a; 1447 CompletableFuture<U> b; 1448 if ((d = dep) == null || 1449 !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this)) 1450 return null; 1451 dep = null; src = null; snd = null; fn = null; 1452 return d.postFire(a, b, mode); 1453 } 1454 } 1455 orAccept(CompletableFuture<R> a, CompletableFuture<S> b, Consumer<? super R> f, OrAccept<R,S> c)1456 final <R,S extends R> boolean orAccept(CompletableFuture<R> a, 1457 CompletableFuture<S> b, 1458 Consumer<? super R> f, 1459 OrAccept<R,S> c) { 1460 Object r; Throwable x; 1461 if (a == null || b == null || 1462 ((r = a.result) == null && (r = b.result) == null) || f == null) 1463 return false; 1464 tryComplete: if (result == null) { 1465 try { 1466 if (c != null && !c.claim()) 1467 return false; 1468 if (r instanceof AltResult) { 1469 if ((x = ((AltResult)r).ex) != null) { 1470 completeThrowable(x, r); 1471 break tryComplete; 1472 } 1473 r = null; 1474 } 1475 @SuppressWarnings("unchecked") R rr = (R) r; 1476 f.accept(rr); 1477 completeNull(); 1478 } catch (Throwable ex) { 1479 completeThrowable(ex); 1480 } 1481 } 1482 return true; 1483 } 1484 orAcceptStage( Executor e, CompletionStage<U> o, Consumer<? super T> f)1485 private <U extends T> CompletableFuture<Void> orAcceptStage( 1486 Executor e, CompletionStage<U> o, Consumer<? super T> f) { 1487 CompletableFuture<U> b; 1488 if (f == null || (b = o.toCompletableFuture()) == null) 1489 throw new NullPointerException(); 1490 CompletableFuture<Void> d = newIncompleteFuture(); 1491 if (e != null || !d.orAccept(this, b, f, null)) { 1492 OrAccept<T,U> c = new OrAccept<T,U>(e, d, this, b, f); 1493 orpush(b, c); 1494 c.tryFire(SYNC); 1495 } 1496 return d; 1497 } 1498 1499 @SuppressWarnings("serial") 1500 static final class OrRun<T,U> extends BiCompletion<T,U,Void> { 1501 Runnable fn; OrRun(Executor executor, CompletableFuture<Void> dep, CompletableFuture<T> src, CompletableFuture<U> snd, Runnable fn)1502 OrRun(Executor executor, CompletableFuture<Void> dep, 1503 CompletableFuture<T> src, 1504 CompletableFuture<U> snd, 1505 Runnable fn) { 1506 super(executor, dep, src, snd); this.fn = fn; 1507 } tryFire(int mode)1508 final CompletableFuture<Void> tryFire(int mode) { 1509 CompletableFuture<Void> d; 1510 CompletableFuture<T> a; 1511 CompletableFuture<U> b; 1512 if ((d = dep) == null || 1513 !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this)) 1514 return null; 1515 dep = null; src = null; snd = null; fn = null; 1516 return d.postFire(a, b, mode); 1517 } 1518 } 1519 orRun(CompletableFuture<?> a, CompletableFuture<?> b, Runnable f, OrRun<?,?> c)1520 final boolean orRun(CompletableFuture<?> a, CompletableFuture<?> b, 1521 Runnable f, OrRun<?,?> c) { 1522 Object r; Throwable x; 1523 if (a == null || b == null || 1524 ((r = a.result) == null && (r = b.result) == null) || f == null) 1525 return false; 1526 if (result == null) { 1527 try { 1528 if (c != null && !c.claim()) 1529 return false; 1530 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) 1531 completeThrowable(x, r); 1532 else { 1533 f.run(); 1534 completeNull(); 1535 } 1536 } catch (Throwable ex) { 1537 completeThrowable(ex); 1538 } 1539 } 1540 return true; 1541 } 1542 orRunStage(Executor e, CompletionStage<?> o, Runnable f)1543 private CompletableFuture<Void> orRunStage(Executor e, CompletionStage<?> o, 1544 Runnable f) { 1545 CompletableFuture<?> b; 1546 if (f == null || (b = o.toCompletableFuture()) == null) 1547 throw new NullPointerException(); 1548 CompletableFuture<Void> d = newIncompleteFuture(); 1549 if (e != null || !d.orRun(this, b, f, null)) { 1550 OrRun<T,?> c = new OrRun<>(e, d, this, b, f); 1551 orpush(b, c); 1552 c.tryFire(SYNC); 1553 } 1554 return d; 1555 } 1556 1557 @SuppressWarnings("serial") 1558 static final class OrRelay<T,U> extends BiCompletion<T,U,Object> { // for Or OrRelay(CompletableFuture<Object> dep, CompletableFuture<T> src, CompletableFuture<U> snd)1559 OrRelay(CompletableFuture<Object> dep, CompletableFuture<T> src, 1560 CompletableFuture<U> snd) { 1561 super(null, dep, src, snd); 1562 } tryFire(int mode)1563 final CompletableFuture<Object> tryFire(int mode) { 1564 CompletableFuture<Object> d; 1565 CompletableFuture<T> a; 1566 CompletableFuture<U> b; 1567 if ((d = dep) == null || !d.orRelay(a = src, b = snd)) 1568 return null; 1569 src = null; snd = null; dep = null; 1570 return d.postFire(a, b, mode); 1571 } 1572 } 1573 orRelay(CompletableFuture<?> a, CompletableFuture<?> b)1574 final boolean orRelay(CompletableFuture<?> a, CompletableFuture<?> b) { 1575 Object r; 1576 if (a == null || b == null || 1577 ((r = a.result) == null && (r = b.result) == null)) 1578 return false; 1579 if (result == null) 1580 completeRelay(r); 1581 return true; 1582 } 1583 1584 /** Recursively constructs a tree of completions. */ orTree(CompletableFuture<?>[] cfs, int lo, int hi)1585 static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs, 1586 int lo, int hi) { 1587 CompletableFuture<Object> d = new CompletableFuture<Object>(); 1588 if (lo <= hi) { 1589 CompletableFuture<?> a, b; 1590 int mid = (lo + hi) >>> 1; 1591 if ((a = (lo == mid ? cfs[lo] : 1592 orTree(cfs, lo, mid))) == null || 1593 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] : 1594 orTree(cfs, mid+1, hi))) == null) 1595 throw new NullPointerException(); 1596 if (!d.orRelay(a, b)) { 1597 OrRelay<?,?> c = new OrRelay<>(d, a, b); 1598 a.orpush(b, c); 1599 c.tryFire(SYNC); 1600 } 1601 } 1602 return d; 1603 } 1604 1605 /* ------------- Zero-input Async forms -------------- */ 1606 1607 @SuppressWarnings("serial") 1608 static final class AsyncSupply<T> extends ForkJoinTask<Void> 1609 implements Runnable, AsynchronousCompletionTask { 1610 CompletableFuture<T> dep; Supplier<? extends T> fn; AsyncSupply(CompletableFuture<T> dep, Supplier<? extends T> fn)1611 AsyncSupply(CompletableFuture<T> dep, Supplier<? extends T> fn) { 1612 this.dep = dep; this.fn = fn; 1613 } 1614 getRawResult()1615 public final Void getRawResult() { return null; } setRawResult(Void v)1616 public final void setRawResult(Void v) {} exec()1617 public final boolean exec() { run(); return true; } 1618 run()1619 public void run() { 1620 CompletableFuture<T> d; Supplier<? extends T> f; 1621 if ((d = dep) != null && (f = fn) != null) { 1622 dep = null; fn = null; 1623 if (d.result == null) { 1624 try { 1625 d.completeValue(f.get()); 1626 } catch (Throwable ex) { 1627 d.completeThrowable(ex); 1628 } 1629 } 1630 d.postComplete(); 1631 } 1632 } 1633 } 1634 asyncSupplyStage(Executor e, Supplier<U> f)1635 static <U> CompletableFuture<U> asyncSupplyStage(Executor e, 1636 Supplier<U> f) { 1637 if (f == null) throw new NullPointerException(); 1638 CompletableFuture<U> d = new CompletableFuture<U>(); 1639 e.execute(new AsyncSupply<U>(d, f)); 1640 return d; 1641 } 1642 1643 @SuppressWarnings("serial") 1644 static final class AsyncRun extends ForkJoinTask<Void> 1645 implements Runnable, AsynchronousCompletionTask { 1646 CompletableFuture<Void> dep; Runnable fn; AsyncRun(CompletableFuture<Void> dep, Runnable fn)1647 AsyncRun(CompletableFuture<Void> dep, Runnable fn) { 1648 this.dep = dep; this.fn = fn; 1649 } 1650 getRawResult()1651 public final Void getRawResult() { return null; } setRawResult(Void v)1652 public final void setRawResult(Void v) {} exec()1653 public final boolean exec() { run(); return true; } 1654 run()1655 public void run() { 1656 CompletableFuture<Void> d; Runnable f; 1657 if ((d = dep) != null && (f = fn) != null) { 1658 dep = null; fn = null; 1659 if (d.result == null) { 1660 try { 1661 f.run(); 1662 d.completeNull(); 1663 } catch (Throwable ex) { 1664 d.completeThrowable(ex); 1665 } 1666 } 1667 d.postComplete(); 1668 } 1669 } 1670 } 1671 asyncRunStage(Executor e, Runnable f)1672 static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) { 1673 if (f == null) throw new NullPointerException(); 1674 CompletableFuture<Void> d = new CompletableFuture<Void>(); 1675 e.execute(new AsyncRun(d, f)); 1676 return d; 1677 } 1678 1679 /* ------------- Signallers -------------- */ 1680 1681 /** 1682 * Completion for recording and releasing a waiting thread. This 1683 * class implements ManagedBlocker to avoid starvation when 1684 * blocking actions pile up in ForkJoinPools. 1685 */ 1686 @SuppressWarnings("serial") 1687 static final class Signaller extends Completion 1688 implements ForkJoinPool.ManagedBlocker { 1689 long nanos; // remaining wait time if timed 1690 final long deadline; // non-zero if timed 1691 final boolean interruptible; 1692 boolean interrupted; 1693 volatile Thread thread; 1694 Signaller(boolean interruptible, long nanos, long deadline)1695 Signaller(boolean interruptible, long nanos, long deadline) { 1696 this.thread = Thread.currentThread(); 1697 this.interruptible = interruptible; 1698 this.nanos = nanos; 1699 this.deadline = deadline; 1700 } tryFire(int ignore)1701 final CompletableFuture<?> tryFire(int ignore) { 1702 Thread w; // no need to atomically claim 1703 if ((w = thread) != null) { 1704 thread = null; 1705 LockSupport.unpark(w); 1706 } 1707 return null; 1708 } isReleasable()1709 public boolean isReleasable() { 1710 if (Thread.interrupted()) 1711 interrupted = true; 1712 return ((interrupted && interruptible) || 1713 (deadline != 0L && 1714 (nanos <= 0L || 1715 (nanos = deadline - System.nanoTime()) <= 0L)) || 1716 thread == null); 1717 } block()1718 public boolean block() { 1719 while (!isReleasable()) { 1720 if (deadline == 0L) 1721 LockSupport.park(this); 1722 else 1723 LockSupport.parkNanos(this, nanos); 1724 } 1725 return true; 1726 } isLive()1727 final boolean isLive() { return thread != null; } 1728 } 1729 1730 /** 1731 * Returns raw result after waiting, or null if interruptible and 1732 * interrupted. 1733 */ waitingGet(boolean interruptible)1734 private Object waitingGet(boolean interruptible) { 1735 Signaller q = null; 1736 boolean queued = false; 1737 int spins = SPINS; 1738 Object r; 1739 while ((r = result) == null) { 1740 if (spins > 0) { 1741 if (ThreadLocalRandom.nextSecondarySeed() >= 0) 1742 --spins; 1743 } 1744 else if (q == null) 1745 q = new Signaller(interruptible, 0L, 0L); 1746 else if (!queued) 1747 queued = tryPushStack(q); 1748 else { 1749 try { 1750 ForkJoinPool.managedBlock(q); 1751 } catch (InterruptedException ie) { // currently cannot happen 1752 q.interrupted = true; 1753 } 1754 if (q.interrupted && interruptible) 1755 break; 1756 } 1757 } 1758 if (q != null) { 1759 q.thread = null; 1760 if (q.interrupted) { 1761 if (interruptible) 1762 cleanStack(); 1763 else 1764 Thread.currentThread().interrupt(); 1765 } 1766 } 1767 if (r != null) 1768 postComplete(); 1769 return r; 1770 } 1771 1772 /** 1773 * Returns raw result after waiting, or null if interrupted, or 1774 * throws TimeoutException on timeout. 1775 */ timedGet(long nanos)1776 private Object timedGet(long nanos) throws TimeoutException { 1777 if (Thread.interrupted()) 1778 return null; 1779 if (nanos > 0L) { 1780 long d = System.nanoTime() + nanos; 1781 long deadline = (d == 0L) ? 1L : d; // avoid 0 1782 Signaller q = null; 1783 boolean queued = false; 1784 Object r; 1785 while ((r = result) == null) { // similar to untimed, without spins 1786 if (q == null) 1787 q = new Signaller(true, nanos, deadline); 1788 else if (!queued) 1789 queued = tryPushStack(q); 1790 else if (q.nanos <= 0L) 1791 break; 1792 else { 1793 try { 1794 ForkJoinPool.managedBlock(q); 1795 } catch (InterruptedException ie) { 1796 q.interrupted = true; 1797 } 1798 if (q.interrupted) 1799 break; 1800 } 1801 } 1802 if (q != null) 1803 q.thread = null; 1804 if (r != null) 1805 postComplete(); 1806 else 1807 cleanStack(); 1808 if (r != null || (q != null && q.interrupted)) 1809 return r; 1810 } 1811 throw new TimeoutException(); 1812 } 1813 1814 /* ------------- public methods -------------- */ 1815 1816 /** 1817 * Creates a new incomplete CompletableFuture. 1818 */ CompletableFuture()1819 public CompletableFuture() { 1820 } 1821 1822 /** 1823 * Creates a new complete CompletableFuture with given encoded result. 1824 */ CompletableFuture(Object r)1825 CompletableFuture(Object r) { 1826 this.result = r; 1827 } 1828 1829 /** 1830 * Returns a new CompletableFuture that is asynchronously completed 1831 * by a task running in the {@link ForkJoinPool#commonPool()} with 1832 * the value obtained by calling the given Supplier. 1833 * 1834 * @param supplier a function returning the value to be used 1835 * to complete the returned CompletableFuture 1836 * @param <U> the function's return type 1837 * @return the new CompletableFuture 1838 */ supplyAsync(Supplier<U> supplier)1839 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) { 1840 return asyncSupplyStage(ASYNC_POOL, supplier); 1841 } 1842 1843 /** 1844 * Returns a new CompletableFuture that is asynchronously completed 1845 * by a task running in the given executor with the value obtained 1846 * by calling the given Supplier. 1847 * 1848 * @param supplier a function returning the value to be used 1849 * to complete the returned CompletableFuture 1850 * @param executor the executor to use for asynchronous execution 1851 * @param <U> the function's return type 1852 * @return the new CompletableFuture 1853 */ supplyAsync(Supplier<U> supplier, Executor executor)1854 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, 1855 Executor executor) { 1856 return asyncSupplyStage(screenExecutor(executor), supplier); 1857 } 1858 1859 /** 1860 * Returns a new CompletableFuture that is asynchronously completed 1861 * by a task running in the {@link ForkJoinPool#commonPool()} after 1862 * it runs the given action. 1863 * 1864 * @param runnable the action to run before completing the 1865 * returned CompletableFuture 1866 * @return the new CompletableFuture 1867 */ runAsync(Runnable runnable)1868 public static CompletableFuture<Void> runAsync(Runnable runnable) { 1869 return asyncRunStage(ASYNC_POOL, runnable); 1870 } 1871 1872 /** 1873 * Returns a new CompletableFuture that is asynchronously completed 1874 * by a task running in the given executor after it runs the given 1875 * action. 1876 * 1877 * @param runnable the action to run before completing the 1878 * returned CompletableFuture 1879 * @param executor the executor to use for asynchronous execution 1880 * @return the new CompletableFuture 1881 */ runAsync(Runnable runnable, Executor executor)1882 public static CompletableFuture<Void> runAsync(Runnable runnable, 1883 Executor executor) { 1884 return asyncRunStage(screenExecutor(executor), runnable); 1885 } 1886 1887 /** 1888 * Returns a new CompletableFuture that is already completed with 1889 * the given value. 1890 * 1891 * @param value the value 1892 * @param <U> the type of the value 1893 * @return the completed CompletableFuture 1894 */ completedFuture(U value)1895 public static <U> CompletableFuture<U> completedFuture(U value) { 1896 return new CompletableFuture<U>((value == null) ? NIL : value); 1897 } 1898 1899 /** 1900 * Returns {@code true} if completed in any fashion: normally, 1901 * exceptionally, or via cancellation. 1902 * 1903 * @return {@code true} if completed 1904 */ isDone()1905 public boolean isDone() { 1906 return result != null; 1907 } 1908 1909 /** 1910 * Waits if necessary for this future to complete, and then 1911 * returns its result. 1912 * 1913 * @return the result value 1914 * @throws CancellationException if this future was cancelled 1915 * @throws ExecutionException if this future completed exceptionally 1916 * @throws InterruptedException if the current thread was interrupted 1917 * while waiting 1918 */ get()1919 public T get() throws InterruptedException, ExecutionException { 1920 Object r; 1921 return reportGet((r = result) == null ? waitingGet(true) : r); 1922 } 1923 1924 /** 1925 * Waits if necessary for at most the given time for this future 1926 * to complete, and then returns its result, if available. 1927 * 1928 * @param timeout the maximum time to wait 1929 * @param unit the time unit of the timeout argument 1930 * @return the result value 1931 * @throws CancellationException if this future was cancelled 1932 * @throws ExecutionException if this future completed exceptionally 1933 * @throws InterruptedException if the current thread was interrupted 1934 * while waiting 1935 * @throws TimeoutException if the wait timed out 1936 */ get(long timeout, TimeUnit unit)1937 public T get(long timeout, TimeUnit unit) 1938 throws InterruptedException, ExecutionException, TimeoutException { 1939 Object r; 1940 long nanos = unit.toNanos(timeout); 1941 return reportGet((r = result) == null ? timedGet(nanos) : r); 1942 } 1943 1944 /** 1945 * Returns the result value when complete, or throws an 1946 * (unchecked) exception if completed exceptionally. To better 1947 * conform with the use of common functional forms, if a 1948 * computation involved in the completion of this 1949 * CompletableFuture threw an exception, this method throws an 1950 * (unchecked) {@link CompletionException} with the underlying 1951 * exception as its cause. 1952 * 1953 * @return the result value 1954 * @throws CancellationException if the computation was cancelled 1955 * @throws CompletionException if this future completed 1956 * exceptionally or a completion computation threw an exception 1957 */ join()1958 public T join() { 1959 Object r; 1960 return reportJoin((r = result) == null ? waitingGet(false) : r); 1961 } 1962 1963 /** 1964 * Returns the result value (or throws any encountered exception) 1965 * if completed, else returns the given valueIfAbsent. 1966 * 1967 * @param valueIfAbsent the value to return if not completed 1968 * @return the result value, if completed, else the given valueIfAbsent 1969 * @throws CancellationException if the computation was cancelled 1970 * @throws CompletionException if this future completed 1971 * exceptionally or a completion computation threw an exception 1972 */ getNow(T valueIfAbsent)1973 public T getNow(T valueIfAbsent) { 1974 Object r; 1975 return ((r = result) == null) ? valueIfAbsent : reportJoin(r); 1976 } 1977 1978 /** 1979 * If not already completed, sets the value returned by {@link 1980 * #get()} and related methods to the given value. 1981 * 1982 * @param value the result value 1983 * @return {@code true} if this invocation caused this CompletableFuture 1984 * to transition to a completed state, else {@code false} 1985 */ complete(T value)1986 public boolean complete(T value) { 1987 boolean triggered = completeValue(value); 1988 postComplete(); 1989 return triggered; 1990 } 1991 1992 /** 1993 * If not already completed, causes invocations of {@link #get()} 1994 * and related methods to throw the given exception. 1995 * 1996 * @param ex the exception 1997 * @return {@code true} if this invocation caused this CompletableFuture 1998 * to transition to a completed state, else {@code false} 1999 */ completeExceptionally(Throwable ex)2000 public boolean completeExceptionally(Throwable ex) { 2001 if (ex == null) throw new NullPointerException(); 2002 boolean triggered = internalComplete(new AltResult(ex)); 2003 postComplete(); 2004 return triggered; 2005 } 2006 thenApply( Function<? super T,? extends U> fn)2007 public <U> CompletableFuture<U> thenApply( 2008 Function<? super T,? extends U> fn) { 2009 return uniApplyStage(null, fn); 2010 } 2011 thenApplyAsync( Function<? super T,? extends U> fn)2012 public <U> CompletableFuture<U> thenApplyAsync( 2013 Function<? super T,? extends U> fn) { 2014 return uniApplyStage(defaultExecutor(), fn); 2015 } 2016 thenApplyAsync( Function<? super T,? extends U> fn, Executor executor)2017 public <U> CompletableFuture<U> thenApplyAsync( 2018 Function<? super T,? extends U> fn, Executor executor) { 2019 return uniApplyStage(screenExecutor(executor), fn); 2020 } 2021 thenAccept(Consumer<? super T> action)2022 public CompletableFuture<Void> thenAccept(Consumer<? super T> action) { 2023 return uniAcceptStage(null, action); 2024 } 2025 thenAcceptAsync(Consumer<? super T> action)2026 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) { 2027 return uniAcceptStage(defaultExecutor(), action); 2028 } 2029 thenAcceptAsync(Consumer<? super T> action, Executor executor)2030 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, 2031 Executor executor) { 2032 return uniAcceptStage(screenExecutor(executor), action); 2033 } 2034 thenRun(Runnable action)2035 public CompletableFuture<Void> thenRun(Runnable action) { 2036 return uniRunStage(null, action); 2037 } 2038 thenRunAsync(Runnable action)2039 public CompletableFuture<Void> thenRunAsync(Runnable action) { 2040 return uniRunStage(defaultExecutor(), action); 2041 } 2042 thenRunAsync(Runnable action, Executor executor)2043 public CompletableFuture<Void> thenRunAsync(Runnable action, 2044 Executor executor) { 2045 return uniRunStage(screenExecutor(executor), action); 2046 } 2047 thenCombine( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)2048 public <U,V> CompletableFuture<V> thenCombine( 2049 CompletionStage<? extends U> other, 2050 BiFunction<? super T,? super U,? extends V> fn) { 2051 return biApplyStage(null, other, fn); 2052 } 2053 thenCombineAsync( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)2054 public <U,V> CompletableFuture<V> thenCombineAsync( 2055 CompletionStage<? extends U> other, 2056 BiFunction<? super T,? super U,? extends V> fn) { 2057 return biApplyStage(defaultExecutor(), other, fn); 2058 } 2059 thenCombineAsync( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)2060 public <U,V> CompletableFuture<V> thenCombineAsync( 2061 CompletionStage<? extends U> other, 2062 BiFunction<? super T,? super U,? extends V> fn, Executor executor) { 2063 return biApplyStage(screenExecutor(executor), other, fn); 2064 } 2065 thenAcceptBoth( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)2066 public <U> CompletableFuture<Void> thenAcceptBoth( 2067 CompletionStage<? extends U> other, 2068 BiConsumer<? super T, ? super U> action) { 2069 return biAcceptStage(null, other, action); 2070 } 2071 thenAcceptBothAsync( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)2072 public <U> CompletableFuture<Void> thenAcceptBothAsync( 2073 CompletionStage<? extends U> other, 2074 BiConsumer<? super T, ? super U> action) { 2075 return biAcceptStage(defaultExecutor(), other, action); 2076 } 2077 thenAcceptBothAsync( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor)2078 public <U> CompletableFuture<Void> thenAcceptBothAsync( 2079 CompletionStage<? extends U> other, 2080 BiConsumer<? super T, ? super U> action, Executor executor) { 2081 return biAcceptStage(screenExecutor(executor), other, action); 2082 } 2083 runAfterBoth(CompletionStage<?> other, Runnable action)2084 public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, 2085 Runnable action) { 2086 return biRunStage(null, other, action); 2087 } 2088 runAfterBothAsync(CompletionStage<?> other, Runnable action)2089 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, 2090 Runnable action) { 2091 return biRunStage(defaultExecutor(), other, action); 2092 } 2093 runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor)2094 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, 2095 Runnable action, 2096 Executor executor) { 2097 return biRunStage(screenExecutor(executor), other, action); 2098 } 2099 applyToEither( CompletionStage<? extends T> other, Function<? super T, U> fn)2100 public <U> CompletableFuture<U> applyToEither( 2101 CompletionStage<? extends T> other, Function<? super T, U> fn) { 2102 return orApplyStage(null, other, fn); 2103 } 2104 applyToEitherAsync( CompletionStage<? extends T> other, Function<? super T, U> fn)2105 public <U> CompletableFuture<U> applyToEitherAsync( 2106 CompletionStage<? extends T> other, Function<? super T, U> fn) { 2107 return orApplyStage(defaultExecutor(), other, fn); 2108 } 2109 applyToEitherAsync( CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor)2110 public <U> CompletableFuture<U> applyToEitherAsync( 2111 CompletionStage<? extends T> other, Function<? super T, U> fn, 2112 Executor executor) { 2113 return orApplyStage(screenExecutor(executor), other, fn); 2114 } 2115 acceptEither( CompletionStage<? extends T> other, Consumer<? super T> action)2116 public CompletableFuture<Void> acceptEither( 2117 CompletionStage<? extends T> other, Consumer<? super T> action) { 2118 return orAcceptStage(null, other, action); 2119 } 2120 acceptEitherAsync( CompletionStage<? extends T> other, Consumer<? super T> action)2121 public CompletableFuture<Void> acceptEitherAsync( 2122 CompletionStage<? extends T> other, Consumer<? super T> action) { 2123 return orAcceptStage(defaultExecutor(), other, action); 2124 } 2125 acceptEitherAsync( CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)2126 public CompletableFuture<Void> acceptEitherAsync( 2127 CompletionStage<? extends T> other, Consumer<? super T> action, 2128 Executor executor) { 2129 return orAcceptStage(screenExecutor(executor), other, action); 2130 } 2131 runAfterEither(CompletionStage<?> other, Runnable action)2132 public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, 2133 Runnable action) { 2134 return orRunStage(null, other, action); 2135 } 2136 runAfterEitherAsync(CompletionStage<?> other, Runnable action)2137 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, 2138 Runnable action) { 2139 return orRunStage(defaultExecutor(), other, action); 2140 } 2141 runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor)2142 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, 2143 Runnable action, 2144 Executor executor) { 2145 return orRunStage(screenExecutor(executor), other, action); 2146 } 2147 thenCompose( Function<? super T, ? extends CompletionStage<U>> fn)2148 public <U> CompletableFuture<U> thenCompose( 2149 Function<? super T, ? extends CompletionStage<U>> fn) { 2150 return uniComposeStage(null, fn); 2151 } 2152 thenComposeAsync( Function<? super T, ? extends CompletionStage<U>> fn)2153 public <U> CompletableFuture<U> thenComposeAsync( 2154 Function<? super T, ? extends CompletionStage<U>> fn) { 2155 return uniComposeStage(defaultExecutor(), fn); 2156 } 2157 thenComposeAsync( Function<? super T, ? extends CompletionStage<U>> fn, Executor executor)2158 public <U> CompletableFuture<U> thenComposeAsync( 2159 Function<? super T, ? extends CompletionStage<U>> fn, 2160 Executor executor) { 2161 return uniComposeStage(screenExecutor(executor), fn); 2162 } 2163 whenComplete( BiConsumer<? super T, ? super Throwable> action)2164 public CompletableFuture<T> whenComplete( 2165 BiConsumer<? super T, ? super Throwable> action) { 2166 return uniWhenCompleteStage(null, action); 2167 } 2168 whenCompleteAsync( BiConsumer<? super T, ? super Throwable> action)2169 public CompletableFuture<T> whenCompleteAsync( 2170 BiConsumer<? super T, ? super Throwable> action) { 2171 return uniWhenCompleteStage(defaultExecutor(), action); 2172 } 2173 whenCompleteAsync( BiConsumer<? super T, ? super Throwable> action, Executor executor)2174 public CompletableFuture<T> whenCompleteAsync( 2175 BiConsumer<? super T, ? super Throwable> action, Executor executor) { 2176 return uniWhenCompleteStage(screenExecutor(executor), action); 2177 } 2178 handle( BiFunction<? super T, Throwable, ? extends U> fn)2179 public <U> CompletableFuture<U> handle( 2180 BiFunction<? super T, Throwable, ? extends U> fn) { 2181 return uniHandleStage(null, fn); 2182 } 2183 handleAsync( BiFunction<? super T, Throwable, ? extends U> fn)2184 public <U> CompletableFuture<U> handleAsync( 2185 BiFunction<? super T, Throwable, ? extends U> fn) { 2186 return uniHandleStage(defaultExecutor(), fn); 2187 } 2188 handleAsync( BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)2189 public <U> CompletableFuture<U> handleAsync( 2190 BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) { 2191 return uniHandleStage(screenExecutor(executor), fn); 2192 } 2193 2194 /** 2195 * Returns this CompletableFuture. 2196 * 2197 * @return this CompletableFuture 2198 */ toCompletableFuture()2199 public CompletableFuture<T> toCompletableFuture() { 2200 return this; 2201 } 2202 2203 // not in interface CompletionStage 2204 2205 /** 2206 * Returns a new CompletableFuture that is completed when this 2207 * CompletableFuture completes, with the result of the given 2208 * function of the exception triggering this CompletableFuture's 2209 * completion when it completes exceptionally; otherwise, if this 2210 * CompletableFuture completes normally, then the returned 2211 * CompletableFuture also completes normally with the same value. 2212 * Note: More flexible versions of this functionality are 2213 * available using methods {@code whenComplete} and {@code handle}. 2214 * 2215 * @param fn the function to use to compute the value of the 2216 * returned CompletableFuture if this CompletableFuture completed 2217 * exceptionally 2218 * @return the new CompletableFuture 2219 */ exceptionally( Function<Throwable, ? extends T> fn)2220 public CompletableFuture<T> exceptionally( 2221 Function<Throwable, ? extends T> fn) { 2222 return uniExceptionallyStage(fn); 2223 } 2224 2225 2226 /* ------------- Arbitrary-arity constructions -------------- */ 2227 2228 /** 2229 * Returns a new CompletableFuture that is completed when all of 2230 * the given CompletableFutures complete. If any of the given 2231 * CompletableFutures complete exceptionally, then the returned 2232 * CompletableFuture also does so, with a CompletionException 2233 * holding this exception as its cause. Otherwise, the results, 2234 * if any, of the given CompletableFutures are not reflected in 2235 * the returned CompletableFuture, but may be obtained by 2236 * inspecting them individually. If no CompletableFutures are 2237 * provided, returns a CompletableFuture completed with the value 2238 * {@code null}. 2239 * 2240 * <p>Among the applications of this method is to await completion 2241 * of a set of independent CompletableFutures before continuing a 2242 * program, as in: {@code CompletableFuture.allOf(c1, c2, 2243 * c3).join();}. 2244 * 2245 * @param cfs the CompletableFutures 2246 * @return a new CompletableFuture that is completed when all of the 2247 * given CompletableFutures complete 2248 * @throws NullPointerException if the array or any of its elements are 2249 * {@code null} 2250 */ allOf(CompletableFuture<?>.... cfs)2251 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) { 2252 return andTree(cfs, 0, cfs.length - 1); 2253 } 2254 2255 /** 2256 * Returns a new CompletableFuture that is completed when any of 2257 * the given CompletableFutures complete, with the same result. 2258 * Otherwise, if it completed exceptionally, the returned 2259 * CompletableFuture also does so, with a CompletionException 2260 * holding this exception as its cause. If no CompletableFutures 2261 * are provided, returns an incomplete CompletableFuture. 2262 * 2263 * @param cfs the CompletableFutures 2264 * @return a new CompletableFuture that is completed with the 2265 * result or exception of any of the given CompletableFutures when 2266 * one completes 2267 * @throws NullPointerException if the array or any of its elements are 2268 * {@code null} 2269 */ anyOf(CompletableFuture<?>.... cfs)2270 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) { 2271 return orTree(cfs, 0, cfs.length - 1); 2272 } 2273 2274 /* ------------- Control and status methods -------------- */ 2275 2276 /** 2277 * If not already completed, completes this CompletableFuture with 2278 * a {@link CancellationException}. Dependent CompletableFutures 2279 * that have not already completed will also complete 2280 * exceptionally, with a {@link CompletionException} caused by 2281 * this {@code CancellationException}. 2282 * 2283 * @param mayInterruptIfRunning this value has no effect in this 2284 * implementation because interrupts are not used to control 2285 * processing. 2286 * 2287 * @return {@code true} if this task is now cancelled 2288 */ cancel(boolean mayInterruptIfRunning)2289 public boolean cancel(boolean mayInterruptIfRunning) { 2290 boolean cancelled = (result == null) && 2291 internalComplete(new AltResult(new CancellationException())); 2292 postComplete(); 2293 return cancelled || isCancelled(); 2294 } 2295 2296 /** 2297 * Returns {@code true} if this CompletableFuture was cancelled 2298 * before it completed normally. 2299 * 2300 * @return {@code true} if this CompletableFuture was cancelled 2301 * before it completed normally 2302 */ isCancelled()2303 public boolean isCancelled() { 2304 Object r; 2305 return ((r = result) instanceof AltResult) && 2306 (((AltResult)r).ex instanceof CancellationException); 2307 } 2308 2309 /** 2310 * Returns {@code true} if this CompletableFuture completed 2311 * exceptionally, in any way. Possible causes include 2312 * cancellation, explicit invocation of {@code 2313 * completeExceptionally}, and abrupt termination of a 2314 * CompletionStage action. 2315 * 2316 * @return {@code true} if this CompletableFuture completed 2317 * exceptionally 2318 */ isCompletedExceptionally()2319 public boolean isCompletedExceptionally() { 2320 Object r; 2321 return ((r = result) instanceof AltResult) && r != NIL; 2322 } 2323 2324 /** 2325 * Forcibly sets or resets the value subsequently returned by 2326 * method {@link #get()} and related methods, whether or not 2327 * already completed. This method is designed for use only in 2328 * error recovery actions, and even in such situations may result 2329 * in ongoing dependent completions using established versus 2330 * overwritten outcomes. 2331 * 2332 * @param value the completion value 2333 */ obtrudeValue(T value)2334 public void obtrudeValue(T value) { 2335 result = (value == null) ? NIL : value; 2336 postComplete(); 2337 } 2338 2339 /** 2340 * Forcibly causes subsequent invocations of method {@link #get()} 2341 * and related methods to throw the given exception, whether or 2342 * not already completed. This method is designed for use only in 2343 * error recovery actions, and even in such situations may result 2344 * in ongoing dependent completions using established versus 2345 * overwritten outcomes. 2346 * 2347 * @param ex the exception 2348 * @throws NullPointerException if the exception is null 2349 */ obtrudeException(Throwable ex)2350 public void obtrudeException(Throwable ex) { 2351 if (ex == null) throw new NullPointerException(); 2352 result = new AltResult(ex); 2353 postComplete(); 2354 } 2355 2356 /** 2357 * Returns the estimated number of CompletableFutures whose 2358 * completions are awaiting completion of this CompletableFuture. 2359 * This method is designed for use in monitoring system state, not 2360 * for synchronization control. 2361 * 2362 * @return the number of dependent CompletableFutures 2363 */ getNumberOfDependents()2364 public int getNumberOfDependents() { 2365 int count = 0; 2366 for (Completion p = stack; p != null; p = p.next) 2367 ++count; 2368 return count; 2369 } 2370 2371 /** 2372 * Returns a string identifying this CompletableFuture, as well as 2373 * its completion state. The state, in brackets, contains the 2374 * String {@code "Completed Normally"} or the String {@code 2375 * "Completed Exceptionally"}, or the String {@code "Not 2376 * completed"} followed by the number of CompletableFutures 2377 * dependent upon its completion, if any. 2378 * 2379 * @return a string identifying this CompletableFuture, as well as its state 2380 */ toString()2381 public String toString() { 2382 Object r = result; 2383 int count = 0; // avoid call to getNumberOfDependents in case disabled 2384 for (Completion p = stack; p != null; p = p.next) 2385 ++count; 2386 return super.toString() + 2387 ((r == null) ? 2388 ((count == 0) ? 2389 "[Not completed]" : 2390 "[Not completed, " + count + " dependents]") : 2391 (((r instanceof AltResult) && ((AltResult)r).ex != null) ? 2392 "[Completed exceptionally]" : 2393 "[Completed normally]")); 2394 } 2395 2396 // jdk9 additions 2397 2398 /** 2399 * Returns a new incomplete CompletableFuture of the type to be 2400 * returned by a CompletionStage method. Subclasses should 2401 * normally override this method to return an instance of the same 2402 * class as this CompletableFuture. The default implementation 2403 * returns an instance of class CompletableFuture. 2404 * 2405 * @param <U> the type of the value 2406 * @return a new CompletableFuture 2407 * @since 9 2408 * @hide 2409 */ 2410 // Android-changed: hidden newIncompleteFuture()2411 public <U> CompletableFuture<U> newIncompleteFuture() { 2412 return new CompletableFuture<U>(); 2413 } 2414 2415 /** 2416 * Returns the default Executor used for async methods that do not 2417 * specify an Executor. This class uses the {@link 2418 * ForkJoinPool#commonPool()} if it supports more than one 2419 * parallel thread, or else an Executor using one thread per async 2420 * task. This method may be overridden in subclasses to return 2421 * an Executor that provides at least one independent thread. 2422 * 2423 * @return the executor 2424 * @since 9 2425 * @hide 2426 */ 2427 // Android-changed: hidden defaultExecutor()2428 public Executor defaultExecutor() { 2429 return ASYNC_POOL; 2430 } 2431 2432 /** 2433 * Returns a new CompletableFuture that is completed normally with 2434 * the same value as this CompletableFuture when it completes 2435 * normally. If this CompletableFuture completes exceptionally, 2436 * then the returned CompletableFuture completes exceptionally 2437 * with a CompletionException with this exception as cause. The 2438 * behavior is equivalent to {@code thenApply(x -> x)}. This 2439 * method may be useful as a form of "defensive copying", to 2440 * prevent clients from completing, while still being able to 2441 * arrange dependent actions. 2442 * 2443 * @return the new CompletableFuture 2444 * @since 9 2445 * @hide 2446 */ 2447 // Android-changed: hidden copy()2448 public CompletableFuture<T> copy() { 2449 return uniCopyStage(); 2450 } 2451 2452 /** 2453 * Returns a new CompletionStage that is completed normally with 2454 * the same value as this CompletableFuture when it completes 2455 * normally, and cannot be independently completed or otherwise 2456 * used in ways not defined by the methods of interface {@link 2457 * CompletionStage}. If this CompletableFuture completes 2458 * exceptionally, then the returned CompletionStage completes 2459 * exceptionally with a CompletionException with this exception as 2460 * cause. 2461 * 2462 * @return the new CompletionStage 2463 * @since 9 2464 * @hide 2465 */ 2466 // Android-changed: hidden minimalCompletionStage()2467 public CompletionStage<T> minimalCompletionStage() { 2468 return uniAsMinimalStage(); 2469 } 2470 2471 /** 2472 * Completes this CompletableFuture with the result of 2473 * the given Supplier function invoked from an asynchronous 2474 * task using the given executor. 2475 * 2476 * @param supplier a function returning the value to be used 2477 * to complete this CompletableFuture 2478 * @param executor the executor to use for asynchronous execution 2479 * @return this CompletableFuture 2480 * @since 9 2481 * @hide 2482 */ 2483 // Android-changed: hidden completeAsync(Supplier<? extends T> supplier, Executor executor)2484 public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier, 2485 Executor executor) { 2486 if (supplier == null || executor == null) 2487 throw new NullPointerException(); 2488 executor.execute(new AsyncSupply<T>(this, supplier)); 2489 return this; 2490 } 2491 2492 /** 2493 * Completes this CompletableFuture with the result of the given 2494 * Supplier function invoked from an asynchronous task using the 2495 * default executor. 2496 * 2497 * @param supplier a function returning the value to be used 2498 * to complete this CompletableFuture 2499 * @return this CompletableFuture 2500 * @since 9 2501 * @hide 2502 */ 2503 // Android-changed: hidden completeAsync(Supplier<? extends T> supplier)2504 public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier) { 2505 return completeAsync(supplier, defaultExecutor()); 2506 } 2507 2508 /** 2509 * Exceptionally completes this CompletableFuture with 2510 * a {@link TimeoutException} if not otherwise completed 2511 * before the given timeout. 2512 * 2513 * @param timeout how long to wait before completing exceptionally 2514 * with a TimeoutException, in units of {@code unit} 2515 * @param unit a {@code TimeUnit} determining how to interpret the 2516 * {@code timeout} parameter 2517 * @return this CompletableFuture 2518 * @since 9 2519 * @hide 2520 */ 2521 // Android-changed: hidden orTimeout(long timeout, TimeUnit unit)2522 public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) { 2523 if (unit == null) 2524 throw new NullPointerException(); 2525 if (result == null) 2526 whenComplete(new Canceller(Delayer.delay(new Timeout(this), 2527 timeout, unit))); 2528 return this; 2529 } 2530 2531 /** 2532 * Completes this CompletableFuture with the given value if not 2533 * otherwise completed before the given timeout. 2534 * 2535 * @param value the value to use upon timeout 2536 * @param timeout how long to wait before completing normally 2537 * with the given value, in units of {@code unit} 2538 * @param unit a {@code TimeUnit} determining how to interpret the 2539 * {@code timeout} parameter 2540 * @return this CompletableFuture 2541 * @since 9 2542 * @hide 2543 */ 2544 // Android-changed: hidden completeOnTimeout(T value, long timeout, TimeUnit unit)2545 public CompletableFuture<T> completeOnTimeout(T value, long timeout, 2546 TimeUnit unit) { 2547 if (unit == null) 2548 throw new NullPointerException(); 2549 if (result == null) 2550 whenComplete(new Canceller(Delayer.delay( 2551 new DelayedCompleter<T>(this, value), 2552 timeout, unit))); 2553 return this; 2554 } 2555 2556 /** 2557 * Returns a new Executor that submits a task to the given base 2558 * executor after the given delay (or no delay if non-positive). 2559 * Each delay commences upon invocation of the returned executor's 2560 * {@code execute} method. 2561 * 2562 * @param delay how long to delay, in units of {@code unit} 2563 * @param unit a {@code TimeUnit} determining how to interpret the 2564 * {@code delay} parameter 2565 * @param executor the base executor 2566 * @return the new delayed executor 2567 * @since 9 2568 * @hide 2569 */ 2570 // Android-changed: hidden delayedExecutor(long delay, TimeUnit unit, Executor executor)2571 public static Executor delayedExecutor(long delay, TimeUnit unit, 2572 Executor executor) { 2573 if (unit == null || executor == null) 2574 throw new NullPointerException(); 2575 return new DelayedExecutor(delay, unit, executor); 2576 } 2577 2578 /** 2579 * Returns a new Executor that submits a task to the default 2580 * executor after the given delay (or no delay if non-positive). 2581 * Each delay commences upon invocation of the returned executor's 2582 * {@code execute} method. 2583 * 2584 * @param delay how long to delay, in units of {@code unit} 2585 * @param unit a {@code TimeUnit} determining how to interpret the 2586 * {@code delay} parameter 2587 * @return the new delayed executor 2588 * @since 9 2589 * @hide 2590 */ 2591 // Android-changed: hidden delayedExecutor(long delay, TimeUnit unit)2592 public static Executor delayedExecutor(long delay, TimeUnit unit) { 2593 if (unit == null) 2594 throw new NullPointerException(); 2595 return new DelayedExecutor(delay, unit, ASYNC_POOL); 2596 } 2597 2598 /** 2599 * Returns a new CompletionStage that is already completed with 2600 * the given value and supports only those methods in 2601 * interface {@link CompletionStage}. 2602 * 2603 * @param value the value 2604 * @param <U> the type of the value 2605 * @return the completed CompletionStage 2606 * @since 9 2607 * @hide 2608 */ 2609 // Android-changed: hidden completedStage(U value)2610 public static <U> CompletionStage<U> completedStage(U value) { 2611 return new MinimalStage<U>((value == null) ? NIL : value); 2612 } 2613 2614 /** 2615 * Returns a new CompletableFuture that is already completed 2616 * exceptionally with the given exception. 2617 * 2618 * @param ex the exception 2619 * @param <U> the type of the value 2620 * @return the exceptionally completed CompletableFuture 2621 * @since 9 2622 * @hide 2623 */ 2624 // Android-changed: hidden failedFuture(Throwable ex)2625 public static <U> CompletableFuture<U> failedFuture(Throwable ex) { 2626 if (ex == null) throw new NullPointerException(); 2627 return new CompletableFuture<U>(new AltResult(ex)); 2628 } 2629 2630 /** 2631 * Returns a new CompletionStage that is already completed 2632 * exceptionally with the given exception and supports only those 2633 * methods in interface {@link CompletionStage}. 2634 * 2635 * @param ex the exception 2636 * @param <U> the type of the value 2637 * @return the exceptionally completed CompletionStage 2638 * @since 9 2639 * @hide 2640 */ 2641 // Android-changed: hidden failedStage(Throwable ex)2642 public static <U> CompletionStage<U> failedStage(Throwable ex) { 2643 if (ex == null) throw new NullPointerException(); 2644 return new MinimalStage<U>(new AltResult(ex)); 2645 } 2646 2647 /** 2648 * Singleton delay scheduler, used only for starting and 2649 * cancelling tasks. 2650 */ 2651 static final class Delayer { delay(Runnable command, long delay, TimeUnit unit)2652 static ScheduledFuture<?> delay(Runnable command, long delay, 2653 TimeUnit unit) { 2654 return delayer.schedule(command, delay, unit); 2655 } 2656 2657 static final class DaemonThreadFactory implements ThreadFactory { newThread(Runnable r)2658 public Thread newThread(Runnable r) { 2659 Thread t = new Thread(r); 2660 t.setDaemon(true); 2661 t.setName("CompletableFutureDelayScheduler"); 2662 return t; 2663 } 2664 } 2665 2666 static final ScheduledThreadPoolExecutor delayer; 2667 static { 2668 (delayer = new ScheduledThreadPoolExecutor( 2669 1, new DaemonThreadFactory())). 2670 setRemoveOnCancelPolicy(true); 2671 } 2672 } 2673 2674 // Little class-ified lambdas to better support monitoring 2675 2676 static final class DelayedExecutor implements Executor { 2677 final long delay; 2678 final TimeUnit unit; 2679 final Executor executor; DelayedExecutor(long delay, TimeUnit unit, Executor executor)2680 DelayedExecutor(long delay, TimeUnit unit, Executor executor) { 2681 this.delay = delay; this.unit = unit; this.executor = executor; 2682 } execute(Runnable r)2683 public void execute(Runnable r) { 2684 Delayer.delay(new TaskSubmitter(executor, r), delay, unit); 2685 } 2686 } 2687 2688 /** Action to submit user task */ 2689 static final class TaskSubmitter implements Runnable { 2690 final Executor executor; 2691 final Runnable action; TaskSubmitter(Executor executor, Runnable action)2692 TaskSubmitter(Executor executor, Runnable action) { 2693 this.executor = executor; 2694 this.action = action; 2695 } run()2696 public void run() { executor.execute(action); } 2697 } 2698 2699 /** Action to completeExceptionally on timeout */ 2700 static final class Timeout implements Runnable { 2701 final CompletableFuture<?> f; Timeout(CompletableFuture<?> f)2702 Timeout(CompletableFuture<?> f) { this.f = f; } run()2703 public void run() { 2704 if (f != null && !f.isDone()) 2705 f.completeExceptionally(new TimeoutException()); 2706 } 2707 } 2708 2709 /** Action to complete on timeout */ 2710 static final class DelayedCompleter<U> implements Runnable { 2711 final CompletableFuture<U> f; 2712 final U u; DelayedCompleter(CompletableFuture<U> f, U u)2713 DelayedCompleter(CompletableFuture<U> f, U u) { this.f = f; this.u = u; } run()2714 public void run() { 2715 if (f != null) 2716 f.complete(u); 2717 } 2718 } 2719 2720 /** Action to cancel unneeded timeouts */ 2721 static final class Canceller implements BiConsumer<Object, Throwable> { 2722 final Future<?> f; Canceller(Future<?> f)2723 Canceller(Future<?> f) { this.f = f; } accept(Object ignore, Throwable ex)2724 public void accept(Object ignore, Throwable ex) { 2725 if (ex == null && f != null && !f.isDone()) 2726 f.cancel(false); 2727 } 2728 } 2729 2730 /** 2731 * A subclass that just throws UOE for most non-CompletionStage methods. 2732 */ 2733 static final class MinimalStage<T> extends CompletableFuture<T> { MinimalStage()2734 MinimalStage() { } MinimalStage(Object r)2735 MinimalStage(Object r) { super(r); } newIncompleteFuture()2736 @Override public <U> CompletableFuture<U> newIncompleteFuture() { 2737 return new MinimalStage<U>(); } get()2738 @Override public T get() { 2739 throw new UnsupportedOperationException(); } get(long timeout, TimeUnit unit)2740 @Override public T get(long timeout, TimeUnit unit) { 2741 throw new UnsupportedOperationException(); } getNow(T valueIfAbsent)2742 @Override public T getNow(T valueIfAbsent) { 2743 throw new UnsupportedOperationException(); } join()2744 @Override public T join() { 2745 throw new UnsupportedOperationException(); } complete(T value)2746 @Override public boolean complete(T value) { 2747 throw new UnsupportedOperationException(); } completeExceptionally(Throwable ex)2748 @Override public boolean completeExceptionally(Throwable ex) { 2749 throw new UnsupportedOperationException(); } cancel(boolean mayInterruptIfRunning)2750 @Override public boolean cancel(boolean mayInterruptIfRunning) { 2751 throw new UnsupportedOperationException(); } obtrudeValue(T value)2752 @Override public void obtrudeValue(T value) { 2753 throw new UnsupportedOperationException(); } obtrudeException(Throwable ex)2754 @Override public void obtrudeException(Throwable ex) { 2755 throw new UnsupportedOperationException(); } isDone()2756 @Override public boolean isDone() { 2757 throw new UnsupportedOperationException(); } isCancelled()2758 @Override public boolean isCancelled() { 2759 throw new UnsupportedOperationException(); } isCompletedExceptionally()2760 @Override public boolean isCompletedExceptionally() { 2761 throw new UnsupportedOperationException(); } getNumberOfDependents()2762 @Override public int getNumberOfDependents() { 2763 throw new UnsupportedOperationException(); } completeAsync(Supplier<? extends T> supplier, Executor executor)2764 @Override public CompletableFuture<T> completeAsync 2765 (Supplier<? extends T> supplier, Executor executor) { 2766 throw new UnsupportedOperationException(); } completeAsync(Supplier<? extends T> supplier)2767 @Override public CompletableFuture<T> completeAsync 2768 (Supplier<? extends T> supplier) { 2769 throw new UnsupportedOperationException(); } orTimeout(long timeout, TimeUnit unit)2770 @Override public CompletableFuture<T> orTimeout 2771 (long timeout, TimeUnit unit) { 2772 throw new UnsupportedOperationException(); } completeOnTimeout(T value, long timeout, TimeUnit unit)2773 @Override public CompletableFuture<T> completeOnTimeout 2774 (T value, long timeout, TimeUnit unit) { 2775 throw new UnsupportedOperationException(); } 2776 } 2777 2778 // Unsafe mechanics 2779 private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe(); 2780 private static final long RESULT; 2781 private static final long STACK; 2782 private static final long NEXT; 2783 static { 2784 try { 2785 RESULT = U.objectFieldOffset 2786 (CompletableFuture.class.getDeclaredField("result")); 2787 STACK = U.objectFieldOffset 2788 (CompletableFuture.class.getDeclaredField("stack")); 2789 NEXT = U.objectFieldOffset 2790 (Completion.class.getDeclaredField("next")); 2791 } catch (ReflectiveOperationException e) { 2792 throw new Error(e); 2793 } 2794 2795 // Reduce the risk of rare disastrous classloading in first call to 2796 // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773 2797 Class<?> ensureLoaded = LockSupport.class; 2798 } 2799 } 2800