ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.189
Committed: Tue Apr 19 22:55:29 2016 UTC (8 years, 1 month ago) by jsr166
Branch: MAIN
Changes since 1.188: +2 -2 lines
Log Message:
s~\bsun\.(misc\.Unsafe)\b~jdk.internal.$1~g;
s~\bputOrdered([A-Za-z]+)\b~put${1}Release~g

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