ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.140
Committed: Tue Jan 6 18:03:27 2015 UTC (9 years, 4 months ago) by dl
Branch: MAIN
Changes since 1.139: +9 -2 lines
Log Message:
Fix previous change

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.</li>
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). To simplify
41 * monitoring, debugging, and tracking, all generated asynchronous
42 * tasks are instances of the marker interface {@link
43 * AsynchronousCompletionTask}. </li>
44 *
45 * <li>All CompletionStage methods are implemented independently of
46 * other public methods, so the behavior of one method is not impacted
47 * by overrides of others in subclasses. </li> </ul>
48 *
49 * <p>CompletableFuture also implements {@link Future} with the following
50 * policies: <ul>
51 *
52 * <li>Since (unlike {@link FutureTask}) this class has no direct
53 * control over the computation that causes it to be completed,
54 * cancellation is treated as just another form of exceptional
55 * completion. Method {@link #cancel cancel} has the same effect as
56 * {@code completeExceptionally(new CancellationException())}. Method
57 * {@link #isCompletedExceptionally} can be used to determine if a
58 * CompletableFuture completed in any exceptional fashion.</li>
59 *
60 * <li>In case of exceptional completion with a CompletionException,
61 * methods {@link #get()} and {@link #get(long, TimeUnit)} throw an
62 * {@link ExecutionException} with the same cause as held in the
63 * corresponding CompletionException. To simplify usage in most
64 * contexts, this class also defines methods {@link #join()} and
65 * {@link #getNow} that instead throw the CompletionException directly
66 * in these cases.</li> </ul>
67 *
68 * @author Doug Lea
69 * @since 1.8
70 */
71 public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
72
73 /*
74 * Overview:
75 *
76 * A CompletableFuture may have dependent completion actions,
77 * collected in a linked stack. It atomically completes by CASing
78 * a result field, and then pops off and runs those actions. This
79 * applies across normal vs exceptional outcomes, sync vs async
80 * actions, binary triggers, and various forms of completions.
81 *
82 * Non-nullness of field result (set via CAS) indicates done. An
83 * AltResult is used to box null as a result, as well as to hold
84 * exceptions. Using a single field makes completion simple to
85 * detect and trigger. Encoding and decoding is straightforward
86 * but adds to the sprawl of trapping and associating exceptions
87 * with targets. Minor simplifications rely on (static) NIL (to
88 * box null results) being the only AltResult with a null
89 * exception field, so we don't usually need explicit comparisons.
90 * Even though some of the generics casts are unchecked (see
91 * SuppressWarnings annotations), they are placed to be
92 * appropriate even if checked.
93 *
94 * Dependent actions are represented by Completion objects linked
95 * as Treiber stacks headed by field "stack". There are Completion
96 * classes for each kind of action, grouped into single-input
97 * (UniCompletion), two-input (BiCompletion), projected
98 * (BiCompletions using either (not both) of two inputs), shared
99 * (CoCompletion, used by the second of two sources), zero-input
100 * source actions, and Signallers that unblock waiters. Class
101 * Completion extends ForkJoinTask to enable async execution
102 * (adding no space overhead because we exploit its "tag" methods
103 * to maintain claims). It is also declared as Runnable to allow
104 * usage with arbitrary executors.
105 *
106 * Support for each kind of CompletionStage relies on a separate
107 * class, along with two CompletableFuture methods:
108 *
109 * * A Completion class with name X corresponding to function,
110 * prefaced with "Uni", "Bi", or "Or". Each class contains
111 * fields for source(s), actions, and dependent. They are
112 * boringly similar, differing from others only with respect to
113 * underlying functional forms. We do this so that users don't
114 * encounter layers of adaptors in common usages. We also
115 * include "Relay" classes/methods that don't correspond to user
116 * methods; they copy results from one stage to another.
117 *
118 * * Boolean CompletableFuture method x(...) (for example
119 * uniApply) takes all of the arguments needed to check that an
120 * action is triggerable, and then either runs the action or
121 * arranges its async execution by executing its Completion
122 * argument, if present. The method returns true if known to be
123 * complete.
124 *
125 * * Completion method tryFire(int mode) invokes the associated x
126 * method with its held arguments, and on success cleans up.
127 * The mode argument allows tryFire to be called twice (SYNC,
128 * then ASYNC); the first to screen and trap exceptions while
129 * arranging to execute, and the second when called from a
130 * task. (A few classes are not used async so take slightly
131 * different forms.) The claim() callback suppresses function
132 * invocation if already claimed by another thread.
133 *
134 * * CompletableFuture method xStage(...) is called from a public
135 * stage method of CompletableFuture x. It screens user
136 * arguments and invokes and/or creates the stage object. If
137 * not async and x is already complete, the action is run
138 * immediately. Otherwise a Completion c is created, pushed to
139 * x's stack (unless done), and started or triggered via
140 * c.tryFire. This also covers races possible if x completes
141 * while pushing. Classes with two inputs (for example BiApply)
142 * deal with races across both while pushing actions. The
143 * second completion is a CoCompletion pointing to the first,
144 * shared so that at most one performs the action. The
145 * multiple-arity methods allOf and anyOf do this pairwise to
146 * form trees of completions.
147 *
148 * Note that the generic type parameters of methods vary according
149 * to whether "this" is a source, dependent, or completion.
150 *
151 * Method postComplete is called upon completion unless the target
152 * is guaranteed not to be observable (i.e., not yet returned or
153 * linked). Multiple threads can call postComplete, which
154 * atomically pops each dependent action, and tries to trigger it
155 * via method tryFire, in NESTED mode. Triggering can propagate
156 * recursively, so NESTED mode returns its completed dependent (if
157 * one exists) for further processing by its caller (see method
158 * postFire).
159 *
160 * Blocking methods get() and join() rely on Signaller Completions
161 * that wake up waiting threads. The mechanics are similar to
162 * Treiber stack wait-nodes used in FutureTask, Phaser, and
163 * SynchronousQueue. See their internal documentation for
164 * algorithmic details.
165 *
166 * Without precautions, CompletableFutures would be prone to
167 * garbage accumulation as chains of Completions build up, each
168 * pointing back to its sources. So we null out fields as soon as
169 * possible (see especially method Completion.detach). The
170 * screening checks needed anyway harmlessly ignore null arguments
171 * that may have been obtained during races with threads nulling
172 * out fields. We also try to unlink fired Completions from
173 * stacks that might never be popped (see method postFire).
174 * Completion fields need not be declared as final or volatile
175 * because they are only visible to other threads upon safe
176 * publication.
177 */
178
179 volatile Object result; // Either the result or boxed AltResult
180 volatile Completion stack; // Top of Treiber stack of dependent actions
181
182 final boolean internalComplete(Object r) { // CAS from null to r
183 return U.compareAndSwapObject(this, RESULT, null, r);
184 }
185
186 final boolean casStack(Completion cmp, Completion val) {
187 return U.compareAndSwapObject(this, STACK, cmp, val);
188 }
189
190 /** Returns true if successfully pushed c onto stack. */
191 final boolean tryPushStack(Completion c) {
192 Completion h = stack;
193 lazySetNext(c, h);
194 return U.compareAndSwapObject(this, STACK, h, c);
195 }
196
197 /** Unconditionally pushes c onto stack, retrying if necessary. */
198 final void pushStack(Completion c) {
199 do {} while (!tryPushStack(c));
200 }
201
202 /* ------------- Encoding and decoding outcomes -------------- */
203
204 static final class AltResult { // See above
205 final Throwable ex; // null only for NIL
206 AltResult(Throwable x) { this.ex = x; }
207 }
208
209 /** The encoding of the null value. */
210 static final AltResult NIL = new AltResult(null);
211
212 /** Completes with the null value, unless already completed. */
213 final boolean completeNull() {
214 return U.compareAndSwapObject(this, RESULT, null,
215 NIL);
216 }
217
218 /** Returns the encoding of the given non-exceptional value. */
219 final Object encodeValue(T t) {
220 return (t == null) ? NIL : t;
221 }
222
223 /** Completes with a non-exceptional result, unless already completed. */
224 final boolean completeValue(T t) {
225 return U.compareAndSwapObject(this, RESULT, null,
226 (t == null) ? NIL : t);
227 }
228
229 /**
230 * Returns the encoding of the given (non-null) exception as a
231 * wrapped CompletionException unless it is one already.
232 */
233 static AltResult encodeThrowable(Throwable x) {
234 return new AltResult((x instanceof CompletionException) ? x :
235 new CompletionException(x));
236 }
237
238 /** Completes with an exceptional result, unless already completed. */
239 final boolean completeThrowable(Throwable x) {
240 return U.compareAndSwapObject(this, RESULT, null,
241 encodeThrowable(x));
242 }
243
244 /**
245 * Returns the encoding of the given (non-null) exception as a
246 * wrapped CompletionException unless it is one already. May
247 * return the given Object r (which must have been the result of a
248 * source future) if it is equivalent, i.e. if this is a simple
249 * relay of an existing CompletionException.
250 */
251 static Object encodeThrowable(Throwable x, Object r) {
252 if (!(x instanceof CompletionException))
253 x = new CompletionException(x);
254 else if (r instanceof AltResult && x == ((AltResult)r).ex)
255 return r;
256 return new AltResult(x);
257 }
258
259 /**
260 * Completes with the given (non-null) exceptional result as a
261 * wrapped CompletionException unless it is one already, unless
262 * already completed. May complete with the given Object r
263 * (which must have been the result of a source future) if it is
264 * equivalent, i.e. if this is a simple propagation of an
265 * existing CompletionException.
266 */
267 final boolean completeThrowable(Throwable x, Object r) {
268 return U.compareAndSwapObject(this, RESULT, null,
269 encodeThrowable(x, r));
270 }
271
272 /**
273 * Returns the encoding of the given arguments: if the exception
274 * is non-null, encodes as AltResult. Otherwise uses the given
275 * value, boxed as NIL if null.
276 */
277 Object encodeOutcome(T t, Throwable x) {
278 return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x);
279 }
280
281 /**
282 * Returns the encoding of a copied outcome; if exceptional,
283 * rewraps as a CompletionException, else returns argument.
284 */
285 static Object encodeRelay(Object r) {
286 Throwable x;
287 return (((r instanceof AltResult) &&
288 (x = ((AltResult)r).ex) != null &&
289 !(x instanceof CompletionException)) ?
290 new AltResult(new CompletionException(x)) : r);
291 }
292
293 /**
294 * Completes with r or a copy of r, unless already completed.
295 * If exceptional, r is first coerced to a CompletionException.
296 */
297 final boolean completeRelay(Object r) {
298 return U.compareAndSwapObject(this, RESULT, null,
299 encodeRelay(r));
300 }
301
302 /**
303 * Reports result using Future.get conventions.
304 */
305 private static <T> T reportGet(Object r)
306 throws InterruptedException, ExecutionException {
307 if (r == null) // by convention below, null means interrupted
308 throw new InterruptedException();
309 if (r instanceof AltResult) {
310 Throwable x, cause;
311 if ((x = ((AltResult)r).ex) == null)
312 return null;
313 if (x instanceof CancellationException)
314 throw (CancellationException)x;
315 if ((x instanceof CompletionException) &&
316 (cause = x.getCause()) != null)
317 x = cause;
318 throw new ExecutionException(x);
319 }
320 @SuppressWarnings("unchecked") T t = (T) r;
321 return t;
322 }
323
324 /**
325 * Decodes outcome to return result or throw unchecked exception.
326 */
327 private static <T> T reportJoin(Object r) {
328 if (r instanceof AltResult) {
329 Throwable x;
330 if ((x = ((AltResult)r).ex) == null)
331 return null;
332 if (x instanceof CancellationException)
333 throw (CancellationException)x;
334 if (x instanceof CompletionException)
335 throw (CompletionException)x;
336 throw new CompletionException(x);
337 }
338 @SuppressWarnings("unchecked") T t = (T) r;
339 return t;
340 }
341
342 /* ------------- Async task preliminaries -------------- */
343
344 /**
345 * A marker interface identifying asynchronous tasks produced by
346 * {@code async} methods. This may be useful for monitoring,
347 * debugging, and tracking asynchronous activities.
348 *
349 * @since 1.8
350 */
351 public static interface AsynchronousCompletionTask {
352 }
353
354 private static final boolean useCommonPool =
355 (ForkJoinPool.getCommonPoolParallelism() > 1);
356
357 /**
358 * Default executor -- ForkJoinPool.commonPool() unless it cannot
359 * support parallelism.
360 */
361 private static final Executor asyncPool = useCommonPool ?
362 ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
363
364 /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
365 static final class ThreadPerTaskExecutor implements Executor {
366 public void execute(Runnable r) { new Thread(r).start(); }
367 }
368
369 /**
370 * Null-checks user executor argument, and translates uses of
371 * commonPool to asyncPool in case parallelism disabled.
372 */
373 static Executor screenExecutor(Executor e) {
374 if (!useCommonPool && e == ForkJoinPool.commonPool())
375 return asyncPool;
376 if (e == null) throw new NullPointerException();
377 return e;
378 }
379
380 // Modes for Completion.tryFire. Signedness matters.
381 static final int SYNC = 0;
382 static final int ASYNC = 1;
383 static final int NESTED = -1;
384
385 /* ------------- Base Completion classes and operations -------------- */
386
387 @SuppressWarnings("serial")
388 abstract static class Completion extends ForkJoinTask<Void>
389 implements Runnable, AsynchronousCompletionTask {
390 volatile Completion next; // Treiber stack link
391
392 /**
393 * Performs completion action if triggered, returning a
394 * dependent that may need propagation, if one exists.
395 *
396 * @param mode SYNC, ASYNC, or NESTED
397 */
398 abstract CompletableFuture<?> tryFire(int mode);
399
400 /** Returns true if possibly still triggerable. Used by cleanStack. */
401 abstract boolean isLive();
402
403 public final void run() { tryFire(ASYNC); }
404 public final boolean exec() { tryFire(ASYNC); return true; }
405 public final Void getRawResult() { return null; }
406 public final void setRawResult(Void v) {}
407 }
408
409 static void lazySetNext(Completion c, Completion next) {
410 U.putOrderedObject(c, NEXT, next);
411 }
412
413 /**
414 * Pops and tries to trigger all reachable dependents. Call only
415 * when known to be done.
416 */
417 final void postComplete() {
418 /*
419 * On each step, variable f holds current dependents to pop
420 * and run. It is extended along only one path at a time,
421 * pushing others to avoid unbounded recursion.
422 */
423 CompletableFuture<?> f = this; Completion h;
424 while ((h = f.stack) != null ||
425 (f != this && (h = (f = this).stack) != null)) {
426 CompletableFuture<?> d; Completion t;
427 if (f.casStack(h, t = h.next)) {
428 if (t != null) {
429 if (f != this) {
430 pushStack(h);
431 continue;
432 }
433 h.next = null; // detach
434 }
435 f = (d = h.tryFire(NESTED)) == null ? this : d;
436 }
437 }
438 }
439
440 /** Traverses stack and unlinks dead Completions. */
441 final void cleanStack() {
442 for (Completion p = null, q = stack; q != null;) {
443 Completion s = q.next;
444 if (q.isLive()) {
445 p = q;
446 q = s;
447 }
448 else if (p == null) {
449 casStack(q, s);
450 q = stack;
451 }
452 else {
453 p.next = s;
454 if (p.isLive())
455 q = s;
456 else {
457 p = null; // restart
458 q = stack;
459 }
460 }
461 }
462 }
463
464 /* ------------- One-input Completions -------------- */
465
466 /** A Completion with a source, dependent, and executor. */
467 @SuppressWarnings("serial")
468 abstract static class UniCompletion<T,V> extends Completion {
469 Executor executor; // executor to use (null if none)
470 CompletableFuture<V> dep; // the dependent to complete
471 CompletableFuture<T> src; // source for action
472
473 UniCompletion(Executor executor, CompletableFuture<V> dep,
474 CompletableFuture<T> src) {
475 this.executor = executor; this.dep = dep; this.src = src;
476 }
477
478 /**
479 * Returns true if action can be run. Call only when known to
480 * be triggerable. Uses FJ tag bit to ensure that only one
481 * thread claims ownership. If async, starts as task -- a
482 * later call to tryFire will run action.
483 */
484 final boolean claim() {
485 Executor e = executor;
486 if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {
487 if (e == null)
488 return true;
489 executor = null; // disable
490 e.execute(this);
491 }
492 return false;
493 }
494
495 final boolean isLive() { return dep != null; }
496 }
497
498 /** Pushes the given completion (if it exists) unless done. */
499 final void push(UniCompletion<?,?> c) {
500 if (c != null) {
501 while (result == null && !tryPushStack(c))
502 lazySetNext(c, null); // clear on failure
503 }
504 }
505
506 /**
507 * Post-processing by dependent after successful UniCompletion
508 * tryFire. Tries to clean stack of source a, and then either runs
509 * postComplete or returns this to caller, depending on mode.
510 */
511 final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
512 if (a != null && a.stack != null) {
513 if (mode < 0 || a.result == null)
514 a.cleanStack();
515 else
516 a.postComplete();
517 }
518 if (result != null && stack != null) {
519 if (mode < 0)
520 return this;
521 else
522 postComplete();
523 }
524 return null;
525 }
526
527 @SuppressWarnings("serial")
528 static final class UniApply<T,V> extends UniCompletion<T,V> {
529 Function<? super T,? extends V> fn;
530 UniApply(Executor executor, CompletableFuture<V> dep,
531 CompletableFuture<T> src,
532 Function<? super T,? extends V> fn) {
533 super(executor, dep, src); this.fn = fn;
534 }
535 final CompletableFuture<V> tryFire(int mode) {
536 CompletableFuture<V> d; CompletableFuture<T> a;
537 if ((d = dep) == null ||
538 !d.uniApply(a = src, fn, mode > 0 ? null : this))
539 return null;
540 dep = null; src = null; fn = null;
541 return d.postFire(a, mode);
542 }
543 }
544
545 final <S> boolean uniApply(CompletableFuture<S> a,
546 Function<? super S,? extends T> f,
547 UniApply<S,T> c) {
548 Object r; Throwable x;
549 if (a == null || (r = a.result) == null || f == null)
550 return false;
551 tryComplete: if (result == null) {
552 if (r instanceof AltResult) {
553 if ((x = ((AltResult)r).ex) != null) {
554 completeThrowable(x, r);
555 break tryComplete;
556 }
557 r = null;
558 }
559 try {
560 if (c != null && !c.claim())
561 return false;
562 @SuppressWarnings("unchecked") S s = (S) r;
563 completeValue(f.apply(s));
564 } catch (Throwable ex) {
565 completeThrowable(ex);
566 }
567 }
568 return true;
569 }
570
571 private <V> CompletableFuture<V> uniApplyStage(
572 Executor e, Function<? super T,? extends V> f) {
573 if (f == null) throw new NullPointerException();
574 CompletableFuture<V> d = new CompletableFuture<V>();
575 if (e != null || !d.uniApply(this, f, null)) {
576 UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
577 push(c);
578 c.tryFire(SYNC);
579 }
580 return d;
581 }
582
583 @SuppressWarnings("serial")
584 static final class UniAccept<T> extends UniCompletion<T,Void> {
585 Consumer<? super T> fn;
586 UniAccept(Executor executor, CompletableFuture<Void> dep,
587 CompletableFuture<T> src, Consumer<? super T> fn) {
588 super(executor, dep, src); this.fn = fn;
589 }
590 final CompletableFuture<Void> tryFire(int mode) {
591 CompletableFuture<Void> d; CompletableFuture<T> a;
592 if ((d = dep) == null ||
593 !d.uniAccept(a = src, fn, mode > 0 ? null : this))
594 return null;
595 dep = null; src = null; fn = null;
596 return d.postFire(a, mode);
597 }
598 }
599
600 final <S> boolean uniAccept(CompletableFuture<S> a,
601 Consumer<? super S> f, UniAccept<S> c) {
602 Object r; Throwable x;
603 if (a == null || (r = a.result) == null || f == null)
604 return false;
605 tryComplete: if (result == null) {
606 if (r instanceof AltResult) {
607 if ((x = ((AltResult)r).ex) != null) {
608 completeThrowable(x, r);
609 break tryComplete;
610 }
611 r = null;
612 }
613 try {
614 if (c != null && !c.claim())
615 return false;
616 @SuppressWarnings("unchecked") S s = (S) r;
617 f.accept(s);
618 completeNull();
619 } catch (Throwable ex) {
620 completeThrowable(ex);
621 }
622 }
623 return true;
624 }
625
626 private CompletableFuture<Void> uniAcceptStage(Executor e,
627 Consumer<? super T> f) {
628 if (f == null) throw new NullPointerException();
629 CompletableFuture<Void> d = new CompletableFuture<Void>();
630 if (e != null || !d.uniAccept(this, f, null)) {
631 UniAccept<T> c = new UniAccept<T>(e, d, this, f);
632 push(c);
633 c.tryFire(SYNC);
634 }
635 return d;
636 }
637
638 @SuppressWarnings("serial")
639 static final class UniRun<T> extends UniCompletion<T,Void> {
640 Runnable fn;
641 UniRun(Executor executor, CompletableFuture<Void> dep,
642 CompletableFuture<T> src, Runnable fn) {
643 super(executor, dep, src); this.fn = fn;
644 }
645 final CompletableFuture<Void> tryFire(int mode) {
646 CompletableFuture<Void> d; CompletableFuture<T> a;
647 if ((d = dep) == null ||
648 !d.uniRun(a = src, fn, mode > 0 ? null : this))
649 return null;
650 dep = null; src = null; fn = null;
651 return d.postFire(a, mode);
652 }
653 }
654
655 final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {
656 Object r; Throwable x;
657 if (a == null || (r = a.result) == null || f == null)
658 return false;
659 if (result == null) {
660 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
661 completeThrowable(x, r);
662 else
663 try {
664 if (c != null && !c.claim())
665 return false;
666 f.run();
667 completeNull();
668 } catch (Throwable ex) {
669 completeThrowable(ex);
670 }
671 }
672 return true;
673 }
674
675 private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
676 if (f == null) throw new NullPointerException();
677 CompletableFuture<Void> d = new CompletableFuture<Void>();
678 if (e != null || !d.uniRun(this, f, null)) {
679 UniRun<T> c = new UniRun<T>(e, d, this, f);
680 push(c);
681 c.tryFire(SYNC);
682 }
683 return d;
684 }
685
686 @SuppressWarnings("serial")
687 static final class UniWhenComplete<T> extends UniCompletion<T,T> {
688 BiConsumer<? super T, ? super Throwable> fn;
689 UniWhenComplete(Executor executor, CompletableFuture<T> dep,
690 CompletableFuture<T> src,
691 BiConsumer<? super T, ? super Throwable> fn) {
692 super(executor, dep, src); this.fn = fn;
693 }
694 final CompletableFuture<T> tryFire(int mode) {
695 CompletableFuture<T> d; CompletableFuture<T> a;
696 if ((d = dep) == null ||
697 !d.uniWhenComplete(a = src, fn, mode > 0 ? null : this))
698 return null;
699 dep = null; src = null; fn = null;
700 return d.postFire(a, mode);
701 }
702 }
703
704 final boolean uniWhenComplete(CompletableFuture<T> a,
705 BiConsumer<? super T,? super Throwable> f,
706 UniWhenComplete<T> c) {
707 Object r; T t; Throwable x = null;
708 if (a == null || (r = a.result) == null || f == null)
709 return false;
710 if (result == null) {
711 try {
712 if (c != null && !c.claim())
713 return false;
714 if (r instanceof AltResult) {
715 x = ((AltResult)r).ex;
716 t = null;
717 } else {
718 @SuppressWarnings("unchecked") T tr = (T) r;
719 t = tr;
720 }
721 f.accept(t, x);
722 if (x == null) {
723 internalComplete(r);
724 return true;
725 }
726 } catch (Throwable ex) {
727 if (x == null)
728 x = ex;
729 }
730 completeThrowable(x, r);
731 }
732 return true;
733 }
734
735 private CompletableFuture<T> uniWhenCompleteStage(
736 Executor e, BiConsumer<? super T, ? super Throwable> f) {
737 if (f == null) throw new NullPointerException();
738 CompletableFuture<T> d = new CompletableFuture<T>();
739 if (e != null || !d.uniWhenComplete(this, f, null)) {
740 UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);
741 push(c);
742 c.tryFire(SYNC);
743 }
744 return d;
745 }
746
747 @SuppressWarnings("serial")
748 static final class UniHandle<T,V> extends UniCompletion<T,V> {
749 BiFunction<? super T, Throwable, ? extends V> fn;
750 UniHandle(Executor executor, CompletableFuture<V> dep,
751 CompletableFuture<T> src,
752 BiFunction<? super T, Throwable, ? extends V> fn) {
753 super(executor, dep, src); this.fn = fn;
754 }
755 final CompletableFuture<V> tryFire(int mode) {
756 CompletableFuture<V> d; CompletableFuture<T> a;
757 if ((d = dep) == null ||
758 !d.uniHandle(a = src, fn, mode > 0 ? null : this))
759 return null;
760 dep = null; src = null; fn = null;
761 return d.postFire(a, mode);
762 }
763 }
764
765 final <S> boolean uniHandle(CompletableFuture<S> a,
766 BiFunction<? super S, Throwable, ? extends T> f,
767 UniHandle<S,T> c) {
768 Object r; S s; Throwable x;
769 if (a == null || (r = a.result) == null || f == null)
770 return false;
771 if (result == null) {
772 try {
773 if (c != null && !c.claim())
774 return false;
775 if (r instanceof AltResult) {
776 x = ((AltResult)r).ex;
777 s = null;
778 } else {
779 x = null;
780 @SuppressWarnings("unchecked") S ss = (S) r;
781 s = ss;
782 }
783 completeValue(f.apply(s, x));
784 } catch (Throwable ex) {
785 completeThrowable(ex);
786 }
787 }
788 return true;
789 }
790
791 private <V> CompletableFuture<V> uniHandleStage(
792 Executor e, BiFunction<? super T, Throwable, ? extends V> f) {
793 if (f == null) throw new NullPointerException();
794 CompletableFuture<V> d = new CompletableFuture<V>();
795 if (e != null || !d.uniHandle(this, f, null)) {
796 UniHandle<T,V> c = new UniHandle<T,V>(e, d, this, f);
797 push(c);
798 c.tryFire(SYNC);
799 }
800 return d;
801 }
802
803 @SuppressWarnings("serial")
804 static final class UniExceptionally<T> extends UniCompletion<T,T> {
805 Function<? super Throwable, ? extends T> fn;
806 UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src,
807 Function<? super Throwable, ? extends T> fn) {
808 super(null, dep, src); this.fn = fn;
809 }
810 final CompletableFuture<T> tryFire(int mode) { // never ASYNC
811 // assert mode != ASYNC;
812 CompletableFuture<T> d; CompletableFuture<T> a;
813 if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this))
814 return null;
815 dep = null; src = null; fn = null;
816 return d.postFire(a, mode);
817 }
818 }
819
820 final boolean uniExceptionally(CompletableFuture<T> a,
821 Function<? super Throwable, ? extends T> f,
822 UniExceptionally<T> c) {
823 Object r; Throwable x;
824 if (a == null || (r = a.result) == null || f == null)
825 return false;
826 if (result == null) {
827 try {
828 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) {
829 if (c != null && !c.claim())
830 return false;
831 completeValue(f.apply(x));
832 } else
833 internalComplete(r);
834 } catch (Throwable ex) {
835 completeThrowable(ex);
836 }
837 }
838 return true;
839 }
840
841 private CompletableFuture<T> uniExceptionallyStage(
842 Function<Throwable, ? extends T> f) {
843 if (f == null) throw new NullPointerException();
844 CompletableFuture<T> d = new CompletableFuture<T>();
845 if (!d.uniExceptionally(this, f, null)) {
846 UniExceptionally<T> c = new UniExceptionally<T>(d, this, f);
847 push(c);
848 c.tryFire(SYNC);
849 }
850 return d;
851 }
852
853 @SuppressWarnings("serial")
854 static final class UniRelay<T> extends UniCompletion<T,T> { // for Compose
855 UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src) {
856 super(null, dep, src);
857 }
858 final CompletableFuture<T> tryFire(int mode) {
859 CompletableFuture<T> d; CompletableFuture<T> a;
860 if ((d = dep) == null || !d.uniRelay(a = src))
861 return null;
862 src = null; dep = null;
863 return d.postFire(a, mode);
864 }
865 }
866
867 final boolean uniRelay(CompletableFuture<T> a) {
868 Object r;
869 if (a == null || (r = a.result) == null)
870 return false;
871 if (result == null) // no need to claim
872 completeRelay(r);
873 return true;
874 }
875
876 @SuppressWarnings("serial")
877 static final class UniCompose<T,V> extends UniCompletion<T,V> {
878 Function<? super T, ? extends CompletionStage<V>> fn;
879 UniCompose(Executor executor, CompletableFuture<V> dep,
880 CompletableFuture<T> src,
881 Function<? super T, ? extends CompletionStage<V>> fn) {
882 super(executor, dep, src); this.fn = fn;
883 }
884 final CompletableFuture<V> tryFire(int mode) {
885 CompletableFuture<V> d; CompletableFuture<T> a;
886 if ((d = dep) == null ||
887 !d.uniCompose(a = src, fn, mode > 0 ? null : this))
888 return null;
889 dep = null; src = null; fn = null;
890 return d.postFire(a, mode);
891 }
892 }
893
894 final <S> boolean uniCompose(
895 CompletableFuture<S> a,
896 Function<? super S, ? extends CompletionStage<T>> f,
897 UniCompose<S,T> c) {
898 Object r; Throwable x;
899 if (a == null || (r = a.result) == null || f == null)
900 return false;
901 tryComplete: if (result == null) {
902 if (r instanceof AltResult) {
903 if ((x = ((AltResult)r).ex) != null) {
904 completeThrowable(x, r);
905 break tryComplete;
906 }
907 r = null;
908 }
909 try {
910 if (c != null && !c.claim())
911 return false;
912 @SuppressWarnings("unchecked") S s = (S) r;
913 CompletableFuture<T> g = f.apply(s).toCompletableFuture();
914 if (g.result == null || !uniRelay(g)) {
915 UniRelay<T> copy = new UniRelay<T>(this, g);
916 g.push(copy);
917 copy.tryFire(SYNC);
918 if (result == null)
919 return false;
920 }
921 } catch (Throwable ex) {
922 completeThrowable(ex);
923 }
924 }
925 return true;
926 }
927
928 private <V> CompletableFuture<V> uniComposeStage(
929 Executor e, Function<? super T, ? extends CompletionStage<V>> f) {
930 if (f == null) throw new NullPointerException();
931 Object r; Throwable x;
932 if (e == null && (r = result) != null) {
933 // try to return function result directly
934 if (r instanceof AltResult) {
935 if ((x = ((AltResult)r).ex) != null) {
936 return new CompletableFuture<V>(encodeThrowable(x, r));
937 }
938 r = null;
939 }
940 try {
941 @SuppressWarnings("unchecked") T t = (T) r;
942 CompletableFuture<V> g = f.apply(t).toCompletableFuture();
943 Object s = g.result;
944 if (s != null)
945 return new CompletableFuture<V>(encodeRelay(s));
946 CompletableFuture<V> d = new CompletableFuture<V>();
947 UniRelay<V> copy = new UniRelay<V>(d, g);
948 g.push(copy);
949 copy.tryFire(SYNC);
950 return d;
951 } catch (Throwable ex) {
952 return new CompletableFuture<V>(encodeThrowable(ex));
953 }
954 }
955 CompletableFuture<V> d = new CompletableFuture<V>();
956 UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
957 push(c);
958 c.tryFire(SYNC);
959 return d;
960 }
961
962 /* ------------- Two-input Completions -------------- */
963
964 /** A Completion for an action with two sources */
965 @SuppressWarnings("serial")
966 abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> {
967 CompletableFuture<U> snd; // second source for action
968 BiCompletion(Executor executor, CompletableFuture<V> dep,
969 CompletableFuture<T> src, CompletableFuture<U> snd) {
970 super(executor, dep, src); this.snd = snd;
971 }
972 }
973
974 /** A Completion delegating to a BiCompletion */
975 @SuppressWarnings("serial")
976 static final class CoCompletion extends Completion {
977 BiCompletion<?,?,?> base;
978 CoCompletion(BiCompletion<?,?,?> base) { this.base = base; }
979 final CompletableFuture<?> tryFire(int mode) {
980 BiCompletion<?,?,?> c; CompletableFuture<?> d;
981 if ((c = base) == null || (d = c.tryFire(mode)) == null)
982 return null;
983 base = null; // detach
984 return d;
985 }
986 final boolean isLive() {
987 BiCompletion<?,?,?> c;
988 return (c = base) != null && c.dep != null;
989 }
990 }
991
992 /** Pushes completion to this and b unless both done. */
993 final void bipush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
994 if (c != null) {
995 Object r;
996 while ((r = result) == null && !tryPushStack(c))
997 lazySetNext(c, null); // clear on failure
998 if (b != null && b != this && b.result == null) {
999 Completion q = (r != null) ? c : new CoCompletion(c);
1000 while (b.result == null && !b.tryPushStack(q))
1001 lazySetNext(q, null); // clear on failure
1002 }
1003 }
1004 }
1005
1006 /** Post-processing after successful BiCompletion tryFire. */
1007 final CompletableFuture<T> postFire(CompletableFuture<?> a,
1008 CompletableFuture<?> b, int mode) {
1009 if (b != null && b.stack != null) { // clean second source
1010 if (mode < 0 || b.result == null)
1011 b.cleanStack();
1012 else
1013 b.postComplete();
1014 }
1015 return postFire(a, mode);
1016 }
1017
1018 @SuppressWarnings("serial")
1019 static final class BiApply<T,U,V> extends BiCompletion<T,U,V> {
1020 BiFunction<? super T,? super U,? extends V> fn;
1021 BiApply(Executor executor, CompletableFuture<V> dep,
1022 CompletableFuture<T> src, CompletableFuture<U> snd,
1023 BiFunction<? super T,? super U,? extends V> fn) {
1024 super(executor, dep, src, snd); this.fn = fn;
1025 }
1026 final CompletableFuture<V> tryFire(int mode) {
1027 CompletableFuture<V> d;
1028 CompletableFuture<T> a;
1029 CompletableFuture<U> b;
1030 if ((d = dep) == null ||
1031 !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this))
1032 return null;
1033 dep = null; src = null; snd = null; fn = null;
1034 return d.postFire(a, b, mode);
1035 }
1036 }
1037
1038 final <R,S> boolean biApply(CompletableFuture<R> a,
1039 CompletableFuture<S> b,
1040 BiFunction<? super R,? super S,? extends T> f,
1041 BiApply<R,S,T> c) {
1042 Object r, s; Throwable x;
1043 if (a == null || (r = a.result) == null ||
1044 b == null || (s = b.result) == null || f == null)
1045 return false;
1046 tryComplete: if (result == null) {
1047 if (r instanceof AltResult) {
1048 if ((x = ((AltResult)r).ex) != null) {
1049 completeThrowable(x, r);
1050 break tryComplete;
1051 }
1052 r = null;
1053 }
1054 if (s instanceof AltResult) {
1055 if ((x = ((AltResult)s).ex) != null) {
1056 completeThrowable(x, s);
1057 break tryComplete;
1058 }
1059 s = null;
1060 }
1061 try {
1062 if (c != null && !c.claim())
1063 return false;
1064 @SuppressWarnings("unchecked") R rr = (R) r;
1065 @SuppressWarnings("unchecked") S ss = (S) s;
1066 completeValue(f.apply(rr, ss));
1067 } catch (Throwable ex) {
1068 completeThrowable(ex);
1069 }
1070 }
1071 return true;
1072 }
1073
1074 private <U,V> CompletableFuture<V> biApplyStage(
1075 Executor e, CompletionStage<U> o,
1076 BiFunction<? super T,? super U,? extends V> f) {
1077 CompletableFuture<U> b;
1078 if (f == null || (b = o.toCompletableFuture()) == null)
1079 throw new NullPointerException();
1080 CompletableFuture<V> d = new CompletableFuture<V>();
1081 if (e != null || !d.biApply(this, b, f, null)) {
1082 BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);
1083 bipush(b, c);
1084 c.tryFire(SYNC);
1085 }
1086 return d;
1087 }
1088
1089 @SuppressWarnings("serial")
1090 static final class BiAccept<T,U> extends BiCompletion<T,U,Void> {
1091 BiConsumer<? super T,? super U> fn;
1092 BiAccept(Executor executor, CompletableFuture<Void> dep,
1093 CompletableFuture<T> src, CompletableFuture<U> snd,
1094 BiConsumer<? super T,? super U> fn) {
1095 super(executor, dep, src, snd); this.fn = fn;
1096 }
1097 final CompletableFuture<Void> tryFire(int mode) {
1098 CompletableFuture<Void> d;
1099 CompletableFuture<T> a;
1100 CompletableFuture<U> b;
1101 if ((d = dep) == null ||
1102 !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1103 return null;
1104 dep = null; src = null; snd = null; fn = null;
1105 return d.postFire(a, b, mode);
1106 }
1107 }
1108
1109 final <R,S> boolean biAccept(CompletableFuture<R> a,
1110 CompletableFuture<S> b,
1111 BiConsumer<? super R,? super S> f,
1112 BiAccept<R,S> c) {
1113 Object r, s; Throwable x;
1114 if (a == null || (r = a.result) == null ||
1115 b == null || (s = b.result) == null || f == null)
1116 return false;
1117 tryComplete: if (result == null) {
1118 if (r instanceof AltResult) {
1119 if ((x = ((AltResult)r).ex) != null) {
1120 completeThrowable(x, r);
1121 break tryComplete;
1122 }
1123 r = null;
1124 }
1125 if (s instanceof AltResult) {
1126 if ((x = ((AltResult)s).ex) != null) {
1127 completeThrowable(x, s);
1128 break tryComplete;
1129 }
1130 s = null;
1131 }
1132 try {
1133 if (c != null && !c.claim())
1134 return false;
1135 @SuppressWarnings("unchecked") R rr = (R) r;
1136 @SuppressWarnings("unchecked") S ss = (S) s;
1137 f.accept(rr, ss);
1138 completeNull();
1139 } catch (Throwable ex) {
1140 completeThrowable(ex);
1141 }
1142 }
1143 return true;
1144 }
1145
1146 private <U> CompletableFuture<Void> biAcceptStage(
1147 Executor e, CompletionStage<U> o,
1148 BiConsumer<? super T,? super U> f) {
1149 CompletableFuture<U> b;
1150 if (f == null || (b = o.toCompletableFuture()) == null)
1151 throw new NullPointerException();
1152 CompletableFuture<Void> d = new CompletableFuture<Void>();
1153 if (e != null || !d.biAccept(this, b, f, null)) {
1154 BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
1155 bipush(b, c);
1156 c.tryFire(SYNC);
1157 }
1158 return d;
1159 }
1160
1161 @SuppressWarnings("serial")
1162 static final class BiRun<T,U> extends BiCompletion<T,U,Void> {
1163 Runnable fn;
1164 BiRun(Executor executor, CompletableFuture<Void> dep,
1165 CompletableFuture<T> src,
1166 CompletableFuture<U> snd,
1167 Runnable fn) {
1168 super(executor, dep, src, snd); this.fn = fn;
1169 }
1170 final CompletableFuture<Void> tryFire(int mode) {
1171 CompletableFuture<Void> d;
1172 CompletableFuture<T> a;
1173 CompletableFuture<U> b;
1174 if ((d = dep) == null ||
1175 !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this))
1176 return null;
1177 dep = null; src = null; snd = null; fn = null;
1178 return d.postFire(a, b, mode);
1179 }
1180 }
1181
1182 final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b,
1183 Runnable f, BiRun<?,?> c) {
1184 Object r, s; Throwable x;
1185 if (a == null || (r = a.result) == null ||
1186 b == null || (s = b.result) == null || f == null)
1187 return false;
1188 if (result == null) {
1189 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1190 completeThrowable(x, r);
1191 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1192 completeThrowable(x, s);
1193 else
1194 try {
1195 if (c != null && !c.claim())
1196 return false;
1197 f.run();
1198 completeNull();
1199 } catch (Throwable ex) {
1200 completeThrowable(ex);
1201 }
1202 }
1203 return true;
1204 }
1205
1206 private CompletableFuture<Void> biRunStage(Executor e, CompletionStage<?> o,
1207 Runnable f) {
1208 CompletableFuture<?> b;
1209 if (f == null || (b = o.toCompletableFuture()) == null)
1210 throw new NullPointerException();
1211 CompletableFuture<Void> d = new CompletableFuture<Void>();
1212 if (e != null || !d.biRun(this, b, f, null)) {
1213 BiRun<T,?> c = new BiRun<>(e, d, this, b, f);
1214 bipush(b, c);
1215 c.tryFire(SYNC);
1216 }
1217 return d;
1218 }
1219
1220 @SuppressWarnings("serial")
1221 static final class BiRelay<T,U> extends BiCompletion<T,U,Void> { // for And
1222 BiRelay(CompletableFuture<Void> dep,
1223 CompletableFuture<T> src,
1224 CompletableFuture<U> snd) {
1225 super(null, dep, src, snd);
1226 }
1227 final CompletableFuture<Void> tryFire(int mode) {
1228 CompletableFuture<Void> d;
1229 CompletableFuture<T> a;
1230 CompletableFuture<U> b;
1231 if ((d = dep) == null || !d.biRelay(a = src, b = snd))
1232 return null;
1233 src = null; snd = null; dep = null;
1234 return d.postFire(a, b, mode);
1235 }
1236 }
1237
1238 boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1239 Object r, s; Throwable x;
1240 if (a == null || (r = a.result) == null ||
1241 b == null || (s = b.result) == null)
1242 return false;
1243 if (result == null) {
1244 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1245 completeThrowable(x, r);
1246 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1247 completeThrowable(x, s);
1248 else
1249 completeNull();
1250 }
1251 return true;
1252 }
1253
1254 /** Recursively constructs a tree of completions. */
1255 static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs,
1256 int lo, int hi) {
1257 CompletableFuture<Void> d = new CompletableFuture<Void>();
1258 if (lo > hi) // empty
1259 d.result = NIL;
1260 else {
1261 CompletableFuture<?> a, b;
1262 int mid = (lo + hi) >>> 1;
1263 if ((a = (lo == mid ? cfs[lo] :
1264 andTree(cfs, lo, mid))) == null ||
1265 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1266 andTree(cfs, mid+1, hi))) == null)
1267 throw new NullPointerException();
1268 if (!d.biRelay(a, b)) {
1269 BiRelay<?,?> c = new BiRelay<>(d, a, b);
1270 a.bipush(b, c);
1271 c.tryFire(SYNC);
1272 }
1273 }
1274 return d;
1275 }
1276
1277 /* ------------- Projected (Ored) BiCompletions -------------- */
1278
1279 /** Pushes completion to this and b unless either done. */
1280 final void orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
1281 if (c != null) {
1282 while ((b == null || b.result == null) && result == null) {
1283 if (tryPushStack(c)) {
1284 if (b != null && b != this && b.result == null) {
1285 Completion q = new CoCompletion(c);
1286 while (result == null && b.result == null &&
1287 !b.tryPushStack(q))
1288 lazySetNext(q, null); // clear on failure
1289 }
1290 break;
1291 }
1292 lazySetNext(c, null); // clear on failure
1293 }
1294 }
1295 }
1296
1297 @SuppressWarnings("serial")
1298 static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> {
1299 Function<? super T,? extends V> fn;
1300 OrApply(Executor executor, CompletableFuture<V> dep,
1301 CompletableFuture<T> src,
1302 CompletableFuture<U> snd,
1303 Function<? super T,? extends V> fn) {
1304 super(executor, dep, src, snd); this.fn = fn;
1305 }
1306 final CompletableFuture<V> tryFire(int mode) {
1307 CompletableFuture<V> d;
1308 CompletableFuture<T> a;
1309 CompletableFuture<U> b;
1310 if ((d = dep) == null ||
1311 !d.orApply(a = src, b = snd, fn, mode > 0 ? null : this))
1312 return null;
1313 dep = null; src = null; snd = null; fn = null;
1314 return d.postFire(a, b, mode);
1315 }
1316 }
1317
1318 final <R,S extends R> boolean orApply(CompletableFuture<R> a,
1319 CompletableFuture<S> b,
1320 Function<? super R, ? extends T> f,
1321 OrApply<R,S,T> c) {
1322 Object r; Throwable x;
1323 if (a == null || b == null ||
1324 ((r = a.result) == null && (r = b.result) == null) || f == null)
1325 return false;
1326 tryComplete: if (result == null) {
1327 try {
1328 if (c != null && !c.claim())
1329 return false;
1330 if (r instanceof AltResult) {
1331 if ((x = ((AltResult)r).ex) != null) {
1332 completeThrowable(x, r);
1333 break tryComplete;
1334 }
1335 r = null;
1336 }
1337 @SuppressWarnings("unchecked") R rr = (R) r;
1338 completeValue(f.apply(rr));
1339 } catch (Throwable ex) {
1340 completeThrowable(ex);
1341 }
1342 }
1343 return true;
1344 }
1345
1346 private <U extends T,V> CompletableFuture<V> orApplyStage(
1347 Executor e, CompletionStage<U> o,
1348 Function<? super T, ? extends V> f) {
1349 CompletableFuture<U> b;
1350 if (f == null || (b = o.toCompletableFuture()) == null)
1351 throw new NullPointerException();
1352 CompletableFuture<V> d = new CompletableFuture<V>();
1353 if (e != null || !d.orApply(this, b, f, null)) {
1354 OrApply<T,U,V> c = new OrApply<T,U,V>(e, d, this, b, f);
1355 orpush(b, c);
1356 c.tryFire(SYNC);
1357 }
1358 return d;
1359 }
1360
1361 @SuppressWarnings("serial")
1362 static final class OrAccept<T,U extends T> extends BiCompletion<T,U,Void> {
1363 Consumer<? super T> fn;
1364 OrAccept(Executor executor, CompletableFuture<Void> dep,
1365 CompletableFuture<T> src,
1366 CompletableFuture<U> snd,
1367 Consumer<? super T> fn) {
1368 super(executor, dep, src, snd); this.fn = fn;
1369 }
1370 final CompletableFuture<Void> tryFire(int mode) {
1371 CompletableFuture<Void> d;
1372 CompletableFuture<T> a;
1373 CompletableFuture<U> b;
1374 if ((d = dep) == null ||
1375 !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1376 return null;
1377 dep = null; src = null; snd = null; fn = null;
1378 return d.postFire(a, b, mode);
1379 }
1380 }
1381
1382 final <R,S extends R> boolean orAccept(CompletableFuture<R> a,
1383 CompletableFuture<S> b,
1384 Consumer<? super R> f,
1385 OrAccept<R,S> c) {
1386 Object r; Throwable x;
1387 if (a == null || b == null ||
1388 ((r = a.result) == null && (r = b.result) == null) || f == null)
1389 return false;
1390 tryComplete: if (result == null) {
1391 try {
1392 if (c != null && !c.claim())
1393 return false;
1394 if (r instanceof AltResult) {
1395 if ((x = ((AltResult)r).ex) != null) {
1396 completeThrowable(x, r);
1397 break tryComplete;
1398 }
1399 r = null;
1400 }
1401 @SuppressWarnings("unchecked") R rr = (R) r;
1402 f.accept(rr);
1403 completeNull();
1404 } catch (Throwable ex) {
1405 completeThrowable(ex);
1406 }
1407 }
1408 return true;
1409 }
1410
1411 private <U extends T> CompletableFuture<Void> orAcceptStage(
1412 Executor e, CompletionStage<U> o, Consumer<? super T> f) {
1413 CompletableFuture<U> b;
1414 if (f == null || (b = o.toCompletableFuture()) == null)
1415 throw new NullPointerException();
1416 CompletableFuture<Void> d = new CompletableFuture<Void>();
1417 if (e != null || !d.orAccept(this, b, f, null)) {
1418 OrAccept<T,U> c = new OrAccept<T,U>(e, d, this, b, f);
1419 orpush(b, c);
1420 c.tryFire(SYNC);
1421 }
1422 return d;
1423 }
1424
1425 @SuppressWarnings("serial")
1426 static final class OrRun<T,U> extends BiCompletion<T,U,Void> {
1427 Runnable fn;
1428 OrRun(Executor executor, CompletableFuture<Void> dep,
1429 CompletableFuture<T> src,
1430 CompletableFuture<U> snd,
1431 Runnable fn) {
1432 super(executor, dep, src, snd); this.fn = fn;
1433 }
1434 final CompletableFuture<Void> tryFire(int mode) {
1435 CompletableFuture<Void> d;
1436 CompletableFuture<T> a;
1437 CompletableFuture<U> b;
1438 if ((d = dep) == null ||
1439 !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this))
1440 return null;
1441 dep = null; src = null; snd = null; fn = null;
1442 return d.postFire(a, b, mode);
1443 }
1444 }
1445
1446 final boolean orRun(CompletableFuture<?> a, CompletableFuture<?> b,
1447 Runnable f, OrRun<?,?> c) {
1448 Object r; Throwable x;
1449 if (a == null || b == null ||
1450 ((r = a.result) == null && (r = b.result) == null) || f == null)
1451 return false;
1452 if (result == null) {
1453 try {
1454 if (c != null && !c.claim())
1455 return false;
1456 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1457 completeThrowable(x, r);
1458 else {
1459 f.run();
1460 completeNull();
1461 }
1462 } catch (Throwable ex) {
1463 completeThrowable(ex);
1464 }
1465 }
1466 return true;
1467 }
1468
1469 private CompletableFuture<Void> orRunStage(Executor e, CompletionStage<?> o,
1470 Runnable f) {
1471 CompletableFuture<?> b;
1472 if (f == null || (b = o.toCompletableFuture()) == null)
1473 throw new NullPointerException();
1474 CompletableFuture<Void> d = new CompletableFuture<Void>();
1475 if (e != null || !d.orRun(this, b, f, null)) {
1476 OrRun<T,?> c = new OrRun<>(e, d, this, b, f);
1477 orpush(b, c);
1478 c.tryFire(SYNC);
1479 }
1480 return d;
1481 }
1482
1483 @SuppressWarnings("serial")
1484 static final class OrRelay<T,U> extends BiCompletion<T,U,Object> { // for Or
1485 OrRelay(CompletableFuture<Object> dep, CompletableFuture<T> src,
1486 CompletableFuture<U> snd) {
1487 super(null, dep, src, snd);
1488 }
1489 final CompletableFuture<Object> tryFire(int mode) {
1490 CompletableFuture<Object> d;
1491 CompletableFuture<T> a;
1492 CompletableFuture<U> b;
1493 if ((d = dep) == null || !d.orRelay(a = src, b = snd))
1494 return null;
1495 src = null; snd = null; dep = null;
1496 return d.postFire(a, b, mode);
1497 }
1498 }
1499
1500 final boolean orRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1501 Object r;
1502 if (a == null || b == null ||
1503 ((r = a.result) == null && (r = b.result) == null))
1504 return false;
1505 if (result == null)
1506 completeRelay(r);
1507 return true;
1508 }
1509
1510 /** Recursively constructs a tree of completions. */
1511 static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs,
1512 int lo, int hi) {
1513 CompletableFuture<Object> d = new CompletableFuture<Object>();
1514 if (lo <= hi) {
1515 CompletableFuture<?> a, b;
1516 int mid = (lo + hi) >>> 1;
1517 if ((a = (lo == mid ? cfs[lo] :
1518 orTree(cfs, lo, mid))) == null ||
1519 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1520 orTree(cfs, mid+1, hi))) == null)
1521 throw new NullPointerException();
1522 if (!d.orRelay(a, b)) {
1523 OrRelay<?,?> c = new OrRelay<>(d, a, b);
1524 a.orpush(b, c);
1525 c.tryFire(SYNC);
1526 }
1527 }
1528 return d;
1529 }
1530
1531 /* ------------- Zero-input Async forms -------------- */
1532
1533 @SuppressWarnings("serial")
1534 static final class AsyncSupply<T> extends ForkJoinTask<Void>
1535 implements Runnable, AsynchronousCompletionTask {
1536 CompletableFuture<T> dep; Supplier<T> fn;
1537 AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
1538 this.dep = dep; this.fn = fn;
1539 }
1540
1541 public final Void getRawResult() { return null; }
1542 public final void setRawResult(Void v) {}
1543 public final boolean exec() { run(); return true; }
1544
1545 public void run() {
1546 CompletableFuture<T> d; Supplier<T> f;
1547 if ((d = dep) != null && (f = fn) != null) {
1548 dep = null; fn = null;
1549 if (d.result == null) {
1550 try {
1551 d.completeValue(f.get());
1552 } catch (Throwable ex) {
1553 d.completeThrowable(ex);
1554 }
1555 }
1556 d.postComplete();
1557 }
1558 }
1559 }
1560
1561 static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
1562 Supplier<U> f) {
1563 if (f == null) throw new NullPointerException();
1564 CompletableFuture<U> d = new CompletableFuture<U>();
1565 e.execute(new AsyncSupply<U>(d, f));
1566 return d;
1567 }
1568
1569 @SuppressWarnings("serial")
1570 static final class AsyncRun extends ForkJoinTask<Void>
1571 implements Runnable, AsynchronousCompletionTask {
1572 CompletableFuture<Void> dep; Runnable fn;
1573 AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
1574 this.dep = dep; this.fn = fn;
1575 }
1576
1577 public final Void getRawResult() { return null; }
1578 public final void setRawResult(Void v) {}
1579 public final boolean exec() { run(); return true; }
1580
1581 public void run() {
1582 CompletableFuture<Void> d; Runnable f;
1583 if ((d = dep) != null && (f = fn) != null) {
1584 dep = null; fn = null;
1585 if (d.result == null) {
1586 try {
1587 f.run();
1588 d.completeNull();
1589 } catch (Throwable ex) {
1590 d.completeThrowable(ex);
1591 }
1592 }
1593 d.postComplete();
1594 }
1595 }
1596 }
1597
1598 static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
1599 if (f == null) throw new NullPointerException();
1600 CompletableFuture<Void> d = new CompletableFuture<Void>();
1601 e.execute(new AsyncRun(d, f));
1602 return d;
1603 }
1604
1605 /* ------------- Signallers -------------- */
1606
1607 /**
1608 * Completion for recording and releasing a waiting thread. This
1609 * class implements ManagedBlocker to avoid starvation when
1610 * blocking actions pile up in ForkJoinPools.
1611 */
1612 @SuppressWarnings("serial")
1613 static final class Signaller extends Completion
1614 implements ForkJoinPool.ManagedBlocker {
1615 long nanos; // wait time if timed
1616 final long deadline; // non-zero if timed
1617 volatile int interruptControl; // > 0: interruptible, < 0: interrupted
1618 volatile Thread thread;
1619
1620 Signaller(boolean interruptible, long nanos, long deadline) {
1621 this.thread = Thread.currentThread();
1622 this.interruptControl = interruptible ? 1 : 0;
1623 this.nanos = nanos;
1624 this.deadline = deadline;
1625 }
1626 final CompletableFuture<?> tryFire(int ignore) {
1627 Thread w; // no need to atomically claim
1628 if ((w = thread) != null) {
1629 thread = null;
1630 LockSupport.unpark(w);
1631 }
1632 return null;
1633 }
1634 public boolean isReleasable() {
1635 if (thread == null)
1636 return true;
1637 if (Thread.interrupted()) {
1638 int i = interruptControl;
1639 interruptControl = -1;
1640 if (i > 0)
1641 return true;
1642 }
1643 if (deadline != 0L &&
1644 (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {
1645 thread = null;
1646 return true;
1647 }
1648 return false;
1649 }
1650 public boolean block() {
1651 if (isReleasable())
1652 return true;
1653 else if (deadline == 0L)
1654 LockSupport.park(this);
1655 else if (nanos > 0L)
1656 LockSupport.parkNanos(this, nanos);
1657 return isReleasable();
1658 }
1659 final boolean isLive() { return thread != null; }
1660 }
1661
1662 /**
1663 * Returns raw result after waiting, or null if interruptible and
1664 * interrupted.
1665 */
1666 private Object waitingGet(boolean interruptible) {
1667 Signaller q = null;
1668 boolean queued = false;
1669 int spins = -1;
1670 Object r;
1671 while ((r = result) == null) {
1672 if (spins < 0)
1673 spins = (Runtime.getRuntime().availableProcessors() > 1) ?
1674 1 << 8 : 0; // Use brief spin-wait on multiprocessors
1675 else if (spins > 0) {
1676 if (ThreadLocalRandom.nextSecondarySeed() >= 0)
1677 --spins;
1678 }
1679 else if (q == null)
1680 q = new Signaller(interruptible, 0L, 0L);
1681 else if (!queued)
1682 queued = tryPushStack(q);
1683 else if (interruptible && q.interruptControl < 0) {
1684 q.thread = null;
1685 cleanStack();
1686 return null;
1687 }
1688 else if (q.thread != null && result == null) {
1689 try {
1690 ForkJoinPool.managedBlock(q);
1691 } catch (InterruptedException ie) {
1692 q.interruptControl = -1;
1693 }
1694 }
1695 }
1696 if (q != null) {
1697 q.thread = null;
1698 if (q.interruptControl < 0) {
1699 if (interruptible)
1700 r = null; // report interruption
1701 else
1702 Thread.currentThread().interrupt();
1703 }
1704 }
1705 postComplete();
1706 return r;
1707 }
1708
1709 /**
1710 * Returns raw result after waiting, or null if interrupted, or
1711 * throws TimeoutException on timeout.
1712 */
1713 private Object timedGet(long nanos) throws TimeoutException {
1714 if (Thread.interrupted())
1715 return null;
1716 if (nanos <= 0L)
1717 throw new TimeoutException();
1718 long d = System.nanoTime() + nanos;
1719 Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0
1720 boolean queued = false;
1721 Object r;
1722 // We intentionally don't spin here (as waitingGet does) because
1723 // the call to nanoTime() above acts much like a spin.
1724 while ((r = result) == null) {
1725 if (!queued)
1726 queued = tryPushStack(q);
1727 else if (q.interruptControl < 0 || q.nanos <= 0L) {
1728 q.thread = null;
1729 cleanStack();
1730 if (q.interruptControl < 0)
1731 return null;
1732 throw new TimeoutException();
1733 }
1734 else if (q.thread != null && result == null) {
1735 try {
1736 ForkJoinPool.managedBlock(q);
1737 } catch (InterruptedException ie) {
1738 q.interruptControl = -1;
1739 }
1740 }
1741 }
1742 if (q.interruptControl < 0)
1743 r = null;
1744 q.thread = null;
1745 postComplete();
1746 return r;
1747 }
1748
1749 /* ------------- public methods -------------- */
1750
1751 /**
1752 * Creates a new incomplete CompletableFuture.
1753 */
1754 public CompletableFuture() {
1755 }
1756
1757 /**
1758 * Creates a new complete CompletableFuture with given encoded result.
1759 */
1760 private CompletableFuture(Object r) {
1761 this.result = r;
1762 }
1763
1764 /**
1765 * Returns a new CompletableFuture that is asynchronously completed
1766 * by a task running in the {@link ForkJoinPool#commonPool()} with
1767 * the value obtained by calling the given Supplier.
1768 *
1769 * @param supplier a function returning the value to be used
1770 * to complete the returned CompletableFuture
1771 * @param <U> the function's return type
1772 * @return the new CompletableFuture
1773 */
1774 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
1775 return asyncSupplyStage(asyncPool, supplier);
1776 }
1777
1778 /**
1779 * Returns a new CompletableFuture that is asynchronously completed
1780 * by a task running in the given executor with the value obtained
1781 * by calling the given Supplier.
1782 *
1783 * @param supplier a function returning the value to be used
1784 * to complete the returned CompletableFuture
1785 * @param executor the executor to use for asynchronous execution
1786 * @param <U> the function's return type
1787 * @return the new CompletableFuture
1788 */
1789 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
1790 Executor executor) {
1791 return asyncSupplyStage(screenExecutor(executor), supplier);
1792 }
1793
1794 /**
1795 * Returns a new CompletableFuture that is asynchronously completed
1796 * by a task running in the {@link ForkJoinPool#commonPool()} after
1797 * it runs the given action.
1798 *
1799 * @param runnable the action to run before completing the
1800 * returned CompletableFuture
1801 * @return the new CompletableFuture
1802 */
1803 public static CompletableFuture<Void> runAsync(Runnable runnable) {
1804 return asyncRunStage(asyncPool, runnable);
1805 }
1806
1807 /**
1808 * Returns a new CompletableFuture that is asynchronously completed
1809 * by a task running in the given executor after it runs the given
1810 * action.
1811 *
1812 * @param runnable the action to run before completing the
1813 * returned CompletableFuture
1814 * @param executor the executor to use for asynchronous execution
1815 * @return the new CompletableFuture
1816 */
1817 public static CompletableFuture<Void> runAsync(Runnable runnable,
1818 Executor executor) {
1819 return asyncRunStage(screenExecutor(executor), runnable);
1820 }
1821
1822 /**
1823 * Returns a new CompletableFuture that is already completed with
1824 * the given value.
1825 *
1826 * @param value the value
1827 * @param <U> the type of the value
1828 * @return the completed CompletableFuture
1829 */
1830 public static <U> CompletableFuture<U> completedFuture(U value) {
1831 return new CompletableFuture<U>((value == null) ? NIL : value);
1832 }
1833
1834 /**
1835 * Returns {@code true} if completed in any fashion: normally,
1836 * exceptionally, or via cancellation.
1837 *
1838 * @return {@code true} if completed
1839 */
1840 public boolean isDone() {
1841 return result != null;
1842 }
1843
1844 /**
1845 * Waits if necessary for this future to complete, and then
1846 * returns its result.
1847 *
1848 * @return the result value
1849 * @throws CancellationException if this future was cancelled
1850 * @throws ExecutionException if this future completed exceptionally
1851 * @throws InterruptedException if the current thread was interrupted
1852 * while waiting
1853 */
1854 public T get() throws InterruptedException, ExecutionException {
1855 Object r;
1856 return reportGet((r = result) == null ? waitingGet(true) : r);
1857 }
1858
1859 /**
1860 * Waits if necessary for at most the given time for this future
1861 * to complete, and then returns its result, if available.
1862 *
1863 * @param timeout the maximum time to wait
1864 * @param unit the time unit of the timeout argument
1865 * @return the result value
1866 * @throws CancellationException if this future was cancelled
1867 * @throws ExecutionException if this future completed exceptionally
1868 * @throws InterruptedException if the current thread was interrupted
1869 * while waiting
1870 * @throws TimeoutException if the wait timed out
1871 */
1872 public T get(long timeout, TimeUnit unit)
1873 throws InterruptedException, ExecutionException, TimeoutException {
1874 Object r;
1875 long nanos = unit.toNanos(timeout);
1876 return reportGet((r = result) == null ? timedGet(nanos) : r);
1877 }
1878
1879 /**
1880 * Returns the result value when complete, or throws an
1881 * (unchecked) exception if completed exceptionally. To better
1882 * conform with the use of common functional forms, if a
1883 * computation involved in the completion of this
1884 * CompletableFuture threw an exception, this method throws an
1885 * (unchecked) {@link CompletionException} with the underlying
1886 * exception as its cause.
1887 *
1888 * @return the result value
1889 * @throws CancellationException if the computation was cancelled
1890 * @throws CompletionException if this future completed
1891 * exceptionally or a completion computation threw an exception
1892 */
1893 public T join() {
1894 Object r;
1895 return reportJoin((r = result) == null ? waitingGet(false) : r);
1896 }
1897
1898 /**
1899 * Returns the result value (or throws any encountered exception)
1900 * if completed, else returns the given valueIfAbsent.
1901 *
1902 * @param valueIfAbsent the value to return if not completed
1903 * @return the result value, if completed, else the given valueIfAbsent
1904 * @throws CancellationException if the computation was cancelled
1905 * @throws CompletionException if this future completed
1906 * exceptionally or a completion computation threw an exception
1907 */
1908 public T getNow(T valueIfAbsent) {
1909 Object r;
1910 return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
1911 }
1912
1913 /**
1914 * If not already completed, sets the value returned by {@link
1915 * #get()} and related methods to the given value.
1916 *
1917 * @param value the result value
1918 * @return {@code true} if this invocation caused this CompletableFuture
1919 * to transition to a completed state, else {@code false}
1920 */
1921 public boolean complete(T value) {
1922 boolean triggered = completeValue(value);
1923 postComplete();
1924 return triggered;
1925 }
1926
1927 /**
1928 * If not already completed, causes invocations of {@link #get()}
1929 * and related methods to throw the given exception.
1930 *
1931 * @param ex the exception
1932 * @return {@code true} if this invocation caused this CompletableFuture
1933 * to transition to a completed state, else {@code false}
1934 */
1935 public boolean completeExceptionally(Throwable ex) {
1936 if (ex == null) throw new NullPointerException();
1937 boolean triggered = internalComplete(new AltResult(ex));
1938 postComplete();
1939 return triggered;
1940 }
1941
1942 public <U> CompletableFuture<U> thenApply(
1943 Function<? super T,? extends U> fn) {
1944 return uniApplyStage(null, fn);
1945 }
1946
1947 public <U> CompletableFuture<U> thenApplyAsync(
1948 Function<? super T,? extends U> fn) {
1949 return uniApplyStage(asyncPool, fn);
1950 }
1951
1952 public <U> CompletableFuture<U> thenApplyAsync(
1953 Function<? super T,? extends U> fn, Executor executor) {
1954 return uniApplyStage(screenExecutor(executor), fn);
1955 }
1956
1957 public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
1958 return uniAcceptStage(null, action);
1959 }
1960
1961 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
1962 return uniAcceptStage(asyncPool, action);
1963 }
1964
1965 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
1966 Executor executor) {
1967 return uniAcceptStage(screenExecutor(executor), action);
1968 }
1969
1970 public CompletableFuture<Void> thenRun(Runnable action) {
1971 return uniRunStage(null, action);
1972 }
1973
1974 public CompletableFuture<Void> thenRunAsync(Runnable action) {
1975 return uniRunStage(asyncPool, action);
1976 }
1977
1978 public CompletableFuture<Void> thenRunAsync(Runnable action,
1979 Executor executor) {
1980 return uniRunStage(screenExecutor(executor), action);
1981 }
1982
1983 public <U,V> CompletableFuture<V> thenCombine(
1984 CompletionStage<? extends U> other,
1985 BiFunction<? super T,? super U,? extends V> fn) {
1986 return biApplyStage(null, other, fn);
1987 }
1988
1989 public <U,V> CompletableFuture<V> thenCombineAsync(
1990 CompletionStage<? extends U> other,
1991 BiFunction<? super T,? super U,? extends V> fn) {
1992 return biApplyStage(asyncPool, other, fn);
1993 }
1994
1995 public <U,V> CompletableFuture<V> thenCombineAsync(
1996 CompletionStage<? extends U> other,
1997 BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
1998 return biApplyStage(screenExecutor(executor), other, fn);
1999 }
2000
2001 public <U> CompletableFuture<Void> thenAcceptBoth(
2002 CompletionStage<? extends U> other,
2003 BiConsumer<? super T, ? super U> action) {
2004 return biAcceptStage(null, other, action);
2005 }
2006
2007 public <U> CompletableFuture<Void> thenAcceptBothAsync(
2008 CompletionStage<? extends U> other,
2009 BiConsumer<? super T, ? super U> action) {
2010 return biAcceptStage(asyncPool, other, action);
2011 }
2012
2013 public <U> CompletableFuture<Void> thenAcceptBothAsync(
2014 CompletionStage<? extends U> other,
2015 BiConsumer<? super T, ? super U> action, Executor executor) {
2016 return biAcceptStage(screenExecutor(executor), other, action);
2017 }
2018
2019 public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
2020 Runnable action) {
2021 return biRunStage(null, other, action);
2022 }
2023
2024 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2025 Runnable action) {
2026 return biRunStage(asyncPool, other, action);
2027 }
2028
2029 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2030 Runnable action,
2031 Executor executor) {
2032 return biRunStage(screenExecutor(executor), other, action);
2033 }
2034
2035 public <U> CompletableFuture<U> applyToEither(
2036 CompletionStage<? extends T> other, Function<? super T, U> fn) {
2037 return orApplyStage(null, other, fn);
2038 }
2039
2040 public <U> CompletableFuture<U> applyToEitherAsync(
2041 CompletionStage<? extends T> other, Function<? super T, U> fn) {
2042 return orApplyStage(asyncPool, other, fn);
2043 }
2044
2045 public <U> CompletableFuture<U> applyToEitherAsync(
2046 CompletionStage<? extends T> other, Function<? super T, U> fn,
2047 Executor executor) {
2048 return orApplyStage(screenExecutor(executor), other, fn);
2049 }
2050
2051 public CompletableFuture<Void> acceptEither(
2052 CompletionStage<? extends T> other, Consumer<? super T> action) {
2053 return orAcceptStage(null, other, action);
2054 }
2055
2056 public CompletableFuture<Void> acceptEitherAsync(
2057 CompletionStage<? extends T> other, Consumer<? super T> action) {
2058 return orAcceptStage(asyncPool, other, action);
2059 }
2060
2061 public CompletableFuture<Void> acceptEitherAsync(
2062 CompletionStage<? extends T> other, Consumer<? super T> action,
2063 Executor executor) {
2064 return orAcceptStage(screenExecutor(executor), other, action);
2065 }
2066
2067 public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
2068 Runnable action) {
2069 return orRunStage(null, other, action);
2070 }
2071
2072 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2073 Runnable action) {
2074 return orRunStage(asyncPool, other, action);
2075 }
2076
2077 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2078 Runnable action,
2079 Executor executor) {
2080 return orRunStage(screenExecutor(executor), other, action);
2081 }
2082
2083 public <U> CompletableFuture<U> thenCompose(
2084 Function<? super T, ? extends CompletionStage<U>> fn) {
2085 return uniComposeStage(null, fn);
2086 }
2087
2088 public <U> CompletableFuture<U> thenComposeAsync(
2089 Function<? super T, ? extends CompletionStage<U>> fn) {
2090 return uniComposeStage(asyncPool, fn);
2091 }
2092
2093 public <U> CompletableFuture<U> thenComposeAsync(
2094 Function<? super T, ? extends CompletionStage<U>> fn,
2095 Executor executor) {
2096 return uniComposeStage(screenExecutor(executor), fn);
2097 }
2098
2099 public CompletableFuture<T> whenComplete(
2100 BiConsumer<? super T, ? super Throwable> action) {
2101 return uniWhenCompleteStage(null, action);
2102 }
2103
2104 public CompletableFuture<T> whenCompleteAsync(
2105 BiConsumer<? super T, ? super Throwable> action) {
2106 return uniWhenCompleteStage(asyncPool, action);
2107 }
2108
2109 public CompletableFuture<T> whenCompleteAsync(
2110 BiConsumer<? super T, ? super Throwable> action, Executor executor) {
2111 return uniWhenCompleteStage(screenExecutor(executor), action);
2112 }
2113
2114 public <U> CompletableFuture<U> handle(
2115 BiFunction<? super T, Throwable, ? extends U> fn) {
2116 return uniHandleStage(null, fn);
2117 }
2118
2119 public <U> CompletableFuture<U> handleAsync(
2120 BiFunction<? super T, Throwable, ? extends U> fn) {
2121 return uniHandleStage(asyncPool, fn);
2122 }
2123
2124 public <U> CompletableFuture<U> handleAsync(
2125 BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
2126 return uniHandleStage(screenExecutor(executor), fn);
2127 }
2128
2129 /**
2130 * Returns this CompletableFuture.
2131 *
2132 * @return this CompletableFuture
2133 */
2134 public CompletableFuture<T> toCompletableFuture() {
2135 return this;
2136 }
2137
2138 // not in interface CompletionStage
2139
2140 /**
2141 * Returns a new CompletableFuture that is completed when this
2142 * CompletableFuture completes, with the result of the given
2143 * function of the exception triggering this CompletableFuture's
2144 * completion when it completes exceptionally; otherwise, if this
2145 * CompletableFuture completes normally, then the returned
2146 * CompletableFuture also completes normally with the same value.
2147 * Note: More flexible versions of this functionality are
2148 * available using methods {@code whenComplete} and {@code handle}.
2149 *
2150 * @param fn the function to use to compute the value of the
2151 * returned CompletableFuture if this CompletableFuture completed
2152 * exceptionally
2153 * @return the new CompletableFuture
2154 */
2155 public CompletableFuture<T> exceptionally(
2156 Function<Throwable, ? extends T> fn) {
2157 return uniExceptionallyStage(fn);
2158 }
2159
2160 /* ------------- Arbitrary-arity constructions -------------- */
2161
2162 /**
2163 * Returns a new CompletableFuture that is completed when all of
2164 * the given CompletableFutures complete. If any of the given
2165 * CompletableFutures complete exceptionally, then the returned
2166 * CompletableFuture also does so, with a CompletionException
2167 * holding this exception as its cause. Otherwise, the results,
2168 * if any, of the given CompletableFutures are not reflected in
2169 * the returned CompletableFuture, but may be obtained by
2170 * inspecting them individually. If no CompletableFutures are
2171 * provided, returns a CompletableFuture completed with the value
2172 * {@code null}.
2173 *
2174 * <p>Among the applications of this method is to await completion
2175 * of a set of independent CompletableFutures before continuing a
2176 * program, as in: {@code CompletableFuture.allOf(c1, c2,
2177 * c3).join();}.
2178 *
2179 * @param cfs the CompletableFutures
2180 * @return a new CompletableFuture that is completed when all of the
2181 * given CompletableFutures complete
2182 * @throws NullPointerException if the array or any of its elements are
2183 * {@code null}
2184 */
2185 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
2186 return andTree(cfs, 0, cfs.length - 1);
2187 }
2188
2189 /**
2190 * Returns a new CompletableFuture that is completed when any of
2191 * the given CompletableFutures complete, with the same result.
2192 * Otherwise, if it completed exceptionally, the returned
2193 * CompletableFuture also does so, with a CompletionException
2194 * holding this exception as its cause. If no CompletableFutures
2195 * are provided, returns an incomplete CompletableFuture.
2196 *
2197 * @param cfs the CompletableFutures
2198 * @return a new CompletableFuture that is completed with the
2199 * result or exception of any of the given CompletableFutures when
2200 * one completes
2201 * @throws NullPointerException if the array or any of its elements are
2202 * {@code null}
2203 */
2204 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
2205 return orTree(cfs, 0, cfs.length - 1);
2206 }
2207
2208 /* ------------- Control and status methods -------------- */
2209
2210 /**
2211 * If not already completed, completes this CompletableFuture with
2212 * a {@link CancellationException}. Dependent CompletableFutures
2213 * that have not already completed will also complete
2214 * exceptionally, with a {@link CompletionException} caused by
2215 * this {@code CancellationException}.
2216 *
2217 * @param mayInterruptIfRunning this value has no effect in this
2218 * implementation because interrupts are not used to control
2219 * processing.
2220 *
2221 * @return {@code true} if this task is now cancelled
2222 */
2223 public boolean cancel(boolean mayInterruptIfRunning) {
2224 boolean cancelled = (result == null) &&
2225 internalComplete(new AltResult(new CancellationException()));
2226 postComplete();
2227 return cancelled || isCancelled();
2228 }
2229
2230 /**
2231 * Returns {@code true} if this CompletableFuture was cancelled
2232 * before it completed normally.
2233 *
2234 * @return {@code true} if this CompletableFuture was cancelled
2235 * before it completed normally
2236 */
2237 public boolean isCancelled() {
2238 Object r;
2239 return ((r = result) instanceof AltResult) &&
2240 (((AltResult)r).ex instanceof CancellationException);
2241 }
2242
2243 /**
2244 * Returns {@code true} if this CompletableFuture completed
2245 * exceptionally, in any way. Possible causes include
2246 * cancellation, explicit invocation of {@code
2247 * completeExceptionally}, and abrupt termination of a
2248 * CompletionStage action.
2249 *
2250 * @return {@code true} if this CompletableFuture completed
2251 * exceptionally
2252 */
2253 public boolean isCompletedExceptionally() {
2254 Object r;
2255 return ((r = result) instanceof AltResult) && r != NIL;
2256 }
2257
2258 /**
2259 * Forcibly sets or resets the value subsequently returned by
2260 * method {@link #get()} and related methods, whether or not
2261 * already completed. This method is designed for use only in
2262 * error recovery actions, and even in such situations may result
2263 * in ongoing dependent completions using established versus
2264 * overwritten outcomes.
2265 *
2266 * @param value the completion value
2267 */
2268 public void obtrudeValue(T value) {
2269 result = (value == null) ? NIL : value;
2270 postComplete();
2271 }
2272
2273 /**
2274 * Forcibly causes subsequent invocations of method {@link #get()}
2275 * and related methods to throw the given exception, whether or
2276 * not already completed. This method is designed for use only in
2277 * error recovery actions, and even in such situations may result
2278 * in ongoing dependent completions using established versus
2279 * overwritten outcomes.
2280 *
2281 * @param ex the exception
2282 * @throws NullPointerException if the exception is null
2283 */
2284 public void obtrudeException(Throwable ex) {
2285 if (ex == null) throw new NullPointerException();
2286 result = new AltResult(ex);
2287 postComplete();
2288 }
2289
2290 /**
2291 * Returns the estimated number of CompletableFutures whose
2292 * completions are awaiting completion of this CompletableFuture.
2293 * This method is designed for use in monitoring system state, not
2294 * for synchronization control.
2295 *
2296 * @return the number of dependent CompletableFutures
2297 */
2298 public int getNumberOfDependents() {
2299 int count = 0;
2300 for (Completion p = stack; p != null; p = p.next)
2301 ++count;
2302 return count;
2303 }
2304
2305 /**
2306 * Returns a string identifying this CompletableFuture, as well as
2307 * its completion state. The state, in brackets, contains the
2308 * String {@code "Completed Normally"} or the String {@code
2309 * "Completed Exceptionally"}, or the String {@code "Not
2310 * completed"} followed by the number of CompletableFutures
2311 * dependent upon its completion, if any.
2312 *
2313 * @return a string identifying this CompletableFuture, as well as its state
2314 */
2315 public String toString() {
2316 Object r = result;
2317 int count;
2318 return super.toString() +
2319 ((r == null) ?
2320 (((count = getNumberOfDependents()) == 0) ?
2321 "[Not completed]" :
2322 "[Not completed, " + count + " dependents]") :
2323 (((r instanceof AltResult) && ((AltResult)r).ex != null) ?
2324 "[Completed exceptionally]" :
2325 "[Completed normally]"));
2326 }
2327
2328 // Unsafe mechanics
2329 private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
2330 private static final long RESULT;
2331 private static final long STACK;
2332 private static final long NEXT;
2333 static {
2334 try {
2335 RESULT = U.objectFieldOffset
2336 (CompletableFuture.class.getDeclaredField("result"));
2337 STACK = U.objectFieldOffset
2338 (CompletableFuture.class.getDeclaredField("stack"));
2339 NEXT = U.objectFieldOffset
2340 (Completion.class.getDeclaredField("next"));
2341 } catch (ReflectiveOperationException e) {
2342 throw new Error(e);
2343 }
2344 }
2345 }