ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.133
Committed: Thu Aug 28 11:40:51 2014 UTC (9 years, 9 months ago) by jsr166
Branch: MAIN
Changes since 1.132: +12 -2 lines
Log Message:
AsyncSupply, AsyncRun should extend ForkJoinTask to avoid wrapping when submitted to ForkJoinPool

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 ForkJoinTask<Void>
1552 implements Runnable, AsynchronousCompletionTask {
1553 CompletableFuture<T> dep; Supplier<T> fn;
1554 AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
1555 this.dep = dep; this.fn = fn;
1556 }
1557
1558 public final Void getRawResult() { return null; }
1559 public final void setRawResult(Void v) {}
1560 public final boolean exec() { run(); return true; }
1561
1562 public void run() {
1563 CompletableFuture<T> d; Supplier<T> f;
1564 if ((d = dep) != null && (f = fn) != null) {
1565 dep = null; fn = null;
1566 if (d.result == null) {
1567 try {
1568 d.completeValue(f.get());
1569 } catch (Throwable ex) {
1570 d.completeThrowable(ex);
1571 }
1572 }
1573 d.postComplete();
1574 }
1575 }
1576 }
1577
1578 static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
1579 Supplier<U> f) {
1580 if (f == null) throw new NullPointerException();
1581 CompletableFuture<U> d = new CompletableFuture<U>();
1582 e.execute(new AsyncSupply<U>(d, f));
1583 return d;
1584 }
1585
1586 @SuppressWarnings("serial")
1587 static final class AsyncRun extends ForkJoinTask<Void>
1588 implements Runnable, AsynchronousCompletionTask {
1589 CompletableFuture<Void> dep; Runnable fn;
1590 AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
1591 this.dep = dep; this.fn = fn;
1592 }
1593
1594 public final Void getRawResult() { return null; }
1595 public final void setRawResult(Void v) {}
1596 public final boolean exec() { run(); return true; }
1597
1598 public void run() {
1599 CompletableFuture<Void> d; Runnable f;
1600 if ((d = dep) != null && (f = fn) != null) {
1601 dep = null; fn = null;
1602 if (d.result == null) {
1603 try {
1604 f.run();
1605 d.completeNull();
1606 } catch (Throwable ex) {
1607 d.completeThrowable(ex);
1608 }
1609 }
1610 d.postComplete();
1611 }
1612 }
1613 }
1614
1615 static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
1616 if (f == null) throw new NullPointerException();
1617 CompletableFuture<Void> d = new CompletableFuture<Void>();
1618 e.execute(new AsyncRun(d, f));
1619 return d;
1620 }
1621
1622 /* ------------- Signallers -------------- */
1623
1624 /**
1625 * Completion for recording and releasing a waiting thread. This
1626 * class implements ManagedBlocker to avoid starvation when
1627 * blocking actions pile up in ForkJoinPools.
1628 */
1629 @SuppressWarnings("serial")
1630 static final class Signaller extends Completion
1631 implements ForkJoinPool.ManagedBlocker {
1632 long nanos; // wait time if timed
1633 final long deadline; // non-zero if timed
1634 volatile int interruptControl; // > 0: interruptible, < 0: interrupted
1635 volatile Thread thread;
1636
1637 Signaller(boolean interruptible, long nanos, long deadline) {
1638 this.thread = Thread.currentThread();
1639 this.interruptControl = interruptible ? 1 : 0;
1640 this.nanos = nanos;
1641 this.deadline = deadline;
1642 }
1643 final CompletableFuture<?> tryFire(int ignore) {
1644 Thread w; // no need to atomically claim
1645 if ((w = thread) != null) {
1646 thread = null;
1647 LockSupport.unpark(w);
1648 }
1649 return null;
1650 }
1651 public boolean isReleasable() {
1652 if (thread == null)
1653 return true;
1654 if (Thread.interrupted()) {
1655 int i = interruptControl;
1656 interruptControl = -1;
1657 if (i > 0)
1658 return true;
1659 }
1660 if (deadline != 0L &&
1661 (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {
1662 thread = null;
1663 return true;
1664 }
1665 return false;
1666 }
1667 public boolean block() {
1668 if (isReleasable())
1669 return true;
1670 else if (deadline == 0L)
1671 LockSupport.park(this);
1672 else if (nanos > 0L)
1673 LockSupport.parkNanos(this, nanos);
1674 return isReleasable();
1675 }
1676 final boolean isLive() { return thread != null; }
1677 }
1678
1679 /**
1680 * Returns raw result after waiting, or null if interruptible and
1681 * interrupted.
1682 */
1683 private Object waitingGet(boolean interruptible) {
1684 Signaller q = null;
1685 boolean queued = false;
1686 int spins = -1;
1687 Object r;
1688 while ((r = result) == null) {
1689 if (spins < 0)
1690 spins = (Runtime.getRuntime().availableProcessors() > 1) ?
1691 1 << 8 : 0; // Use brief spin-wait on multiprocessors
1692 else if (spins > 0) {
1693 if (ThreadLocalRandom.nextSecondarySeed() >= 0)
1694 --spins;
1695 }
1696 else if (q == null)
1697 q = new Signaller(interruptible, 0L, 0L);
1698 else if (!queued)
1699 queued = tryPushStack(q);
1700 else if (interruptible && q.interruptControl < 0) {
1701 q.thread = null;
1702 cleanStack();
1703 return null;
1704 }
1705 else if (q.thread != null && result == null) {
1706 try {
1707 ForkJoinPool.managedBlock(q);
1708 } catch (InterruptedException ie) {
1709 q.interruptControl = -1;
1710 }
1711 }
1712 }
1713 if (q != null) {
1714 q.thread = null;
1715 if (q.interruptControl < 0) {
1716 if (interruptible)
1717 r = null; // report interruption
1718 else
1719 Thread.currentThread().interrupt();
1720 }
1721 }
1722 postComplete();
1723 return r;
1724 }
1725
1726 /**
1727 * Returns raw result after waiting, or null if interrupted, or
1728 * throws TimeoutException on timeout.
1729 */
1730 private Object timedGet(long nanos) throws TimeoutException {
1731 if (Thread.interrupted())
1732 return null;
1733 if (nanos <= 0L)
1734 throw new TimeoutException();
1735 long d = System.nanoTime() + nanos;
1736 Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0
1737 boolean queued = false;
1738 Object r;
1739 // We intentionally don't spin here (as waitingGet does) because
1740 // the call to nanoTime() above acts much like a spin.
1741 while ((r = result) == null) {
1742 if (!queued)
1743 queued = tryPushStack(q);
1744 else if (q.interruptControl < 0 || q.nanos <= 0L) {
1745 q.thread = null;
1746 cleanStack();
1747 if (q.interruptControl < 0)
1748 return null;
1749 throw new TimeoutException();
1750 }
1751 else if (q.thread != null && result == null) {
1752 try {
1753 ForkJoinPool.managedBlock(q);
1754 } catch (InterruptedException ie) {
1755 q.interruptControl = -1;
1756 }
1757 }
1758 }
1759 if (q.interruptControl < 0)
1760 r = null;
1761 q.thread = null;
1762 postComplete();
1763 return r;
1764 }
1765
1766 /* ------------- public methods -------------- */
1767
1768 /**
1769 * Creates a new incomplete CompletableFuture.
1770 */
1771 public CompletableFuture() {
1772 }
1773
1774 /**
1775 * Creates a new complete CompletableFuture with given encoded result.
1776 */
1777 private CompletableFuture(Object r) {
1778 this.result = r;
1779 }
1780
1781 /**
1782 * Returns a new CompletableFuture that is asynchronously completed
1783 * by a task running in the {@link ForkJoinPool#commonPool()} with
1784 * the value obtained by calling the given Supplier.
1785 *
1786 * @param supplier a function returning the value to be used
1787 * to complete the returned CompletableFuture
1788 * @param <U> the function's return type
1789 * @return the new CompletableFuture
1790 */
1791 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
1792 return asyncSupplyStage(asyncPool, supplier);
1793 }
1794
1795 /**
1796 * Returns a new CompletableFuture that is asynchronously completed
1797 * by a task running in the given executor with the value obtained
1798 * by calling the given Supplier.
1799 *
1800 * @param supplier a function returning the value to be used
1801 * to complete the returned CompletableFuture
1802 * @param executor the executor to use for asynchronous execution
1803 * @param <U> the function's return type
1804 * @return the new CompletableFuture
1805 */
1806 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
1807 Executor executor) {
1808 return asyncSupplyStage(screenExecutor(executor), supplier);
1809 }
1810
1811 /**
1812 * Returns a new CompletableFuture that is asynchronously completed
1813 * by a task running in the {@link ForkJoinPool#commonPool()} after
1814 * it runs the given action.
1815 *
1816 * @param runnable the action to run before completing the
1817 * returned CompletableFuture
1818 * @return the new CompletableFuture
1819 */
1820 public static CompletableFuture<Void> runAsync(Runnable runnable) {
1821 return asyncRunStage(asyncPool, runnable);
1822 }
1823
1824 /**
1825 * Returns a new CompletableFuture that is asynchronously completed
1826 * by a task running in the given executor after it runs the given
1827 * action.
1828 *
1829 * @param runnable the action to run before completing the
1830 * returned CompletableFuture
1831 * @param executor the executor to use for asynchronous execution
1832 * @return the new CompletableFuture
1833 */
1834 public static CompletableFuture<Void> runAsync(Runnable runnable,
1835 Executor executor) {
1836 return asyncRunStage(screenExecutor(executor), runnable);
1837 }
1838
1839 /**
1840 * Returns a new CompletableFuture that is already completed with
1841 * the given value.
1842 *
1843 * @param value the value
1844 * @param <U> the type of the value
1845 * @return the completed CompletableFuture
1846 */
1847 public static <U> CompletableFuture<U> completedFuture(U value) {
1848 return new CompletableFuture<U>((value == null) ? NIL : value);
1849 }
1850
1851 /**
1852 * Returns {@code true} if completed in any fashion: normally,
1853 * exceptionally, or via cancellation.
1854 *
1855 * @return {@code true} if completed
1856 */
1857 public boolean isDone() {
1858 return result != null;
1859 }
1860
1861 /**
1862 * Waits if necessary for this future to complete, and then
1863 * returns its result.
1864 *
1865 * @return the result value
1866 * @throws CancellationException if this future was cancelled
1867 * @throws ExecutionException if this future completed exceptionally
1868 * @throws InterruptedException if the current thread was interrupted
1869 * while waiting
1870 */
1871 public T get() throws InterruptedException, ExecutionException {
1872 Object r;
1873 return reportGet((r = result) == null ? waitingGet(true) : r);
1874 }
1875
1876 /**
1877 * Waits if necessary for at most the given time for this future
1878 * to complete, and then returns its result, if available.
1879 *
1880 * @param timeout the maximum time to wait
1881 * @param unit the time unit of the timeout argument
1882 * @return the result value
1883 * @throws CancellationException if this future was cancelled
1884 * @throws ExecutionException if this future completed exceptionally
1885 * @throws InterruptedException if the current thread was interrupted
1886 * while waiting
1887 * @throws TimeoutException if the wait timed out
1888 */
1889 public T get(long timeout, TimeUnit unit)
1890 throws InterruptedException, ExecutionException, TimeoutException {
1891 Object r;
1892 long nanos = unit.toNanos(timeout);
1893 return reportGet((r = result) == null ? timedGet(nanos) : r);
1894 }
1895
1896 /**
1897 * Returns the result value when complete, or throws an
1898 * (unchecked) exception if completed exceptionally. To better
1899 * conform with the use of common functional forms, if a
1900 * computation involved in the completion of this
1901 * CompletableFuture threw an exception, this method throws an
1902 * (unchecked) {@link CompletionException} with the underlying
1903 * exception as its cause.
1904 *
1905 * @return the result value
1906 * @throws CancellationException if the computation was cancelled
1907 * @throws CompletionException if this future completed
1908 * exceptionally or a completion computation threw an exception
1909 */
1910 public T join() {
1911 Object r;
1912 return reportJoin((r = result) == null ? waitingGet(false) : r);
1913 }
1914
1915 /**
1916 * Returns the result value (or throws any encountered exception)
1917 * if completed, else returns the given valueIfAbsent.
1918 *
1919 * @param valueIfAbsent the value to return if not completed
1920 * @return the result value, if completed, else the given valueIfAbsent
1921 * @throws CancellationException if the computation was cancelled
1922 * @throws CompletionException if this future completed
1923 * exceptionally or a completion computation threw an exception
1924 */
1925 public T getNow(T valueIfAbsent) {
1926 Object r;
1927 return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
1928 }
1929
1930 /**
1931 * If not already completed, sets the value returned by {@link
1932 * #get()} and related methods to the given value.
1933 *
1934 * @param value the result value
1935 * @return {@code true} if this invocation caused this CompletableFuture
1936 * to transition to a completed state, else {@code false}
1937 */
1938 public boolean complete(T value) {
1939 boolean triggered = completeValue(value);
1940 postComplete();
1941 return triggered;
1942 }
1943
1944 /**
1945 * If not already completed, causes invocations of {@link #get()}
1946 * and related methods to throw the given exception.
1947 *
1948 * @param ex the exception
1949 * @return {@code true} if this invocation caused this CompletableFuture
1950 * to transition to a completed state, else {@code false}
1951 */
1952 public boolean completeExceptionally(Throwable ex) {
1953 if (ex == null) throw new NullPointerException();
1954 boolean triggered = internalComplete(new AltResult(ex));
1955 postComplete();
1956 return triggered;
1957 }
1958
1959 public <U> CompletableFuture<U> thenApply(
1960 Function<? super T,? extends U> fn) {
1961 return uniApplyStage(null, fn);
1962 }
1963
1964 public <U> CompletableFuture<U> thenApplyAsync(
1965 Function<? super T,? extends U> fn) {
1966 return uniApplyStage(asyncPool, fn);
1967 }
1968
1969 public <U> CompletableFuture<U> thenApplyAsync(
1970 Function<? super T,? extends U> fn, Executor executor) {
1971 return uniApplyStage(screenExecutor(executor), fn);
1972 }
1973
1974 public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
1975 return uniAcceptStage(null, action);
1976 }
1977
1978 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
1979 return uniAcceptStage(asyncPool, action);
1980 }
1981
1982 public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
1983 Executor executor) {
1984 return uniAcceptStage(screenExecutor(executor), action);
1985 }
1986
1987 public CompletableFuture<Void> thenRun(Runnable action) {
1988 return uniRunStage(null, action);
1989 }
1990
1991 public CompletableFuture<Void> thenRunAsync(Runnable action) {
1992 return uniRunStage(asyncPool, action);
1993 }
1994
1995 public CompletableFuture<Void> thenRunAsync(Runnable action,
1996 Executor executor) {
1997 return uniRunStage(screenExecutor(executor), action);
1998 }
1999
2000 public <U,V> CompletableFuture<V> thenCombine(
2001 CompletionStage<? extends U> other,
2002 BiFunction<? super T,? super U,? extends V> fn) {
2003 return biApplyStage(null, 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) {
2009 return biApplyStage(asyncPool, other, fn);
2010 }
2011
2012 public <U,V> CompletableFuture<V> thenCombineAsync(
2013 CompletionStage<? extends U> other,
2014 BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
2015 return biApplyStage(screenExecutor(executor), other, fn);
2016 }
2017
2018 public <U> CompletableFuture<Void> thenAcceptBoth(
2019 CompletionStage<? extends U> other,
2020 BiConsumer<? super T, ? super U> action) {
2021 return biAcceptStage(null, other, action);
2022 }
2023
2024 public <U> CompletableFuture<Void> thenAcceptBothAsync(
2025 CompletionStage<? extends U> other,
2026 BiConsumer<? super T, ? super U> action) {
2027 return biAcceptStage(asyncPool, other, action);
2028 }
2029
2030 public <U> CompletableFuture<Void> thenAcceptBothAsync(
2031 CompletionStage<? extends U> other,
2032 BiConsumer<? super T, ? super U> action, Executor executor) {
2033 return biAcceptStage(screenExecutor(executor), other, action);
2034 }
2035
2036 public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
2037 Runnable action) {
2038 return biRunStage(null, other, action);
2039 }
2040
2041 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2042 Runnable action) {
2043 return biRunStage(asyncPool, other, action);
2044 }
2045
2046 public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2047 Runnable action,
2048 Executor executor) {
2049 return biRunStage(screenExecutor(executor), other, action);
2050 }
2051
2052 public <U> CompletableFuture<U> applyToEither(
2053 CompletionStage<? extends T> other, Function<? super T, U> fn) {
2054 return orApplyStage(null, other, fn);
2055 }
2056
2057 public <U> CompletableFuture<U> applyToEitherAsync(
2058 CompletionStage<? extends T> other, Function<? super T, U> fn) {
2059 return orApplyStage(asyncPool, other, fn);
2060 }
2061
2062 public <U> CompletableFuture<U> applyToEitherAsync(
2063 CompletionStage<? extends T> other, Function<? super T, U> fn,
2064 Executor executor) {
2065 return orApplyStage(screenExecutor(executor), other, fn);
2066 }
2067
2068 public CompletableFuture<Void> acceptEither(
2069 CompletionStage<? extends T> other, Consumer<? super T> action) {
2070 return orAcceptStage(null, other, action);
2071 }
2072
2073 public CompletableFuture<Void> acceptEitherAsync(
2074 CompletionStage<? extends T> other, Consumer<? super T> action) {
2075 return orAcceptStage(asyncPool, other, action);
2076 }
2077
2078 public CompletableFuture<Void> acceptEitherAsync(
2079 CompletionStage<? extends T> other, Consumer<? super T> action,
2080 Executor executor) {
2081 return orAcceptStage(screenExecutor(executor), other, action);
2082 }
2083
2084 public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
2085 Runnable action) {
2086 return orRunStage(null, other, action);
2087 }
2088
2089 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2090 Runnable action) {
2091 return orRunStage(asyncPool, other, action);
2092 }
2093
2094 public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2095 Runnable action,
2096 Executor executor) {
2097 return orRunStage(screenExecutor(executor), other, action);
2098 }
2099
2100 public <U> CompletableFuture<U> thenCompose(
2101 Function<? super T, ? extends CompletionStage<U>> fn) {
2102 return uniComposeStage(null, fn);
2103 }
2104
2105 public <U> CompletableFuture<U> thenComposeAsync(
2106 Function<? super T, ? extends CompletionStage<U>> fn) {
2107 return uniComposeStage(asyncPool, fn);
2108 }
2109
2110 public <U> CompletableFuture<U> thenComposeAsync(
2111 Function<? super T, ? extends CompletionStage<U>> fn,
2112 Executor executor) {
2113 return uniComposeStage(screenExecutor(executor), fn);
2114 }
2115
2116 public CompletableFuture<T> whenComplete(
2117 BiConsumer<? super T, ? super Throwable> action) {
2118 return uniWhenCompleteStage(null, action);
2119 }
2120
2121 public CompletableFuture<T> whenCompleteAsync(
2122 BiConsumer<? super T, ? super Throwable> action) {
2123 return uniWhenCompleteStage(asyncPool, action);
2124 }
2125
2126 public CompletableFuture<T> whenCompleteAsync(
2127 BiConsumer<? super T, ? super Throwable> action, Executor executor) {
2128 return uniWhenCompleteStage(screenExecutor(executor), action);
2129 }
2130
2131 public <U> CompletableFuture<U> handle(
2132 BiFunction<? super T, Throwable, ? extends U> fn) {
2133 return uniHandleStage(null, fn);
2134 }
2135
2136 public <U> CompletableFuture<U> handleAsync(
2137 BiFunction<? super T, Throwable, ? extends U> fn) {
2138 return uniHandleStage(asyncPool, fn);
2139 }
2140
2141 public <U> CompletableFuture<U> handleAsync(
2142 BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
2143 return uniHandleStage(screenExecutor(executor), fn);
2144 }
2145
2146 /**
2147 * Returns this CompletableFuture.
2148 *
2149 * @return this CompletableFuture
2150 */
2151 public CompletableFuture<T> toCompletableFuture() {
2152 return this;
2153 }
2154
2155 // not in interface CompletionStage
2156
2157 /**
2158 * Returns a new CompletableFuture that is completed when this
2159 * CompletableFuture completes, with the result of the given
2160 * function of the exception triggering this CompletableFuture's
2161 * completion when it completes exceptionally; otherwise, if this
2162 * CompletableFuture completes normally, then the returned
2163 * CompletableFuture also completes normally with the same value.
2164 * Note: More flexible versions of this functionality are
2165 * available using methods {@code whenComplete} and {@code handle}.
2166 *
2167 * @param fn the function to use to compute the value of the
2168 * returned CompletableFuture if this CompletableFuture completed
2169 * exceptionally
2170 * @return the new CompletableFuture
2171 */
2172 public CompletableFuture<T> exceptionally(
2173 Function<Throwable, ? extends T> fn) {
2174 return uniExceptionallyStage(fn);
2175 }
2176
2177 /* ------------- Arbitrary-arity constructions -------------- */
2178
2179 /**
2180 * Returns a new CompletableFuture that is completed when all of
2181 * the given CompletableFutures complete. If any of the given
2182 * CompletableFutures complete exceptionally, then the returned
2183 * CompletableFuture also does so, with a CompletionException
2184 * holding this exception as its cause. Otherwise, the results,
2185 * if any, of the given CompletableFutures are not reflected in
2186 * the returned CompletableFuture, but may be obtained by
2187 * inspecting them individually. If no CompletableFutures are
2188 * provided, returns a CompletableFuture completed with the value
2189 * {@code null}.
2190 *
2191 * <p>Among the applications of this method is to await completion
2192 * of a set of independent CompletableFutures before continuing a
2193 * program, as in: {@code CompletableFuture.allOf(c1, c2,
2194 * c3).join();}.
2195 *
2196 * @param cfs the CompletableFutures
2197 * @return a new CompletableFuture that is completed when all of the
2198 * given CompletableFutures complete
2199 * @throws NullPointerException if the array or any of its elements are
2200 * {@code null}
2201 */
2202 public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
2203 return andTree(cfs, 0, cfs.length - 1);
2204 }
2205
2206 /**
2207 * Returns a new CompletableFuture that is completed when any of
2208 * the given CompletableFutures complete, with the same result.
2209 * Otherwise, if it completed exceptionally, the returned
2210 * CompletableFuture also does so, with a CompletionException
2211 * holding this exception as its cause. If no CompletableFutures
2212 * are provided, returns an incomplete CompletableFuture.
2213 *
2214 * @param cfs the CompletableFutures
2215 * @return a new CompletableFuture that is completed with the
2216 * result or exception of any of the given CompletableFutures when
2217 * one completes
2218 * @throws NullPointerException if the array or any of its elements are
2219 * {@code null}
2220 */
2221 public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
2222 return orTree(cfs, 0, cfs.length - 1);
2223 }
2224
2225 /* ------------- Control and status methods -------------- */
2226
2227 /**
2228 * If not already completed, completes this CompletableFuture with
2229 * a {@link CancellationException}. Dependent CompletableFutures
2230 * that have not already completed will also complete
2231 * exceptionally, with a {@link CompletionException} caused by
2232 * this {@code CancellationException}.
2233 *
2234 * @param mayInterruptIfRunning this value has no effect in this
2235 * implementation because interrupts are not used to control
2236 * processing.
2237 *
2238 * @return {@code true} if this task is now cancelled
2239 */
2240 public boolean cancel(boolean mayInterruptIfRunning) {
2241 boolean cancelled = (result == null) &&
2242 internalComplete(new AltResult(new CancellationException()));
2243 postComplete();
2244 return cancelled || isCancelled();
2245 }
2246
2247 /**
2248 * Returns {@code true} if this CompletableFuture was cancelled
2249 * before it completed normally.
2250 *
2251 * @return {@code true} if this CompletableFuture was cancelled
2252 * before it completed normally
2253 */
2254 public boolean isCancelled() {
2255 Object r;
2256 return ((r = result) instanceof AltResult) &&
2257 (((AltResult)r).ex instanceof CancellationException);
2258 }
2259
2260 /**
2261 * Returns {@code true} if this CompletableFuture completed
2262 * exceptionally, in any way. Possible causes include
2263 * cancellation, explicit invocation of {@code
2264 * completeExceptionally}, and abrupt termination of a
2265 * CompletionStage action.
2266 *
2267 * @return {@code true} if this CompletableFuture completed
2268 * exceptionally
2269 */
2270 public boolean isCompletedExceptionally() {
2271 Object r;
2272 return ((r = result) instanceof AltResult) && r != NIL;
2273 }
2274
2275 /**
2276 * Forcibly sets or resets the value subsequently returned by
2277 * method {@link #get()} and related methods, whether or not
2278 * already completed. This method is designed for use only in
2279 * error recovery actions, and even in such situations may result
2280 * in ongoing dependent completions using established versus
2281 * overwritten outcomes.
2282 *
2283 * @param value the completion value
2284 */
2285 public void obtrudeValue(T value) {
2286 result = (value == null) ? NIL : value;
2287 postComplete();
2288 }
2289
2290 /**
2291 * Forcibly causes subsequent invocations of method {@link #get()}
2292 * and related methods to throw the given exception, whether or
2293 * not already completed. This method is designed for use only in
2294 * error recovery actions, and even in such situations may result
2295 * in ongoing dependent completions using established versus
2296 * overwritten outcomes.
2297 *
2298 * @param ex the exception
2299 * @throws NullPointerException if the exception is null
2300 */
2301 public void obtrudeException(Throwable ex) {
2302 if (ex == null) throw new NullPointerException();
2303 result = new AltResult(ex);
2304 postComplete();
2305 }
2306
2307 /**
2308 * Returns the estimated number of CompletableFutures whose
2309 * completions are awaiting completion of this CompletableFuture.
2310 * This method is designed for use in monitoring system state, not
2311 * for synchronization control.
2312 *
2313 * @return the number of dependent CompletableFutures
2314 */
2315 public int getNumberOfDependents() {
2316 int count = 0;
2317 for (Completion p = stack; p != null; p = p.next)
2318 ++count;
2319 return count;
2320 }
2321
2322 /**
2323 * Returns a string identifying this CompletableFuture, as well as
2324 * its completion state. The state, in brackets, contains the
2325 * String {@code "Completed Normally"} or the String {@code
2326 * "Completed Exceptionally"}, or the String {@code "Not
2327 * completed"} followed by the number of CompletableFutures
2328 * dependent upon its completion, if any.
2329 *
2330 * @return a string identifying this CompletableFuture, as well as its state
2331 */
2332 public String toString() {
2333 Object r = result;
2334 int count;
2335 return super.toString() +
2336 ((r == null) ?
2337 (((count = getNumberOfDependents()) == 0) ?
2338 "[Not completed]" :
2339 "[Not completed, " + count + " dependents]") :
2340 (((r instanceof AltResult) && ((AltResult)r).ex != null) ?
2341 "[Completed exceptionally]" :
2342 "[Completed normally]"));
2343 }
2344
2345 // Unsafe mechanics
2346 private static final sun.misc.Unsafe UNSAFE;
2347 private static final long RESULT;
2348 private static final long STACK;
2349 static {
2350 try {
2351 UNSAFE = sun.misc.Unsafe.getUnsafe();
2352 Class<?> k = CompletableFuture.class;
2353 RESULT = UNSAFE.objectFieldOffset
2354 (k.getDeclaredField("result"));
2355 STACK = UNSAFE.objectFieldOffset
2356 (k.getDeclaredField("stack"));
2357 } catch (Exception x) {
2358 throw new Error(x);
2359 }
2360 }
2361 }