ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.134
Committed: Thu Aug 28 12:13:25 2014 UTC (9 years, 9 months ago) by jsr166
Branch: MAIN
Changes since 1.133: +12 -24 lines
Log Message:
simplify Unsafe mechanics

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