ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.225
Committed: Tue Jan 19 00:58:58 2021 UTC (3 years, 3 months ago) by jsr166
Branch: MAIN
CVS Tags: HEAD
Changes since 1.224: +6 -5 lines
Log Message:
8259796: timedGet: slighly cleaner code

File Contents

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