ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.180
Committed: Sun Nov 15 23:31:51 2015 UTC (8 years, 6 months ago) by jsr166
Branch: MAIN
Changes since 1.179: +2 -0 lines
Log Message:
use suppressed exception facility with whenComplete

File Contents

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