ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.127
Committed: Sat Jun 14 14:24:06 2014 UTC (9 years, 11 months ago) by jsr166
Branch: MAIN
Changes since 1.126: +215 -255 lines
Log Message:
more readable and compact mechanics;
*Either methods should always claim before running action to avoid race

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