ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jdk8/java/util/concurrent/CompletableFuture.java
Revision: 1.10
Committed: Mon Oct 1 03:58:18 2018 UTC (5 years, 7 months ago) by jsr166
Branch: MAIN
CVS Tags: HEAD
Changes since 1.9: +112 -30 lines
Log Message:
backport exceptionally* methods to fix 4jdk8-tck target

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