ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.125
Committed: Sat May 31 05:52:50 2014 UTC (10 years ago) by jsr166
Branch: MAIN
Changes since 1.124: +33 -33 lines
Log Message:
better variable names

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