ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.138
Committed: Sun Jan 4 09:15:11 2015 UTC (9 years, 5 months ago) by jsr166
Branch: MAIN
Changes since 1.137: +15 -16 lines
Log Message:
standardize Unsafe mechanics; slightly smaller bytecode

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 return f.apply(t).toCompletableFuture();
943 } catch (Throwable ex) {
944 return new CompletableFuture<V>(encodeThrowable(ex));
945 }
946 }
947 CompletableFuture<V> d = new CompletableFuture<V>();
948 UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
949 push(c);
950 c.tryFire(SYNC);
951 return d;
952 }
953
954 /* ------------- Two-input Completions -------------- */
955
956 /** A Completion for an action with two sources */
957 @SuppressWarnings("serial")
958 abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> {
959 CompletableFuture<U> snd; // second source for action
960 BiCompletion(Executor executor, CompletableFuture<V> dep,
961 CompletableFuture<T> src, CompletableFuture<U> snd) {
962 super(executor, dep, src); this.snd = snd;
963 }
964 }
965
966 /** A Completion delegating to a BiCompletion */
967 @SuppressWarnings("serial")
968 static final class CoCompletion extends Completion {
969 BiCompletion<?,?,?> base;
970 CoCompletion(BiCompletion<?,?,?> base) { this.base = base; }
971 final CompletableFuture<?> tryFire(int mode) {
972 BiCompletion<?,?,?> c; CompletableFuture<?> d;
973 if ((c = base) == null || (d = c.tryFire(mode)) == null)
974 return null;
975 base = null; // detach
976 return d;
977 }
978 final boolean isLive() {
979 BiCompletion<?,?,?> c;
980 return (c = base) != null && c.dep != null;
981 }
982 }
983
984 /** Pushes completion to this and b unless both done. */
985 final void bipush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
986 if (c != null) {
987 Object r;
988 while ((r = result) == null && !tryPushStack(c))
989 lazySetNext(c, null); // clear on failure
990 if (b != null && b != this && b.result == null) {
991 Completion q = (r != null) ? c : new CoCompletion(c);
992 while (b.result == null && !b.tryPushStack(q))
993 lazySetNext(q, null); // clear on failure
994 }
995 }
996 }
997
998 /** Post-processing after successful BiCompletion tryFire. */
999 final CompletableFuture<T> postFire(CompletableFuture<?> a,
1000 CompletableFuture<?> b, int mode) {
1001 if (b != null && b.stack != null) { // clean second source
1002 if (mode < 0 || b.result == null)
1003 b.cleanStack();
1004 else
1005 b.postComplete();
1006 }
1007 return postFire(a, mode);
1008 }
1009
1010 @SuppressWarnings("serial")
1011 static final class BiApply<T,U,V> extends BiCompletion<T,U,V> {
1012 BiFunction<? super T,? super U,? extends V> fn;
1013 BiApply(Executor executor, CompletableFuture<V> dep,
1014 CompletableFuture<T> src, CompletableFuture<U> snd,
1015 BiFunction<? super T,? super U,? extends V> fn) {
1016 super(executor, dep, src, snd); this.fn = fn;
1017 }
1018 final CompletableFuture<V> tryFire(int mode) {
1019 CompletableFuture<V> d;
1020 CompletableFuture<T> a;
1021 CompletableFuture<U> b;
1022 if ((d = dep) == null ||
1023 !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this))
1024 return null;
1025 dep = null; src = null; snd = null; fn = null;
1026 return d.postFire(a, b, mode);
1027 }
1028 }
1029
1030 final <R,S> boolean biApply(CompletableFuture<R> a,
1031 CompletableFuture<S> b,
1032 BiFunction<? super R,? super S,? extends T> f,
1033 BiApply<R,S,T> c) {
1034 Object r, s; Throwable x;
1035 if (a == null || (r = a.result) == null ||
1036 b == null || (s = b.result) == null || f == null)
1037 return false;
1038 tryComplete: if (result == null) {
1039 if (r instanceof AltResult) {
1040 if ((x = ((AltResult)r).ex) != null) {
1041 completeThrowable(x, r);
1042 break tryComplete;
1043 }
1044 r = null;
1045 }
1046 if (s instanceof AltResult) {
1047 if ((x = ((AltResult)s).ex) != null) {
1048 completeThrowable(x, s);
1049 break tryComplete;
1050 }
1051 s = null;
1052 }
1053 try {
1054 if (c != null && !c.claim())
1055 return false;
1056 @SuppressWarnings("unchecked") R rr = (R) r;
1057 @SuppressWarnings("unchecked") S ss = (S) s;
1058 completeValue(f.apply(rr, ss));
1059 } catch (Throwable ex) {
1060 completeThrowable(ex);
1061 }
1062 }
1063 return true;
1064 }
1065
1066 private <U,V> CompletableFuture<V> biApplyStage(
1067 Executor e, CompletionStage<U> o,
1068 BiFunction<? super T,? super U,? extends V> f) {
1069 CompletableFuture<U> b;
1070 if (f == null || (b = o.toCompletableFuture()) == null)
1071 throw new NullPointerException();
1072 CompletableFuture<V> d = new CompletableFuture<V>();
1073 if (e != null || !d.biApply(this, b, f, null)) {
1074 BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);
1075 bipush(b, c);
1076 c.tryFire(SYNC);
1077 }
1078 return d;
1079 }
1080
1081 @SuppressWarnings("serial")
1082 static final class BiAccept<T,U> extends BiCompletion<T,U,Void> {
1083 BiConsumer<? super T,? super U> fn;
1084 BiAccept(Executor executor, CompletableFuture<Void> dep,
1085 CompletableFuture<T> src, CompletableFuture<U> snd,
1086 BiConsumer<? super T,? super U> fn) {
1087 super(executor, dep, src, snd); this.fn = fn;
1088 }
1089 final CompletableFuture<Void> tryFire(int mode) {
1090 CompletableFuture<Void> d;
1091 CompletableFuture<T> a;
1092 CompletableFuture<U> b;
1093 if ((d = dep) == null ||
1094 !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1095 return null;
1096 dep = null; src = null; snd = null; fn = null;
1097 return d.postFire(a, b, mode);
1098 }
1099 }
1100
1101 final <R,S> boolean biAccept(CompletableFuture<R> a,
1102 CompletableFuture<S> b,
1103 BiConsumer<? super R,? super S> f,
1104 BiAccept<R,S> c) {
1105 Object r, s; Throwable x;
1106 if (a == null || (r = a.result) == null ||
1107 b == null || (s = b.result) == null || f == null)
1108 return false;
1109 tryComplete: if (result == null) {
1110 if (r instanceof AltResult) {
1111 if ((x = ((AltResult)r).ex) != null) {
1112 completeThrowable(x, r);
1113 break tryComplete;
1114 }
1115 r = null;
1116 }
1117 if (s instanceof AltResult) {
1118 if ((x = ((AltResult)s).ex) != null) {
1119 completeThrowable(x, s);
1120 break tryComplete;
1121 }
1122 s = null;
1123 }
1124 try {
1125 if (c != null && !c.claim())
1126 return false;
1127 @SuppressWarnings("unchecked") R rr = (R) r;
1128 @SuppressWarnings("unchecked") S ss = (S) s;
1129 f.accept(rr, ss);
1130 completeNull();
1131 } catch (Throwable ex) {
1132 completeThrowable(ex);
1133 }
1134 }
1135 return true;
1136 }
1137
1138 private <U> CompletableFuture<Void> biAcceptStage(
1139 Executor e, CompletionStage<U> o,
1140 BiConsumer<? super T,? super U> f) {
1141 CompletableFuture<U> b;
1142 if (f == null || (b = o.toCompletableFuture()) == null)
1143 throw new NullPointerException();
1144 CompletableFuture<Void> d = new CompletableFuture<Void>();
1145 if (e != null || !d.biAccept(this, b, f, null)) {
1146 BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
1147 bipush(b, c);
1148 c.tryFire(SYNC);
1149 }
1150 return d;
1151 }
1152
1153 @SuppressWarnings("serial")
1154 static final class BiRun<T,U> extends BiCompletion<T,U,Void> {
1155 Runnable fn;
1156 BiRun(Executor executor, CompletableFuture<Void> dep,
1157 CompletableFuture<T> src,
1158 CompletableFuture<U> snd,
1159 Runnable fn) {
1160 super(executor, dep, src, snd); this.fn = fn;
1161 }
1162 final CompletableFuture<Void> tryFire(int mode) {
1163 CompletableFuture<Void> d;
1164 CompletableFuture<T> a;
1165 CompletableFuture<U> b;
1166 if ((d = dep) == null ||
1167 !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this))
1168 return null;
1169 dep = null; src = null; snd = null; fn = null;
1170 return d.postFire(a, b, mode);
1171 }
1172 }
1173
1174 final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b,
1175 Runnable f, BiRun<?,?> c) {
1176 Object r, s; Throwable x;
1177 if (a == null || (r = a.result) == null ||
1178 b == null || (s = b.result) == null || f == null)
1179 return false;
1180 if (result == null) {
1181 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1182 completeThrowable(x, r);
1183 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1184 completeThrowable(x, s);
1185 else
1186 try {
1187 if (c != null && !c.claim())
1188 return false;
1189 f.run();
1190 completeNull();
1191 } catch (Throwable ex) {
1192 completeThrowable(ex);
1193 }
1194 }
1195 return true;
1196 }
1197
1198 private CompletableFuture<Void> biRunStage(Executor e, CompletionStage<?> o,
1199 Runnable f) {
1200 CompletableFuture<?> b;
1201 if (f == null || (b = o.toCompletableFuture()) == null)
1202 throw new NullPointerException();
1203 CompletableFuture<Void> d = new CompletableFuture<Void>();
1204 if (e != null || !d.biRun(this, b, f, null)) {
1205 BiRun<T,?> c = new BiRun<>(e, d, this, b, f);
1206 bipush(b, c);
1207 c.tryFire(SYNC);
1208 }
1209 return d;
1210 }
1211
1212 @SuppressWarnings("serial")
1213 static final class BiRelay<T,U> extends BiCompletion<T,U,Void> { // for And
1214 BiRelay(CompletableFuture<Void> dep,
1215 CompletableFuture<T> src,
1216 CompletableFuture<U> snd) {
1217 super(null, dep, src, snd);
1218 }
1219 final CompletableFuture<Void> tryFire(int mode) {
1220 CompletableFuture<Void> d;
1221 CompletableFuture<T> a;
1222 CompletableFuture<U> b;
1223 if ((d = dep) == null || !d.biRelay(a = src, b = snd))
1224 return null;
1225 src = null; snd = null; dep = null;
1226 return d.postFire(a, b, mode);
1227 }
1228 }
1229
1230 boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1231 Object r, s; Throwable x;
1232 if (a == null || (r = a.result) == null ||
1233 b == null || (s = b.result) == null)
1234 return false;
1235 if (result == null) {
1236 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1237 completeThrowable(x, r);
1238 else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1239 completeThrowable(x, s);
1240 else
1241 completeNull();
1242 }
1243 return true;
1244 }
1245
1246 /** Recursively constructs a tree of completions. */
1247 static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs,
1248 int lo, int hi) {
1249 CompletableFuture<Void> d = new CompletableFuture<Void>();
1250 if (lo > hi) // empty
1251 d.result = NIL;
1252 else {
1253 CompletableFuture<?> a, b;
1254 int mid = (lo + hi) >>> 1;
1255 if ((a = (lo == mid ? cfs[lo] :
1256 andTree(cfs, lo, mid))) == null ||
1257 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1258 andTree(cfs, mid+1, hi))) == null)
1259 throw new NullPointerException();
1260 if (!d.biRelay(a, b)) {
1261 BiRelay<?,?> c = new BiRelay<>(d, a, b);
1262 a.bipush(b, c);
1263 c.tryFire(SYNC);
1264 }
1265 }
1266 return d;
1267 }
1268
1269 /* ------------- Projected (Ored) BiCompletions -------------- */
1270
1271 /** Pushes completion to this and b unless either done. */
1272 final void orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
1273 if (c != null) {
1274 while ((b == null || b.result == null) && result == null) {
1275 if (tryPushStack(c)) {
1276 if (b != null && b != this && b.result == null) {
1277 Completion q = new CoCompletion(c);
1278 while (result == null && b.result == null &&
1279 !b.tryPushStack(q))
1280 lazySetNext(q, null); // clear on failure
1281 }
1282 break;
1283 }
1284 lazySetNext(c, null); // clear on failure
1285 }
1286 }
1287 }
1288
1289 @SuppressWarnings("serial")
1290 static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> {
1291 Function<? super T,? extends V> fn;
1292 OrApply(Executor executor, CompletableFuture<V> dep,
1293 CompletableFuture<T> src,
1294 CompletableFuture<U> snd,
1295 Function<? super T,? extends V> fn) {
1296 super(executor, dep, src, snd); this.fn = fn;
1297 }
1298 final CompletableFuture<V> tryFire(int mode) {
1299 CompletableFuture<V> d;
1300 CompletableFuture<T> a;
1301 CompletableFuture<U> b;
1302 if ((d = dep) == null ||
1303 !d.orApply(a = src, b = snd, fn, mode > 0 ? null : this))
1304 return null;
1305 dep = null; src = null; snd = null; fn = null;
1306 return d.postFire(a, b, mode);
1307 }
1308 }
1309
1310 final <R,S extends R> boolean orApply(CompletableFuture<R> a,
1311 CompletableFuture<S> b,
1312 Function<? super R, ? extends T> f,
1313 OrApply<R,S,T> c) {
1314 Object r; Throwable x;
1315 if (a == null || b == null ||
1316 ((r = a.result) == null && (r = b.result) == null) || f == null)
1317 return false;
1318 tryComplete: if (result == null) {
1319 try {
1320 if (c != null && !c.claim())
1321 return false;
1322 if (r instanceof AltResult) {
1323 if ((x = ((AltResult)r).ex) != null) {
1324 completeThrowable(x, r);
1325 break tryComplete;
1326 }
1327 r = null;
1328 }
1329 @SuppressWarnings("unchecked") R rr = (R) r;
1330 completeValue(f.apply(rr));
1331 } catch (Throwable ex) {
1332 completeThrowable(ex);
1333 }
1334 }
1335 return true;
1336 }
1337
1338 private <U extends T,V> CompletableFuture<V> orApplyStage(
1339 Executor e, CompletionStage<U> o,
1340 Function<? super T, ? extends V> f) {
1341 CompletableFuture<U> b;
1342 if (f == null || (b = o.toCompletableFuture()) == null)
1343 throw new NullPointerException();
1344 CompletableFuture<V> d = new CompletableFuture<V>();
1345 if (e != null || !d.orApply(this, b, f, null)) {
1346 OrApply<T,U,V> c = new OrApply<T,U,V>(e, d, this, b, f);
1347 orpush(b, c);
1348 c.tryFire(SYNC);
1349 }
1350 return d;
1351 }
1352
1353 @SuppressWarnings("serial")
1354 static final class OrAccept<T,U extends T> extends BiCompletion<T,U,Void> {
1355 Consumer<? super T> fn;
1356 OrAccept(Executor executor, CompletableFuture<Void> dep,
1357 CompletableFuture<T> src,
1358 CompletableFuture<U> snd,
1359 Consumer<? super T> fn) {
1360 super(executor, dep, src, snd); this.fn = fn;
1361 }
1362 final CompletableFuture<Void> tryFire(int mode) {
1363 CompletableFuture<Void> d;
1364 CompletableFuture<T> a;
1365 CompletableFuture<U> b;
1366 if ((d = dep) == null ||
1367 !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1368 return null;
1369 dep = null; src = null; snd = null; fn = null;
1370 return d.postFire(a, b, mode);
1371 }
1372 }
1373
1374 final <R,S extends R> boolean orAccept(CompletableFuture<R> a,
1375 CompletableFuture<S> b,
1376 Consumer<? super R> f,
1377 OrAccept<R,S> c) {
1378 Object r; Throwable x;
1379 if (a == null || b == null ||
1380 ((r = a.result) == null && (r = b.result) == null) || f == null)
1381 return false;
1382 tryComplete: if (result == null) {
1383 try {
1384 if (c != null && !c.claim())
1385 return false;
1386 if (r instanceof AltResult) {
1387 if ((x = ((AltResult)r).ex) != null) {
1388 completeThrowable(x, r);
1389 break tryComplete;
1390 }
1391 r = null;
1392 }
1393 @SuppressWarnings("unchecked") R rr = (R) r;
1394 f.accept(rr);
1395 completeNull();
1396 } catch (Throwable ex) {
1397 completeThrowable(ex);
1398 }
1399 }
1400 return true;
1401 }
1402
1403 private <U extends T> CompletableFuture<Void> orAcceptStage(
1404 Executor e, CompletionStage<U> o, Consumer<? super T> f) {
1405 CompletableFuture<U> b;
1406 if (f == null || (b = o.toCompletableFuture()) == null)
1407 throw new NullPointerException();
1408 CompletableFuture<Void> d = new CompletableFuture<Void>();
1409 if (e != null || !d.orAccept(this, b, f, null)) {
1410 OrAccept<T,U> c = new OrAccept<T,U>(e, d, this, b, f);
1411 orpush(b, c);
1412 c.tryFire(SYNC);
1413 }
1414 return d;
1415 }
1416
1417 @SuppressWarnings("serial")
1418 static final class OrRun<T,U> extends BiCompletion<T,U,Void> {
1419 Runnable fn;
1420 OrRun(Executor executor, CompletableFuture<Void> dep,
1421 CompletableFuture<T> src,
1422 CompletableFuture<U> snd,
1423 Runnable fn) {
1424 super(executor, dep, src, snd); this.fn = fn;
1425 }
1426 final CompletableFuture<Void> tryFire(int mode) {
1427 CompletableFuture<Void> d;
1428 CompletableFuture<T> a;
1429 CompletableFuture<U> b;
1430 if ((d = dep) == null ||
1431 !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this))
1432 return null;
1433 dep = null; src = null; snd = null; fn = null;
1434 return d.postFire(a, b, mode);
1435 }
1436 }
1437
1438 final boolean orRun(CompletableFuture<?> a, CompletableFuture<?> b,
1439 Runnable f, OrRun<?,?> c) {
1440 Object r; Throwable x;
1441 if (a == null || b == null ||
1442 ((r = a.result) == null && (r = b.result) == null) || f == null)
1443 return false;
1444 if (result == null) {
1445 try {
1446 if (c != null && !c.claim())
1447 return false;
1448 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1449 completeThrowable(x, r);
1450 else {
1451 f.run();
1452 completeNull();
1453 }
1454 } catch (Throwable ex) {
1455 completeThrowable(ex);
1456 }
1457 }
1458 return true;
1459 }
1460
1461 private CompletableFuture<Void> orRunStage(Executor e, CompletionStage<?> o,
1462 Runnable f) {
1463 CompletableFuture<?> b;
1464 if (f == null || (b = o.toCompletableFuture()) == null)
1465 throw new NullPointerException();
1466 CompletableFuture<Void> d = new CompletableFuture<Void>();
1467 if (e != null || !d.orRun(this, b, f, null)) {
1468 OrRun<T,?> c = new OrRun<>(e, d, this, b, f);
1469 orpush(b, c);
1470 c.tryFire(SYNC);
1471 }
1472 return d;
1473 }
1474
1475 @SuppressWarnings("serial")
1476 static final class OrRelay<T,U> extends BiCompletion<T,U,Object> { // for Or
1477 OrRelay(CompletableFuture<Object> dep, CompletableFuture<T> src,
1478 CompletableFuture<U> snd) {
1479 super(null, dep, src, snd);
1480 }
1481 final CompletableFuture<Object> tryFire(int mode) {
1482 CompletableFuture<Object> d;
1483 CompletableFuture<T> a;
1484 CompletableFuture<U> b;
1485 if ((d = dep) == null || !d.orRelay(a = src, b = snd))
1486 return null;
1487 src = null; snd = null; dep = null;
1488 return d.postFire(a, b, mode);
1489 }
1490 }
1491
1492 final boolean orRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1493 Object r;
1494 if (a == null || b == null ||
1495 ((r = a.result) == null && (r = b.result) == null))
1496 return false;
1497 if (result == null)
1498 completeRelay(r);
1499 return true;
1500 }
1501
1502 /** Recursively constructs a tree of completions. */
1503 static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs,
1504 int lo, int hi) {
1505 CompletableFuture<Object> d = new CompletableFuture<Object>();
1506 if (lo <= hi) {
1507 CompletableFuture<?> a, b;
1508 int mid = (lo + hi) >>> 1;
1509 if ((a = (lo == mid ? cfs[lo] :
1510 orTree(cfs, lo, mid))) == null ||
1511 (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1512 orTree(cfs, mid+1, hi))) == null)
1513 throw new NullPointerException();
1514 if (!d.orRelay(a, b)) {
1515 OrRelay<?,?> c = new OrRelay<>(d, a, b);
1516 a.orpush(b, c);
1517 c.tryFire(SYNC);
1518 }
1519 }
1520 return d;
1521 }
1522
1523 /* ------------- Zero-input Async forms -------------- */
1524
1525 @SuppressWarnings("serial")
1526 static final class AsyncSupply<T> extends ForkJoinTask<Void>
1527 implements Runnable, AsynchronousCompletionTask {
1528 CompletableFuture<T> dep; Supplier<T> fn;
1529 AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
1530 this.dep = dep; this.fn = fn;
1531 }
1532
1533 public final Void getRawResult() { return null; }
1534 public final void setRawResult(Void v) {}
1535 public final boolean exec() { run(); return true; }
1536
1537 public void run() {
1538 CompletableFuture<T> d; Supplier<T> f;
1539 if ((d = dep) != null && (f = fn) != null) {
1540 dep = null; fn = null;
1541 if (d.result == null) {
1542 try {
1543 d.completeValue(f.get());
1544 } catch (Throwable ex) {
1545 d.completeThrowable(ex);
1546 }
1547 }
1548 d.postComplete();
1549 }
1550 }
1551 }
1552
1553 static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
1554 Supplier<U> f) {
1555 if (f == null) throw new NullPointerException();
1556 CompletableFuture<U> d = new CompletableFuture<U>();
1557 e.execute(new AsyncSupply<U>(d, f));
1558 return d;
1559 }
1560
1561 @SuppressWarnings("serial")
1562 static final class AsyncRun extends ForkJoinTask<Void>
1563 implements Runnable, AsynchronousCompletionTask {
1564 CompletableFuture<Void> dep; Runnable fn;
1565 AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
1566 this.dep = dep; this.fn = fn;
1567 }
1568
1569 public final Void getRawResult() { return null; }
1570 public final void setRawResult(Void v) {}
1571 public final boolean exec() { run(); return true; }
1572
1573 public void run() {
1574 CompletableFuture<Void> d; Runnable f;
1575 if ((d = dep) != null && (f = fn) != null) {
1576 dep = null; fn = null;
1577 if (d.result == null) {
1578 try {
1579 f.run();
1580 d.completeNull();
1581 } catch (Throwable ex) {
1582 d.completeThrowable(ex);
1583 }
1584 }
1585 d.postComplete();
1586 }
1587 }
1588 }
1589
1590 static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
1591 if (f == null) throw new NullPointerException();
1592 CompletableFuture<Void> d = new CompletableFuture<Void>();
1593 e.execute(new AsyncRun(d, f));
1594 return d;
1595 }
1596
1597 /* ------------- Signallers -------------- */
1598
1599 /**
1600 * Completion for recording and releasing a waiting thread. This
1601 * class implements ManagedBlocker to avoid starvation when
1602 * blocking actions pile up in ForkJoinPools.
1603 */
1604 @SuppressWarnings("serial")
1605 static final class Signaller extends Completion
1606 implements ForkJoinPool.ManagedBlocker {
1607 long nanos; // wait time if timed
1608 final long deadline; // non-zero if timed
1609 volatile int interruptControl; // > 0: interruptible, < 0: interrupted
1610 volatile Thread thread;
1611
1612 Signaller(boolean interruptible, long nanos, long deadline) {
1613 this.thread = Thread.currentThread();
1614 this.interruptControl = interruptible ? 1 : 0;
1615 this.nanos = nanos;
1616 this.deadline = deadline;
1617 }
1618 final CompletableFuture<?> tryFire(int ignore) {
1619 Thread w; // no need to atomically claim
1620 if ((w = thread) != null) {
1621 thread = null;
1622 LockSupport.unpark(w);
1623 }
1624 return null;
1625 }
1626 public boolean isReleasable() {
1627 if (thread == null)
1628 return true;
1629 if (Thread.interrupted()) {
1630 int i = interruptControl;
1631 interruptControl = -1;
1632 if (i > 0)
1633 return true;
1634 }
1635 if (deadline != 0L &&
1636 (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {
1637 thread = null;
1638 return true;
1639 }
1640 return false;
1641 }
1642 public boolean block() {
1643 if (isReleasable())
1644 return true;
1645 else if (deadline == 0L)
1646 LockSupport.park(this);
1647 else if (nanos > 0L)
1648 LockSupport.parkNanos(this, nanos);
1649 return isReleasable();
1650 }
1651 final boolean isLive() { return thread != null; }
1652 }
1653
1654 /**
1655 * Returns raw result after waiting, or null if interruptible and
1656 * interrupted.
1657 */
1658 private Object waitingGet(boolean interruptible) {
1659 Signaller q = null;
1660 boolean queued = false;
1661 int spins = -1;
1662 Object r;
1663 while ((r = result) == null) {
1664 if (spins < 0)
1665 spins = (Runtime.getRuntime().availableProcessors() > 1) ?
1666 1 << 8 : 0; // Use brief spin-wait on multiprocessors
1667 else if (spins > 0) {
1668 if (ThreadLocalRandom.nextSecondarySeed() >= 0)
1669 --spins;
1670 }
1671 else if (q == null)
1672 q = new Signaller(interruptible, 0L, 0L);
1673 else if (!queued)
1674 queued = tryPushStack(q);
1675 else if (interruptible && q.interruptControl < 0) {
1676 q.thread = null;
1677 cleanStack();
1678 return null;
1679 }
1680 else if (q.thread != null && result == null) {
1681 try {
1682 ForkJoinPool.managedBlock(q);
1683 } catch (InterruptedException ie) {
1684 q.interruptControl = -1;
1685 }
1686 }
1687 }
1688 if (q != null) {
1689 q.thread = null;
1690 if (q.interruptControl < 0) {
1691 if (interruptible)
1692 r = null; // report interruption
1693 else
1694 Thread.currentThread().interrupt();
1695 }
1696 }
1697 postComplete();
1698 return r;
1699 }
1700
1701 /**
1702 * Returns raw result after waiting, or null if interrupted, or
1703 * throws TimeoutException on timeout.
1704 */
1705 private Object timedGet(long nanos) throws TimeoutException {
1706 if (Thread.interrupted())
1707 return null;
1708 if (nanos <= 0L)
1709 throw new TimeoutException();
1710 long d = System.nanoTime() + nanos;
1711 Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0
1712 boolean queued = false;
1713 Object r;
1714 // We intentionally don't spin here (as waitingGet does) because
1715 // the call to nanoTime() above acts much like a spin.
1716 while ((r = result) == null) {
1717 if (!queued)
1718 queued = tryPushStack(q);
1719 else if (q.interruptControl < 0 || q.nanos <= 0L) {
1720 q.thread = null;
1721 cleanStack();
1722 if (q.interruptControl < 0)
1723 return null;
1724 throw new TimeoutException();
1725 }
1726 else if (q.thread != null && result == null) {
1727 try {
1728 ForkJoinPool.managedBlock(q);
1729 } catch (InterruptedException ie) {
1730 q.interruptControl = -1;
1731 }
1732 }
1733 }
1734 if (q.interruptControl < 0)
1735 r = null;
1736 q.thread = null;
1737 postComplete();
1738 return r;
1739 }
1740
1741 /* ------------- public methods -------------- */
1742
1743 /**
1744 * Creates a new incomplete CompletableFuture.
1745 */
1746 public CompletableFuture() {
1747 }
1748
1749 /**
1750 * Creates a new complete CompletableFuture with given encoded result.
1751 */
1752 private CompletableFuture(Object r) {
1753 this.result = r;
1754 }
1755
1756 /**
1757 * Returns a new CompletableFuture that is asynchronously completed
1758 * by a task running in the {@link ForkJoinPool#commonPool()} with
1759 * the value obtained by calling the given Supplier.
1760 *
1761 * @param supplier a function returning the value to be used
1762 * to complete the returned CompletableFuture
1763 * @param <U> the function's return type
1764 * @return the new CompletableFuture
1765 */
1766 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
1767 return asyncSupplyStage(asyncPool, supplier);
1768 }
1769
1770 /**
1771 * Returns a new CompletableFuture that is asynchronously completed
1772 * by a task running in the given executor with the value obtained
1773 * by calling the given Supplier.
1774 *
1775 * @param supplier a function returning the value to be used
1776 * to complete the returned CompletableFuture
1777 * @param executor the executor to use for asynchronous execution
1778 * @param <U> the function's return type
1779 * @return the new CompletableFuture
1780 */
1781 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
1782 Executor executor) {
1783 return asyncSupplyStage(screenExecutor(executor), supplier);
1784 }
1785
1786 /**
1787 * Returns a new CompletableFuture that is asynchronously completed
1788 * by a task running in the {@link ForkJoinPool#commonPool()} after
1789 * it runs the given action.
1790 *
1791 * @param runnable the action to run before completing the
1792 * returned CompletableFuture
1793 * @return the new CompletableFuture
1794 */
1795 public static CompletableFuture<Void> runAsync(Runnable runnable) {
1796 return asyncRunStage(asyncPool, runnable);
1797 }
1798
1799 /**
1800 * Returns a new CompletableFuture that is asynchronously completed
1801 * by a task running in the given executor after it runs the given
1802 * action.
1803 *
1804 * @param runnable the action to run before completing the
1805 * returned CompletableFuture
1806 * @param executor the executor to use for asynchronous execution
1807 * @return the new CompletableFuture
1808 */
1809 public static CompletableFuture<Void> runAsync(Runnable runnable,
1810 Executor executor) {
1811 return asyncRunStage(screenExecutor(executor), runnable);
1812 }
1813
1814 /**
1815 * Returns a new CompletableFuture that is already completed with
1816 * the given value.
1817 *
1818 * @param value the value
1819 * @param <U> the type of the value
1820 * @return the completed CompletableFuture
1821 */
1822 public static <U> CompletableFuture<U> completedFuture(U value) {
1823 return new CompletableFuture<U>((value == null) ? NIL : value);
1824 }
1825
1826 /**
1827 * Returns {@code true} if completed in any fashion: normally,
1828 * exceptionally, or via cancellation.
1829 *
1830 * @return {@code true} if completed
1831 */
1832 public boolean isDone() {
1833 return result != null;
1834 }
1835
1836 /**
1837 * Waits if necessary for this future to complete, and then
1838 * returns its result.
1839 *
1840 * @return the result value
1841 * @throws CancellationException if this future was cancelled
1842 * @throws ExecutionException if this future completed exceptionally
1843 * @throws InterruptedException if the current thread was interrupted
1844 * while waiting
1845 */
1846 public T get() throws InterruptedException, ExecutionException {
1847 Object r;
1848 return reportGet((r = result) == null ? waitingGet(true) : r);
1849 }
1850
1851 /**
1852 * Waits if necessary for at most the given time for this future
1853 * to complete, and then returns its result, if available.
1854 *
1855 * @param timeout the maximum time to wait
1856 * @param unit the time unit of the timeout argument
1857 * @return the result value
1858 * @throws CancellationException if this future was cancelled
1859 * @throws ExecutionException if this future completed exceptionally
1860 * @throws InterruptedException if the current thread was interrupted
1861 * while waiting
1862 * @throws TimeoutException if the wait timed out
1863 */
1864 public T get(long timeout, TimeUnit unit)
1865 throws InterruptedException, ExecutionException, TimeoutException {
1866 Object r;
1867 long nanos = unit.toNanos(timeout);
1868 return reportGet((r = result) == null ? timedGet(nanos) : r);
1869 }
1870
1871 /**
1872 * Returns the result value when complete, or throws an
1873 * (unchecked) exception if completed exceptionally. To better
1874 * conform with the use of common functional forms, if a
1875 * computation involved in the completion of this
1876 * CompletableFuture threw an exception, this method throws an
1877 * (unchecked) {@link CompletionException} with the underlying
1878 * exception as its cause.
1879 *
1880 * @return the result value
1881 * @throws CancellationException if the computation was cancelled
1882 * @throws CompletionException if this future completed
1883 * exceptionally or a completion computation threw an exception
1884 */
1885 public T join() {
1886 Object r;
1887 return reportJoin((r = result) == null ? waitingGet(false) : r);
1888 }
1889
1890 /**
1891 * Returns the result value (or throws any encountered exception)
1892 * if completed, else returns the given valueIfAbsent.
1893 *
1894 * @param valueIfAbsent the value to return if not completed
1895 * @return the result value, if completed, else the given valueIfAbsent
1896 * @throws CancellationException if the computation was cancelled
1897 * @throws CompletionException if this future completed
1898 * exceptionally or a completion computation threw an exception
1899 */
1900 public T getNow(T valueIfAbsent) {
1901 Object r;
1902 return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
1903 }
1904
1905 /**
1906 * If not already completed, sets the value returned by {@link
1907 * #get()} and related methods to the given value.
1908 *
1909 * @param value the result value
1910 * @return {@code true} if this invocation caused this CompletableFuture
1911 * to transition to a completed state, else {@code false}
1912 */
1913 public boolean complete(T value) {
1914 boolean triggered = completeValue(value);
1915 postComplete();
1916 return triggered;
1917 }
1918
1919 /**
1920 * If not already completed, causes invocations of {@link #get()}
1921 * and related methods to throw the given exception.
1922 *
1923 * @param ex the exception
1924 * @return {@code true} if this invocation caused this CompletableFuture
1925 * to transition to a completed state, else {@code false}
1926 */
1927 public boolean completeExceptionally(Throwable ex) {
1928 if (ex == null) throw new NullPointerException();
1929 boolean triggered = internalComplete(new AltResult(ex));
1930 postComplete();
1931 return triggered;
1932 }
1933
1934 public <U> CompletableFuture<U> thenApply(
1935 Function<? super T,? extends U> fn) {
1936 return uniApplyStage(null, fn);
1937 }
1938
1939 public <U> CompletableFuture<U> thenApplyAsync(
1940 Function<? super T,? extends U> fn) {
1941 return uniApplyStage(asyncPool, fn);
1942 }
1943
1944 public <U> CompletableFuture<U> thenApplyAsync(
1945 Function<? super T,? extends U> fn, Executor executor) {
1946 return uniApplyStage(screenExecutor(executor), fn);
1947 }
1948
1949 public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
1950 return uniAcceptStage(null, action);
1951 }
1952
1953 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
1954 return uniAcceptStage(asyncPool, action);
1955 }
1956
1957 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
1958 Executor executor) {
1959 return uniAcceptStage(screenExecutor(executor), action);
1960 }
1961
1962 public CompletableFuture<Void> thenRun(Runnable action) {
1963 return uniRunStage(null, action);
1964 }
1965
1966 public CompletableFuture<Void> thenRunAsync(Runnable action) {
1967 return uniRunStage(asyncPool, action);
1968 }
1969
1970 public CompletableFuture<Void> thenRunAsync(Runnable action,
1971 Executor executor) {
1972 return uniRunStage(screenExecutor(executor), action);
1973 }
1974
1975 public <U,V> CompletableFuture<V> thenCombine(
1976 CompletionStage<? extends U> other,
1977 BiFunction<? super T,? super U,? extends V> fn) {
1978 return biApplyStage(null, other, fn);
1979 }
1980
1981 public <U,V> CompletableFuture<V> thenCombineAsync(
1982 CompletionStage<? extends U> other,
1983 BiFunction<? super T,? super U,? extends V> fn) {
1984 return biApplyStage(asyncPool, other, fn);
1985 }
1986
1987 public <U,V> CompletableFuture<V> thenCombineAsync(
1988 CompletionStage<? extends U> other,
1989 BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
1990 return biApplyStage(screenExecutor(executor), other, fn);
1991 }
1992
1993 public <U> CompletableFuture<Void> thenAcceptBoth(
1994 CompletionStage<? extends U> other,
1995 BiConsumer<? super T, ? super U> action) {
1996 return biAcceptStage(null, other, action);
1997 }
1998
1999 public <U> CompletableFuture<Void> thenAcceptBothAsync(
2000 CompletionStage<? extends U> other,
2001 BiConsumer<? super T, ? super U> action) {
2002 return biAcceptStage(asyncPool, other, action);
2003 }
2004
2005 public <U> CompletableFuture<Void> thenAcceptBothAsync(
2006 CompletionStage<? extends U> other,
2007 BiConsumer<? super T, ? super U> action, Executor executor) {
2008 return biAcceptStage(screenExecutor(executor), other, action);
2009 }
2010
2011 public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
2012 Runnable action) {
2013 return biRunStage(null, other, action);
2014 }
2015
2016 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2017 Runnable action) {
2018 return biRunStage(asyncPool, other, action);
2019 }
2020
2021 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2022 Runnable action,
2023 Executor executor) {
2024 return biRunStage(screenExecutor(executor), other, action);
2025 }
2026
2027 public <U> CompletableFuture<U> applyToEither(
2028 CompletionStage<? extends T> other, Function<? super T, U> fn) {
2029 return orApplyStage(null, other, fn);
2030 }
2031
2032 public <U> CompletableFuture<U> applyToEitherAsync(
2033 CompletionStage<? extends T> other, Function<? super T, U> fn) {
2034 return orApplyStage(asyncPool, other, fn);
2035 }
2036
2037 public <U> CompletableFuture<U> applyToEitherAsync(
2038 CompletionStage<? extends T> other, Function<? super T, U> fn,
2039 Executor executor) {
2040 return orApplyStage(screenExecutor(executor), other, fn);
2041 }
2042
2043 public CompletableFuture<Void> acceptEither(
2044 CompletionStage<? extends T> other, Consumer<? super T> action) {
2045 return orAcceptStage(null, other, action);
2046 }
2047
2048 public CompletableFuture<Void> acceptEitherAsync(
2049 CompletionStage<? extends T> other, Consumer<? super T> action) {
2050 return orAcceptStage(asyncPool, other, action);
2051 }
2052
2053 public CompletableFuture<Void> acceptEitherAsync(
2054 CompletionStage<? extends T> other, Consumer<? super T> action,
2055 Executor executor) {
2056 return orAcceptStage(screenExecutor(executor), other, action);
2057 }
2058
2059 public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
2060 Runnable action) {
2061 return orRunStage(null, other, action);
2062 }
2063
2064 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2065 Runnable action) {
2066 return orRunStage(asyncPool, other, action);
2067 }
2068
2069 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2070 Runnable action,
2071 Executor executor) {
2072 return orRunStage(screenExecutor(executor), other, action);
2073 }
2074
2075 public <U> CompletableFuture<U> thenCompose(
2076 Function<? super T, ? extends CompletionStage<U>> fn) {
2077 return uniComposeStage(null, fn);
2078 }
2079
2080 public <U> CompletableFuture<U> thenComposeAsync(
2081 Function<? super T, ? extends CompletionStage<U>> fn) {
2082 return uniComposeStage(asyncPool, fn);
2083 }
2084
2085 public <U> CompletableFuture<U> thenComposeAsync(
2086 Function<? super T, ? extends CompletionStage<U>> fn,
2087 Executor executor) {
2088 return uniComposeStage(screenExecutor(executor), fn);
2089 }
2090
2091 public CompletableFuture<T> whenComplete(
2092 BiConsumer<? super T, ? super Throwable> action) {
2093 return uniWhenCompleteStage(null, action);
2094 }
2095
2096 public CompletableFuture<T> whenCompleteAsync(
2097 BiConsumer<? super T, ? super Throwable> action) {
2098 return uniWhenCompleteStage(asyncPool, action);
2099 }
2100
2101 public CompletableFuture<T> whenCompleteAsync(
2102 BiConsumer<? super T, ? super Throwable> action, Executor executor) {
2103 return uniWhenCompleteStage(screenExecutor(executor), action);
2104 }
2105
2106 public <U> CompletableFuture<U> handle(
2107 BiFunction<? super T, Throwable, ? extends U> fn) {
2108 return uniHandleStage(null, fn);
2109 }
2110
2111 public <U> CompletableFuture<U> handleAsync(
2112 BiFunction<? super T, Throwable, ? extends U> fn) {
2113 return uniHandleStage(asyncPool, fn);
2114 }
2115
2116 public <U> CompletableFuture<U> handleAsync(
2117 BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
2118 return uniHandleStage(screenExecutor(executor), fn);
2119 }
2120
2121 /**
2122 * Returns this CompletableFuture.
2123 *
2124 * @return this CompletableFuture
2125 */
2126 public CompletableFuture<T> toCompletableFuture() {
2127 return this;
2128 }
2129
2130 // not in interface CompletionStage
2131
2132 /**
2133 * Returns a new CompletableFuture that is completed when this
2134 * CompletableFuture completes, with the result of the given
2135 * function of the exception triggering this CompletableFuture's
2136 * completion when it completes exceptionally; otherwise, if this
2137 * CompletableFuture completes normally, then the returned
2138 * CompletableFuture also completes normally with the same value.
2139 * Note: More flexible versions of this functionality are
2140 * available using methods {@code whenComplete} and {@code handle}.
2141 *
2142 * @param fn the function to use to compute the value of the
2143 * returned CompletableFuture if this CompletableFuture completed
2144 * exceptionally
2145 * @return the new CompletableFuture
2146 */
2147 public CompletableFuture<T> exceptionally(
2148 Function<Throwable, ? extends T> fn) {
2149 return uniExceptionallyStage(fn);
2150 }
2151
2152 /* ------------- Arbitrary-arity constructions -------------- */
2153
2154 /**
2155 * Returns a new CompletableFuture that is completed when all of
2156 * the given CompletableFutures complete. If any of the given
2157 * CompletableFutures complete exceptionally, then the returned
2158 * CompletableFuture also does so, with a CompletionException
2159 * holding this exception as its cause. Otherwise, the results,
2160 * if any, of the given CompletableFutures are not reflected in
2161 * the returned CompletableFuture, but may be obtained by
2162 * inspecting them individually. If no CompletableFutures are
2163 * provided, returns a CompletableFuture completed with the value
2164 * {@code null}.
2165 *
2166 * <p>Among the applications of this method is to await completion
2167 * of a set of independent CompletableFutures before continuing a
2168 * program, as in: {@code CompletableFuture.allOf(c1, c2,
2169 * c3).join();}.
2170 *
2171 * @param cfs the CompletableFutures
2172 * @return a new CompletableFuture that is completed when all of the
2173 * given CompletableFutures complete
2174 * @throws NullPointerException if the array or any of its elements are
2175 * {@code null}
2176 */
2177 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
2178 return andTree(cfs, 0, cfs.length - 1);
2179 }
2180
2181 /**
2182 * Returns a new CompletableFuture that is completed when any of
2183 * the given CompletableFutures complete, with the same result.
2184 * Otherwise, if it completed exceptionally, the returned
2185 * CompletableFuture also does so, with a CompletionException
2186 * holding this exception as its cause. If no CompletableFutures
2187 * are provided, returns an incomplete CompletableFuture.
2188 *
2189 * @param cfs the CompletableFutures
2190 * @return a new CompletableFuture that is completed with the
2191 * result or exception of any of the given CompletableFutures when
2192 * one completes
2193 * @throws NullPointerException if the array or any of its elements are
2194 * {@code null}
2195 */
2196 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
2197 return orTree(cfs, 0, cfs.length - 1);
2198 }
2199
2200 /* ------------- Control and status methods -------------- */
2201
2202 /**
2203 * If not already completed, completes this CompletableFuture with
2204 * a {@link CancellationException}. Dependent CompletableFutures
2205 * that have not already completed will also complete
2206 * exceptionally, with a {@link CompletionException} caused by
2207 * this {@code CancellationException}.
2208 *
2209 * @param mayInterruptIfRunning this value has no effect in this
2210 * implementation because interrupts are not used to control
2211 * processing.
2212 *
2213 * @return {@code true} if this task is now cancelled
2214 */
2215 public boolean cancel(boolean mayInterruptIfRunning) {
2216 boolean cancelled = (result == null) &&
2217 internalComplete(new AltResult(new CancellationException()));
2218 postComplete();
2219 return cancelled || isCancelled();
2220 }
2221
2222 /**
2223 * Returns {@code true} if this CompletableFuture was cancelled
2224 * before it completed normally.
2225 *
2226 * @return {@code true} if this CompletableFuture was cancelled
2227 * before it completed normally
2228 */
2229 public boolean isCancelled() {
2230 Object r;
2231 return ((r = result) instanceof AltResult) &&
2232 (((AltResult)r).ex instanceof CancellationException);
2233 }
2234
2235 /**
2236 * Returns {@code true} if this CompletableFuture completed
2237 * exceptionally, in any way. Possible causes include
2238 * cancellation, explicit invocation of {@code
2239 * completeExceptionally}, and abrupt termination of a
2240 * CompletionStage action.
2241 *
2242 * @return {@code true} if this CompletableFuture completed
2243 * exceptionally
2244 */
2245 public boolean isCompletedExceptionally() {
2246 Object r;
2247 return ((r = result) instanceof AltResult) && r != NIL;
2248 }
2249
2250 /**
2251 * Forcibly sets or resets the value subsequently returned by
2252 * method {@link #get()} and related methods, whether or not
2253 * already completed. This method is designed for use only in
2254 * error recovery actions, and even in such situations may result
2255 * in ongoing dependent completions using established versus
2256 * overwritten outcomes.
2257 *
2258 * @param value the completion value
2259 */
2260 public void obtrudeValue(T value) {
2261 result = (value == null) ? NIL : value;
2262 postComplete();
2263 }
2264
2265 /**
2266 * Forcibly causes subsequent invocations of method {@link #get()}
2267 * and related methods to throw the given exception, whether or
2268 * not already completed. This method is designed for use only in
2269 * error recovery actions, and even in such situations may result
2270 * in ongoing dependent completions using established versus
2271 * overwritten outcomes.
2272 *
2273 * @param ex the exception
2274 * @throws NullPointerException if the exception is null
2275 */
2276 public void obtrudeException(Throwable ex) {
2277 if (ex == null) throw new NullPointerException();
2278 result = new AltResult(ex);
2279 postComplete();
2280 }
2281
2282 /**
2283 * Returns the estimated number of CompletableFutures whose
2284 * completions are awaiting completion of this CompletableFuture.
2285 * This method is designed for use in monitoring system state, not
2286 * for synchronization control.
2287 *
2288 * @return the number of dependent CompletableFutures
2289 */
2290 public int getNumberOfDependents() {
2291 int count = 0;
2292 for (Completion p = stack; p != null; p = p.next)
2293 ++count;
2294 return count;
2295 }
2296
2297 /**
2298 * Returns a string identifying this CompletableFuture, as well as
2299 * its completion state. The state, in brackets, contains the
2300 * String {@code "Completed Normally"} or the String {@code
2301 * "Completed Exceptionally"}, or the String {@code "Not
2302 * completed"} followed by the number of CompletableFutures
2303 * dependent upon its completion, if any.
2304 *
2305 * @return a string identifying this CompletableFuture, as well as its state
2306 */
2307 public String toString() {
2308 Object r = result;
2309 int count;
2310 return super.toString() +
2311 ((r == null) ?
2312 (((count = getNumberOfDependents()) == 0) ?
2313 "[Not completed]" :
2314 "[Not completed, " + count + " dependents]") :
2315 (((r instanceof AltResult) && ((AltResult)r).ex != null) ?
2316 "[Completed exceptionally]" :
2317 "[Completed normally]"));
2318 }
2319
2320 // Unsafe mechanics
2321 private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
2322 private static final long RESULT;
2323 private static final long STACK;
2324 private static final long NEXT;
2325 static {
2326 try {
2327 RESULT = U.objectFieldOffset
2328 (CompletableFuture.class.getDeclaredField("result"));
2329 STACK = U.objectFieldOffset
2330 (CompletableFuture.class.getDeclaredField("stack"));
2331 NEXT = U.objectFieldOffset
2332 (Completion.class.getDeclaredField("next"));
2333 } catch (ReflectiveOperationException e) {
2334 throw new Error(e);
2335 }
2336 }
2337 }