ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.130
Committed: Mon Jun 23 21:52:17 2014 UTC (9 years, 11 months ago) by jsr166
Branch: MAIN
Changes since 1.129: +7 -6 lines
Log Message:
update stale internal docs; s/exec/tryFire/g

File Contents

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