ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.145
Committed: Wed Jan 14 18:24:10 2015 UTC (9 years, 4 months ago) by jsr166
Branch: MAIN
Changes since 1.144: +1 -1 lines
Log Message:
typo

File Contents

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