ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.183
Committed: Wed Mar 23 23:05:51 2016 UTC (8 years, 2 months ago) by dl
Branch: MAIN
Changes since 1.182: +4 -4 lines
Log Message:
Avoid unnecessary cleanStack calls

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