ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.190
Committed: Sun May 1 04:26:48 2016 UTC (8 years, 1 month ago) by jsr166
Branch: MAIN
Changes since 1.189: +12 -10 lines
Log Message:
move casts around, removing some errorprone warnings

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