ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.163
Committed: Fri Sep 4 20:58:17 2015 UTC (8 years, 8 months ago) by jsr166
Branch: MAIN
Changes since 1.162: +2 -1 lines
Log Message:
whitespace

File Contents

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