ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jdk8/java/util/concurrent/CompletableFuture.java
Revision: 1.2
Committed: Sat Apr 2 17:45:34 2016 UTC (8 years, 1 month ago) by dl
Branch: MAIN
Changes since 1.1: +140 -39 lines
Log Message:
reduce async overhead

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 fired
211 * Completions from stacks that might never be popped (see method
212 * postFire). Completion fields need not be declared as final or
213 * volatile because they are only visible to other threads upon
214 * safe publication.
215 */
216
217 volatile Object result; // Either the result or boxed AltResult
218 volatile Completion stack; // Top of Treiber stack of dependent actions
219
220 final boolean internalComplete(Object r) { // CAS from null to r
221 return U.compareAndSwapObject(this, RESULT, null, r);
222 }
223
224 final boolean casStack(Completion cmp, Completion val) {
225 return U.compareAndSwapObject(this, STACK, cmp, val);
226 }
227
228 /** Returns true if successfully pushed c onto stack. */
229 final boolean tryPushStack(Completion c) {
230 Completion h = stack;
231 lazySetNext(c, h);
232 return U.compareAndSwapObject(this, STACK, h, c);
233 }
234
235 /** Unconditionally pushes c onto stack, retrying if necessary. */
236 final void pushStack(Completion c) {
237 do {} while (!tryPushStack(c));
238 }
239
240 /* ------------- Encoding and decoding outcomes -------------- */
241
242 static final class AltResult { // See above
243 final Throwable ex; // null only for NIL
244 AltResult(Throwable x) { this.ex = x; }
245 }
246
247 /** The encoding of the null value. */
248 static final AltResult NIL = new AltResult(null);
249
250 /** Completes with the null value, unless already completed. */
251 final boolean completeNull() {
252 return U.compareAndSwapObject(this, RESULT, null,
253 NIL);
254 }
255
256 /** Returns the encoding of the given non-exceptional value. */
257 final Object encodeValue(T t) {
258 return (t == null) ? NIL : t;
259 }
260
261 /** Completes with a non-exceptional result, unless already completed. */
262 final boolean completeValue(T t) {
263 return U.compareAndSwapObject(this, RESULT, null,
264 (t == null) ? NIL : t);
265 }
266
267 /**
268 * Returns the encoding of the given (non-null) exception as a
269 * wrapped CompletionException unless it is one already.
270 */
271 static AltResult encodeThrowable(Throwable x) {
272 return new AltResult((x instanceof CompletionException) ? x :
273 new CompletionException(x));
274 }
275
276 /** Completes with an exceptional result, unless already completed. */
277 final boolean completeThrowable(Throwable x) {
278 return U.compareAndSwapObject(this, RESULT, null,
279 encodeThrowable(x));
280 }
281
282 /**
283 * Returns the encoding of the given (non-null) exception as a
284 * wrapped CompletionException unless it is one already. May
285 * return the given Object r (which must have been the result of a
286 * source future) if it is equivalent, i.e. if this is a simple
287 * relay of an existing CompletionException.
288 */
289 static Object encodeThrowable(Throwable x, Object r) {
290 if (!(x instanceof CompletionException))
291 x = new CompletionException(x);
292 else if (r instanceof AltResult && x == ((AltResult)r).ex)
293 return r;
294 return new AltResult(x);
295 }
296
297 /**
298 * Completes with the given (non-null) exceptional result as a
299 * wrapped CompletionException unless it is one already, unless
300 * already completed. May complete with the given Object r
301 * (which must have been the result of a source future) if it is
302 * equivalent, i.e. if this is a simple propagation of an
303 * existing CompletionException.
304 */
305 final boolean completeThrowable(Throwable x, Object r) {
306 return U.compareAndSwapObject(this, RESULT, null,
307 encodeThrowable(x, r));
308 }
309
310 /**
311 * Returns the encoding of the given arguments: if the exception
312 * is non-null, encodes as AltResult. Otherwise uses the given
313 * value, boxed as NIL if null.
314 */
315 Object encodeOutcome(T t, Throwable x) {
316 return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x);
317 }
318
319 /**
320 * Returns the encoding of a copied outcome; if exceptional,
321 * rewraps as a CompletionException, else returns argument.
322 */
323 static Object encodeRelay(Object r) {
324 Throwable x;
325 return (((r instanceof AltResult) &&
326 (x = ((AltResult)r).ex) != null &&
327 !(x instanceof CompletionException)) ?
328 new AltResult(new CompletionException(x)) : r);
329 }
330
331 /**
332 * Completes with r or a copy of r, unless already completed.
333 * If exceptional, r is first coerced to a CompletionException.
334 */
335 final boolean completeRelay(Object r) {
336 return U.compareAndSwapObject(this, RESULT, null,
337 encodeRelay(r));
338 }
339
340 /**
341 * Reports result using Future.get conventions.
342 */
343 private static <T> T reportGet(Object r)
344 throws InterruptedException, ExecutionException {
345 if (r == null) // by convention below, null means interrupted
346 throw new InterruptedException();
347 if (r instanceof AltResult) {
348 Throwable x, cause;
349 if ((x = ((AltResult)r).ex) == null)
350 return null;
351 if (x instanceof CancellationException)
352 throw (CancellationException)x;
353 if ((x instanceof CompletionException) &&
354 (cause = x.getCause()) != null)
355 x = cause;
356 throw new ExecutionException(x);
357 }
358 @SuppressWarnings("unchecked") T t = (T) r;
359 return t;
360 }
361
362 /**
363 * Decodes outcome to return result or throw unchecked exception.
364 */
365 private static <T> T reportJoin(Object r) {
366 if (r instanceof AltResult) {
367 Throwable x;
368 if ((x = ((AltResult)r).ex) == null)
369 return null;
370 if (x instanceof CancellationException)
371 throw (CancellationException)x;
372 if (x instanceof CompletionException)
373 throw (CompletionException)x;
374 throw new CompletionException(x);
375 }
376 @SuppressWarnings("unchecked") T t = (T) r;
377 return t;
378 }
379
380 /* ------------- Async task preliminaries -------------- */
381
382 /**
383 * A marker interface identifying asynchronous tasks produced by
384 * {@code async} methods. This may be useful for monitoring,
385 * debugging, and tracking asynchronous activities.
386 *
387 * @since 1.8
388 */
389 public static interface AsynchronousCompletionTask {
390 }
391
392 private static final boolean USE_COMMON_POOL =
393 (ForkJoinPool.getCommonPoolParallelism() > 1);
394
395 /**
396 * Default executor -- ForkJoinPool.commonPool() unless it cannot
397 * support parallelism.
398 */
399 private static final Executor ASYNC_POOL = USE_COMMON_POOL ?
400 ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
401
402 /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
403 static final class ThreadPerTaskExecutor implements Executor {
404 public void execute(Runnable r) { new Thread(r).start(); }
405 }
406
407 /**
408 * Null-checks user executor argument, and translates uses of
409 * commonPool to ASYNC_POOL in case parallelism disabled.
410 */
411 static Executor screenExecutor(Executor e) {
412 if (!USE_COMMON_POOL && e == ForkJoinPool.commonPool())
413 return ASYNC_POOL;
414 if (e == null) throw new NullPointerException();
415 return e;
416 }
417
418 // Modes for Completion.tryFire. Signedness matters.
419 static final int SYNC = 0;
420 static final int ASYNC = 1;
421 static final int NESTED = -1;
422
423 /* ------------- Base Completion classes and operations -------------- */
424
425 @SuppressWarnings("serial")
426 abstract static class Completion extends ForkJoinTask<Void>
427 implements Runnable, AsynchronousCompletionTask {
428 volatile Completion next; // Treiber stack link
429
430 /**
431 * Performs completion action if triggered, returning a
432 * dependent that may need propagation, if one exists.
433 *
434 * @param mode SYNC, ASYNC, or NESTED
435 */
436 abstract CompletableFuture<?> tryFire(int mode);
437
438 /** Returns true if possibly still triggerable. Used by cleanStack. */
439 abstract boolean isLive();
440
441 public final void run() { tryFire(ASYNC); }
442 public final boolean exec() { tryFire(ASYNC); return false; }
443 public final Void getRawResult() { return null; }
444 public final void setRawResult(Void v) {}
445 }
446
447 static void lazySetNext(Completion c, Completion next) {
448 U.putOrderedObject(c, NEXT, next);
449 }
450
451 /**
452 * Pops and tries to trigger all reachable dependents. Call only
453 * when known to be done.
454 */
455 final void postComplete() {
456 /*
457 * On each step, variable f holds current dependents to pop
458 * and run. It is extended along only one path at a time,
459 * pushing others to avoid unbounded recursion.
460 */
461 CompletableFuture<?> f = this; Completion h;
462 while ((h = f.stack) != null ||
463 (f != this && (h = (f = this).stack) != null)) {
464 CompletableFuture<?> d; Completion t;
465 if (f.casStack(h, t = h.next)) {
466 if (t != null) {
467 if (f != this) {
468 pushStack(h);
469 continue;
470 }
471 h.next = null; // detach
472 }
473 f = (d = h.tryFire(NESTED)) == null ? this : d;
474 }
475 }
476 }
477
478 /** Traverses stack and unlinks dead Completions. */
479 final void cleanStack() {
480 for (Completion p = null, q = stack; q != null;) {
481 Completion s = q.next;
482 if (q.isLive()) {
483 p = q;
484 q = s;
485 }
486 else if (p == null) {
487 casStack(q, s);
488 q = stack;
489 }
490 else {
491 p.next = s;
492 if (p.isLive())
493 q = s;
494 else {
495 p = null; // restart
496 q = stack;
497 }
498 }
499 }
500 }
501
502 /* ------------- One-input Completions -------------- */
503
504 /** A Completion with a source, dependent, and executor. */
505 @SuppressWarnings("serial")
506 abstract static class UniCompletion<T,V> extends Completion {
507 Executor executor; // executor to use (null if none)
508 CompletableFuture<V> dep; // the dependent to complete
509 CompletableFuture<T> src; // source for action
510
511 UniCompletion(Executor executor, CompletableFuture<V> dep,
512 CompletableFuture<T> src) {
513 this.executor = executor; this.dep = dep; this.src = src;
514 }
515
516 /**
517 * Returns true if action can be run. Call only when known to
518 * be triggerable. Uses FJ tag bit to ensure that only one
519 * thread claims ownership. If async, starts as task -- a
520 * later call to tryFire will run action.
521 */
522 final boolean claim() {
523 Executor e = executor;
524 if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {
525 if (e == null)
526 return true;
527 executor = null; // disable
528 e.execute(this);
529 }
530 return false;
531 }
532
533 final boolean isLive() { return dep != null; }
534 }
535
536 /** Pushes the given completion (if it exists) unless done. */
537 final void push(UniCompletion<?,?> c) {
538 if (c != null) {
539 while (result == null && !tryPushStack(c))
540 lazySetNext(c, null); // clear on failure
541 }
542 }
543
544 /**
545 * Post-processing by dependent after successful UniCompletion
546 * tryFire. Tries to clean stack of source a, and then either runs
547 * postComplete or returns this to caller, depending on mode.
548 */
549 final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
550 if (a != null && a.stack != null) {
551 if (a.result == null)
552 a.cleanStack();
553 else if (mode >= 0)
554 a.postComplete();
555 }
556 if (result != null && stack != null) {
557 if (mode < 0)
558 return this;
559 else
560 postComplete();
561 }
562 return null;
563 }
564
565 @SuppressWarnings("serial")
566 static final class UniApply<T,V> extends UniCompletion<T,V> {
567 Function<? super T,? extends V> fn;
568 UniApply(Executor executor, CompletableFuture<V> dep,
569 CompletableFuture<T> src,
570 Function<? super T,? extends V> fn) {
571 super(executor, dep, src); this.fn = fn;
572 }
573 final CompletableFuture<V> tryFire(int mode) {
574 CompletableFuture<V> d; CompletableFuture<T> a;
575 if ((d = dep) == null ||
576 !d.uniApply(a = src, fn, mode > 0 ? null : this))
577 return null;
578 dep = null; src = null; fn = null;
579 return d.postFire(a, mode);
580 }
581 }
582
583 final <S> boolean uniApply(CompletableFuture<S> a,
584 Function<? super S,? extends T> f,
585 UniApply<S,T> c) {
586 Object r; Throwable x;
587 if (a == null || (r = a.result) == null || f == null)
588 return false;
589 tryComplete: if (result == null) {
590 if (r instanceof AltResult) {
591 if ((x = ((AltResult)r).ex) != null) {
592 completeThrowable(x, r);
593 break tryComplete;
594 }
595 r = null;
596 }
597 try {
598 if (c != null && !c.claim())
599 return false;
600 @SuppressWarnings("unchecked") S s = (S) r;
601 completeValue(f.apply(s));
602 } catch (Throwable ex) {
603 completeThrowable(ex);
604 }
605 }
606 return true;
607 }
608
609 private <V> CompletableFuture<V> uniApplyStage(
610 Executor e, Function<? super T,? extends V> f) {
611 if (f == null) throw new NullPointerException();
612 CompletableFuture<V> d = newIncompleteFuture();
613 if (e != null || !d.uniApply(this, f, null)) {
614 UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
615 if (e != null && result != null) {
616 try {
617 e.execute(c);
618 } catch (Throwable ex) {
619 d.completeThrowable(ex);
620 }
621 }
622 else {
623 push(c);
624 c.tryFire(SYNC);
625 }
626 }
627 return d;
628 }
629
630 @SuppressWarnings("serial")
631 static final class UniAccept<T> extends UniCompletion<T,Void> {
632 Consumer<? super T> fn;
633 UniAccept(Executor executor, CompletableFuture<Void> dep,
634 CompletableFuture<T> src, Consumer<? super T> fn) {
635 super(executor, dep, src); this.fn = fn;
636 }
637 final CompletableFuture<Void> tryFire(int mode) {
638 CompletableFuture<Void> d; CompletableFuture<T> a;
639 if ((d = dep) == null ||
640 !d.uniAccept(a = src, fn, mode > 0 ? null : this))
641 return null;
642 dep = null; src = null; fn = null;
643 return d.postFire(a, mode);
644 }
645 }
646
647 final <S> boolean uniAccept(CompletableFuture<S> a,
648 Consumer<? super S> f, UniAccept<S> c) {
649 Object r; Throwable x;
650 if (a == null || (r = a.result) == null || f == null)
651 return false;
652 tryComplete: if (result == null) {
653 if (r instanceof AltResult) {
654 if ((x = ((AltResult)r).ex) != null) {
655 completeThrowable(x, r);
656 break tryComplete;
657 }
658 r = null;
659 }
660 try {
661 if (c != null && !c.claim())
662 return false;
663 @SuppressWarnings("unchecked") S s = (S) r;
664 f.accept(s);
665 completeNull();
666 } catch (Throwable ex) {
667 completeThrowable(ex);
668 }
669 }
670 return true;
671 }
672
673 private CompletableFuture<Void> uniAcceptStage(Executor e,
674 Consumer<? super T> f) {
675 if (f == null) throw new NullPointerException();
676 CompletableFuture<Void> d = newIncompleteFuture();
677 if (e != null || !d.uniAccept(this, f, null)) {
678 UniAccept<T> c = new UniAccept<T>(e, d, this, f);
679 if (e != null && result != null) {
680 try {
681 e.execute(c);
682 } catch (Throwable ex) {
683 d.completeThrowable(ex);
684 }
685 }
686 else {
687 push(c);
688 c.tryFire(SYNC);
689 }
690 }
691 return d;
692 }
693
694 @SuppressWarnings("serial")
695 static final class UniRun<T> extends UniCompletion<T,Void> {
696 Runnable fn;
697 UniRun(Executor executor, CompletableFuture<Void> dep,
698 CompletableFuture<T> src, Runnable fn) {
699 super(executor, dep, src); this.fn = fn;
700 }
701 final CompletableFuture<Void> tryFire(int mode) {
702 CompletableFuture<Void> d; CompletableFuture<T> a;
703 if ((d = dep) == null ||
704 !d.uniRun(a = src, fn, mode > 0 ? null : this))
705 return null;
706 dep = null; src = null; fn = null;
707 return d.postFire(a, mode);
708 }
709 }
710
711 final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {
712 Object r; Throwable x;
713 if (a == null || (r = a.result) == null || f == null)
714 return false;
715 if (result == null) {
716 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
717 completeThrowable(x, r);
718 else
719 try {
720 if (c != null && !c.claim())
721 return false;
722 f.run();
723 completeNull();
724 } catch (Throwable ex) {
725 completeThrowable(ex);
726 }
727 }
728 return true;
729 }
730
731 private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
732 if (f == null) throw new NullPointerException();
733 CompletableFuture<Void> d = newIncompleteFuture();
734 if (e != null || !d.uniRun(this, f, null)) {
735 UniRun<T> c = new UniRun<T>(e, d, this, f);
736 if (e != null && result != null) {
737 try {
738 e.execute(c);
739 } catch (Throwable ex) {
740 d.completeThrowable(ex);
741 }
742 }
743 else {
744 push(c);
745 c.tryFire(SYNC);
746 }
747 }
748 return d;
749 }
750
751 @SuppressWarnings("serial")
752 static final class UniWhenComplete<T> extends UniCompletion<T,T> {
753 BiConsumer<? super T, ? super Throwable> fn;
754 UniWhenComplete(Executor executor, CompletableFuture<T> dep,
755 CompletableFuture<T> src,
756 BiConsumer<? super T, ? super Throwable> fn) {
757 super(executor, dep, src); this.fn = fn;
758 }
759 final CompletableFuture<T> tryFire(int mode) {
760 CompletableFuture<T> d; CompletableFuture<T> a;
761 if ((d = dep) == null ||
762 !d.uniWhenComplete(a = src, fn, mode > 0 ? null : this))
763 return null;
764 dep = null; src = null; fn = null;
765 return d.postFire(a, mode);
766 }
767 }
768
769 final boolean uniWhenComplete(CompletableFuture<T> a,
770 BiConsumer<? super T,? super Throwable> f,
771 UniWhenComplete<T> c) {
772 Object r; T t; Throwable x = null;
773 if (a == null || (r = a.result) == null || f == null)
774 return false;
775 if (result == null) {
776 try {
777 if (c != null && !c.claim())
778 return false;
779 if (r instanceof AltResult) {
780 x = ((AltResult)r).ex;
781 t = null;
782 } else {
783 @SuppressWarnings("unchecked") T tr = (T) r;
784 t = tr;
785 }
786 f.accept(t, x);
787 if (x == null) {
788 internalComplete(r);
789 return true;
790 }
791 } catch (Throwable ex) {
792 if (x == null)
793 x = ex;
794 else if (x != ex)
795 x.addSuppressed(ex);
796 }
797 completeThrowable(x, r);
798 }
799 return true;
800 }
801
802 private CompletableFuture<T> uniWhenCompleteStage(
803 Executor e, BiConsumer<? super T, ? super Throwable> f) {
804 if (f == null) throw new NullPointerException();
805 CompletableFuture<T> d = newIncompleteFuture();
806 if (e != null || !d.uniWhenComplete(this, f, null)) {
807 UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);
808 if (e != null && result != null) {
809 try {
810 e.execute(c);
811 } catch (Throwable ex) {
812 d.completeThrowable(ex);
813 }
814 }
815 else {
816 push(c);
817 c.tryFire(SYNC);
818 }
819 }
820 return d;
821 }
822
823 @SuppressWarnings("serial")
824 static final class UniHandle<T,V> extends UniCompletion<T,V> {
825 BiFunction<? super T, Throwable, ? extends V> fn;
826 UniHandle(Executor executor, CompletableFuture<V> dep,
827 CompletableFuture<T> src,
828 BiFunction<? super T, Throwable, ? extends V> fn) {
829 super(executor, dep, src); this.fn = fn;
830 }
831 final CompletableFuture<V> tryFire(int mode) {
832 CompletableFuture<V> d; CompletableFuture<T> a;
833 if ((d = dep) == null ||
834 !d.uniHandle(a = src, fn, mode > 0 ? null : this))
835 return null;
836 dep = null; src = null; fn = null;
837 return d.postFire(a, mode);
838 }
839 }
840
841 final <S> boolean uniHandle(CompletableFuture<S> a,
842 BiFunction<? super S, Throwable, ? extends T> f,
843 UniHandle<S,T> c) {
844 Object r; S s; Throwable x;
845 if (a == null || (r = a.result) == null || f == null)
846 return false;
847 if (result == null) {
848 try {
849 if (c != null && !c.claim())
850 return false;
851 if (r instanceof AltResult) {
852 x = ((AltResult)r).ex;
853 s = null;
854 } else {
855 x = null;
856 @SuppressWarnings("unchecked") S ss = (S) r;
857 s = ss;
858 }
859 completeValue(f.apply(s, x));
860 } catch (Throwable ex) {
861 completeThrowable(ex);
862 }
863 }
864 return true;
865 }
866
867 private <V> CompletableFuture<V> uniHandleStage(
868 Executor e, BiFunction<? super T, Throwable, ? extends V> f) {
869 if (f == null) throw new NullPointerException();
870 CompletableFuture<V> d = newIncompleteFuture();
871 if (e != null || !d.uniHandle(this, f, null)) {
872 UniHandle<T,V> c = new UniHandle<T,V>(e, d, this, f);
873 if (e != null && result != null) {
874 try {
875 e.execute(c);
876 } catch (Throwable ex) {
877 d.completeThrowable(ex);
878 }
879 }
880 else {
881 push(c);
882 c.tryFire(SYNC);
883 }
884 }
885 return d;
886 }
887
888 @SuppressWarnings("serial")
889 static final class UniExceptionally<T> extends UniCompletion<T,T> {
890 Function<? super Throwable, ? extends T> fn;
891 UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src,
892 Function<? super Throwable, ? extends T> fn) {
893 super(null, dep, src); this.fn = fn;
894 }
895 final CompletableFuture<T> tryFire(int mode) { // never ASYNC
896 // assert mode != ASYNC;
897 CompletableFuture<T> d; CompletableFuture<T> a;
898 if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this))
899 return null;
900 dep = null; src = null; fn = null;
901 return d.postFire(a, mode);
902 }
903 }
904
905 final boolean uniExceptionally(CompletableFuture<T> a,
906 Function<? super Throwable, ? extends T> f,
907 UniExceptionally<T> c) {
908 Object r; Throwable x;
909 if (a == null || (r = a.result) == null || f == null)
910 return false;
911 if (result == null) {
912 try {
913 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) {
914 if (c != null && !c.claim())
915 return false;
916 completeValue(f.apply(x));
917 } else
918 internalComplete(r);
919 } catch (Throwable ex) {
920 completeThrowable(ex);
921 }
922 }
923 return true;
924 }
925
926 private CompletableFuture<T> uniExceptionallyStage(
927 Function<Throwable, ? extends T> f) {
928 if (f == null) throw new NullPointerException();
929 CompletableFuture<T> d = newIncompleteFuture();
930 if (!d.uniExceptionally(this, f, null)) {
931 UniExceptionally<T> c = new UniExceptionally<T>(d, this, f);
932 push(c);
933 c.tryFire(SYNC);
934 }
935 return d;
936 }
937
938 @SuppressWarnings("serial")
939 static final class UniRelay<T> extends UniCompletion<T,T> { // for Compose
940 UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src) {
941 super(null, dep, src);
942 }
943 final CompletableFuture<T> tryFire(int mode) {
944 CompletableFuture<T> d; CompletableFuture<T> a;
945 if ((d = dep) == null || !d.uniRelay(a = src))
946 return null;
947 src = null; dep = null;
948 return d.postFire(a, mode);
949 }
950 }
951
952 final boolean uniRelay(CompletableFuture<T> a) {
953 Object r;
954 if (a == null || (r = a.result) == null)
955 return false;
956 if (result == null) // no need to claim
957 completeRelay(r);
958 return true;
959 }
960
961 private CompletableFuture<T> uniCopyStage() {
962 Object r;
963 CompletableFuture<T> d = newIncompleteFuture();
964 if ((r = result) != null)
965 d.completeRelay(r);
966 else {
967 UniRelay<T> c = new UniRelay<T>(d, this);
968 push(c);
969 c.tryFire(SYNC);
970 }
971 return d;
972 }
973
974 private MinimalStage<T> uniAsMinimalStage() {
975 Object r;
976 if ((r = result) != null)
977 return new MinimalStage<T>(encodeRelay(r));
978 MinimalStage<T> d = new MinimalStage<T>();
979 UniRelay<T> c = new UniRelay<T>(d, this);
980 push(c);
981 c.tryFire(SYNC);
982 return d;
983 }
984
985 @SuppressWarnings("serial")
986 static final class UniCompose<T,V> extends UniCompletion<T,V> {
987 Function<? super T, ? extends CompletionStage<V>> fn;
988 UniCompose(Executor executor, CompletableFuture<V> dep,
989 CompletableFuture<T> src,
990 Function<? super T, ? extends CompletionStage<V>> fn) {
991 super(executor, dep, src); this.fn = fn;
992 }
993 final CompletableFuture<V> tryFire(int mode) {
994 CompletableFuture<V> d; CompletableFuture<T> a;
995 if ((d = dep) == null ||
996 !d.uniCompose(a = src, fn, mode > 0 ? null : this))
997 return null;
998 dep = null; src = null; fn = null;
999 return d.postFire(a, mode);
1000 }
1001 }
1002
1003 final <S> boolean uniCompose(
1004 CompletableFuture<S> a,
1005 Function<? super S, ? extends CompletionStage<T>> f,
1006 UniCompose<S,T> c) {
1007 Object r; Throwable x;
1008 if (a == null || (r = a.result) == null || f == null)
1009 return false;
1010 tryComplete: if (result == null) {
1011 if (r instanceof AltResult) {
1012 if ((x = ((AltResult)r).ex) != null) {
1013 completeThrowable(x, r);
1014 break tryComplete;
1015 }
1016 r = null;
1017 }
1018 try {
1019 if (c != null && !c.claim())
1020 return false;
1021 @SuppressWarnings("unchecked") S s = (S) r;
1022 CompletableFuture<T> g = f.apply(s).toCompletableFuture();
1023 if (g.result == null || !uniRelay(g)) {
1024 UniRelay<T> copy = new UniRelay<T>(this, g);
1025 g.push(copy);
1026 copy.tryFire(SYNC);
1027 if (result == null)
1028 return false;
1029 }
1030 } catch (Throwable ex) {
1031 completeThrowable(ex);
1032 }
1033 }
1034 return true;
1035 }
1036
1037 private <V> CompletableFuture<V> uniComposeStage(
1038 Executor e, Function<? super T, ? extends CompletionStage<V>> f) {
1039 if (f == null) throw new NullPointerException();
1040 Object r, s; Throwable x;
1041 CompletableFuture<V> d = newIncompleteFuture();
1042 if ((r = result) != null && e == null) {
1043 if (r instanceof AltResult) {
1044 if ((x = ((AltResult)r).ex) != null) {
1045 d.result = encodeThrowable(x, r);
1046 return d;
1047 }
1048 r = null;
1049 }
1050 try {
1051 @SuppressWarnings("unchecked") T t = (T) r;
1052 CompletableFuture<V> g = f.apply(t).toCompletableFuture();
1053 if ((s = g.result) != null)
1054 d.completeRelay(s);
1055 else {
1056 UniRelay<V> c = new UniRelay<V>(d, g);
1057 g.push(c);
1058 c.tryFire(SYNC);
1059 }
1060 return d;
1061 } catch (Throwable ex) {
1062 d.result = encodeThrowable(ex);
1063 return d;
1064 }
1065 }
1066 UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
1067 if (r != null && e != null) {
1068 try {
1069 e.execute(new UniCompose<T,V>(null, d, this, f));
1070 } catch (Throwable ex) {
1071 d.completeThrowable(ex);
1072 }
1073 }
1074 else {
1075 push(c);
1076 c.tryFire(SYNC);
1077 }
1078 return d;
1079 }
1080
1081 /* ------------- Two-input Completions -------------- */
1082
1083 /** A Completion for an action with two sources */
1084 @SuppressWarnings("serial")
1085 abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> {
1086 CompletableFuture<U> snd; // second source for action
1087 BiCompletion(Executor executor, CompletableFuture<V> dep,
1088 CompletableFuture<T> src, CompletableFuture<U> snd) {
1089 super(executor, dep, src); this.snd = snd;
1090 }
1091 }
1092
1093 /** A Completion delegating to a BiCompletion */
1094 @SuppressWarnings("serial")
1095 static final class CoCompletion extends Completion {
1096 BiCompletion<?,?,?> base;
1097 CoCompletion(BiCompletion<?,?,?> base) { this.base = base; }
1098 final CompletableFuture<?> tryFire(int mode) {
1099 BiCompletion<?,?,?> c; CompletableFuture<?> d;
1100 if ((c = base) == null || (d = c.tryFire(mode)) == null)
1101 return null;
1102 base = null; // detach
1103 return d;
1104 }
1105 final boolean isLive() {
1106 BiCompletion<?,?,?> c;
1107 return (c = base) != null && c.dep != null;
1108 }
1109 }
1110
1111 /** Pushes completion to this and b unless both done. */
1112 final void bipush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
1113 if (c != null) {
1114 Object r;
1115 while ((r = result) == null && !tryPushStack(c))
1116 lazySetNext(c, null); // clear on failure
1117 if (b != null && b != this && b.result == null) {
1118 Completion q = (r != null) ? c : new CoCompletion(c);
1119 while (b.result == null && !b.tryPushStack(q))
1120 lazySetNext(q, null); // clear on failure
1121 }
1122 }
1123 }
1124
1125 /** Post-processing after successful BiCompletion tryFire. */
1126 final CompletableFuture<T> postFire(CompletableFuture<?> a,
1127 CompletableFuture<?> b, int mode) {
1128 if (b != null && b.stack != null) { // clean second source
1129 if (b.result == null)
1130 b.cleanStack();
1131 else if (mode >= 0)
1132 b.postComplete();
1133 }
1134 return postFire(a, mode);
1135 }
1136
1137 @SuppressWarnings("serial")
1138 static final class BiApply<T,U,V> extends BiCompletion<T,U,V> {
1139 BiFunction<? super T,? super U,? extends V> fn;
1140 BiApply(Executor executor, CompletableFuture<V> dep,
1141 CompletableFuture<T> src, CompletableFuture<U> snd,
1142 BiFunction<? super T,? super U,? extends V> fn) {
1143 super(executor, dep, src, snd); this.fn = fn;
1144 }
1145 final CompletableFuture<V> tryFire(int mode) {
1146 CompletableFuture<V> d;
1147 CompletableFuture<T> a;
1148 CompletableFuture<U> b;
1149 if ((d = dep) == null ||
1150 !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this))
1151 return null;
1152 dep = null; src = null; snd = null; fn = null;
1153 return d.postFire(a, b, mode);
1154 }
1155 }
1156
1157 final <R,S> boolean biApply(CompletableFuture<R> a,
1158 CompletableFuture<S> b,
1159 BiFunction<? super R,? super S,? extends T> f,
1160 BiApply<R,S,T> c) {
1161 Object r, s; Throwable x;
1162 if (a == null || (r = a.result) == null ||
1163 b == null || (s = b.result) == null || f == null)
1164 return false;
1165 tryComplete: if (result == null) {
1166 if (r instanceof AltResult) {
1167 if ((x = ((AltResult)r).ex) != null) {
1168 completeThrowable(x, r);
1169 break tryComplete;
1170 }
1171 r = null;
1172 }
1173 if (s instanceof AltResult) {
1174 if ((x = ((AltResult)s).ex) != null) {
1175 completeThrowable(x, s);
1176 break tryComplete;
1177 }
1178 s = null;
1179 }
1180 try {
1181 if (c != null && !c.claim())
1182 return false;
1183 @SuppressWarnings("unchecked") R rr = (R) r;
1184 @SuppressWarnings("unchecked") S ss = (S) s;
1185 completeValue(f.apply(rr, ss));
1186 } catch (Throwable ex) {
1187 completeThrowable(ex);
1188 }
1189 }
1190 return true;
1191 }
1192
1193 private <U,V> CompletableFuture<V> biApplyStage(
1194 Executor e, CompletionStage<U> o,
1195 BiFunction<? super T,? super U,? extends V> f) {
1196 CompletableFuture<U> b;
1197 if (f == null || (b = o.toCompletableFuture()) == null)
1198 throw new NullPointerException();
1199 CompletableFuture<V> d = newIncompleteFuture();
1200 if (e != null || !d.biApply(this, b, f, null)) {
1201 BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);
1202 if (e != null && result != null && b.result != null) {
1203 try {
1204 e.execute(c);
1205 } catch (Throwable ex) {
1206 d.completeThrowable(ex);
1207 }
1208 }
1209 else {
1210 bipush(b, c);
1211 c.tryFire(SYNC);
1212 }
1213 }
1214 return d;
1215 }
1216
1217 @SuppressWarnings("serial")
1218 static final class BiAccept<T,U> extends BiCompletion<T,U,Void> {
1219 BiConsumer<? super T,? super U> fn;
1220 BiAccept(Executor executor, CompletableFuture<Void> dep,
1221 CompletableFuture<T> src, CompletableFuture<U> snd,
1222 BiConsumer<? super T,? super U> fn) {
1223 super(executor, dep, src, snd); this.fn = fn;
1224 }
1225 final CompletableFuture<Void> tryFire(int mode) {
1226 CompletableFuture<Void> d;
1227 CompletableFuture<T> a;
1228 CompletableFuture<U> b;
1229 if ((d = dep) == null ||
1230 !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1231 return null;
1232 dep = null; src = null; snd = null; fn = null;
1233 return d.postFire(a, b, mode);
1234 }
1235 }
1236
1237 final <R,S> boolean biAccept(CompletableFuture<R> a,
1238 CompletableFuture<S> b,
1239 BiConsumer<? super R,? super S> f,
1240 BiAccept<R,S> c) {
1241 Object r, s; Throwable x;
1242 if (a == null || (r = a.result) == null ||
1243 b == null || (s = b.result) == null || f == null)
1244 return false;
1245 tryComplete: if (result == null) {
1246 if (r instanceof AltResult) {
1247 if ((x = ((AltResult)r).ex) != null) {
1248 completeThrowable(x, r);
1249 break tryComplete;
1250 }
1251 r = null;
1252 }
1253 if (s instanceof AltResult) {
1254 if ((x = ((AltResult)s).ex) != null) {
1255 completeThrowable(x, s);
1256 break tryComplete;
1257 }
1258 s = null;
1259 }
1260 try {
1261 if (c != null && !c.claim())
1262 return false;
1263 @SuppressWarnings("unchecked") R rr = (R) r;
1264 @SuppressWarnings("unchecked") S ss = (S) s;
1265 f.accept(rr, ss);
1266 completeNull();
1267 } catch (Throwable ex) {
1268 completeThrowable(ex);
1269 }
1270 }
1271 return true;
1272 }
1273
1274 private <U> CompletableFuture<Void> biAcceptStage(
1275 Executor e, CompletionStage<U> o,
1276 BiConsumer<? super T,? super U> f) {
1277 CompletableFuture<U> b;
1278 if (f == null || (b = o.toCompletableFuture()) == null)
1279 throw new NullPointerException();
1280 CompletableFuture<Void> d = newIncompleteFuture();
1281 if (e != null || !d.biAccept(this, b, f, null)) {
1282 BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
1283 if (e != null && result != null && b.result != null) {
1284 try {
1285 e.execute(c);
1286 } catch (Throwable ex) {
1287 d.completeThrowable(ex);
1288 }
1289 }
1290 else {
1291 bipush(b, c);
1292 c.tryFire(SYNC);
1293 }
1294 }
1295 return d;
1296 }
1297
1298 @SuppressWarnings("serial")
1299 static final class BiRun<T,U> extends BiCompletion<T,U,Void> {
1300 Runnable fn;
1301 BiRun(Executor executor, CompletableFuture<Void> dep,
1302 CompletableFuture<T> src,
1303 CompletableFuture<U> snd,
1304 Runnable fn) {
1305 super(executor, dep, src, snd); this.fn = fn;
1306 }
1307 final CompletableFuture<Void> tryFire(int mode) {
1308 CompletableFuture<Void> d;
1309 CompletableFuture<T> a;
1310 CompletableFuture<U> b;
1311 if ((d = dep) == null ||
1312 !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this))
1313 return null;
1314 dep = null; src = null; snd = null; fn = null;
1315 return d.postFire(a, b, mode);
1316 }
1317 }
1318
1319 final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b,
1320 Runnable f, BiRun<?,?> c) {
1321 Object r, s; Throwable x;
1322 if (a == null || (r = a.result) == null ||
1323 b == null || (s = b.result) == null || f == null)
1324 return false;
1325 if (result == null) {
1326 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1327 completeThrowable(x, r);
1328 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1329 completeThrowable(x, s);
1330 else
1331 try {
1332 if (c != null && !c.claim())
1333 return false;
1334 f.run();
1335 completeNull();
1336 } catch (Throwable ex) {
1337 completeThrowable(ex);
1338 }
1339 }
1340 return true;
1341 }
1342
1343 private CompletableFuture<Void> biRunStage(Executor e, CompletionStage<?> o,
1344 Runnable f) {
1345 CompletableFuture<?> b;
1346 if (f == null || (b = o.toCompletableFuture()) == null)
1347 throw new NullPointerException();
1348 CompletableFuture<Void> d = newIncompleteFuture();
1349 if (e != null || !d.biRun(this, b, f, null)) {
1350 BiRun<T,?> c = new BiRun<>(e, d, this, b, f);
1351 if (e != null && result != null && b.result != null) {
1352 try {
1353 e.execute(c);
1354 } catch (Throwable ex) {
1355 d.completeThrowable(ex);
1356 }
1357 }
1358 else {
1359 bipush(b, c);
1360 c.tryFire(SYNC);
1361 }
1362 }
1363 return d;
1364 }
1365
1366 @SuppressWarnings("serial")
1367 static final class BiRelay<T,U> extends BiCompletion<T,U,Void> { // for And
1368 BiRelay(CompletableFuture<Void> dep,
1369 CompletableFuture<T> src,
1370 CompletableFuture<U> snd) {
1371 super(null, dep, src, snd);
1372 }
1373 final CompletableFuture<Void> tryFire(int mode) {
1374 CompletableFuture<Void> d;
1375 CompletableFuture<T> a;
1376 CompletableFuture<U> b;
1377 if ((d = dep) == null || !d.biRelay(a = src, b = snd))
1378 return null;
1379 src = null; snd = null; dep = null;
1380 return d.postFire(a, b, mode);
1381 }
1382 }
1383
1384 boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1385 Object r, s; Throwable x;
1386 if (a == null || (r = a.result) == null ||
1387 b == null || (s = b.result) == null)
1388 return false;
1389 if (result == null) {
1390 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1391 completeThrowable(x, r);
1392 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1393 completeThrowable(x, s);
1394 else
1395 completeNull();
1396 }
1397 return true;
1398 }
1399
1400 /** Recursively constructs a tree of completions. */
1401 static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs,
1402 int lo, int hi) {
1403 CompletableFuture<Void> d = new CompletableFuture<Void>();
1404 if (lo > hi) // empty
1405 d.result = NIL;
1406 else {
1407 CompletableFuture<?> a, b;
1408 int mid = (lo + hi) >>> 1;
1409 if ((a = (lo == mid ? cfs[lo] :
1410 andTree(cfs, lo, mid))) == null ||
1411 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1412 andTree(cfs, mid+1, hi))) == null)
1413 throw new NullPointerException();
1414 if (!d.biRelay(a, b)) {
1415 BiRelay<?,?> c = new BiRelay<>(d, a, b);
1416 a.bipush(b, c);
1417 c.tryFire(SYNC);
1418 }
1419 }
1420 return d;
1421 }
1422
1423 /* ------------- Projected (Ored) BiCompletions -------------- */
1424
1425 /** Pushes completion to this and b unless either done. */
1426 final void orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
1427 if (c != null) {
1428 while ((b == null || b.result == null) && result == null) {
1429 if (tryPushStack(c)) {
1430 if (b != null && b != this && b.result == null) {
1431 Completion q = new CoCompletion(c);
1432 while (result == null && b.result == null &&
1433 !b.tryPushStack(q))
1434 lazySetNext(q, null); // clear on failure
1435 }
1436 break;
1437 }
1438 lazySetNext(c, null); // clear on failure
1439 }
1440 }
1441 }
1442
1443 @SuppressWarnings("serial")
1444 static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> {
1445 Function<? super T,? extends V> fn;
1446 OrApply(Executor executor, CompletableFuture<V> dep,
1447 CompletableFuture<T> src,
1448 CompletableFuture<U> snd,
1449 Function<? super T,? extends V> fn) {
1450 super(executor, dep, src, snd); this.fn = fn;
1451 }
1452 final CompletableFuture<V> tryFire(int mode) {
1453 CompletableFuture<V> d;
1454 CompletableFuture<T> a;
1455 CompletableFuture<U> b;
1456 if ((d = dep) == null ||
1457 !d.orApply(a = src, b = snd, fn, mode > 0 ? null : this))
1458 return null;
1459 dep = null; src = null; snd = null; fn = null;
1460 return d.postFire(a, b, mode);
1461 }
1462 }
1463
1464 final <R,S extends R> boolean orApply(CompletableFuture<R> a,
1465 CompletableFuture<S> b,
1466 Function<? super R, ? extends T> f,
1467 OrApply<R,S,T> c) {
1468 Object r; Throwable x;
1469 if (a == null || b == null ||
1470 ((r = a.result) == null && (r = b.result) == null) || f == null)
1471 return false;
1472 tryComplete: if (result == null) {
1473 try {
1474 if (c != null && !c.claim())
1475 return false;
1476 if (r instanceof AltResult) {
1477 if ((x = ((AltResult)r).ex) != null) {
1478 completeThrowable(x, r);
1479 break tryComplete;
1480 }
1481 r = null;
1482 }
1483 @SuppressWarnings("unchecked") R rr = (R) r;
1484 completeValue(f.apply(rr));
1485 } catch (Throwable ex) {
1486 completeThrowable(ex);
1487 }
1488 }
1489 return true;
1490 }
1491
1492 private <U extends T,V> CompletableFuture<V> orApplyStage(
1493 Executor e, CompletionStage<U> o,
1494 Function<? super T, ? extends V> f) {
1495 CompletableFuture<U> b;
1496 if (f == null || (b = o.toCompletableFuture()) == null)
1497 throw new NullPointerException();
1498 CompletableFuture<V> d = newIncompleteFuture();
1499 if (e != null || !d.orApply(this, b, f, null)) {
1500 OrApply<T,U,V> c = new OrApply<T,U,V>(e, d, this, b, f);
1501 if (e != null && (result != null || b.result != null)) {
1502 try {
1503 e.execute(c);
1504 } catch (Throwable ex) {
1505 d.completeThrowable(ex);
1506 }
1507 }
1508 else {
1509 orpush(b, c);
1510 c.tryFire(SYNC);
1511 }
1512 }
1513 return d;
1514 }
1515
1516 @SuppressWarnings("serial")
1517 static final class OrAccept<T,U extends T> extends BiCompletion<T,U,Void> {
1518 Consumer<? super T> fn;
1519 OrAccept(Executor executor, CompletableFuture<Void> dep,
1520 CompletableFuture<T> src,
1521 CompletableFuture<U> snd,
1522 Consumer<? super T> fn) {
1523 super(executor, dep, src, snd); this.fn = fn;
1524 }
1525 final CompletableFuture<Void> tryFire(int mode) {
1526 CompletableFuture<Void> d;
1527 CompletableFuture<T> a;
1528 CompletableFuture<U> b;
1529 if ((d = dep) == null ||
1530 !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1531 return null;
1532 dep = null; src = null; snd = null; fn = null;
1533 return d.postFire(a, b, mode);
1534 }
1535 }
1536
1537 final <R,S extends R> boolean orAccept(CompletableFuture<R> a,
1538 CompletableFuture<S> b,
1539 Consumer<? super R> f,
1540 OrAccept<R,S> c) {
1541 Object r; Throwable x;
1542 if (a == null || b == null ||
1543 ((r = a.result) == null && (r = b.result) == null) || f == null)
1544 return false;
1545 tryComplete: if (result == null) {
1546 try {
1547 if (c != null && !c.claim())
1548 return false;
1549 if (r instanceof AltResult) {
1550 if ((x = ((AltResult)r).ex) != null) {
1551 completeThrowable(x, r);
1552 break tryComplete;
1553 }
1554 r = null;
1555 }
1556 @SuppressWarnings("unchecked") R rr = (R) r;
1557 f.accept(rr);
1558 completeNull();
1559 } catch (Throwable ex) {
1560 completeThrowable(ex);
1561 }
1562 }
1563 return true;
1564 }
1565
1566 private <U extends T> CompletableFuture<Void> orAcceptStage(
1567 Executor e, CompletionStage<U> o, Consumer<? super T> f) {
1568 CompletableFuture<U> b;
1569 if (f == null || (b = o.toCompletableFuture()) == null)
1570 throw new NullPointerException();
1571 CompletableFuture<Void> d = newIncompleteFuture();
1572 if (e != null || !d.orAccept(this, b, f, null)) {
1573 OrAccept<T,U> c = new OrAccept<T,U>(e, d, this, b, f);
1574 if (e != null && (result != null || b.result != null)) {
1575 try {
1576 e.execute(c);
1577 } catch (Throwable ex) {
1578 d.completeThrowable(ex);
1579 }
1580 }
1581 else {
1582 orpush(b, c);
1583 c.tryFire(SYNC);
1584 }
1585 }
1586 return d;
1587 }
1588
1589 @SuppressWarnings("serial")
1590 static final class OrRun<T,U> extends BiCompletion<T,U,Void> {
1591 Runnable fn;
1592 OrRun(Executor executor, CompletableFuture<Void> dep,
1593 CompletableFuture<T> src,
1594 CompletableFuture<U> snd,
1595 Runnable fn) {
1596 super(executor, dep, src, snd); this.fn = fn;
1597 }
1598 final CompletableFuture<Void> tryFire(int mode) {
1599 CompletableFuture<Void> d;
1600 CompletableFuture<T> a;
1601 CompletableFuture<U> b;
1602 if ((d = dep) == null ||
1603 !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this))
1604 return null;
1605 dep = null; src = null; snd = null; fn = null;
1606 return d.postFire(a, b, mode);
1607 }
1608 }
1609
1610 final boolean orRun(CompletableFuture<?> a, CompletableFuture<?> b,
1611 Runnable f, OrRun<?,?> c) {
1612 Object r; Throwable x;
1613 if (a == null || b == null ||
1614 ((r = a.result) == null && (r = b.result) == null) || f == null)
1615 return false;
1616 if (result == null) {
1617 try {
1618 if (c != null && !c.claim())
1619 return false;
1620 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1621 completeThrowable(x, r);
1622 else {
1623 f.run();
1624 completeNull();
1625 }
1626 } catch (Throwable ex) {
1627 completeThrowable(ex);
1628 }
1629 }
1630 return true;
1631 }
1632
1633 private CompletableFuture<Void> orRunStage(Executor e, CompletionStage<?> o,
1634 Runnable f) {
1635 CompletableFuture<?> b;
1636 if (f == null || (b = o.toCompletableFuture()) == null)
1637 throw new NullPointerException();
1638 CompletableFuture<Void> d = newIncompleteFuture();
1639 if (e != null || !d.orRun(this, b, f, null)) {
1640 OrRun<T,?> c = new OrRun<>(e, d, this, b, f);
1641 if (e != null && (result != null || b.result != null)) {
1642 try {
1643 e.execute(c);
1644 } catch (Throwable ex) {
1645 d.completeThrowable(ex);
1646 }
1647 }
1648 else {
1649 orpush(b, c);
1650 c.tryFire(SYNC);
1651 }
1652 }
1653 return d;
1654 }
1655
1656 @SuppressWarnings("serial")
1657 static final class OrRelay<T,U> extends BiCompletion<T,U,Object> { // for Or
1658 OrRelay(CompletableFuture<Object> dep, CompletableFuture<T> src,
1659 CompletableFuture<U> snd) {
1660 super(null, dep, src, snd);
1661 }
1662 final CompletableFuture<Object> tryFire(int mode) {
1663 CompletableFuture<Object> d;
1664 CompletableFuture<T> a;
1665 CompletableFuture<U> b;
1666 if ((d = dep) == null || !d.orRelay(a = src, b = snd))
1667 return null;
1668 src = null; snd = null; dep = null;
1669 return d.postFire(a, b, mode);
1670 }
1671 }
1672
1673 final boolean orRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1674 Object r;
1675 if (a == null || b == null ||
1676 ((r = a.result) == null && (r = b.result) == null))
1677 return false;
1678 if (result == null)
1679 completeRelay(r);
1680 return true;
1681 }
1682
1683 /** Recursively constructs a tree of completions. */
1684 static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs,
1685 int lo, int hi) {
1686 CompletableFuture<Object> d = new CompletableFuture<Object>();
1687 if (lo <= hi) {
1688 CompletableFuture<?> a, b;
1689 int mid = (lo + hi) >>> 1;
1690 if ((a = (lo == mid ? cfs[lo] :
1691 orTree(cfs, lo, mid))) == null ||
1692 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1693 orTree(cfs, mid+1, hi))) == null)
1694 throw new NullPointerException();
1695 if (!d.orRelay(a, b)) {
1696 OrRelay<?,?> c = new OrRelay<>(d, a, b);
1697 a.orpush(b, c);
1698 c.tryFire(SYNC);
1699 }
1700 }
1701 return d;
1702 }
1703
1704 /* ------------- Zero-input Async forms -------------- */
1705
1706 @SuppressWarnings("serial")
1707 static final class AsyncSupply<T> extends ForkJoinTask<Void>
1708 implements Runnable, AsynchronousCompletionTask {
1709 CompletableFuture<T> dep; Supplier<? extends T> fn;
1710 AsyncSupply(CompletableFuture<T> dep, Supplier<? extends T> fn) {
1711 this.dep = dep; this.fn = fn;
1712 }
1713
1714 public final Void getRawResult() { return null; }
1715 public final void setRawResult(Void v) {}
1716 public final boolean exec() { run(); return true; }
1717
1718 public void run() {
1719 CompletableFuture<T> d; Supplier<? extends T> f;
1720 if ((d = dep) != null && (f = fn) != null) {
1721 dep = null; fn = null;
1722 if (d.result == null) {
1723 try {
1724 d.completeValue(f.get());
1725 } catch (Throwable ex) {
1726 d.completeThrowable(ex);
1727 }
1728 }
1729 d.postComplete();
1730 }
1731 }
1732 }
1733
1734 static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
1735 Supplier<U> f) {
1736 if (f == null) throw new NullPointerException();
1737 CompletableFuture<U> d = new CompletableFuture<U>();
1738 e.execute(new AsyncSupply<U>(d, f));
1739 return d;
1740 }
1741
1742 @SuppressWarnings("serial")
1743 static final class AsyncRun extends ForkJoinTask<Void>
1744 implements Runnable, AsynchronousCompletionTask {
1745 CompletableFuture<Void> dep; Runnable fn;
1746 AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
1747 this.dep = dep; this.fn = fn;
1748 }
1749
1750 public final Void getRawResult() { return null; }
1751 public final void setRawResult(Void v) {}
1752 public final boolean exec() { run(); return true; }
1753
1754 public void run() {
1755 CompletableFuture<Void> d; Runnable f;
1756 if ((d = dep) != null && (f = fn) != null) {
1757 dep = null; fn = null;
1758 if (d.result == null) {
1759 try {
1760 f.run();
1761 d.completeNull();
1762 } catch (Throwable ex) {
1763 d.completeThrowable(ex);
1764 }
1765 }
1766 d.postComplete();
1767 }
1768 }
1769 }
1770
1771 static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
1772 if (f == null) throw new NullPointerException();
1773 CompletableFuture<Void> d = new CompletableFuture<Void>();
1774 e.execute(new AsyncRun(d, f));
1775 return d;
1776 }
1777
1778 /* ------------- Signallers -------------- */
1779
1780 /**
1781 * Completion for recording and releasing a waiting thread. This
1782 * class implements ManagedBlocker to avoid starvation when
1783 * blocking actions pile up in ForkJoinPools.
1784 */
1785 @SuppressWarnings("serial")
1786 static final class Signaller extends Completion
1787 implements ForkJoinPool.ManagedBlocker {
1788 long nanos; // remaining wait time if timed
1789 final long deadline; // non-zero if timed
1790 final boolean interruptible;
1791 boolean interrupted;
1792 volatile Thread thread;
1793
1794 Signaller(boolean interruptible, long nanos, long deadline) {
1795 this.thread = Thread.currentThread();
1796 this.interruptible = interruptible;
1797 this.nanos = nanos;
1798 this.deadline = deadline;
1799 }
1800 final CompletableFuture<?> tryFire(int ignore) {
1801 Thread w; // no need to atomically claim
1802 if ((w = thread) != null) {
1803 thread = null;
1804 LockSupport.unpark(w);
1805 }
1806 return null;
1807 }
1808 public boolean isReleasable() {
1809 if (Thread.interrupted())
1810 interrupted = true;
1811 return ((interrupted && interruptible) ||
1812 (deadline != 0L &&
1813 (nanos <= 0L ||
1814 (nanos = deadline - System.nanoTime()) <= 0L)) ||
1815 thread == null);
1816 }
1817 public boolean block() {
1818 while (!isReleasable()) {
1819 if (deadline == 0L)
1820 LockSupport.park(this);
1821 else
1822 LockSupport.parkNanos(this, nanos);
1823 }
1824 return true;
1825 }
1826 final boolean isLive() { return thread != null; }
1827 }
1828
1829 /**
1830 * Returns raw result after waiting, or null if interruptible and
1831 * interrupted.
1832 */
1833 private Object waitingGet(boolean interruptible) {
1834 Signaller q = null;
1835 boolean queued = false;
1836 Object r;
1837 while ((r = result) == null) {
1838 if (q == null) {
1839 q = new Signaller(interruptible, 0L, 0L);
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) {
1855 q.thread = null;
1856 if (q.interrupted) {
1857 if (interruptible)
1858 cleanStack();
1859 else
1860 Thread.currentThread().interrupt();
1861 }
1862 }
1863 if (r != null)
1864 postComplete();
1865 return r;
1866 }
1867
1868 /**
1869 * Returns raw result after waiting, or null if interrupted, or
1870 * throws TimeoutException on timeout.
1871 */
1872 private Object timedGet(long nanos) throws TimeoutException {
1873 if (Thread.interrupted())
1874 return null;
1875 if (nanos > 0L) {
1876 long d = System.nanoTime() + nanos;
1877 long deadline = (d == 0L) ? 1L : d; // avoid 0
1878 Signaller q = null;
1879 boolean queued = false;
1880 Object r;
1881 while ((r = result) == null) { // similar to untimed
1882 if (q == null) {
1883 q = new Signaller(true, nanos, deadline);
1884 ForkJoinPool.helpAsyncBlocker(defaultExecutor(), q);
1885 }
1886 else if (!queued)
1887 queued = tryPushStack(q);
1888 else if (q.nanos <= 0L)
1889 break;
1890 else {
1891 try {
1892 ForkJoinPool.managedBlock(q);
1893 } catch (InterruptedException ie) {
1894 q.interrupted = true;
1895 }
1896 if (q.interrupted)
1897 break;
1898 }
1899 }
1900 if (q != null)
1901 q.thread = null;
1902 if (r != null)
1903 postComplete();
1904 else
1905 cleanStack();
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 sun.misc.Unsafe U = sun.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 }