ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.178
Committed: Sat Oct 3 18:17:51 2015 UTC (8 years, 8 months ago) by jsr166
Branch: MAIN
Changes since 1.177: +1 -1 lines
Log Message:
compare longs against 0L, not 0

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