ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jdk8/java/util/concurrent/CompletableFuture.java
Revision: 1.6
Committed: Thu May 5 16:26:20 2016 UTC (8 years ago) by jsr166
Branch: MAIN
Changes since 1.5: +49 -49 lines
Log Message:
sync src/main to src/jdk8

File Contents

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