ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.213
Committed: Mon Sep 17 11:50:12 2018 UTC (5 years, 8 months ago) by dl
Branch: MAIN
Changes since 1.212: +120 -30 lines
Log Message:
Add exceptionally{Compose}{Async}

File Contents

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